diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index edc9d1d6e..11acdd68f 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - [修复] 修复 Windows 桌面端转抄后端 stdout/stderr 时中文日志可能乱码的问题，统一优先使用 UTF-8 并兼容本地代码页回退 - [改进] Docker 发布工作流收敛为更清晰的正式发布与手动补发链路，并统一官方 Docker Hub 镜像名为 `zhulinsen/daily_stock_analysis` - [文档] 补充官方镜像拉取、`docker run` 用法与 `.env` / 数据目录映射说明，不再仅覆盖 Compose 部署路径 +- [改进] Agent 日线工具优先复用本地缓存，并持久化新获取的日线与新闻情报 - [修复] GitHub Actions 每日分析工作流补齐 `LLM_CHANNELS`、多 Key 与常用 `LLM__*` 渠道变量透传，避免本地可用的多模型配置在云端定时任务中失效（Fixes #1063, #872） - [文档] 修正 `feishu_sender.py` 中飞书自定义机器人 Webhook 消息格式示例为 interactive card JSON，并补充飞书自动化 Webhook 触发器配置教程（参数 JSON 与 `card.elements[0].text.content` 字段映射）。 diff --git a/docs/full-guide.md b/docs/full-guide.md index afddcaf12..33f9ed540 100644 --- a/docs/full-guide.md +++ b/docs/full-guide.md @@ -1192,6 +1192,14 @@ A: 检查是否启用了 Actions，以及 cron 表达式是否正确（注意是 - On online failure, fallback to latest cached rate and mark `is_stale=true`. - Main snapshot/risk pipeline stays available even when online FX fetch is unavailable. +## Agent 工具数据缓存与持久化 + +- `get_daily_history` 会先尝试复用本地 `stock_daily` 日线缓存；缓存新鲜且至少覆盖首页默认的 30 条记录时，不再重复请求外部数据源。 +- 当 Agent 请求的天数多于本地缓存记录数时，工具会返回实际可用记录，并通过 `partial_cache=true`、`requested_days`、`actual_records` 标明这是部分缓存命中。 +- 缓存缺失或过期时，工具仍会按原逻辑从数据源获取日线数据；获取成功后会 best-effort 写回 `stock_daily`，保存失败不会阻断 Agent 回复。 +- `search_stock_news` 与 `search_comprehensive_intel` 成功返回后会 best-effort 写入 `news_intel`，复用现有 URL / fallback key 去重逻辑。 +- `get_realtime_quote` 不复用 `stock_daily` 作为实时行情缓存，也不会把盘中实时行情写入日线表；如需实时行情缓存，应单独设计实时行情存储。 + ## Portfolio P0 PR3 (Web + Agent Consumption) ### Web consumption page diff --git a/docs/full-guide_EN.md b/docs/full-guide_EN.md index 9774dd444..9bf9d580d 100644 --- a/docs/full-guide_EN.md +++ b/docs/full-guide_EN.md @@ -976,6 +976,14 @@ A: Check if Actions is enabled, and if cron expression is correct (note it's UTC - If upstream FX fetch fails, the page may still remain stale after refresh and will explain the fallback result inline. - When `PORTFOLIO_FX_UPDATE_ENABLED=false`, the refresh API returns an explicit disabled status and the page shows that online FX refresh is disabled instead of implying that no refreshable pairs exist. +## Agent Tool Data Cache And Persistence + +- `get_daily_history` first tries to reuse local `stock_daily` daily-bar cache; when the cache is fresh and contains at least the dashboard default of 30 records, it avoids another external data-source request. +- If Agent asks for more days than the local cache contains, the tool returns the available records and marks the response with `partial_cache=true`, `requested_days`, and `actual_records`. +- When the cache is missing or stale, the tool keeps the original data-source fetch path; successful fetches are written back to `stock_daily` on a best-effort basis, and write failures do not block the Agent response. +- `search_stock_news` and `search_comprehensive_intel` persist successful results to `news_intel` on a best-effort basis, reusing the existing URL / fallback-key deduplication logic. +- `get_realtime_quote` does not use `stock_daily` as a realtime-quote cache and does not write intraday quotes into the daily-bar table; realtime quote caching should use a dedicated realtime store if needed. + --- For more questions, please [submit an Issue](https://github.com/ZhuLinsen/daily_stock_analysis/issues) diff --git a/src/agent/tools/data_tools.py b/src/agent/tools/data_tools.py index 8aa3b25c5..eceb581d6 100644 --- a/src/agent/tools/data_tools.py +++ b/src/agent/tools/data_tools.py @@ -12,7 +12,7 @@ Tools: import logging from datetime import date from threading import Lock -from typing import Optional +from typing import Any, Dict, List, Optional, Tuple from src.agent.tools.registry import ToolParameter, ToolDefinition @@ -20,6 +20,8 @@ logger = logging.getLogger(__name__) _fetcher_manager_singleton = None _fetcher_manager_lock = Lock() +_DAILY_HISTORY_DEFAULT_DAYS = 60 +_DAILY_HISTORY_MAX_DAYS = 365 def _get_fetcher_manager(): @@ -51,6 +53,59 @@ def _get_db(): return get_db() +def _normalize_history_days(days: Any) -> Tuple[int, Dict[str, Any]]: + """Normalize LLM-provided history window and return response metadata.""" + requested_days = days + warning = None + try: + if isinstance(days, bool): + raise ValueError("bool is not a valid days value") + effective_days = int(days) + except (TypeError, ValueError): + effective_days = _DAILY_HISTORY_DEFAULT_DAYS + warning = ( + f"Invalid days value {requested_days!r}; " + f"using default {_DAILY_HISTORY_DEFAULT_DAYS}." + ) + + if effective_days < 1: + effective_days = 1 + warning = f"days must be >= 1; using {effective_days}." + elif effective_days > _DAILY_HISTORY_MAX_DAYS: + effective_days = _DAILY_HISTORY_MAX_DAYS + warning = f"days exceeds max {_DAILY_HISTORY_MAX_DAYS}; truncated." + + metadata: Dict[str, Any] = {} + if warning is not None: + metadata.update( + { + "warning": warning, + "requested_days": requested_days, + "effective_days": effective_days, + } + ) + return effective_days, metadata + + +def _history_code_candidates(stock_code: str) -> Tuple[List[str], str]: + """Return cache lookup candidates plus canonical write code.""" + from data_provider.base import canonical_stock_code, normalize_stock_code + + raw_code = str(stock_code or "").strip() + normalized_code = canonical_stock_code(normalize_stock_code(raw_code)) + candidates: List[str] = [] + for candidate in (canonical_stock_code(raw_code), normalized_code): + if candidate and candidate not in candidates: + candidates.append(candidate) + return candidates, normalized_code + + +def _append_history_metadata(response: dict, metadata: Dict[str, Any]) -> dict: + if metadata: + response.update(metadata) + return response + + def _compact_fundamental_context(fundamental_context: dict) -> dict: """Reduce token footprint for tool responses while keeping key semantics.""" if not isinstance(fundamental_context, dict): @@ -234,25 +289,56 @@ get_realtime_quote_tool = ToolDefinition( def _handle_get_daily_history(stock_code: str, days: int = 60) -> dict: """Get daily OHLCV history data.""" + effective_days, metadata = _normalize_history_days(days) + from src.services.history_loader import load_history_df - df, source = load_history_df(stock_code, days=days) + df, source = load_history_df(stock_code, days=effective_days) if df is None or df.empty: - return {"error": f"No historical data available for {stock_code}"} + return _append_history_metadata( + {"error": f"No historical data available for {stock_code}"}, + metadata, + ) + + if source != "db_cache": + _, normalized_code = _history_code_candidates(stock_code) + try: + saved_count = _get_db().save_daily_data(df, normalized_code, source) + logger.info( + "Agent daily history persisted for %s (source=%s, new_records=%s)", + normalized_code, + source, + saved_count, + ) + except Exception as exc: + logger.warning( + "Agent daily history persistence failed for %s: %s", + normalized_code, + exc, + ) # Convert DataFrame to list of dicts (last N records) - records = df.tail(min(days, len(df))).to_dict(orient="records") + records = df.tail(min(effective_days, len(df))).to_dict(orient="records") # Ensure date is string for r in records: if "date" in r: r["date"] = str(r["date"]) - return { - "code": stock_code, + response_code = stock_code + if source == "db_cache" and records: + response_code = records[-1].get("code") or response_code + + return _append_history_metadata({ + "code": response_code, "source": source, + "cache_hit": source == "db_cache", + "requested_days": effective_days, + "effective_days": effective_days, + "actual_records": len(records), + "partial_cache": source == "db_cache" and len(records) < effective_days, "total_records": len(records), "data": records, - } + }, metadata) get_daily_history_tool = ToolDefinition( diff --git a/src/agent/tools/search_tools.py b/src/agent/tools/search_tools.py index abe7daf8d..848d49283 100644 --- a/src/agent/tools/search_tools.py +++ b/src/agent/tools/search_tools.py @@ -8,19 +8,66 @@ Tools: """ import logging -from typing import Optional from src.agent.tools.registry import ToolParameter, ToolDefinition logger = logging.getLogger(__name__) +def _get_db(): + """Lazy import for DatabaseManager.""" + from src.storage import get_db + return get_db() + + def _get_search_service(): """Return shared SearchService singleton.""" from src.search_service import get_search_service return get_search_service() +def _canonical_search_code(stock_code: str) -> str: + from data_provider.base import canonical_stock_code, normalize_stock_code + + return canonical_stock_code(normalize_stock_code(str(stock_code or "").strip())) + + +def _persist_news_response( + *, + stock_code: str, + stock_name: str, + dimension: str, + response, +) -> None: + """Best-effort news persistence for Agent search tools.""" + if not response or not getattr(response, "success", False) or not getattr(response, "results", None): + return + + code = _canonical_search_code(stock_code) + try: + saved_count = _get_db().save_news_intel( + code=code, + name=stock_name, + dimension=dimension, + query=response.query, + response=response, + query_context=None, + ) + logger.info( + "Agent news intel persisted for %s (dimension=%s, new_records=%s)", + code, + dimension, + saved_count, + ) + except Exception as exc: + logger.warning( + "Agent news intel persistence failed for %s (dimension=%s): %s", + code, + dimension, + exc, + ) + + def _handle_search_stock_news(stock_code: str, stock_name: str) -> dict: """Search latest news for a stock.""" service = _get_search_service() @@ -37,6 +84,13 @@ def _handle_search_stock_news(stock_code: str, stock_name: str) -> dict: "error": response.error_message, } + _persist_news_response( + stock_code=stock_code, + stock_name=stock_name, + dimension="latest_news", + response=response, + ) + return { "query": response.query, "provider": response.provider, @@ -104,6 +158,12 @@ def _handle_search_comprehensive_intel(stock_code: str, stock_name: str) -> dict dimensions = {} for dim_name, response in intel_results.items(): if response and response.success: + _persist_news_response( + stock_code=stock_code, + stock_name=stock_name, + dimension=dim_name, + response=response, + ) dimensions[dim_name] = { "query": response.query, "results_count": len(response.results), diff --git a/src/services/history_loader.py b/src/services/history_loader.py index e189d5651..b4c7b4289 100644 --- a/src/services/history_loader.py +++ b/src/services/history_loader.py @@ -10,13 +10,14 @@ from __future__ import annotations import contextvars import logging -from datetime import date, timedelta +from datetime import date, datetime, timedelta from threading import Lock -from typing import Optional, Tuple +from typing import Any, List, Optional, Tuple import pandas as pd logger = logging.getLogger(__name__) +_CACHE_MIN_RECORDS = 30 # --------------------------------------------------------------------------- # Frozen target date (ContextVar) – set once per stock in pipeline, read by @@ -59,6 +60,69 @@ def _get_fetcher_manager(): # --------------------------------------------------------------------------- # DB-first history loader # --------------------------------------------------------------------------- +def _history_code_candidates(stock_code: str) -> Tuple[List[str], str]: + from data_provider.base import canonical_stock_code, normalize_stock_code + + raw_code = str(stock_code or "").strip() + normalized_code = canonical_stock_code(normalize_stock_code(raw_code)) + candidates: List[str] = [] + for candidate in (canonical_stock_code(raw_code), normalized_code): + if candidate and candidate not in candidates: + candidates.append(candidate) + return candidates, normalized_code + + +def _coerce_bar_date(value: Any) -> date: + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + if isinstance(value, str): + try: + return datetime.strptime(value[:10], "%Y-%m-%d").date() + except ValueError: + return date.min + if hasattr(value, "date"): + try: + coerced = value.date() + return coerced if isinstance(coerced, date) else date.min + except Exception: + return date.min + return date.min + + +def _bar_date(bar: Any) -> date: + row_date = _coerce_bar_date(getattr(bar, "date", None)) + if row_date != date.min: + return row_date + if hasattr(bar, "to_dict"): + try: + return _coerce_bar_date((bar.to_dict() or {}).get("date")) + except Exception: + return date.min + return date.min + + +def _select_best_bars(db, stock_code: str, start: date, end: date) -> Tuple[Optional[str], list]: + candidates, normalized_code = _history_code_candidates(stock_code) + best_code = None + best_bars = [] + best_key = None + + for candidate in candidates: + bars = list(db.get_data_range(candidate, start, end) or []) + if not bars: + continue + latest_date = max(_bar_date(bar) for bar in bars) + key = (latest_date, len(bars), candidate == normalized_code) + if best_key is None or key > best_key: + best_key = key + best_code = candidate + best_bars = bars + + return best_code, best_bars + + def load_history_df( stock_code: str, days: int = 60, @@ -70,7 +134,6 @@ def load_history_df( actual provider name on network fallback. Returns ``(None, "none")`` when both paths fail. """ - from data_provider.base import canonical_stock_code, normalize_stock_code from src.storage import get_db # Resolve effective end date @@ -84,15 +147,12 @@ def load_history_df( start = end - timedelta(days=int(days * 1.8) + 10) # --- 1. DB lookup (canonical code, then prefix-stripped fallback) ------ - code = canonical_stock_code(stock_code) try: db = get_db() - bars = db.get_data_range(code, start, end) - if not bars: - alt = normalize_stock_code(stock_code) - if alt != code: - bars = db.get_data_range(alt, start, end) - if bars and len(bars) >= max(int(days * 0.3), 5): + _code, bars = _select_best_bars(db, stock_code, start, end) + required_records = max(min(days, _CACHE_MIN_RECORDS), 1) + latest_date = max((_bar_date(bar) for bar in bars), default=date.min) + if bars and latest_date >= end and len(bars) >= required_records: df = pd.DataFrame([b.to_dict() for b in bars]) logger.debug( "load_history_df(%s): %d bars from DB (requested %d)", diff --git a/tests/test_data_tools_daily_history_cache.py b/tests/test_data_tools_daily_history_cache.py new file mode 100644 index 000000000..0c47429c4 --- /dev/null +++ b/tests/test_data_tools_daily_history_cache.py @@ -0,0 +1,230 @@ +# -*- coding: utf-8 -*- +"""Tests for Agent get_daily_history DB cache reuse.""" + +from datetime import date, timedelta +from types import SimpleNamespace +from typing import Optional +import unittest +from unittest.mock import MagicMock, patch + +import pandas as pd + +from src.agent.tools.data_tools import _handle_get_daily_history +from src.services.history_loader import reset_frozen_target_date, set_frozen_target_date + + +class _DailyRow: + def __init__(self, code: str, row_date: date, close: float) -> None: + self.code = code + self.date = row_date + self.open = close - 1 + self.high = close + 1 + self.low = close - 2 + self.close = close + self.volume = 1000 + self.amount = 10000 + self.pct_chg = 1.2 + self.ma5 = close - 0.5 + self.ma10 = close - 1 + self.ma20 = close - 2 + self.volume_ratio = 1.1 + self.data_source = "unit-test" + + def to_dict(self): + return { + "code": self.code, + "date": self.date, + "open": self.open, + "high": self.high, + "low": self.low, + "close": self.close, + "volume": self.volume, + "amount": self.amount, + "pct_chg": self.pct_chg, + "ma5": self.ma5, + "ma10": self.ma10, + "ma20": self.ma20, + "volume_ratio": self.volume_ratio, + "data_source": self.data_source, + } + + +def _rows(code: str, latest: date, count: int): + return [ + _DailyRow(code, latest - timedelta(days=offset), close=100 + offset) + for offset in range(count) + ] + + +class _FakeDb: + def __init__(self, rows_by_code=None, save_error: Optional[Exception] = None) -> None: + self.rows_by_code = rows_by_code or {} + self.save_error = save_error + self.save_daily_data = MagicMock(side_effect=self._save_daily_data) + + def get_data_range(self, code: str, start_date: date, end_date: date): + rows = [ + row + for row in self.rows_by_code.get(code, []) + if start_date <= row.date <= end_date + ] + return sorted(rows, key=lambda row: row.date) + + def _save_daily_data(self, df, code: str, source: str): + if self.save_error: + raise self.save_error + return len(df) + + +class DailyHistoryCacheToolTest(unittest.TestCase): + def _run_with_frozen_date(self, target: date, stock_code: str, days: int): + token = set_frozen_target_date(target) + try: + return _handle_get_daily_history(stock_code, days=days) + finally: + reset_frozen_target_date(token) + + def test_uses_fresh_partial_db_cache_without_fetching(self) -> None: + target = date(2026, 4, 24) + db = _FakeDb({"600519": _rows("600519", target, 30)}) + manager = SimpleNamespace(get_daily_data=MagicMock()) + + with patch("src.storage.get_db", return_value=db), \ + patch("src.services.history_loader._get_fetcher_manager", return_value=manager): + result = self._run_with_frozen_date(target, "600519", days=60) + + self.assertEqual(result["source"], "db_cache") + self.assertTrue(result["cache_hit"]) + self.assertTrue(result["partial_cache"]) + self.assertEqual(result["actual_records"], 30) + self.assertEqual(result["requested_days"], 60) + self.assertEqual(result["data"][0]["date"], str(target - timedelta(days=29))) + self.assertEqual(result["data"][-1]["date"], str(target)) + manager.get_daily_data.assert_not_called() + + def test_prefers_fuller_candidate_when_dates_tie(self) -> None: + target = date(2026, 4, 24) + db = _FakeDb( + { + "1810.HK": _rows("1810.HK", target, 40), + "HK01810": _rows("HK01810", target, 30), + } + ) + manager = SimpleNamespace(get_daily_data=MagicMock()) + + with patch("src.storage.get_db", return_value=db), \ + patch("src.services.history_loader._get_fetcher_manager", return_value=manager): + result = self._run_with_frozen_date(target, "1810.HK", days=60) + + self.assertEqual(result["code"], "1810.HK") + self.assertEqual(result["actual_records"], 40) + manager.get_daily_data.assert_not_called() + + def test_prefers_normalized_candidate_when_dates_and_counts_tie(self) -> None: + target = date(2026, 4, 24) + db = _FakeDb( + { + "1810.HK": _rows("1810.HK", target, 30), + "HK01810": _rows("HK01810", target, 30), + } + ) + manager = SimpleNamespace(get_daily_data=MagicMock()) + + with patch("src.storage.get_db", return_value=db), \ + patch("src.services.history_loader._get_fetcher_manager", return_value=manager): + result = self._run_with_frozen_date(target, "1810.HK", days=60) + + self.assertEqual(result["code"], "HK01810") + self.assertEqual(result["actual_records"], 30) + manager.get_daily_data.assert_not_called() + + def test_fetches_and_persists_when_cache_is_stale(self) -> None: + target = date(2026, 4, 24) + db = _FakeDb({"600519": _rows("600519", target - timedelta(days=1), 30)}) + df = pd.DataFrame( + [ + {"date": target, "open": 1, "high": 2, "low": 0.5, "close": 1.5}, + ] + ) + manager = SimpleNamespace(get_daily_data=MagicMock(return_value=(df, "Fetcher"))) + + with patch("src.storage.get_db", return_value=db), \ + patch("src.agent.tools.data_tools._get_db", return_value=db), \ + patch("src.services.history_loader._get_fetcher_manager", return_value=manager): + result = self._run_with_frozen_date(target, "600519", days=60) + + manager.get_daily_data.assert_called_once_with("600519", days=60) + db.save_daily_data.assert_called_once_with(df, "600519", "Fetcher") + self.assertFalse(result["cache_hit"]) + self.assertEqual(result["source"], "Fetcher") + + def test_save_failure_does_not_hide_fetched_data(self) -> None: + target = date(2026, 4, 24) + db = _FakeDb(save_error=RuntimeError("db locked")) + df = pd.DataFrame( + [ + {"date": target, "open": 1, "high": 2, "low": 0.5, "close": 1.5}, + ] + ) + manager = SimpleNamespace(get_daily_data=MagicMock(return_value=(df, "Fetcher"))) + + with patch("src.storage.get_db", return_value=db), \ + patch("src.agent.tools.data_tools._get_db", return_value=db), \ + patch("src.services.history_loader._get_fetcher_manager", return_value=manager): + result = self._run_with_frozen_date(target, "600519", days=60) + + self.assertEqual(result["total_records"], 1) + self.assertEqual(result["data"][0]["date"], str(target)) + + def test_db_read_exception_falls_back_to_fetch(self) -> None: + target = date(2026, 4, 24) + df = pd.DataFrame( + [{"date": target, "open": 1, "high": 2, "low": 0.5, "close": 1.5}] + ) + manager = SimpleNamespace(get_daily_data=MagicMock(return_value=(df, "Fetcher"))) + broken_db = MagicMock() + broken_db.get_data_range.side_effect = RuntimeError("db corrupted") + broken_db.save_daily_data.return_value = 1 + + with patch("src.storage.get_db", return_value=broken_db), \ + patch("src.agent.tools.data_tools._get_db", return_value=broken_db), \ + patch("src.services.history_loader._get_fetcher_manager", return_value=manager): + result = self._run_with_frozen_date(target, "600519", days=60) + + manager.get_daily_data.assert_called_once_with("600519", days=60) + self.assertFalse(result["cache_hit"]) + self.assertEqual(result["source"], "Fetcher") + + def test_days_one_cache_hit_with_single_fresh_record(self) -> None: + target = date(2026, 4, 24) + db = _FakeDb({"600519": _rows("600519", target, 1)}) + manager = SimpleNamespace(get_daily_data=MagicMock()) + + with patch("src.storage.get_db", return_value=db), \ + patch("src.services.history_loader._get_fetcher_manager", return_value=manager): + result = self._run_with_frozen_date(target, "600519", days=1) + + self.assertTrue(result["cache_hit"]) + self.assertEqual(result["actual_records"], 1) + self.assertFalse(result["partial_cache"]) + manager.get_daily_data.assert_not_called() + + def test_days_are_normalized_with_warning(self) -> None: + target = date(2026, 4, 24) + db = _FakeDb() + df = pd.DataFrame([{"date": target, "close": 1.5}]) + manager = SimpleNamespace(get_daily_data=MagicMock(return_value=(df, "Fetcher"))) + + with patch("src.storage.get_db", return_value=db), \ + patch("src.agent.tools.data_tools._get_db", return_value=db), \ + patch("src.services.history_loader._get_fetcher_manager", return_value=manager): + result = self._run_with_frozen_date(target, "600519", days=999) + + manager.get_daily_data.assert_called_once_with("600519", days=365) + self.assertEqual(result["requested_days"], 999) + self.assertEqual(result["effective_days"], 365) + self.assertIn("warning", result) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_search_tools_persistence.py b/tests/test_search_tools_persistence.py new file mode 100644 index 000000000..3eb23d6a5 --- /dev/null +++ b/tests/test_search_tools_persistence.py @@ -0,0 +1,121 @@ +# -*- coding: utf-8 -*- +"""Tests for Agent search tool news persistence.""" + +import unittest +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from src.agent.tools.search_tools import ( + _handle_search_comprehensive_intel, + _handle_search_stock_news, +) +from src.search_service import SearchResponse, SearchResult + + +def _response(query: str, *, success: bool = True) -> SearchResponse: + return SearchResponse( + query=query, + provider="UnitSearch", + success=success, + error_message=None if success else "search failed", + results=[ + SearchResult( + title="新闻标题", + snippet="新闻摘要", + url="https://example.com/news", + source="example.com", + published_date="2026-04-24", + ) + ] if success else [], + ) + + +class SearchToolsPersistenceTest(unittest.TestCase): + def test_search_stock_news_persists_successful_response(self) -> None: + response = _response("贵州茅台 600519 latest news") + service = SimpleNamespace( + is_available=True, + search_stock_news=MagicMock(return_value=response), + ) + db = SimpleNamespace(save_news_intel=MagicMock(return_value=1)) + + with patch("src.agent.tools.search_tools._get_search_service", return_value=service), \ + patch("src.agent.tools.search_tools._get_db", return_value=db): + result = _handle_search_stock_news("600519", "贵州茅台") + + self.assertTrue(result["success"]) + db.save_news_intel.assert_called_once_with( + code="600519", + name="贵州茅台", + dimension="latest_news", + query=response.query, + response=response, + query_context=None, + ) + + def test_search_comprehensive_intel_persists_successful_dimensions_only(self) -> None: + latest = _response("latest") + failed = _response("risk", success=False) + service = SimpleNamespace( + is_available=True, + search_comprehensive_intel=MagicMock( + return_value={"latest_news": latest, "risk_check": failed} + ), + format_intel_report=MagicMock(return_value="report"), + ) + db = SimpleNamespace(save_news_intel=MagicMock(return_value=1)) + + with patch("src.agent.tools.search_tools._get_search_service", return_value=service), \ + patch("src.agent.tools.search_tools._get_db", return_value=db): + result = _handle_search_comprehensive_intel("600519", "贵州茅台") + + self.assertEqual(result["report"], "report") + self.assertEqual(list(result["dimensions"].keys()), ["latest_news"]) + db.save_news_intel.assert_called_once_with( + code="600519", + name="贵州茅台", + dimension="latest_news", + query=latest.query, + response=latest, + query_context=None, + ) + + def test_persistence_failure_keeps_search_result(self) -> None: + response = _response("贵州茅台 600519 latest news") + service = SimpleNamespace( + is_available=True, + search_stock_news=MagicMock(return_value=response), + ) + db = SimpleNamespace(save_news_intel=MagicMock(side_effect=RuntimeError("db locked"))) + + with patch("src.agent.tools.search_tools._get_search_service", return_value=service), \ + patch("src.agent.tools.search_tools._get_db", return_value=db): + result = _handle_search_stock_news("600519", "贵州茅台") + + self.assertTrue(result["success"]) + self.assertEqual(result["results_count"], 1) + + def test_unavailable_or_failed_search_does_not_persist(self) -> None: + unavailable = SimpleNamespace(is_available=False) + db = SimpleNamespace(save_news_intel=MagicMock()) + with patch("src.agent.tools.search_tools._get_search_service", return_value=unavailable), \ + patch("src.agent.tools.search_tools._get_db", return_value=db): + result = _handle_search_stock_news("600519", "贵州茅台") + + self.assertIn("error", result) + db.save_news_intel.assert_not_called() + + failed = SimpleNamespace( + is_available=True, + search_stock_news=MagicMock(return_value=_response("latest", success=False)), + ) + with patch("src.agent.tools.search_tools._get_search_service", return_value=failed), \ + patch("src.agent.tools.search_tools._get_db", return_value=db): + result = _handle_search_stock_news("600519", "贵州茅台") + + self.assertFalse(result["success"]) + db.save_news_intel.assert_not_called() + + +if __name__ == "__main__": + unittest.main()