"""AI 分析核心服务 负责： 1. 调用 OpenAI 兼容 API 进行分析 2. Function Calling 循环（AI 可主动获取数据） 3. 采集当日盘面快照（供次日环比） 4. 保存报告到数据库 """ import asyncio import json import traceback 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: 获取个股资金流向 - get_news: 获取财经快讯（新浪7x24，用于重要消息面） 重要规则： 1. 你必须先调用工具获取数据，然后基于数据进行分析 2. 不要凭空编造数据，所有数据必须来自工具返回 3. 如果工具返回空数据，如实说明数据不可用 4. 分析完成后给出明确的结论和建议""" DAILY_ANALYSIS_PROMPT = """请对 {trade_date} 的A股市场进行收盘分析，生成一份完整的分析报告。 {prev_report_section} {prev_snapshot_section} 请先调用以下工具获取数据： 1. get_market_dashboard - 获取市场整体数据（含涨跌统计、市场温度、连板梯队 limitLadder、板块资金流 sectorFundFlow、两融、海外与期货 globalMarkets） 2. get_theme_history(date="{trade_date}") - 获取今日题材涨幅 3. get_active_core_stocks - 获取核心股数据 4. get_news(limit=30) - 获取今日财经快讯 生成报告时必须遵守以下格式规则： 1. 报告标题（# 一级标题）之后的第一行，必须是一个引用块"定调摘要"，格式严格为： > 今日定调：<一句话核心结论，不超过80字，必须包含1-2个关键数字（如成交额、涨停家数、市场温度）> 2. 量能与情绪类数字必须给环比：若上方提供了"前一交易日盘面数据快照"，成交额、涨跌家数、涨停数、两融余额、市场温度等在与昨日对比后表述（如"成交额2.05万亿，较昨日缩量约700亿"）；没有昨日快照则如实说明"暂无昨日数据"。 3. 连板梯队必须完整呈现 get_market_dashboard 返回的 limitLadder：从最高连板到2连板逐级列表格，每只标注涨停原因（reason字段）与封单金额（sealWan，单位万，为空则不写）；首板只挑3-5只人气最高的点评。 4. 板块资金面必须引用 get_market_dashboard 返回的 sectorFundFlow：行业主力净流入TOP3、净流出TOP3、概念净流入TOP3（单位亿元），结合题材分析说明资金动向。 5. 海外市场与国内期货必须引用 get_market_dashboard 返回的 globalMarkets：overseas 为海外主要指数（纳斯达克/道琼斯/标普500/恒生/日经/富时），futures 为国内期货主力合约（按成交额降序，已含价格与涨跌幅）；点评与A股关联度高的品种（股指期货、原油、贵金属、黑色系），数据缺失则如实说明。 6. 重要消息面必须基于 get_news 返回的快讯整理：挑5-8条对次日盘面影响最大的消息，每条格式为"【分类】一句话新闻 —— 一句影响解读"（分类用：宏观/政策/行业/公司/海外）；快讯中若没有某方面的重要消息，如实说明，严禁编造工具中不存在的新闻。 7. 适当使用表格展示数据对比。 然后基于数据生成报告。报告共八章，将由系统分三次调用完成，每次调用只负责其中一部分，具体写作指令由后续消息给出。 请用 Markdown 格式输出，适当使用表格展示数据对比。""" # 分段生成指令：网关对单次 LLM 请求有约120s硬超时，整篇报告一次生成必被掐断， # 故拆为三段（每段约1200-1600字），各自独立调用后拼接 REPORT_PARTS = [ """现在写报告的【第1部分】，只输出这一部分，直接输出 Markdown，不要任何开场白或说明： 1. 以 `# {title} A股收盘分析报告` 一级标题开头 2. 写 `## 一、市场总览`（指数表格、涨跌统计须环比、市场温度）与 `## 二、题材热点分析`（涨幅前5、持续活跃、新兴热点、退潮警示、板块主力资金流TOP3） 全文控制在1600字以内。""", """现在写报告的【第2部分】，只输出这一部分，直接输出 Markdown，不要重复之前内容： - `## 三、核心股追踪`（连板梯队完整表格：层级/股票/涨停原因/封单，首板挑3-5只人气股点评；核心股表现；龙头辨识） - `## 四、关注方向`（明日题材方向、潜在交易机会） 全文控制在1300字以内。""", """现在写报告的【第3部分】，只输出这一部分，直接输出 Markdown，不要重复之前内容： - `## 五、下个交易日建议`（大盘预判、题材方向、核心股、仓位策略、规避方向） - `## 六、重要消息面`（5-8条，格式【分类】新闻——影响解读） - `## 七、海外市场与国内期货`（点评对次日A股的影响） - `## 八、风险提示` 全文控制在1900字以内。""", ] async def _consume_sse(resp: httpx.Response) -> dict: """消费 OpenAI 兼容 SSE 流，拼装为与非流式响应相同的结构""" content_parts: list[str] = [] finish_reason = "" usage: dict = {} # tool_calls 按 index 拼装（流式下 arguments 分片到达） tool_acc: dict[int, dict] = {} async for line in resp.aiter_lines(): if not line.startswith("data:"): continue data = line[5:].strip() if not data or data == "[DONE]": continue try: chunk = json.loads(data) except json.JSONDecodeError: continue if chunk.get("usage"): usage = chunk["usage"] choices = chunk.get("choices") or [] if not choices: continue delta = choices[0].get("delta") or {} if delta.get("content"): content_parts.append(delta["content"]) for tc in delta.get("tool_calls") or []: idx = tc.get("index", 0) slot = tool_acc.setdefault(idx, {"id": "", "type": "function", "function": {"name": "", "arguments": ""}}) if tc.get("id"): slot["id"] = tc["id"] fn = tc.get("function") or {} if fn.get("name"): slot["function"]["name"] += fn["name"] if fn.get("arguments"): slot["function"]["arguments"] += fn["arguments"] if choices[0].get("finish_reason"): finish_reason = choices[0]["finish_reason"] message: dict = {"role": "assistant", "content": "".join(content_parts) or None} if tool_acc: message["tool_calls"] = [ {"id": tool_acc[i]["id"], "type": "function", "function": tool_acc[i]["function"]} for i in sorted(tool_acc) ] return {"choices": [{"message": message, "finish_reason": finish_reason}], "usage": usage} async def call_llm(messages: list, tools: list = None) -> dict: """调用 OpenAI 兼容 API（流式）。 必须用 stream：网关对非流式请求有约120s的代理超时，长生成会被 502 掐断； 流式下字节持续到达不会被判定超时。返回结构与非流式一致。 """ async with httpx.AsyncClient() as client: payload = { "model": AI_MODEL, "messages": messages, "stream": True, } if AI_MAX_TOKENS is not None: payload["max_tokens"] = AI_MAX_TOKENS if AI_TEMPERATURE is not None: payload["temperature"] = AI_TEMPERATURE if tools: payload["tools"] = tools payload["tool_choice"] = "auto" max_attempts = 4 for attempt in range(1, max_attempts + 1): try: async with client.stream( "POST", f"{AI_API_BASE}/chat/completions", headers={ "Authorization": f"Bearer {AI_API_KEY}", "Content-Type": "application/json", "Accept": "text/event-stream", }, json=payload, timeout=600, ) as resp: if resp.status_code == 429 and attempt < max_attempts: wait = min(30 * attempt, 90) print(f"[ai-service] LLM 429 限流，{wait}s 后重试（第 {attempt}/{max_attempts - 1} 次）") await asyncio.sleep(wait) continue if resp.status_code >= 500 and attempt < max_attempts: print(f"[ai-service] LLM {resp.status_code}，{min(15 * attempt, 60)}s 后重试") await asyncio.sleep(min(15 * attempt, 60)) continue resp.raise_for_status() return await _consume_sse(resp) except httpx.TransportError as e: # 网络层错误（超时/断连）也值得重试 if attempt < max_attempts: wait = min(15 * attempt, 60) print(f"[ai-service] LLM 网络错误（{type(e).__name__}），{wait}s 后重试") await asyncio.sleep(wait) continue raise raise RuntimeError("LLM 调用重试次数耗尽") 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) # 采集当日盘面快照（供次日环比），并读取上一交易日快照注入 prompt await _capture_market_snapshot(trade_date) prev_snapshot_section = _get_prev_snapshot_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, prev_snapshot_section=prev_snapshot_section, )} ] # ── 阶段一：工具轮（只获取数据，模型若直接开写报告则丢弃，由阶段二重写） ── tools_used = [] total_tokens = 0 llm_calls = 0 was_truncated = False max_rounds = 8 for i in range(max_rounds): response = await call_llm(messages, tools=TOOLS) llm_calls += 1 total_tokens += response.get("usage", {}).get("total_tokens", 0) choice = response["choices"][0] message = choice["message"] finish_reason = choice.get("finish_reason", "") print(f"[ai-service] tool round {i}: finish={finish_reason}, " f"content_len={len(message.get('content') or '')}, " f"tool_calls={len(message.get('tool_calls') or [])}") if finish_reason == "tool_calls" and message.get("tool_calls"): messages.append(message) for tool_call in message["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 }) break # 工具已齐，立即进入分段写作（再问一轮模型只会空转120s） # 非工具轮（模型直接开写/空返回）：只要有工具结果就直接进入分段写作； # 一轮工具都没拿到则重试 if tools_used: break print(f"[ai-service] 未获取到工具数据，重试（{i + 1}/{max_rounds}）") await asyncio.sleep(5) if not tools_used: raise RuntimeError("工具数据获取失败，无法生成报告") # ── 阶段二：分段生成报告（绕开网关单请求约120s硬超时） ── final_content = "" for part_idx, part_prompt in enumerate(REPORT_PARTS): part_content = "" for attempt, backoff in ((1, 0), (2, 10), (3, 30)): if backoff: await asyncio.sleep(backoff) part_messages = messages + [{ "role": "user", "content": part_prompt.format(title=trade_date) if part_idx == 0 else part_prompt, }] response = await call_llm(part_messages) llm_calls += 1 total_tokens += response.get("usage", {}).get("total_tokens", 0) choice = response["choices"][0] part_content = choice["message"].get("content") or "" finish = choice.get("finish_reason", "") print(f"[ai-service] part {part_idx + 1} attempt {attempt}: finish={finish}, len={len(part_content)}") if part_content and finish in ("stop", "length"): break print(f"[ai-service] part {part_idx + 1} 生成异常，重试") if not part_content: was_truncated = True print(f"[ai-service] 警告：part {part_idx + 1} 三次尝试均失败") part_content = f"\n\n> （第{part_idx + 1}部分生成失败，请稍后重新生成）\n" final_content += (final_content and "\n\n" or "") + part_content summary = _extract_summary(final_content) report_id = _save_report(trade_date, final_content, summary, tools_used, total_tokens, llm_calls) print(f"[ai-service] 生成完成：LLM调用 {llm_calls} 次，tokens {total_tokens}") return {"id": report_id, "tokens_used": total_tokens, "llm_calls": llm_calls, "tools_used": tools_used, "truncated": was_truncated} def _extract_summary(content: str) -> str: """摘要：正文截断前200字（供管理列表展示）""" return (content or "")[:200].replace("\n", " ").strip() def _num(v) -> str: """快照数值安全转字符串""" if isinstance(v, float): return f"{v:g}" return str(v if v is not None else "-") def _fmt_amount(v) -> str: """成交额（元）→ 万亿/亿 可读格式""" try: v = float(v) except (TypeError, ValueError): return "-" if v >= 1e12: return f"{v / 1e12:.2f}万亿" if v >= 1e8: return f"{v / 1e8:.0f}亿" return f"{v:.0f}元" def _fmt_index(idx: dict) -> str: if not idx: return "-" try: pct = round(float(idx.get("changePct", 0)), 2) except (TypeError, ValueError): pct = 0 sign = "+" if pct >= 0 else "" return f"{idx.get('price', '-')}（{sign}{pct}%）" def _fmt_temperature(v) -> str: """温度可能是 dict（score/label/factors），取分数与标签""" if isinstance(v, dict): score = v.get("score") label = v.get("label") or "" return f"{score}分{('（' + label + '）') if label else ''}" return _num(v) async def _capture_market_snapshot(trade_date: str) -> bool: """采集当日盘面快照入库（指数+涨跌统计），供次日分析做环比""" try: from routes.market_dashboard import _build_dashboard data = await _build_dashboard() stats = data.get("marketStats") or {} # 数据有效性校验：fuyao 拉取失败时涨跌统计全 0，空快照会污染次日环比 if not data.get("indices") or (stats.get("upCount", 0) + stats.get("downCount", 0) == 0): print(f"[ai-service] 盘面数据无效，跳过快照入库 {trade_date}") return False payload = json.dumps( {"indices": data.get("indices", []), "marketStats": stats}, ensure_ascii=False, default=str, ) conn = get_connection() try: conn.execute( """INSERT INTO daily_market_stats (trade_date, payload) VALUES (?, ?) ON CONFLICT(trade_date) DO UPDATE SET payload = excluded.payload""", (trade_date, payload), ) conn.commit() finally: conn.close() return True except Exception: print(f"[ai-service] 市场快照采集失败 {trade_date}：") traceback.print_exc() return False def _get_prev_snapshot_section(trade_date: str) -> str: """读取上一交易日盘面快照，格式化为 prompt 中的环比数据段""" conn = get_connection() try: row = conn.execute( "SELECT trade_date, payload FROM daily_market_stats WHERE trade_date < ? ORDER BY trade_date DESC LIMIT 1", (trade_date,), ).fetchone() if not row: return "" try: snap = json.loads(row["payload"]) except (TypeError, ValueError): return "" stats = snap.get("marketStats") or {} indices = {i.get("name"): i for i in (snap.get("indices") or []) if isinstance(i, dict)} idx_line = "、".join( f"{name} {_fmt_index(indices.get(name))}" for name in ("上证指数", "深证成指", "创业板指", "科创50") ) margin_line = "" if stats.get("marginBalanceYi"): change = stats.get("marginChangeYi") change_txt = "" if change is not None: sign = "+" if float(change) >= 0 else "" change_txt = f"（较前一日 {sign}{_num(change)}亿）" margin_line = f"\n- 两融余额：{_num(stats.get('marginBalanceYi'))}亿{change_txt}，数据日期 {stats.get('marginDate') or '-'}（T+1）" return f"""以下是前一交易日（{row["trade_date"]}）的盘面数据快照，报告中的量能与情绪数字必须给出与它的环比对比： - 两市成交额：{_fmt_amount(stats.get("totalTurnover"))} - 上涨/下跌/平盘：{_num(stats.get("upCount"))}/{_num(stats.get("downCount"))}/{_num(stats.get("flatCount"))}，涨停 {_num(stats.get("limitUp"))} 家、跌停 {_num(stats.get("limitDown"))} 家、炸板 {_num(stats.get("limitBreak"))} 家（炸板率 {_num(stats.get("breakRate"))}%） - 强势/弱势股：{_num(stats.get("strongCount"))}/{_num(stats.get("weakCount"))}，市场宽度 {_num(stats.get("marketBreadth"))}% - 市场温度：{_fmt_temperature(stats.get("temperature"))}，竞价信号：{stats.get("auctionSignal") or "-"}{margin_line} - 指数收盘：{idx_line} --- """ finally: conn.close() def _save_report(trade_date: str, content: str, summary: str, tools_used: list, tokens_used: int, llm_calls: int = 0) -> int: """保存报告到数据库（同日重生成：覆盖内容、generation_count+1、tokens 记当次消耗）""" conn = get_connection() try: conn.execute( """INSERT INTO ai_reports (trade_date, report_type, title, content, summary, tools_used, model, tokens_used, llm_calls, updated_at) VALUES (?, 'daily', ?, ?, ?, ?, ?, ?, ?, datetime('now','localtime')) ON CONFLICT(trade_date, report_type) DO UPDATE SET title = excluded.title, content = excluded.content, summary = excluded.summary, tools_used = excluded.tools_used, model = excluded.model, tokens_used = excluded.tokens_used, llm_calls = excluded.llm_calls, updated_at = excluded.updated_at, generation_count = ai_reports.generation_count + 1""", ( trade_date, f"{trade_date} A股收盘分析", content, summary, json.dumps(tools_used), AI_MODEL, tokens_used, llm_calls, ), ) conn.commit() row = conn.execute( "SELECT id FROM ai_reports WHERE trade_date = ? AND report_type = 'daily'", (trade_date,), ).fetchone() return row["id"] 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 "" return f"""以下是前一个交易日（{prev_date}）的分析报告，请参考其中的分析逻辑和关注方向，结合今日数据进行对比分析： {prev_content} --- """ finally: conn.close()