From 9d1e1c9af5c36945f164fa9a1d64534000e0027c Mon Sep 17 00:00:00 2001 From: Sakurasan <26715255+Sakurasan@users.noreply.github.com> Date: Thu, 17 Sep 2026 23:07:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20AI=E6=97=A5=E6=8A=A5=E6=A0=B8=E5=BF=83?= =?UTF-8?q?=E8=82=A1=E8=BF=BD=E8=B8=AA=E7=AB=A0=E8=8A=82=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=E8=BF=9E=E7=BB=AD=E4=B8=8A=E6=A6=9C=E8=82=A1=E7=A5=A8=EF=BC=88?= =?UTF-8?q?=E8=BF=913=E6=97=A5+=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/routes/core_stocks.py | 71 +++++++++++++++++++++++++ backend/services/ai_service.py | 10 ++-- backend/services/ai_tools.py | 94 ++++++++++++++++++++++++++++++++++ 3 files changed, 171 insertions(+), 4 deletions(-) diff --git a/backend/routes/core_stocks.py b/backend/routes/core_stocks.py index 22d7a7e..e8f6bd7 100644 --- a/backend/routes/core_stocks.py +++ b/backend/routes/core_stocks.py @@ -90,6 +90,77 @@ async def active_core_stocks(days: int = Query(10, ge=1, le=30, description="交 conn.close() +@router.get("/consecutive", summary="连续上榜的核心股") +async def consecutive_core_stocks(min_days: int = Query(3, ge=2, le=10, description="最小连续天数")): + from datetime import date as date_cls + conn = get_connection() + try: + # 最近30个有数据的交易日 + rows = conn.execute( + "SELECT DISTINCT trade_date FROM daily_core_stocks ORDER BY trade_date DESC LIMIT 30" + ).fetchall() + all_dates = [r["trade_date"] for r in reversed(rows)] + + if not all_dates: + return JSONResponse({"stocks": []}, headers=_NO_CACHE_HEADERS) + + # 查询所有上榜记录 + placeholders = ",".join("?" * len(all_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 stock_code, trade_date""", + all_dates, + ).fetchall() + + # 按股票分组，记录上榜日期 + stock_dates: dict[str, dict] = {} + for r in rows: + code = r["stock_code"] + s = stock_dates.setdefault(code, { + "stockCode": code, + "stockName": r["stock_name"], + "dates": [], + "dailyGains": {}, + }) + s["dates"].append(r["trade_date"]) + s["dailyGains"][r["trade_date"]] = r["f3"] + + # 计算连续上榜天数（从最新交易日往回数） + results = [] + for code, s in stock_dates.items(): + dates_list = sorted(s["dates"]) + consecutive = 0 + last_n = [] + for i in range(len(dates_list) - 1, -1, -1): + if i == len(dates_list) - 1: + consecutive = 1 + last_n.append(dates_list[i]) + else: + prev_idx = all_dates.index(dates_list[i + 1]) if dates_list[i + 1] in all_dates else -1 + curr_idx = all_dates.index(dates_list[i]) if dates_list[i] in all_dates else -1 + if prev_idx - curr_idx <= 2: + consecutive += 1 + last_n.append(dates_list[i]) + else: + break + + if consecutive >= min_days: + results.append({ + "stockCode": code, + "stockName": s["stockName"], + "consecutiveDays": consecutive, + "recentDates": last_n, + "dailyGains": {d: s["dailyGains"].get(d) for d in last_n}, + }) + + results.sort(key=lambda x: -x["consecutiveDays"]) + return JSONResponse({"stocks": results}, headers=_NO_CACHE_HEADERS) + finally: + conn.close() + + @router.get("/history", summary="指定交易日核心股(含所属题材)") async def core_stock_history(date: str = Query(..., description="交易日 YYYY-MM-DD")): conn = get_connection() diff --git a/backend/services/ai_service.py b/backend/services/ai_service.py index 0b0643e..23c558a 100644 --- a/backend/services/ai_service.py +++ b/backend/services/ai_service.py @@ -33,6 +33,7 @@ SYSTEM_PROMPT = """你是一位专业的A股市场分析师，擅长从数据中 - get_market_dashboard: 获取市场整体数据（指数/涨跌统计/市场温度/连板梯队/行业强度/板块资金流/两融/海外指数与国内期货/事件情报） - get_theme_history: 获取指定日期的题材涨幅排行 - get_active_core_stocks: 获取核心股追踪数据（10日涨幅矩阵+所属题材） +- get_consecutive_core_stocks: 获取连续上榜的核心股列表（近3日及以上连续上榜），用于识别持续活跃的热点股 - get_stock_quote: 获取个股实时行情 - get_fund_flow: 获取个股资金流向 - get_news: 获取财经快讯（新浪7x24，用于重要消息面） @@ -56,7 +57,7 @@ SYSTEM_PROMPT = """你是一位专业的A股市场分析师，擅长从数据中 # 一次工具轮就必须拿全的核心工具：模型常只调 1-2 个就开写，缺失会让对应章节数据空洞 REQUIRED_TOOLS = ("get_market_dashboard", "get_theme_history", - "get_active_core_stocks", "get_news") + "get_active_core_stocks", "get_consecutive_core_stocks", "get_news") DAILY_ANALYSIS_PROMPT = """请对 {trade_date}（A股交易日） 的A股市场进行收盘分析，生成一份完整的分析报告。 @@ -76,11 +77,12 @@ DAILY_ANALYSIS_PROMPT = """请对 {trade_date}（A股交易日） 的A股市场 4. 分析完成后给出明确的结论和建议 ■ 第一步 · 取数（必须一次并行完成） -在同一次回复中并行发起以下 4 个工具调用，不要拆成多轮、不要只调其中一部分： +在同一次回复中并行发起以下 5 个工具调用，不要拆成多轮、不要只调其中一部分： 1. get_market_dashboard() —— 指数、涨跌统计、市场温度、连板梯队 limitLadder、板块资金流 sectorFundFlow、两融、海外指数与国内期货 globalMarkets 2. get_theme_history(date="{trade_date}") —— 今日题材涨幅榜 3. get_active_core_stocks() —— 核心股 10 日涨幅矩阵与所属题材 -4. get_news(limit=30) —— 今日财经快讯 +4. get_consecutive_core_stocks(min_days=3) —— 连续3日及以上上榜的核心股 +5. get_news(limit=30) —— 今日财经快讯 只有在需要核实某只具体个股时，才额外调用 get_stock_quote / get_fund_flow，合计不超过 2 次。 ■ 第二步 · 写作（以下规约对每一部分都生效） @@ -109,7 +111,7 @@ REPORT_PARTS = [ 3. `## 二、题材热点分析`（涨幅前5、持续活跃、新兴热点、退潮警示、板块主力资金流TOP3） 全文控制在1600字以内。""", """现在写报告的【第2部分】，只输出这一部分，直接输出 Markdown，不要重复之前内容： -- `## 三、核心股追踪`（连板梯队完整表格：层级/股票/涨停原因/封单，首板挑3-5只人气股点评；核心股表现；龙头辨识） +- `## 三、核心股追踪`（连板梯队完整表格：层级/股票/涨停原因/封单，首板挑3-5只人气股点评；连续上榜股票表格：股票/连续天数/近3日涨幅，点评持续活跃的核心标的；核心股表现；龙头辨识） - `## 四、资金与筹码`：sectorFundFlow 的行业净流入 TOP3 / 净流出 TOP3 / 概念净流入 TOP3（亿元，带符号）、两融余额与变化、事件情报中值得注意的筹码信号 全文控制在 1500 字以内。""", """前两部分已在上下文中，现在续写【第3部分】，直接输出 Markdown，不要复述前文、不要写过渡句： diff --git a/backend/services/ai_tools.py b/backend/services/ai_tools.py index 9546807..ffbff9d 100644 --- a/backend/services/ai_tools.py +++ b/backend/services/ai_tools.py @@ -42,6 +42,20 @@ TOOLS = [ "parameters": {"type": "object", "properties": {}, "required": []} } }, + { + "type": "function", + "function": { + "name": "get_consecutive_core_stocks", + "description": "获取连续上榜的核心股列表（近3日及以上连续上榜），用于识别持续活跃的热点股", + "parameters": { + "type": "object", + "properties": { + "min_days": {"type": "integer", "description": "最小连续天数，默认3"} + }, + "required": [] + } + } + }, { "type": "function", "function": { @@ -98,6 +112,8 @@ async def execute_tool(tool_name: str, arguments: dict) -> str: 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_consecutive_core_stocks": + return await _get_consecutive_core_stocks(arguments.get("min_days", 3)) elif tool_name == "get_stock_quote": return await _get_stock_quote(arguments.get("code", "")) elif tool_name == "get_fund_flow": @@ -219,6 +235,84 @@ async def _get_active_core_stocks() -> str: conn.close() +async def _get_consecutive_core_stocks(min_days: int = 3) -> str: + """获取连续上榜的核心股列表""" + from database import get_connection + conn = get_connection() + try: + # 最近30个有数据的交易日 + rows = conn.execute( + "SELECT DISTINCT trade_date FROM daily_core_stocks ORDER BY trade_date DESC LIMIT 30" + ).fetchall() + all_dates = [r["trade_date"] for r in reversed(rows)] + + if not all_dates: + return json.dumps({"dates": [], "stocks": []}, ensure_ascii=False) + + # 查询所有上榜记录 + placeholders = ",".join("?" * len(all_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 stock_code, trade_date""", + all_dates, + ).fetchall() + + # 按股票分组，记录上榜日期 + stock_dates: dict[str, dict] = {} + for r in rows: + code = r["stock_code"] + s = stock_dates.setdefault(code, { + "stockCode": code, + "stockName": r["stock_name"], + "dates": [], + "dailyGains": {}, + }) + s["dates"].append(r["trade_date"]) + s["dailyGains"][r["trade_date"]] = r["f3"] + + # 计算连续上榜天数（从最新交易日往回数） + results = [] + for code, s in stock_dates.items(): + dates_list = sorted(s["dates"]) + # 从最新的日期往回数连续天数 + consecutive = 0 + last_n = [] + for i in range(len(dates_list) - 1, -1, -1): + if i == len(dates_list) - 1: + consecutive = 1 + last_n.append(dates_list[i]) + else: + # 检查是否连续（允许间隔1天非交易日） + prev_idx = all_dates.index(dates_list[i + 1]) if dates_list[i + 1] in all_dates else -1 + curr_idx = all_dates.index(dates_list[i]) if dates_list[i] in all_dates else -1 + if prev_idx - curr_idx <= 2: # 允许间隔1天 + consecutive += 1 + last_n.append(dates_list[i]) + else: + break + + if consecutive >= min_days: + results.append({ + "stockCode": code, + "stockName": s["stockName"], + "consecutiveDays": consecutive, + "recentDates": last_n, + "dailyGains": {d: s["dailyGains"].get(d) for d in last_n}, + }) + + # 按连续天数降序排列 + results.sort(key=lambda x: -x["consecutiveDays"]) + + return json.dumps({ + "minDays": min_days, + "stocks": results[:30], + }, ensure_ascii=False) + finally: + conn.close() + + async def _get_stock_quote(code: str) -> str: """获取个股实时行情""" from services.tencent import fetch_quote