Merge pull request #107

CHANGELOG.md

@@ -22,6 +22,11 @@ Breaking changes within the 0.x line are called out explicitly.
 
 ## [0.5.16] — 2026-09-02
 
+### 兼容性：`langgraph` 最低版本提升至 1.0
+
+缺失数据追踪在 `ToolNode` 上使用 `wrap_tool_call`，该参数从 LangGraph 1.0 起提供。
+因此项目依赖由 `langgraph>=0.4.8` 提升为 `langgraph>=1.0`；使用旧版 LangGraph 的用户需要先升级依赖。
+
 ### 修复：只传部分 `config` 时 `TradingAgentsGraph` 直接 KeyError（#101）
 
 README「快速开始 · 3. 运行分析」的示例只给了 4 个键（`llm_provider` / 两个模型 / `output_language`），

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "langchain-anthropic>=0.3.15",
     "langchain-experimental>=0.3.4",
     "langchain-openai>=0.3.23",
-    "langgraph>=0.4.8",
+    "langgraph>=1.0",
     "langgraph-checkpoint-sqlite>=2.0.0",
     "pandas>=2.3.0",
     "parsel>=1.10.0",

tests/test_missing_data_tasks.py

@@ -0,0 +1,326 @@
+"""Tests for missing-data task recording and retry helpers."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from tradingagents.dataflows import missing_data
+
+
+@pytest.fixture()
+def missing_index(tmp_path, monkeypatch):
+    index = tmp_path / "missing_data_tasks.json"
+    cache_dir = tmp_path / "missing_data_cache"
+    monkeypatch.setattr(missing_data, "_MISSING_DATA_TASKS_FILE", index)
+    monkeypatch.setattr(missing_data, "_MISSING_DATA_CACHE_DIR", cache_dir)
+    return index
+
+
+def test_records_explicit_missing_items(missing_index):
+    task = missing_data.record_tool_result(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="fundamentals",
+        tool_name="get_fundamentals",
+        args={"ticker": "300476", "curr_date": "2026-06-03"},
+        content="| ROE | -- | [数据缺失: ROE] |\n| EPS | -- | [数据缺失: 机构一致预期EPS] |",
+    )
+
+    assert task is not None
+    assert task["ticker"] == "300476"
+    assert task["stage_label"] == "基本面"
+    assert task["tool_label"] == "综合基本面"
+    assert task["missing_items"] == ["ROE", "机构一致预期EPS"]
+
+    tasks = missing_data.get_missing_tasks("300476", "2026-06-03")
+    assert [t["id"] for t in tasks] == [task["id"]]
+
+
+def test_news_title_starting_with_error_is_not_recorded_as_failure(missing_index):
+    task = missing_data.record_tool_result(
+        ticker="600602",
+        trade_date="2026-06-05",
+        stage="news",
+        tool_name="get_news",
+        args={"ticker": "600602"},
+        content="Error in the market: regulators announce a new policy",
+    )
+
+    assert task is None
+    assert missing_data.get_missing_tasks("600602", "2026-06-05") == []
+
+
+def test_structured_error_prefix_is_recorded_as_failure(missing_index):
+    task = missing_data.record_tool_result(
+        ticker="600602",
+        trade_date="2026-06-05",
+        stage="market",
+        tool_name="get_indicators",
+        args={"ticker": "600602"},
+        content="Error: upstream timeout",
+    )
+
+    assert task is not None
+    assert task["missing_items"] == ["接口返回失败或空数据"]
+
+
+def test_no_realtime_northbound_data_is_recorded_as_missing(missing_index):
+    task = missing_data.record_tool_result(
+        ticker="600602",
+        trade_date="2026-06-05",
+        stage="market",
+        tool_name="get_northbound_flow",
+        args={"curr_date": "2026-06-05", "include_history": True},
+        content="# Northbound Capital Flow\nNo realtime data (non-trading hours or holiday)",
+    )
+
+    assert task is not None
+    assert task["tool_label"] == "北向资金"
+    assert task["missing_items"] == ["接口返回失败或空数据"]
+
+
+def test_successful_retry_resolves_task(missing_index, monkeypatch):
+    task = missing_data.record_tool_result(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="market",
+        tool_name="get_indicators",
+        args={"symbol": "300476", "curr_date": "2026-06-03", "look_back_days": 90},
+        content="Error retrieving technical summary for 300476: timeout",
+    )
+    assert task is not None
+
+    fake_tool = MagicMock()
+    fake_tool.invoke.return_value = "最新收盘价: 12.34\n| 指标 | 数值 |\n|---|---|\n| RSI | 55 |"
+    monkeypatch.setattr(missing_data, "_tool_registry", lambda: {"get_indicators": fake_tool})
+    monkeypatch.setattr(
+        missing_data,
+        "sync_missing_snapshot_to_analysis_log",
+        lambda ticker, date, **kwargs: None,
+    )
+
+    result = missing_data.retry_missing_tasks("300476", "2026-06-03")
+
+    assert result["attempted"] == 1
+    assert result["resolved_count"] == 1
+    assert result["remaining_count"] == 0
+    fake_tool.invoke.assert_called_once_with(task["args"])
+
+    all_tasks = missing_data.get_missing_tasks("300476", "2026-06-03", active_only=False)
+    assert all_tasks[0]["status"] == "resolved"
+    assert all_tasks[0]["retry_attempts"] == 1
+
+
+def test_retry_keeps_unresolved_task_when_still_missing(missing_index, monkeypatch):
+    missing_data.record_tool_result(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="fundamentals",
+        tool_name="get_profit_forecast",
+        args={"ticker": "300476"},
+        content="[数据缺失: 机构一致预期EPS]",
+    )
+
+    fake_tool = MagicMock()
+    fake_tool.invoke.return_value = "[数据缺失: 机构一致预期EPS]"
+    monkeypatch.setattr(missing_data, "_tool_registry", lambda: {"get_profit_forecast": fake_tool})
+    monkeypatch.setattr(
+        missing_data,
+        "sync_missing_snapshot_to_analysis_log",
+        lambda ticker, date, **kwargs: None,
+    )
+
+    result = missing_data.retry_missing_tasks("300476", "2026-06-03")
+
+    assert result["resolved_count"] == 0
+    assert result["remaining_count"] == 1
+    assert result["remaining"][0]["status"] == "active"
+    assert result["remaining"][0]["retry_attempts"] == 1
+
+
+def test_snapshot_sync_updates_saved_report(missing_index, tmp_path, monkeypatch):
+    log_path = tmp_path / "full_states_log_2026-06-03.json"
+    log_path.write_text('{"final_trade_decision": "HOLD"}', encoding="utf-8")
+    monkeypatch.setattr(missing_data, "analysis_log_path", lambda ticker, date: Path(log_path))
+
+    missing_data.record_tool_result(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="news",
+        tool_name="get_news",
+        args={"ticker": "300476", "start_date": "2026-05-27", "end_date": "2026-06-03"},
+        content="Error fetching news for 300476: timeout",
+    )
+
+    missing_data.sync_missing_snapshot_to_analysis_log("300476", "2026-06-03")
+
+    text = log_path.read_text(encoding="utf-8")
+    assert '"missing_data_complete": false' in text
+    assert '"tool_name": "get_news"' in text
+
+
+def test_snapshot_sync_preserves_reanalysis_flag(missing_index, tmp_path, monkeypatch):
+    log_path = tmp_path / "full_states_log_2026-06-03.json"
+    log_path.write_text(
+        '{"missing_data_requires_reanalysis": true}', encoding="utf-8"
+    )
+    monkeypatch.setattr(
+        missing_data, "analysis_log_path", lambda ticker, date: log_path
+    )
+
+    missing_data.sync_missing_snapshot_to_analysis_log("300476", "2026-06-03")
+
+    saved = json.loads(log_path.read_text(encoding="utf-8"))
+    assert saved["missing_data_requires_reanalysis"] is True
+
+
+def test_cached_retry_output_is_available_for_reanalysis(missing_index, monkeypatch):
+    args = {"symbol": "300476", "curr_date": "2026-06-03", "look_back_days": 90}
+    missing_data.record_tool_result(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="market",
+        tool_name="get_indicators",
+        args=args,
+        content="Error retrieving technical summary for 300476: timeout",
+    )
+
+    fake_tool = MagicMock()
+    fake_tool.invoke.return_value = "补齐后的技术摘要"
+    monkeypatch.setattr(missing_data, "_tool_registry", lambda: {"get_indicators": fake_tool})
+    monkeypatch.setattr(
+        missing_data,
+        "sync_missing_snapshot_to_analysis_log",
+        lambda ticker, date, **kwargs: None,
+    )
+
+    missing_data.retry_missing_tasks("300476", "2026-06-03")
+
+    output = missing_data.cached_retry_output(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="market",
+        tool_name="get_indicators",
+        args=args,
+    )
+
+    assert output == "补齐后的技术摘要"
+    all_tasks = missing_data.get_missing_tasks("300476", "2026-06-03", active_only=False)
+    assert all_tasks[0]["used_in_reanalysis_at"] is not None
+
+
+def test_consumed_resolved_task_removes_cached_output(missing_index, monkeypatch):
+    args = {"symbol": "300476", "curr_date": "2026-06-03", "look_back_days": 90}
+    task = missing_data.record_tool_result(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="market",
+        tool_name="get_indicators",
+        args=args,
+        content="Error retrieving technical summary for 300476: timeout",
+    )
+    assert task is not None
+
+    fake_tool = MagicMock()
+    fake_tool.invoke.return_value = "补齐后的技术摘要"
+    monkeypatch.setattr(missing_data, "_tool_registry", lambda: {"get_indicators": fake_tool})
+    monkeypatch.setattr(
+        missing_data,
+        "sync_missing_snapshot_to_analysis_log",
+        lambda ticker, date, **kwargs: None,
+    )
+
+    missing_data.retry_missing_tasks("300476", "2026-06-03")
+    resolved = [
+        task
+        for task in missing_data.get_missing_tasks(
+            "300476", "2026-06-03", active_only=False
+        )
+        if task["status"] == "resolved"
+    ]
+    cache_path = Path(resolved[0]["resolved_output_path"])
+    assert cache_path.exists()
+
+    assert (
+        missing_data.cached_retry_output(
+            ticker="300476",
+            trade_date="2026-06-03",
+            stage="market",
+            tool_name="get_indicators",
+            args=args,
+        )
+        == "补齐后的技术摘要"
+    )
+    missing_data.mark_resolved_tasks_consumed("300476", "2026-06-03")
+
+    assert not cache_path.exists()
+    all_tasks = missing_data.get_missing_tasks("300476", "2026-06-03", active_only=False)
+    assert all_tasks[0]["status"] == "consumed"
+    assert (
+        missing_data.cached_retry_output(
+            ticker="300476",
+            trade_date="2026-06-03",
+            stage="market",
+            tool_name="get_indicators",
+            args=args,
+        )
+        is None
+    )
+
+
+def test_unconsumed_retry_output_survives_unrelated_completed_run(
+    missing_index, monkeypatch
+):
+    args = {"symbol": "300476", "curr_date": "2026-06-03"}
+    missing_data.record_tool_result(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="market",
+        tool_name="get_indicators",
+        args=args,
+        content="Error fetching indicators",
+    )
+    fake_tool = MagicMock()
+    fake_tool.invoke.return_value = "补齐后的技术摘要"
+    monkeypatch.setattr(
+        missing_data, "_tool_registry", lambda: {"get_indicators": fake_tool}
+    )
+    monkeypatch.setattr(
+        missing_data,
+        "sync_missing_snapshot_to_analysis_log",
+        lambda ticker, date, **kwargs: None,
+    )
+
+    missing_data.retry_missing_tasks("300476", "2026-06-03")
+    missing_data.mark_resolved_tasks_consumed("300476", "2026-06-03")
+
+    tasks = missing_data.get_missing_tasks(
+        "300476", "2026-06-03", active_only=False
+    )
+    assert tasks[0]["status"] == "resolved"
+    assert Path(tasks[0]["resolved_output_path"]).exists()
+
+
+def test_fresh_run_removes_stale_active_and_consumed_tasks(missing_index):
+    args = {"symbol": "300476"}
+    task = missing_data.record_tool_result(
+        ticker="300476",
+        trade_date="2026-06-03",
+        stage="market",
+        tool_name="get_indicators",
+        args=args,
+        content="Error fetching indicators",
+    )
+    assert task is not None
+    task["status"] = "consumed"
+    missing_data._save_index([task])
+
+    missing_data.reset_missing_tasks_for_run("300476", "2026-06-03")
+
+    assert missing_data.get_missing_tasks(
+        "300476", "2026-06-03", active_only=False
+    ) == []

tests/test_report_viewer_pdf_gate.py

@@ -0,0 +1,270 @@
+"""Regression tests for report PDF controls with missing data."""
+
+from __future__ import annotations
+
+import importlib
+import sys
+import types
+from typing import Any
+
+
+class _Context:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+
+class _FakeStreamlit:
+    def __init__(self) -> None:
+        self.session_state: dict[str, Any] = {}
+        self.buttons: list[dict[str, Any]] = []
+        self.downloads: list[dict[str, Any]] = []
+        self.captions: list[str] = []
+        self.markdowns: list[str] = []
+        self.messages: list[tuple[str, str]] = []
+        self.button_returns: dict[str, bool] = {}
+        self.events: list[tuple[str, str]] = []
+
+    def columns(self, spec):
+        count = len(spec) if isinstance(spec, list) else int(spec)
+        return [_Context() for _ in range(count)]
+
+    def download_button(self, label: str, **kwargs):
+        self.downloads.append({"label": label, **kwargs})
+        return False
+
+    def button(self, label: str, **kwargs):
+        self.buttons.append({"label": label, **kwargs})
+        self.events.append(("button", label))
+        return self.button_returns.get(str(kwargs.get("key")), False)
+
+    def caption(self, text: str):
+        self.captions.append(text)
+        self.events.append(("caption", text))
+
+    def markdown(self, text: str, *args, **kwargs):
+        self.markdowns.append(str(text))
+        self.events.append(("markdown", str(text)))
+        return None
+
+    def error(self, text: str, **kwargs):
+        self.messages.append(("error", text))
+
+    def warning(self, text: str, **kwargs):
+        self.messages.append(("warning", text))
+
+    def success(self, text: str, **kwargs):
+        self.messages.append(("success", text))
+
+    def info(self, text: str, **kwargs):
+        self.messages.append(("info", text))
+
+    def expander(self, *args, **kwargs):
+        return _Context()
+
+    def spinner(self, *args, **kwargs):
+        return _Context()
+
+    def tabs(self, labels):
+        return [_Context() for _ in labels]
+
+    def rerun(self):
+        raise AssertionError("rerun should not be triggered in this test")
+
+
+def _load_report_viewer(monkeypatch, fake_st: _FakeStreamlit):
+    streamlit_stub = types.ModuleType("streamlit")
+    for name in (
+        "button",
+        "caption",
+        "columns",
+        "download_button",
+        "error",
+        "expander",
+        "info",
+        "markdown",
+        "rerun",
+        "spinner",
+        "success",
+        "tabs",
+        "warning",
+    ):
+        setattr(streamlit_stub, name, getattr(fake_st, name))
+    streamlit_stub.session_state = fake_st.session_state
+
+    messages_stub = types.ModuleType("langchain_core.messages")
+    messages_stub.ToolMessage = type("ToolMessage", (), {})
+    langchain_stub = types.ModuleType("langchain_core")
+    langchain_stub.messages = messages_stub
+    pdf_export_stub = types.ModuleType("web.pdf_export")
+    pdf_export_stub.generate_markdown = lambda *args, **kwargs: "markdown"
+    pdf_export_stub.generate_pdf = lambda *args, **kwargs: b"%PDF-stub"
+    missing_data_stub = types.ModuleType("tradingagents.dataflows.missing_data")
+    missing_data_stub.get_missing_tasks = lambda *args, **kwargs: []
+    missing_data_stub.retry_missing_tasks = lambda *args, **kwargs: {
+        "attempted": 0,
+        "resolved_count": 0,
+        "remaining_count": 0,
+        "remaining": [],
+    }
+    stock_display_stub = types.ModuleType("web.stock_display")
+    stock_display_stub.normalize_stock_mentions = (
+        lambda text, ticker, final_state=None: text
+    )
+    stock_display_stub.stock_display_label = (
+        lambda ticker, final_state=None: str(ticker)
+    )
+    dataflow_utils_stub = types.ModuleType("tradingagents.dataflows.utils")
+    dataflow_utils_stub.safe_ticker_component = lambda value: str(value).strip().upper()
+
+    monkeypatch.setitem(sys.modules, "streamlit", streamlit_stub)
+    monkeypatch.setitem(sys.modules, "langchain_core", langchain_stub)
+    monkeypatch.setitem(sys.modules, "langchain_core.messages", messages_stub)
+    monkeypatch.setitem(sys.modules, "web.pdf_export", pdf_export_stub)
+    monkeypatch.setitem(sys.modules, "tradingagents.dataflows.missing_data", missing_data_stub)
+    monkeypatch.setitem(sys.modules, "tradingagents.dataflows.utils", dataflow_utils_stub)
+    monkeypatch.setitem(sys.modules, "web.stock_display", stock_display_stub)
+    sys.modules.pop("web.components.report_viewer", None)
+    return importlib.import_module("web.components.report_viewer")
+
+
+def _install_report_viewer_fakes(monkeypatch, report_viewer, pdf_calls: list[tuple[Any, ...]]) -> None:
+    monkeypatch.setattr(report_viewer, "generate_markdown", lambda *args: "markdown")
+    monkeypatch.setattr(
+        report_viewer,
+        "generate_pdf",
+        lambda *args: pdf_calls.append(args) or b"%PDF-current-report",
+    )
+    monkeypatch.setattr(
+        report_viewer,
+        "get_missing_tasks",
+        lambda ticker, trade_date, active_only=True: [
+            {
+                "status": "active",
+                "stage_label": "基本面",
+                "tool_label": "综合基本面",
+                "missing_items": ["ROE"],
+            }
+        ],
+    )
+
+
+def _install_report_viewer_fakes_without_missing(
+    monkeypatch,
+    report_viewer,
+    pdf_calls: list[tuple[Any, ...]],
+) -> None:
+    _install_report_viewer_fakes(monkeypatch, report_viewer, pdf_calls)
+    monkeypatch.setattr(
+        report_viewer,
+        "get_missing_tasks",
+        lambda ticker, trade_date, active_only=True: [],
+    )
+
+
+def test_pdf_is_generated_and_downloadable_when_missing_data_remains(monkeypatch):
+    fake_st = _FakeStreamlit()
+    report_viewer = _load_report_viewer(monkeypatch, fake_st)
+    pdf_calls: list[tuple[Any, ...]] = []
+    _install_report_viewer_fakes(monkeypatch, report_viewer, pdf_calls)
+
+    report_viewer.render_report(
+        {"missing_data_complete": False},
+        ticker="300476",
+        trade_date="2026-06-04",
+        signal="HOLD",
+    )
+
+    assert len(pdf_calls) == 1
+    assert any(
+        download["label"] == "📄 下载 PDF"
+        and download["data"] == b"%PDF-current-report"
+        for download in fake_st.downloads
+    )
+    missing_buttons = [button for button in fake_st.buttons if button["label"] == "缺失项 (1)"]
+    assert missing_buttons
+    assert missing_buttons[0].get("disabled") is False
+    assert any("PDF 会按当前已有内容生成" in caption for caption in fake_st.captions)
+
+
+def test_requires_reanalysis_keeps_current_pdf_downloadable(monkeypatch):
+    fake_st = _FakeStreamlit()
+    report_viewer = _load_report_viewer(monkeypatch, fake_st)
+    pdf_calls: list[tuple[Any, ...]] = []
+    _install_report_viewer_fakes_without_missing(monkeypatch, report_viewer, pdf_calls)
+
+    report_viewer.render_report(
+        {"missing_data_requires_reanalysis": True},
+        ticker="300476",
+        trade_date="2026-06-04",
+        signal="HOLD",
+    )
+
+    assert len(pdf_calls) == 1
+    assert any(download["label"] == "📄 下载 PDF" for download in fake_st.downloads)
+    assert any(
+        button["label"] == "需重新分析" and button.get("disabled") is False
+        for button in fake_st.buttons
+    )
+    assert any("当前 PDF 仍基于旧报告内容" in caption for caption in fake_st.captions)
+
+
+def test_no_missing_data_shows_disabled_clean_status(monkeypatch):
+    fake_st = _FakeStreamlit()
+    report_viewer = _load_report_viewer(monkeypatch, fake_st)
+    pdf_calls: list[tuple[Any, ...]] = []
+    _install_report_viewer_fakes_without_missing(monkeypatch, report_viewer, pdf_calls)
+
+    report_viewer.render_report(
+        {"missing_data_complete": True, "missing_data_tasks": []},
+        ticker="300476",
+        trade_date="2026-06-04",
+        signal="HOLD",
+    )
+
+    assert any(
+        button["label"] == "✅ 无缺失项" and button.get("disabled") is True
+        for button in fake_st.buttons
+    )
+    assert any(download["label"] == "📄 下载 PDF" for download in fake_st.downloads)
+
+
+def test_retry_missing_button_is_above_missing_task_details(monkeypatch):
+    fake_st = _FakeStreamlit()
+    report_viewer = _load_report_viewer(monkeypatch, fake_st)
+    pdf_calls: list[tuple[Any, ...]] = []
+    _install_report_viewer_fakes(monkeypatch, report_viewer, pdf_calls)
+
+    report_viewer._render_missing_data_panel(
+        {"missing_data_complete": False},
+        ticker="300476",
+        trade_date="2026-06-04",
+        key_suffix="missing-panel",
+    )
+
+    retry_index = fake_st.events.index(("button", "重新取数这些缺失项"))
+    first_task_index = fake_st.events.index(("markdown", "**1. 基本面 · 综合基本面**"))
+
+    assert retry_index < first_task_index
+
+
+def test_reanalysis_starts_a_fresh_run(monkeypatch):
+    fake_st = _FakeStreamlit()
+    report_viewer = _load_report_viewer(monkeypatch, fake_st)
+    key = "reanalyze_after_missing_300476_2026-06-04_panel"
+    fake_st.button_returns[key] = True
+
+    try:
+        report_viewer._start_reanalysis_button(
+            "300476", "2026-06-04", "panel"
+        )
+    except AssertionError as exc:
+        assert str(exc) == "rerun should not be triggered in this test"
+
+    assert fake_st.session_state["start_analysis"] == {
+        "ticker": "300476",
+        "trade_date": "2026-06-04",
+        "fresh": True,
+    }

tests/test_sentiment_data_tools.py

@@ -38,7 +38,12 @@ def test_graph_tool_node_matches_analyst_tools():
     import tradingagents.graph.trading_graph as tg
 
     src = inspect.getsource(tg.TradingAgentsGraph._create_tool_nodes)
-    social_block = re.search(r'"social": ToolNode\(\s*\[(.*?)\]\s*\)', src, re.S)
+    social_block = re.search(
+        r'"social": ToolNode\(\s*\[(.*?)\]\s*'
+        r'(?:,\s*wrap_tool_call\s*=\s*make_tool_call_recorder\("social"\))?\s*,?\s*\)',
+        src,
+        re.S,
+    )
     assert social_block, "找不到 social 的 ToolNode 定义"
     registered = {
         t.strip().rstrip(",")

tradingagents/dataflows/missing_data.py

@@ -0,0 +1,660 @@
+"""Track data tool calls that returned missing or failed data.
+
+The analysis graph can continue even when a vendor endpoint returns an error or
+partial result.  This module keeps those gaps visible after the LLM report is
+finished, and lets the Web UI retry the exact tool calls later.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import threading
+import time
+from functools import wraps
+from pathlib import Path
+from typing import Any
+
+from langchain_core.messages import ToolMessage
+
+from tradingagents.default_config import DEFAULT_CONFIG
+from tradingagents.dataflows.utils import safe_ticker_component
+
+
+_MISSING_DATA_TASKS_FILE = Path.home() / ".tradingagents" / "missing_data_tasks.json"
+_MISSING_DATA_CACHE_DIR = Path.home() / ".tradingagents" / "missing_data_cache"
+_INDEX_LOCK = threading.RLock()
+# Keep the parsed index in memory between ToolNode calls.  The cache is
+# invalidated after writes and also notices external changes through the file
+# stat signature, so a long-running Streamlit process does not repeatedly parse
+# the same JSON file while still seeing edits made by another process.
+_INDEX_CACHE: dict[Path, tuple[tuple[int, int] | None, list[dict[str, Any]]]] = {}
+
+
+STAGE_LABELS = {
+    "market": ("技术分析", "market_report"),
+    "social": ("情绪分析", "sentiment_report"),
+    "news": ("新闻舆情", "news_report"),
+    "fundamentals": ("基本面", "fundamentals_report"),
+    "policy": ("政策分析", "policy_report"),
+    "hot_money": ("游资追踪", "hot_money_report"),
+    "lockup": ("解禁监控", "lockup_report"),
+}
+
+TOOL_LABELS = {
+    "get_stock_data": "K线行情",
+    "get_indicators": "技术指标",
+    "get_fundamentals": "综合基本面",
+    "get_balance_sheet": "资产负债表",
+    "get_cashflow": "现金流量表",
+    "get_income_statement": "利润表",
+    "get_news": "个股新闻",
+    "get_global_news": "宏观新闻",
+    "get_insider_transactions": "股东/内部人交易",
+    "get_profit_forecast": "机构一致预期",
+    "get_hot_stocks": "热门股题材",
+    "get_northbound_flow": "北向资金",
+    "get_concept_blocks": "概念板块",
+    "get_fund_flow": "个股资金流",
+    "get_dragon_tiger_board": "龙虎榜",
+    "get_lockup_expiry": "限售解禁",
+    "get_industry_comparison": "行业对比",
+}
+
+_EXPLICIT_MISSING_RE = re.compile(r"\[数据缺失:\s*([^\]]+?)\s*\]")
+
+_ERROR_PREFIX_RE = re.compile(
+    r"^Error\s*(?:[:：]|(?:retrieving|fetching|loading|reading|requesting|"
+    r"calling|connecting|getting|accessing)\b)",
+    re.IGNORECASE,
+)
+
+_FAILURE_PATTERNS = [
+    re.compile(r"\bError (?:retrieving|fetching)\b", re.IGNORECASE),
+    re.compile(r"\bNo data found\b", re.IGNORECASE),
+    re.compile(r"\bNo data from\b", re.IGNORECASE),
+    re.compile(r"\bNo realtime data\b", re.IGNORECASE),
+    re.compile(r"\bNo OHLCV data\b", re.IGNORECASE),
+    re.compile(r"\bAPI error\b", re.IGNORECASE),
+    re.compile(r"查询失败"),
+    re.compile(r"获取失败"),
+    re.compile(r"获取为空"),
+    re.compile(r"工具调用失败"),
+    re.compile(r"无法获取"),
+]
+
+
+def _index_transaction(func):
+    """Serialize read-modify-write operations from parallel graph nodes."""
+
+    @wraps(func)
+    def _locked(*args, **kwargs):
+        with _INDEX_LOCK:
+            return func(*args, **kwargs)
+
+    return _locked
+
+
+def _cache_path(task_id: str) -> Path:
+    return _MISSING_DATA_CACHE_DIR / f"{task_id}.txt"
+
+
+def _write_cached_output(task_id: str, content: Any) -> Path:
+    path = _cache_path(task_id)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".tmp")
+    with open(tmp, "w", encoding="utf-8") as f:
+        f.write("" if content is None else str(content))
+    tmp.replace(path)
+    return path
+
+
+def _read_cached_output(task_id: str) -> str | None:
+    path = _cache_path(task_id)
+    try:
+        return path.read_text(encoding="utf-8")
+    except OSError:
+        return None
+
+
+def _index_file_signature(path: Path) -> tuple[int, int] | None:
+    try:
+        stat = path.stat()
+    except OSError:
+        return None
+    return stat.st_mtime_ns, stat.st_size
+
+
+def _load_index() -> list[dict[str, Any]]:
+    path = _MISSING_DATA_TASKS_FILE
+    signature = _index_file_signature(path)
+    cached = _INDEX_CACHE.get(path)
+    if cached is not None and cached[0] == signature:
+        return cached[1]
+
+    if signature is None:
+        entries: list[dict[str, Any]] = []
+    else:
+        try:
+            with open(path, encoding="utf-8") as f:
+                data = json.load(f)
+        except (OSError, json.JSONDecodeError):
+            data = []
+
+        entries = []
+        if isinstance(data, list):
+            for item in data:
+                if not isinstance(item, dict):
+                    continue
+                ticker = str(item.get("ticker", "")).strip().upper()
+                trade_date = str(item.get("trade_date", "")).strip()
+                task_id = str(item.get("id", "")).strip()
+                if not ticker or not trade_date or not task_id:
+                    continue
+                item["ticker"] = ticker
+                item["trade_date"] = trade_date
+                item["id"] = task_id
+                entries.append(item)
+
+    _INDEX_CACHE[path] = (signature, entries)
+    return entries
+
+
+def _save_index(entries: list[dict[str, Any]]) -> None:
+    path = _MISSING_DATA_TASKS_FILE
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".tmp")
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(entries, f, ensure_ascii=False, indent=2, default=str)
+    tmp.replace(path)
+    # The next read can reuse the just-written in-memory list without parsing
+    # the JSON again.  Copy the list so callers cannot mutate the cache behind
+    # the index lock.
+    _INDEX_CACHE[path] = (_index_file_signature(path), list(entries))
+
+
+def _normalize_args(args: Any) -> dict[str, Any]:
+    if isinstance(args, dict):
+        return dict(args)
+    return {"value": args}
+
+
+def _task_id(
+    ticker: str,
+    trade_date: str,
+    stage: str,
+    tool_name: str,
+    args: dict[str, Any],
+) -> str:
+    raw = json.dumps(
+        {
+            "ticker": ticker.upper(),
+            "trade_date": trade_date,
+            "stage": stage,
+            "tool_name": tool_name,
+            "args": args,
+        },
+        ensure_ascii=False,
+        sort_keys=True,
+        default=str,
+    )
+    return "md_" + hashlib.sha256(raw.encode("utf-8")).hexdigest()[:20]
+
+
+def _text_preview(text: str, limit: int = 500) -> str:
+    cleaned = re.sub(r"\s+", " ", text).strip()
+    if len(cleaned) <= limit:
+        return cleaned
+    return cleaned[: limit - 3] + "..."
+
+
+def _diagnose_result(content: Any, *, status: str = "success") -> dict[str, Any] | None:
+    text = "" if content is None else str(content)
+    markers = sorted({m.strip() for m in _EXPLICIT_MISSING_RE.findall(text) if m.strip()})
+    stripped = text.strip()
+    # A successful tool may return a news article whose title starts with
+    # ``Error``.  Treat that prefix as a failure only for short, structured
+    # error responses; actual ToolMessage errors are covered by ``status`` and
+    # the more specific patterns below.
+    error_prefix = bool(
+        stripped
+        and len(stripped) <= 200
+        and _ERROR_PREFIX_RE.search(stripped)
+    )
+    failed = (
+        status == "error"
+        or error_prefix
+        or any(pattern.search(text) for pattern in _FAILURE_PATTERNS)
+    )
+
+    if not markers and not failed:
+        return None
+
+    if not markers:
+        markers = ["接口返回失败或空数据"]
+
+    severity = "failed" if failed and not _EXPLICIT_MISSING_RE.search(text) else "partial"
+    return {
+        "severity": severity,
+        "missing_items": markers,
+        "message": _text_preview(text) or "工具调用没有返回可用内容",
+    }
+
+
+def _same_run(entry: dict[str, Any], ticker: str, trade_date: str) -> bool:
+    return (
+        str(entry.get("ticker", "")).upper() == ticker.upper()
+        and str(entry.get("trade_date", "")) == trade_date
+    )
+
+
+@_index_transaction
+def reset_missing_tasks_for_run(ticker: str, trade_date: str) -> None:
+    """Clear prior active tasks when a fresh full analysis starts.
+
+    Resolved retry outputs are preserved so the next full reanalysis can reuse
+    the data that was just recovered from flaky endpoints.
+    """
+    ticker = ticker.strip().upper()
+    trade_date = trade_date.strip()
+    entries = [
+        entry
+        for entry in _load_index()
+        if not _same_run(entry, ticker, trade_date) or entry.get("status") == "resolved"
+    ]
+    _save_index(entries)
+
+
+@_index_transaction
+def record_tool_result(
+    *,
+    ticker: str,
+    trade_date: str,
+    stage: str,
+    tool_name: str,
+    args: Any,
+    content: Any,
+    status: str = "success",
+    retry_attempt: bool = False,
+) -> dict[str, Any] | None:
+    """Record or resolve a missing-data task based on one tool result."""
+    ticker = (ticker or "").strip().upper()
+    trade_date = (trade_date or "").strip()
+    if not ticker or not trade_date or not tool_name:
+        return None
+
+    normalized_args = _normalize_args(args)
+    task_id = _task_id(ticker, trade_date, stage, tool_name, normalized_args)
+    diagnosis = _diagnose_result(content, status=status)
+    entries = _load_index()
+    now = time.time()
+
+    if diagnosis is None:
+        changed = False
+        for entry in entries:
+            if entry.get("id") != task_id or entry.get("status") != "active":
+                continue
+            entry["status"] = "resolved"
+            entry["resolved_at"] = now
+            entry["updated_at"] = now
+            entry["last_result_preview"] = _text_preview(str(content))
+            if retry_attempt:
+                cached_path = _write_cached_output(task_id, content)
+                entry["resolved_output_path"] = str(cached_path)
+                entry["resolved_output_chars"] = len("" if content is None else str(content))
+                entry["retry_attempts"] = int(entry.get("retry_attempts", 0)) + 1
+                entry["last_retry_at"] = now
+            changed = True
+        if changed:
+            _save_index(entries)
+        return None
+
+    stage_label, report_key = STAGE_LABELS.get(
+        stage, (stage or "未知分析段", "")
+    )
+    existing = next((entry for entry in entries if entry.get("id") == task_id), None)
+    task = existing or {
+        "id": task_id,
+        "ticker": ticker,
+        "trade_date": trade_date,
+        "stage": stage,
+        "stage_label": stage_label,
+        "report_key": report_key,
+        "tool_name": tool_name,
+        "tool_label": TOOL_LABELS.get(tool_name, tool_name),
+        "args": normalized_args,
+        "first_seen": now,
+        "retry_attempts": 0,
+    }
+
+    task.update(
+        {
+            "status": "active",
+            "severity": diagnosis["severity"],
+            "missing_items": diagnosis["missing_items"],
+            "message": diagnosis["message"],
+            "updated_at": now,
+            "resolved_at": None,
+        }
+    )
+    if retry_attempt:
+        task["retry_attempts"] = int(task.get("retry_attempts", 0)) + 1
+        task["last_retry_at"] = now
+
+    if existing is None:
+        entries.append(task)
+
+    entries.sort(key=lambda item: float(item.get("updated_at", 0)), reverse=True)
+    _save_index(entries)
+    return task
+
+
+@_index_transaction
+def cached_retry_output(
+    *,
+    ticker: str,
+    trade_date: str,
+    stage: str,
+    tool_name: str,
+    args: Any,
+) -> str | None:
+    """Return a successful retry output for the exact tool call, if any."""
+    ticker = (ticker or "").strip().upper()
+    trade_date = (trade_date or "").strip()
+    if not ticker or not trade_date or not tool_name:
+        return None
+
+    normalized_args = _normalize_args(args)
+    task_id = _task_id(ticker, trade_date, stage, tool_name, normalized_args)
+    entries = _load_index()
+    for entry in entries:
+        if entry.get("id") != task_id or entry.get("status") != "resolved":
+            continue
+        output = _read_cached_output(task_id)
+        if output is not None:
+            entry["used_in_reanalysis_at"] = time.time()
+            _save_index(entries)
+            return output
+    return None
+
+
+def _request_context(stage: str, request: Any) -> tuple[str, str, str, dict[str, Any], str]:
+    state = getattr(request, "state", None) or {}
+    ticker = str(state.get("company_of_interest", "")).strip().upper()
+    trade_date = str(state.get("trade_date", "")).strip()
+    tool_call = getattr(request, "tool_call", {}) or {}
+    tool_name = str(tool_call.get("name", ""))
+    args = tool_call.get("args", {})
+    tool_call_id = str(tool_call.get("id", ""))
+    return ticker, trade_date, tool_name, _normalize_args(args), tool_call_id
+
+
+def make_tool_call_recorder(stage: str):
+    """Create a ToolNode wrapper that records missing data after execution."""
+
+    def _wrapper(request: Any, execute: Any) -> Any:
+        ticker, trade_date, tool_name, args, tool_call_id = _request_context(stage, request)
+        cached = (
+            cached_retry_output(
+                ticker=ticker,
+                trade_date=trade_date,
+                stage=stage,
+                tool_name=tool_name,
+                args=args,
+            )
+            if tool_call_id
+            else None
+        )
+        if cached is not None and tool_call_id:
+            return ToolMessage(
+                content=cached,
+                name=tool_name,
+                tool_call_id=tool_call_id,
+                status="success",
+            )
+
+        response = execute(request)
+        try:
+            for item in response if isinstance(response, list) else [response]:
+                record_tool_result(
+                    ticker=ticker,
+                    trade_date=trade_date,
+                    stage=stage,
+                    tool_name=tool_name,
+                    args=args,
+                    content=getattr(item, "content", ""),
+                    status=getattr(item, "status", "success"),
+                )
+        except Exception:
+            # Missing-data bookkeeping must never break the analysis graph.
+            pass
+        return response
+
+    return _wrapper
+
+
+@_index_transaction
+def get_missing_tasks(
+    ticker: str,
+    trade_date: str,
+    *,
+    active_only: bool = True,
+) -> list[dict[str, Any]]:
+    """Return missing-data tasks for one ticker/date."""
+    ticker = ticker.strip().upper()
+    trade_date = trade_date.strip()
+    tasks = [
+        entry
+        for entry in _load_index()
+        if _same_run(entry, ticker, trade_date)
+        and (not active_only or entry.get("status") == "active")
+    ]
+    tasks.sort(key=lambda item: (str(item.get("stage", "")), str(item.get("tool_name", ""))))
+    return tasks
+
+
+@_index_transaction
+def mark_resolved_tasks_consumed(ticker: str, trade_date: str) -> None:
+    """Archive resolved retry outputs after a fresh full analysis completes."""
+    ticker = ticker.strip().upper()
+    trade_date = trade_date.strip()
+    entries = _load_index()
+    changed = False
+    now = time.time()
+
+    for entry in entries:
+        if (
+            not _same_run(entry, ticker, trade_date)
+            or entry.get("status") != "resolved"
+            or not entry.get("used_in_reanalysis_at")
+        ):
+            continue
+        entry["status"] = "consumed"
+        entry["consumed_at"] = now
+        cache_path = entry.get("resolved_output_path")
+        if cache_path:
+            try:
+                Path(cache_path).unlink(missing_ok=True)
+            except OSError:
+                pass
+        changed = True
+
+    if changed:
+        _save_index(entries)
+
+
+def attach_missing_data_snapshot(
+    final_state: dict[str, Any],
+    ticker: str,
+    trade_date: str,
+    *,
+    requires_reanalysis: bool = False,
+) -> dict[str, Any]:
+    """Attach the current active missing-data snapshot to a final state."""
+    tasks = get_missing_tasks(ticker, trade_date, active_only=True)
+    final_state["missing_data_tasks"] = tasks
+    final_state["missing_data_complete"] = len(tasks) == 0
+    final_state["missing_data_requires_reanalysis"] = bool(requires_reanalysis)
+    final_state["missing_data_updated_at"] = time.time()
+    return final_state
+
+
+def analysis_log_path(ticker: str, trade_date: str) -> Path:
+    safe_ticker = safe_ticker_component(ticker)
+    return (
+        Path(DEFAULT_CONFIG["results_dir"])
+        / safe_ticker
+        / "TradingAgentsStrategy_logs"
+        / f"full_states_log_{trade_date}.json"
+    )
+
+
+def sync_missing_snapshot_to_analysis_log(
+    ticker: str,
+    trade_date: str,
+    *,
+    requires_reanalysis: bool | None = None,
+) -> None:
+    """Update an already-saved report JSON with the latest missing-data snapshot."""
+    path = analysis_log_path(ticker, trade_date)
+    if not path.exists():
+        return
+
+    try:
+        with open(path, encoding="utf-8") as f:
+            state = json.load(f)
+    except (OSError, json.JSONDecodeError):
+        return
+
+    if requires_reanalysis is None:
+        requires_reanalysis = bool(state.get("missing_data_requires_reanalysis", False))
+    attach_missing_data_snapshot(
+        state,
+        ticker,
+        trade_date,
+        requires_reanalysis=bool(requires_reanalysis),
+    )
+    tmp = path.with_suffix(".tmp")
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(state, f, ensure_ascii=False, indent=4, default=str)
+    tmp.replace(path)
+
+
+def _tool_registry() -> dict[str, Any]:
+    from tradingagents.agents.utils.agent_utils import (
+        get_balance_sheet,
+        get_cashflow,
+        get_concept_blocks,
+        get_dragon_tiger_board,
+        get_fund_flow,
+        get_fundamentals,
+        get_global_news,
+        get_hot_stocks,
+        get_income_statement,
+        get_indicators,
+        get_industry_comparison,
+        get_insider_transactions,
+        get_lockup_expiry,
+        get_news,
+        get_northbound_flow,
+        get_profit_forecast,
+        get_stock_data,
+    )
+
+    tools = [
+        get_stock_data,
+        get_indicators,
+        get_fundamentals,
+        get_balance_sheet,
+        get_cashflow,
+        get_income_statement,
+        get_news,
+        get_global_news,
+        get_insider_transactions,
+        get_profit_forecast,
+        get_hot_stocks,
+        get_northbound_flow,
+        get_concept_blocks,
+        get_fund_flow,
+        get_dragon_tiger_board,
+        get_lockup_expiry,
+        get_industry_comparison,
+    ]
+    return {tool.name: tool for tool in tools}
+
+
+def retry_missing_tasks(
+    ticker: str,
+    trade_date: str,
+    *,
+    task_ids: list[str] | None = None,
+) -> dict[str, Any]:
+    """Retry active missing-data tool calls.
+
+    Returns a summary with resolved/unresolved task records.  The analysis report
+    itself is not mutated here; if all missing tasks resolve, the Web UI can kick
+    off a fresh full analysis so downstream decisions are regenerated from the
+    now-complete data.
+    """
+    selected = set(task_ids or [])
+    tasks = [
+        task
+        for task in get_missing_tasks(ticker, trade_date, active_only=True)
+        if not selected or task["id"] in selected
+    ]
+
+    registry = _tool_registry()
+    resolved: list[dict[str, Any]] = []
+    unresolved: list[dict[str, Any]] = []
+
+    for task in tasks:
+        tool = registry.get(task.get("tool_name"))
+        if tool is None:
+            task["message"] = f"找不到工具: {task.get('tool_name')}"
+            unresolved.append(task)
+            continue
+
+        try:
+            output = tool.invoke(task.get("args") or {})
+            new_task = record_tool_result(
+                ticker=ticker,
+                trade_date=trade_date,
+                stage=str(task.get("stage", "")),
+                tool_name=str(task.get("tool_name", "")),
+                args=task.get("args") or {},
+                content=output,
+                status="success",
+                retry_attempt=True,
+            )
+            if new_task is None:
+                resolved.append({**task, "last_result_preview": _text_preview(str(output))})
+            else:
+                unresolved.append(new_task)
+        except Exception as exc:
+            new_task = record_tool_result(
+                ticker=ticker,
+                trade_date=trade_date,
+                stage=str(task.get("stage", "")),
+                tool_name=str(task.get("tool_name", "")),
+                args=task.get("args") or {},
+                content=f"{type(exc).__name__}: {exc}",
+                status="error",
+                retry_attempt=True,
+            )
+            unresolved.append(new_task or task)
+
+    remaining = get_missing_tasks(ticker, trade_date, active_only=True)
+    sync_missing_snapshot_to_analysis_log(
+        ticker,
+        trade_date,
+        requires_reanalysis=True if resolved else None,
+    )
+    return {
+        "attempted": len(tasks),
+        "resolved": resolved,
+        "unresolved": unresolved,
+        "remaining": remaining,
+        "remaining_count": len(remaining),
+        "resolved_count": len(resolved),
+    }

tradingagents/graph/trading_graph.py

@@ -25,6 +25,12 @@ from tradingagents.agents.utils.agent_states import (
     RiskDebateState,
 )
 from tradingagents.dataflows.config import set_config
+from tradingagents.dataflows.missing_data import (
+    attach_missing_data_snapshot,
+    mark_resolved_tasks_consumed,
+    make_tool_call_recorder,
+    reset_missing_tasks_for_run,
+)
 
 # Import the new abstract tool methods from agent_utils
 from tradingagents.agents.utils.agent_utils import (
@@ -443,7 +449,8 @@ class TradingAgentsGraph:
                     get_stock_data,
                     # Technical indicators
                     get_indicators,
-                ]
+                ],
+                wrap_tool_call=make_tool_call_recorder("market"),
             ),
             "social": ToolNode(
                 [
@@ -453,7 +460,8 @@ class TradingAgentsGraph:
                     get_fund_flow,
                     get_hot_stocks,
                     get_stock_data,
-                ]
+                ],
+                wrap_tool_call=make_tool_call_recorder("social"),
             ),
             "news": ToolNode(
                 [
@@ -461,7 +469,8 @@ class TradingAgentsGraph:
                     get_news,
                     get_global_news,
                     get_insider_transactions,
-                ]
+                ],
+                wrap_tool_call=make_tool_call_recorder("news"),
             ),
             "fundamentals": ToolNode(
                 [
@@ -471,13 +480,15 @@ class TradingAgentsGraph:
                     get_income_statement,
                     get_profit_forecast,
                     get_industry_comparison,
-                ]
+                ],
+                wrap_tool_call=make_tool_call_recorder("fundamentals"),
             ),
             "policy": ToolNode(
                 [
                     get_news,
                     get_global_news,
-                ]
+                ],
+                wrap_tool_call=make_tool_call_recorder("policy"),
             ),
             "hot_money": ToolNode(
                 [
@@ -490,7 +501,8 @@ class TradingAgentsGraph:
                     get_fund_flow,
                     get_dragon_tiger_board,
                     get_industry_comparison,
-                ]
+                ],
+                wrap_tool_call=make_tool_call_recorder("hot_money"),
             ),
             "lockup": ToolNode(
                 [
@@ -498,7 +510,8 @@ class TradingAgentsGraph:
                     get_news,
                     get_fundamentals,
                     get_lockup_expiry,
-                ]
+                ],
+                wrap_tool_call=make_tool_call_recorder("lockup"),
             ),
         }
 
@@ -649,6 +662,7 @@ class TradingAgentsGraph:
 
         # Initialize state only for fresh runs. Passing a new initial state to
         # LangGraph would start a new run and replay completed nodes.
+        reset_missing_tasks_for_run(company_name, str(trade_date))
         past_context = self.memory_log.get_past_context(company_name)
         init_agent_state = self.propagator.create_initial_state(
             company_name, trade_date, past_context=past_context
@@ -657,10 +671,12 @@ class TradingAgentsGraph:
 
     def finalize_graph_run(self, company_name, trade_date, final_state):
         """Persist a completed run and clear its checkpoint."""
+        attach_missing_data_snapshot(final_state, company_name, str(trade_date))
         self.curr_state = final_state
 
         # Log state to disk.
         self._log_state(trade_date, final_state)
+        mark_resolved_tasks_consumed(company_name, str(trade_date))
 
         # Store decision for deferred reflection on the next same-ticker run.
         self.memory_log.store_decision(
@@ -718,6 +734,10 @@ class TradingAgentsGraph:
             "policy_report": final_state.get("policy_report", ""),
             "hot_money_report": final_state.get("hot_money_report", ""),
             "lockup_report": final_state.get("lockup_report", ""),
+            "missing_data_tasks": final_state.get("missing_data_tasks", []),
+            "missing_data_complete": final_state.get("missing_data_complete", True),
+            "missing_data_requires_reanalysis": final_state.get("missing_data_requires_reanalysis", False),
+            "missing_data_updated_at": final_state.get("missing_data_updated_at"),
             "investment_debate_state": {
                 "bull_history": final_state["investment_debate_state"]["bull_history"],
                 "bear_history": final_state["investment_debate_state"]["bear_history"],

web/components/report_viewer.py

@@ -2,11 +2,16 @@
 
 from __future__ import annotations
 
+from datetime import datetime
 import re
 from typing import Any
 
 import streamlit as st
 
+from tradingagents.dataflows.missing_data import (
+    get_missing_tasks,
+    retry_missing_tasks,
+)
 from web.pdf_export import generate_markdown, generate_pdf
 from web.stock_display import normalize_stock_mentions, stock_display_label
 
@@ -45,6 +50,121 @@ def _display_report_text(text: Any, ticker: str, final_state: dict[str, Any]) ->
     return normalize_stock_mentions(cleaned, ticker, final_state)
 
 
+def _active_missing_tasks(
+    final_state: dict[str, Any],
+    ticker: str,
+    trade_date: str,
+) -> list[dict[str, Any]]:
+    live_tasks = get_missing_tasks(ticker, trade_date, active_only=True)
+    if live_tasks:
+        return live_tasks
+
+    tasks = final_state.get("missing_data_tasks")
+    if not isinstance(tasks, list):
+        return []
+    return [
+        task
+        for task in tasks
+        if isinstance(task, dict) and task.get("status", "active") == "active"
+    ]
+
+
+def _format_task_args(args: Any) -> str:
+    if not isinstance(args, dict):
+        return str(args)
+    return ", ".join(f"{key}={value}" for key, value in args.items())
+
+
+def _start_reanalysis_button(
+    ticker: str,
+    trade_date: str,
+    key_suffix: str,
+) -> None:
+    if st.button(
+        "用补齐后的数据重新分析",
+        key=f"reanalyze_after_missing_{ticker}_{trade_date}_{key_suffix}",
+        type="primary",
+        use_container_width=True,
+    ):
+        st.session_state["start_analysis"] = {
+            "ticker": ticker,
+            "trade_date": trade_date,
+            "fresh": True,
+        }
+        st.session_state["viewing_history"] = None
+        st.rerun()
+
+
+def _render_missing_data_panel(
+    final_state: dict[str, Any],
+    ticker: str,
+    trade_date: str,
+    key_suffix: str,
+) -> None:
+    tasks = _active_missing_tasks(final_state, ticker, trade_date)
+    count = len(tasks)
+    requires_reanalysis = bool(final_state.get("missing_data_requires_reanalysis"))
+
+    if count == 0:
+        if requires_reanalysis:
+            st.info("缺失接口已经补到数据，但当前报告仍是补数前生成的。请重新分析一次，让分析师、辩论和最终决策使用补齐后的数据。")
+            _start_reanalysis_button(ticker, trade_date, key_suffix)
+        elif (
+            final_state.get("missing_data_tasks") == []
+            or final_state.get("missing_data_complete") is True
+        ):
+            st.success("缺失数据清单为空：当前没有记录到取数失败或必采项缺失。", icon="✅")
+        return
+
+    st.warning(
+        f"当前报告仍有 {count} 个取数缺口。你仍可先下载按现有内容生成的 PDF；补齐后再重新分析，可得到吸收补齐数据的完整版。",
+        icon="⚠️",
+    )
+
+    with st.expander(f"查看缺失/失败取数项（{count}）", expanded=False):
+        retry_key = f"retry_missing_{ticker}_{trade_date}_{key_suffix}"
+        if st.button("重新取数这些缺失项", key=retry_key, type="primary", use_container_width=True):
+            with st.spinner("正在重新调用缺失数据接口..."):
+                result = retry_missing_tasks(ticker, trade_date)
+            final_state["missing_data_tasks"] = result.get("remaining", [])
+            final_state["missing_data_complete"] = result.get("remaining_count", 0) == 0
+            if result.get("resolved_count", 0):
+                final_state["missing_data_requires_reanalysis"] = True
+            final_state["missing_data_updated_at"] = datetime.now().timestamp()
+            st.session_state[f"{retry_key}_result"] = result
+            st.rerun()
+
+        retry_result = st.session_state.get(f"{retry_key}_result")
+        if retry_result:
+            resolved_count = retry_result.get("resolved_count", 0)
+            remaining_count = retry_result.get("remaining_count", 0)
+            if resolved_count:
+                st.success(f"本次已补到 {resolved_count} 个缺失接口的数据。")
+            if remaining_count:
+                st.info(f"仍有 {remaining_count} 个缺失项没有补齐，可稍后继续重试。")
+            else:
+                st.success("所有缺失取数项都已补齐。现在可以重新分析，生成基于补齐数据的完整报告。")
+                _start_reanalysis_button(ticker, trade_date, key_suffix)
+
+        for idx, task in enumerate(tasks, start=1):
+            stage = task.get("stage_label") or task.get("stage") or "未知分析段"
+            tool = task.get("tool_label") or task.get("tool_name") or "未知接口"
+            missing_items = task.get("missing_items") or []
+            if isinstance(missing_items, list):
+                missing_text = "、".join(str(item) for item in missing_items)
+            else:
+                missing_text = str(missing_items)
+            st.markdown(f"**{idx}. {stage} · {tool}**")
+            st.caption(f"调用参数：{_format_task_args(task.get('args', {}))}")
+            if missing_text:
+                st.markdown(f"- 缺失内容：{missing_text}")
+            if task.get("message"):
+                st.markdown(f"- 返回信息：{task['message']}")
+            retry_attempts = int(task.get("retry_attempts", 0) or 0)
+            if retry_attempts:
+                st.caption(f"已重试 {retry_attempts} 次")
+
+
 def render_report(
     final_state: dict[str, Any],
     ticker: str,
@@ -87,9 +207,11 @@ def render_report(
 
     st.caption("⚠️ 本报告由 AI 自动生成，仅供学习研究，不构成投资建议。")
 
-    # Markdown export always works (no font dependency); PDF is generated
-    # lazily and guarded so a PDF/font failure never crashes the results page.
-    col_md, col_pdf, col_spacer = st.columns([1, 1, 2])
+    missing_count = len(_active_missing_tasks(final_state, ticker, trade_date))
+    requires_reanalysis = bool(final_state.get("missing_data_requires_reanalysis"))
+    missing_panel_key = f"missing_panel_{ticker}_{trade_date}"
+
+    col_md, col_pdf, col_missing, col_spacer = st.columns([1, 1, 1, 1])
     with col_md:
         md_text = generate_markdown(final_state, ticker, trade_date, signal)
         st.download_button(
@@ -117,6 +239,32 @@ def render_report(
                 help=f"PDF 生成失败，请改用 Markdown 导出。原因：{exc}",
             )
 
+    with col_missing:
+        if missing_count:
+            missing_label = f"缺失项 ({missing_count})"
+            disabled = False
+        elif requires_reanalysis:
+            missing_label = "需重新分析"
+            disabled = False
+        else:
+            missing_label = "✅ 无缺失项"
+            disabled = True
+        if st.button(
+            missing_label,
+            key=f"missing_toggle_{ticker}_{trade_date}",
+            use_container_width=True,
+            disabled=disabled,
+        ):
+            st.session_state[missing_panel_key] = not st.session_state.get(missing_panel_key, False)
+
+    if missing_count:
+        st.caption(f"⚠️ 仍有 {missing_count} 个取数缺口；PDF 会按当前已有内容生成，可能缺少部分分析。")
+    elif requires_reanalysis:
+        st.caption("✅ 缺失接口已补到数据；当前 PDF 仍基于旧报告内容，重新分析后可生成吸收补齐数据的完整版。")
+
+    if st.session_state.get(missing_panel_key):
+        _render_missing_data_panel(final_state, ticker, trade_date, missing_panel_key)
+
     st.markdown("---")
 
     inv_plan = final_state.get("investment_plan", "")

web/pdf_export.py

@@ -329,6 +329,24 @@ def _signal_color(signal: str) -> tuple[int, int, int]:
     return (251, 191, 36)
 
 
+def _missing_data_warning(final_state: dict[str, Any]) -> str | None:
+    """Return a concise warning when a report used incomplete inputs."""
+    tasks = final_state.get("missing_data_tasks")
+    active_count = (
+        sum(
+            1 for task in tasks
+            if isinstance(task, dict) and task.get("status", "active") == "active"
+        )
+        if isinstance(tasks, list)
+        else 0
+    )
+    if active_count:
+        return f"⚠️ 数据不完整：仍有 {active_count} 个取数缺口，本报告按当前已有内容生成。"
+    if final_state.get("missing_data_requires_reanalysis"):
+        return "⚠️ 数据已补齐但尚未重新分析：本报告仍基于补数前的分析结果。"
+    return None
+
+
 _REPORT_SECTIONS = [
     ("market_report", "技术分析报告"),
     ("sentiment_report", "市场情绪报告"),
@@ -353,6 +371,7 @@ class _ReportPDF(FPDF):
         self.ticker_label = stock_display_label(ticker, final_state)
         self.trade_date = trade_date
         self.signal = signal
+        self.final_state = final_state or {}
         regular_font, bold_font = _find_cjk_fonts()
 
         try:
@@ -414,6 +433,13 @@ class _ReportPDF(FPDF):
         self.cell(0, 12, "A股多Agent投研分析报告", align="C")
         self.ln(20)
 
+        warning = _missing_data_warning(self.final_state)
+        if warning:
+            self._use_font("B", 11)
+            self.set_text_color(190, 70, 20)
+            self._write_multicell(6, warning, align="C")
+            self.ln(8)
+
         self._use_font("B", 36)
         self.set_text_color(30, 30, 30)
         self.cell(0, 18, self.ticker_label, align="C")
@@ -664,6 +690,9 @@ def generate_markdown(final_state: dict[str, Any], ticker: str, trade_date: str,
         "---",
         "",
     ]
+    warning = _missing_data_warning(final_state)
+    if warning:
+        out.extend([f"> {warning}", ""])
     for title, content in _collect_sections(final_state, ticker):
         out.append(f"## {title}")
         out.append("")