feat: 支持通过 --stocks 一次性分析板块指数 (#2300)

* feat: add one-shot index analysis via --stocks with structured AnalysisTarget pipeline

* fix: render index data sources in aggregate reports

* fix: refresh index registry before --stocks parsing and filter indices on CN holidays

---------

Co-authored-by: zhulinsen <42829555+ZhuLinsen@users.noreply.github.com>
This commit is contained in:
Elvis Wang
2026-08-28 21:56:48 +08:00
committed by GitHub
parent ee635a35bf
commit eb1333a678
27 changed files with 2512 additions and 104 deletions

View File

@@ -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)

View File

@@ -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,

View File

@@ -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 实时行情

View File

@@ -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 只显示数字和英文、中文或韩文内容消失。

View File

@@ -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` 可强制在非交易日执行。
指数实时行情使用独立固定链:腾讯 → 新浪 → 东财单股接口 → TickFlowSH/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 后运行:

View File

@@ -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:

69
main.py
View File

@@ -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:
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

View File

@@ -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):

View File

@@ -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,
)

View File

@@ -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,6 +498,16 @@ class StockAnalysisPipeline:
self._emit_progress(18, f"{code}:正在获取行情与筹码数据")
# 获取股票名称(先走轻量名称路径,后续若 realtime_quote 有 name 再覆盖)
# 指数目标使用注册表中文显示名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: 获取实时行情(量比、换手率等)- 使用统一入口,自动故障切换
@@ -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,7 +537,11 @@ class StockAnalysisPipeline:
stock_name = f'股票{code}'
# Step 2: 获取筹码分布 - 使用统一入口,带熔断保护
# 指数目标跳过筹码分布INDEX_SKIP_MODULES 能力矩阵)
chip_data = None
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:
@@ -531,7 +574,22 @@ class StockAnalysisPipeline:
# Step 2.5: 基本面能力聚合(统一入口,异常降级)
# - 失败时返回 partial/failed不影响既有技术面/新闻链路
# - 关闭开关时仍返回 not_supported 结构
# - 指数目标跳过基本面/板块/资金流/龙虎榜/公司事件INDEX_SKIP_MODULES
fundamental_context = None
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,
"index target: fundamental modules skipped",
)
else:
try:
fundamental_context = self.fetcher_manager.get_fundamental_context(
code,
@@ -545,6 +603,12 @@ class StockAnalysisPipeline:
logger.warning(f"{stock_name}({code}) 基本面聚合失败: {e}")
fundamental_context = self.fetcher_manager.build_failed_fundamental_context(code, str(e))
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,
@@ -559,6 +623,7 @@ class StockAnalysisPipeline:
)
# P0: write-only snapshot, fail-open, no read dependency on this table.
if not is_index:
try:
self.db.save_fundamental_snapshot(
query_id=query_id,
@@ -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
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,12 +1961,17 @@ 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
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,6 +2754,7 @@ class StockAnalysisPipeline:
)
if not enable_realtime_tech:
return df
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):
@@ -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,11 +3209,19 @@ 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.
"""
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)
@@ -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("跳过重复分析目标 %scanonical_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)

View File

@@ -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)

View File

@@ -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": "정배열",

View File

@@ -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

View File

@@ -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":

View File

@@ -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):

View File

@@ -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 }}*

View File

@@ -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 %}

View File

@@ -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

View File

@@ -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.

View File

@@ -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()

View File

@@ -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(

View File

@@ -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 开市保留;
# 市场未知的非指数 codeXYZ123继续 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(

View File

@@ -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)

View File

@@ -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()

View File

@@ -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()

View File

@@ -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"), "중립")

View File

@@ -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(