fix(instruments): 数据源不提供股本时保留既有维表字段

部分行情源的 get_instruments 只返回代码/名称/市场/类型, 没有股本与交易规则字段。维表同步整表覆盖后 float_shares/total_shares 变 null, enriched 的 turnover_rate 随之全空, 依赖换手率与流通市值的策略 (起爆、5连阳等) 基础过滤会筛出 0 只 —— 换数据源即策略结果不同。改为按 symbol 逐行 coalesce: provider 有值用 provider, 为空才回落既有维表值; 新上市标的保持 null 不伪造; provider 自带字段时以其为准。 provider 完全缺字段且无历史维表时打 warning 提示换手率将为空。验证: 新增 5 个用例; 真实维表同步后 5580 只仅 3 只新标的为空。
2026-11-02 10:26:13 +00:00 · 2026-10-20 13:59:07 +08:00
parent 05817654d3
commit eab70c402f
2 changed files with 190 additions and 0 deletions
@@ -95,6 +95,7 @@ def sync_instruments(data_dir: Path) -> int:
        return 0

    df = pl.DataFrame(all_rows)
+    df = _preserve_share_fields(df, data_dir)
    df = df.with_columns(pl.lit(date.today()).alias("as_of"))

    out = data_dir / "instruments" / "instruments.parquet"
@@ -105,6 +106,70 @@ def sync_instruments(data_dir: Path) -> int:
    return df.height


+# 股本与交易规则字段: 部分行情源的标的维表只有代码/名称/市场/类型。
+# 这些字段缺失会让 enriched 的 turnover_rate / 市值 / 涨跌停判定全空, 进而使
+# 依赖换手率与流通市值的策略筛不出任何标的 —— 因此绝不能用 null 覆盖已有值。
+_SHARE_FIELDS = (
+    "listing_date", "total_shares", "float_shares", "tick_size",
+    "limit_up", "limit_down",
+)
+
+
+def _preserve_share_fields(df: pl.DataFrame, data_dir: Path) -> pl.DataFrame:
+    """用现有维表补全 provider 未提供的股本/交易规则字段 (按 symbol 左连接)。
+
+    新增标的仍保留 null (没有历史值可用, 不伪造); 只在同一 symbol 上把
+    provider 的空值替换为既有值, 并在日志里给出补全规模。
+    """
+    available = [c for c in _SHARE_FIELDS if c in df.columns]
+    if not available:
+        return df
+    if all(df[c].null_count() == 0 for c in available):
+        return df  # provider 已提供全部字段, 无需补全
+
+    path = data_dir / "instruments" / "instruments.parquet"
+    if not path.exists():
+        logger.warning(
+            "instruments provider 未提供股本字段且无历史维表可补全: "
+            "enriched 换手率/市值将为 null, 依赖它们的策略会筛不出标的"
+        )
+        return df
+    try:
+        existing = pl.read_parquet(path)
+    except Exception as e:  # noqa: BLE001
+        logger.warning("instruments 历史维表读取失败, 无法补全股本字段: %s", e)
+        return df
+    if "symbol" not in existing.columns:
+        return df
+
+    keep = ["symbol", *[c for c in available if c in existing.columns]]
+    if len(keep) <= 1:
+        return df
+    merged = df.join(existing.select(keep), on="symbol", how="left", suffix="_prev")
+    filled: list[str] = []
+    for col in keep[1:]:
+        prev = f"{col}_prev"
+        if prev not in merged.columns:
+            continue
+        # 逐行 coalesce: provider 有值用 provider, 为空才回落历史值。这样
+        # 「部分标的缺字段」也能补齐, 而不只处理整列为空的源。
+        if merged[col].null_count() and merged[prev].is_not_null().any():
+            merged = merged.with_columns(
+                pl.coalesce(pl.col(col), pl.col(prev)).alias(col)
+            )
+            filled.append(col)
+    merged = merged.drop([f"{c}_prev" for c in keep[1:] if f"{c}_prev" in merged.columns])
+    if filled:
+        restored = int(merged.select(
+            [pl.col(c).is_not_null().sum() for c in filled]
+        ).row(0)[0])
+        logger.info(
+            "instruments provider 缺 %s, 已逐行从历史维表补全 (首列非空 %d/%d)",
+            "/".join(filled), restored, merged.height,
+        )
+    return merged
+
+
 def enrich_names_from_quotes(
    data_dir: Path,
    quotes_data: list[dict],
@@ -0,0 +1,125 @@
+"""标的维表同步: provider 不提供股本时不覆盖已有值。
+
+回归背景: 部分行情源的 get_instruments 只返回代码/名称/市场/类型, 没有
+股本与交易规则字段。旧实现整表覆盖后 float_shares/total_shares 变 null,
+enriched 的 turnover_rate 随之全空, 依赖换手率与流通市值的策略 (起爆/5连阳)
+直接筛出 0 只 —— 同一个策略换数据源结果就不同。
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import polars as pl
+import pytest
+
+from app.services import instrument_sync as isync
+
+
+def _write_existing(data_dir: Path) -> None:
+    out = data_dir / "instruments"
+    out.mkdir(parents=True, exist_ok=True)
+    pl.DataFrame({
+        "symbol": ["600000.SH", "000001.SZ"],
+        "name": ["浦发银行", "平安银行"],
+        "code": ["600000", "000001"],
+        "exchange": ["SH", "SZ"],
+        "region": ["CN", "CN"],
+        "type": ["stock", "stock"],
+        "listing_date": ["1999-11-10", "1991-04-03"],
+        "total_shares": [2.9e10, 1.9e10],
+        "float_shares": [2.9e10, 1.9e10],
+        "tick_size": [0.01, 0.01],
+        "limit_up": [11.0, 12.0],
+        "limit_down": [9.0, 10.0],
+    }).write_parquet(out / "instruments.parquet")
+
+
+def _provider_rows(*symbols: str) -> list[dict]:
+    """模拟只提供标识字段的行情源 (无股本/交易规则)。"""
+    return [
+        {
+            "symbol": s, "name": s, "code": s.split(".")[0],
+            "exchange": s.split(".")[1], "region": "CN", "type": "stock",
+            "listing_date": None, "total_shares": None, "float_shares": None,
+            "tick_size": None, "limit_up": None, "limit_down": None,
+        }
+        for s in symbols
+    ]
+
+
+def test_share_fields_restored_from_existing_table(tmp_path, monkeypatch) -> None:
+    _write_existing(tmp_path)
+    monkeypatch.setattr(
+        isync, "_fetch_instruments_via_provider",
+        lambda: _provider_rows("600000.SH", "000001.SZ"),
+    )
+
+    assert isync.sync_instruments(tmp_path) == 2
+
+    df = pl.read_parquet(tmp_path / "instruments" / "instruments.parquet")
+    by_symbol = {r["symbol"]: r for r in df.to_dicts()}
+    assert by_symbol["600000.SH"]["float_shares"] == pytest.approx(2.9e10)
+    assert by_symbol["600000.SH"]["total_shares"] == pytest.approx(2.9e10)
+    assert by_symbol["600000.SH"]["tick_size"] == pytest.approx(0.01)
+    assert by_symbol["000001.SZ"]["float_shares"] == pytest.approx(1.9e10)
+
+
+def test_new_symbol_keeps_null_shares_instead_of_faking(tmp_path, monkeypatch) -> None:
+    """新上市标的没有历史值 → 保持 null, 不伪造股本。"""
+    _write_existing(tmp_path)
+    monkeypatch.setattr(
+        isync, "_fetch_instruments_via_provider",
+        lambda: _provider_rows("600000.SH", "301999.SZ"),
+    )
+
+    isync.sync_instruments(tmp_path)
+
+    df = pl.read_parquet(tmp_path / "instruments" / "instruments.parquet")
+    new = df.filter(pl.col("symbol") == "301999.SZ")
+    assert new.height == 1
+    assert new["float_shares"][0] is None
+    assert new["total_shares"][0] is None
+
+
+def test_provider_fields_win_when_present(tmp_path, monkeypatch) -> None:
+    """provider 自带股本时以 provider 为准 (不复用旧值)。"""
+    _write_existing(tmp_path)
+    rows = _provider_rows("600000.SH")
+    rows[0]["float_shares"] = 1.0e9
+    rows[0]["total_shares"] = 2.0e9
+    monkeypatch.setattr(isync, "_fetch_instruments_via_provider", lambda: rows)
+
+    isync.sync_instruments(tmp_path)
+
+    df = pl.read_parquet(tmp_path / "instruments" / "instruments.parquet")
+    assert df["float_shares"][0] == pytest.approx(1.0e9)
+    assert df["total_shares"][0] == pytest.approx(2.0e9)
+
+
+def test_no_existing_table_is_tolerated(tmp_path, monkeypatch) -> None:
+    """首次同步且无历史维表: 不报错, 字段保持 null。"""
+    monkeypatch.setattr(
+        isync, "_fetch_instruments_via_provider",
+        lambda: _provider_rows("600000.SH"),
+    )
+
+    assert isync.sync_instruments(tmp_path) == 1
+
+    df = pl.read_parquet(tmp_path / "instruments" / "instruments.parquet")
+    assert df["float_shares"][0] is None
+
+
+def test_partial_provider_fields_filled_per_row(tmp_path, monkeypatch) -> None:
+    """provider 只给部分标的股本时, 也要逐行补全 (不能整列跳过)。"""
+    _write_existing(tmp_path)
+    rows = _provider_rows("600000.SH", "000001.SZ")
+    rows[0]["float_shares"] = 5.0e8  # 第一只有值, 第二只仍为空
+    rows[0]["total_shares"] = 6.0e8
+    monkeypatch.setattr(isync, "_fetch_instruments_via_provider", lambda: rows)
+
+    isync.sync_instruments(tmp_path)
+
+    df = pl.read_parquet(tmp_path / "instruments" / "instruments.parquet")
+    by_symbol = {r["symbol"]: r for r in df.to_dicts()}
+    assert by_symbol["600000.SH"]["float_shares"] == pytest.approx(5.0e8)
+    assert by_symbol["000001.SZ"]["float_shares"] == pytest.approx(1.9e10)