feat: AI分析第二档——get_news消息面、板块主力资金流TOPS、两融余额、截断续写机制

AGENTS.md

@@ -78,3 +78,9 @@
 - 连板梯队：`get_market_dashboard` 返回 `limitLadder` 字段（完整版，2连板以上全量+首板前8），涨停原因/封单金额来自涨停池按 thscode 匹配；看板 events 里的天梯仍只取每层前2（保持 UI 精简），两处用途不同不要合并
 - tokens 口径：`ai_reports.tokens_used` 只记"当次生成"消耗（约7万/份）；同日重新生成走 UPSERT，`generation_count` 自增、`updated_at` 刷新、`created_at` 保留首次生成时间；表有 UNIQUE(trade_date, report_type)，禁止改回 INSERT OR REPLACE 之外还要注意别用 lastrowid（UPSERT 更新时不可靠，须回查 id）
 - 旧库补列用 init_db 里的 try/except ALTER TABLE 轻量迁移（CREATE TABLE IF NOT EXISTS 不会更新旧表结构）
+
+- 补充数据源（services/market_extra.py）：①财经快讯 get_news=新浪7x24 zhibo.sina.cn（feed.list.rich_text）；②板块主力资金流=东财 push2 的 clist 接口（f62 主力净额），**必须用 push2delay.eastmoney.com 镜像**——push2 对部分客户端 TLS 指纹拦截（peer closed），查询串保持字面量 `+` 号；③两融=datacenter-web 的 RPTA_RZRQ_LSHJ（T+1 披露），两融余额=RZYE+RQYE
+- AI 报告重要消息面规则：必须基于 get_news 快讯，5-8条，格式【宏观/政策/行业/公司/海外】新闻——影响解读，禁止编造；板块资金面规则：必须引用 sectorFundFlow 的行业净流入/流出TOP3+概念TOP3
+- 截断续写：报告长导致 finish_reason=length 时，拼接已有内容并向 messages 追加"继续"指令让模型续写（最多3次），truncated 仅在续写后仍截断时为 true；call_llm 读超时 300s（续写携带全部上下文）
+- 快照入库防护：_capture_market_snapshot 校验 indices 非空且涨跌统计不全 0，fuyao 失败时跳过入库，防止空快照污染环比
+- 调试注意：独立脚本跑 backend 代码必须显式 `load_dotenv("/path/to/repo/.env")`——fuyao_apikey 在仓库根目录 .env（uvicorn 靠 --env-file 参数加载），backend/.env 只有 MX keys；且 stdin 脚本里 load_dotenv() 无参调用会因 frame 断言报错，须显式传路径

backend/routes/market_dashboard.py

@@ -17,7 +17,7 @@ from datetime import datetime, timezone, timedelta
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import JSONResponse
 
-from services import fuyao_client
+from services import fuyao_client, market_extra
 
 router = APIRouter()
 
@@ -163,12 +163,16 @@ async def _build_dashboard() -> dict:
             anomaly_data,
             limit_ladder_data,
             skyrocket_data,
+            sector_flow_data,
+            margin_data,
         ) = await asyncio.gather(
             fuyao_client.hot_stock_list("day"),
             fuyao_client.dragon_tiger_list("all"),
             fuyao_client.anomaly_analysis_list(["SHARP_RISE", "RAPID_RALLY", "LIMIT_UP"]),
             fuyao_client.limit_up_ladder(),
             fuyao_client.skyrocket_list("day"),
+            market_extra.fetch_sector_fund_flow(),
+            market_extra.fetch_margin_summary(),
             return_exceptions=True,
         )
     except Exception:
@@ -177,6 +181,8 @@ async def _build_dashboard() -> dict:
         anomaly_data = {}
         limit_ladder_data = {}
         skyrocket_data = {}
+        sector_flow_data = None
+        margin_data = None
 
     # ── 解析指数 ──
     indices = []
@@ -291,6 +297,12 @@ async def _build_dashboard() -> dict:
         "auctionSignal": auction_signal,
     }
 
+    # 两融余额（交易所 T+1 披露），供看板展示、AI 分析与次日环比快照使用
+    if isinstance(margin_data, dict):
+        market_stats["marginBalanceYi"] = margin_data.get("balanceYi")
+        market_stats["marginChangeYi"] = margin_data.get("changeYi")
+        market_stats["marginDate"] = margin_data.get("date")
+
     # ── 行业强度榜（使用行业指数真实涨幅） ──
     sector_strength = []
     industries = industry_catalog if isinstance(industry_catalog, list) else []
@@ -552,6 +564,7 @@ async def _build_dashboard() -> dict:
         "marketStats": market_stats,
         "sectorStrength": sector_strength[:31],
         "conceptStrength": concept_strength[:10],
+        "sectorFundFlow": sector_flow_data if isinstance(sector_flow_data, dict) else None,
         "events": events,
         "limitLadder": limit_ladder,
         "updateTime": datetime.now(BJT).strftime("%Y-%m-%d %H:%M:%S"),

backend/services/ai_service.py

@@ -29,11 +29,12 @@ SYSTEM_PROMPT = """你是一位专业的A股市场分析师，擅长从数据中
 - 风险提示，每次推荐都需说明风险点
 
 可用工具：
-- get_market_dashboard: 获取市场整体数据（指数/涨跌统计/市场温度/连板梯队/行业强度/事件情报）
+- get_market_dashboard: 获取市场整体数据（指数/涨跌统计/市场温度/连板梯队/行业强度/板块资金流/两融/事件情报）
 - get_theme_history: 获取指定日期的题材涨幅排行
 - get_active_core_stocks: 获取核心股追踪数据（10日涨幅矩阵+所属题材）
 - get_stock_quote: 获取个股实时行情
 - get_fund_flow: 获取个股资金流向
+- get_news: 获取财经快讯（新浪7x24，用于重要消息面）
 
 重要规则：
 1. 你必须先调用工具获取数据，然后基于数据进行分析
@@ -47,20 +48,25 @@ DAILY_ANALYSIS_PROMPT = """请对 {trade_date} 的A股市场进行收盘分析
 {prev_snapshot_section}
 
 请先调用以下工具获取数据：
-1. get_market_dashboard - 获取市场整体数据（含涨跌统计、市场温度、连板梯队 limitLadder）
+1. get_market_dashboard - 获取市场整体数据（含涨跌统计、市场温度、连板梯队 limitLadder、板块资金流 sectorFundFlow、两融）
 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亿"）；没有昨日快照则如实说明"暂无昨日数据"。
+2. 量能与情绪类数字必须给环比：若上方提供了"前一交易日盘面数据快照"，成交额、涨跌家数、涨停数、两融余额、市场温度等在与昨日对比后表述（如"成交额2.05万亿，较昨日缩量约700亿"）；没有昨日快照则如实说明"暂无昨日数据"。
 
 3. 连板梯队必须完整呈现 get_market_dashboard 返回的 limitLadder：从最高连板到2连板逐级列表格，每只标注涨停原因（reason字段）与封单金额（sealWan，单位万，为空则不写）；首板只挑3-5只人气最高的点评。
 
-4. 适当使用表格展示数据对比。
+4. 板块资金面必须引用 get_market_dashboard 返回的 sectorFundFlow：行业主力净流入TOP3、净流出TOP3、概念净流入TOP3（单位亿元），结合题材分析说明资金动向。
+
+5. 重要消息面必须基于 get_news 返回的快讯整理：挑5-8条对次日盘面影响最大的消息，每条格式为"【分类】一句话新闻 —— 一句影响解读"（分类用：宏观/政策/行业/公司/海外）；快讯中若没有某方面的重要消息，如实说明，严禁编造工具中不存在的新闻。
+
+6. 适当使用表格展示数据对比。
 
 然后基于数据生成报告，结构如下：
 
@@ -74,6 +80,7 @@ DAILY_ANALYSIS_PROMPT = """请对 {trade_date} 的A股市场进行收盘分析
 - 持续活跃的题材
 - 新兴热点题材
 - 明显退潮的题材（警示）
+- 板块主力资金流（按格式规则4引用数据）
 
 ## 三、核心股追踪
 - 连板梯队分析（按格式规则3完整呈现）
@@ -91,7 +98,10 @@ DAILY_ANALYSIS_PROMPT = """请对 {trade_date} 的A股市场进行收盘分析
 - 操作策略（仓位建议、买卖时机）
 - 需要规避的方向
 
-## 六、风险提示
+## 六、重要消息面
+- 基于 get_news 快讯整理（按格式规则5）
+
+## 七、风险提示
 - 需要警惕的风险因素
 - 操作建议
 
@@ -123,7 +133,7 @@ async def call_llm(messages: list, tools: list = None) -> dict:
                 "Content-Type": "application/json",
             },
             json=payload,
-            timeout=120,
+            timeout=300,  # 续写调用携带全部上下文且输出很长，需要较宽的读超时
         )
         resp.raise_for_status()
         return resp.json()
@@ -156,6 +166,7 @@ async def collect_ai_analysis(trade_date: str) -> dict:
     total_tokens = 0
     final_content = ""
     was_truncated = False
+    continuations = 0
     max_rounds = 30  # 安全上限，正常分析约 3-8 轮
 
     for i in range(max_rounds):
@@ -167,20 +178,27 @@ async def collect_ai_analysis(trade_date: str) -> dict:
         messages.append(message)
 
         finish_reason = choice.get("finish_reason", "")
+        print(f"[ai-service] 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 == "stop":
-            final_content = message.get("content") or ""
+            final_content += message.get("content") or ""
             break
 
         if finish_reason == "length":
+            # 单次输出上限截断：拼接已有内容并让模型续写（最多3次）
+            final_content += message.get("content") or ""
+            if continuations < 3:
+                continuations += 1
+                print(f"[ai-service] 响应被截断，第 {continuations} 次续写")
+                messages.append({
+                    "role": "user",
+                    "content": "报告输出被截断了。请从截断处无缝续写剩余内容：直接接着写，不要重复已输出的部分，也不要重新输出标题。",
+                })
+                continue
             was_truncated = True
-            final_content = message.get("content") or ""
+            print("[ai-service] 警告：多次续写后仍被截断")
-            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":
@@ -235,8 +253,21 @@ def _fmt_amount(v) -> str:
 def _fmt_index(idx: dict) -> str:
     if not idx:
         return "-"
-    sign = "+" if idx.get("changePct", 0) >= 0 else ""
+    try:
-    return f"{idx.get('price', '-')}（{sign}{idx.get('changePct', 0)}%）"
+        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:
@@ -244,8 +275,13 @@ 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": data.get("marketStats", {})},
+            {"indices": data.get("indices", []), "marketStats": stats},
             ensure_ascii=False,
             default=str,
         )
@@ -286,12 +322,20 @@ def _get_prev_snapshot_section(trade_date: str) -> str:
             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"))}%
-- 市场温度：{_num(stats.get("temperature"))} 分，竞价信号：{stats.get("auctionSignal") or "-"}
+- 市场温度：{_fmt_temperature(stats.get("temperature"))}，竞价信号：{stats.get("auctionSignal") or "-"}{margin_line}
 - 指数收盘：{idx_line}
 
 ---

backend/services/ai_tools.py

@@ -71,6 +71,20 @@ TOOLS = [
                 "required": ["code", "name"]
             }
         }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "get_news",
+            "description": "获取最近的财经快讯（新浪7x24，含宏观/行业/公司/海外动态），用于重要消息面梳理",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "limit": {"type": "integer", "description": "获取条数，默认30，最大50"}
+                },
+                "required": []
+            }
+        }
     }
 ]
 
@@ -92,12 +106,23 @@ async def execute_tool(tool_name: str, arguments: dict) -> str:
                 arguments.get("name", ""),
                 arguments.get("days", 30)
             )
+        elif tool_name == "get_news":
+            return await _get_news(arguments.get("limit", 30))
         else:
             return json.dumps({"error": f"未知工具: {tool_name}"})
     except Exception as e:
         return json.dumps({"error": str(e)})
 
 
+async def _get_news(limit: int = 30) -> str:
+    """获取财经快讯"""
+    from services.market_extra import fetch_news
+    news = await fetch_news(limit)
+    if not news:
+        return json.dumps({"error": "快讯获取失败或暂无数据"}, ensure_ascii=False)
+    return json.dumps({"count": len(news), "items": news}, ensure_ascii=False)
+
+
 async def _get_market_dashboard() -> str:
     """获取市场看板数据"""
     from routes.market_dashboard import _build_dashboard

backend/services/market_extra.py

@@ -0,0 +1,143 @@
+"""市场补充数据源（供 AI 分析与看板扩展）
+
+- fetch_news: 新浪财经 7x24 快讯
+- fetch_sector_fund_flow: 东方财富板块主力资金流排行（行业/概念）
+- fetch_margin_summary: 东方财富两融余额汇总（T+1 数据）
+
+均为公开接口，失败时返回 []/None，不阻塞主流程。
+"""
+
+import asyncio
+
+import httpx
+
+_TIMEOUT = 10.0
+_UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
+       "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36")
+
+
+async def fetch_news(limit: int = 30) -> list[dict]:
+    """新浪财经 7x24 快讯，返回 [{time: 'MM-DD HH:MM', content}]；失败返回 []"""
+    try:
+        limit = max(1, min(int(limit or 30), 50))
+    except (TypeError, ValueError):
+        limit = 30
+    try:
+        async with httpx.AsyncClient(timeout=_TIMEOUT, headers={"User-Agent": _UA}) as client:
+            resp = await client.get(
+                "https://zhibo.sina.com.cn/api/zhibo/feed",
+                params={"page": 1, "page_size": limit, "zhibo_id": 152, "tag_id": 0},
+            )
+            resp.raise_for_status()
+            items = resp.json()["result"]["data"]["feed"]["list"]
+        news = []
+        for it in items or []:
+            text = (it.get("rich_text") or "").strip()
+            if not text:
+                continue
+            news.append({
+                "time": (it.get("create_time") or "")[5:16],  # 'MM-DD HH:MM'
+                "content": text[:300],
+            })
+        return news
+    except Exception:
+        return []
+
+
+_FLOW_HOSTS = (
+    # push2 对部分客户端有 TLS 指纹拦截（peer closed），delay 镜像同接口且稳定
+    "https://push2delay.eastmoney.com",
+    "https://push2.eastmoney.com",
+)
+
+
+async def _fetch_flow_boards(client: httpx.AsyncClient, fs: str, po: int, pz: int) -> list[dict]:
+    """拉取一类板块的主力净流入排行。po=1 降序（净流入最多），po=0 升序（净流出最多）
+
+    查询串保持字面量 + 号（与东财网页请求一致），逐 host 尝试。
+    """
+    qs = (f"/api/qt/clist/get?fid=f62&po={po}&pz={pz}&pn=1&np=1"
+          f"&fltt=2&invt=2&fs={fs}&fields=f12,f14,f62,f184")
+    last_err: Exception | None = None
+    for host in _FLOW_HOSTS:
+        try:
+            resp = await client.get(host + qs)
+            resp.raise_for_status()
+            diff = (resp.json().get("data") or {}).get("diff") or []
+            break
+        except Exception as e:
+            last_err = e
+            diff = []
+    else:
+        raise ConnectionError(f"板块资金流全部数据源失败: {last_err}")
+    if isinstance(diff, dict):  # 兼容旧版 {index: item} 结构
+        diff = list(diff.values())
+    boards = []
+    for d in diff:
+        amt = d.get("f62")
+        if not isinstance(amt, (int, float)):
+            continue
+        boards.append({
+            "code": d.get("f12", ""),
+            "name": d.get("f14", ""),
+            "mainNet": round(amt / 1e8, 1),  # 亿元
+            "mainNetPct": d.get("f184"),     # 主力净占比 %
+        })
+    return boards
+
+
+async def fetch_sector_fund_flow() -> dict | None:
+    """板块主力资金流排行：行业净流入/净流出 TOP6 + 概念净流入 TOP6；失败返回 None"""
+    try:
+        async with httpx.AsyncClient(timeout=_TIMEOUT, headers={"User-Agent": _UA}) as client:
+            industry_in, industry_out, concept_in = await asyncio.gather(
+                _fetch_flow_boards(client, "m:90+t:2", 1, 6),
+                _fetch_flow_boards(client, "m:90+t:2", 0, 6),
+                _fetch_flow_boards(client, "m:90+t:3", 1, 6),
+            )
+        return {
+            "industryInflow": industry_in,    # 主力净流入降序
+            "industryOutflow": industry_out,  # 升序（净流出最多在前）
+            "conceptInflow": concept_in,
+            "unit": "亿元",
+        }
+    except Exception:
+        return None
+
+
+async def fetch_margin_summary() -> dict | None:
+    """沪深北两融余额汇总（交易所 T+1 披露）；失败返回 None"""
+    try:
+        async with httpx.AsyncClient(timeout=_TIMEOUT, headers={"User-Agent": _UA}) as client:
+            resp = await client.get(
+                "https://datacenter-web.eastmoney.com/api/data/v1/get",
+                params={
+                    "reportName": "RPTA_RZRQ_LSHJ",
+                    "columns": "ALL",
+                    "source": "WEB",
+                    "sortColumns": "dim_date",
+                    "sortTypes": "-1",
+                    "pageSize": 2,
+                    "pageNumber": 1,
+                },
+            )
+            resp.raise_for_status()
+            rows = ((resp.json().get("result") or {}).get("data")) or []
+        if not rows:
+            return None
+
+        def _balance(row: dict) -> float:
+            return float(row.get("RZYE") or 0) + float(row.get("RQYE") or 0)
+
+        latest = rows[0]
+        prev = rows[1] if len(rows) > 1 else None
+        balance = _balance(latest)
+        change = (balance - _balance(prev)) if prev else None
+        return {
+            "date": (latest.get("DIM_DATE") or "")[:10],
+            "balanceYi": round(balance / 1e8),  # 亿元
+            "changeYi": round(change / 1e8) if change is not None else None,
+            "rzjmeYi": round(float(latest.get("RZJME") or 0) / 1e8, 1),  # 融资净买入
+        }
+    except Exception:
+        return None