220 lines
7.1 KiB
Python
220 lines
7.1 KiB
Python
"""AI 分析核心服务
|
|
|
|
负责：
|
|
1. 调用 OpenAI 兼容 API 进行分析
|
|
2. Function Calling 循环（AI 可主动获取数据）
|
|
3. 保存报告到数据库
|
|
"""
|
|
|
|
import json
|
|
import httpx
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
from services.ai_config import AI_API_BASE, AI_API_KEY, AI_MODEL, AI_MAX_TOKENS, AI_TEMPERATURE
|
|
from services.ai_tools import TOOLS, execute_tool
|
|
from database import get_connection
|
|
|
|
_CST = timezone(timedelta(hours=8))
|
|
|
|
SYSTEM_PROMPT = """你是一位专业的A股市场分析师，擅长从数据中发现投资机会。
|
|
|
|
你的分析风格：
|
|
- 数据驱动，基于真实数据而非主观臆断
|
|
- 逻辑清晰，先总后分，层层递进
|
|
- 观点明确，给出具体的操作建议
|
|
- 风险提示，每次推荐都需说明风险点
|
|
|
|
可用工具：
|
|
- get_market_dashboard: 获取市场整体数据（指数/涨跌统计/行业强度/事件情报/市场温度）
|
|
- get_theme_history: 获取指定日期的题材涨幅排行
|
|
- get_active_core_stocks: 获取核心股追踪数据（10日涨幅矩阵+所属题材）
|
|
- get_stock_quote: 获取个股实时行情
|
|
- get_fund_flow: 获取个股资金流向
|
|
|
|
重要规则：
|
|
1. 你必须先调用工具获取数据，然后基于数据进行分析
|
|
2. 不要凭空编造数据，所有数据必须来自工具返回
|
|
3. 如果工具返回空数据，如实说明数据不可用
|
|
4. 分析完成后给出明确的结论和建议"""
|
|
|
|
DAILY_ANALYSIS_PROMPT = """请对 {trade_date} 的A股市场进行收盘分析，生成一份完整的分析报告。
|
|
|
|
{prev_report_section}
|
|
|
|
请先调用以下工具获取数据：
|
|
1. get_market_dashboard - 获取市场整体数据
|
|
2. get_theme_history(date="{trade_date}") - 获取今日题材涨幅
|
|
3. get_active_core_stocks - 获取核心股数据
|
|
|
|
然后基于数据生成报告，结构如下：
|
|
|
|
## 一、市场总览
|
|
- 主要指数表现（上证、深证、创业板）
|
|
- 涨跌家数统计
|
|
- 市场温度评估
|
|
|
|
## 二、题材热点分析
|
|
- 今日涨幅前5题材
|
|
- 持续活跃的题材
|
|
- 新兴热点题材
|
|
|
|
## 三、核心股追踪
|
|
- 连板股分析
|
|
- 核心股表现
|
|
- 龙头股辨识
|
|
|
|
## 四、关注方向
|
|
- 明日值得关注的题材方向
|
|
- 潜在的交易机会
|
|
|
|
## 五、风险提示
|
|
- 需要警惕的风险因素
|
|
- 操作建议
|
|
|
|
请用 Markdown 格式输出，适当使用表格展示数据对比。"""
|
|
|
|
|
|
async def call_llm(messages: list, tools: list = None) -> dict:
|
|
"""调用 OpenAI 兼容 API"""
|
|
async with httpx.AsyncClient() as client:
|
|
payload = {
|
|
"model": AI_MODEL,
|
|
"messages": messages,
|
|
"max_tokens": AI_MAX_TOKENS,
|
|
"temperature": AI_TEMPERATURE,
|
|
}
|
|
if tools:
|
|
payload["tools"] = tools
|
|
payload["tool_choice"] = "auto"
|
|
|
|
resp = await client.post(
|
|
f"{AI_API_BASE}/chat/completions",
|
|
headers={
|
|
"Authorization": f"Bearer {AI_API_KEY}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json=payload,
|
|
timeout=120,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
async def collect_ai_analysis(trade_date: str) -> dict:
|
|
"""AI 分析主流程
|
|
|
|
Args:
|
|
trade_date: 交易日 YYYY-MM-DD
|
|
|
|
Returns:
|
|
{"id": int, "tokens_used": int, "tools_used": list}
|
|
"""
|
|
prev_report_section = _get_prev_report_section(trade_date)
|
|
messages = [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": DAILY_ANALYSIS_PROMPT.format(
|
|
trade_date=trade_date,
|
|
prev_report_section=prev_report_section
|
|
)}
|
|
]
|
|
|
|
tools_used = []
|
|
total_tokens = 0
|
|
final_content = ""
|
|
was_truncated = False
|
|
|
|
for i in range(8):
|
|
response = await call_llm(messages, tools=TOOLS)
|
|
total_tokens += response.get("usage", {}).get("total_tokens", 0)
|
|
|
|
choice = response["choices"][0]
|
|
message = choice["message"]
|
|
messages.append(message)
|
|
|
|
finish_reason = choice.get("finish_reason", "")
|
|
|
|
if finish_reason == "stop":
|
|
final_content = message.get("content") or ""
|
|
break
|
|
|
|
if finish_reason == "length":
|
|
# 响应被截断，保存已生成的内容
|
|
was_truncated = True
|
|
final_content = message.get("content") or ""
|
|
if not final_content:
|
|
# 如果当前消息没有内容，尝试从之前的消息中提取
|
|
for msg in reversed(messages):
|
|
if msg.get("role") == "assistant" and msg.get("content"):
|
|
final_content = msg["content"]
|
|
break
|
|
print(f"[ai-service] 警告：响应被截断 (finish_reason=length)")
|
|
break
|
|
|
|
if finish_reason == "tool_calls":
|
|
for tool_call in message.get("tool_calls", []):
|
|
func_name = tool_call["function"]["name"]
|
|
func_args = json.loads(tool_call["function"]["arguments"])
|
|
tools_used.append(func_name)
|
|
|
|
result = await execute_tool(func_name, func_args)
|
|
messages.append({
|
|
"role": "tool",
|
|
"tool_call_id": tool_call["id"],
|
|
"content": result
|
|
})
|
|
|
|
summary = final_content[:200].replace("\n", " ") if final_content else ""
|
|
report_id = _save_report(trade_date, final_content, summary, tools_used, total_tokens)
|
|
|
|
return {"id": report_id, "tokens_used": total_tokens, "tools_used": tools_used, "truncated": was_truncated}
|
|
|
|
|
|
def _save_report(trade_date: str, content: str, summary: str, tools_used: list, tokens_used: int) -> int:
|
|
"""保存报告到数据库"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.execute(
|
|
"""INSERT OR REPLACE INTO ai_reports
|
|
(trade_date, report_type, title, content, summary, tools_used, model, tokens_used)
|
|
VALUES (?, 'daily', ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
trade_date,
|
|
f"{trade_date} A股收盘分析",
|
|
content,
|
|
summary,
|
|
json.dumps(tools_used),
|
|
AI_MODEL,
|
|
tokens_used,
|
|
)
|
|
)
|
|
conn.commit()
|
|
return cursor.lastrowid
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _get_prev_report_section(trade_date: str) -> str:
|
|
"""获取前一个交易日的报告摘要，用于上下文参考"""
|
|
conn = get_connection()
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT trade_date, content FROM ai_reports WHERE trade_date < ? AND report_type = 'daily' ORDER BY trade_date DESC LIMIT 1",
|
|
(trade_date,)
|
|
).fetchone()
|
|
if not row:
|
|
return ""
|
|
prev_date = row["trade_date"]
|
|
prev_content = row["content"] or ""
|
|
# 取报告的前1500字符作为参考上下文
|
|
truncated = prev_content[:1500]
|
|
if len(prev_content) > 1500:
|
|
truncated += "\n...(已截断)"
|
|
return f"""以下是前一个交易日（{prev_date}）的分析报告，请参考其中的分析逻辑和关注方向，结合今日数据进行对比分析：
|
|
|
|
{truncated}
|
|
|
|
---
|
|
"""
|
|
finally:
|
|
conn.close()
|