diff --git a/CHANGELOG.md b/CHANGELOG.md index 9127a71..15e554e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,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`）， diff --git a/pyproject.toml b/pyproject.toml index 7320af9..daac221 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/test_missing_data_tasks.py b/tests/test_missing_data_tasks.py new file mode 100644 index 0000000..351bd48 --- /dev/null +++ b/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 + ) == [] diff --git a/tests/test_report_viewer_pdf_gate.py b/tests/test_report_viewer_pdf_gate.py new file mode 100644 index 0000000..4784398 --- /dev/null +++ b/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, + } diff --git a/tests/test_sentiment_data_tools.py b/tests/test_sentiment_data_tools.py index 4d51790..83eb26b 100644 --- a/tests/test_sentiment_data_tools.py +++ b/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(",") diff --git a/tradingagents/dataflows/missing_data.py b/tradingagents/dataflows/missing_data.py new file mode 100644 index 0000000..7d22046 --- /dev/null +++ b/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), + } diff --git a/tradingagents/graph/trading_graph.py b/tradingagents/graph/trading_graph.py index 889f82d..a1bb149 100644 --- a/tradingagents/graph/trading_graph.py +++ b/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"], diff --git a/web/components/report_viewer.py b/web/components/report_viewer.py index eda5946..84c1a65 100644 --- a/web/components/report_viewer.py +++ b/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", "")