mirrors2/auv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: AI市场分析功能

Browse Source 
- 后端：AI分析服务、工具调用、定时任务
- 前端：报告列表、详情页、Mermaid图表支持
- 支持OpenAI API兼容模型
- 收盘后自动分析生成报告
This commit is contained in:
Sakurasan
2026-09-01 04:21:29 +08:00
parent d7d019c2c4
commit 912a224b6b
17 changed files with 2525 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/database.py
+14
View File
@@ -71,6 +71,20 @@ CREATE TABLE IF NOT EXISTS daily_top_themes (
created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
UNIQUE(trade_date, theme_code)
);
CREATE TABLE IF NOT EXISTS ai_reports (
id INTEGER PRIMARY KEY AUTOINCREMENT,
trade_date TEXT NOT NULL,
report_type TEXT NOT NULL DEFAULT 'daily',
title TEXT NOT NULL,
content TEXT NOT NULL,
summary TEXT,
tools_used TEXT,
model TEXT NOT NULL,
tokens_used INTEGER,
created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
UNIQUE(trade_date, report_type)
);
"""
backend/main.py
+6 -3
View File
@@ -7,8 +7,9 @@ from contextlib import asynccontextmanager
from dotenv import load_dotenv
from database import init_db
from routes import stock, collections, shares, themes, core_stocks, fuyao, market_dashboard
from routes import stock, collections, shares, themes, core_stocks, fuyao, market_dashboard, ai_analysis
from services.daily_collector import collector_loop, cache_cleanup_loop
from services.ai_collector import ai_analysis_loop
load_dotenv()
@@ -18,12 +19,13 @@ async def lifespan(app: FastAPI):
init_db()
collector_task = asyncio.create_task(collector_loop())
cache_cleanup_task = asyncio.create_task(cache_cleanup_loop())
ai_analysis_task = asyncio.create_task(ai_analysis_loop())
try:
yield
finally:
for t in (collector_task, cache_cleanup_task):
for t in (collector_task, cache_cleanup_task, ai_analysis_task):
t.cancel()
for t in (collector_task, cache_cleanup_task):
for t in (collector_task, cache_cleanup_task, ai_analysis_task):
try:
await t
except asyncio.CancelledError:
@@ -47,6 +49,7 @@ app.include_router(themes.router, prefix="/api/themes")
app.include_router(core_stocks.router, prefix="/api/core-stocks")
app.include_router(fuyao.router, prefix="/api/v2")
app.include_router(market_dashboard.router, prefix="/api")
app.include_router(ai_analysis.router, prefix="/api")
# 生产模式：后端同时托管前端静态文件
# catch-all 路由在 API 路由之后注册，所以 API 优先级更高
backend/routes/ai_analysis.py
+127
View File
@@ -0,0 +1,127 @@
"""AI 分析报告路由（/api/ai-analysis）
提供：
- 报告列表
- 报告详情
- 手动触发分析（12小时频率限制）
- 重新生成报告
- 检查某日是否有报告
"""
import json
from datetime import datetime, timezone, timedelta
from fastapi import APIRouter, HTTPException
from fastapi.responses import JSONResponse
from database import get_connection, dict_from_row
_CST = timezone(timedelta(hours=8))
_NO_CACHE_HEADERS = {"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0"}
router = APIRouter()
def _has_ai_report(trade_date: str) -> bool:
conn = get_connection()
try:
row = conn.execute(
"SELECT 1 FROM ai_reports WHERE trade_date = ? AND report_type = 'daily' LIMIT 1",
(trade_date,)
).fetchone()
return row is not None
finally:
conn.close()
def _can_trigger(trade_date: str) -> tuple[bool, str]:
"""检查是否可以触发分析（12小时频率限制）"""
conn = get_connection()
try:
row = conn.execute(
"SELECT created_at FROM ai_reports WHERE trade_date = ? ORDER BY id DESC LIMIT 1",
(trade_date,)
).fetchone()
if row:
last_time = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S")
now = datetime.now(_CST)
if (now - last_time).total_seconds() < 12 * 3600:
return False, "12小时内已触发过，请稍后再试"
return True, ""
finally:
conn.close()
@router.get("/ai-analysis", summary="AI 分析报告列表")
async def list_reports():
conn = get_connection()
try:
rows = conn.execute(
"SELECT id, trade_date, report_type, title, summary, tools_used, model, tokens_used, created_at "
"FROM ai_reports ORDER BY trade_date DESC, id DESC LIMIT 50"
).fetchall()
items = []
for r in rows:
item = dict_from_row(r)
item["toolsUsed"] = json.loads(item.pop("tools_used") or "[]")
items.append(item)
return JSONResponse({"data": items}, headers=_NO_CACHE_HEADERS)
finally:
conn.close()
@router.get("/ai-analysis/{report_id}", summary="AI 分析报告详情")
async def get_report(report_id: int):
conn = get_connection()
try:
row = conn.execute("SELECT * FROM ai_reports WHERE id = ?", (report_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="报告不存在")
item = dict_from_row(row)
item["toolsUsed"] = json.loads(item.pop("tools_used") or "[]")
return JSONResponse({"data": item}, headers=_NO_CACHE_HEADERS)
finally:
conn.close()
@router.get("/ai-analysis/check/{trade_date}", summary="检查某日是否有 AI 分析报告")
async def check_report(trade_date: str):
has = _has_ai_report(trade_date)
return JSONResponse({"data": {"hasReport": has, "tradeDate": trade_date}})
@router.post("/ai-analysis/trigger", summary="手动触发今日 AI 分析")
async def trigger_analysis():
now = datetime.now(_CST)
trade_date = now.strftime("%Y-%m-%d")
if _has_ai_report(trade_date):
raise HTTPException(status_code=400, detail="今日已有分析报告")
can, msg = _can_trigger(trade_date)
if not can:
raise HTTPException(status_code=429, detail=msg)
from services.ai_service import collect_ai_analysis
result = await collect_ai_analysis(trade_date)
return JSONResponse({"data": result})
@router.post("/ai-analysis/{report_id}/regenerate", summary="重新生成 AI 分析报告")
async def regenerate_report(report_id: int):
conn = get_connection()
try:
row = conn.execute("SELECT trade_date FROM ai_reports WHERE id = ?", (report_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="报告不存在")
trade_date = row["trade_date"]
finally:
conn.close()
can, msg = _can_trigger(trade_date)
if not can:
raise HTTPException(status_code=429, detail=msg)
from services.ai_service import collect_ai_analysis
result = await collect_ai_analysis(trade_date)
return JSONResponse({"data": result})
backend/services/ai_collector.py
+56
View File
@@ -0,0 +1,56 @@
"""AI 分析定时任务
收盘后自动触发 AI 分析：
- 15:01 collector_loop 采集核心股+题材数据
- 15:10 ai_analysis_loop 触发 AI 分析（等数据采集完成）
"""
import asyncio
import traceback
from datetime import datetime, time as dtime, timezone, timedelta
from typing import Optional
from database import get_connection
_CST = timezone(timedelta(hours=8))
CHECK_INTERVAL_SECONDS = 300 # 每5分钟检查
AI_ANALYSIS_TIME = dtime(15, 10) # 15:10 后触发
def _is_trading_day(d: datetime) -> bool:
"""仅按工作日判断"""
return d.weekday() < 5
def _has_ai_report(trade_date: str) -> bool:
"""检查某日是否已有 AI 分析报告"""
conn = get_connection()
try:
row = conn.execute(
"SELECT 1 FROM ai_reports WHERE trade_date = ? AND report_type = 'daily' LIMIT 1",
(trade_date,)
).fetchone()
return row is not None
finally:
conn.close()
async def ai_analysis_loop(stop: Optional[asyncio.Event] = None) -> None:
"""后台循环：每个交易日 15:10 后自动触发 AI 分析（幂等）"""
while True:
try:
now = datetime.now(_CST)
if _is_trading_day(now) and now.time() >= AI_ANALYSIS_TIME:
trade_date = now.strftime("%Y-%m-%d")
if not _has_ai_report(trade_date):
print(f"[ai-collector] 开始 AI 分析 {trade_date}")
from services.ai_service import collect_ai_analysis
result = await collect_ai_analysis(trade_date)
print(f"[ai-collector] AI 分析完成 {trade_date}，token 消耗: {result.get('tokens_used', 0)}")
except Exception:
print("[ai-collector] AI 分析异常：")
traceback.print_exc()
if stop is not None and stop.is_set():
break
await asyncio.sleep(CHECK_INTERVAL_SECONDS)
backend/services/ai_config.py
+12
View File
@@ -0,0 +1,12 @@
"""AI 分析配置管理（从环境变量读取）"""
import os
from dotenv import load_dotenv
load_dotenv()
AI_API_BASE = os.getenv("AI_API_BASE", "https://openteam.oneisall.xyz/v1")
AI_API_KEY = os.getenv("AI_API_KEY", "")
AI_MODEL = os.getenv("AI_MODEL", "deepseek-v4-flash")
AI_MAX_TOKENS = int(os.getenv("AI_MAX_TOKENS", "4096"))
AI_TEMPERATURE = float(os.getenv("AI_TEMPERATURE", "0.7"))
backend/services/ai_service.py
+187
View File
@@ -0,0 +1,187 @@
"""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股市场进行收盘分析，生成一份完整的分析报告。
请先调用以下工具获取数据：
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}
"""
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": DAILY_ANALYSIS_PROMPT.format(trade_date=trade_date)}
]
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()
backend/services/ai_tools.py
+234
View File
@@ -0,0 +1,234 @@
"""Function Calling 工具定义和执行器
让 AI 可以主动调用现有 API 获取市场数据：
- get_market_dashboard: 市场看板
- get_theme_history: 题材热点历史
- get_active_core_stocks: 核心股追踪
- get_stock_quote: 个股实时行情
- get_fund_flow: 资金流向
"""
import json
from services.ai_config import AI_API_BASE, AI_API_KEY, AI_MODEL, AI_MAX_TOKENS, AI_TEMPERATURE
TOOLS = [
{
"type": "function",
"function": {
"name": "get_market_dashboard",
"description": "获取A股市场看板数据，包含主要指数行情、涨跌统计、行业强度、概念热度、事件情报、市场温度评分",
"parameters": {"type": "object", "properties": {}, "required": []}
}
},
{
"type": "function",
"function": {
"name": "get_theme_history",
"description": "获取指定交易日的题材涨幅排行前20",
"parameters": {
"type": "object",
"properties": {
"date": {"type": "string", "description": "交易日 YYYY-MM-DD"}
},
"required": ["date"]
}
}
},
{
"type": "function",
"function": {
"name": "get_active_core_stocks",
"description": "获取活跃核心股列表，包含最近10日涨幅矩阵和所属题材",
"parameters": {"type": "object", "properties": {}, "required": []}
}
},
{
"type": "function",
"function": {
"name": "get_stock_quote",
"description": "获取单只股票实时行情（价格、涨跌幅、成交量、换手率等）",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "6位股票代码，如 600519"}
},
"required": ["code"]
}
}
},
{
"type": "function",
"function": {
"name": "get_fund_flow",
"description": "获取个股资金流向数据（主力净流入、超大单/大单/中单/小单流入流出）",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "6位股票代码"},
"name": {"type": "string", "description": "股票名称"},
"days": {"type": "integer", "description": "获取天数，默认30"}
},
"required": ["code", "name"]
}
}
}
]
async def execute_tool(tool_name: str, arguments: dict) -> str:
"""执行工具调用，返回 JSON 字符串"""
try:
if tool_name == "get_market_dashboard":
return await _get_market_dashboard()
elif tool_name == "get_theme_history":
return await _get_theme_history(arguments.get("date", ""))
elif tool_name == "get_active_core_stocks":
return await _get_active_core_stocks()
elif tool_name == "get_stock_quote":
return await _get_stock_quote(arguments.get("code", ""))
elif tool_name == "get_fund_flow":
return await _get_fund_flow(
arguments.get("code", ""),
arguments.get("name", ""),
arguments.get("days", 30)
)
else:
return json.dumps({"error": f"未知工具: {tool_name}"})
except Exception as e:
return json.dumps({"error": str(e)})
async def _get_market_dashboard() -> str:
"""获取市场看板数据"""
from routes.market_dashboard import _build_dashboard
data = await _build_dashboard()
return json.dumps(data, ensure_ascii=False, default=str)
async def _get_theme_history(date: str) -> str:
"""获取指定交易日题材涨幅前20"""
from database import get_connection
conn = get_connection()
try:
rows = conn.execute(
"SELECT * FROM daily_top_themes WHERE trade_date = ? ORDER BY rank ASC",
(date,)
).fetchall()
items = [dict(r) for r in rows]
return json.dumps({"date": date, "items": items}, ensure_ascii=False)
finally:
conn.close()
async def _get_active_core_stocks() -> str:
"""获取活跃核心股列表"""
from database import get_connection
from datetime import date as date_cls
conn = get_connection()
try:
# 最近10个有数据的交易日
rows = conn.execute(
"SELECT DISTINCT trade_date FROM daily_core_stocks ORDER BY trade_date DESC LIMIT 10"
).fetchall()
dates = [r["trade_date"] for r in reversed(rows)]
if not dates:
return json.dumps({"dates": [], "stocks": []}, ensure_ascii=False)
placeholders = ",".join("?" * len(dates))
rows = conn.execute(
f"""SELECT trade_date, stock_code, stock_name, f3, cover_count FROM daily_core_stocks
WHERE trade_date IN ({placeholders})
ORDER BY trade_date DESC, rank ASC""",
dates,
).fetchall()
stock_days = {}
for r in rows:
code = r["stock_code"]
s = stock_days.setdefault(code, {
"stockCode": code,
"stockName": r["stock_name"],
"coverCount": r["cover_count"],
"dailyGains": {},
"appearCount": 0,
"lastAppear": None,
})
s["dailyGains"][r["trade_date"]] = r["f3"]
s["appearCount"] += 1
if s["lastAppear"] is None or r["trade_date"] > s["lastAppear"]:
s["lastAppear"] = r["trade_date"]
stocks = list(stock_days.values())
stocks.sort(key=lambda x: x.get("lastAppear") or "", reverse=True)
stocks.sort(key=lambda x: -x["appearCount"])
themes_rows = conn.execute(
f"""SELECT stock_code, theme_code, theme_name FROM daily_core_stock_themes
WHERE trade_date IN ({placeholders})""",
dates,
).fetchall()
themes_by_stock = {}
for t in themes_rows:
per = themes_by_stock.setdefault(t["stock_code"], {})
per.setdefault(t["theme_code"], {"theme_code": t["theme_code"], "theme_name": t["theme_name"]})
for s in stocks:
s["themes"] = list(themes_by_stock.get(s["stockCode"], {}).values())
latest = dates[-1] if dates else None
for s in stocks:
if s.get("lastAppear") and latest:
try:
d1 = date_cls.fromisoformat(latest)
d2 = date_cls.fromisoformat(s["lastAppear"])
s["daysSinceLastAppear"] = (d1 - d2).days
except ValueError:
s["daysSinceLastAppear"] = 0
else:
s["daysSinceLastAppear"] = 0
return json.dumps({"dates": dates, "stocks": stocks}, ensure_ascii=False)
finally:
conn.close()
async def _get_stock_quote(code: str) -> str:
"""获取个股实时行情"""
from services.tencent import fetch_quote
data = await fetch_quote(code)
if not data:
return json.dumps({"error": f"未找到股票 {code} 的数据"})
return json.dumps(data, ensure_ascii=False)
async def _get_fund_flow(code: str, name: str, days: int) -> str:
"""获取个股资金流向"""
from services.tencent import fetch_quote, get_market_prefix
from database import get_connection
# 先尝试从本地数据库获取历史资金流向
conn = get_connection()
try:
rows = conn.execute(
"SELECT * FROM cache WHERE key LIKE ? AND expires_at > datetime('now','localtime')",
(f"fund_flow:{code}%",)
).fetchall()
if rows:
return rows[0]["value"]
finally:
conn.close()
# 降级：通过腾讯获取基础行情数据
quote = await fetch_quote(code)
if not quote:
return json.dumps({"error": f"未找到股票 {code} 的数据"})
return json.dumps({
"code": code,
"name": name or quote.get("name", ""),
"currentPrice": quote.get("currentPrice", 0),
"changePercent": quote.get("changePercent", 0),
"volume": quote.get("volume", 0),
"amount": quote.get("amount", 0),
"note": "资金流向详细数据需通过东方财富API获取"
}, ensure_ascii=False)