diff --git a/data_provider/akshare_fetcher.py b/data_provider/akshare_fetcher.py index 06704eba1..4b3faa5a8 100644 --- a/data_provider/akshare_fetcher.py +++ b/data_provider/akshare_fetcher.py @@ -249,8 +249,20 @@ def _is_us_code(stock_code: str) -> bool: def _to_sina_tx_symbol(stock_code: str) -> str: - """Convert 6-digit A-share code to sh/sz/bj prefixed symbol for Sina/Tencent APIs.""" - base = (stock_code.strip().split(".")[0] if "." in stock_code else stock_code).strip() + """Convert 6-digit A-share code to sh/sz/bj prefixed symbol for Sina/Tencent APIs. + + Explicit sh/sz/bj prefixes are preserved (``sh000016`` -> ``sh000016``) so + registered index codes keep their index identity instead of degrading into + the colliding stock symbol (Story 1.5). + """ + raw = (stock_code or "").strip() + lower = raw.lower() + for prefix in ("sh", "sz", "bj"): + if lower.startswith(prefix) and len(raw) > len(prefix): + candidate = raw[len(prefix):] + if candidate.isdigit() and len(candidate) == 6: + return f"{prefix}{candidate}" + base = (raw.split(".")[0] if "." in raw else raw).strip() if is_bse_code(base): return f"bj{base}" # Shanghai: 60xxxx, 5xxxx (ETF), 90xxxx (B-shares) diff --git a/data_provider/base.py b/data_provider/base.py index 9237af149..ba5e28e32 100644 --- a/data_provider/base.py +++ b/data_provider/base.py @@ -637,6 +637,12 @@ class DataFetcherManager: "TickFlowFetcher", "YfinanceFetcher", ) + _CN_INDEX_REALTIME_SOURCE_ORDER = ( + ("AkshareFetcher", "tencent"), + ("AkshareFetcher", "sina"), + ("EfinanceFetcher", "index"), + ("TickFlowFetcher", "tickflow"), + ) _CN_INDEX_NAME_SOURCE_ORDER = ( "TencentFetcher", "AkshareFetcher", @@ -1139,6 +1145,236 @@ class DataFetcherManager: ) return pd.DataFrame(columns=STANDARD_COLUMNS), "" + def _get_cn_index_realtime_quote( + self, + target: AnalysisTarget, + *, + log_final_failure: bool = True, + ): + """Fetch a realtime quote for a registered CN index via a fixed chain. + + Chain (Story 1.5): + 1. Tencent via AkshareFetcher (``sh000016``/``sz399001`` prefixed symbol) + 2. Sina via AkshareFetcher (same prefixed symbol) + 3. Eastmoney single-stock secid via EfinanceFetcher (SH/SZ/CSI) + 4. TickFlow (``000016.SH`` symbol; SH/SZ only) + + The explicit index identity is preserved end-to-end: provider symbols + are derived from the registry entry, never from ``normalize_stock_code``, + so ``sh000016`` can never degrade into the stock ``000016`` path. + """ + fetchers_by_name = { + fetcher.name: fetcher for fetcher in self._get_fetchers_snapshot() + } + errors: List[str] = [] + request_start = time.time() + source_order = self._CN_INDEX_REALTIME_SOURCE_ORDER + + for index, (source_name, source_kind) in enumerate(source_order): + fallback_to = ( + source_order[index + 1][0] if index + 1 < len(source_order) else None + ) + fetcher = fetchers_by_name.get(source_name) + provider_symbol = self._cn_index_realtime_provider_symbol( + target, source_name, source_kind + ) + + if not provider_symbol: + reason = ( + "unsupported index provider symbol: " + f"{target.canonical_id} -> {source_name}" + ) + record_provider_run( + data_type="realtime_quote", + provider=source_name, + operation="get_realtime_quote", + success=False, + latency_ms=0, + error_type="unsupported", + error_message=reason, + fallback_to=fallback_to, + record_count=0, + ) + logger.warning( + "[指数实时行情不支持 %d/%d] [%s] %s: %s", + index + 1, + len(source_order), + source_name, + target.canonical_id, + reason, + ) + errors.append(f"[{source_name}] {reason}") + continue + + if fetcher is None or not self._is_fetcher_available( + fetcher, capability="realtime_quote" + ): + reason = "数据源未配置或暂不可用" + record_provider_run( + data_type="realtime_quote", + provider=source_name, + operation="get_realtime_quote", + success=False, + latency_ms=0, + error_type="unavailable", + error_message=reason, + fallback_to=fallback_to, + record_count=0, + ) + logger.warning( + "[指数实时行情失败 %d/%d] [%s] %s: %s", + index + 1, + len(source_order), + source_name, + target.canonical_id, + reason, + ) + errors.append(f"[{source_name}] {reason}") + continue + + attempt_start = time.time() + try: + logger.info( + "[指数实时行情尝试 %d/%d] [%s] %s -> %s", + index + 1, + len(source_order), + source_name, + target.canonical_id, + provider_symbol, + ) + record_provider_run_started( + data_type="realtime_quote", + provider=source_name, + operation="get_realtime_quote", + ) + if source_kind == "index": + quote = self._call_fetcher_method( + fetcher, + "get_index_realtime_quote", + target.canonical_id, + ) + elif source_kind == "tickflow": + quote = self._call_fetcher_method( + fetcher, + "get_realtime_quote", + provider_symbol, + ) + else: + quote = self._call_fetcher_method( + fetcher, + "get_realtime_quote", + provider_symbol, + source=source_kind, + ) + duration_ms = int((time.time() - attempt_start) * 1000) + if quote is not None and quote.has_basic_data(): + record_provider_run( + data_type="realtime_quote", + provider=source_name, + operation="get_realtime_quote", + success=True, + latency_ms=duration_ms, + record_count=1, + ) + logger.info( + "[指数实时行情完成] %s 使用 [%s] 获取成功: elapsed=%.2fs", + target.canonical_id, + source_name, + time.time() - request_start, + ) + from src.config import get_config as _get_config_safe + + return self._enrich_realtime_quote( + quote, + realtime_cache_ttl=getattr( + _get_config_safe(), "realtime_cache_ttl", None + ), + ) + + reason = "empty or incomplete quote" + record_provider_run( + data_type="realtime_quote", + provider=source_name, + operation="get_realtime_quote", + success=False, + latency_ms=duration_ms, + error_type="empty", + error_message=reason, + fallback_to=fallback_to, + record_count=0, + ) + logger.warning( + "[指数实时行情失败 %d/%d] [%s] %s: %s", + index + 1, + len(source_order), + source_name, + target.canonical_id, + reason, + ) + errors.append(f"[{source_name}] {reason}") + except Exception as exc: + error_type, error_reason = summarize_exception(exc) + duration_ms = int((time.time() - attempt_start) * 1000) + record_provider_run( + data_type="realtime_quote", + provider=source_name, + operation="get_realtime_quote", + success=False, + latency_ms=duration_ms, + error_type=error_type, + error_message=error_reason, + fallback_to=fallback_to, + record_count=0, + ) + logger.warning( + "[指数实时行情失败 %d/%d] [%s] %s: error_type=%s, reason=%s", + index + 1, + len(source_order), + source_name, + target.canonical_id, + error_type, + error_reason, + ) + errors.append(f"[{source_name}] ({error_type}) {error_reason}") + + if log_final_failure: + logger.warning( + "[指数实时行情终止] %s 所有指数实时行情数据源均失败: elapsed=%.2fs; %s", + target.canonical_id, + time.time() - request_start, + "; ".join(errors) or "暂无可用数据源", + ) + return None + + @classmethod + def _cn_index_realtime_provider_symbol( + cls, + target: AnalysisTarget, + fetcher_name: str, + source_kind: str, + ) -> str: + """Derive the provider symbol for the CN index realtime chain.""" + entry = target.matched_index + if entry is None: + return "" + exchange = entry.exchange.upper() + if exchange == "CSI": + # CSI indices are only supported by the Eastmoney single-stock + # secid endpoint (EfinanceFetcher); other providers return an + # empty symbol so the caller records ``unsupported`` and skips. + if fetcher_name == "EfinanceFetcher" and source_kind == "index": + return target.canonical_id + return "" + if exchange not in {"SH", "SZ"}: + return "" + if fetcher_name == "AkshareFetcher" and source_kind in ("tencent", "sina"): + return f"{exchange.lower()}{entry.bare_code}" + if fetcher_name == "EfinanceFetcher" and source_kind == "index": + return target.canonical_id + if fetcher_name == "TickFlowFetcher" and source_kind == "tickflow": + return f"{entry.bare_code}.{exchange}" + return "" + def _get_cn_index_name(self, target: AnalysisTarget) -> str: cache_key = target.canonical_id entry = target.matched_index @@ -1929,8 +2165,17 @@ class DataFetcherManager: Returns: 预取的股票数量（0 表示跳过预取） """ - # Normalize all codes - stock_codes = [normalize_stock_code(c) for c in stock_codes] + # Normalize all codes, preserving explicit index identities + # (``sh000016`` / ``csi930955``) so the prefetch never degrades an + # index into the colliding stock bucket (Story 1.5). + normalized_codes: List[str] = [] + for code in stock_codes: + target = parse_analysis_target(code) + if target.asset_type == ParseStatus.INDEX: + normalized_codes.append(target.canonical_id) + else: + normalized_codes.append(normalize_stock_code(code)) + stock_codes = normalized_codes from src.config import get_config @@ -2169,6 +2414,17 @@ class DataFetcherManager: logger.debug(f"[实时行情] 功能已禁用，跳过 {stock_code}") return None + # ---------------------------------------------------------- + # 已登记 A 股指数 — 固定实时行情链（Story 1.5） + # 显式指数身份在 normalize 前解析，避免 sh000016 被剥成 + # 股票 000016 后误取同名股票行情。 + # ---------------------------------------------------------- + index_target = parse_analysis_target(raw_stock_code) + if index_target.asset_type == ParseStatus.INDEX: + return self._get_cn_index_realtime_quote( + index_target, log_final_failure=log_final_failure + ) + # ---------------------------------------------------------- # 美股 (指数 + 个股) / 港股 — 专用双源路由 # 配置长桥后: Longbridge 首选, YFinance/AkShare 补充 @@ -3905,6 +4161,12 @@ class DataFetcherManager: **blocks, } + def build_not_supported_fundamental_context( + self, stock_code: str, reason: str + ) -> Dict[str, Any]: + """Build a consistent not-supported payload without calling providers.""" + return self._build_market_not_supported(_market_tag(stock_code), reason) + def get_fundamental_context( self, stock_code: str, diff --git a/data_provider/efinance_fetcher.py b/data_provider/efinance_fetcher.py index ae760b6e7..1c4bbfd66 100644 --- a/data_provider/efinance_fetcher.py +++ b/data_provider/efinance_fetcher.py @@ -53,6 +53,7 @@ except (ValueError, TypeError): from src.patches.eastmoney_patch import eastmoney_patch from src.config import get_config +from src.services.stock_list_parser import ParseStatus, parse_analysis_target from .base import ( BaseFetcher, DataFetchError, @@ -626,6 +627,7 @@ class EfinanceFetcher(BaseFetcher): 数据来源：ef.stock.get_realtime_quotes() ETF 数据源：ef.stock.get_realtime_quotes(['ETF']) + 已登记指数：东财单股 secid 接口（get_index_realtime_quote） Args: stock_code: 股票代码 @@ -633,6 +635,11 @@ class EfinanceFetcher(BaseFetcher): Returns: UnifiedRealtimeQuote 对象，获取失败返回 None """ + # 已登记指数走东财单股 secid 接口（Story 1.5） + target = parse_analysis_target(stock_code) + if target.asset_type == ParseStatus.INDEX: + return self.get_index_realtime_quote(target.canonical_id) + # ETF 需要单独请求 ETF 实时行情接口 if _is_etf_code(stock_code): return self._get_etf_realtime_quote(stock_code) @@ -739,6 +746,87 @@ class EfinanceFetcher(BaseFetcher): circuit_breaker.record_failure(source_key, str(e)) return None + def get_index_realtime_quote( + self, stock_code: str + ) -> Optional[UnifiedRealtimeQuote]: + """Fetch a registered CN index quote via the Eastmoney single-stock secid API. + + Supports SH (``1.{code}``), SZ (``0.{code}``) and CSI (``2.{code}``) + secid forms. This is the only realtime source for CSI indices and the + fallback for SH/SZ indices when Tencent/Sina fail (Story 1.5). + """ + target = parse_analysis_target(stock_code) + if target.asset_type != ParseStatus.INDEX or target.matched_index is None: + return None + entry = target.matched_index + exchange = entry.exchange.upper() + if exchange == "SH": + secid = f"1.{entry.bare_code}" + elif exchange == "SZ": + secid = f"0.{entry.bare_code}" + elif exchange == "CSI": + secid = f"2.{entry.bare_code}" + else: + return None + + circuit_breaker = get_realtime_circuit_breaker() + source_key = "efinance_index" + if not circuit_breaker.is_available(source_key): + logger.info(f"[熔断] 数据源 {source_key} 处于熔断状态，跳过") + return None + + try: + self._set_random_user_agent() + self._enforce_rate_limit() + response = requests.get( + "https://push2.eastmoney.com/api/qt/stock/get", + params={ + "secid": secid, + "fltt": "2", + "fields": "f43,f44,f45,f46,f47,f48,f57,f58,f60,f168,f169,f170,f171", + }, + timeout=10, + ) + if response.status_code != 200: + circuit_breaker.record_failure(source_key, f"HTTP {response.status_code}") + return None + payload = response.json() + data = payload.get("data") or {} + price = safe_float(data.get("f43")) + if price is None or price <= 0: + circuit_breaker.record_failure(source_key, "empty quote payload") + return None + circuit_breaker.record_success(source_key) + quote = UnifiedRealtimeQuote( + code=target.canonical_id, + name=str(data.get("f58") or entry.display_name or ""), + source=RealtimeSource.EFINANCE, + price=price, + change_pct=safe_float(data.get("f170")), + change_amount=safe_float(data.get("f169")), + volume=safe_int(data.get("f47")), + amount=safe_float(data.get("f48")), + amplitude=safe_float(data.get("f171")), + high=safe_float(data.get("f44")), + low=safe_float(data.get("f45")), + open_price=safe_float(data.get("f46")), + pre_close=safe_float(data.get("f60")), + volume_ratio=safe_float(data.get("f168")), + ) + logger.info( + "[实时行情-东财指数] %s %s: 价格=%s, 涨跌=%s%%, secid=%s", + target.canonical_id, + quote.name, + quote.price, + quote.change_pct, + secid, + ) + return quote + except Exception as e: + logger.info(f"[API错误] 获取 {target.canonical_id} 指数实时行情(东财)失败: {e}") + circuit_breaker.record_failure(source_key, str(e)) + return None + def _get_etf_realtime_quote(self, stock_code: str) -> Optional[UnifiedRealtimeQuote]: """ 获取 ETF 实时行情 diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 3764390d6..91e61c714 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +- [新功能] 支持通过 `main.py --stocks` 一次性分析已登记板块指数，自动使用指数适用的数据与分析能力，并保持报告、历史和决策信号兼容。 +- [修复] `main.py --stocks` 在解析股票列表前先 best-effort 刷新股票索引注册表，保证首次运行能吃到刷新后的指数 alias/身份；刷新失败、超时或禁用不阻断分析。 +- [修复] 交易日过滤对市场未知的指数 code（如 `sh000016`/`csi930955`/`930955.CSI`）按 `market=cn` 参与 A 股休市过滤，避免休市日指数被 fail-open 保留；市场仍未知的非指数 code 继续保留。 +- [修复] 指数分析将实际命中的日线数据源归因保存到历史记录，并由 Dashboard/Brief aggregate 报告展示；来源无效时保持原有输出。 - [文档] 在中英繁 README 顶部关联 DSA arXiv 论文，并新增 `CITATION.cff` 统一项目引用信息。 - [改进] PR CI 增加文档路径检测：仅修改普通文档、非治理 Markdown 或 LICENSE 时跳过后端测试分片、Docker、Web 与桌面打包，保留轻量治理和门禁汇总；契约文档、静态 API 规格与测试 fixture 仍执行后端回归。 - [修复] Linux/Docker 分享图补齐 Noto CJK 字体与中韩文字体栈，避免 PNG 只显示数字和英文、中文或韩文内容消失。 diff --git a/docs/full-guide.md b/docs/full-guide.md index 40892a8a9..330d5ce6c 100644 --- a/docs/full-guide.md +++ b/docs/full-guide.md @@ -717,6 +717,27 @@ python main.py --debug # 调试模式（详细日志） python main.py --workers 5 # 指定并发数 ``` +### 指数一次性分析（Phase 1） + +`--stocks` 一次性入口支持对已登记的 SH、SZ 与 CSI 指数执行完整分析。指数目标使用 `sh`/`sz` 前缀（如 `sh000016`）或 `.CSI` alias（如 `000300.CSI`、`930955.CSI`）显式指定；裸六码（如 `000016`）始终按股票处理。 + +```bash +# 同批分析上证50、沪深300、红利低波100 三个指数 +python main.py --stocks sh000016,000300.CSI,930955.CSI +# 指数与普通股票同批分析（000016 按股票身份执行） +python main.py --stocks sh000016,000016 +# 仅获取指数数据，不执行 AI 分析 +python main.py --stocks sh000016 --dry-run +``` + +指数目标在 Pipeline 内以 `market=cn` 统一处理市场阶段、日线目标日期、断点续传日期、历史窗口与 `DecisionSignal`；筹码分布、基本面、板块归属、资金流、龙虎榜与公司事件等个股专属模块会被集中跳过。未登记的 `.CSI` 输入（如 `930956.CSI`）会在任何行情数据 provider 请求前明确拒绝，且不影响同批其他目标。搜索与报告使用注册表中文指数名称，不携带机器码。 + +指数与 A 股共享交易日语义：启用交易日检查时，已登记指数（`sh`/`sz` 前缀或 `.CSI` alias）按 `market=cn` 参与 A 股休市过滤，A 股休市日指数会被跳过；市场仍无法识别的非指数 code 保持既有 fail-open 行为。`--force-run` 可强制在非交易日执行。 + +指数实时行情使用独立固定链：腾讯 → 新浪 → 东财单股接口 → TickFlow；SH/SZ 指数按 `sh000016`/`sz399001` 显式符号请求，CSI 指数仅由东财单股接口提供（`2.{code}` secid）。显式指数身份全程保留，不会退化为同码股票行情。 + +> **Phase 2 边界**：默认 `STOCK_LIST`、`--schedule`、Web/API 自动补全与分析入口、Bot 与 GitHub Actions 每日工作流暂不开放指数入口；本能力仅通过一次性 `--stocks` 提供。 + ### Futu 真实持仓作为分析列表 标准源码安装（`pip install -r requirements.txt`）、官方 Docker 镜像和 Windows/macOS Desktop backend 已默认包含锁定的 `futu-api==10.8.6808`。仅在使用裁剪过的自定义 Python 环境时，才需要按 [Futu OpenAPI SDK 安装说明](https://openapi.futunn.com/futu-api-doc/en/intro/intro.html) 手动补装。启动并登录 Futu OpenD 后运行： diff --git a/docs/full-guide_EN.md b/docs/full-guide_EN.md index fabc786d3..f5e8587a3 100644 --- a/docs/full-guide_EN.md +++ b/docs/full-guide_EN.md @@ -657,6 +657,27 @@ python main.py --debug # Debug mode (verbose logging) python main.py --workers 5 # Specify concurrency ``` +### One-shot index analysis (Phase 1) + +The one-shot `--stocks` entry supports full analysis of registered SH, SZ, and CSI indices. Index targets are specified explicitly with an `sh`/`sz` prefix (e.g. `sh000016`) or a `.CSI` alias (e.g. `000300.CSI`, `930955.CSI`); a bare six-digit code (e.g. `000016`) is always treated as a stock. + +```bash +# Analyze three indices in one batch: SSE 50, CSI 300, CSI Dividend Low Vol 100 +python main.py --stocks sh000016,000300.CSI,930955.CSI +# Mix indices and stocks in one batch (000016 retains stock identity) +python main.py --stocks sh000016,000016 +# Fetch index data only, no AI analysis +python main.py --stocks sh000016 --dry-run +``` + +Index targets are handled with `market=cn` throughout the Pipeline for market phase, daily-bar target date, resume/checkpoint date, history window, and `DecisionSignal`. Stock-only modules (chip distribution, fundamentals, board membership, capital flow, LHB, corporate events) are centrally skipped. An unregistered `.CSI` input (e.g. `930956.CSI`) is rejected before any market-data provider request without affecting other targets in the batch. Search and reports use the registry Chinese index name and never carry machine codes. + +Indices share the A-share trading-day semantics: when the trading-day check is enabled, registered indices (`sh`/`sz` prefix or `.CSI` alias) participate in CN holiday filtering as `market=cn`, so indices are skipped on A-share holidays; a market-unknown non-index code keeps the existing fail-open behavior. `--force-run` forces execution on non-trading days. + +Index realtime quotes use a dedicated fixed chain: Tencent → Sina → Eastmoney single-stock endpoint → TickFlow. SH/SZ indices are requested with explicit symbols (`sh000016`/`sz399001`); CSI indices are served only by the Eastmoney single-stock endpoint (`2.{code}` secid). The explicit index identity is preserved end-to-end and never degrades into the colliding stock quote. + +> **Phase 2 boundary**: default `STOCK_LIST`, `--schedule`, Web/API autocomplete and analysis entrypoints, Bot, and the GitHub Actions daily workflow do not yet expose index entrypoints; this capability is available only through the one-shot `--stocks` entry. + ### Use real Futu holdings as the analysis list Standard source installs (`pip install -r requirements.txt`), official Docker images, and Windows/macOS Desktop backends already include the pinned `futu-api==10.8.6808`. Install it manually from the [Futu OpenAPI SDK guide](https://openapi.futunn.com/futu-api-doc/en/intro/intro.html) only when using a reduced custom Python environment. After starting and signing in to Futu OpenD, run: diff --git a/main.py b/main.py index 3fbbae3ea..b8cf6d342 100644 --- a/main.py +++ b/main.py @@ -74,7 +74,12 @@ from src.config import get_config, Config from src.logging_config import setup_logging from src.brokers.futu.portfolio import FutuPortfolioError from data_provider.base import canonical_stock_code -from src.services.stock_list_parser import split_stock_list +from src.services.stock_list_parser import ( + AnalysisTarget, + ParseStatus, + parse_analysis_target, + split_stock_list, +) from src.services.stock_code_utils import resolve_index_stock_code_for_analysis @@ -280,6 +285,7 @@ def parse_arguments() -> argparse.Namespace: python main.py --debug # 调试模式 python main.py --dry-run # 仅获取数据，不进行 AI 分析 python main.py --stocks 600519,000001 # 指定分析特定股票 + python main.py --stocks sh000016,000300.CSI,930955.CSI # 指定分析已登记指数（sh/sz 前缀或 .CSI alias） python main.py --portfolio futu # 使用 Futu 真实正股持仓（覆盖 --stocks） python main.py --no-notify # 不发送推送通知 python main.py --check-notify # 检查通知配置，不发送通知 @@ -304,7 +310,7 @@ def parse_arguments() -> argparse.Namespace: parser.add_argument( '--stocks', type=str, - help='指定要分析的股票代码，逗号分隔（覆盖配置文件）' + help='指定要分析的股票代码，逗号分隔（覆盖配置文件）；支持已登记指数：sh/sz 前缀（如 sh000016）或 .CSI alias（如 000300.CSI、930955.CSI）' ) parser.add_argument( @@ -471,6 +477,15 @@ def _compute_trading_day_filter( filtered_codes = [] for code in stock_codes: mkt = get_market_for_stock(code) + if mkt is None: + # 指数 code（如 sh000016/csi930955/930955.CSI）的 get_market_for_stock + # 返回 None，若直接 fail-open 保留，A 股休市日指数不会被过滤，破坏 + # per-stock 交易日契约。对市场未知的 code 用 parse_analysis_target + # 判型，已登记指数按 market=cn 参与 CN 交易日过滤；仍未知的非指数 + # code 继续 fail-open 保留。 + target = parse_analysis_target(code) + if target.asset_type == ParseStatus.INDEX: + mkt = "cn" if mkt in open_markets or mkt is None: filtered_codes.append(code) @@ -722,12 +737,15 @@ def run_full_analysis( stock_codes: Optional[List[str]] = None, *, raise_errors: bool = False, + analysis_targets: Optional[List[AnalysisTarget]] = None, ) -> bool: """ 执行完整的分析流程（个股 + 大盘复盘） 这是定时任务调用的主函数。Futu 持仓解析失败始终传播给调用方； ``raise_errors`` 只控制持仓解析成功后的分析流程异常语义。 + ``analysis_targets`` 与 ``stock_codes`` 对齐，携带结构化分析目标 + （指数目标用于推导 market=cn 与能力矩阵）。 """ # Portfolio resolution is its own CLI contract boundary. A broker import # failure must reach the one-shot caller, while all later work keeps the @@ -762,6 +780,7 @@ def run_full_analysis( _refresh_stock_index_cache_for_analysis(config) if portfolio_stock_codes is not None: stock_codes = portfolio_stock_codes + analysis_targets = None # Issue #529: Hot-reload STOCK_LIST from .env on each scheduled run if stock_codes is None and portfolio_stock_codes is None: @@ -802,6 +821,18 @@ def run_full_analysis( if set(filtered_codes) != set(effective_codes): skipped = set(effective_codes) - set(filtered_codes) logger.info("今日休市股票已跳过: %s", skipped) + if analysis_targets is not None: + if len(analysis_targets) != len(effective_codes): + raise ValueError("analysis_targets must align with stock_codes") + remaining_pairs = list(zip(effective_codes, analysis_targets)) + filtered_targets = [] + for filtered_code in filtered_codes: + for index, (code, target) in enumerate(remaining_pairs): + if code == filtered_code: + filtered_targets.append(target) + remaining_pairs.pop(index) + break + analysis_targets = filtered_targets stock_codes = filtered_codes skip_futu_stock_analysis = ( portfolio_stock_codes is not None and not stock_codes @@ -909,6 +940,7 @@ def run_full_analysis( send_notification=not args.no_notify, merge_notification=merge_notification, current_time=analysis_reference_time, + analysis_targets=analysis_targets, ) if should_use_daily_market_context and not market_context_summary: @@ -1152,6 +1184,7 @@ def _run_analysis_with_runtime_scheduler_lock( config: Config, args: argparse.Namespace, stock_codes: Optional[List[str]] = None, + analysis_targets: Optional[List[AnalysisTarget]] = None, ) -> bool: from src.services.runtime_scheduler import run_with_global_analysis_lock @@ -1162,7 +1195,15 @@ def _run_analysis_with_runtime_scheduler_lock( locked_args: argparse.Namespace, locked_stock_codes: Optional[List[str]] = None, ) -> bool: - result = run_full_analysis(locked_config, locked_args, locked_stock_codes) + if analysis_targets is None: + result = run_full_analysis(locked_config, locked_args, locked_stock_codes) + else: + result = run_full_analysis( + locked_config, + locked_args, + locked_stock_codes, + analysis_targets=analysis_targets, + ) task_result["ok"] = bool(result) return task_result["ok"] @@ -1456,13 +1497,27 @@ def main() -> int: return 0 if result.ok else 1 # 解析股票列表（统一为大写 Issue #355） + # Story 1.5: 一次性 --stocks 入口使用 parse_analysis_target 构造结构化 + # AnalysisTarget 列表，指数目标（sh/sz/csi 前缀与 .CSI alias）在入口即保留 + # 身份语义；unsupported 目标在 Pipeline 内于 provider 调用前拒绝。 stock_codes = None + analysis_targets = None if args.stocks: + # 在解析 --stocks 前先 best-effort 刷新股票索引注册表，保证首次运行能吃到 + # 刷新后的 alias/身份；失败/超时/禁用不阻断分析。仅 --stocks 入口需要， + # 其他模式由 run_full_analysis 内的既有刷新覆盖。 + _refresh_stock_index_cache_for_analysis(config) + tokens = [c for c in split_stock_list(args.stocks) if (c or "").strip()] + targets = [parse_analysis_target(t) for t in tokens] + # 指数目标使用 parser canonical；非指数目标沿用既有 + # resolve_index_stock_code_for_analysis 语义（保留 JP/KR 等解析行为）。 stock_codes = [ - resolve_index_stock_code_for_analysis(c) - for c in split_stock_list(args.stocks) - if (c or "").strip() + t.canonical_id + if t.asset_type == ParseStatus.INDEX + else resolve_index_stock_code_for_analysis(raw) + for t, raw in zip(targets, tokens) ] + analysis_targets = targets logger.info(f"使用命令行指定的股票列表: {stock_codes}") if getattr(args, "portfolio", None): logger.info("同时指定了 --portfolio；实际分析时 portfolio 将覆盖 --stocks") @@ -1686,7 +1741,9 @@ def main() -> int: # 模式3: 正常单次运行 if config.run_immediately: try: - analysis_ok = _run_analysis_with_runtime_scheduler_lock(config, args, stock_codes) + analysis_ok = _run_analysis_with_runtime_scheduler_lock( + config, args, stock_codes, analysis_targets + ) except FutuPortfolioError as exc: if not start_serve: raise diff --git a/src/agent/tools/data_tools.py b/src/agent/tools/data_tools.py index 30934d620..8b567e78c 100644 --- a/src/agent/tools/data_tools.py +++ b/src/agent/tools/data_tools.py @@ -116,8 +116,15 @@ def _normalize_history_days(days: Any) -> Tuple[int, Dict[str, Any]]: 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 + from src.services.stock_list_parser import ParseStatus, parse_analysis_target raw_code = str(stock_code or "").strip() + target = parse_analysis_target(raw_code) + if target.asset_type == ParseStatus.INDEX: + # Explicit index identities keep their canonical bucket (``sh000016`` + # / ``csi930955``) so index bars never land in the colliding stock + # bucket (Story 1.5). + return [target.canonical_id], target.canonical_id normalized_code = canonical_stock_code(normalize_stock_code(raw_code)) candidates: List[str] = [] for candidate in (canonical_stock_code(raw_code), normalized_code): diff --git a/src/agent/tools/search_tools.py b/src/agent/tools/search_tools.py index 7a8c0a3f5..4bd9f5ffc 100644 --- a/src/agent/tools/search_tools.py +++ b/src/agent/tools/search_tools.py @@ -42,8 +42,22 @@ def _get_search_service(): def _canonical_search_code(stock_code: str) -> str: from data_provider.base import canonical_stock_code, normalize_stock_code + from src.services.stock_list_parser import ParseStatus, parse_analysis_target - return canonical_stock_code(normalize_stock_code(str(stock_code or "").strip())) + raw = str(stock_code or "").strip() + target = parse_analysis_target(raw) + if target.asset_type == ParseStatus.INDEX and target.canonical_id: + return target.canonical_id + return canonical_stock_code(normalize_stock_code(raw)) + + +def _resolve_search_subject(stock_code: str, stock_name: str) -> tuple[str, str]: + from src.services.stock_list_parser import ParseStatus, parse_analysis_target + + target = parse_analysis_target(stock_code) + if target.asset_type == ParseStatus.INDEX and target.matched_index is not None: + return "", target.matched_index.display_name + return stock_code, stock_name def _persist_news_response( @@ -85,11 +99,12 @@ def _persist_news_response( def _handle_search_stock_news(stock_code: str, stock_name: str) -> dict: """Search latest news for a stock.""" service = _get_search_service() + query_code, query_name = _resolve_search_subject(stock_code, stock_name) if not service.is_available: return {"error": "No search engine available (no API keys configured)"} - response = service.search_stock_news(stock_code, stock_name, max_results=5) + response = service.search_stock_news(query_code, query_name, max_results=5) if not response.success: # 检索已发起但失败：Agent 这一轮没有拿到新闻证据，必须记 0 而不是不记， @@ -105,7 +120,7 @@ def _handle_search_stock_news(stock_code: str, stock_name: str) -> dict: _persist_news_response( stock_code=stock_code, - stock_name=stock_name, + stock_name=query_name, dimension="latest_news", response=response, ) @@ -158,13 +173,14 @@ search_stock_news_tool = ToolDefinition( def _handle_search_comprehensive_intel(stock_code: str, stock_name: str) -> dict: """Multi-dimensional intelligence search.""" service = _get_search_service() + query_code, query_name = _resolve_search_subject(stock_code, stock_name) if not service.is_available: return {"error": "No search engine available (no API keys configured)"} intel_results = service.search_comprehensive_intel( - stock_code=stock_code, - stock_name=stock_name, + stock_code=query_code, + stock_name=query_name, max_searches=6, ) @@ -174,7 +190,7 @@ def _handle_search_comprehensive_intel(stock_code: str, stock_name: str) -> dict return {"error": "Comprehensive intel search returned no results"} # Format into readable report - report = service.format_intel_report(intel_results, stock_name) + report = service.format_intel_report(intel_results, query_name) # 本次真正交给 Agent 的证据条数，按维度累计后一次性记录。 evidence_count = 0 @@ -186,7 +202,7 @@ def _handle_search_comprehensive_intel(stock_code: str, stock_name: str) -> dict evidence_count += len(response.results) _persist_news_response( stock_code=stock_code, - stock_name=stock_name, + stock_name=query_name, dimension=dim_name, response=response, ) diff --git a/src/core/pipeline.py b/src/core/pipeline.py index 03ff2619d..2978b1085 100644 --- a/src/core/pipeline.py +++ b/src/core/pipeline.py @@ -95,6 +95,7 @@ from src.services.decision_signal_extractor import ( extract_and_persist_from_analysis_result, resolve_decision_signal_action_fields, ) +from src.services.stock_list_parser import AnalysisTarget, ParseStatus from src.services.decision_signal_summary import summarize_decision_signal from src.enums import ReportType from src.stock_analyzer import StockTrendAnalyzer, TrendAnalysisResult @@ -112,6 +113,22 @@ from bot.models import BotMessage logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Index capability matrix — the single authority for which analysis modules +# are skipped for index targets (Story 1.5). Skipping must happen before any +# composite service, snapshot persistence, or the traditional/Agent branch +# split, so the underlying provider calls are never reached. +# --------------------------------------------------------------------------- +INDEX_SKIP_MODULES = frozenset({ + "chip_distribution", + "fundamental", + "belong_boards", + "capital_flow", + "lhb", + "corporate_events", +}) + + def _share_image_payload(result: Any) -> Optional[Dict[str, Any]]: """Return structured poster data when the result exposes the real contract.""" @@ -360,6 +377,7 @@ class StockAnalysisPipeline: code: str, force_refresh: bool = False, current_time: Optional[datetime] = None, + analysis_target: Optional[AnalysisTarget] = None, ) -> Tuple[bool, Optional[str]]: """ 获取并保存单只股票数据 @@ -373,6 +391,7 @@ class StockAnalysisPipeline: code: 股票代码 force_refresh: 是否强制刷新（忽略本地缓存） current_time: 本轮运行冻结的参考时间，用于统一断点续传目标交易日判断 + analysis_target: 结构化分析目标（指数目标用于推导 market=cn 与日期语义） Returns: Tuple[是否成功, 错误信息] @@ -383,7 +402,7 @@ class StockAnalysisPipeline: stock_name = self.fetcher_manager.get_stock_name(code, allow_realtime=False) target_date = self._resolve_resume_target_date( - code, current_time=current_time + code, current_time=current_time, analysis_target=analysis_target ) # 断点续传检查：如果最新可复用交易日的数据已存在，则跳过 @@ -417,6 +436,7 @@ class StockAnalysisPipeline: report_type: ReportType, query_id: str, current_time: Optional[datetime] = None, + analysis_target: Optional[AnalysisTarget] = None, ) -> Optional[AnalysisResult]: """ 分析单只股票（增强版：含量比、换手率、筹码分析、多维度情报） @@ -434,6 +454,7 @@ class StockAnalysisPipeline: code: 股票代码 report_type: 报告类型 current_time: 本轮运行冻结的参考时间，用于统一市场阶段上下文 + analysis_target: 结构化分析目标（指数目标用于推导 market=cn 与能力矩阵） Returns: AnalysisResult 或 None（如果分析失败） @@ -443,7 +464,15 @@ class StockAnalysisPipeline: portfolio_context = getattr(self, "portfolio_context", None) if not isinstance(portfolio_context, dict): portfolio_context = None - market = get_market_for_stock(normalize_stock_code(code)) + is_index = ( + analysis_target is not None + and analysis_target.asset_type == ParseStatus.INDEX + ) + market = ( + "cn" + if is_index + else get_market_for_stock(normalize_stock_code(code)) + ) market_phase_context = build_market_phase_context( market=market, current_time=current_time, @@ -469,7 +498,17 @@ class StockAnalysisPipeline: self._emit_progress(18, f"{code}：正在获取行情与筹码数据") # 获取股票名称（先走轻量名称路径，后续若 realtime_quote 有 name 再覆盖） - stock_name = self.fetcher_manager.get_stock_name(code, allow_realtime=False) + # 指数目标使用注册表中文显示名（matched_index.display_name），避免把 + # raw alias / canonical code / 六码机器码带入搜索查询（Story 1.5 V7）。 + if is_index and analysis_target is not None: + index_name = getattr( + getattr(analysis_target, "matched_index", None), + "display_name", + None, + ) + stock_name = index_name or analysis_target.display_code or code + else: + stock_name = self.fetcher_manager.get_stock_name(code, allow_realtime=False) # Step 1: 获取实时行情（量比、换手率等）- 使用统一入口，自动故障切换 realtime_quote = None @@ -477,8 +516,8 @@ class StockAnalysisPipeline: if self.config.enable_realtime_quote: realtime_quote = self.fetcher_manager.get_realtime_quote(code, log_final_failure=False) if realtime_quote: - # 使用实时行情返回的真实股票名称 - if realtime_quote.name: + # 股票使用实时行情名称；指数保持注册表中文名为权威显示名。 + if realtime_quote.name and not is_index: stock_name = realtime_quote.name # 兼容不同数据源的字段（有些数据源可能没有 volume_ratio） volume_ratio = getattr(realtime_quote, 'volume_ratio', None) @@ -498,16 +537,20 @@ class StockAnalysisPipeline: stock_name = f'股票{code}' # Step 2: 获取筹码分布 - 使用统一入口，带熔断保护 + # 指数目标跳过筹码分布（INDEX_SKIP_MODULES 能力矩阵） chip_data = None - try: - chip_data = self.fetcher_manager.get_chip_distribution(code) - if chip_data: - logger.info(f"{stock_name}({code}) 筹码分布: 获利比例={chip_data.profit_ratio:.1%}, " - f"90%集中度={chip_data.concentration_90:.2%}") - else: - logger.debug(f"{stock_name}({code}) 筹码分布获取失败或已禁用") - except Exception as e: - logger.warning(f"{stock_name}({code}) 获取筹码分布失败: {e}") + if is_index and "chip_distribution" in INDEX_SKIP_MODULES: + logger.debug(f"{stock_name}({code}) 指数目标跳过筹码分布") + else: + try: + chip_data = self.fetcher_manager.get_chip_distribution(code) + if chip_data: + logger.info(f"{stock_name}({code}) 筹码分布: 获利比例={chip_data.profit_ratio:.1%}, " + f"90%集中度={chip_data.concentration_90:.2%}") + else: + logger.debug(f"{stock_name}({code}) 筹码分布获取失败或已禁用") + except Exception as e: + logger.warning(f"{stock_name}({code}) 获取筹码分布失败: {e}") # If agent mode is explicitly enabled, or specific agent skills are configured, use the Agent analysis pipeline. # NOTE: use config.agent_mode (explicit opt-in) instead of @@ -531,24 +574,45 @@ class StockAnalysisPipeline: # Step 2.5: 基本面能力聚合（统一入口，异常降级） # - 失败时返回 partial/failed，不影响既有技术面/新闻链路 # - 关闭开关时仍返回 not_supported 结构 + # - 指数目标跳过基本面/板块/资金流/龙虎榜/公司事件（INDEX_SKIP_MODULES） fundamental_context = None - try: - fundamental_context = self.fetcher_manager.get_fundamental_context( + fundamental_modules = { + "fundamental", + "belong_boards", + "capital_flow", + "lhb", + "corporate_events", + } + if is_index and INDEX_SKIP_MODULES.intersection(fundamental_modules): + logger.debug(f"{stock_name}({code}) 指数目标跳过基本面聚合") + fundamental_context = self.fetcher_manager.build_not_supported_fundamental_context( code, - budget_seconds=getattr( - self.config, - 'fundamental_stage_timeout_seconds', - FUNDAMENTAL_STAGE_TIMEOUT_SECONDS_DEFAULT, - ), + "index target: fundamental modules skipped", ) - except Exception as e: - logger.warning(f"{stock_name}({code}) 基本面聚合失败: {e}") - fundamental_context = self.fetcher_manager.build_failed_fundamental_context(code, str(e)) + else: + try: + fundamental_context = self.fetcher_manager.get_fundamental_context( + code, + budget_seconds=getattr( + self.config, + 'fundamental_stage_timeout_seconds', + FUNDAMENTAL_STAGE_TIMEOUT_SECONDS_DEFAULT, + ), + ) + except Exception as e: + logger.warning(f"{stock_name}({code}) 基本面聚合失败: {e}") + fundamental_context = self.fetcher_manager.build_failed_fundamental_context(code, str(e)) - fundamental_context = self._attach_belong_boards_to_fundamental_context( - code, - fundamental_context, - ) + if is_index and "belong_boards" in INDEX_SKIP_MODULES: + logger.debug(f"{stock_name}({code}) 指数目标跳过板块归属") + if isinstance(fundamental_context, dict): + fundamental_context = dict(fundamental_context) + fundamental_context["belong_boards"] = [] + else: + fundamental_context = self._attach_belong_boards_to_fundamental_context( + code, + fundamental_context, + ) market_structure_context = self._build_market_structure_context( code=code, stock_name=stock_name, @@ -559,16 +623,17 @@ class StockAnalysisPipeline: ) # P0: write-only snapshot, fail-open, no read dependency on this table. - try: - self.db.save_fundamental_snapshot( - query_id=query_id, - code=code, - payload=fundamental_context, - source_chain=fundamental_context.get("source_chain", []), - coverage=fundamental_context.get("coverage", {}), - ) - except Exception as e: - logger.debug(f"{stock_name}({code}) 基本面快照写入失败: {e}") + if not is_index: + try: + self.db.save_fundamental_snapshot( + query_id=query_id, + code=code, + payload=fundamental_context, + source_chain=fundamental_context.get("source_chain", []), + coverage=fundamental_context.get("coverage", {}), + ) + except Exception as e: + logger.debug(f"{stock_name}({code}) 基本面快照写入失败: {e}") # Step 3: 趋势分析（基于交易理念）— 在 Agent 分支之前执行，供两条路径共用 trend_result: Optional[TrendAnalysisResult] = None @@ -583,7 +648,9 @@ class StockAnalysisPipeline: df = pd.DataFrame([bar.to_dict() for bar in historical_bars]) # Issue #234: Augment with realtime for intraday MA calculation if self.config.enable_realtime_quote and realtime_quote: - df = self._augment_historical_with_realtime(df, realtime_quote, code) + df = self._augment_historical_with_realtime( + df, realtime_quote, code, market=market + ) trend_result = self.trend_analyzer.analyze(df, code) logger.info(f"{stock_name}({code}) 趋势分析: {trend_result.trend_status.value}, " f"买入信号={trend_result.buy_signal.value}, 评分={trend_result.signal_score}") @@ -607,6 +674,7 @@ class StockAnalysisPipeline: daily_market_context=daily_market_context, portfolio_context=portfolio_context, market_structure_context=market_structure_context, + analysis_target=analysis_target, ) # Step 4: 多维度情报搜索（最新消息+风险排查+业绩预期） @@ -627,8 +695,10 @@ class StockAnalysisPipeline: news_result_count = 0 # 使用多维度搜索（最多5次搜索） + # 指数目标：查询 subject 仅用注册表中文名称，不把 canonical code / + # 六码机器码带入查询（Story 1.5 V7）。 intel_results = self.search_service.search_comprehensive_intel( - stock_code=code, + stock_code=("" if is_index else code), stock_name=stock_name, max_searches=5 ) @@ -685,7 +755,9 @@ class StockAnalysisPipeline: # Step 5: 获取分析上下文（技术面数据） self._emit_progress(58, f"{stock_name}：正在整理分析上下文") - context = self._get_analysis_context_with_market_fallback(code) + context = self._get_analysis_context_with_market_fallback( + code, analysis_target=analysis_target + ) if context is None: logger.warning(f"{stock_name}({code}) 无法获取历史行情数据，将仅基于新闻和实时行情分析") @@ -864,6 +936,9 @@ class StockAnalysisPipeline: previous_operation_advice=action_source_advice, ) + if result: + self._append_daily_data_source(result, context, analysis_target) + # Step 8: 保存分析历史记录 if result and result.success: try: @@ -906,6 +981,7 @@ class StockAnalysisPipeline: report_type=report_type.value, context_snapshot=context_snapshot, portfolio_context=portfolio_context, + analysis_target=analysis_target, ) except Exception as e: record_history_run( @@ -1316,13 +1392,20 @@ class StockAnalysisPipeline: ) return None - def _ensure_agent_history(self, code: str, min_days: int = 240) -> None: + def _ensure_agent_history( + self, + code: str, + min_days: int = 240, + analysis_target: Optional[AnalysisTarget] = None, + ) -> None: """Ensure at least *min_days* of K-line history is in DB for agent tools.""" from src.services.history_loader import get_frozen_target_date target = get_frozen_target_date() if target is None: - target = self._resolve_resume_target_date(code) + target = self._resolve_resume_target_date( + code, analysis_target=analysis_target + ) start = target - timedelta(days=int(min_days * 1.8)) bars = self.db.get_data_range(code, start, target) if bars and len(bars) >= min(min_days, 200): @@ -1336,6 +1419,43 @@ class StockAnalysisPipeline: except Exception as e: logger.warning("[%s] Agent history prefetch failed: %s", code, e) + def _filter_agent_tools_for_index(self, executor: Any) -> Any: + """Return an executor whose tool registry excludes index-incompatible tools. + + Maps ``INDEX_SKIP_MODULES`` to the agent tool names that would otherwise + invoke the skipped bottom-layer providers (chip distribution, fundamental + aggregation, capital flow). The filtered registry carries the source + category-timeout map so per-category ceilings survive the subset copy. + """ + tool_modules = { + "get_chip_distribution": {"chip_distribution"}, + "get_stock_info": { + "fundamental", + "belong_boards", + "lhb", + "corporate_events", + }, + "get_capital_flow": {"capital_flow"}, + } + index_skip_tool_names = { + name + for name, modules in tool_modules.items() + if INDEX_SKIP_MODULES.intersection(modules) + } + registry = getattr(executor, "tool_registry", None) + if registry is None: + return executor + from src.agent.tools.registry import ToolRegistry as _TR + filtered = _TR(category_timeout_map=registry.category_timeout_map) + for name in registry.list_names(): + if name in index_skip_tool_names: + continue + tool_def = registry.get(name) + if tool_def is not None: + filtered.register(tool_def) + executor.tool_registry = filtered + return executor + def _analyze_with_agent( self, code: str, @@ -1352,6 +1472,7 @@ class StockAnalysisPipeline: daily_market_context: Optional[DailyMarketContext] = None, portfolio_context: Optional[Dict[str, Any]] = None, market_structure_context: Optional[Dict[str, Any]] = None, + analysis_target: Optional[AnalysisTarget] = None, ) -> Optional[AnalysisResult]: """ 使用 Agent 模式分析单只股票。 @@ -1360,6 +1481,11 @@ class StockAnalysisPipeline: from src.agent.factory import build_agent_executor report_language = normalize_report_language(getattr(self.config, "report_language", "zh")) + is_index = ( + analysis_target is not None + and analysis_target.asset_type == ParseStatus.INDEX + ) + requested_skills = ( self.analysis_skills if self.analysis_skills is not None @@ -1368,6 +1494,11 @@ class StockAnalysisPipeline: # Build executor from shared factory (ToolRegistry and SkillManager prototype are cached) executor = build_agent_executor(self.config, requested_skills) + # 指数目标：从 Agent 工具面剔除与 INDEX_SKIP_MODULES 对应的底层 provider + # 工具（筹码/基本面/资金流），确保 Agent 分支同样零调用（Story 1.5 V6）。 + if is_index: + executor = self._filter_agent_tools_for_index(executor) + # Build initial context to avoid redundant tool calls initial_context = { "stock_code": code, @@ -1418,7 +1549,7 @@ class StockAnalysisPipeline: persisted_intelligence_context = self._load_persisted_intelligence_context( code=code, stock_name=stock_name, - market=get_market_for_stock(normalize_stock_code(code)) or "cn", + market=("cn" if is_index else get_market_for_stock(normalize_stock_code(code)) or "cn"), ) if persisted_intelligence_context: existing = initial_context.get("news_context") @@ -1430,10 +1561,15 @@ class StockAnalysisPipeline: logger.info(f"[{code}] Agent mode: local intelligence evidence injected into news_context") # Issue #1066: ensure deep history is in DB before agent tools run - self._ensure_agent_history(code) + if analysis_target is None: + self._ensure_agent_history(code) + else: + self._ensure_agent_history(code, analysis_target=analysis_target) - analysis_context = self._load_agent_analysis_context(code, stock_name) - market = get_market_for_stock(normalize_stock_code(code)) + analysis_context = self._load_agent_analysis_context( + code, stock_name, analysis_target=analysis_target + ) + market = "cn" if is_index else get_market_for_stock(normalize_stock_code(code)) ( analysis_context_pack_summary, analysis_context_pack_overview, @@ -1687,6 +1823,13 @@ class StockAnalysisPipeline: ) ) + if result: + self._append_daily_data_source( + result, + analysis_context, + analysis_target, + ) + resolved_stock_name = result.name if result and result.name else stock_name # 保存新闻情报到数据库（Agent 工具结果仅用于 LLM 上下文，未持久化，Fixes #396） @@ -1694,7 +1837,7 @@ class StockAnalysisPipeline: if self.search_service is not None and self.search_service.is_available: try: news_response = self.search_service.search_stock_news( - stock_code=code, + stock_code=("" if is_index else code), stock_name=resolved_stock_name, max_results=5 ) @@ -1766,6 +1909,7 @@ class StockAnalysisPipeline: report_type=report_type.value, context_snapshot=agent_context_snapshot, portfolio_context=portfolio_context, + analysis_target=analysis_target, ) latest_diagnostic_snapshot = current_diagnostic_snapshot() if latest_diagnostic_snapshot is not None: @@ -1786,10 +1930,14 @@ class StockAnalysisPipeline: logger.exception(f"[{code}] Agent 详细错误信息:") return None - def _load_agent_analysis_context(self, code: str, stock_name: str) -> Dict[str, Any]: + def _load_agent_analysis_context( + self, code: str, stock_name: str, analysis_target: Optional[AnalysisTarget] = None + ) -> Dict[str, Any]: """Load daily-bar context for Agent pack summaries without blocking analysis.""" try: - context = self._get_analysis_context_with_market_fallback(code) + context = self._get_analysis_context_with_market_fallback( + code, analysis_target=analysis_target + ) except Exception as exc: logger.warning( "[%s] Agent analysis context load failed; daily_bars will be marked missing: %s", @@ -1813,13 +1961,18 @@ class StockAnalysisPipeline: "yesterday": {}, } - def _get_analysis_context_with_market_fallback(self, code: str) -> Optional[Dict[str, Any]]: + def _get_analysis_context_with_market_fallback( + self, code: str, analysis_target: Optional[AnalysisTarget] = None + ) -> Optional[Dict[str, Any]]: """Load analysis context, fetching JP/KR/TW daily bars when DB has no context.""" context = self.db.get_analysis_context(code) if isinstance(context, dict) and context: return context - market = get_market_for_stock(normalize_stock_code(code)) + if analysis_target is not None and analysis_target.asset_type == ParseStatus.INDEX: + market = "cn" + else: + market = get_market_for_stock(normalize_stock_code(code)) if market not in {"jp", "kr", "tw"}: return context @@ -2316,6 +2469,48 @@ class StockAnalysisPipeline: "strong_sell": "sell", }.get(signal_name) + @staticmethod + def _append_daily_data_source( + result: AnalysisResult, + context: Any, + analysis_target: Optional[AnalysisTarget], + ) -> None: + """Append an index's persisted daily provider without inferring sources.""" + if ( + analysis_target is None + or analysis_target.asset_type != ParseStatus.INDEX + ): + return + if not isinstance(context, dict): + return + today = context.get("today") + if not isinstance(today, dict): + return + source = today.get("data_source") + if not isinstance(source, str): + return + source = source.strip() + if ( + not source + or source.casefold() == "unknown" + or source.casefold().startswith("realtime:") + ): + return + + existing = result.data_sources + if existing is None: + existing_tokens = [] + elif isinstance(existing, str): + existing_tokens = [ + item.strip() for item in existing.split(",") if item.strip() + ] + else: + return + token = f"daily:{source}" + if token in existing_tokens: + return + result.data_sources = ",".join([*existing_tokens, token]) + @staticmethod def _mark_trend_fallback_source(result: AnalysisResult) -> None: if "trend:fallback" in (result.data_sources or ""): @@ -2535,7 +2730,11 @@ class StockAnalysisPipeline: return "震荡整理 ↔️" def _augment_historical_with_realtime( - self, df: pd.DataFrame, realtime_quote: Any, code: str + self, + df: pd.DataFrame, + realtime_quote: Any, + code: str, + market: Optional[str] = None, ) -> pd.DataFrame: """ 使用当日实时行情补齐历史 OHLCV，用于盘中 MA 计算。 @@ -2555,7 +2754,8 @@ class StockAnalysisPipeline: ) if not enable_realtime_tech: return df - market = get_market_for_stock(code) + if market is None: + market = get_market_for_stock(code) market_today = get_market_now(market).date() if market and not is_market_open(market, market_today): return df @@ -2691,6 +2891,7 @@ class StockAnalysisPipeline: report_type: str, context_snapshot: Dict[str, Any], portfolio_context: Optional[Dict[str, Any]] = None, + analysis_target: Optional[AnalysisTarget] = None, ) -> None: """Best-effort DecisionSignal extraction after analysis history is saved.""" @@ -2707,6 +2908,13 @@ class StockAnalysisPipeline: or getattr(self, "trace_id", None) or query_id ) + # 指数目标：信号以 market=cn 持久化（canonical code / 中文名由 result 携带） + market_override = ( + "cn" + if analysis_target is not None + and analysis_target.asset_type == ParseStatus.INDEX + else None + ) signal_result = extract_and_persist_from_analysis_result( result, context_snapshot=context_snapshot, @@ -2716,6 +2924,7 @@ class StockAnalysisPipeline: report_type=report_type, portfolio_context=portfolio_context, profile_source="auto_default", + market_override=market_override, ) if isinstance(signal_result, dict): summary = summarize_decision_signal(signal_result.get("item")) @@ -3000,12 +3209,20 @@ class StockAnalysisPipeline: @staticmethod def _resolve_resume_target_date( - code: str, current_time: Optional[datetime] = None + code: str, current_time: Optional[datetime] = None, + analysis_target: Optional[AnalysisTarget] = None, ) -> date: """ Resolve the trading date used by checkpoint/resume checks. + + For index targets the market is derived from the structured target + (``market=cn``) rather than from ``get_market_for_stock`` on the + canonical code, which would return ``None`` for CSI indices. """ - market = get_market_for_stock(normalize_stock_code(code)) + if analysis_target is not None and analysis_target.asset_type == ParseStatus.INDEX: + market = "cn" + else: + market = get_market_for_stock(normalize_stock_code(code)) return get_effective_trading_date(market, current_time=current_time) @staticmethod @@ -3082,6 +3299,7 @@ class StockAnalysisPipeline: report_type: ReportType = ReportType.SIMPLE, analysis_query_id: Optional[str] = None, current_time: Optional[datetime] = None, + analysis_target: Optional[AnalysisTarget] = None, ) -> Optional[AnalysisResult]: """ 处理单只股票的完整流程 @@ -3101,6 +3319,7 @@ class StockAnalysisPipeline: single_stock_notify: 是否启用单股推送模式（每分析完一只立即推送） report_type: 报告类型枚举（从配置读取，Issue #119） current_time: 本轮运行冻结的参考时间，用于统一断点续传目标交易日判断 + analysis_target: 结构化分析目标（指数目标用于推导 market=cn 与能力矩阵） Returns: AnalysisResult 或 None @@ -3108,7 +3327,9 @@ class StockAnalysisPipeline: logger.info(f"========== 开始处理 {code} ==========") from src.services.history_loader import set_frozen_target_date, reset_frozen_target_date - frozen_td = self._resolve_resume_target_date(code, current_time=current_time) + frozen_td = self._resolve_resume_target_date( + code, current_time=current_time, analysis_target=analysis_target + ) token = set_frozen_target_date(frozen_td) effective_query_id = analysis_query_id or getattr(self, "query_id", None) or uuid.uuid4().hex effective_trace_id = getattr(self, "trace_id", None) or effective_query_id @@ -3124,7 +3345,7 @@ class StockAnalysisPipeline: self._emit_progress(12, f"{code}：正在准备分析任务") # Step 1: 获取并保存数据 success, error = self.fetch_and_save_stock_data( - code, current_time=current_time + code, current_time=current_time, analysis_target=analysis_target ) if not success: @@ -3141,6 +3362,8 @@ class StockAnalysisPipeline: analyze_kwargs = {"query_id": effective_query_id} if current_time is not None: analyze_kwargs["current_time"] = current_time + if analysis_target is not None: + analyze_kwargs["analysis_target"] = analysis_target result = self.analyze_stock(code, report_type, **analyze_kwargs) if result and result.success: @@ -3178,6 +3401,7 @@ class StockAnalysisPipeline: send_notification: bool = True, merge_notification: bool = False, current_time: Optional[datetime] = None, + analysis_targets: Optional[List[AnalysisTarget]] = None, ) -> List[AnalysisResult]: """ 运行完整的分析流程 @@ -3194,6 +3418,7 @@ class StockAnalysisPipeline: send_notification: 是否发送推送通知 merge_notification: 是否合并推送（跳过本次推送，由 main 层合并个股+大盘后统一发送，Issue #190） current_time: 本轮运行冻结的参考时间；为空时在 run 内生成 + analysis_targets: 与 stock_codes 对齐的结构化分析目标列表（可选，指数目标用于推导 market=cn 与能力矩阵） Returns: 分析结果列表 @@ -3209,6 +3434,39 @@ class StockAnalysisPipeline: logger.error("未配置自选股列表，请在 .env 文件中设置 STOCK_LIST") return [] + # 过滤 unsupported 目标并按 canonical identity 去重，均发生在 + # prefetch/provider 调用前；无结构化 target 的旧入口保持原行为。 + if analysis_targets is not None: + if len(analysis_targets) != len(stock_codes): + raise ValueError("analysis_targets must align with stock_codes") + supported_codes = [] + supported_targets = [] + seen_identities = set() + for code, target in zip(stock_codes, analysis_targets): + if target is not None and target.asset_type == ParseStatus.UNSUPPORTED: + logger.warning( + "跳过 unsupported 目标 %r（%s），不触发网络请求", + code, + target.unsupported_reason or "unsupported", + ) + continue + identity = ( + target.canonical_id + if target is not None and target.asset_type == ParseStatus.INDEX + else code + ) + if identity in seen_identities: + logger.info("跳过重复分析目标 %s（canonical_id=%s）", code, identity) + continue + seen_identities.add(identity) + supported_codes.append(code) + supported_targets.append(target) + stock_codes = supported_codes + analysis_targets = supported_targets + if not stock_codes: + logger.warning("批次中所有目标均 unsupported，无任务可执行") + return [] + logger.info(f"===== 开始分析 {len(stock_codes)} 只股票 =====") logger.info(f"股票列表: {', '.join(stock_codes)}") logger.info(f"并发数: {self.max_workers}, 模式: {'仅获取数据' if dry_run else '完整分析'}") @@ -3262,18 +3520,28 @@ class StockAnalysisPipeline: # 注意：max_workers 设置较低（默认3）以避免触发反爬 with ThreadPoolExecutor(max_workers=self.max_workers) as executor: # 提交任务 - future_to_code = { - executor.submit( + future_to_code = {} + for idx, code in enumerate(stock_codes): + target = ( + analysis_targets[idx] + if analysis_targets is not None and idx < len(analysis_targets) + else None + ) + submit_kwargs = { + "skip_analysis": dry_run, + "single_stock_notify": False, + "report_type": report_type, + "analysis_query_id": uuid.uuid4().hex, + "current_time": resume_reference_time, + } + if target is not None: + submit_kwargs["analysis_target"] = target + future = executor.submit( self.process_single_stock, code, - skip_analysis=dry_run, - single_stock_notify=False, - report_type=report_type, # Issue #119: 传递报告类型 - analysis_query_id=uuid.uuid4().hex, - current_time=resume_reference_time, - ): code - for code in stock_codes - } + **submit_kwargs, + ) + future_to_code[future] = code # 收集结果 for idx, future in enumerate(as_completed(future_to_code)): @@ -3312,16 +3580,22 @@ class StockAnalysisPipeline: # dry-run 模式下，数据获取成功即视为成功 if dry_run: # 检查哪些股票的最新可复用交易日数据已存在 - success_count = sum( - 1 - for code in stock_codes + success_count = 0 + for idx, code in enumerate(stock_codes): + target = ( + analysis_targets[idx] + if analysis_targets is not None and idx < len(analysis_targets) + else None + ) if self.db.has_today_data( code, self._resolve_resume_target_date( - code, current_time=resume_reference_time + code, + current_time=resume_reference_time, + analysis_target=target, ), - ) - ) + ): + success_count += 1 fail_count = len(stock_codes) - success_count else: success_count = len(results) diff --git a/src/notification.py b/src/notification.py index e501ad5be..736816851 100644 --- a/src/notification.py +++ b/src/notification.py @@ -421,6 +421,21 @@ class NotificationService( return empty_news_disclosure(result, language) + @staticmethod + def _append_data_sources_line( + lines: List[str], + result: AnalysisResult, + labels: Dict[str, str], + ) -> bool: + data_sources = getattr(result, "data_sources", None) + if not isinstance(data_sources, str): + return False + data_sources = data_sources.strip() + if not data_sources: + return False + lines.append(f"*📋 {labels['data_sources_label']}：{data_sources}*") + return True + def generate_aggregate_report( self, results: List[AnalysisResult], @@ -1071,8 +1086,7 @@ class NotificationService( # 数据来源说明 if hasattr(result, 'search_performed') and result.search_performed: report_lines.append("*🔍 已执行联网搜索*") - if hasattr(result, 'data_sources') and result.data_sources: - report_lines.append(f"*📋 数据来源：{result.data_sources}*") + self._append_data_sources_line(report_lines, result, labels) # 错误信息（如果有） if not result.success and result.error_message: @@ -1322,6 +1336,7 @@ class NotificationService( news_disclosure = self._empty_news_disclosure(r, report_language) if news_disclosure: report_lines.append(news_disclosure) + self._append_data_sources_line(report_lines, r, labels) report_lines.extend([ "", "---", @@ -1587,6 +1602,9 @@ class NotificationService( report_lines.append(f"{result.news_summary}") report_lines.append("") + if self._append_data_sources_line(report_lines, result, labels): + report_lines.append("") + report_lines.extend([ "---", "", @@ -1929,6 +1947,8 @@ class NotificationService( news_disclosure = self._empty_news_disclosure(r, report_language) if news_disclosure: lines.append(news_disclosure) + if self._append_data_sources_line(lines, r, labels): + lines.append("") lines.append("") lines.append(f"*{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*") models = self._collect_models_used(results) diff --git a/src/report_language.py b/src/report_language.py index 02b8a3e36..7699b97c2 100644 --- a/src/report_language.py +++ b/src/report_language.py @@ -400,6 +400,7 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = { "volume_ratio_label": "量比", "turnover_rate_label": "换手率", "source_label": "行情来源", + "data_sources_label": "数据来源", "data_perspective_heading": "数据透视", "ma_alignment_label": "均线排列", "bullish_alignment_label": "多头排列", @@ -534,6 +535,7 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = { "volume_ratio_label": "Volume Ratio", "turnover_rate_label": "Turnover Rate", "source_label": "Source", + "data_sources_label": "Data Sources", "data_perspective_heading": "Data View", "ma_alignment_label": "MA Alignment", "bullish_alignment_label": "Bullish Alignment", @@ -668,6 +670,7 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = { "volume_ratio_label": "거래량비", "turnover_rate_label": "회전율", "source_label": "시세 출처", + "data_sources_label": "데이터 출처", "data_perspective_heading": "데이터 분석", "ma_alignment_label": "이동평균 배열", "bullish_alignment_label": "정배열", diff --git a/src/services/decision_signal_extractor.py b/src/services/decision_signal_extractor.py index d7ac53157..db6075221 100644 --- a/src/services/decision_signal_extractor.py +++ b/src/services/decision_signal_extractor.py @@ -50,8 +50,14 @@ def build_decision_signal_payload_from_report( query_source: str, report_type: str, profile_source: ProfileSource, + market_override: Optional[str] = None, ) -> Dict[str, Any] | None: - """Build a DecisionSignal payload from a completed stock analysis report.""" + """Build a DecisionSignal payload from a completed stock analysis report. + + ``market_override`` lets callers supply a target-derived market (e.g. + ``cn`` for index targets whose canonical code would otherwise resolve to + ``None`` via ``get_market_for_stock``). + """ if result is None or not getattr(result, "success", True): return None @@ -72,7 +78,7 @@ def build_decision_signal_payload_from_report( return None raw_code = str(getattr(result, "code", "") or "").strip() - market = get_market_for_stock(normalize_stock_code(raw_code)) + market = market_override or get_market_for_stock(normalize_stock_code(raw_code)) if not market: logger.warning("Skip decision signal extraction: unrecognized market stock_code=%s", raw_code) return None @@ -209,6 +215,7 @@ def extract_and_persist_from_analysis_result( report_type: str, profile_source: ProfileSource, service: Optional[DecisionSignalService] = None, + market_override: Optional[str] = None, ) -> Dict[str, Any] | None: """Best-effort extract and persist a DecisionSignal from an analysis result.""" @@ -222,6 +229,7 @@ def extract_and_persist_from_analysis_result( query_source=query_source, report_type=report_type, profile_source=profile_source, + market_override=market_override, ) if payload is None: return None diff --git a/src/services/decision_signal_service.py b/src/services/decision_signal_service.py index e515522a0..e9e8d753b 100644 --- a/src/services/decision_signal_service.py +++ b/src/services/decision_signal_service.py @@ -11,6 +11,7 @@ from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Literal, Optional, Tuple, get_args from data_provider.base import canonical_stock_code, normalize_stock_code +from src.services.stock_list_parser import ParseStatus, parse_analysis_target from src.core.trading_calendar import MarketPhase from src.repositories.decision_signal_repo import ( DecisionSignalCreateResult, @@ -1108,6 +1109,9 @@ class DecisionSignalService: @classmethod def _normalize_stock_code(cls, value: Any, *, market: Optional[str] = None) -> str: raw = str(value or "").strip() + target = parse_analysis_target(raw) + if target.asset_type == ParseStatus.INDEX and target.canonical_id: + return target.canonical_id if market == "us": code = canonical_stock_code(raw) elif market == "hk": diff --git a/src/services/history_loader.py b/src/services/history_loader.py index b4c7b4289..260904aad 100644 --- a/src/services/history_loader.py +++ b/src/services/history_loader.py @@ -62,8 +62,15 @@ def _get_fetcher_manager(): # --------------------------------------------------------------------------- def _history_code_candidates(stock_code: str) -> Tuple[List[str], str]: from data_provider.base import canonical_stock_code, normalize_stock_code + from src.services.stock_list_parser import ParseStatus, parse_analysis_target raw_code = str(stock_code or "").strip() + target = parse_analysis_target(raw_code) + if target.asset_type == ParseStatus.INDEX: + # Explicit index identities keep their canonical bucket (``sh000016`` + # / ``csi930955``) so index bars never land in the colliding stock + # bucket (Story 1.5). + return [target.canonical_id], target.canonical_id normalized_code = canonical_stock_code(normalize_stock_code(raw_code)) candidates: List[str] = [] for candidate in (canonical_stock_code(raw_code), normalized_code): diff --git a/templates/report_brief.j2 b/templates/report_brief.j2 index 9187188bf..3eaac09b5 100644 --- a/templates/report_brief.j2 +++ b/templates/report_brief.j2 @@ -13,6 +13,9 @@ {% if e.empty_news_disclosure %} {{ e.empty_news_disclosure }} {% endif %} +{% if e.result.data_sources and e.result.data_sources|trim %} +*📋 {{ labels.data_sources_label }}：{{ e.result.data_sources|trim }}* +{% endif %} {% endfor %} *{{ report_timestamp }}* diff --git a/templates/report_markdown.j2 b/templates/report_markdown.j2 index 33619686c..0316aa2e0 100644 --- a/templates/report_markdown.j2 +++ b/templates/report_markdown.j2 @@ -13,6 +13,9 @@ {% if summary_only and e.empty_news_disclosure %} {{ e.empty_news_disclosure }} {% endif %} +{% if summary_only and e.result.data_sources and e.result.data_sources|trim %} +*📋 {{ labels.data_sources_label }}：{{ e.result.data_sources|trim }}* +{% endif %} {% endfor %} --- @@ -240,6 +243,10 @@ {% endfor %} {% endif %} +{% if result.data_sources and result.data_sources|trim %} +*📋 {{ labels.data_sources_label }}：{{ result.data_sources|trim }}* +{% endif %} + --- {% endfor %} {% endif %} diff --git a/tests/test_decision_signal_extractor.py b/tests/test_decision_signal_extractor.py index f92d8efe7..1b64d90dc 100644 --- a/tests/test_decision_signal_extractor.py +++ b/tests/test_decision_signal_extractor.py @@ -624,3 +624,69 @@ def test_extract_and_persist_missing_price_plan_does_not_fabricate_fields(isolat assert item["entry_high"] is None assert item["stop_loss"] is None assert item["target_price"] is None + + +def test_build_payload_index_uses_market_override_cn() -> None: + """V9 — an index target whose canonical code would resolve to None market + (e.g. CSI) must persist with market=cn via the market_override.""" + result = _result(code="csi930955", name="红利低波100") + + payload = build_decision_signal_payload_from_report( + result, + context_snapshot=None, + portfolio_context=None, + source_report_id=955, + trace_id="trace-index-csi", + query_source="cli", + report_type="full", + profile_source=BUILD_PROFILE_SOURCE, + market_override="cn", + ) + + assert payload is not None + assert payload["stock_code"] == "csi930955" + assert payload["stock_name"] == "红利低波100" + assert payload["market"] == "cn" + assert payload["source_type"] == "analysis" + assert payload["source_report_id"] == 955 + + +def test_extract_and_persist_index_signal_with_market_override(isolated_db) -> None: + """V9 — end-to-end: an index signal persists with market=cn, source_type, + and source_report_id linkage (not skipped because CSI market was None).""" + service = DecisionSignalService(db_manager=isolated_db) + result = _result(code="csi930955", name="红利低波100") + + created = extract_and_persist_from_analysis_result( + result, + context_snapshot={"market_phase_summary": {"phase": "intraday"}}, + portfolio_context={"quantity": 0}, + source_report_id=955, + trace_id="trace-index-csi", + query_source="cli", + report_type="full", + profile_source="auto_default", + service=service, + market_override="cn", + ) + + assert created is not None + assert created["created"] is True + item = created["item"] + assert item["stock_code"] == "csi930955" + assert item["market"] == "cn" + assert item["source_type"] == "analysis" + assert item["source_report_id"] == 955 + + listed = service.list_signals(source_report_id=955) + assert listed["total"] == 1 + assert listed["items"][0]["stock_code"] == "csi930955" + assert listed["items"][0]["market"] == "cn" + + by_alias = service.list_signals(stock_code="930955.CSI", market="cn") + assert by_alias["total"] == 1 + assert by_alias["items"][0]["stock_code"] == "csi930955" + + latest = service.get_latest_active(stock_code="csi930955", market="cn") + assert latest["total"] == 1 + assert latest["items"][0]["source_report_id"] == 955 diff --git a/tests/test_fundamental_context.py b/tests/test_fundamental_context.py index 8e43253b4..590def9a7 100644 --- a/tests/test_fundamental_context.py +++ b/tests/test_fundamental_context.py @@ -39,6 +39,22 @@ class _DummyBoardFetcher: class TestFundamentalContext(unittest.TestCase): + def test_not_supported_builder_uses_existing_fundamental_schema(self) -> None: + manager = DataFetcherManager(fetchers=[]) + + context = manager.build_not_supported_fundamental_context( + "sh000016", "index target: fundamental modules skipped" + ) + + self.assertEqual(context["status"], "not_supported") + self.assertTrue(context["coverage"]) + self.assertTrue( + all(status == "not_supported" for status in context["coverage"].values()) + ) + self.assertEqual( + context["errors"], ["index target: fundamental modules skipped"] + ) + def test_offshore_market_returns_not_supported_when_adapter_empty(self) -> None: """When yfinance adapter has no data, offshore (US/HK) status is not_supported. diff --git a/tests/test_index_realtime_routing.py b/tests/test_index_realtime_routing.py new file mode 100644 index 000000000..d29704038 --- /dev/null +++ b/tests/test_index_realtime_routing.py @@ -0,0 +1,314 @@ +# -*- coding: utf-8 -*- +"""Story 1.5 — index realtime quote routing regression tests. + +Covers the smoke-found contract violations (2026-08-26 network smoke): +- ``get_realtime_quote`` must route registered index codes through a fixed + index chain instead of stripping sh/sz prefixes into the stock path + (``sh000016`` was resolved as *ST康佳A at 2.33). +- CSI indices use the Eastmoney single-stock secid endpoint only. +- ``prefetch_realtime_quotes`` must preserve explicit index identities. +- ``_to_sina_tx_symbol`` must preserve explicit sh/sz prefixes. +- history code candidates must keep the index canonical bucket. +- ``_augment_historical_with_realtime`` must use the caller-provided market. +""" + +from __future__ import annotations + +import unittest +from datetime import date, datetime +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pandas as pd + +from data_provider.akshare_fetcher import _to_sina_tx_symbol +from data_provider.base import DataFetcherManager +from data_provider.efinance_fetcher import EfinanceFetcher +from data_provider.realtime_types import RealtimeSource, UnifiedRealtimeQuote +from src.core.pipeline import StockAnalysisPipeline + + +def _quote(code: str, price: float = 3000.0) -> UnifiedRealtimeQuote: + return UnifiedRealtimeQuote( + code=code, + name="上证50", + price=price, + change_pct=1.0, + source=RealtimeSource.TENCENT, + ) + + +class _FakeAkshareFetcher: + name = "AkshareFetcher" + priority = 1 + + def __init__(self, tencent_quote: bool = True): + self.calls = [] + self.tencent_quote = tencent_quote + + def get_realtime_quote(self, stock_code, source="em"): + self.calls.append((stock_code, source)) + if source == "tencent" and self.tencent_quote: + return _quote(stock_code) + return None + + +class _FakeEfinanceFetcher: + name = "EfinanceFetcher" + priority = 0 + + def __init__(self): + self.calls = [] + + def get_realtime_quote(self, stock_code): + self.calls.append(stock_code) + return None + + def get_index_realtime_quote(self, stock_code): + self.calls.append(stock_code) + return None + + +class _FakeTickFlowFetcher: + name = "TickFlowFetcher" + priority = 2 + + def __init__(self): + self.calls = [] + self.prefetch_calls = [] + + def get_realtime_quote(self, stock_code): + self.calls.append(stock_code) + return None + + def prefetch_realtime_quotes(self, stock_codes, batch_size=None): + self.prefetch_calls.append((list(stock_codes), batch_size)) + return len(stock_codes) + + +class IndexRealtimeRoutingTestCase(unittest.TestCase): + def _manager(self, fetchers): + return DataFetcherManager(fetchers=fetchers) + + def _config(self, priority="tencent,akshare_sina,efinance,akshare_em"): + return SimpleNamespace( + enable_realtime_quote=True, + realtime_source_priority=priority, + realtime_cache_ttl=600, + ) + + def test_sh_index_routes_to_index_chain_with_prefix_preserved(self): + akshare = _FakeAkshareFetcher() + manager = self._manager( + [_FakeEfinanceFetcher(), akshare, _FakeTickFlowFetcher()] + ) + with patch("src.config.get_config", return_value=self._config()): + quote = manager.get_realtime_quote("sh000016") + self.assertIsNotNone(quote) + self.assertEqual(quote.code, "sh000016") + # First chain step: Tencent via AkshareFetcher with the prefixed symbol. + self.assertEqual(akshare.calls[0], ("sh000016", "tencent")) + + def test_sh_index_chain_falls_through_all_sources(self): + akshare = _FakeAkshareFetcher(tencent_quote=False) + efinance = _FakeEfinanceFetcher() + tickflow = _FakeTickFlowFetcher() + manager = self._manager([efinance, akshare, tickflow]) + with patch("src.config.get_config", return_value=self._config()): + quote = manager.get_realtime_quote("sh000016") + self.assertIsNone(quote) + self.assertEqual( + akshare.calls, [("sh000016", "tencent"), ("sh000016", "sina")] + ) + self.assertEqual(efinance.calls, ["sh000016"]) + self.assertEqual(tickflow.calls, ["000016.SH"]) + + def test_csi_index_uses_efinance_only(self): + akshare = _FakeAkshareFetcher() + efinance = _FakeEfinanceFetcher() + manager = self._manager([efinance, akshare, _FakeTickFlowFetcher()]) + with patch("src.config.get_config", return_value=self._config()): + quote = manager.get_realtime_quote("csi930955") + self.assertIsNone(quote) + self.assertEqual(efinance.calls, ["csi930955"]) + self.assertEqual(akshare.calls, []) + + def test_bare_code_stays_on_stock_path(self): + akshare = _FakeAkshareFetcher() + manager = self._manager([_FakeEfinanceFetcher(), akshare]) + with patch("src.config.get_config", return_value=self._config()): + manager.get_realtime_quote("000016") + # Bare 000016 is a stock: generic path normalizes and calls the + # configured priority sources, never the index chain. + self.assertNotIn(("sh000016", "tencent"), akshare.calls) + self.assertTrue(any(code == "000016" for code, _ in akshare.calls)) + + def test_prefetch_preserves_index_codes(self): + tickflow = _FakeTickFlowFetcher() + manager = self._manager([tickflow]) + with patch( + "src.config.get_config", return_value=self._config("tickflow,tencent") + ): + manager.prefetch_realtime_quotes( + ["sh000016", "600519", "000001", "AAPL", "hk00700"] + ) + self.assertEqual( + tickflow.prefetch_calls[0][0], + ["sh000016", "600519", "000001", "AAPL", "HK00700"], + ) + + +class SinaTxSymbolPrefixTestCase(unittest.TestCase): + def test_explicit_prefix_preserved(self): + self.assertEqual(_to_sina_tx_symbol("sh000016"), "sh000016") + self.assertEqual(_to_sina_tx_symbol("sz399001"), "sz399001") + self.assertEqual(_to_sina_tx_symbol("SH000300"), "sh000300") + self.assertEqual(_to_sina_tx_symbol("bj920748"), "bj920748") + + def test_bare_codes_unchanged(self): + self.assertEqual(_to_sina_tx_symbol("600519"), "sh600519") + self.assertEqual(_to_sina_tx_symbol("000001"), "sz000001") + self.assertEqual(_to_sina_tx_symbol("920748"), "bj920748") + self.assertEqual(_to_sina_tx_symbol("900901"), "sh900901") + + +class HistoryCodeCandidatesIndexTestCase(unittest.TestCase): + def test_index_canonical_bucket_preserved(self): + from src.services.history_loader import _history_code_candidates as hc + + candidates, normalized = hc("sh000016") + self.assertEqual(normalized, "sh000016") + self.assertIn("sh000016", candidates) + + candidates, normalized = hc("csi930955") + self.assertEqual(normalized, "csi930955") + self.assertIn("csi930955", candidates) + + def test_stock_candidates_unchanged(self): + from src.services.history_loader import _history_code_candidates as hc + + _, normalized = hc("600519") + self.assertEqual(normalized, "600519") + _, normalized = hc("1810.HK") + self.assertEqual(normalized, "HK01810") + + def test_data_tools_candidates_preserve_index_canonical(self): + from src.agent.tools.data_tools import _history_code_candidates as dc + + _, normalized = dc("sh000016") + self.assertEqual(normalized, "sh000016") + _, normalized = dc("csi930955") + self.assertEqual(normalized, "csi930955") + + +class EfinanceIndexQuoteTestCase(unittest.TestCase): + def _fetcher(self): + with patch( + "data_provider.efinance_fetcher.get_config", + return_value=SimpleNamespace(enable_eastmoney_patch=False), + ): + return EfinanceFetcher(sleep_min=0, sleep_max=0) + + @patch("data_provider.efinance_fetcher.requests.get") + def test_sh_index_quote_parsed(self, mock_get): + mock_get.return_value = MagicMock( + status_code=200, + json=MagicMock( + return_value={ + "data": { + "f43": 2905.08, + "f44": 2914.18, + "f45": 2870.42, + "f46": 2871.89, + "f47": 45323454, + "f48": 140152858404.0, + "f57": "000016", + "f58": "上证50", + "f60": 2875.51, + "f168": 0.28, + "f169": 29.57, + "f170": 1.03, + "f171": 1.52, + } + } + ), + ) + quote = self._fetcher().get_index_realtime_quote("sh000016") + self.assertIsNotNone(quote) + self.assertEqual(quote.code, "sh000016") + self.assertEqual(quote.name, "上证50") + self.assertEqual(quote.price, 2905.08) + self.assertEqual(quote.source, RealtimeSource.EFINANCE) + self.assertEqual(mock_get.call_args.kwargs["params"]["secid"], "1.000016") + + @patch("data_provider.efinance_fetcher.requests.get") + def test_csi_index_secid(self, mock_get): + mock_get.return_value = MagicMock( + status_code=200, + json=MagicMock( + return_value={"data": {"f43": 11365.47, "f58": "红利低波100"}} + ), + ) + quote = self._fetcher().get_index_realtime_quote("csi930955") + self.assertIsNotNone(quote) + self.assertEqual(quote.price, 11365.47) + self.assertEqual(mock_get.call_args.kwargs["params"]["secid"], "2.930955") + + def test_non_index_returns_none(self): + self.assertIsNone(self._fetcher().get_index_realtime_quote("600519")) + + def test_get_realtime_quote_delegates_index(self): + fetcher = self._fetcher() + with patch.object( + fetcher, "get_index_realtime_quote", return_value=_quote("sh000016") + ) as mock_idx: + quote = fetcher.get_realtime_quote("sh000016") + mock_idx.assert_called_once_with("sh000016") + self.assertIsNotNone(quote) + + +class AugmentRealtimeIndexMarketTestCase(unittest.TestCase): + def test_index_market_passed_avoids_market_for_stock(self): + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = SimpleNamespace(enable_realtime_technical_indicators=True) + df = pd.DataFrame( + [ + { + "code": "csi930955", + "date": date(2026, 8, 25), + "open": 100.0, + "high": 101.0, + "low": 99.0, + "close": 100.5, + "volume": 100, + "amount": 0, + "pct_chg": 0, + } + ] + ) + quote = SimpleNamespace( + price=101.0, + open_price=100.0, + high=102.0, + low=99.0, + volume=200, + amount=None, + change_pct=1.0, + pre_close=None, + ) + with patch("src.core.pipeline.is_market_open", return_value=True), patch( + "src.core.pipeline.get_market_now", + return_value=datetime(2026, 8, 26, 15, 0), + ) as mock_now, patch( + "src.core.pipeline.get_market_for_stock", return_value=None + ) as mock_market: + result = pipeline._augment_historical_with_realtime( + df, quote, "csi930955", market="cn" + ) + self.assertEqual(len(result), 2) + mock_market.assert_not_called() + mock_now.assert_called_once_with("cn") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_main_portfolio.py b/tests/test_main_portfolio.py index f33b17912..ea8ce3a36 100644 --- a/tests/test_main_portfolio.py +++ b/tests/test_main_portfolio.py @@ -474,8 +474,13 @@ class MainPortfolioTest(unittest.TestCase): "src.feishu_doc.FeishuDocManager", ) as feishu_manager: feishu_manager.return_value.is_configured.return_value = False - first_result = main.run_full_analysis(config, args, ["600519"]) - second_result = main.run_full_analysis(config, args, ["600519"]) + cli_target = MagicMock() + first_result = main.run_full_analysis( + config, args, ["600519"], analysis_targets=[cli_target] + ) + second_result = main.run_full_analysis( + config, args, ["600519"], analysis_targets=[cli_target] + ) self.assertTrue(first_result) self.assertTrue(second_result) @@ -491,6 +496,7 @@ class MainPortfolioTest(unittest.TestCase): self.assertEqual(pipeline.run.call_count, 2) for invocation in pipeline.run.call_args_list: self.assertEqual(invocation.kwargs["stock_codes"], ["AAPL", "HK00700"]) + self.assertIsNone(invocation.kwargs["analysis_targets"]) def test_run_full_analysis_skips_empty_futu_portfolio_without_fallback(self): args = SimpleNamespace( diff --git a/tests/test_main_schedule_mode.py b/tests/test_main_schedule_mode.py index 3ecf8c175..24744d775 100644 --- a/tests/test_main_schedule_mode.py +++ b/tests/test_main_schedule_mode.py @@ -20,6 +20,7 @@ _ENV_BEFORE_MAIN_IMPORT = dict(os.environ) import main from src.brokers.futu.portfolio import FutuPortfolioError from src.config import Config +from src.services.stock_list_parser import ParseStatus, parse_analysis_target _MAIN_IMPORT_ENV_ADDITIONS = frozenset(set(os.environ) - set(_ENV_BEFORE_MAIN_IMPORT)) _MAIN_IMPORT_ENV_OVERRIDES = { @@ -191,6 +192,40 @@ class MainScheduleModeTestCase(unittest.TestCase): self.assertEqual(effective_region, "jp,kr") self.assertFalse(should_skip_all) + def test_compute_trading_day_filter_filters_registered_indices_on_cn_holiday(self) -> None: + """Index codes whose ``get_market_for_stock`` returns None must still + participate in CN trading-day filtering via ``parse_analysis_target`` + (INDEX -> market=cn). On a CN holiday the indices are dropped, a US + stock whose market is open stays, and a market-unknown non-index code + stays (fail-open unchanged).""" + args = self._make_args() + config = self._make_config( + trading_day_check_enabled=True, + market_review_enabled=False, + database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"), + ) + + stock_codes = ["sh000016", "csi930955", "930955.CSI", "AAPL", "XYZ123"] + + def fake_market(code: str): + return "us" if code == "AAPL" else None + + with patch( + "src.core.trading_calendar.get_market_for_stock", + side_effect=fake_market, + ), patch("src.core.trading_calendar.get_open_markets_today", return_value={"us"}): + filtered_codes, effective_region, should_skip_all = main._compute_trading_day_filter( + config, + args, + stock_codes, + ) + + # 已登记指数按 market=cn 参与过滤，CN 休市时被剔除；AAPL 因 US 开市保留； + # 市场未知的非指数 code（XYZ123）继续 fail-open 保留。 + self.assertEqual(filtered_codes, ["AAPL", "XYZ123"]) + self.assertIsNone(effective_region) + self.assertFalse(should_skip_all) + def test_public_webui_bind_warns_when_auth_is_disabled(self) -> None: with patch("src.auth.is_auth_enabled", return_value=False), \ patch("main.logger.warning") as warning_log: @@ -393,6 +428,7 @@ class MainScheduleModeTestCase(unittest.TestCase): patch("main.setup_logging"), \ patch("main.run_full_analysis") as run_full_analysis, \ patch("main.logger.warning") as warning_log, \ + patch("main._refresh_stock_index_cache_for_analysis"), \ patch("src.scheduler.run_with_schedule", side_effect=fake_run_with_schedule): exit_code = main.main() @@ -418,13 +454,79 @@ class MainScheduleModeTestCase(unittest.TestCase): with patch("main.parse_arguments", return_value=args), \ patch("main.get_config", return_value=config), \ patch("main.setup_logging"), \ + patch("main._refresh_stock_index_cache_for_analysis"), \ patch("main.run_full_analysis") as run_full_analysis: exit_code = main.main() self.assertEqual(exit_code, 0) run_full_analysis.assert_called_once() _, _, stock_codes = run_full_analysis.call_args.args + analysis_targets = run_full_analysis.call_args.kwargs.get("analysis_targets") self.assertEqual(stock_codes, ["005930.KS"]) + self.assertEqual(len(analysis_targets), 1) + self.assertEqual(analysis_targets[0].asset_type, "stock") + + def test_standalone_run_builds_structured_index_targets(self) -> None: + args = self._make_args( + stocks="sh000016,000300.CSI,930955.CSI,000016" + ) + config = self._make_config(run_immediately=True) + + with patch("main.parse_arguments", return_value=args), \ + patch("main.get_config", return_value=config), \ + patch("main.setup_logging"), \ + patch("main._refresh_stock_index_cache_for_analysis"), \ + patch("main.run_full_analysis") as run_full_analysis: + exit_code = main.main() + + self.assertEqual(exit_code, 0) + _, _, stock_codes = run_full_analysis.call_args.args + analysis_targets = run_full_analysis.call_args.kwargs["analysis_targets"] + self.assertEqual( + stock_codes, + ["sh000016", "sh000300", "csi930955", "000016"], + ) + self.assertEqual( + [target.asset_type for target in analysis_targets], + ["index", "index", "index", "stock"], + ) + self.assertEqual( + [target.canonical_id for target in analysis_targets], + ["sh000016", "sh000300", "csi930955", "sz000016"], + ) + + def test_standalone_run_refreshes_index_cache_before_parsing_stocks(self) -> None: + """``main.main()`` must refresh the stock-index registry (best-effort) + BEFORE parsing ``--stocks`` so a first run with a stale local registry + resolves a newly-registered alias to an index target in the same run.""" + args = self._make_args(stocks="930955.CSI") + config = self._make_config(run_immediately=True) + calls = [] + + def fake_refresh(cfg): + calls.append(("refresh", cfg)) + + def fake_run_full_analysis(cfg, a, stock_codes, **kwargs): + calls.append(("parse", stock_codes, kwargs.get("analysis_targets"))) + return 0 + + with patch("main.parse_arguments", return_value=args), \ + patch("main.get_config", return_value=config), \ + patch("main.setup_logging"), \ + patch( + "main._refresh_stock_index_cache_for_analysis", + side_effect=fake_refresh, + ), \ + patch("main.run_full_analysis", side_effect=fake_run_full_analysis): + exit_code = main.main() + + self.assertEqual(exit_code, 0) + # 刷新必须先于 --stocks 解析执行。 + self.assertEqual(calls[0][0], "refresh") + self.assertEqual(calls[1][0], "parse") + self.assertEqual(calls[1][1], ["csi930955"]) + self.assertEqual(calls[1][2][0].asset_type, ParseStatus.INDEX) + self.assertEqual(calls[1][2][0].canonical_id, "csi930955") def test_standalone_run_returns_nonzero_when_startup_analysis_reports_failure(self) -> None: args = self._make_args() @@ -438,7 +540,7 @@ class MainScheduleModeTestCase(unittest.TestCase): exit_code = main.main() self.assertEqual(exit_code, 1) - run_with_lock.assert_called_once_with(config, args, None) + run_with_lock.assert_called_once_with(config, args, None, None) def test_standalone_futu_portfolio_failure_returns_nonzero(self) -> None: args = self._make_args(portfolio="futu") @@ -740,7 +842,7 @@ class MainScheduleModeTestCase(unittest.TestCase): self.assertEqual(exit_code, 0) start_bots.assert_not_called() - run_with_lock.assert_called_once_with(config, args, None) + run_with_lock.assert_called_once_with(config, args, None, None) run_full_analysis.assert_not_called() error_log.assert_called_once() @@ -866,7 +968,7 @@ class MainScheduleModeTestCase(unittest.TestCase): self.assertEqual(exit_code, 0) self.assertEqual(run_with_lock.call_count, 1) - run_with_lock.assert_called_once_with(config, args, None) + run_with_lock.assert_called_once_with(config, args, None, None) run_full_analysis.assert_not_called() start_bots.assert_called_once_with(config) @@ -898,7 +1000,7 @@ class MainScheduleModeTestCase(unittest.TestCase): exit_code = main.main() self.assertEqual(exit_code, 0) - run_with_lock.assert_called_once_with(config, args, None) + run_with_lock.assert_called_once_with(config, args, None, None) start_bots.assert_called_once_with(config) exception_log.assert_any_call( "Futu 持仓导入失败，Web/API 服务继续运行: %s", @@ -1198,11 +1300,18 @@ class MainScheduleModeTestCase(unittest.TestCase): with patch("main.parse_arguments", return_value=args), \ patch("main.get_config", return_value=config), \ patch("main.setup_logging"), \ + patch("main._refresh_stock_index_cache_for_analysis"), \ patch("main.run_full_analysis") as run_full_analysis: exit_code = main.main() self.assertEqual(exit_code, 0) - run_full_analysis.assert_called_once_with(config, args, ["600519", "000001"]) + run_full_analysis.assert_called_once() + _, _, stock_codes = run_full_analysis.call_args.args + analysis_targets = run_full_analysis.call_args.kwargs.get("analysis_targets") + self.assertEqual(stock_codes, ["600519", "000001"]) + self.assertEqual(len(analysis_targets), 2) + self.assertEqual(analysis_targets[0].asset_type, "stock") + self.assertEqual(analysis_targets[1].asset_type, "stock") def test_run_full_analysis_skips_market_review_when_shared_lock_is_held(self) -> None: from src.core.market_review_lock import ( @@ -1956,6 +2065,7 @@ class MainScheduleModeTestCase(unittest.TestCase): send_notification=True, merge_notification=True, current_time=unittest.mock.ANY, + analysis_targets=None, ) notifier_message = pipeline.notifier.send.call_args.args[0] self.assertIn("## 完整大盘复盘", notifier_message) @@ -2065,6 +2175,43 @@ class MainScheduleModeTestCase(unittest.TestCase): self.assertIs(call_args.args[1], run_market_review) self.assertEqual(call_args.kwargs["trigger_source"], "schedule") + def test_run_full_analysis_keeps_targets_aligned_after_trading_day_filter(self) -> None: + args = self._make_args(dry_run=True, no_market_review=True) + config = self._make_config( + trading_day_check_enabled=True, + market_review_enabled=False, + daily_market_context_enabled=False, + single_stock_notify=False, + merge_email_notification=False, + analysis_delay=0, + database_path=str(Path(self.temp_dir.name) / "stock_analysis.db"), + ) + pipeline = MagicMock() + pipeline.run.return_value = [] + targets = [ + parse_analysis_target("sh000016"), + parse_analysis_target("AAPL"), + ] + + with patch.object(main, "_refresh_stock_index_cache_for_analysis"), \ + patch.object( + main, + "_compute_trading_day_filter", + return_value=(["AAPL"], "us", False), + ), \ + patch("src.core.pipeline.StockAnalysisPipeline", return_value=pipeline), \ + patch("src.core.market_review.run_market_review"): + main.run_full_analysis( + config, + args, + ["sh000016", "AAPL"], + analysis_targets=targets, + ) + + run_kwargs = pipeline.run.call_args.kwargs + self.assertEqual(run_kwargs["stock_codes"], ["AAPL"]) + self.assertEqual(run_kwargs["analysis_targets"], [targets[1]]) + def test_market_review_mode_uses_shared_runtime_assembly(self) -> None: args = self._make_args(market_review=True) config = self._make_config( diff --git a/tests/test_notification.py b/tests/test_notification.py index d10a5a052..e2091aa1c 100644 --- a/tests/test_notification.py +++ b/tests/test_notification.py @@ -713,6 +713,27 @@ class TestNotificationServiceSendToMethods(unittest.TestCase): class TestNotificationServiceReportGeneration(unittest.TestCase): """报告生成与选路相关测试。""" + @staticmethod + def _source_result( + code: str, + name: str, + score: int, + data_sources: Optional[str], + ) -> AnalysisResult: + result = AnalysisResult( + code=code, + name=name, + sentiment_score=score, + trend_prediction="震荡", + operation_advice="观望", + analysis_summary=f"{name} summary", + report_language="zh", + data_sources=data_sources or "", + ) + if data_sources is None: + setattr(result, "data_sources", None) + return result + def test_signal_metadata_uses_resolved_eight_state_action(self): service = NotificationService() cases = [ @@ -824,6 +845,200 @@ class TestNotificationServiceReportGeneration(unittest.TestCase): self.assertEqual(mock_dashboard.call_count, 3) mock_brief.assert_called_once() + @mock.patch("src.notification.get_config") + def test_dashboard_sources_are_isolated_in_builtin_and_template_renderers( + self, mock_get_config: mock.MagicMock + ): + alpha = self._source_result("AAA", "Alpha", 80, "agent:a,daily:AlphaFetcher") + empty = self._source_result("NONE", "NoSource", 70, None) + beta = self._source_result("BBB", "Beta", 60, "agent:b,daily:BetaFetcher") + + for renderer_enabled in (False, True): + with self.subTest(renderer_enabled=renderer_enabled): + mock_get_config.return_value = _make_config( + report_renderer_enabled=renderer_enabled + ) + service = NotificationService() + out = service.generate_dashboard_report( + [beta, empty, alpha], report_date="2026-08-27" + ) + + alpha_start = out.index("Alpha (AAA)") + empty_start = out.index("NoSource (NONE)") + beta_start = out.index("Beta (BBB)") + alpha_block = out[alpha_start:empty_start] + empty_block = out[empty_start:beta_start] + beta_block = out[beta_start:] + + self.assertIn("agent:a,daily:AlphaFetcher", alpha_block) + self.assertNotIn("daily:BetaFetcher", alpha_block) + self.assertNotIn("数据来源", empty_block) + self.assertIn("agent:b,daily:BetaFetcher", beta_block) + self.assertNotIn("daily:AlphaFetcher", beta_block) + self.assertEqual(out.count("agent:a,daily:AlphaFetcher"), 1) + self.assertEqual(out.count("agent:b,daily:BetaFetcher"), 1) + if not renderer_enabled: + self.assertIn( + "agent:a,daily:AlphaFetcher*\n\n---", + alpha_block, + ) + + @mock.patch("src.notification.get_config") + def test_brief_sources_are_isolated_in_builtin_and_template_renderers( + self, mock_get_config: mock.MagicMock + ): + alpha = self._source_result("AAA", "Alpha", 80, "daily:AlphaFetcher") + empty = self._source_result("NONE", "NoSource", 70, "") + beta = self._source_result("BBB", "Beta", 60, "daily:BetaFetcher") + + for renderer_enabled in (False, True): + with self.subTest(renderer_enabled=renderer_enabled): + mock_get_config.return_value = _make_config( + report_renderer_enabled=renderer_enabled + ) + service = NotificationService() + out = service.generate_brief_report( + [beta, empty, alpha], report_date="2026-08-27" + ) + + alpha_start = out.index("**Alpha(AAA)**") + empty_start = out.index("**NoSource(NONE)**") + beta_start = out.index("**Beta(BBB)**") + alpha_block = out[alpha_start:empty_start] + empty_block = out[empty_start:beta_start] + beta_block = out[beta_start:] + + self.assertIn("daily:AlphaFetcher", alpha_block) + self.assertNotIn("daily:BetaFetcher", alpha_block) + self.assertNotIn("数据来源", empty_block) + self.assertIn("daily:BetaFetcher", beta_block) + self.assertNotIn("daily:AlphaFetcher", beta_block) + self.assertEqual(out.count("daily:AlphaFetcher"), 1) + self.assertEqual(out.count("daily:BetaFetcher"), 1) + + @mock.patch("src.notification.get_config") + def test_dashboard_summary_only_keeps_sources_without_restoring_details( + self, mock_get_config: mock.MagicMock + ): + result = self._source_result("AAA", "Alpha", 80, "daily:AlphaFetcher") + result.dashboard = { + "core_conclusion": {"one_sentence": "summary-only conclusion"}, + "battle_plan": {"entry_plan": "detail-only plan"}, + } + + for renderer_enabled in (False, True): + with self.subTest(renderer_enabled=renderer_enabled): + mock_get_config.return_value = _make_config( + report_renderer_enabled=renderer_enabled + ) + service = NotificationService() + service._report_summary_only = True + out = service.generate_dashboard_report( + [result], report_date="2026-08-27" + ) + + self.assertEqual(out.count("daily:AlphaFetcher"), 1) + self.assertNotIn("核心结论", out) + self.assertNotIn("detail-only plan", out) + + @mock.patch("src.notification.get_config") + def test_empty_template_output_falls_back_to_source_aware_reports( + self, mock_get_config: mock.MagicMock + ): + mock_get_config.return_value = _make_config(report_renderer_enabled=True) + result = self._source_result("AAA", "Alpha", 80, "daily:AlphaFetcher") + service = NotificationService() + + with mock.patch("src.services.report_renderer.render", return_value=None) as render: + dashboard = service.generate_dashboard_report( + [result], report_date="2026-08-27" + ) + brief = service.generate_brief_report([result], report_date="2026-08-27") + + self.assertEqual(render.call_count, 2) + self.assertEqual(dashboard.count("daily:AlphaFetcher"), 1) + self.assertEqual(brief.count("daily:AlphaFetcher"), 1) + self.assertIn("*📋 数据来源：daily:AlphaFetcher*", dashboard) + self.assertIn("*📋 数据来源：daily:AlphaFetcher*", brief) + + @mock.patch("src.notification.get_config") + def test_source_labels_render_in_all_supported_languages( + self, mock_get_config: mock.MagicMock + ): + labels_by_language = { + "zh": "数据来源", + "en": "Data Sources", + "ko": "데이터 출처", + } + for language, expected_label in labels_by_language.items(): + for renderer_enabled in (False, True): + for report_type in ("simple", "brief"): + with self.subTest( + language=language, + renderer_enabled=renderer_enabled, + report_type=report_type, + ): + mock_get_config.return_value = _make_config( + report_renderer_enabled=renderer_enabled + ) + service = NotificationService() + result = self._source_result( + "AAA", "Alpha", 80, "daily:AlphaFetcher" + ) + result.report_language = language + + out = service.generate_aggregate_report( + [result], report_type, report_date="2026-08-27" + ) + + self.assertIn( + f"*📋 {expected_label}：daily:AlphaFetcher*", + out, + ) + + @mock.patch("src.notification.get_config") + def test_reports_omit_source_label_when_data_sources_are_missing( + self, mock_get_config: mock.MagicMock + ): + for data_sources in (None, "", " "): + for renderer_enabled in (False, True): + for report_type in ("simple", "brief"): + with self.subTest( + data_sources=data_sources, + renderer_enabled=renderer_enabled, + report_type=report_type, + ): + mock_get_config.return_value = _make_config( + report_renderer_enabled=renderer_enabled + ) + service = NotificationService() + result = self._source_result( + "AAA", "Alpha", 80, data_sources + ) + + out = service.generate_aggregate_report( + [result], report_type, report_date="2026-08-27" + ) + + self.assertNotIn("数据来源", out) + + @mock.patch("src.notification.get_config") + def test_generate_daily_report_keeps_existing_source_line( + self, mock_get_config: mock.MagicMock + ): + mock_get_config.return_value = _make_config(report_renderer_enabled=False) + service = NotificationService() + result = self._source_result( + "AAA", "Alpha", 80, "agent:a,daily:AlphaFetcher" + ) + + out = service.generate_daily_report([result], report_date="2026-08-27") + + self.assertEqual( + out.count("*📋 数据来源：agent:a,daily:AlphaFetcher*"), + 1, + ) + @mock.patch("src.notification.get_config") def test_generate_single_stock_report_keeps_legacy_simple_format(self, mock_get_config: mock.MagicMock): mock_get_config.return_value = _make_config(report_renderer_enabled=True) diff --git a/tests/test_pipeline_fetch_error.py b/tests/test_pipeline_fetch_error.py index fa5bbeae4..25a46109d 100644 --- a/tests/test_pipeline_fetch_error.py +++ b/tests/test_pipeline_fetch_error.py @@ -43,7 +43,9 @@ class PipelineFetchErrorTestCase(unittest.TestCase): self.assertTrue(success) self.assertIsNone(error) - _mock_target.assert_called_once_with("600519", current_time=current_time) + _mock_target.assert_called_once_with( + "600519", current_time=current_time, analysis_target=None + ) pipeline.db.has_today_data.assert_called_once_with("600519", date(2026, 3, 27)) pipeline.fetcher_manager.get_daily_data.assert_not_called() diff --git a/tests/test_pipeline_index_targets.py b/tests/test_pipeline_index_targets.py new file mode 100644 index 000000000..e157be70b --- /dev/null +++ b/tests/test_pipeline_index_targets.py @@ -0,0 +1,766 @@ +# -*- coding: utf-8 -*- +"""Story 1.5 — focused Pipeline/CLI index integration tests. + +Covers the deterministic acceptance matrix from +``_bmad-output/specs/spec-pipeline-cli-integration/verification.md``: +- V1 CLI target construction +- V2 Unsupported boundary (reject before provider calls, batch continues) +- V3 Canonical batch identity (alias dedupe, stock/index isolation) +- V4 Market/date propagation (SH/SZ/CSI -> cn) +- V5 Traditional capability matrix (skip modules zero bottom-layer calls) +- V6 Agent capability matrix (same negative/positive assertions) +- V7 Search semantics (name-only query subject for indices) +- V10 Daily data-source attribution (report/history wiring and fail-open cases) +- V11 Dry-run target/date propagation + +Cross-layer V8-V9/V11-V12 compatibility remains covered by the existing history, +data-routing, task-service, schedule, and stock-regression suites. +""" + +from __future__ import annotations + +import re +import unittest +from datetime import date, datetime, timezone +from unittest.mock import MagicMock, patch + +from src.analyzer import AnalysisResult +from src.config import Config +from src.core.pipeline import StockAnalysisPipeline, INDEX_SKIP_MODULES +from src.enums import ReportType +from src.notification import NotificationService +from src.search_service import SearchResponse, SearchService +from src.services.stock_list_parser import ( + AnalysisTarget, + ParseStatus, + parse_analysis_target, +) + + +def _index_target(raw: str) -> AnalysisTarget: + target = parse_analysis_target(raw) + assert target.asset_type == ParseStatus.INDEX, f"{raw} should be index" + return target + + +def _stock_target(raw: str) -> AnalysisTarget: + target = parse_analysis_target(raw) + assert target.asset_type == ParseStatus.STOCK, f"{raw} should be stock" + return target + + +def _analysis_result( + code: str, name: str, data_sources: str | None +) -> AnalysisResult: + result = AnalysisResult( + code=code, + name=name, + sentiment_score=60, + trend_prediction="震荡", + operation_advice="观望", + data_sources=data_sources or "", + ) + if data_sources is None: + setattr(result, "data_sources", None) + return result + + +def _render_aggregate_report( + result: AnalysisResult, + report_type: ReportType = ReportType.SIMPLE, +) -> str: + with patch( + "src.notification.get_config", + return_value=Config(stock_list=[], report_renderer_enabled=False), + ): + service = NotificationService() + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.notifier = service + return pipeline._generate_aggregate_report([result], report_type) + + +def _analysis_pipeline( + code: str, + name: str, + *, + enable_search: bool = False, + realtime_name: str | None = None, +) -> StockAnalysisPipeline: + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.fetcher_manager = MagicMock() + pipeline.fetcher_manager.build_not_supported_fundamental_context.return_value = { + "status": "not_supported", + "coverage": {}, + "source_chain": [], + "belong_boards": [], + } + pipeline.db = MagicMock() + pipeline.config = MagicMock() + pipeline.config.enable_realtime_quote = realtime_name is not None + pipeline.config.agent_mode = False + pipeline.config.agent_skills = [] + pipeline.config.report_language = "zh" + pipeline.config.fundamental_stage_timeout_seconds = 5.0 + pipeline.config.market_review_enabled = False + pipeline.config.daily_market_context_enabled = False + pipeline.config.report_type = "simple" + pipeline.search_service = MagicMock() if enable_search else None + if pipeline.search_service is not None: + pipeline.search_service.is_available = True + pipeline.search_service.search_comprehensive_intel.return_value = {} + pipeline.social_sentiment_service = None + pipeline.trend_analyzer = MagicMock() + pipeline.analyzer = MagicMock() + pipeline.analysis_skills = None + pipeline.query_source = "cli" + pipeline.save_context_snapshot = False + pipeline.progress_callback = None + pipeline.portfolio_context = None + pipeline.analysis_phase = "auto" + pipeline._emit_progress = MagicMock() + pipeline._load_daily_market_context = MagicMock(return_value=None) + pipeline._build_market_structure_context = MagicMock(return_value=None) + pipeline._load_persisted_intelligence_context = MagicMock(return_value=None) + pipeline._get_analysis_context_with_market_fallback = MagicMock( + return_value={"code": code, "stock_name": name} + ) + pipeline._enhance_context = MagicMock( + return_value={"code": code, "stock_name": name} + ) + pipeline._build_analysis_context_pack_outputs = MagicMock( + return_value=("", None) + ) + pipeline._build_legacy_analysis_artifacts = MagicMock() + pipeline._refresh_decision_action_for_final_result = MagicMock() + pipeline._build_query_context = MagicMock(return_value={}) + pipeline._build_context_snapshot = MagicMock(return_value={}) + pipeline._extract_decision_signal_after_history_save = MagicMock() + pipeline.db.save_analysis_history.return_value = 1 + pipeline.db.save_fundamental_snapshot = MagicMock() + pipeline.db.get_data_range.return_value = [] + pipeline.db.get_analysis_context.return_value = None + pipeline.analyzer.analyze.return_value = MagicMock( + success=True, + code=code, + name=name, + sentiment_score=60, + operation_advice="观望", + decision_type="hold", + confidence_level="中", + report_language="zh", + dashboard={}, + ) + if realtime_name is not None: + realtime_quote = MagicMock() + realtime_quote.name = realtime_name + realtime_quote.price = 3000.0 + realtime_quote.volume_ratio = 1.0 + realtime_quote.turnover_rate = 0.0 + realtime_quote.pe_ratio = None + realtime_quote.pb_ratio = None + realtime_quote.total_mv = None + realtime_quote.circ_mv = None + pipeline.fetcher_manager.get_realtime_quote.return_value = realtime_quote + return pipeline + + +class PipelineIndexTargetsTestCase(unittest.TestCase): + """V1 — CLI target construction produces expected type + canonical identity.""" + + def test_v1_sh_sz_csi_and_bare_stock_construct(self) -> None: + cases = [ + ("sh000016", ParseStatus.INDEX, "sh000016"), + ("000300.CSI", ParseStatus.INDEX, "sh000300"), + ("930955.CSI", ParseStatus.INDEX, "csi930955"), + ("000016", ParseStatus.STOCK, "sz000016"), + ("600519", ParseStatus.STOCK, "sh600519"), + ] + for raw, expected_type, expected_canonical in cases: + target = parse_analysis_target(raw) + self.assertEqual(target.asset_type, expected_type, raw) + self.assertEqual(target.canonical_id, expected_canonical, raw) + + def test_v1_index_display_name_is_chinese(self) -> None: + target = _index_target("sh000016") + self.assertEqual(target.display_code, "上证50") + self.assertIsNotNone(target.matched_index) + self.assertEqual(target.matched_index.display_name, "上证50") + + def test_v2_unregistered_csi_is_unsupported(self) -> None: + target = parse_analysis_target("930956.CSI") + self.assertEqual(target.asset_type, ParseStatus.UNSUPPORTED) + self.assertIsNotNone(target.unsupported_reason) + + def test_v3_alias_dedupe_by_canonical_id(self) -> None: + # Equivalent explicit aliases share one canonical scheduling key. + a = _index_target("sh000300") + b = _index_target("000300.CSI") + self.assertEqual(a.canonical_id, b.canonical_id) + self.assertEqual(a.canonical_id, "sh000300") + + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = MagicMock() + pipeline.config.single_stock_notify = False + pipeline.config.report_type = "simple" + pipeline.config.analysis_delay = 0 + pipeline.max_workers = 1 + pipeline.fetcher_manager = MagicMock() + pipeline.db = MagicMock() + pipeline.db.has_today_data.return_value = True + pipeline._save_local_report = MagicMock() + pipeline._send_notifications = MagicMock() + pipeline.process_single_stock = MagicMock(return_value=None) + + pipeline.run( + stock_codes=[a.canonical_id, b.canonical_id], + analysis_targets=[a, b], + dry_run=True, + send_notification=False, + ) + + self.assertEqual(pipeline.process_single_stock.call_count, 1) + submitted = pipeline.process_single_stock.call_args + self.assertEqual(submitted.args[0], "sh000300") + self.assertIs(submitted.kwargs["analysis_target"], a) + + def test_v3_stock_and_index_do_not_collapse(self) -> None: + # sh000016 (index) and bare 000016 (stock) are distinct identities. + index = _index_target("sh000016") + stock = _stock_target("000016") + self.assertEqual(index.canonical_id, "sh000016") + self.assertEqual(stock.canonical_id, "sz000016") + self.assertNotEqual(index.canonical_id, stock.canonical_id) + + +class PipelineMarketDatePropagationTestCase(unittest.TestCase): + """V4 — SH/SZ/CSI indices all use cn for market/date semantics.""" + + def _pipeline(self) -> StockAnalysisPipeline: + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.fetcher_manager = MagicMock() + pipeline.db = MagicMock() + return pipeline + + def test_v4_resume_target_date_uses_cn_for_csi_index(self) -> None: + target = _index_target("930955.CSI") + with patch( + "src.core.pipeline.get_effective_trading_date", + return_value=date(2026, 8, 26), + ) as mock_target, patch( + "src.core.pipeline.get_market_for_stock", return_value=None + ) as mock_market: + result = StockAnalysisPipeline._resolve_resume_target_date( + "csi930955", analysis_target=target + ) + self.assertEqual(result, date(2026, 8, 26)) + # market=cn must be passed to get_effective_trading_date, not None. + self.assertEqual(mock_target.call_args.args[0], "cn") + # get_market_for_stock must NOT be consulted for index targets. + mock_market.assert_not_called() + + def test_v4_resume_target_date_uses_cn_for_sh_index(self) -> None: + target = _index_target("sh000016") + with patch( + "src.core.pipeline.get_effective_trading_date", + return_value=date(2026, 8, 26), + ) as mock_target, patch( + "src.core.pipeline.get_market_for_stock", return_value=None + ) as mock_market: + StockAnalysisPipeline._resolve_resume_target_date( + "sh000016", analysis_target=target + ) + self.assertEqual(mock_target.call_args.args[0], "cn") + mock_market.assert_not_called() + + def test_v4_stock_still_uses_market_for_stock(self) -> None: + with patch( + "src.core.pipeline.get_effective_trading_date", + return_value=date(2026, 8, 26), + ), patch( + "src.core.pipeline.get_market_for_stock", return_value="cn" + ) as mock_market: + StockAnalysisPipeline._resolve_resume_target_date("600519") + mock_market.assert_called_once_with("600519") + + def test_v4_analysis_context_fallback_uses_cn_for_index(self) -> None: + pipeline = self._pipeline() + pipeline.db.get_analysis_context.return_value = None + target = _index_target("930955.CSI") + with patch( + "src.core.pipeline.get_market_for_stock", return_value=None + ) as mock_market: + result = pipeline._get_analysis_context_with_market_fallback( + "csi930955", analysis_target=target + ) + self.assertIsNone(result) + # For index targets the market is derived as cn, so the JP/KR/TW + # fallback branch is not entered and get_market_for_stock is not called. + mock_market.assert_not_called() + + +class PipelineCapabilityMatrixTestCase(unittest.TestCase): + """V5/V6 — INDEX_SKIP_MODULES zero bottom-layer calls for index targets.""" + + def test_index_skip_modules_contains_all_six(self) -> None: + self.assertEqual( + INDEX_SKIP_MODULES, + frozenset({ + "chip_distribution", + "fundamental", + "belong_boards", + "capital_flow", + "lhb", + "corporate_events", + }), + ) + + def test_v5_traditional_branch_skips_chip_and_fundamental(self) -> None: + pipeline = _analysis_pipeline("sh000016", "上证50") + + target = _index_target("sh000016") + result = pipeline.analyze_stock( + "sh000016", MagicMock(), "q1", analysis_target=target + ) + + # Bottom-layer provider calls for skipped modules must be zero. + pipeline.fetcher_manager.get_chip_distribution.assert_not_called() + pipeline.fetcher_manager.get_fundamental_context.assert_not_called() + pipeline.fetcher_manager.build_not_supported_fundamental_context.assert_called_once_with( + "sh000016", "index target: fundamental modules skipped" + ) + pipeline.fetcher_manager.build_failed_fundamental_context.assert_not_called() + pipeline.fetcher_manager.get_belong_boards.assert_not_called() + pipeline.db.save_fundamental_snapshot.assert_not_called() + # Supported modules still execute. + pipeline.analyzer.analyze.assert_called_once() + + def test_v6_agent_tool_filter_removes_index_incompatible_tools(self) -> None: + from src.agent.tools.registry import ToolRegistry, ToolDefinition, ToolParameter + + registry = ToolRegistry() + for name in ( + "get_realtime_quote", + "get_daily_history", + "get_chip_distribution", + "get_stock_info", + "get_capital_flow", + "get_analysis_context", + ): + registry.register( + ToolDefinition( + name=name, + description=name, + parameters=[ + ToolParameter( + name="stock_code", + type="string", + description="stock code", + ) + ], + handler=lambda **kw: {}, + ) + ) + + executor = MagicMock() + executor.tool_registry = registry + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + filtered = pipeline._filter_agent_tools_for_index(executor) + + names = set(filtered.tool_registry.list_names()) + self.assertIn("get_realtime_quote", names) + self.assertIn("get_daily_history", names) + self.assertIn("get_analysis_context", names) + self.assertNotIn("get_chip_distribution", names) + self.assertNotIn("get_stock_info", names) + self.assertNotIn("get_capital_flow", names) + + @patch("src.agent.factory.build_agent_executor") + def test_v6_agent_branch_applies_filter_and_target_aware_history( + self, mock_build_executor: MagicMock + ) -> None: + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = MagicMock() + pipeline.config.report_language = "zh" + pipeline.config.agent_skills = [] + pipeline.config.agent_litellm_model = "test-model" + pipeline.config.report_integrity_enabled = False + pipeline.analysis_skills = None + pipeline.social_sentiment_service = None + pipeline.search_service = None + pipeline._load_persisted_intelligence_context = MagicMock(return_value=None) + pipeline._ensure_agent_history = MagicMock() + pipeline._load_agent_analysis_context = MagicMock(return_value={}) + pipeline._build_agent_analysis_artifacts = MagicMock(return_value={}) + pipeline._build_analysis_context_pack_outputs = MagicMock( + return_value=("", None) + ) + pipeline._agent_result_to_analysis_result = MagicMock(return_value=None) + + executor = MagicMock() + executor.run.return_value = MagicMock(model="test-model", runtime_facts=None) + mock_build_executor.return_value = executor + pipeline._filter_agent_tools_for_index = MagicMock(return_value=executor) + target = _index_target("930955.CSI") + + result = pipeline._analyze_with_agent( + code="csi930955", + report_type=MagicMock(value="simple"), + query_id="q-agent-index", + stock_name="红利低波100", + realtime_quote=None, + chip_data=None, + analysis_target=target, + ) + + self.assertIsNone(result) + pipeline._filter_agent_tools_for_index.assert_called_once_with(executor) + pipeline._ensure_agent_history.assert_called_once_with( + "csi930955", analysis_target=target + ) + + +class PipelineDailySourceAttributionTestCase(unittest.TestCase): + """V10 — persisted daily providers reach history and existing reports.""" + + def test_v10_traditional_branch_persists_and_renders_daily_source(self) -> None: + pipeline = _analysis_pipeline("sh000016", "上证50") + pipeline._get_analysis_context_with_market_fallback.return_value = { + "today": {"data_source": "TencentFetcher"} + } + pipeline.analyzer.analyze.return_value = _analysis_result( + "sh000016", "上证50", "analysis:litellm" + ) + + result = pipeline.analyze_stock( + "sh000016", + ReportType.SIMPLE, + "q-daily-traditional", + analysis_target=_index_target("sh000016"), + ) + + self.assertIsNotNone(result) + self.assertEqual( + result.data_sources, + "analysis:litellm,daily:TencentFetcher", + ) + history_result = pipeline.db.save_analysis_history.call_args.kwargs["result"] + self.assertIs(history_result, result) + self.assertEqual(history_result.data_sources, result.data_sources) + # V8 — persisted history uses canonical code plus registry Chinese name. + self.assertEqual(history_result.code, "sh000016") + self.assertEqual(history_result.name, "上证50") + self.assertIn( + "*📋 数据来源：analysis:litellm,daily:TencentFetcher*", + _render_aggregate_report(result), + ) + + @patch("src.agent.factory.build_agent_executor") + def test_v10_agent_branch_persists_and_renders_daily_source( + self, mock_build_executor: MagicMock + ) -> None: + pipeline = _analysis_pipeline("csi930955", "红利低波100") + pipeline.config.agent_litellm_model = "test-model" + pipeline.config.report_integrity_enabled = False + pipeline._ensure_agent_history = MagicMock() + pipeline._load_agent_analysis_context = MagicMock( + return_value={"today": {"data_source": "AkshareFetcher"}} + ) + pipeline._build_agent_analysis_artifacts = MagicMock(return_value={}) + pipeline._agent_result_to_analysis_result = MagicMock( + return_value=_analysis_result( + "csi930955", "红利低波100", "agent:openai" + ) + ) + pipeline._persist_skill_opinion_samples_after_history_save = MagicMock() + + agent_result = MagicMock() + agent_result.model = "test-model" + agent_result.runtime_facts = None + executor = MagicMock() + executor.run.return_value = agent_result + mock_build_executor.return_value = executor + pipeline._filter_agent_tools_for_index = MagicMock(return_value=executor) + + result = pipeline._analyze_with_agent( + code="csi930955", + report_type=ReportType.SIMPLE, + query_id="q-daily-agent", + stock_name="红利低波100", + realtime_quote=None, + chip_data=None, + analysis_target=_index_target("930955.CSI"), + ) + + self.assertIsNotNone(result) + self.assertEqual( + result.data_sources, + "agent:openai,daily:AkshareFetcher", + ) + history_result = pipeline.db.save_analysis_history.call_args.kwargs["result"] + self.assertIs(history_result, result) + self.assertEqual(history_result.data_sources, result.data_sources) + # V8 — persisted history uses canonical code plus registry Chinese name. + self.assertEqual(history_result.code, "csi930955") + self.assertEqual(history_result.name, "红利低波100") + self.assertIn( + "*📋 数据来源：agent:openai,daily:AkshareFetcher*", + _render_aggregate_report(result, ReportType.BRIEF), + ) + + def test_v10_daily_source_deduplicates_complete_tokens_only(self) -> None: + cases = ( + ( + None, + "daily:AkshareFetcher", + ), + ( + "agent:openai,daily:AkshareFetcher", + "agent:openai,daily:AkshareFetcher", + ), + ( + "agent:openai,daily:AkshareFetcherV2", + "agent:openai,daily:AkshareFetcherV2,daily:AkshareFetcher", + ), + ) + + for existing_sources, expected_sources in cases: + with self.subTest(existing_sources=existing_sources): + pipeline = _analysis_pipeline("sh000016", "上证50") + pipeline._get_analysis_context_with_market_fallback.return_value = { + "today": {"data_source": "AkshareFetcher"} + } + pipeline.analyzer.analyze.return_value = _analysis_result( + "sh000016", "上证50", existing_sources + ) + + result = pipeline.analyze_stock( + "sh000016", + ReportType.SIMPLE, + "q-daily-dedupe", + analysis_target=_index_target("sh000016"), + ) + + self.assertIsNotNone(result) + self.assertEqual(result.data_sources, expected_sources) + + def test_v10_invalid_daily_sources_fail_open(self) -> None: + invalid_contexts = ( + {}, + {"today": {}}, + {"today": {"data_source": None}}, + {"today": {"data_source": 42}}, + {"today": {"data_source": " "}}, + {"today": {"data_source": " Unknown "}}, + {"today": {"data_source": " realtime:TencentFetcher "}}, + {"today": {"data_source": "Realtime:TencentFetcher"}}, + {"today": {"data_source": "REALTIME:TencentFetcher"}}, + ) + + for context in invalid_contexts: + with self.subTest(context=context): + pipeline = _analysis_pipeline("sh000016", "上证50") + pipeline._get_analysis_context_with_market_fallback.return_value = context + pipeline.analyzer.analyze.return_value = _analysis_result( + "sh000016", "上证50", "analysis:litellm" + ) + + result = pipeline.analyze_stock( + "sh000016", + ReportType.SIMPLE, + "q-daily-invalid", + analysis_target=_index_target("sh000016"), + ) + + self.assertIsNotNone(result) + self.assertEqual(result.data_sources, "analysis:litellm") + + def test_v11_stock_result_does_not_gain_index_daily_attribution(self) -> None: + pipeline = _analysis_pipeline("600519", "贵州茅台") + pipeline._get_analysis_context_with_market_fallback.return_value = { + "today": {"data_source": "AkshareFetcher"} + } + pipeline.analyzer.analyze.return_value = _analysis_result( + "600519", "贵州茅台", "analysis:litellm" + ) + + result = pipeline.analyze_stock( + "600519", + ReportType.SIMPLE, + "q-stock-daily-source", + analysis_target=_stock_target("600519"), + ) + + self.assertIsNotNone(result) + self.assertEqual(result.data_sources, "analysis:litellm") + + +class PipelineSearchSemanticsTestCase(unittest.TestCase): + """V7 — index search query subject is the Chinese name only.""" + + def test_v7_traditional_search_uses_name_only_for_index(self) -> None: + pipeline = _analysis_pipeline( + "sh000016", + "上证50", + enable_search=True, + realtime_name="provider alias", + ) + + target = _index_target("sh000016") + pipeline.analyze_stock("sh000016", MagicMock(), "q1", analysis_target=target) + + # The search query subject must be the Chinese name only — no canonical + # code / six-digit machine code in the provider query. + call_kwargs = pipeline.search_service.search_comprehensive_intel.call_args.kwargs + self.assertEqual(call_kwargs["stock_code"], "") + self.assertEqual(call_kwargs["stock_name"], "上证50") + + def test_v7_stock_search_keeps_code(self) -> None: + pipeline = _analysis_pipeline("600519", "贵州茅台", enable_search=True) + + pipeline.analyze_stock("600519", MagicMock(), "q1") + + call_kwargs = pipeline.search_service.search_comprehensive_intel.call_args.kwargs + self.assertEqual(call_kwargs["stock_code"], "600519") + + def test_v7_provider_queries_contain_only_chinese_index_name(self) -> None: + service = SearchService( + searxng_public_instances_enabled=False, + news_max_age_days=3, + news_strategy_profile="short", + ) + provider = MagicMock() + provider.name = "DummyProvider" + provider.is_available = True + + def search(query: str, **_kwargs) -> SearchResponse: + return SearchResponse( + query=query, + results=[], + provider="DummyProvider", + success=True, + ) + + provider.search.side_effect = search + service._providers = [provider] + + service.search_comprehensive_intel( + stock_code="", + stock_name="上证50", + max_searches=5, + ) + + queries = [call.args[0] for call in provider.search.call_args_list] + self.assertTrue(queries) + for query in queries: + self.assertIn("上证50", query) + self.assertNotRegex(query, re.compile(r"\d{6}")) + self.assertNotIn("sh000016", query.casefold()) + + +class PipelineBatchFailureIsolationTestCase(unittest.TestCase): + """V2/V10 — unsupported targets rejected before provider calls; batch continues.""" + + def test_v2_run_filters_unsupported_before_prefetch(self) -> None: + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = MagicMock() + pipeline.config.refresh_stock_list = MagicMock() + pipeline.config.stock_list = [] + pipeline.config.single_stock_notify = False + pipeline.config.report_type = "simple" + pipeline.config.analysis_delay = 0 + pipeline.max_workers = 2 + pipeline.fetcher_manager = MagicMock() + pipeline.fetcher_manager.prefetch_daily_klines.return_value = 0 + pipeline.fetcher_manager.prefetch_realtime_quotes.return_value = 0 + pipeline.db = MagicMock() + pipeline._save_local_report = MagicMock() + pipeline._send_notifications = MagicMock() + pipeline._send_single_stock_notification = MagicMock() + pipeline._resolve_resume_target_date = MagicMock( + return_value=date(2026, 8, 26) + ) + pipeline.process_single_stock = MagicMock(return_value=None) + + unsupported = parse_analysis_target("930956.CSI") + supported = [ + _stock_target("600519"), + _stock_target("000016"), + _stock_target("AAPL"), + _stock_target("TSLA"), + _stock_target("hk00700"), + ] + codes = ["930956.CSI", *[target.canonical_id for target in supported]] + targets = [unsupported, *supported] + + with patch.object( + StockAnalysisPipeline, "process_single_stock", pipeline.process_single_stock + ): + results = pipeline.run( + stock_codes=codes, + analysis_targets=targets, + send_notification=False, + dry_run=False, + ) + + # Five supported targets still cross the prefetch threshold, proving the + # unsupported target was removed from every provider request first. + supported_codes = codes[1:] + pipeline.fetcher_manager.prefetch_daily_klines.assert_called_once_with( + supported_codes, days=30 + ) + pipeline.fetcher_manager.prefetch_realtime_quotes.assert_called_once_with( + supported_codes + ) + pipeline.fetcher_manager.prefetch_stock_names.assert_called_once_with( + supported_codes, use_bulk=False + ) + self.assertEqual(pipeline.process_single_stock.call_count, 5) + submitted_codes = { + call.args[0] for call in pipeline.process_single_stock.call_args_list + } + self.assertEqual(submitted_codes, set(supported_codes)) + self.assertEqual(results, []) + + +class PipelineDryRunTestCase(unittest.TestCase): + """V11 — dry-run preserves target identity through resume checks.""" + + def test_v11_dry_run_uses_target_aware_resume_date(self) -> None: + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = MagicMock() + pipeline.config.refresh_stock_list = MagicMock() + pipeline.config.stock_list = [] + pipeline.config.single_stock_notify = False + pipeline.config.report_type = "simple" + pipeline.config.analysis_delay = 0 + pipeline.max_workers = 1 + pipeline.fetcher_manager = MagicMock() + pipeline.db = MagicMock() + pipeline.db.has_today_data.return_value = True + pipeline._save_local_report = MagicMock() + pipeline._send_notifications = MagicMock() + pipeline._send_single_stock_notification = MagicMock() + pipeline.process_single_stock = MagicMock(return_value=None) + + target = _index_target("930955.CSI") + with patch.object( + StockAnalysisPipeline, + "_resolve_resume_target_date", + return_value=date(2026, 8, 26), + ) as mock_resolve: + pipeline.run( + stock_codes=["csi930955"], + dry_run=True, + send_notification=False, + analysis_targets=[target], + ) + + # The resume date must be resolved with the index target so market=cn + # governs the dry-run success check. + self.assertEqual(mock_resolve.call_count, 1) # dry-run success count + for call in mock_resolve.call_args_list: + self.assertEqual(call.kwargs.get("analysis_target"), target) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_report_language.py b/tests/test_report_language.py index 166ded2f1..e0a221639 100644 --- a/tests/test_report_language.py +++ b/tests/test_report_language.py @@ -104,10 +104,17 @@ class KoreanReportLanguageTestCase(unittest.TestCase): def test_korean_labels_cover_full_english_key_set(self) -> None: ko_labels = get_report_labels("ko") en_labels = get_report_labels("en") + zh_labels = get_report_labels("zh") self.assertEqual(set(ko_labels.keys()), set(en_labels.keys())) + self.assertEqual(set(zh_labels.keys()), set(en_labels.keys())) self.assertEqual(ko_labels["dashboard_title"], "결정 대시보드") self.assertEqual(ko_labels["risk_alerts_label"], "리스크 경보") + def test_data_sources_label_is_localized(self) -> None: + self.assertEqual(get_report_labels("zh")["data_sources_label"], "数据来源") + self.assertEqual(get_report_labels("en")["data_sources_label"], "Data Sources") + self.assertEqual(get_report_labels("ko")["data_sources_label"], "데이터 출처") + def test_korean_sentiment_label_bands(self) -> None: self.assertEqual(get_sentiment_label(80, "ko"), "매우 낙관") self.assertEqual(get_sentiment_label(40, "ko"), "중립") diff --git a/tests/test_search_tools_persistence.py b/tests/test_search_tools_persistence.py index 3eb23d6a5..97e94ec94 100644 --- a/tests/test_search_tools_persistence.py +++ b/tests/test_search_tools_persistence.py @@ -80,6 +80,61 @@ class SearchToolsPersistenceTest(unittest.TestCase): query_context=None, ) + def test_index_news_search_uses_name_only_and_persists_canonical_code(self) -> None: + response = _response("红利低波100 最新消息") + 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("csi930955", "stale name") + + self.assertTrue(result["success"]) + service.search_stock_news.assert_called_once_with( + "", "红利低波100", max_results=5 + ) + db.save_news_intel.assert_called_once_with( + code="csi930955", + name="红利低波100", + dimension="latest_news", + query=response.query, + response=response, + query_context=None, + ) + + def test_index_comprehensive_search_uses_name_only(self) -> None: + latest = _response("上证50 最新消息") + service = SimpleNamespace( + is_available=True, + search_comprehensive_intel=MagicMock( + return_value={"latest_news": latest} + ), + 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("sh000016", "stale name") + + self.assertEqual(result["report"], "report") + service.search_comprehensive_intel.assert_called_once_with( + stock_code="", + stock_name="上证50", + max_searches=6, + ) + db.save_news_intel.assert_called_once_with( + code="sh000016", + name="上证50", + 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(