541 lines
27 KiB
Python
541 lines
27 KiB
Python
"""AI 分析核心服务
|
||
|
||
负责：
|
||
1. 调用 OpenAI 兼容 API 进行分析
|
||
2. Function Calling 循环（AI 可主动获取数据）
|
||
3. 采集当日盘面快照（供次日环比）
|
||
4. 保存报告到数据库
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import re
|
||
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_consecutive_core_stocks: 获取最近N个交易日每天都上榜的核心股（缺一日都不行），用于识别持续活跃的热点股
|
||
- get_stock_quote: 获取个股实时行情
|
||
- get_fund_flow: 获取个股资金流向
|
||
- get_news: 获取财经快讯（新浪7x24，用于重要消息面）
|
||
|
||
重要规则：
|
||
1. 你必须先调用工具获取数据，然后基于数据进行分析
|
||
2. 不要凭空编造数据，所有数据必须来自工具返回
|
||
3. 如果工具返回空数据，如实说明数据不可用
|
||
4. 分析完成后给出明确的结论和建议
|
||
|
||
工作原则：
|
||
1. 事实优先：所有数字、股票名称与代码、涨停原因、封单金额必须能在工具返回中找到出处；工具没有的写「—（今日无数据）」，严禁凭记忆、常识或推理补全。
|
||
2. 时间锚定：只分析给定交易日当天及之前的真实数据，禁止引用训练记忆中的行情、政策或题材。
|
||
3. 判断必须挂数字：每个结论后面要有价格/涨跌幅/金额/家数支撑，不做无数据的定性判断。
|
||
4. 说人话：结论先行、短句、少形容词；禁用「整体来看」「值得注意的是」「情绪有所回暖」这类无信息量的套话。
|
||
|
||
输出格式（前端按 ## 二级标题切卡片渲染，红涨绿跌依赖数字符号）：
|
||
- 章节标题严格用 `## 一、xxx` 形式，标题不超过 12 字
|
||
- 涨跌幅、环比变化、资金净额等有方向的数字必须带 + 或 - 号（如 +2.31%、-15.6亿）；不带符号的数字不会被着色
|
||
- 多维度对比一律用 Markdown 表格，表头标注单位"""
|
||
|
||
# 一次工具轮就必须拿全的核心工具：模型常只调 1-2 个就开写，缺失会让对应章节数据空洞
|
||
REQUIRED_TOOLS = ("get_market_dashboard", "get_theme_history",
|
||
"get_active_core_stocks", "get_consecutive_core_stocks", "get_news")
|
||
|
||
DAILY_ANALYSIS_PROMPT = """请对 {trade_date}（A股交易日） 的A股市场进行收盘分析，生成一份完整的分析报告。
|
||
|
||
{prev_report_section}
|
||
{prev_snapshot_section}
|
||
|
||
你的分析风格：
|
||
- 数据驱动，基于真实数据而非主观臆断
|
||
- 逻辑清晰，先总后分，层层递进
|
||
- 观点明确，给出具体的操作建议
|
||
- 风险提示，每次推荐都需说明风险点
|
||
|
||
重要规则：
|
||
1. 你必须先调用工具获取数据，然后基于数据进行分析
|
||
2. 不要凭空编造数据，所有数据必须来自工具返回
|
||
3. 如果工具返回空数据，如实说明数据不可用
|
||
4. 分析完成后给出明确的结论和建议
|
||
|
||
■ 第一步 · 取数（必须一次并行完成）
|
||
在同一次回复中并行发起以下 5 个工具调用，不要拆成多轮、不要只调其中一部分：
|
||
1. get_market_dashboard() —— 指数、涨跌统计、市场温度、连板梯队 limitLadder、板块资金流 sectorFundFlow、两融、海外指数与国内期货 globalMarkets
|
||
2. get_theme_history(date="{trade_date}") —— 今日题材涨幅榜
|
||
3. get_active_core_stocks() —— 核心股 10 日涨幅矩阵与所属题材
|
||
4. get_consecutive_core_stocks(days=3) —— 最近3个交易日每天都上榜的核心股（缺一日都不行）
|
||
5. get_news(limit=30) —— 今日财经快讯
|
||
只有在需要核实某只具体个股时，才额外调用 get_stock_quote / get_fund_flow，合计不超过 2 次。
|
||
|
||
■ 第二步 · 写作（以下规约对每一部分都生效）
|
||
0. 报告标题（# 一级标题）之后的第一行，必须是一个引用块"定调摘要"，格式严格为：
|
||
> 今日定调：<一句话核心结论，不超过80字，必须包含1-2个关键数字（如成交额、涨停家数、市场温度）>
|
||
1. 事实约束：所有数字、个股代码与名称、涨停原因、封单金额必须出自工具返回，找不到就写「—（今日无数据）」，禁止推测或用常识填补；禁止用「预计/有望/大概率」代替数据。
|
||
2. 环比约束：量能与情绪类数字必须给环比,若上方提供了"前一交易日盘面数据快照"（成交额、涨跌家数、涨停/跌停/炸板、市场温度、两融余额、主要指数），首次提到时必须写成「今日值（较昨日 ±变化）」；快照缺失时写「暂无昨日基准」，不得编造环比。
|
||
3. 单位口径（全文统一，表头标注单位）：成交额→亿元（≥1万亿时写 x.xx 万亿）；资金净额→亿元；封单金额→万元（limitLadder 的 sealWan 字段）；指数→点。
|
||
4. 符号规范：涨跌幅、环比变化、净流入/流出一律带 + 或 -，这是前端红涨绿跌的着色依据。
|
||
5. 连板梯队：连板梯队必须完整呈现 get_market_dashboard 返回的 limitLadder：从最高连板到2连板逐级列表格，每只标注涨停原因（reason字段）与封单金额（sealWan，单位万，为空则不写）；首板只挑3-5只人气最高的点评。
|
||
6. 板块资金：板块资金面必须引用 get_market_dashboard 返回的 sectorFundFlow：行业主力净流入TOP5、净流出TOP5、概念净流入TOP5（亿元，带符号），结合题材分析说明资金动向。
|
||
7. 海外市场与国内期货：必须引用 get_market_dashboard 返回的 globalMarkets：overseas 为海外主要指数（纳斯达克/道琼斯/标普500/恒生/日经/富时），futures 为国内期货主力合约（按成交额降序，已含价格与涨跌幅）；点评与A股关联度高的品种（股指期货、原油、贵金属、黑色系），数据缺失则如实说明。
|
||
8. 消息面：重要消息面必须基于 get_news 返回的快讯整理：挑5-8条对次日盘面影响最大的消息，每条格式为"【分类】一句话新闻 —— 一句影响解读"（分类用：宏观/政策/行业/公司/海外）；快讯中若没有某方面的重要消息，如实说明「今日无重要消息」，严禁编造工具中不存在的新闻。
|
||
9. 适当使用表格展示数据对比。
|
||
基于数据生成报告。报告共八章，固定顺序：一、市场总览｜二、题材热点分析｜三、核心股追踪｜四、资金与筹码｜五、重要消息面｜六、海外市场与国内期货｜七、下个交易日建议｜八、风险提示。
|
||
同一段数据只允许在它归属的那一章出现：指数与量能在第一章、题材涨幅在第二章、连板与核心股在第三章、资金流与两融在第四章，其他章节只做结论引用，不重复铺数据。
|
||
系统会把八章拆成三次生成，每次只写被指定的部分。
|
||
"""
|
||
|
||
# 分段生成指令：网关对单次 LLM 请求有约120s硬超时，整篇报告一次生成必被掐断，
|
||
# 故拆为三段（每段约1200-1600字），各自独立调用后拼接
|
||
REPORT_PARTS = [
|
||
"""现在写报告的【第1部分】，只输出这一部分，直接输出 Markdown，不要任何开场白或说明：
|
||
1. 以 `# {title} A股收盘分析报告` 一级标题开头
|
||
2. `## 一、市场总览`（指数表格、涨跌统计须环比、市场温度）
|
||
3. `## 二、题材热点分析`（涨幅前5、持续活跃、新兴热点、退潮警示、板块主力资金流TOP3）
|
||
全文控制在1600字以内。""",
|
||
"""现在写报告的【第2部分】，只输出这一部分，直接输出 Markdown，不要重复之前内容：
|
||
- `## 三、核心股追踪`（连板梯队完整表格：层级/股票/涨停原因/封单，首板挑3-5只人气股点评；连续上榜股票表格：股票/连续天数/近3日涨幅，点评持续活跃的核心标的；核心股表现；龙头辨识）
|
||
- `## 四、资金与筹码`：sectorFundFlow 的行业净流入 TOP3 / 净流出 TOP3 / 概念净流入 TOP3（亿元，带符号）、两融余额与变化、事件情报中值得注意的筹码信号
|
||
全文控制在 1500 字以内。""",
|
||
"""前两部分已在上下文中，现在续写【第3部分】，直接输出 Markdown，不要复述前文、不要写过渡句：
|
||
- `## 五、重要消息面`：5-8条，格式【分类】新闻——影响解读
|
||
- `## 六、海外市场与国内期货`（点评对次日A股的影响）
|
||
全文控制在1300字以内。""",
|
||
"""现在写报告的【第3部分】，只输出这一部分，直接输出 Markdown，不要重复之前内容：
|
||
- `## 七、下个交易日建议`（大盘预判、题材方向、核心股、仓位策略、潜在交易机会、规避方向，结合市场情绪、资金流向、热点题材、核心股表现、消息面、海外市场与国内期货、风险提示，给出明确的结论和建议。)
|
||
- `## 八、风险提示`
|
||
全文控制在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
|
||
tool_rounds = 0
|
||
|
||
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"):
|
||
tool_rounds += 1
|
||
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
|
||
})
|
||
# 核心工具齐了就立即进分段写作（再问一轮只会空转120s）；
|
||
# 不齐则最多再补一轮，否则对应章节会无数据可写
|
||
if all(t in tools_used for t in REQUIRED_TOOLS):
|
||
break
|
||
if tool_rounds >= 2:
|
||
print(f"[ai-service] 核心工具仍不齐（已取：{sorted(set(tools_used))}），直接进入写作")
|
||
break
|
||
print(f"[ai-service] 核心工具不全，补一轮取数（{tool_rounds}/2）")
|
||
missing = [t for t in REQUIRED_TOOLS if t not in tools_used]
|
||
messages.append({
|
||
"role": "user",
|
||
"content": f"还缺以下数据，请立即调用后停止取数：{', '.join(missing)}"
|
||
f"（get_theme_history 的 date 参数为 {trade_date}）"
|
||
})
|
||
continue
|
||
|
||
# 非工具轮（模型直接开写/空返回）：只要有工具结果就直接进入分段写作；
|
||
# 一轮工具都没拿到则重试
|
||
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"
|
||
else:
|
||
# 回灌前文：后段看不到前段内容会导致重复铺数据、数字口径打架
|
||
messages.append({"role": "assistant", "content": part_content})
|
||
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:
|
||
"""摘要：去掉标题行与 Markdown 标记后截断 200 字（供管理列表展示）
|
||
|
||
正文首行是 `# XXXX A股收盘分析报告`，直接截断会让它占掉摘要大半。
|
||
"""
|
||
text = re.sub(r"^#.*$", "", content or "", flags=re.MULTILINE)
|
||
text = re.sub(r"[>#*`|]", "", text)
|
||
return re.sub(r"\s+", " ", text).strip()[:200]
|
||
|
||
|
||
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()
|