feat: 扩展 A 股指数注册表与 CSI provider 路由 (#2267)

* feat: add JSON-backed index registry

* fix: address index registry review feedback

* fix: validate raw duplicate index aliases

* fix: converge CSI task identities and registry fallback

* docs: align index identity terminology
This commit is contained in:
Elvis Wang
2026-08-25 22:32:04 +08:00
committed by GitHub
parent 262cf5d5de
commit 1b42907605
21 changed files with 2772 additions and 147 deletions
+32 -1
View File
@@ -31676,5 +31676,36 @@
["018260.KS","018260.KS","三星SDS","sanxingSDS","sxSDS",["Samsung SDS","삼성에스디에스"],"KR","stock",true,100],
["011200.KS","011200.KS","HMM","HMM","HMM",["现代商船"],"KR","stock",true,100],
["090430.KS","090430.KS","爱茉莉太平洋","aimolitaipingyang","amltpy",["Amorepacific","아모레퍼시픽"],"KR","stock",true,100],
["086520.KQ","086520.KQ","Ecopro","Ecopro","Ecopro",["에코프로"],"KR","stock",true,100]
["086520.KQ","086520.KQ","Ecopro","Ecopro","Ecopro",["에코프로"],"KR","stock",true,100],
["csi930955","930955.CSI","红利低波100","honglidibo100","hldb100",[],"CN","index",true,100],
["csi931052","931052.CSI","国信价值","guoxinjiazhi","gxjz",[],"CN","index",true,100],
["csi931446","931446.CSI","东证红利低波","dongzhenghonglidibo","dzhldb",[],"CN","index",true,100],
["csi931643","931643.CSI","科创创业50","kechuangchuangye50","kccy50",[],"CN","index",true,100],
["csi932365","932365.CSI","中证现金流","zhongzhengxianjinliu","zzxjl",[],"CN","index",true,100],
["csi932366","932366.CSI","300现金流","300xianjinliu","300xjl",[],"CN","index",true,100],
["sh000001","sh000001","上证指数","shangzhengzhishu","szzs",["000001.SH"],"CN","index",true,100],
["sh000009","sh000009","上证380","shangzheng380","sz380",["000009.SH"],"CN","index",true,100],
["sh000010","sh000010","上证180","shangzheng180","sz180",["000010.SH"],"CN","index",true,100],
["sh000015","sh000015","红利指数","honglizhishu","hlzs",["000015.SH"],"CN","index",true,100],
["sh000016","sh000016","上证50","shangzheng50","sz50",["000016.SH"],"CN","index",true,100],
["sh000300","sh000300","沪深300","hushen300","hs300",["sz399300","000300.SH","000300.CSI"],"CN","index",true,100],
["sh000688","sh000688","科创50","kechuang50","kc50",["000688.SH"],"CN","index",true,100],
["sh000852","sh000852","中证1000","zhongzheng1000","zz1000",["000852.SH"],"CN","index",true,100],
["sh000905","sh000905","中证500","zhongzheng500","zz500",["sz399905","000905.SH","000905.CSI"],"CN","index",true,100],
["sh000906","sh000906","中证800","zhongzheng800","zz800",["000906.SH"],"CN","index",true,100],
["sh000922","sh000922","中证红利","zhongzhenghongli","zzhl",["sz399922","000922.SH","000922.CSI"],"CN","index",true,100],
["sh000941","sh000941","新能源","xinnengyuan","xny",["000941.SH"],"CN","index",true,100],
["sz399001","sz399001","深证成指","shenzhengchengzhi","szcz",["399001.SZ"],"CN","index",true,100],
["sz399005","sz399005","中小100","zhongxiao100","zx100",["399005.SZ"],"CN","index",true,100],
["sz399006","sz399006","创业板指","chuangyebanzhi","cybz",["399006.SZ"],"CN","index",true,100],
["sz399296","sz399296","创成长","chuangchengzhang","ccz",["399296.SZ"],"CN","index",true,100],
["sz399303","sz399303","国证2000","guozheng2000","gz2000",["399303.SZ"],"CN","index",true,100],
["sz399324","sz399324","深证红利","shenzhenghongli","szhl",["399324.SZ"],"CN","index",true,100],
["sz399330","sz399330","深证100","shenzheng100","sz100",["399330.SZ"],"CN","index",true,100],
["sz399967","sz399967","中证军工","zhongzhengjungong","zzjg",["399967.SZ"],"CN","index",true,100],
["sz399975","sz399975","证券公司","zhengquangongsi","zqgs",["399975.SZ"],"CN","index",true,100],
["sz399986","sz399986","中证银行","zhongzhengyinhang","zzyh",["399986.SZ"],"CN","index",true,100],
["sz399989","sz399989","中证医疗","zhongzhengyiliao","zzyl",["399989.SZ"],"CN","index",true,100],
["sz399997","sz399997","中证白酒","zhongzhengbaijiu","zzbj",["399997.SZ"],"CN","index",true,100],
["sz980092","sz980092","自由现金流","ziyouxianjinliu","zyxjl",["980092.SZ"],"CN","index",true,100]
]
@@ -187,6 +187,56 @@ describe('stockIndexLoader', () => {
const fetchCallArgs = mockFetch.mock.calls[0][0];
expect(fetchCallArgs).toContain('?_t=');
});
test('filters out assetType=index rows from the returned data', async () => {
const withIndex = [
...mockIndexData,
{
canonicalCode: 'sh000300',
displayCode: 'sh000300',
nameZh: '沪深300',
pinyinFull: 'hushen300',
pinyinAbbr: 'hs300',
aliases: ['000300.SH'],
market: 'CN',
assetType: 'index',
active: true,
popularity: 100,
},
];
mockFetch.mockResolvedValueOnce({
ok: true,
json: async () => withIndex,
} as unknown as Response);
const result = await loadStockIndex();
expect(result.loaded).toBe(true);
expect(result.fallback).toBe(false);
// Index rows are hidden from the current consumers.
expect(result.data.some(item => item.assetType === 'index')).toBe(false);
// Stock rows are preserved.
expect(result.data).toHaveLength(mockIndexData.length);
});
test('filters index rows from compressed tuple payload', async () => {
const compressedWithIndex = [
['600519.SH', '600519', '贵州茅台', 'guizhoumaotai', 'gzmt', ['茅台'], 'CN', 'stock', true, 100],
['sh000300', 'sh000300', '沪深300', 'hushen300', 'hs300', ['000300.SH'], 'CN', 'index', true, 100],
];
mockFetch.mockResolvedValueOnce({
ok: true,
json: async () => compressedWithIndex,
} as unknown as Response);
const result = await loadStockIndex();
expect(result.loaded).toBe(true);
expect(result.data).toHaveLength(1);
expect(result.data[0].canonicalCode).toBe('600519.SH');
});
});
describe('compressIndex - Compress index', () => {
+7 -1
View File
@@ -39,8 +39,14 @@ export async function loadStockIndex(): Promise<IndexLoadResult> {
? unpackTuples(data as StockIndexTuple[])
: data as StockIndexItem[];
// The shared payload may now carry ``assetType=index`` rows, but the
// current autocomplete/popular/group consumers must not see them. Filter
// index rows out before constructing the successful result so stock/ETF
// behaviour is unchanged.
const visibleItems = items.filter(item => item.assetType !== 'index');
return {
data: items,
data: visibleItems,
loaded: true,
fallback: false,
};
+24
View File
@@ -886,6 +886,13 @@ class DataFetcherManager:
if entry is None:
return ""
exchange = entry.exchange.upper()
if exchange == "CSI":
# CSI indices are only supported by AkShare (``csi{code}``); the
# other providers in the fixed daily chain return an empty symbol
# so the caller records an ``unsupported`` provider-run and skips.
if fetcher_name == "AkshareFetcher":
return f"csi{entry.bare_code}"
return ""
if exchange not in {"SH", "SZ"}:
return ""
if fetcher_name in {"TencentFetcher", "AkshareFetcher"}:
@@ -1637,6 +1644,9 @@ class DataFetcherManager:
raw_stock_code = (stock_code or "").strip()
target = parse_analysis_target(raw_stock_code)
self._warn_bare_index_conflict(target)
if target.asset_type == ParseStatus.UNSUPPORTED:
reason = target.unsupported_reason or "unsupported analysis target"
raise DataFetchError(f"{raw_stock_code}: {reason}")
if target.asset_type == ParseStatus.INDEX:
return self._get_cn_index_daily_data(
target,
@@ -2625,6 +2635,13 @@ class DataFetcherManager:
raw_stock_code = (stock_code or "").strip()
target = parse_analysis_target(raw_stock_code)
self._warn_bare_index_conflict(target)
if target.asset_type == ParseStatus.UNSUPPORTED:
logger.warning(
"[股票名称] 跳过不支持的输入 %s: %s",
raw_stock_code,
target.unsupported_reason or "unsupported analysis target",
)
return ""
if target.asset_type == ParseStatus.INDEX:
return self._get_cn_index_name(target)
@@ -2761,6 +2778,13 @@ class DataFetcherManager:
normalized_codes: List[str] = []
for code in stock_codes:
target = parse_analysis_target(code)
if target.asset_type == ParseStatus.UNSUPPORTED:
logger.warning(
"[股票名称] 预取跳过不支持的输入 %s: %s",
code,
target.unsupported_reason or "unsupported analysis target",
)
continue
normalized_codes.append(
target.canonical_id
if target.asset_type == ParseStatus.INDEX
+11
View File
@@ -14,6 +14,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
<!-- 新条目格式:- [类型] 描述(类型取值:新功能/改进/修复/文档/测试/chore)-->
<!-- 每条独立一行追加到本段末尾,无需分类标题,合并时冲突最小 -->
- [新功能] 建立唯一、可生成、可校验、可降级的指数身份注册表:由 `scripts/stock_index_seeds/index_registry.csv` 的 31 项 manifest 确定性合并进 `apps/dsa-web/public/stocks.index.json`,运行时唯一真源为 JSON 中通过校验的 `active=true`/`assetType=index` 行,移除 `stock_list_parser` 的 5 项硬编码白名单;支持 `--index-only` 生成与字节稳定输出。
- [新功能] 补齐显式 SH/SZ/CSI 指数 alias 收敛与 CSI 身份:`sh000300`/`000300.SH`/`sz399300`/`399300.SZ`/`000300.CSI` 均解析到 `sh000300`,`csi930955`/`930955.CSI` 解析到 `csi930955`;未登记 `.CSI` 输入返回 `unsupported`;裸数字恒为 stock 并仅通过 `matched_index` 暴露歧义。
- [新功能] 数据管理器按 SH/SZ/CSI 支持矩阵映射 provider symbol:CSI 仅 AkShare 支持(`csi{code}`),Tencent/TickFlow/Yahoo 返回空 symbol 并记录 `unsupported` provider-run,不触发指数健康熔断。
- [改进] 存储层 `_derive_canonical_id` 统一为 parser 推导(裸码=stock、显式指数=index),并新增幂等分批修复历史裸码错误 canonical 串桶(`000001`/`000016`/`000688`/`930955` 等),显式指数行与正确 stock 行不受影响,registry 为空时修复 no-op。
- [改进] Web loader 完整解压含 index 的共享 payload,但在返回给 autocomplete/popular/group 消费面前过滤 `assetType=index`,股票与 ETF 行为保持不变。
- [测试] 为生成器、loader、parser、provider 路由、存储修复与 Web 门槛补充 TDD 回归锚点。
- [修复] PR #2267 review 收敛 CSI 显式身份:将 `csi` prefix(canonical)与 `.CSI` suffix(显式 alias)在 parser/build/runtime 规范化器中分离,未登记显式 `csiNNNNNN`/`NNNNNN.CSI` 一律返回 `unsupported`(不再落入美股或猜测 SH/SZ),并防止未登记 `csi000300` 被等价成已登记 `000300.CSI` alias;存储 `_derive_canonical_id` 对 unsupported 输入返回 NULL,避免进入持久化 canonical 桶。
- [修复] 在 seed、build entry 与 runtime candidate 三层严格校验 alias 唯一性与整数 popularity:NFKC/casefold 等价 alias 跨条目冲突被拒绝(无静默覆盖),非负整数之外(小数/布尔/负值/字符串)popularity 一律拒绝,整数 `100` 保持有效。
- [修复] 收敛已登记 CSI 显式身份在 resolver、任务去重键与历史候选中的分裂:`csi930955`/`930955.CSI`/`CSI930955` 统一解析为 parser canonical `csi930955`,未登记 `csi930956`/`930956.CSI` 保持既有降级语义;`is_code_like()`、REST/watchlist 输入边界与完整 Pipeline 透传不变。
- [修复] 阻止任意更新的非 bundled 指数候选(含 legacy `static` 子集)在 remote 缺失/损坏时以 active-index 子集覆盖 bundled baseline:所有非 bundled 候选必须为 bundled active-index canonical 集合的合法超集,否则回退 bundled 并记录 WARNING。
## [3.31.0] - 2026-08-23
### 发布亮点
+405 -17
View File
@@ -11,14 +11,17 @@ Input:
Output: apps/dsa-web/public/stocks.index.json
Usage:
python3 scripts/generate_index_from_csv.py # 默认使用 Tushare
python3 scripts/generate_index_from_csv.py --source akshare
python3 scripts/generate_index_from_csv.py --test # 测试模式
python scripts/generate_index_from_csv.py # 默认使用 Tushare
python scripts/generate_index_from_csv.py --source akshare
python scripts/generate_index_from_csv.py --test # 测试模式
python scripts/generate_index_from_csv.py --index-only --test # 仅合并指数 seed
"""
import argparse
import csv
import json
import math
import os
import re
import sys
import unicodedata
@@ -28,6 +31,8 @@ from typing import List, Dict, Any, Optional
# Add the project root to sys.path.
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.services.stock_index_remote_service import validate_stock_index_payload
try:
from pypinyin import lazy_pinyin, Style
PYPINYIN_AVAILABLE = True
@@ -616,6 +621,355 @@ def compress_index(index: List[Dict[str, Any]]) -> List[List]:
return compressed
# ---------------------------------------------------------------------------
# Index registry seed — build-time manifest merge.
# ---------------------------------------------------------------------------
_INDEX_REGISTRY_SEED_PATH = Path(__file__).parent / "stock_index_seeds" / "index_registry.csv"
_INDEX_NAMESPACE_RE = re.compile(r"^(sh|sz|csi)\d{6}$")
_EXPLICIT_INDEX_ALIAS_RE = re.compile(
r"^(?:(?:sh|sz|csi)\d{6}|\d{6}\.(?:sh|sz|csi))$"
)
def load_index_registry_seed(seed_path: Optional[Path] = None) -> List[Dict[str, Any]]:
"""Load the approved index-registry seed CSV into raw row dicts.
Columns: ``canonical_code,display_code,name_zh,aliases,name_source,popularity``.
``aliases`` uses the existing ``|``-separated ``parse_aliases()`` convention.
"""
path = seed_path or _INDEX_REGISTRY_SEED_PATH
if not path.is_file():
raise FileNotFoundError(f"index registry seed not found: {path}")
rows: List[Dict[str, Any]] = []
# Normalized identity keys must not map to more than one canonical within
# the seed. A NFKC/casefold-equivalent duplicate alias owned by two entries
# (e.g. ``csi930955`` and ``CSI930955`` split across rows) would otherwise
# silently overwrite one identity, so it is rejected at the build-time
# boundary instead of at runtime. A key that equals its own row's canonical
# (e.g. alias ``000300.SH`` on row ``sh000300``) is legitimate and skipped.
seen_identity_keys: Dict[str, str] = {}
with open(path, "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
canonical = (row.get("canonical_code") or "").strip()
display = (row.get("display_code") or "").strip()
name = (row.get("name_zh") or "").strip()
if not canonical or not display or not name:
raise ValueError(f"index registry seed row missing required field: {row}")
raw_aliases = [
alias.strip()
for alias in str(row.get("aliases") or "").split("|")
if alias.strip()
]
_validate_unique_index_aliases(raw_aliases, canonical)
aliases = parse_aliases(row)
_validate_unique_index_aliases(aliases, canonical)
popularity_raw = (row.get("popularity") or "100").strip() or "100"
try:
popularity = int(popularity_raw)
except (TypeError, ValueError) as exc:
raise ValueError(
f"index registry seed popularity must be a plain integer, "
f"got {popularity_raw!r} for canonical {canonical!r}"
) from exc
if popularity < 0:
raise ValueError(
f"index registry seed popularity must be non-negative: "
f"{popularity!r} for canonical {canonical!r}"
)
# Reject a normalized identity key that maps to a different canonical.
for key in [canonical, display] + aliases:
norm_key = _normalize_index_key(key)
if not norm_key:
continue
existing = seen_identity_keys.get(norm_key)
if existing is not None and existing != canonical:
raise ValueError(
f"seed identity key {key!r} normalizes to {norm_key!r} "
f"already owned by canonical {existing!r}"
)
seen_identity_keys[norm_key] = canonical
rows.append({
"canonical_code": canonical,
"display_code": display,
"name_zh": name,
"aliases": aliases,
"name_source": (row.get("name_source") or "").strip(),
"popularity": popularity,
})
return rows
def _normalize_index_key(value: str) -> str:
"""Normalize resolver keys while keeping CSI suffix aliases distinct."""
normalized = unicodedata.normalize(
"NFKC", str(value or "")
).strip().casefold()
prefix_match = re.fullmatch(r"(sh|sz)(\d{6})", normalized)
if prefix_match:
return f"{prefix_match.group(1)}{prefix_match.group(2)}"
suffix_match = re.fullmatch(r"(\d{6})\.(sh|sz)", normalized)
if suffix_match:
return f"{suffix_match.group(2)}{suffix_match.group(1)}"
return normalized
def _validate_unique_index_aliases(aliases: Any, canonical: str) -> None:
"""Reject duplicate aliases after NFKC/case-insensitive normalization."""
if not isinstance(aliases, list):
raise ValueError(f"index aliases must be a list: {canonical!r}")
seen: set[str] = set()
for alias in aliases:
if not isinstance(alias, str):
raise ValueError(f"index alias must be a string: {canonical!r}")
normalized = _normalize_index_key(alias)
if normalized in seen:
raise ValueError(
f"duplicate index alias after normalization for {canonical!r}: {alias!r}"
)
seen.add(normalized)
def build_index_entries_from_seed(seed_rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Convert seed rows into the 10-column index tuple dicts.
``display_code`` is honored from the seed (SH/SZ display equals canonical;
CSI display is ``{code}.CSI``). All rows are ``market=CN``,
``assetType=index``, ``active=True``, ``popularity`` from seed.
"""
entries: List[Dict[str, Any]] = []
for row in seed_rows:
canonical = row["canonical_code"]
_validate_unique_index_aliases(row.get("aliases"), canonical)
popularity = row.get("popularity", 100)
if (
isinstance(popularity, bool)
or not isinstance(popularity, int)
or popularity < 0
):
raise ValueError(
f"index popularity must be a non-negative integer: {canonical!r}"
)
display = (row.get("display_code") or "").strip() or canonical
pinyin_full, pinyin_abbr = generate_pinyin(row["name_zh"])
entries.append({
"canonicalCode": canonical,
"displayCode": display,
"nameZh": row["name_zh"],
"pinyinFull": pinyin_full,
"pinyinAbbr": pinyin_abbr,
"aliases": list(row["aliases"]),
"market": "CN",
"assetType": "index",
"active": True,
"popularity": popularity,
})
return entries
def validate_index_registry(
entries: List[Dict[str, Any]],
non_index_rows: Optional[List[List[Any]]] = None,
) -> None:
"""Semantic validation for the index registry (build-time and candidates).
Rules (implementation-contracts.md §Semantic Validation):
1. canonical matches ``^(sh|sz|csi)\\d{6}$``; SH/SZ display == canonical,
CSI display == ``{code}.CSI``.
2. market=CN, assetType=index, active=True, non-empty name, valid pinyin,
finite numeric popularity.
3. canonical/display/alias normalize to exactly one canonical within the set.
4. Index explicit keys must not collide with active stock/ETF keys; bare
numeric display/alias rejected.
5. Text aliases rejected from identity resolver seed.
6. Each namespace has at least one daily provider in the manifest matrix.
``non_index_rows`` carries the active stock/ETF compressed tuples from the
same payload so rule 4 can reject an index canonical/display/alias that
collides with a stock/ETF identity after normalization.
"""
if not entries:
return
canonical_map: Dict[str, str] = {}
resolver_map: Dict[str, str] = {}
bare_conflicts: Dict[str, str] = {}
# Active stock/ETF identity keys (canonical/display/aliases) that an index
# explicit key must never collide with after normalization.
stock_keys: Dict[str, str] = {}
for row in non_index_rows or []:
if not isinstance(row, list) or len(row) < 10:
continue
if str(row[7] or "").strip() == "index":
continue
if row[8] is not True:
continue
stock_canonical = str(row[0] or "").strip()
for key in [row[0], row[1]] + list(row[5] if isinstance(row[5], list) else []):
norm_key = _normalize_index_key(key)
if norm_key:
stock_keys.setdefault(norm_key, stock_canonical)
for entry in entries:
canonical = str(entry["canonicalCode"] or "").strip()
display = str(entry["displayCode"] or "").strip()
name = str(entry["nameZh"] or "").strip()
market = str(entry["market"] or "").strip()
asset_type = str(entry["assetType"] or "").strip()
active = entry["active"]
popularity = entry["popularity"]
aliases = entry.get("aliases")
if not _INDEX_NAMESPACE_RE.match(canonical):
raise ValueError(f"index canonical must match ^(sh|sz|csi)\\d{{6}}$: {canonical!r}")
namespace = canonical[:3] if canonical.startswith("csi") else canonical[:2]
if namespace in {"sh", "sz"}:
if display != canonical:
raise ValueError(f"SH/SZ index display must equal canonical: {canonical!r} != {display!r}")
elif namespace == "csi":
expected_display = f"{canonical[3:]}.CSI"
if display != expected_display:
raise ValueError(f"CSI index display must be {expected_display!r}, got {display!r}")
if market != "CN":
raise ValueError(f"index market must be CN: {canonical!r}")
if asset_type != "index":
raise ValueError(f"index asset_type must be index: {canonical!r}")
if active is not True:
raise ValueError(f"index active must be True: {canonical!r}")
if not name:
raise ValueError(f"index name must be non-empty: {canonical!r}")
pinyin_full = entry.get("pinyinFull")
pinyin_abbr = entry.get("pinyinAbbr")
if (
not isinstance(pinyin_full, str)
or not pinyin_full.strip()
or not isinstance(pinyin_abbr, str)
or not pinyin_abbr.strip()
):
raise ValueError(f"index pinyin fields must be non-empty: {canonical!r}")
if not isinstance(aliases, list):
raise ValueError(f"index aliases must be a list: {canonical!r}")
_validate_unique_index_aliases(aliases, canonical)
# Popularity must be a plain non-negative integer. Fractional
# (``1.5``), boolean (``True``) and negative values are rejected
# without truncation — only an integer value like ``100`` is valid.
if (
isinstance(popularity, bool)
or not isinstance(popularity, int)
or not math.isfinite(float(popularity))
or popularity < 0
):
raise ValueError(f"index popularity must be a non-negative integer: {canonical!r}")
if namespace not in {"sh", "sz", "csi"}:
raise ValueError(f"index namespace has no provider mapping: {namespace!r}")
# canonical uniqueness
norm_canonical = _normalize_index_key(canonical)
if norm_canonical in canonical_map:
raise ValueError(f"duplicate index canonical: {canonical!r}")
canonical_map[norm_canonical] = canonical
for alias in aliases:
norm_alias = _normalize_index_key(alias)
if norm_alias.isdigit():
raise ValueError(f"bare numeric display/alias rejected for index: {alias!r}")
if not _EXPLICIT_INDEX_ALIAS_RE.fullmatch(norm_alias):
raise ValueError(
f"index aliases must use an explicit code form: {alias!r}"
)
# canonical + display + aliases must resolve to exactly one canonical
for key in [canonical, display] + aliases:
norm_key = _normalize_index_key(key)
if not norm_key:
continue
if norm_key.isdigit():
raise ValueError(f"bare numeric display/alias rejected for index: {key!r}")
if norm_key in resolver_map and resolver_map[norm_key] != canonical:
raise ValueError(
f"index resolver key {key!r} maps to multiple canonicals "
f"({resolver_map[norm_key]} vs {canonical})"
)
if norm_key in stock_keys:
raise ValueError(
f"index resolver key {key!r} collides with active stock/ETF "
f"identity {stock_keys[norm_key]!r}"
)
resolver_map[norm_key] = canonical
# bare-conflict map: numeric base of explicit aliases, for matched_index
for alias in aliases:
base = "".join(ch for ch in alias if ch.isdigit())
if base and base.isdigit() and len(base) == 6:
bare_conflicts.setdefault(base, canonical)
return
def _load_existing_payload(output_path: Path) -> List[List[Any]]:
"""Load the existing compressed JSON payload (must be a list)."""
with open(output_path, "r", encoding="utf-8") as f:
payload = json.load(f)
if not isinstance(payload, list):
raise ValueError(f"existing payload is not a list: {output_path}")
return payload
def _atomic_write_json(output_path: Path, compressed: List[List[Any]]) -> None:
"""Write the compressed payload atomically (temp file + os.replace)."""
output_path.parent.mkdir(parents=True, exist_ok=True)
temp_path = output_path.with_name(f".{output_path.name}.{os.getpid()}.tmp")
try:
with open(temp_path, "w", encoding="utf-8") as f:
f.write("[\n")
for i, item in enumerate(compressed):
json.dump(item, f, ensure_ascii=False, separators=(",", ":"))
if i < len(compressed) - 1:
f.write(",\n")
else:
f.write("\n")
f.write("]\n")
os.replace(temp_path, output_path)
finally:
try:
temp_path.unlink()
except FileNotFoundError:
pass
def run_index_only(output_path: Path, *, test: bool = False) -> List[List[Any]]:
"""Merge the approved index-registry seed into the existing compressed JSON.
Preserves all non-index tuples in their original order, removes any old
index tuples, appends seed-generated index rows sorted by canonicalCode,
validates, and atomically replaces the output (unless ``test``).
"""
seed_rows = load_index_registry_seed()
index_entries = build_index_entries_from_seed(seed_rows)
existing = _load_existing_payload(output_path)
validate_stock_index_payload(existing, min_items=0)
non_index = [item for item in existing if not (len(item) > 7 and item[7] == "index")]
# Validate the seed index rows against the existing active stock/ETF rows so
# an index identity that collides with a stock/ETF key is rejected.
validate_index_registry(index_entries, non_index_rows=non_index)
index_compressed = compress_index(index_entries)
index_compressed.sort(key=lambda item: str(item[0]))
merged = non_index + index_compressed
if not test:
_atomic_write_json(output_path, merged)
return merged
def main():
"""主函数"""
parser = argparse.ArgumentParser(description='从 CSV 生成股票自动补全索引')
@@ -625,6 +979,11 @@ def main():
default='tushare',
help='数据源选择(默认: tushare)'
)
parser.add_argument(
'--index-only',
action='store_true',
help='仅合并指数注册表 seed 到现有压缩 JSON,不重建股票索引'
)
parser.add_argument(
'--test', '-t',
action='store_true',
@@ -635,11 +994,39 @@ def main():
print("=" * 60)
print("股票索引生成工具(从 CSV)")
print("=" * 60)
print(f"数据源:{args.source}")
if not require_pypinyin():
return 1
# 输出路径
output_path = (
Path(__file__).parent.parent / "apps" / "dsa-web" / "public" / "stocks.index.json"
)
output_path.parent.mkdir(parents=True, exist_ok=True)
# --index-only: 只合并指数 seed,不重建股票索引。
if args.index_only:
print(f"数据源:index-only(合并指数注册表 seed)")
print("\n[1/3] 读取指数注册表 seed...")
merged = run_index_only(output_path, test=args.test)
print(f" 合并后共 {len(merged)} 条记录")
index_rows = [item for item in merged if len(item) > 7 and item[7] == "index"]
print(f" 其中指数 {len(index_rows)} 条")
if args.test:
print("\n[2/3] 测试模式:跳过写入文件")
else:
print(f"\n[2/3] 写入文件:{output_path}")
file_size = output_path.stat().st_size
print(f" 文件大小:{file_size / 1024:.2f} KB")
print("\n[3/3] 验证合并结果...")
# In test mode the output file is untouched, so validate/report the
# would-be merged payload returned by ``run_index_only`` rather than
# reopening the unchanged file.
print(f" 验证通过:{len(merged)} 条记录")
return 0
print(f"数据源:{args.source}")
# 加载数据
print("\n[1/5] 读取 CSV 数据...")
if args.source == 'tushare':
@@ -661,11 +1048,20 @@ def main():
print("\n[2/5] 生成索引数据...")
index = build_stock_index(stocks)
# 输出路径
output_path = (
Path(__file__).parent.parent / "apps" / "dsa-web" / "public" / "stocks.index.json"
# 合并指数注册表 seed,防止后续重建擦除 index 行。
print("\n[2.5/5] 合并指数注册表 seed...")
seed_rows = load_index_registry_seed()
index_entries = build_index_entries_from_seed(seed_rows)
# Validate the seed index rows against the freshly built stock/ETF rows so
# an index identity that collides with a stock/ETF key is rejected.
validate_index_registry(
index_entries,
non_index_rows=compress_index(index),
)
output_path.parent.mkdir(parents=True, exist_ok=True)
# Canonical-sort the index rows so the full rebuild is byte-stable and
# matches the ``--index-only`` ordering; non-index rows keep build order.
index_entries.sort(key=lambda entry: str(entry["canonicalCode"]))
index.extend(index_entries)
print("\n[3/5] 压缩索引数据...")
compressed = compress_index(index)
@@ -686,15 +1082,7 @@ def main():
print(f" {i + 1}. {item}")
else:
print(f"\n[4/5] 写入文件:{output_path}")
with open(output_path, 'w', encoding='utf-8') as f:
f.write('[\n')
for i, item in enumerate(compressed):
json.dump(item, f, ensure_ascii=False, separators=(',', ':'))
if i < len(compressed) - 1:
f.write(',\n')
else:
f.write('\n')
f.write(']\n')
_atomic_write_json(output_path, compressed)
file_size = output_path.stat().st_size
print(f" 文件大小:{file_size / 1024:.2f} KB")
@@ -0,0 +1,32 @@
canonical_code,display_code,name_zh,aliases,name_source,popularity
sh000001,sh000001,上证指数,000001.SH,腾讯,100
sh000009,sh000009,上证380,000009.SH,腾讯,100
sh000010,sh000010,上证180,000010.SH,腾讯,100
sh000016,sh000016,上证50,000016.SH,腾讯,100
sh000300,sh000300,沪深300,sz399300|000300.SH|000300.CSI,腾讯,100
sh000688,sh000688,科创50,000688.SH,腾讯,100
sh000852,sh000852,中证1000,000852.SH,腾讯,100
sh000905,sh000905,中证500,sz399905|000905.SH|000905.CSI,腾讯,100
sh000906,sh000906,中证800,000906.SH,腾讯,100
sz399001,sz399001,深证成指,399001.SZ,腾讯,100
sz399005,sz399005,中小100,399005.SZ,腾讯,100
sz399006,sz399006,创业板指,399006.SZ,腾讯,100
sz399303,sz399303,国证2000,399303.SZ,腾讯,100
sz399330,sz399330,深证100,399330.SZ,腾讯,100
sh000015,sh000015,红利指数,000015.SH,腾讯,100
sh000922,sh000922,中证红利,sz399922|000922.SH|000922.CSI,腾讯,100
sz399324,sz399324,深证红利,399324.SZ,腾讯,100
sh000941,sh000941,新能源,000941.SH,腾讯,100
sz399296,sz399296,创成长,399296.SZ,腾讯,100
sz399967,sz399967,中证军工,399967.SZ,腾讯,100
sz399975,sz399975,证券公司,399975.SZ,腾讯,100
sz399986,sz399986,中证银行,399986.SZ,腾讯,100
sz399989,sz399989,中证医疗,399989.SZ,腾讯,100
sz399997,sz399997,中证白酒,399997.SZ,腾讯,100
sz980092,sz980092,自由现金流,980092.SZ,腾讯,100
csi932365,932365.CSI,中证现金流,,东财 2.932365,100
csi930955,930955.CSI,红利低波100,,东财 2.930955,100
csi931052,931052.CSI,国信价值,,东财 2.931052,100
csi931446,931446.CSI,东证红利低波,,东财 2.931446,100
csi931643,931643.CSI,科创创业50,,东财 2.931643,100
csi932366,932366.CSI,300现金流,,东财 2.932366,100
1 canonical_code display_code name_zh aliases name_source popularity
2 sh000001 sh000001 上证指数 000001.SH 腾讯 100
3 sh000009 sh000009 上证380 000009.SH 腾讯 100
4 sh000010 sh000010 上证180 000010.SH 腾讯 100
5 sh000016 sh000016 上证50 000016.SH 腾讯 100
6 sh000300 sh000300 沪深300 sz399300|000300.SH|000300.CSI 腾讯 100
7 sh000688 sh000688 科创50 000688.SH 腾讯 100
8 sh000852 sh000852 中证1000 000852.SH 腾讯 100
9 sh000905 sh000905 中证500 sz399905|000905.SH|000905.CSI 腾讯 100
10 sh000906 sh000906 中证800 000906.SH 腾讯 100
11 sz399001 sz399001 深证成指 399001.SZ 腾讯 100
12 sz399005 sz399005 中小100 399005.SZ 腾讯 100
13 sz399006 sz399006 创业板指 399006.SZ 腾讯 100
14 sz399303 sz399303 国证2000 399303.SZ 腾讯 100
15 sz399330 sz399330 深证100 399330.SZ 腾讯 100
16 sh000015 sh000015 红利指数 000015.SH 腾讯 100
17 sh000922 sh000922 中证红利 sz399922|000922.SH|000922.CSI 腾讯 100
18 sz399324 sz399324 深证红利 399324.SZ 腾讯 100
19 sh000941 sh000941 新能源 000941.SH 腾讯 100
20 sz399296 sz399296 创成长 399296.SZ 腾讯 100
21 sz399967 sz399967 中证军工 399967.SZ 腾讯 100
22 sz399975 sz399975 证券公司 399975.SZ 腾讯 100
23 sz399986 sz399986 中证银行 399986.SZ 腾讯 100
24 sz399989 sz399989 中证医疗 399989.SZ 腾讯 100
25 sz399997 sz399997 中证白酒 399997.SZ 腾讯 100
26 sz980092 sz980092 自由现金流 980092.SZ 腾讯 100
27 csi932365 932365.CSI 中证现金流 东财 2.932365 100
28 csi930955 930955.CSI 红利低波100 东财 2.930955 100
29 csi931052 931052.CSI 国信价值 东财 2.931052 100
30 csi931446 931446.CSI 东证红利低波 东财 2.931446 100
31 csi931643 931643.CSI 科创创业50 东财 2.931643 100
32 csi932366 932366.CSI 300现金流 东财 2.932366 100
+280
View File
@@ -3,6 +3,9 @@ from __future__ import annotations
import json
import logging
import math
import re
import unicodedata
from pathlib import Path
from threading import RLock
from typing import Dict, Iterable, Optional
@@ -18,9 +21,34 @@ from src.services.stock_index_remote_service import (
logger = logging.getLogger(__name__)
_STOCK_INDEX_FILENAME = "stocks.index.json"
_EXPLICIT_INDEX_ALIAS_RE = re.compile(
r"^(?:(?:sh|sz|csi)\d{6}|\d{6}\.(?:sh|sz|csi))$"
)
def _normalize_index_identity_key(value: object) -> str:
"""Normalize an identity key to one resolver identity form.
``sh``/``sz`` prefix and ``{code}.SH`` / ``{code}.SZ`` suffix collapse to
the canonical lowercase-prefixed key. CSI is deliberately **not**
collapsed: ``csi{code}`` (prefix) and ``{code}.CSI`` (suffix) are kept
distinct so an unregistered ``csi`` prefix never conflates with a
registered ``{code}.CSI`` alias — mirroring ``stock_list_parser``.
"""
normalized = unicodedata.normalize(
"NFKC", str(value or "")
).strip().casefold()
prefix_match = re.fullmatch(r"(sh|sz)(\d{6})", normalized)
if prefix_match:
return f"{prefix_match.group(1)}{prefix_match.group(2)}"
suffix_match = re.fullmatch(r"(\d{6})\.(sh|sz)", normalized)
if suffix_match:
return f"{suffix_match.group(2)}{suffix_match.group(1)}"
return normalized
_STOCK_INDEX_CACHE: Dict[str, str] | None = None
_STOCK_CODE_LOOKUP_CACHE: Dict[str, str] | None = None
_STOCK_CODE_CANDIDATES_CACHE: Dict[str, tuple[str, ...]] | None = None
_ACTIVE_INDEX_ROWS_CACHE: list | None = None
_REMOTE_INDEX_VALIDITY_CACHE: tuple[Path, float, int, bool] | None = None
_STOCK_INDEX_CACHE_LOCK = RLock()
@@ -373,14 +401,266 @@ def get_stock_code_index_map() -> Dict[str, str]:
return _STOCK_CODE_LOOKUP_CACHE
def _load_active_index_rows() -> list:
"""Load and cache the active ``assetType=index`` rows from the best candidate.
Returns raw compressed tuples (no parser type dependency) so the loader
stays free of a circular import with ``stock_list_parser``. Each candidate
must pass base + semantic validation before its index rows are used.
Non-regression guard: when the bundled candidate is a valid baseline, every
OTHER candidate — the remote cache AND the legacy ``static`` fallback — must
be a legal superset of the bundled active-index canonical set to be selected.
A candidate that drops any bundled active index canonical is skipped
(WARNING) in favour of the bundled candidate, so a stale or partial
non-bundled file can never bypass the bundled baseline when the remote cache
is missing/invalid.
"""
global _ACTIVE_INDEX_ROWS_CACHE
if _ACTIVE_INDEX_ROWS_CACHE is not None:
return _ACTIVE_INDEX_ROWS_CACHE
with _STOCK_INDEX_CACHE_LOCK:
if _ACTIVE_INDEX_ROWS_CACHE is not None:
return _ACTIVE_INDEX_ROWS_CACHE
remote_path = get_remote_stock_index_cache_path()
candidate_paths = get_stock_index_candidate_paths()
candidates = _get_fresh_stock_index_candidates(candidate_paths, remote_path)
# First pass: find the bundled candidate's active index canonical set
# as the non-regression baseline. The bundled candidate is the
# ``apps/dsa-web/public/stocks.index.json`` path (identified from the
# declared candidate order, not the first non-remote candidate ordered
# by mtime), and its index rows must pass semantic validation before
# they become the baseline.
bundled_path = _get_bundled_stock_index_path(candidate_paths, remote_path)
bundled_rows: list | None = None
bundled_canonicals: set[str] | None = None
if bundled_path is not None:
try:
raw_items = _load_stock_index_payload(bundled_path)
validate_stock_index_payload(raw_items, min_items=0)
rows = _extract_active_index_rows(raw_items)
if rows:
_validate_index_rows_semantics(
rows, _extract_active_non_index_rows(raw_items)
)
bundled_rows = rows
bundled_canonicals = {str(r[0]) for r in rows}
except (OSError, TypeError, ValueError) as exc:
logger.debug("[股票索引] 解析 bundled 指数行失败 %s: %s", bundled_path, exc)
# Second pass: pick the best candidate (remote preferred when it is a
# legal superset of the bundled baseline).
for index_path in candidates:
try:
raw_items = _load_stock_index_payload(index_path)
if _same_path(index_path, remote_path):
validate_stock_index_payload(raw_items)
else:
validate_stock_index_payload(raw_items, min_items=0)
rows = _extract_active_index_rows(raw_items)
if not rows:
continue
_validate_index_rows_semantics(
rows, _extract_active_non_index_rows(raw_items)
)
# Non-regression guard: when the bundled candidate is a valid
# baseline, every OTHER candidate (remote cache or legacy
# ``static`` fallback) must be a legal superset of the bundled
# active-index canonical set. A candidate that drops any
# bundled baseline canonical is skipped (WARNING) so a stale
# or partial legacy-static file can never bypass the bundled
# baseline when the remote cache is missing/invalid.
if bundled_canonicals and bundled_path is not None and not _same_path(index_path, bundled_path):
candidate_canonicals = {str(r[0]) for r in rows}
missing = bundled_canonicals - candidate_canonicals
if missing:
logger.warning(
"[股票索引] 指数候选 %s 缺少 bundled baseline canonical: %s — 回退 bundled",
index_path,
sorted(missing),
)
continue
_ACTIVE_INDEX_ROWS_CACHE = rows
return rows
except (OSError, TypeError, ValueError) as exc:
logger.debug("[股票索引] 解析指数行失败 %s: %s", index_path, exc)
# All candidates failed or were skipped — empty registry + WARNING.
logger.warning("[股票索引] 所有指数候选均失败,指数注册表为空")
_ACTIVE_INDEX_ROWS_CACHE = []
return _ACTIVE_INDEX_ROWS_CACHE
def _get_bundled_stock_index_path(
candidate_paths: Iterable[Path],
remote_path: Path,
) -> Path | None:
"""Return the bundled ``apps/dsa-web/public/stocks.index.json`` path.
The bundled candidate is the deterministic non-remote baseline for the
remote non-regression check, independent of mtime ordering. It is the
declared ``apps/dsa-web/public`` candidate (the non-remote, non-legacy
``static`` candidate in the declared candidate order), not the remote cache
and not the legacy ``static`` fallback.
"""
for candidate in candidate_paths:
if _same_path(candidate, remote_path):
continue
if candidate.as_posix().endswith("static/stocks.index.json"):
continue
if candidate.is_file():
return candidate
return None
def _extract_active_index_rows(raw_items: list) -> list:
"""Return the active ``assetType=index`` rows from a raw payload."""
rows = []
for item in raw_items:
if not isinstance(item, list) or len(item) < 10:
continue
if item[7] != "index":
continue
if item[8] is not True:
continue
rows.append(item)
return rows
def _extract_active_non_index_rows(raw_items: list) -> list:
"""Return the active stock/ETF rows from a raw payload (for collision checks)."""
rows = []
for item in raw_items:
if not isinstance(item, list) or len(item) < 10:
continue
if str(item[7] or "").strip() == "index":
continue
if item[8] is not True:
continue
rows.append(item)
return rows
def _validate_index_rows_semantics(rows: list, non_index_rows: list | None = None) -> None:
"""Semantic validation for active index rows (implementation-contracts §Semantic Validation).
Raises ``ValueError`` on any identity/alias/stock-index conflict so a
malformed candidate is rejected rather than silently loaded. ``non_index_rows``
carries the active stock/ETF rows from the same payload so an index
canonical/display/alias that collides with a stock/ETF identity after
normalization is rejected.
"""
canonical_map: dict[str, str] = {}
resolver_map: dict[str, str] = {}
# Active stock/ETF identity keys an index explicit key must never collide with.
stock_keys: dict[str, str] = {}
for row in non_index_rows or []:
stock_canonical = str(row[0] or "").strip()
for key in [row[0], row[1]] + list(row[5] if isinstance(row[5], list) else []):
norm_key = _normalize_index_identity_key(key)
if norm_key:
stock_keys.setdefault(norm_key, stock_canonical)
for row in rows:
canonical = str(row[0] or "").strip()
display = str(row[1] or "").strip()
name = str(row[2] or "").strip()
pinyin_full = row[3]
pinyin_abbr = row[4]
market = str(row[6] or "").strip()
asset_type = str(row[7] or "").strip()
active = row[8]
popularity = row[9]
aliases = row[5]
if not re.fullmatch(r"(sh|sz|csi)\d{6}", canonical):
raise ValueError(f"index canonical must match ^(sh|sz|csi)\\d{{6}}$: {canonical!r}")
namespace = canonical[:3] if canonical.startswith("csi") else canonical[:2]
if namespace in {"sh", "sz"}:
if display != canonical:
raise ValueError(f"SH/SZ index display must equal canonical: {canonical!r}")
elif namespace == "csi":
if display != f"{canonical[3:]}.CSI":
raise ValueError(f"CSI index display must be {canonical[3:]}.CSI: {display!r}")
if market != "CN":
raise ValueError(f"index market must be CN: {canonical!r}")
if asset_type != "index":
raise ValueError(f"index asset_type must be index: {canonical!r}")
if active is not True:
raise ValueError(f"index active must be True: {canonical!r}")
if not name:
raise ValueError(f"index name must be non-empty: {canonical!r}")
if (
not isinstance(pinyin_full, str)
or not pinyin_full.strip()
or not isinstance(pinyin_abbr, str)
or not pinyin_abbr.strip()
):
raise ValueError(f"index pinyin fields must be non-empty: {canonical!r}")
if not isinstance(aliases, list):
raise ValueError(f"index aliases must be a list: {canonical!r}")
if (
isinstance(popularity, bool)
or not isinstance(popularity, int)
or not math.isfinite(float(popularity))
or popularity < 0
):
raise ValueError(f"index popularity must be a non-negative integer: {canonical!r}")
norm_canonical = _normalize_index_identity_key(canonical)
if norm_canonical in canonical_map:
raise ValueError(f"duplicate index canonical: {canonical!r}")
canonical_map[norm_canonical] = canonical
seen_aliases: set[str] = set()
for alias in aliases:
norm_alias = _normalize_index_identity_key(alias)
if norm_alias in seen_aliases:
raise ValueError(
f"duplicate index alias after normalization for {canonical!r}: {alias!r}"
)
seen_aliases.add(norm_alias)
if norm_alias.isdigit():
raise ValueError(f"bare numeric display/alias rejected for index: {alias!r}")
if not _EXPLICIT_INDEX_ALIAS_RE.fullmatch(norm_alias):
raise ValueError(
f"index aliases must use an explicit code form: {alias!r}"
)
for key in [canonical, display] + aliases:
norm_key = _normalize_index_identity_key(key)
if not norm_key:
continue
if norm_key.isdigit():
raise ValueError(f"bare numeric display/alias rejected for index: {key!r}")
if norm_key in resolver_map and resolver_map[norm_key] != canonical:
raise ValueError(
f"index resolver key {key!r} maps to multiple canonicals "
f"({resolver_map[norm_key]} vs {canonical})"
)
if norm_key in stock_keys:
raise ValueError(
f"index resolver key {key!r} collides with active stock/ETF "
f"identity {stock_keys[norm_key]!r}"
)
resolver_map[norm_key] = canonical
def clear_stock_index_cache() -> None:
"""Clear the in-process stock index lookup cache."""
global _REMOTE_INDEX_VALIDITY_CACHE
global _STOCK_CODE_CANDIDATES_CACHE, _STOCK_CODE_LOOKUP_CACHE, _STOCK_INDEX_CACHE
global _ACTIVE_INDEX_ROWS_CACHE
with _STOCK_INDEX_CACHE_LOCK:
_STOCK_INDEX_CACHE = None
_STOCK_CODE_LOOKUP_CACHE = None
_STOCK_CODE_CANDIDATES_CACHE = None
_ACTIVE_INDEX_ROWS_CACHE = None
_REMOTE_INDEX_VALIDITY_CACHE = None
+33
View File
@@ -122,6 +122,39 @@ class HistoryService:
is_bse_code,
normalize_stock_code,
)
from src.services.stock_code_utils import (
_converge_registered_csi_identity,
)
# PR #2267 review remediation: converge registered CSI aliases
# (``csi930955`` / ``930955.CSI`` / ``CSI930955``) so a record persisted under any
# equivalent form is reachable from every equivalent query input.
# This is a *persisted-read* filter path, so the candidate set must
# include:
# 1. the parser canonical (``csi930955`` — current storage form),
# 2. the old resolver's uppercase canonical (``CSI930955`` — how
# pre-fix records were saved), and
# 3. the IndexEntry's explicit aliases (``930955.CSI``).
converged_csi = _converge_registered_csi_identity(raw_code)
if converged_csi is not None:
from src.data.stock_index_loader import _load_active_index_rows
active_rows = _load_active_index_rows()
alias_keys: List[str] = []
display_keys: List[str] = []
for row in active_rows:
if row and str(row[0] or "").strip() == converged_csi:
display_keys = [str(row[1] or "").strip()]
alias_keys = [
str(a) for a in (row[5] if isinstance(row[5], list) else [])
if str(a).strip()
]
break
canonical_upper = canonical_stock_code(converged_csi) or converged_csi.upper()
add_keys = [converged_csi, canonical_upper] + display_keys + alias_keys
for key in add_keys:
if key and key not in candidates:
candidates.append(key)
return candidates
raw_canonical = canonical_stock_code(raw_code)
normalized = canonical_stock_code(normalize_stock_code(raw_canonical))
+47
View File
@@ -6,6 +6,7 @@ Shared stock code utilities.
from __future__ import annotations
import re
import unicodedata
from importlib import import_module
from dataclasses import dataclass
from typing import List, Optional
@@ -488,6 +489,20 @@ def resolve_index_stock_code_for_analysis(raw: str) -> str:
if not text:
return ""
# PR #2267 review remediation: converge registered CSI aliases
# (``csi930955`` / ``930955.CSI`` / ``CSI930955``) to the parser canonical so
# the resolver, task dedupe key and history candidates do not split the
# same index into distinct keys. The parser canonical is returned verbatim
# (lowercase ``csi{code}``) rather than re-uppercased by
# ``canonical_stock_code`` below.
normalized_csi = unicodedata.normalize("NFKC", text).strip().casefold()
if re.fullmatch(r"csi\d{6}", normalized_csi) or re.fullmatch(r"\d{6}\.csi", normalized_csi):
converged = _converge_registered_csi_identity(text)
# Return the parser canonical verbatim (lowercase ``csi{code}``) for
# every registered CSI form, including the already-canonical input.
if converged is not None:
return converged
if is_code_like(text) or (text.isdigit() and len(text) == 4):
from src.data.stock_index_loader import resolve_index_stock_code
@@ -496,3 +511,35 @@ def resolve_index_stock_code_for_analysis(raw: str) -> str:
return canonical_stock_code(resolved)
return canonical_stock_code(text)
def _converge_registered_csi_identity(raw: str) -> Optional[str]:
"""Converge a registered CSI explicit identity to its parser canonical.
PR #2267 review remediation: the resolver, task dedupe key and history
candidate builders must treat ``csi930955`` / ``930955.CSI`` /
``CSI930955`` as the same registered CSI
index identity, so they do not split into distinct task keys or history
candidates. Returns the parser canonical (lowercase ``csi{code}``) when the
input is a registered CSI explicit identity, or ``None`` when it is not
(so callers keep their existing degradation semantics for unregistered
``csi930956`` / ``930956.CSI``).
"""
text = (raw or "").strip()
if not text:
return None
normalized = unicodedata.normalize("NFKC", text).strip().casefold()
is_csi_form = bool(
re.fullmatch(r"csi\d{6}", normalized)
or re.fullmatch(r"\d{6}\.csi", normalized)
)
if not is_csi_form:
return None
from src.services.stock_list_parser import parse_analysis_target
target = parse_analysis_target(text)
if target.asset_type == "index" and target.canonical_id:
return target.canonical_id
return None
+225 -63
View File
@@ -32,6 +32,7 @@ from __future__ import annotations
import json
import re
import unicodedata
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
@@ -49,6 +50,41 @@ __all__ = [
_STOCK_LIST_SEPARATOR_RE = re.compile(r"[\s,;\uFF0C\u3001\uFF1B]+")
_EXPLICIT_INDEX_ALIAS_RE = re.compile(
r"^(?:(?:sh|sz|csi)\d{6}|\d{6}\.(?:sh|sz|csi))$"
)
# An unregistered numeric CSI form must surface as ``unsupported`` rather than
# degrade into a US ticker. Registered six-digit forms resolve before this
# guard; malformed lengths are rejected here without blocking tickers such as
# ``CSIQ``.
_EXPLICIT_CSI_FORM_RE = re.compile(r"^(?:csi\d+|\d+\.csi)$")
def _normalize_index_key(value: str) -> str:
"""Normalize an identity key to one resolver identity form.
``sh``/``sz`` prefix and ``{code}.SH`` / ``{code}.SZ`` suffix are
interchangeable exchange identities, so they collapse to the canonical
lowercase-prefixed key (``sh000300`` == ``000300.SH``).
CSI is deliberately **not** collapsed: ``csi{code}`` (prefix) and
``{code}.CSI`` (suffix) are kept distinct. The ``csi`` prefix is a canonical
identity that only resolves when a manifest row owns that exact ``csi{code}``
key, while a ``{code}.CSI`` suffix is an *explicit alias* that belongs to the
entry that registered it (e.g. ``000300.CSI`` is an alias of ``sh000300``).
Keeping them separate prevents ``csi000300`` (unregistered) from being
conflated with the registered ``000300.CSI`` alias of ``sh000300``.
"""
normalized = unicodedata.normalize(
"NFKC", str(value or "")
).strip().casefold()
prefix_match = re.fullmatch(r"(sh|sz)(\d{6})", normalized)
if prefix_match:
return f"{prefix_match.group(1)}{prefix_match.group(2)}"
suffix_match = re.fullmatch(r"(\d{6})\.(sh|sz)", normalized)
if suffix_match:
return f"{suffix_match.group(2)}{suffix_match.group(1)}"
return normalized
# ---------------------------------------------------------------------------
@@ -208,7 +244,7 @@ class IndexEntry:
"""A single known index in :class:`IndexRegistry`."""
bare_code: str
exchange: str # 'SH' or 'SZ'
exchange: str # 'SH' / 'SZ' / 'CSI'
canonical_id: str
display_name: str
aliases: Tuple[str, ...] = ()
@@ -240,20 +276,64 @@ class AnalysisTarget:
class IndexRegistry:
"""Authoritative source for ``asset_type=index`` resolution.
The registry is populated with a small built-in white-list of canonical
A-share indices (CSI 300 / SSE 50 / STAR 50 / SZSE Component / ChiNext)
that already exist as hard-coded white-lists across
``data_provider/*_fetcher.py``. PR1 deliberately keeps the registry
in-memory and immutable — later phases of issue #2063 will load the
``asset_type=index`` rows from ``apps/dsa-web/public/stocks.index.json``
once that file carries index rows; the parser contract won't change.
The registry is populated from the validated ``assetType=index`` rows of
``apps/dsa-web/public/stocks.index.json`` (via :func:`default_index_registry`).
It builds three lookup maps at construction time:
* canonical map — ``canonical_id -> entry``
* resolver-key map — every normalized canonical/display/alias key -> entry
* bare-conflict map — the 6-digit numeric base of explicit aliases -> entry
Construction rejects duplicate or ambiguous keys (identity/alias/stock-index
conflicts) by raising ``ValueError`` so a malformed candidate is never
silently loaded.
"""
def __init__(self, entries: Iterable[IndexEntry] = ()):
self._entries: List[IndexEntry] = list(entries)
self._by_canonical: Dict[str, IndexEntry] = {
e.canonical_id: e for e in self._entries
}
self._by_canonical: Dict[str, IndexEntry] = {}
self._by_resolver_key: Dict[str, IndexEntry] = {}
self._by_bare_conflict: Dict[str, IndexEntry] = {}
for entry in self._entries:
norm_canonical = _normalize_index_key(entry.canonical_id)
if norm_canonical in self._by_canonical:
raise ValueError(
f"duplicate index canonical: {entry.canonical_id!r}"
)
self._by_canonical[norm_canonical] = entry
for alias in entry.aliases:
norm_alias = _normalize_index_key(alias)
if norm_alias.isdigit():
raise ValueError(
f"bare numeric display/alias rejected for index: {alias!r}"
)
if not _EXPLICIT_INDEX_ALIAS_RE.fullmatch(norm_alias):
raise ValueError(
f"index aliases must use an explicit code form: {alias!r}"
)
for key in [entry.canonical_id] + list(entry.aliases):
norm_key = _normalize_index_key(key)
if not norm_key:
continue
if norm_key.isdigit():
raise ValueError(
f"bare numeric display/alias rejected for index: {key!r}"
)
existing = self._by_resolver_key.get(norm_key)
if existing is not None and existing is not entry:
raise ValueError(
f"index resolver key {key!r} maps to multiple canonicals "
f"({existing.canonical_id} vs {entry.canonical_id})"
)
self._by_resolver_key[norm_key] = entry
for alias in entry.aliases:
base = "".join(ch for ch in alias if ch.isdigit())
if base and base.isdigit() and len(base) == 6:
self._by_bare_conflict.setdefault(base, entry)
def __len__(self) -> int:
return len(self._entries)
@@ -275,11 +355,16 @@ class IndexRegistry:
if exchange is None:
return None
canonical = f"{prefix}{bare_code}"
entry = self._by_canonical.get(canonical)
entry = self._by_canonical.get(_normalize_index_key(canonical))
if entry is not None:
return entry
# Alias fallback — e.g. user typed ``sz399300`` but registry lists it
# under ``sh000300`` via an alias. Walk the registry once.
# under ``sh000300`` via an alias. Check the resolver map for the
# prefixed form first (``sz399300`` is an explicit alias), then walk
# the registry once for bare-code matches.
entry = self._by_resolver_key.get(_normalize_index_key(canonical))
if entry is not None:
return entry
for entry in self._entries:
if entry.exchange == exchange and entry.matches_code(bare_code):
return entry
@@ -298,64 +383,86 @@ class IndexRegistry:
return entry
return None
def find_by_explicit_key(self, key: str) -> Optional[IndexEntry]:
"""Look up an index by an explicit canonical/display/alias key.
Used by :func:`parse_analysis_target` to resolve ``csi930955`` /
``930955.CSI`` and other explicit index forms before generic stock
normalization runs.
"""
return self._by_resolver_key.get(_normalize_index_key(key))
def find_by_bare_conflict(self, bare_code: str) -> Optional[IndexEntry]:
"""Return the index whose explicit alias base equals ``bare_code``."""
return self._by_bare_conflict.get(_normalize_index_key(bare_code))
# ---------------------------------------------------------------------------
# Default index registry — the built-in white-list.
# Default index registry — built from the validated JSON index rows.
# ---------------------------------------------------------------------------
# Five canonical A-share indices, mirroring the hard-coded lists already
# present in ``data_provider/{efinance,akshare,yfinance,tickflow}_fetcher.py``.
# Keeping this list in one place + giving it a public ``IndexRegistry`` type
# is the whole point of PR1; later phases may move the data into
# ``apps/dsa-web/public/stocks.index.json`` and load it, but the parser
# contract stays stable.
_DEFAULT_INDEX_ENTRIES: Tuple[IndexEntry, ...] = (
IndexEntry(
bare_code="000300",
exchange="SH",
canonical_id="sh000300",
display_name="沪深300",
aliases=("000300.SH", "CSI300", "HS300"),
),
IndexEntry(
bare_code="000016",
exchange="SH",
canonical_id="sh000016",
display_name="上证50",
aliases=("000016.SH", "SSE50"),
),
IndexEntry(
bare_code="000688",
exchange="SH",
canonical_id="sh000688",
display_name="科创50",
aliases=("000688.SH", "STAR50"),
),
IndexEntry(
bare_code="399001",
exchange="SZ",
canonical_id="sz399001",
display_name="深证成指",
aliases=("399001.SZ", "SZSE"),
),
IndexEntry(
bare_code="399006",
exchange="SZ",
canonical_id="sz399006",
display_name="创业板指",
aliases=("399006.SZ", "ChiNext"),
),
)
# The unique runtime identity source is the ``active=true`` / ``assetType=index``
# rows of the candidate ``stocks.index.json``. The old hard-coded
# ``_DEFAULT_INDEX_ENTRIES`` white-list is removed; no second runtime white-list
# is kept. When all candidates fail, the registry is empty and every prefixed
# input degrades to the stock path (fail-open).
def _index_entry_from_row(row) -> Optional[IndexEntry]:
"""Convert one validated active index tuple into an :class:`IndexEntry`."""
if not isinstance(row, list) or len(row) < 10:
return None
canonical = str(row[0] or "").strip()
display = str(row[1] or "").strip()
name = str(row[2] or "").strip()
aliases = row[5] if isinstance(row[5], list) else []
if not canonical or not name:
return None
namespace = canonical[:3] if canonical.startswith("csi") else canonical[:2]
if namespace == "csi":
exchange = "CSI"
bare_code = canonical[3:]
elif namespace in {"sh", "sz"}:
exchange = _EXCHANGE_PREFIX_TO_CODE.get(namespace, "UNKNOWN")
bare_code = canonical[2:]
else:
return None
aliases = [str(a) for a in (row[5] if isinstance(row[5], list) else []) if str(a).strip()]
# The display code (e.g. ``930955.CSI``) is a resolver key even when it is
# not an explicit alias; add it so ``find_by_explicit_key`` and the
# bare-conflict map can resolve it. SH/SZ display equals canonical so it is
# already covered.
display = str(row[1] or "").strip()
if display and display != canonical and display not in aliases:
aliases.append(display)
return IndexEntry(
bare_code=bare_code,
exchange=exchange,
canonical_id=canonical,
display_name=name,
aliases=tuple(aliases),
)
def default_index_registry() -> IndexRegistry:
"""Return the built-in :class:`IndexRegistry` shipped with PR1.
"""Return the :class:`IndexRegistry` built from the validated JSON index rows.
``IndexRegistry`` is intentionally cheap to construct (5 small dataclass
entries); callers may rebuild it on every call rather than caching it
globally. Once the JSON registry carries ``asset_type=index`` rows this
factory should load from disk and stay backward-compatible.
Loads the active ``assetType=index`` rows via the stock-index loader's
cached active-index-row loader, converts each to a lightweight
:class:`IndexEntry`, and constructs a fresh :class:`IndexRegistry`. The
registry is intentionally cheap to construct; callers may rebuild it on
every call rather than caching it globally. When the loader returns no
rows (all candidates failed), the registry is empty and every prefixed
input degrades to the stock path.
"""
return IndexRegistry(_DEFAULT_INDEX_ENTRIES)
from src.data.stock_index_loader import _load_active_index_rows
rows = _load_active_index_rows()
entries = []
for row in rows:
entry = _index_entry_from_row(row)
if entry is not None:
entries.append(entry)
return IndexRegistry(entries)
# ---------------------------------------------------------------------------
@@ -514,6 +621,56 @@ def parse_analysis_target(
if registry is None:
registry = default_index_registry()
# Resolve explicit index keys (canonical/display/alias) BEFORE generic
# stock normalization, so ``csi930955`` / ``930955.CSI`` and other explicit
# index forms resolve to ``index`` without touching
# ``normalize_stock_code()``. An unregistered ``.CSI`` input is surfaced as
# ``unsupported`` (never a US ticker or a guessed SH/SZ index).
explicit_entry = registry.find_by_explicit_key(raw)
if explicit_entry is not None:
canonical, display, exchange = _canonicalize_for_index(explicit_entry, raw)
# Preserve the prefix/bare split for explicit sh/sz prefixed forms so
# the existing ``normalized_prefix``/``normalized_code`` contract stays
# intact (review blocker OR-COR-d24a4e9a family). CSI forms carry no
# sh/sz prefix.
norm_prefix = None
norm_code = raw
lower = raw.lower()
for p in ("sh", "sz"):
if lower.startswith(p) and len(raw) > len(p):
norm_prefix = p
norm_code = raw[len(p):]
break
return AnalysisTarget(
raw_input=raw_input,
asset_type=ParseStatus.INDEX,
canonical_id=canonical,
display_code=display,
exchange=exchange,
normalized_prefix=norm_prefix,
normalized_code=norm_code,
matched_index=explicit_entry,
)
# PR #2267 review fix: an explicit CSI form not owned by the registry must
# surface as ``unsupported`` — never a US ticker or a guessed SH/SZ index. This
# covers both the ``.CSI`` suffix (already handled by the prior check) and
# the ``csi`` prefix form (``csi930956`` / ``CSI930956``), which the generic
# normalizer would otherwise mis-route into the US-ticker branch.
if _EXPLICIT_CSI_FORM_RE.match(unicodedata.normalize("NFKC", raw).strip().casefold()):
return AnalysisTarget(
raw_input=raw_input,
asset_type=ParseStatus.UNSUPPORTED,
canonical_id=raw,
display_code=raw,
exchange="UNKNOWN",
unsupported_reason=(
f"unregistered CSI index: {raw!r} is not in the index registry"
),
normalized_prefix=None,
normalized_code=raw,
)
# Reuse the repository's existing normalization contract
# (``src.services.stock_code_utils._normalize_code_and_exchange``) so
# lowercase tickers (``shop``/``hkd``/``aapl``) and suffix-form inputs
@@ -905,6 +1062,11 @@ def parse_analysis_target(
matched_index = None
if prefix is None and bare:
candidate = registry.find_by_bare_code(bare)
if candidate is None:
# Also surface conflicts via the bare-conflict map so a bare code
# that is the numeric base of an explicit index alias (e.g.
# ``930955`` for ``csi930955``) is advertised as ambiguous.
candidate = registry.find_by_bare_conflict(bare)
if candidate is not None:
matched_index = candidate
+168 -25
View File
@@ -1623,10 +1623,14 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
SQLite can't add NOT NULL/UNIQUE columns via ALTER).
5. Backfill existing rows with ``_derive_canonical_id(code)`` in
id-batched chunks (5000/batch), using ``WHERE canonical_id IS NULL``
so re-runs are safe. ``_derive_canonical_id`` is index-aware: a
bare registered index code (e.g. ``000300``) resolves to the index
canonical_id (``sh000300``) rather than the stock-path key
(``sz000300``), preventing same-index split across buckets.
so re-runs are safe. ``_derive_canonical_id`` mirrors the parser
contract: a bare code always derives the stock-path
canonical_id (e.g. ``000300`` -> ``sz000300``), while only explicit
index forms (``sh000300`` / ``930955.CSI``) derive an index
canonical_id. Historical bare rows whose current canonical_id was
mis-written as an index identity (e.g. ``sh000300``) are not fixed
here — the idempotent ``_backfill_canonical_ids()`` repair corrects
that bucket in a separate pass.
Function-level lazy import keeps the storage layer free of a
circular import on ``src.services.stock_list_parser``.
6. Create a plain (non-unique) index ``ix_stock_daily_canonical_id`` so
@@ -1667,6 +1671,7 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
raise
self._backfill_stock_daily_canonical_id()
self._backfill_canonical_ids()
self._ensure_stock_daily_canonical_id_index()
def _ensure_stock_daily_canonical_id_index(self) -> None:
@@ -1701,17 +1706,15 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
)
def _derive_canonical_id(self, code: str) -> Optional[str]:
"""Derive a Phase 1 ``canonical_id`` for ``code``, index-aware.
"""Derive a Phase 1 ``canonical_id`` for ``code``.
Closes review blocker ``OR-COR-4f9ffc38``: when a bare code hits the
index registry (``parse_analysis_target(code).matched_index is not
None``) we unify to the index's ``canonical_id`` so the same
underlying index doesn't split across buckets — e.g. bare ``000300``
resolves to ``sh000300`` (the CSI-300 index canonical_id) instead of
``sz000300`` (the stock-path canonical_id the classifier would
synthesise for a 6-digit ``0``-prefixed code). When no index is
matched, the parser's stock/explicit-index ``canonical_id`` is
returned unchanged.
This method returns ``parse_analysis_target(code).canonical_id`` (or
``None``) and never reads ``matched_index.canonical_id`` to override
the result. Bare codes always resolve to the stock-path canonical_id
(e.g. bare ``000300`` -> ``sz000300``), while explicit index forms
(``sh000300`` / ``930955.CSI``) resolve to their index canonical_id.
This keeps the storage derivation consistent with the parser contract
(bare = stock, explicit index = index).
Lazy-imports ``parse_analysis_target`` inside the method (the parser
module transitively touches the storage layer) and degrades to
@@ -1719,7 +1722,10 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
persist NULL (D1) without raising.
"""
try:
from src.services.stock_list_parser import parse_analysis_target
from src.services.stock_list_parser import (
ParseStatus,
parse_analysis_target,
)
except Exception as exc:
logger.warning(
"_derive_canonical_id: cannot import parse_analysis_target "
@@ -1728,8 +1734,11 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
)
return None
target = parse_analysis_target(code)
if target.matched_index is not None:
return target.matched_index.canonical_id or None
if target.asset_type == ParseStatus.UNSUPPORTED:
# An unsupported identity (e.g. an unregistered ``csi930956`` /
# ``930956.CSI``) must not enter a persistent canonical bucket.
# Return None so the caller persists NULL instead.
return None
return target.canonical_id or None
def _backfill_stock_daily_canonical_id(self) -> None:
@@ -1745,9 +1754,10 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
skipped past in this run and naturally retried on next startup while
its ``canonical_id`` is still NULL.
Derivation is index-aware via :meth:`_derive_canonical_id` so a bare
index code (e.g. ``000300``) backfills to the index canonical_id
(``sh000300``) rather than the stock-path canonical_id (``sz000300``).
Derivation follows the parser contract: a bare code that collides with
an index identity (e.g. ``000300``) backfills to the stock-path
canonical_id (``sz000300``); only explicit index forms
(``sh000300`` / ``930955.CSI``) derive to an index canonical_id.
"""
_BATCH_SIZE = 5000
@@ -1825,6 +1835,135 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
total_skipped,
)
def _backfill_canonical_ids(self) -> None:
"""Idempotent, batched repair of historical bare-code canonical_id buckets.
The previous ``_derive_canonical_id`` behavior preferred
``matched_index.canonical_id``, so a bare six-digit code that collided
with an index identity (e.g. ``000001`` / ``000016`` / ``000688`` /
``930955``) was written with the *index* canonical_id (``sh000001`` /
``csi930955``) instead of the stock-path canonical_id (``sz000001`` /
``bj930955``). This method repairs those rows.
Only rows matching ALL of the following are touched:
* ``code`` is a bare 6-digit numeric string (no explicit prefix/suffix)
* current ``canonical_id`` is non-NULL and hits an active index identity
* the parser's stock canonical_id differs from the current value
Explicit ``sh``/``sz``/``csi`` prefix and ``.SH``/``.SZ``/``.CSI``
suffix rows are never modified. The repair uses a monotonic ``id``
cursor and a conditional UPDATE (guarded by the current value) so it is
idempotent and safe under concurrent writers. When the registry is
empty the repair is a no-op and logs a WARNING.
"""
if not self._is_sqlite_engine:
return
if not inspect(self._engine).has_table(StockDaily.__tablename__):
return
try:
from src.services.stock_list_parser import parse_analysis_target
from src.data.stock_index_loader import _load_active_index_rows
except Exception as exc:
logger.warning(
"[StockDaily] canonical_id repair cannot import parser/loader: %s — skipping",
exc,
)
return
active_rows = _load_active_index_rows()
if not active_rows:
logger.warning("[StockDaily] canonical_id repair skipped: index registry is empty")
return
# Build the set of active index canonical_ids to detect mis-bucketed rows.
active_index_canonicals = {
str(row[0]).strip() for row in active_rows if row and str(row[0]).strip()
}
if not active_index_canonicals:
logger.warning("[StockDaily] canonical_id repair skipped: no active index canonicals")
return
_BATCH_SIZE = 5000
total_repaired = 0
total_skipped = 0
last_id = 0
# Only rows whose current canonical_id hits an active index identity are
# candidates for repair. Push this filter into the SQL so repeated
# startup repairs do not scan the full table.
canonical_placeholders = ", ".join(f":c{i}" for i in range(len(active_index_canonicals)))
canonical_params = {f"c{i}": c for i, c in enumerate(sorted(active_index_canonicals))}
while True:
with self._engine.begin() as connection:
rows = connection.execute(
text(
f"SELECT id, code, canonical_id FROM {StockDaily.__tablename__} "
"WHERE id > :last_id "
f"AND canonical_id IN ({canonical_placeholders}) "
"AND length(code) = 6 "
"AND code NOT GLOB '*[^0-9]*' "
f"ORDER BY id LIMIT {_BATCH_SIZE}"
),
{"last_id": last_id, **canonical_params},
).fetchall()
if not rows:
break
with self._engine.begin() as connection:
for row_id, code, current_canonical in rows:
last_id = row_id
code_str = str(code or "").strip()
current = str(current_canonical or "").strip()
# Only bare 6-digit numeric codes are candidates for repair.
if not (code_str.isdigit() and len(code_str) == 6):
total_skipped += 1
continue
try:
target = parse_analysis_target(code_str)
except Exception as exc:
logger.warning(
"[StockDaily] canonical_id repair derivation failed for "
"code=%r (id=%s): %s — skipping",
code_str, row_id, exc,
)
total_skipped += 1
continue
derived = target.canonical_id or None
if not derived or derived == current:
total_skipped += 1
continue
result = connection.execute(
text(
f"UPDATE {StockDaily.__tablename__} "
"SET canonical_id = :derived "
"WHERE id = :row_id AND canonical_id = :current"
),
{"derived": derived, "row_id": row_id, "current": current},
)
if result.rowcount == 1:
total_repaired += 1
elif result.rowcount == 0:
# Lost the race with a concurrent writer — safe to skip.
total_skipped += 1
else:
raise RuntimeError(
"canonical_id repair updated an unexpected number "
f"of rows for id={row_id}: {result.rowcount}"
)
logger.info(
"[StockDaily] canonical_id repair stats: repaired_count=%s skipped_count=%s",
total_repaired,
total_skipped,
)
def _ensure_intelligence_items_unique_index(self) -> None:
if not self._is_sqlite_engine:
return
@@ -3168,8 +3307,10 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
- SQLite 分支按 chunk 写入以避免绑定参数上限
- ``canonical_id``(Expand-Contract PR2):显式传入则双写;
未传(``None``)时用 ``_derive_canonical_id(code)`` 延迟推导
(index-aware:裸指数码命中注册表时统一到指数 canonical_id),
推导失败写 NULL 降级(D1)。
(裸码恒为 stock canonical,如 ``000300`` -> ``sz000300``;
仅显式指数形式如 ``sh000300`` / ``930955.CSI`` 推导为指数
canonical_id;历史被错误写成指数桶的裸码行由
``_backfill_canonical_ids()`` 幂等修复),推导失败写 NULL 降级(D1)。
Args:
df: 包含日线数据的 DataFrame
@@ -3188,9 +3329,11 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
# D1: canonical_id=None → derive via the Phase 1 parser; on failure
# degrade to NULL (do NOT raise — read path still uses ``code``).
# Empty string is treated like None (never persisted as a key).
# Derivation is index-aware (OR-COR-4f9ffc38): a bare index code
# (e.g. ``000300``) resolves to the index canonical_id (``sh000300``)
# rather than the stock-path canonical_id (``sz000300``).
# Parser contract (OR-COR-4f9ffc38): a bare code always derives the
# stock canonical_id (e.g. ``000300`` -> ``sz000300``); only explicit
# index forms (``sh000300`` / ``930955.CSI``) derive an index canonical_id.
# Historical bare rows mis-written as an index identity are repaired
# separately by the idempotent ``_backfill_canonical_ids()``.
if not canonical_id:
try:
canonical_id = self._derive_canonical_id(code)
+121 -2
View File
@@ -5,7 +5,12 @@ from unittest.mock import patch
import pandas as pd
import pytest
from data_provider.base import BaseFetcher, DataFetcherManager, STANDARD_COLUMNS
from data_provider.base import (
BaseFetcher,
DataFetchError,
DataFetcherManager,
STANDARD_COLUMNS,
)
from src.services.stock_list_parser import (
AnalysisTarget,
IndexEntry,
@@ -78,6 +83,41 @@ def _manager_without_fetchers() -> DataFetcherManager:
return manager
@pytest.mark.parametrize(
"stock_code",
["csi930956", "CSI930956", "930956.CSI", "csi93095", "93095.CSI"],
)
def test_unregistered_csi_daily_is_rejected_before_provider_calls(stock_code) -> None:
fetcher = _FakeFetcher("YfinanceFetcher", priority=1, daily_result=_daily_frame())
manager = DataFetcherManager(fetchers=[fetcher])
with pytest.raises(DataFetchError, match="unregistered CSI index"):
manager.get_daily_data(stock_code)
assert fetcher.daily_calls == []
@pytest.mark.parametrize(
"stock_code",
["csi930956", "CSI930956", "930956.CSI", "csi93095", "93095.CSI"],
)
def test_unregistered_csi_name_is_rejected_before_provider_calls(stock_code) -> None:
fetcher = _FakeFetcher("YfinanceFetcher", priority=1, name_result="wrong")
manager = DataFetcherManager(fetchers=[fetcher])
assert manager.get_stock_name(stock_code) == ""
assert fetcher.name_calls == []
def test_prefetch_stock_names_skips_unregistered_csi() -> None:
fetcher = _FakeFetcher("YfinanceFetcher", priority=1, name_result="wrong")
manager = DataFetcherManager(fetchers=[fetcher])
manager.prefetch_stock_names(["csi930956", "930956.CSI"])
assert fetcher.name_calls == []
@pytest.fixture(autouse=True)
def _reset_daily_source_health():
DataFetcherManager.reset_daily_source_health()
@@ -470,7 +510,6 @@ def test_index_name_static_fallback_uses_only_canonical_key() -> None:
"000016.SH",
"000016.SS",
"000016.SZ",
"SSE50",
],
),
("sz399001", ["sz399001", "399001", "399001.SZ"]),
@@ -563,3 +602,83 @@ def test_index_and_same_bare_stock_names_use_isolated_cache_keys() -> None:
"000016": "深康佳A",
"sh000016": "上证50",
}
# ---------------------------------------------------------------------------
# CSI provider symbol routing.
# ---------------------------------------------------------------------------
def test_csi_provider_symbols_follow_manifest_matrix() -> None:
csi_target = parse_analysis_target("csi930955")
assert csi_target.asset_type == ParseStatus.INDEX
assert csi_target.exchange == "CSI"
# AkShare is the only supported CSI daily provider.
assert DataFetcherManager._cn_index_provider_symbol(
csi_target, "AkshareFetcher"
) == "csi930955"
# Tencent / TickFlow / Yahoo are unsupported for CSI.
assert DataFetcherManager._cn_index_provider_symbol(
csi_target, "TencentFetcher"
) == ""
assert DataFetcherManager._cn_index_provider_symbol(
csi_target, "TickFlowFetcher"
) == ""
assert DataFetcherManager._cn_index_provider_symbol(
csi_target, "YfinanceFetcher"
) == ""
def test_csi_daily_route_skips_unsupported_providers_without_health_failure() -> None:
tencent = _FakeFetcher("TencentFetcher", priority=0, daily_result=_daily_frame())
akshare = _FakeFetcher("AkshareFetcher", priority=1, daily_result=_daily_frame())
tickflow = _FakeFetcher("TickFlowFetcher", priority=2, daily_result=_daily_frame())
yfinance = _FakeFetcher("YfinanceFetcher", priority=3, daily_result=_daily_frame())
manager = DataFetcherManager(fetchers=[tencent, akshare, tickflow, yfinance])
with patch("data_provider.base.record_provider_run") as record_run, patch.object(
DataFetcherManager, "_record_daily_source_failure"
) as record_health_failure:
df, source = manager.get_daily_data("csi930955")
# AkShare succeeds with the csi symbol.
assert not df.empty
assert source == "AkshareFetcher"
assert tencent.daily_calls == []
assert akshare.daily_calls == ["csi930955"]
assert tickflow.daily_calls == []
assert yfinance.daily_calls == []
# Unsupported providers recorded as unsupported, no health failure.
# AkShare's successful call short-circuits the loop, so only Tencent
# (unsupported) and AkShare (success) are recorded.
assert [item.kwargs.get("error_type") for item in record_run.call_args_list] == [
"unsupported",
None,
]
record_health_failure.assert_not_called()
def test_csi_daily_route_returns_empty_when_akshare_fails() -> None:
tencent = _FakeFetcher("TencentFetcher", priority=0, daily_result=_daily_frame())
akshare = _FakeFetcher(
"AkshareFetcher", priority=1, daily_result=RuntimeError("akshare failed")
)
tickflow = _FakeFetcher("TickFlowFetcher", priority=2, daily_result=_daily_frame())
yfinance = _FakeFetcher("YfinanceFetcher", priority=3, daily_result=_daily_frame())
manager = DataFetcherManager(fetchers=[tencent, akshare, tickflow, yfinance])
with patch("data_provider.base.record_provider_run") as record_run:
df, source = manager.get_daily_data("csi930955")
assert df.empty
assert source == ""
assert tencent.daily_calls == []
assert akshare.daily_calls == ["csi930955"]
assert tickflow.daily_calls == []
assert yfinance.daily_calls == []
# Tencent/TickFlow/Yahoo are unsupported (no network), AkShare failed.
assert [item.kwargs["error_type"] for item in record_run.call_args_list] == [
"unsupported",
"RuntimeError",
"unsupported",
"unsupported",
]
+29
View File
@@ -52,6 +52,35 @@ from src.services.history_service import HistoryService
import src.auth as auth
class TestHistoryCsiCandidateConvergence(unittest.TestCase):
"""PR #2267 review remediation: registered CSI explicit identities must
converge in history filter candidates so a record saved under any
equivalent form is reachable from every equivalent query input."""
def test_registered_csi_forms_include_canonical_uppercase_and_aliases(self):
"""A registered CSI identity is a *persisted-read* filter path: the
candidate set must include the parser canonical (``csi930955``), the
old resolver's uppercase canonical (``CSI930955`` — how pre-fix records
were saved) and the IndexEntry's explicit aliases (``930955.CSI``) so a
record stored under any of them is hit by any equivalent input."""
for code in ("csi930955", "930955.CSI", "CSI930955", " csi930955 "):
candidates = HistoryService._history_code_filter_candidates(code)
self.assertEqual(
set(candidates),
{"csi930955", "CSI930955", "930955.CSI"},
)
self.assertEqual(len(candidates), len(set(candidates)))
def test_bare_csi_base_remains_stock_candidates(self):
candidates = HistoryService._history_code_filter_candidates("930955")
self.assertIn("930955", candidates)
self.assertNotIn("csi930955", candidates)
def test_unregistered_csi_form_is_not_converged(self):
candidates = HistoryService._history_code_filter_candidates("csi930956")
self.assertNotIn("csi930956", candidates)
def _analysis_context_pack_overview() -> dict:
return {
"pack_version": "1.0",
+16
View File
@@ -211,6 +211,22 @@ class TestTaskQueue:
assert len(dups2) == 1
assert isinstance(dups2[0], DuplicateTaskError)
def test_task_queue_csi_equivalent_forms_dedupe_to_one_task(self):
"""Real task-layer regression: submitting both the parser canonical
``csi930955`` and its explicit alias ``930955.CSI`` in one batch must
yield exactly ONE accepted task (canonicalised to ``csi930955``) and
ONE duplicate — not two distinct tasks. This guards the task dedupe key
against splitting a registered CSI identity into distinct keys."""
queue = self._build_queue()
accepted, duplicates = queue.submit_tasks_batch(
stock_codes=["csi930955", "930955.CSI"],
)
assert len(accepted) == 1
assert accepted[0].stock_code == "csi930955"
assert len(duplicates) == 1
assert isinstance(duplicates[0], DuplicateTaskError)
assert duplicates[0].stock_code == "csi930955"
class TestIntegration:
"""Integration Tests"""
+348
View File
@@ -29,6 +29,11 @@ from generate_index_from_csv import (
build_stock_index,
load_tushare_data,
load_akshare_data,
load_index_registry_seed,
build_index_entries_from_seed,
validate_index_registry,
run_index_only,
_normalize_index_key,
)
@@ -654,3 +659,346 @@ class TestPinyin:
monkeypatch.setattr(sys, 'argv', ['generate_index_from_csv.py'])
assert main() == 1
# ---------------------------------------------------------------------------
# Index registry seed generation
# ---------------------------------------------------------------------------
class TestIndexRegistrySeed:
"""Seed generates exactly 31 index rows; index-only merge is stable."""
def test_seed_loads_31_rows(self):
rows = load_index_registry_seed()
assert len(rows) == 31
def test_seed_entries_build_valid_index_tuples(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
assert len(entries) == 31
for entry in entries:
assert entry["market"] == "CN"
assert entry["assetType"] == "index"
assert entry["active"] is True
assert entry["popularity"] == 100
assert entry["nameZh"]
assert entry["pinyinFull"]
assert entry["pinyinAbbr"]
def test_seed_canonical_set_matches_manifest(self):
rows = load_index_registry_seed()
canonicals = {row["canonical_code"] for row in rows}
assert len(canonicals) == 31
# Spot-check the 5 original + CSI entries.
assert {"sh000300", "sh000016", "sh000688", "sz399001", "sz399006"} <= canonicals
assert {"csi930955", "csi932365", "csi931052"} <= canonicals
def test_seed_csi_display_is_code_dot_csi(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
csi = {e["canonicalCode"]: e["displayCode"] for e in entries if e["canonicalCode"].startswith("csi")}
assert csi["csi930955"] == "930955.CSI"
assert csi["csi932365"] == "932365.CSI"
def test_validate_index_registry_accepts_valid_seed(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
validate_index_registry(entries) # should not raise
def test_validate_index_registry_rejects_duplicate_canonical(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries.append(dict(entries[0]))
with pytest.raises(ValueError, match="duplicate index canonical"):
validate_index_registry(entries)
def test_validate_index_registry_rejects_bare_numeric_alias(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries[0]["aliases"] = ["000300"]
with pytest.raises(ValueError, match="bare numeric"):
validate_index_registry(entries)
def test_validate_index_registry_rejects_text_alias(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries[0]["aliases"] = ["CSI300"]
with pytest.raises(ValueError, match="explicit code form"):
validate_index_registry(entries)
def test_validate_index_registry_rejects_unknown_namespace(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries[0]["canonicalCode"] = "xx000300"
with pytest.raises(ValueError, match="canonical must match"):
validate_index_registry(entries)
def test_index_only_preserves_non_index_rows_and_appends_31(self, tmp_path):
output = tmp_path / "stocks.index.json"
output.write_text(
json.dumps(
[
["000001.SZ", "000001", "平安银行", "payh", "payh", [], "CN", "stock", True, 100],
["600519.SH", "600519", "贵州茅台", "gzmt", "gzmt", [], "CN", "stock", True, 100],
],
ensure_ascii=False,
),
encoding="utf-8",
)
merged = run_index_only(output, test=True)
non_index = [x for x in merged if not (len(x) > 7 and x[7] == "index")]
index_rows = [x for x in merged if len(x) > 7 and x[7] == "index"]
# Non-index rows preserved in order.
assert [x[0] for x in non_index] == ["000001.SZ", "600519.SH"]
# Exactly 31 index rows appended.
assert len(index_rows) == 31
# Index rows sorted by canonical.
canonicals = [x[0] for x in index_rows]
assert canonicals == sorted(canonicals)
def test_index_only_is_byte_stable(self, tmp_path):
output = tmp_path / "stocks.index.json"
output.write_text(
json.dumps(
[["000001.SZ", "000001", "平安银行", "payh", "payh", [], "CN", "stock", True, 100]],
ensure_ascii=False,
),
encoding="utf-8",
)
run_index_only(output, test=False)
first = output.read_bytes()
run_index_only(output, test=False)
second = output.read_bytes()
assert first == second
def test_index_only_test_mode_does_not_write(self, tmp_path):
output = tmp_path / "stocks.index.json"
output.write_text(
json.dumps(
[["000001.SZ", "000001", "平安银行", "payh", "payh", [], "CN", "stock", True, 100]],
ensure_ascii=False,
),
encoding="utf-8",
)
before = output.read_bytes()
run_index_only(output, test=True)
assert output.read_bytes() == before
def test_full_path_merge_includes_31_index_rows(self, tmp_path, monkeypatch):
"""The full rebuild path (``main()``) merges the same 31 index
rows before compression, so a full stock-index rebuild never erases the
index registry entries."""
import generate_index_from_csv as gen
# A tiny stock list so the full path runs without network.
stocks = [
{"ts_code": "000001.SZ", "symbol": "000001", "name": "平安银行"},
{"ts_code": "600519.SH", "symbol": "600519", "name": "贵州茅台"},
]
monkeypatch.setattr(gen, "load_tushare_data", lambda data_dir: stocks)
monkeypatch.setattr(gen, "require_pypinyin", lambda: True)
monkeypatch.setattr(
sys, "argv", ["generate_index_from_csv.py", "--source", "tushare", "--test"]
)
assert main() == 0
# Rebuild the same pipeline in-process to inspect the merged index rows.
index = build_stock_index(stocks)
seed_rows = load_index_registry_seed()
index_entries = build_index_entries_from_seed(seed_rows)
validate_index_registry(index_entries)
index.extend(index_entries)
compressed = compress_index(index)
index_rows = [item for item in compressed if len(item) > 7 and item[7] == "index"]
assert len(index_rows) == 31
# Stock rows are preserved alongside the index rows.
stock_rows = [item for item in compressed if len(item) > 7 and item[7] == "stock"]
assert len(stock_rows) == 2
def test_validate_index_registry_rejects_stock_key_collision(self):
"""Gap 3: an index canonical/display/alias that collides with an active
stock/ETF key after normalization is rejected."""
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
# A stock row whose canonical ``sh000300`` collides with the index
# canonical ``sh000300``.
non_index = [["sh000300", "sh000300", "沪深300", "hushen300", "hs300", [], "CN", "stock", True, 100]]
with pytest.raises(ValueError, match="collides with active stock/ETF"):
validate_index_registry(entries, non_index_rows=non_index)
def test_validate_index_registry_rejects_stock_alias_collision(self):
"""Gap 3: an index alias that collides with an active stock alias is
rejected."""
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
# A stock row whose alias ``000300.SH`` collides with the sh000300 index
# alias ``000300.SH``.
non_index = [["600519.SH", "600519", "贵州茅台", "gzmt", "gzmt", ["000300.SH"], "CN", "stock", True, 100]]
with pytest.raises(ValueError, match="collides with active stock/ETF"):
validate_index_registry(entries, non_index_rows=non_index)
def test_validate_index_registry_rejects_csi_canonical_stock_collision(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
non_index = [["csi930955", "930955", "冲突股票", "ctgp", "ctgp", [], "CN", "stock", True, 100]]
with pytest.raises(ValueError, match="collides with active stock/ETF"):
validate_index_registry(entries, non_index_rows=non_index)
def test_validate_index_registry_rejects_equivalent_suffix_stock_collision(self):
rows = load_index_registry_seed()
entry = build_index_entries_from_seed(rows)[0]
entry.update({"canonicalCode": "sh600519", "displayCode": "sh600519", "aliases": []})
non_index = [["600519.SH", "600519", "贵州茅台", "gzmt", "gzmt", [], "CN", "stock", True, 100]]
with pytest.raises(ValueError, match="collides with active stock/ETF"):
validate_index_registry([entry], non_index_rows=non_index)
def test_validate_index_registry_rejects_non_string_pinyin(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries[0]["pinyinFull"] = ["hushen300"]
with pytest.raises(ValueError, match="pinyin fields"):
validate_index_registry(entries)
def test_index_only_rejects_stock_key_collision(self, tmp_path):
"""Gap 3: ``run_index_only`` validates the seed index rows against the
existing active stock/ETF rows and rejects a collision."""
output = tmp_path / "stocks.index.json"
output.write_text(
json.dumps(
[["sh000300", "sh000300", "沪深300", "hushen300", "hs300", [], "CN", "stock", True, 100]],
ensure_ascii=False,
),
encoding="utf-8",
)
with pytest.raises(ValueError, match="collides with active stock/ETF"):
run_index_only(output, test=True)
def test_index_only_rejects_malformed_existing_tuple(self, tmp_path):
output = tmp_path / "stocks.index.json"
output.write_text(json.dumps([["too-short"]]), encoding="utf-8")
with pytest.raises(ValueError, match="compressed tuple"):
run_index_only(output, test=True)
def test_build_index_entries_honors_seed_display_code(self):
"""Gap 5: ``build_index_entries_from_seed`` honors the seed
``display_code`` column rather than recomputing it."""
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
by_canonical = {e["canonicalCode"]: e["displayCode"] for e in entries}
# CSI display comes from the seed ``display_code`` column.
assert by_canonical["csi930955"] == "930955.CSI"
assert by_canonical["csi932365"] == "932365.CSI"
# SH/SZ display equals canonical.
assert by_canonical["sh000300"] == "sh000300"
assert by_canonical["sz399001"] == "sz399001"
def test_validate_index_registry_rejects_non_finite_popularity(self):
"""Gap 5: a non-finite popularity (e.g. NaN) is rejected — it is not a
plain integer (NaN is a float), so it fails the integer check."""
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries[0]["popularity"] = float("nan")
with pytest.raises(ValueError, match="non-negative integer"):
validate_index_registry(entries)
@pytest.mark.parametrize(
"bad_popularity",
[1.5, True, -1, -100, 1.0, "100"],
)
def test_validate_index_registry_rejects_non_integer_popularity(
self, bad_popularity
):
"""PR #2267 review fix: only a plain non-negative integer popularity is
valid. Fractional (``1.5``), boolean (``True``), negative and
string-valued popularities are rejected without truncation."""
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries[0]["popularity"] = bad_popularity
with pytest.raises(ValueError, match="non-negative integer"):
validate_index_registry(entries)
def test_validate_index_registry_accepts_integer_popularity(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries[0]["popularity"] = 100
validate_index_registry(entries) # should not raise
def test_seed_rejects_fractional_popularity(self, tmp_path):
seed = tmp_path / "index_registry.csv"
seed.write_text(
"canonical_code,display_code,name_zh,aliases,name_source,popularity\n"
"sh000300,sh000300,沪深300,,腾讯,1.5\n",
encoding="utf-8",
)
with pytest.raises(ValueError, match="plain integer"):
load_index_registry_seed(seed)
def test_seed_rejects_duplicate_normalized_identity_key(self, tmp_path):
seed = tmp_path / "index_registry.csv"
seed.write_text(
"canonical_code,display_code,name_zh,aliases,name_source,popularity\n"
"csi930955,930955.CSI,红利低波100,,东财,100\n"
"sh000300,sh000300,沪深300,csi930955,腾讯,100\n",
encoding="utf-8",
)
with pytest.raises(ValueError, match="already owned by canonical"):
load_index_registry_seed(seed)
def test_csi_prefix_and_suffix_keep_distinct_resolver_keys(self):
assert _normalize_index_key("csi000300") == "csi000300"
assert _normalize_index_key("000300.CSI") == "000300.csi"
def test_seed_rejects_duplicate_aliases_within_one_row(self, tmp_path):
seed = tmp_path / "index_registry.csv"
seed.write_text(
"canonical_code,display_code,name_zh,aliases,name_source,popularity\n"
"sh000300,sh000300,沪深300,000300.CSI|000300.CSI,腾讯,100\n",
encoding="utf-8",
)
with pytest.raises(ValueError, match="duplicate index alias"):
load_index_registry_seed(seed)
def test_build_rejects_duplicate_aliases_when_seed_loader_is_bypassed(self):
rows = load_index_registry_seed()
rows[0]["aliases"] = ["000001.SH", "000001.SH"]
with pytest.raises(ValueError, match="duplicate index alias"):
build_index_entries_from_seed(rows)
def test_build_rejects_fractional_popularity_when_seed_loader_is_bypassed(self):
rows = load_index_registry_seed()
rows[0]["popularity"] = 1.5
with pytest.raises(ValueError, match="non-negative integer"):
build_index_entries_from_seed(rows)
def test_validate_rejects_duplicate_aliases_within_one_entry(self):
rows = load_index_registry_seed()
entries = build_index_entries_from_seed(rows)
entries[0]["aliases"] = ["000001.SH", "000001.SH"]
with pytest.raises(ValueError, match="duplicate index alias"):
validate_index_registry(entries)
def test_full_path_merge_canonical_sorts_index_rows(self, tmp_path, monkeypatch):
"""Gap 5: the full rebuild path canonical-sorts the index rows so the
output is byte-stable and matches ``--index-only`` ordering."""
import generate_index_from_csv as gen
stocks = [
{"ts_code": "000001.SZ", "symbol": "000001", "name": "平安银行"},
{"ts_code": "600519.SH", "symbol": "600519", "name": "贵州茅台"},
]
monkeypatch.setattr(gen, "load_tushare_data", lambda data_dir: stocks)
monkeypatch.setattr(gen, "require_pypinyin", lambda: True)
monkeypatch.setattr(
sys, "argv", ["generate_index_from_csv.py", "--source", "tushare", "--test"]
)
assert main() == 0
index = build_stock_index(stocks)
seed_rows = load_index_registry_seed()
index_entries = build_index_entries_from_seed(seed_rows)
index_entries.sort(key=lambda entry: str(entry["canonicalCode"]))
index.extend(index_entries)
compressed = compress_index(index)
index_rows = [item for item in compressed if len(item) > 7 and item[7] == "index"]
canonicals = [x[0] for x in index_rows]
assert canonicals == sorted(canonicals)
+33
View File
@@ -361,3 +361,36 @@ class TestResolveIndexStockCodeForAnalysis:
with patch("src.data.stock_index_loader.resolve_index_stock_code", return_value=None):
assert resolve_index_stock_code_for_analysis("005930") == "005930"
assert resolve_index_stock_code_for_analysis("AAPL") == "AAPL"
# ------------------------------------------------------------------
# PR #2267 review remediation — registered CSI explicit identity
# convergence (resolver / task dedupe key / history candidates).
# ------------------------------------------------------------------
@pytest.mark.parametrize(
"code",
["csi930955", "930955.CSI", "CSI930955", " csi930955 ", "930955.csi"],
)
def test_registered_csi_forms_converge_to_parser_canonical(self, code):
"""All registered CSI explicit forms resolve to the parser canonical
``csi930955`` so the resolver and task dedupe key do not split the same
index into distinct keys."""
assert resolve_index_stock_code_for_analysis(code) == "csi930955"
@pytest.mark.parametrize(
"code,expected",
[
("csi930956", "CSI930956"),
("930956.CSI", "930956.CSI"),
("CSI930956", "CSI930956"),
],
)
def test_unregistered_csi_forms_keep_existing_behavior(self, code, expected):
"""An unregistered CSI form is NOT converged; it keeps its existing
canonicalized (uppercased) degradation so it never becomes a guessed
index identity."""
assert resolve_index_stock_code_for_analysis(code) == expected
def test_bare_csi_base_remains_stock(self):
"""A bare numeric base of a CSI index stays a stock (no convergence)."""
assert resolve_index_stock_code_for_analysis("930955") == "930955"
+402
View File
@@ -6,6 +6,8 @@ import unittest
from pathlib import Path
from unittest.mock import patch
import pytest
from src.data import stock_index_loader
@@ -336,6 +338,406 @@ class TestStockIndexLoader(unittest.TestCase):
self.assertIsNone(stock_index_loader.resolve_index_stock_code("2330.TW"))
self.assertIsNone(stock_index_loader.resolve_index_stock_code("6505.TWO"))
# ------------------------------------------------------------------
# Active index row loader
# ------------------------------------------------------------------
def _index_payload(self, canonicals):
rows = []
for c in canonicals:
display = f"{c[3:]}.CSI" if c.startswith("csi") else c
rows.append([c, display, f"指数{c}", "zhishu", "zs", [], "CN", "index", True, 100])
return rows
def _pad_payload(self, rows, size=100):
"""Pad a payload with stock rows so it passes the remote min_items check."""
padded = list(rows)
while len(padded) < size:
i = len(padded)
padded.append([f"{i:06d}.SZ", f"{i:06d}", f"股票{i}", "gupiao", "gp", [], "CN", "stock", True, 100])
return padded
def test_load_active_index_rows_returns_index_rows(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
bundled_path.write_text(
json.dumps(
self._index_payload(["sh000300", "csi930955"])
+ [["000001.SZ", "000001", "平安银行", "payh", "payh", [], "CN", "stock", True, 100]],
ensure_ascii=False,
),
encoding="utf-8",
)
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual({r[0] for r in rows}, {"sh000300", "csi930955"})
def test_load_active_index_rows_remote_superset_wins(self):
with tempfile.TemporaryDirectory() as temp_dir:
remote_cache = Path(temp_dir) / "cache" / "stocks.index.json"
bundled_path = Path(temp_dir) / "apps" / "stocks.index.json"
remote_cache.parent.mkdir(parents=True, exist_ok=True)
remote_cache.write_text(
json.dumps(self._pad_payload(self._index_payload(["sh000300", "sh000016", "csi930955"])), ensure_ascii=False),
encoding="utf-8",
)
bundled_path.parent.mkdir(parents=True, exist_ok=True)
bundled_path.write_text(
json.dumps(self._index_payload(["sh000300", "sh000016"]), ensure_ascii=False),
encoding="utf-8",
)
os.utime(remote_cache, (2_000, 2_000))
os.utime(bundled_path, (1_000, 1_000))
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=remote_cache), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(remote_cache, bundled_path)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual({r[0] for r in rows}, {"sh000300", "sh000016", "csi930955"})
def test_load_active_index_rows_remote_subset_falls_back_to_bundled(self):
with tempfile.TemporaryDirectory() as temp_dir:
remote_cache = Path(temp_dir) / "cache" / "stocks.index.json"
bundled_path = Path(temp_dir) / "apps" / "stocks.index.json"
remote_cache.parent.mkdir(parents=True, exist_ok=True)
# Remote drops sh000016 (a bundled baseline canonical).
remote_cache.write_text(
json.dumps(self._pad_payload(self._index_payload(["sh000300"])), ensure_ascii=False),
encoding="utf-8",
)
bundled_path.parent.mkdir(parents=True, exist_ok=True)
bundled_path.write_text(
json.dumps(self._index_payload(["sh000300", "sh000016"]), ensure_ascii=False),
encoding="utf-8",
)
os.utime(remote_cache, (2_000, 2_000))
os.utime(bundled_path, (1_000, 1_000))
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=remote_cache), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(remote_cache, bundled_path)):
rows = stock_index_loader._load_active_index_rows()
# Falls back to bundled (which has both).
self.assertEqual({r[0] for r in rows}, {"sh000300", "sh000016"})
def test_load_active_index_rows_all_failed_returns_empty(self):
with tempfile.TemporaryDirectory() as temp_dir:
missing_path = Path(temp_dir) / "missing.json"
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=missing_path), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(missing_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_load_active_index_rows_rejects_semantic_conflict(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
# Two index rows share the same canonical — semantic conflict.
bundled_path.write_text(
json.dumps(
self._index_payload(["sh000300"]) + self._index_payload(["sh000300"]),
ensure_ascii=False,
),
encoding="utf-8",
)
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
# Semantic conflict rejects the candidate → empty registry.
self.assertEqual(rows, [])
def test_clear_stock_index_cache_clears_active_index_rows(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
bundled_path.write_text(
json.dumps(self._index_payload(["sh000300"]), ensure_ascii=False),
encoding="utf-8",
)
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
first = stock_index_loader._load_active_index_rows()
stock_index_loader.clear_stock_index_cache()
second = stock_index_loader._load_active_index_rows()
self.assertEqual(first, second)
def test_validate_index_rows_semantics_rejects_malformed_csi_display(self):
"""Gap 1: a CSI row whose display is not ``{code}.CSI`` is rejected."""
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
# csi930955 with a wrong display (not ``930955.CSI``).
bundled_path.write_text(
json.dumps(
[["csi930955", "930955", "红利低波100", "honglidibo100", "hldb100", [], "CN", "index", True, 100]],
ensure_ascii=False,
),
encoding="utf-8",
)
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
# Malformed CSI display rejects the candidate → empty registry.
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_rejects_stock_key_collision(self):
"""Gap 3: an index canonical/display/alias that collides with an active
stock/ETF key after normalization is rejected."""
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
# A stock row whose canonical ``sh000300`` collides with the index
# canonical ``sh000300`` in the same candidate.
bundled_path.write_text(
json.dumps(
self._index_payload(["sh000300"])
+ [["sh000300", "sh000300", "冲突股票", "ctgp", "ctgp", [], "CN", "stock", True, 100]],
ensure_ascii=False,
),
encoding="utf-8",
)
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
# The candidate is rejected because both identities claim one key.
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_rejects_csi_canonical_stock_collision(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
bundled_path.write_text(
json.dumps(
self._index_payload(["csi930955"])
+ [["csi930955", "930955", "冲突股票", "ctgp", "ctgp", [], "CN", "stock", True, 100]],
ensure_ascii=False,
),
encoding="utf-8",
)
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_rejects_text_alias(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
payload = self._index_payload(["sh000300"])
payload[0][5] = ["CSI300"]
bundled_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_rejects_blank_pinyin(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
payload = self._index_payload(["sh000300"])
payload[0][3] = ""
bundled_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_rejects_non_string_pinyin(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
payload = self._index_payload(["sh000300"])
payload[0][3] = ["hushen300"]
bundled_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_rejects_equivalent_suffix_stock_collision(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
payload = self._index_payload(["sh600519"])
payload += [["600519.SH", "600519", "贵州茅台", "gzmt", "gzmt", [], "CN", "stock", True, 100]]
bundled_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_load_active_index_rows_rejects_mixed_valid_and_short_local_rows(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
payload = self._index_payload(["sh000300"]) + [["too-short"]]
bundled_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_rejects_non_integer_popularity(self):
"""PR #2267 review fix: only a plain non-negative integer popularity is
valid at the runtime candidate boundary; fractional/boolean/negative/
string popularities reject the candidate."""
for bad_popularity in (1.5, True, -1, 1.0, "100"):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
payload = self._index_payload(["sh000300"])
payload[0][9] = bad_popularity
bundled_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_accepts_integer_popularity(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
bundled_path.write_text(
json.dumps(self._index_payload(["sh000300"]), ensure_ascii=False),
encoding="utf-8",
)
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual({r[0] for r in rows}, {"sh000300"})
def test_validate_index_rows_semantics_rejects_cross_entry_duplicate_alias(self):
"""PR #2267 review fix: two index rows whose aliases normalize to the
same identity key (e.g. ``csi930955`` and ``CSI930955`` as aliases of
different canonicals) must reject the candidate — no silent overwrite."""
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
payload = self._index_payload(["sh000300", "sz399001"])
payload[0][5] = ["csi930955"]
payload[1][5] = ["CSI930955"]
bundled_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_validate_index_rows_semantics_rejects_duplicate_aliases_in_one_row(self):
with tempfile.TemporaryDirectory() as temp_dir:
bundled_path = Path(temp_dir) / "stocks.index.json"
payload = self._index_payload(["sh000300"])
payload[0][5] = ["000300.CSI", "000300.CSI"]
bundled_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=Path(temp_dir) / "missing.json"), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(bundled_path,)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual(rows, [])
def test_load_active_index_rows_bundled_baseline_ignores_newer_legacy_static(self):
"""Gap 4: the bundled baseline comes from the declared bundled candidate
(``apps/dsa-web/public``), not the first non-remote candidate ordered by
mtime. A newer legacy ``static`` candidate must not become the baseline."""
with tempfile.TemporaryDirectory() as temp_dir:
remote_cache = Path(temp_dir) / "cache" / "stocks.index.json"
bundled_path = Path(temp_dir) / "apps" / "stocks.index.json"
legacy_static = Path(temp_dir) / "static" / "stocks.index.json"
for p in (remote_cache, bundled_path, legacy_static):
p.parent.mkdir(parents=True, exist_ok=True)
# Remote is a superset of the bundled baseline.
remote_cache.write_text(
json.dumps(self._pad_payload(self._index_payload(["sh000300", "sh000016", "csi930955"])), ensure_ascii=False),
encoding="utf-8",
)
bundled_path.write_text(
json.dumps(self._index_payload(["sh000300", "sh000016"]), ensure_ascii=False),
encoding="utf-8",
)
# Legacy static is NEWER than bundled but only carries sh000300.
legacy_static.write_text(
json.dumps(self._index_payload(["sh000300"]), ensure_ascii=False),
encoding="utf-8",
)
os.utime(remote_cache, (3_000, 3_000))
os.utime(legacy_static, (2_000, 2_000))
os.utime(bundled_path, (1_000, 1_000))
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=remote_cache), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(remote_cache, bundled_path, legacy_static)):
rows = stock_index_loader._load_active_index_rows()
# Remote superset of the bundled baseline wins.
self.assertEqual({r[0] for r in rows}, {"sh000300", "sh000016", "csi930955"})
def test_load_active_index_rows_malformed_bundled_falls_back_to_valid_remote(self):
"""Gap 4: a malformed bundled candidate (fails semantic validation) must
not become the baseline; a valid remote superset is then accepted."""
with tempfile.TemporaryDirectory() as temp_dir:
remote_cache = Path(temp_dir) / "cache" / "stocks.index.json"
bundled_path = Path(temp_dir) / "apps" / "stocks.index.json"
remote_cache.parent.mkdir(parents=True, exist_ok=True)
bundled_path.parent.mkdir(parents=True, exist_ok=True)
# Remote is a valid superset.
remote_cache.write_text(
json.dumps(self._pad_payload(self._index_payload(["sh000300", "sh000016", "csi930955"])), ensure_ascii=False),
encoding="utf-8",
)
# Bundled has a malformed CSI display (not ``{code}.CSI``).
bundled_path.write_text(
json.dumps(
[["csi930955", "930955", "红利低波100", "honglidibo100", "hldb100", [], "CN", "index", True, 100]],
ensure_ascii=False,
),
encoding="utf-8",
)
os.utime(remote_cache, (2_000, 2_000))
os.utime(bundled_path, (1_000, 1_000))
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=remote_cache), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(remote_cache, bundled_path)):
rows = stock_index_loader._load_active_index_rows()
# Malformed bundled is skipped; valid remote superset is loaded.
self.assertEqual({r[0] for r in rows}, {"sh000300", "sh000016", "csi930955"})
def test_load_active_index_rows_legacy_static_subset_cannot_bypass_bundled_baseline(self):
"""Review remediation: when the remote cache is missing/invalid, a newer
legacy ``static`` candidate that is a SUBSET of the bundled baseline must
NOT be selected — the bundled baseline wins so no active index is lost."""
with tempfile.TemporaryDirectory() as temp_dir:
remote_cache = Path(temp_dir) / "cache" / "stocks.index.json"
bundled_path = Path(temp_dir) / "apps" / "stocks.index.json"
legacy_static = Path(temp_dir) / "static" / "stocks.index.json"
for p in (remote_cache, bundled_path, legacy_static):
p.parent.mkdir(parents=True, exist_ok=True)
# Remote cache is invalid (not JSON).
remote_cache.write_text("not-json", encoding="utf-8")
# Bundled baseline carries sh000300 + sh000016.
bundled_path.write_text(
json.dumps(self._index_payload(["sh000300", "sh000016"]), ensure_ascii=False),
encoding="utf-8",
)
# Legacy static is NEWER than bundled but only carries sh000300.
legacy_static.write_text(
json.dumps(self._index_payload(["sh000300"]), ensure_ascii=False),
encoding="utf-8",
)
os.utime(remote_cache, (3_000, 3_000))
os.utime(legacy_static, (2_000, 2_000))
os.utime(bundled_path, (1_000, 1_000))
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=remote_cache), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(remote_cache, bundled_path, legacy_static)):
rows = stock_index_loader._load_active_index_rows()
# Bundled baseline wins (both canonicals preserved).
self.assertEqual({r[0] for r in rows}, {"sh000300", "sh000016"})
def test_load_active_index_rows_legacy_static_superset_still_wins(self):
"""A legacy ``static`` candidate that is a legal SUPERSET of the bundled
baseline is still accepted (future supersets are allowed)."""
with tempfile.TemporaryDirectory() as temp_dir:
remote_cache = Path(temp_dir) / "cache" / "stocks.index.json"
bundled_path = Path(temp_dir) / "apps" / "stocks.index.json"
legacy_static = Path(temp_dir) / "static" / "stocks.index.json"
for p in (remote_cache, bundled_path, legacy_static):
p.parent.mkdir(parents=True, exist_ok=True)
remote_cache.write_text("not-json", encoding="utf-8")
bundled_path.write_text(
json.dumps(self._index_payload(["sh000300", "sh000016"]), ensure_ascii=False),
encoding="utf-8",
)
# Legacy static is a superset (adds csi930955).
legacy_static.write_text(
json.dumps(self._index_payload(["sh000300", "sh000016", "csi930955"]), ensure_ascii=False),
encoding="utf-8",
)
os.utime(remote_cache, (3_000, 3_000))
os.utime(legacy_static, (2_000, 2_000))
os.utime(bundled_path, (1_000, 1_000))
with patch.object(stock_index_loader, "get_remote_stock_index_cache_path", return_value=remote_cache), \
patch.object(stock_index_loader, "get_stock_index_candidate_paths", return_value=(remote_cache, bundled_path, legacy_static)):
rows = stock_index_loader._load_active_index_rows()
self.assertEqual({r[0] for r in rows}, {"sh000300", "sh000016", "csi930955"})
if __name__ == "__main__":
unittest.main()
+191 -14
View File
@@ -138,16 +138,19 @@ class TestContract2BareCodeDefaultsToStock:
assert target.matched_index.display_name == "沪深300"
def test_bare_000001_is_stock(self) -> None:
"""Conflict code: ``000001`` is平安银行 (SZ stock) AND the深证成指
isn't this — actually 000001.SZ is the stock, the index is sz399001.
So bare 000001 → stock, no registry hit.
"""Conflict code: ``000001`` is平安银行 (SZ stock) AND the上证指数
(``sh000001``) is now in the registry. Per contract #2 bare codes
always resolve to stock; the index conflict is surfaced via
``matched_index`` without flipping asset_type.
"""
target = parse_analysis_target("000001")
assert target.asset_type == ParseStatus.STOCK
assert target.exchange == "SZ"
assert target.matched_index is None
# canonical_id is round-trippable: sh/sz prefix synthesised from 0/2/3.
assert target.canonical_id == "sz000001"
# The registry now carries sh000001 (上证指数) whose alias base is 000001.
assert target.matched_index is not None
assert target.matched_index.canonical_id == "sh000001"
def test_bare_600519_is_sh_stock(self) -> None:
target = parse_analysis_target("600519")
@@ -369,14 +372,18 @@ class TestEdgeCases:
# Default registry — public API surface.
# ---------------------------------------------------------------------------
class TestDefaultIndexRegistry:
def test_default_registry_has_five_entries(self) -> None:
def test_default_registry_has_31_entries(self) -> None:
registry = default_index_registry()
assert len(registry) == 5
assert len(registry) == 31
def test_default_registry_canonical_ids(self) -> None:
registry = default_index_registry()
ids = {entry.canonical_id for entry in registry}
assert ids == {"sh000300", "sh000016", "sh000688", "sz399001", "sz399006"}
assert len(ids) == 31
# The 5 original hard-coded indices are preserved.
assert {"sh000300", "sh000016", "sh000688", "sz399001", "sz399006"} <= ids
# CSI entries are present.
assert {"csi930955", "csi932365"} <= ids
def test_default_registry_find_by_prefixed_code(self) -> None:
registry = default_index_registry()
@@ -391,6 +398,49 @@ class TestDefaultIndexRegistry:
assert registry.find_by_prefixed_code("hk", "000300") is None
assert registry.find_by_prefixed_code("us", "000300") is None
def test_default_registry_find_by_explicit_key_csi(self) -> None:
registry = default_index_registry()
entry = registry.find_by_explicit_key("csi930955")
assert entry is not None
assert entry.canonical_id == "csi930955"
assert entry.exchange == "CSI"
# display form also resolves
entry2 = registry.find_by_explicit_key("930955.CSI")
assert entry2 is not None
assert entry2.canonical_id == "csi930955"
def test_default_registry_find_by_bare_conflict(self) -> None:
registry = default_index_registry()
# 930955 is the bare base of the csi930955 alias.
entry = registry.find_by_bare_conflict("930955")
assert entry is not None
assert entry.canonical_id == "csi930955"
def test_display_name_is_not_an_identity_alias(self) -> None:
"""Gap 2: a Chinese display name (e.g. ``沪深300``) must never resolve
as an index identity — text names are not identity aliases."""
registry = default_index_registry()
assert registry.find_by_explicit_key("沪深300") is None
assert registry.find_by_explicit_key("上证50") is None
# The canonical/display/alias code forms still resolve.
assert registry.find_by_explicit_key("sh000300") is not None
assert registry.find_by_explicit_key("000300.SH") is not None
# Parsing the Chinese name must not elevate to index.
target = parse_analysis_target("沪深300")
assert target.asset_type != ParseStatus.INDEX
def test_custom_registry_rejects_text_identity_alias(self) -> None:
with pytest.raises(ValueError, match="explicit code form"):
IndexRegistry((
IndexEntry(
bare_code="000300",
exchange="SH",
canonical_id="sh000300",
display_name="沪深300",
aliases=("CSI300",),
),
))
# ---------------------------------------------------------------------------
# Batch parsing helper.
@@ -425,15 +475,14 @@ class TestMaintainerSpecSamples:
assert target.asset_type == ParseStatus.INDEX
assert target.canonical_id == "sh000300"
def test_sz399300_falls_back_to_stock(self) -> None:
"""sz399300 is NOT in the canonical 5-index white-list (we only carry
sz399001 + sz399006 for the SZ side). Per contract #3 it degrades to
stock. This sample guards against accidental future registry expand
that would silently flip behaviour.
def test_sz399300_resolves_to_index(self) -> None:
"""``sz399300`` is a registered cross-market alias of ``sh000300``
(沪深300) per the manifest, so it resolves to the index.
"""
target = parse_analysis_target("sz399300")
assert target.asset_type == ParseStatus.STOCK
assert target.exchange == "SZ"
assert target.asset_type == ParseStatus.INDEX
assert target.canonical_id == "sh000300"
assert target.exchange == "SH"
def test_sh600519(self) -> None:
target = parse_analysis_target("sh600519")
@@ -882,3 +931,131 @@ class TestExplicitExchangeSuffixRejections:
assert target.canonical_id == code
assert target.unsupported_reason is not None
assert expected_exchange in target.unsupported_reason
# ---------------------------------------------------------------------------
# Index registry & CSI provider symbol governance.
# ---------------------------------------------------------------------------
class TestIndexAliasMatrix:
"""Canonical/display/alias converge; CSI explicit; bare stays stock."""
@pytest.mark.parametrize(
"code,expected_canonical,expected_exchange",
[
("sh000300", "sh000300", "SH"),
("000300.SH", "sh000300", "SH"),
("sz399300", "sh000300", "SH"),
("399300.SZ", "sh000300", "SH"),
("000300.CSI", "sh000300", "SH"),
],
)
def test_sh000300_alias_forms_resolve_to_index(
self, code: str, expected_canonical: str, expected_exchange: str
) -> None:
target = parse_analysis_target(code)
assert target.asset_type == ParseStatus.INDEX
assert target.canonical_id == expected_canonical
assert target.exchange == expected_exchange
@pytest.mark.parametrize(
"code",
["csi930955", "930955.CSI", "CSI930955", " csi930955 "],
)
def test_csi930955_forms_resolve_to_index(self, code: str) -> None:
target = parse_analysis_target(code)
assert target.asset_type == ParseStatus.INDEX
assert target.canonical_id == "csi930955"
assert target.exchange == "CSI"
def test_unknown_csi_is_unsupported_not_us_stock(self) -> None:
target = parse_analysis_target("930956.CSI")
assert target.asset_type == ParseStatus.UNSUPPORTED
assert target.exchange == "UNKNOWN"
assert target.unsupported_reason is not None
assert "CSI" in target.unsupported_reason
@pytest.mark.parametrize(
"code",
[
"csi930956",
"CSI930956",
"930956.CSI",
"csi000300",
"CSI000300",
"csi93095",
"csi9309557",
"93095.CSI",
"9309557.CSI",
],
)
def test_unregistered_explicit_csi_prefix_and_suffix_is_unsupported(
self, code: str
) -> None:
"""PR #2267 review fix: an unknown explicit ``csi`` prefix (or an
unregistered ``.CSI`` suffix) must surface as ``unsupported`` — never a
US ticker and never a guessed SH/SZ index. Only a manifest-owned
identity may route/persist as an index."""
target = parse_analysis_target(code)
assert target.asset_type == ParseStatus.UNSUPPORTED
assert target.exchange == "UNKNOWN"
assert target.canonical_id == code
assert target.unsupported_reason is not None
assert "CSI" in target.unsupported_reason
def test_csi_prefix_is_not_overequated_to_csi_suffix_alias(self) -> None:
"""PR #2267 review fix: ``000300.CSI`` is the registered alias of
``sh000300``, but the bare ``csi000300`` prefix is NOT — it must stay
unsupported rather than being promoted to the same-code index."""
target = parse_analysis_target("000300.CSI")
assert target.asset_type == ParseStatus.INDEX
assert target.canonical_id == "sh000300"
assert target.exchange == "SH"
unregistered = parse_analysis_target("csi000300")
assert unregistered.asset_type == ParseStatus.UNSUPPORTED
assert unregistered.exchange == "UNKNOWN"
def test_us_ticker_starting_with_csi_remains_stock(self) -> None:
target = parse_analysis_target("CSIQ")
assert target.asset_type == ParseStatus.STOCK
assert target.canonical_id == "CSIQ"
@pytest.mark.parametrize(
"code,expected_canonical",
[
("000001", "sz000001"),
("000016", "sz000016"),
("000688", "sz000688"),
("000300", "sz000300"),
("399300", "sz399300"),
("930955", "bj930955"),
],
)
def test_bare_conflict_codes_stay_stock_with_matched_index(
self, code: str, expected_canonical: str
) -> None:
target = parse_analysis_target(code)
assert target.asset_type == ParseStatus.STOCK
assert target.canonical_id == expected_canonical
assert target.matched_index is not None
def test_unregistered_sh_sz_prefixed_stock_keeps_stock_path(self) -> None:
target = parse_analysis_target("sh600519")
assert target.asset_type == ParseStatus.STOCK
assert target.canonical_id == "sh600519"
target2 = parse_analysis_target("sz000001")
assert target2.asset_type == ParseStatus.STOCK
assert target2.canonical_id == "sz000001"
def test_old_text_aliases_do_not_resolve_to_index(self) -> None:
# CSI300 / HS300 / SSE50 / STAR50 are not in the manifest aliases.
for code in ("CSI300", "HS300", "SSE50", "STAR50", "SZSE", "ChiNext"):
target = parse_analysis_target(code)
assert target.asset_type != ParseStatus.INDEX
def test_nfkc_and_case_normalization_do_not_create_second_identity(self) -> None:
a = parse_analysis_target("csi930955")
b = parse_analysis_target("CSI930955") # full-width
assert a.asset_type == ParseStatus.INDEX
assert b.asset_type == ParseStatus.INDEX
assert a.canonical_id == b.canonical_id
+289 -24
View File
@@ -1196,6 +1196,39 @@ class TestStorage(unittest.TestCase):
"CREATE INDEX ix_code_date ON stock_daily (code, date)"
)
@staticmethod
def _create_legacy_stock_daily_with_canonical_id(db_path: str) -> None:
"""``stock_daily`` schema WITH a ``canonical_id`` column (for repair tests)."""
with sqlite3.connect(db_path) as conn:
conn.execute(
"""CREATE TABLE stock_daily (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code VARCHAR(10) NOT NULL,
date DATE NOT NULL,
open FLOAT,
high FLOAT,
low FLOAT,
close FLOAT,
volume FLOAT,
amount FLOAT,
pct_chg FLOAT,
ma5 FLOAT,
ma10 FLOAT,
ma20 FLOAT,
volume_ratio FLOAT,
data_source VARCHAR(50),
created_at DATETIME,
updated_at DATETIME,
canonical_id VARCHAR(32)
)"""
)
conn.execute(
"CREATE UNIQUE INDEX uix_code_date ON stock_daily (code, date)"
)
conn.execute(
"CREATE INDEX ix_code_date ON stock_daily (code, date)"
)
@staticmethod
def _make_temp_db_path() -> tuple:
"""Return (db_dir, db_path); Windows-safe cleanup via shutil.rmtree."""
@@ -1329,14 +1362,13 @@ class TestStorage(unittest.TestCase):
Config.reset_instance()
self._cleanup_temp_dir(db_dir)
def test_canonical_id_backfill_bare_index_unifies_to_index_canonical_id(self):
"""Index-aware backfill (OR-COR-4f9ffc38): bare ``000300`` hits the
index registry (``matched_index`` is non-None) and backfills to the
index canonical_id ``sh000300`` — NOT ``sz000300`` (the stock-path
canonical_id the classifier would synthesise for a 6-digit
``0``-prefixed code). Without this unification the same CSI-300 index
would split across two canonical_id buckets depending on whether the
caller passed a bare code or an explicit ``sh000300`` prefix."""
def test_canonical_id_backfill_bare_code_colliding_with_index_stays_stock(self):
"""``_derive_canonical_id`` no longer reads
``matched_index.canonical_id``. A bare ``000300`` (which collides with
the CSI-300 index) now derives to the stock-path canonical_id
``sz000300`` — the parser contract says bare codes are always stock.
The index conflict is surfaced via ``matched_index`` only, never used
to override the canonical_id."""
DatabaseManager.reset_instance()
db_dir, db_path = self._make_temp_db_path()
@@ -1355,10 +1387,8 @@ class TestStorage(unittest.TestCase):
"SELECT canonical_id FROM stock_daily WHERE code='000300'"
).fetchone()[0]
# Bare index code unifies to the index canonical_id via
# ``matched_index.canonical_id`` so bare ``000300`` and explicit
# ``sh000300`` land in the same bucket.
self.assertEqual(canonical_id, "sh000300")
# Bare code resolves to the stock-path canonical_id.
self.assertEqual(canonical_id, "sz000300")
finally:
DatabaseManager.reset_instance()
Config.reset_instance()
@@ -1683,11 +1713,10 @@ class TestStorage(unittest.TestCase):
DatabaseManager.reset_instance()
self._cleanup_temp_dir(db_dir)
def test_save_daily_data_derives_index_aware_canonical_id_for_bare_index_code(self):
"""OR-COR-4f9ffc38: ``save_daily_data(df, code="000300")`` with no
explicit canonical_id writes ``sh000300`` (index canonical_id), not
``sz000300`` (stock-path canonical_id the classifier would synthesise
for a 6-digit ``0``-prefixed code)."""
def test_save_daily_data_derives_stock_canonical_id_for_bare_index_collision(self):
"""``save_daily_data(df, code="000300")`` with no explicit
canonical_id writes ``sz000300`` (stock-path canonical_id), NOT
``sh000300`` (the index canonical_id). Bare codes are always stock."""
DatabaseManager.reset_instance()
db_dir, db_path = self._make_temp_db_path()
db = DatabaseManager(db_url=f"sqlite:///{db_path}")
@@ -1710,16 +1739,16 @@ class TestStorage(unittest.TestCase):
and_(StockDaily.code == "000300", StockDaily.date == date(2026, 4, 5))
)
).scalar_one()
self.assertEqual(row.canonical_id, "sh000300")
self.assertEqual(row.canonical_id, "sz000300")
finally:
DatabaseManager.reset_instance()
self._cleanup_temp_dir(db_dir)
def test_save_daily_data_converges_alias_and_prefix_to_same_canonical_id(self):
"""OR-COR-4f9ffc38: bare ``000300``, ``sh000300`` and ``000300.SH``
all converge to the same canonical_id ``sh000300`` when written via
``save_daily_data`` (no explicit canonical_id), so the same underlying
index is never split across buckets by input form."""
def test_save_daily_data_explicit_index_forms_derive_index_canonical_id(self):
"""Explicit index forms (``sh000300`` / ``000300.SH``)
derive to the index canonical_id ``sh000300``, while the bare code
``000300`` derives to the stock-path ``sz000300``. Explicit index and
bare stock are intentionally different buckets."""
DatabaseManager.reset_instance()
db_dir, db_path = self._make_temp_db_path()
db = DatabaseManager(db_url=f"sqlite:///{db_path}")
@@ -1747,7 +1776,7 @@ class TestStorage(unittest.TestCase):
self.assertEqual(
{row.code: row.canonical_id for row in rows},
{
"000300": "sh000300",
"000300": "sz000300",
"sh000300": "sh000300",
"000300.SH": "sh000300",
},
@@ -1797,5 +1826,241 @@ class TestStorage(unittest.TestCase):
DatabaseManager.reset_instance()
Config.reset_instance()
# ------------------------------------------------------------------
# canonical_id repair
# ------------------------------------------------------------------
def test_derive_canonical_id_bare_conflict_stays_stock(self):
"""``_derive_canonical_id`` returns the parser stock
canonical for bare conflict codes, never the index canonical."""
DatabaseManager.reset_instance()
db = DatabaseManager(db_url="sqlite:///:memory:")
try:
self.assertEqual(db._derive_canonical_id("000016"), "sz000016")
self.assertEqual(db._derive_canonical_id("930955"), "bj930955")
self.assertEqual(db._derive_canonical_id("sh000016"), "sh000016")
self.assertEqual(db._derive_canonical_id("930955.CSI"), "csi930955")
finally:
DatabaseManager.reset_instance()
def test_derive_canonical_id_rejects_unregistered_csi(self):
"""PR #2267 review fix: an unsupported identity (unregistered
``csi`` prefix or ``.CSI`` suffix) must NOT enter a persistent
canonical bucket — ``_derive_canonical_id`` returns None so the
caller persists NULL instead of the raw token."""
DatabaseManager.reset_instance()
db = DatabaseManager(db_url="sqlite:///:memory:")
try:
self.assertIsNone(db._derive_canonical_id("csi930956"))
self.assertIsNone(db._derive_canonical_id("CSI930956"))
self.assertIsNone(db._derive_canonical_id("930956.CSI"))
self.assertIsNone(db._derive_canonical_id("csi000300"))
finally:
DatabaseManager.reset_instance()
def test_canonical_id_repair_fixes_bare_misbucketed_rows(self):
"""Rows whose bare code has an erroneous index canonical are
repaired to the parser stock canonical; explicit index rows and
correct stock rows are untouched."""
DatabaseManager.reset_instance()
db_dir, db_path = self._make_temp_db_path()
try:
self._create_legacy_stock_daily_with_canonical_id(db_path)
with sqlite3.connect(db_path) as conn:
conn.executemany(
"""INSERT INTO stock_daily (code, date, close, canonical_id)
VALUES (?, ?, ?, ?)""",
[
# Mis-bucketed bare codes (old index-aware derivation).
("000001", "2026-01-01", 1.0, "sh000001"),
("000016", "2026-01-02", 2.0, "sh000016"),
("000688", "2026-01-03", 3.0, "sh000688"),
("930955", "2026-01-04", 4.0, "csi930955"),
# Explicit index rows — must NOT be modified.
("sh000016", "2026-01-05", 5.0, "sh000016"),
("930955.CSI", "2026-01-06", 6.0, "csi930955"),
# Correct stock row — must NOT be modified.
("600519", "2026-01-07", 7.0, "sh600519"),
# Unrelated row — must NOT be modified.
("AAPL", "2026-01-08", 8.0, "AAPL"),
],
)
DatabaseManager(db_url=f"sqlite:///{db_path}")
with sqlite3.connect(db_path) as conn:
rows = dict(
conn.execute(
"SELECT code, canonical_id FROM stock_daily ORDER BY id"
).fetchall()
)
self.assertEqual(rows["000001"], "sz000001")
self.assertEqual(rows["000016"], "sz000016")
self.assertEqual(rows["000688"], "sz000688")
self.assertEqual(rows["930955"], "bj930955")
# Explicit index rows preserved.
self.assertEqual(rows["sh000016"], "sh000016")
self.assertEqual(rows["930955.CSI"], "csi930955")
# Correct stock + unrelated rows preserved.
self.assertEqual(rows["600519"], "sh600519")
self.assertEqual(rows["AAPL"], "AAPL")
finally:
DatabaseManager.reset_instance()
Config.reset_instance()
self._cleanup_temp_dir(db_dir)
def test_canonical_id_repair_is_idempotent(self):
"""Running the repair a second time repairs 0 rows."""
DatabaseManager.reset_instance()
db_dir, db_path = self._make_temp_db_path()
try:
self._create_legacy_stock_daily_with_canonical_id(db_path)
with sqlite3.connect(db_path) as conn:
conn.execute(
"INSERT INTO stock_daily (code, date, close, canonical_id) "
"VALUES (?, ?, ?, ?)",
("000016", "2026-01-01", 2.0, "sh000016"),
)
DatabaseManager(db_url=f"sqlite:///{db_path}")
with sqlite3.connect(db_path) as conn:
first = conn.execute(
"SELECT canonical_id FROM stock_daily WHERE code='000016'"
).fetchone()[0]
self.assertEqual(first, "sz000016")
DatabaseManager.reset_instance()
with self.assertLogs("src.storage", level="INFO") as logs:
DatabaseManager(db_url=f"sqlite:///{db_path}")
log_text = "\n".join(logs.output)
self.assertIn("repaired_count=0", log_text)
finally:
DatabaseManager.reset_instance()
Config.reset_instance()
self._cleanup_temp_dir(db_dir)
def test_canonical_id_repair_skips_when_registry_empty(self):
"""When the index registry is empty the repair is a no-op and
logs a WARNING; no rows are modified."""
DatabaseManager.reset_instance()
db_dir, db_path = self._make_temp_db_path()
try:
self._create_legacy_stock_daily_with_canonical_id(db_path)
with sqlite3.connect(db_path) as conn:
conn.execute(
"INSERT INTO stock_daily (code, date, close, canonical_id) "
"VALUES (?, ?, ?, ?)",
("000016", "2026-01-01", 2.0, "sh000016"),
)
with patch(
"src.data.stock_index_loader._load_active_index_rows",
return_value=[],
), self.assertLogs("src.storage", level="WARNING") as logs:
DatabaseManager(db_url=f"sqlite:///{db_path}")
with sqlite3.connect(db_path) as conn:
canonical_id = conn.execute(
"SELECT canonical_id FROM stock_daily WHERE code='000016'"
).fetchone()[0]
self.assertEqual(canonical_id, "sh000016")
self.assertTrue(
any("registry is empty" in record.getMessage() for record in logs.records)
)
finally:
DatabaseManager.reset_instance()
Config.reset_instance()
self._cleanup_temp_dir(db_dir)
def test_canonical_id_repair_handles_concurrent_rewrite_safely(self):
"""A conditional UPDATE that loses the race (row already
rewritten) is skipped, not double-counted."""
DatabaseManager.reset_instance()
db_dir, db_path = self._make_temp_db_path()
try:
self._create_legacy_stock_daily_with_canonical_id(db_path)
with sqlite3.connect(db_path) as conn:
conn.execute(
"INSERT INTO stock_daily (code, date, close, canonical_id) "
"VALUES (?, ?, ?, ?)",
("000016", "2026-01-01", 2.0, "sh000016"),
)
real_parse = stock_list_parser_module.parse_analysis_target
def racing_parse(code):
# Simulate a concurrent writer that already fixed the row.
if code == "000016":
with sqlite3.connect(db_path) as conn:
conn.execute(
"UPDATE stock_daily SET canonical_id='sz000016' "
"WHERE code='000016'"
)
conn.commit()
return real_parse(code)
with patch(
"src.services.stock_list_parser.parse_analysis_target",
side_effect=racing_parse,
):
DatabaseManager(db_url=f"sqlite:///{db_path}")
with sqlite3.connect(db_path) as conn:
canonical_id = conn.execute(
"SELECT canonical_id FROM stock_daily WHERE code='000016'"
).fetchone()[0]
self.assertEqual(canonical_id, "sz000016")
finally:
DatabaseManager.reset_instance()
Config.reset_instance()
self._cleanup_temp_dir(db_dir)
def test_canonical_id_repair_queries_only_relevant_rows(self):
"""Gap 6: the repair SQL filters to rows whose canonical_id is in the
active index set, so rows with unrelated canonical_ids are never
scanned/derived."""
DatabaseManager.reset_instance()
db_dir, db_path = self._make_temp_db_path()
try:
self._create_legacy_stock_daily_with_canonical_id(db_path)
with sqlite3.connect(db_path) as conn:
conn.executemany(
"""INSERT INTO stock_daily (code, date, close, canonical_id)
VALUES (?, ?, ?, ?)""",
[
# Mis-bucketed bare code — must be repaired.
("000016", "2026-01-01", 2.0, "sh000016"),
# Unrelated canonical — must NOT be touched.
("600519", "2026-01-02", 7.0, "sh600519"),
("AAPL", "2026-01-03", 8.0, "AAPL"),
# Explicit index row — must NOT be touched.
("sh000016", "2026-01-04", 5.0, "sh000016"),
],
)
DatabaseManager(db_url=f"sqlite:///{db_path}")
with sqlite3.connect(db_path) as conn:
rows = dict(
conn.execute(
"SELECT code, canonical_id FROM stock_daily ORDER BY id"
).fetchall()
)
# Only the mis-bucketed bare code is repaired.
self.assertEqual(rows["000016"], "sz000016")
self.assertEqual(rows["600519"], "sh600519")
self.assertEqual(rows["AAPL"], "AAPL")
self.assertEqual(rows["sh000016"], "sh000016")
finally:
DatabaseManager.reset_instance()
Config.reset_instance()
self._cleanup_temp_dir(db_dir)
if __name__ == '__main__':
unittest.main()
+29
View File
@@ -87,6 +87,35 @@ class TestTaskService(unittest.TestCase):
self.assertIn("args", captured)
self.assertEqual(captured["args"][1], "005930.KS")
def test_submit_analysis_passes_parser_canonical_to_executor(self):
"""Real task-layer regression: TaskService must hand the executor the
parser canonical ``csi930955`` (not the raw alias ``930955.CSI`` or the
old uppercase ``CSI930955``) so the analysis pipeline receives one
consistent identity for the same registered CSI index."""
service = TaskService()
service._tasks = {}
service._tasks_lock = threading.Lock()
captured = {}
executor = MagicMock()
def capture_submit(*args, **kwargs):
captured["args"] = args
return "future"
executor.submit.side_effect = capture_submit
service._executor = executor
# Use the real resolver (not a mock): ``930955.CSI`` is a registered CSI
# explicit identity and must converge to the parser canonical
# ``csi930955``, which is what gets handed to the executor.
result = service.submit_analysis("930955.CSI", report_type="simple", query_source="cli")
self.assertEqual(result["code"], "csi930955")
self.assertIn("args", captured)
# executor.submit(self._run_analysis, code, task_id, ...) — code is arg[1]
self.assertEqual(captured["args"][1], "csi930955")
if __name__ == "__main__":
import unittest