Merge pull request #311 from shy3130/fix/kline-bare-symbol-fail-loud

fix(kline): 裸符号按维表补全后缀, 批量同步 0 行标的 fail-loud

backend/app/api/kline.py

@@ -1105,8 +1105,12 @@ def sync_symbol(
     """手动触发单股同步(Free 用户在 K 线页用)。"""
     repo = request.app.state.repo
     capset = request.app.state.capabilities
-    n = kline_sync.sync_and_persist_daily_batch([symbol], repo, capset, count=days)
-    return {"symbol": symbol, "rows_written": n}
+    zero: list[str] = []
+    n = kline_sync.sync_and_persist_daily_batch([symbol], repo, capset, count=days, zero_row_out=zero)
+    resp = {"symbol": symbol, "rows_written": n}
+    if zero:
+        resp["zero_row_symbols"] = zero
+    return resp
 
 
 @router.post("/sync_batch")
@@ -1117,8 +1121,14 @@ def sync_batch(
 ):
     repo = request.app.state.repo
     capset = request.app.state.capabilities
-    n = kline_sync.sync_and_persist_daily_batch(symbols, repo, capset, count=days)
-    return {"symbols": symbols, "rows_written": n}
+    zero: list[str] = []
+    n = kline_sync.sync_and_persist_daily_batch(symbols, repo, capset, count=days, zero_row_out=zero)
+    resp = {"symbols": symbols, "rows_written": n}
+    if zero:
+        # fail-loud (#302): 裸符号被跳过/上游 200 空数据的标的显式列出,
+        # 不再"回填显示成功、实际全库 0 行"
+        resp["zero_row_symbols"] = zero
+    return resp
 
 
 @router.post("/refresh_views")

backend/app/services/kline_sync.py

@@ -92,6 +92,73 @@ def _normalize_daily(df_in, default_symbol: str | None = None) -> pl.DataFrame:
     return df.select(keep)
 
 
+def _instruments_symbol_index(data_dir) -> dict[str, list[str]] | None:
+    """裸代码 → instruments 维表完整符号列表; 维表缺失/不可读返回 None。"""
+    path = data_dir / "instruments" / "instruments.parquet"
+    if not path.exists():
+        return None
+    try:
+        symbols = pl.read_parquet(path, columns=["symbol"])["symbol"].drop_nulls().to_list()
+    except Exception as e:
+        logger.warning("instruments 维表读取失败, 裸符号无法补全后缀: %s", e)
+        return None
+    index: dict[str, list[str]] = {}
+    for s in symbols:
+        code, _, _suffix = str(s).partition(".")
+        if code:
+            index.setdefault(code, []).append(str(s))
+    return index
+
+
+def _normalize_bare_symbols(
+    symbols: list[str], data_dir,
+) -> tuple[list[str], dict[str, str], list[str]]:
+    """裸符号(无交易所后缀, 如 600000)按 instruments 维表补全为 600000.SH (#302)。
+
+    返回 (规范化后的去重请求列表, {裸符号: 补全后符号}, 被跳过的裸符号)。
+    上游对裸符号返回 200 空 payload, 本地静默写 0 行 — 补全失败时显式跳过并告警,
+    不再带着已知无效的符号发请求。全部带后缀时不读维表, 行为与开销同旧版;
+    带后缀符号原样透传, 由上游照常处理。
+    """
+    bare = [s for s in symbols if "." not in s]
+    if not bare:
+        return symbols, {}, []
+
+    index = _instruments_symbol_index(data_dir)
+    resolved: dict[str, str] = {}
+    skipped: list[str] = []
+    for s in bare:
+        matches = (index or {}).get(s, [])
+        if len(matches) == 1:
+            resolved[s] = matches[0]
+        else:
+            skipped.append(s)
+
+    if index is None:
+        logger.warning(
+            "日K同步: instruments 维表不可用, %d 个裸符号(无交易所后缀)将被跳过 — "
+            "请先同步标的维表, 或改用带后缀符号 (如 600000.SH)", len(skipped))
+    elif skipped:
+        logger.warning(
+            "日K同步: %d 个裸符号在维表中无唯一匹配(不支持或非股票), 已跳过: %s",
+            len(skipped), skipped[:20])
+    if resolved:
+        logger.info("日K同步: 已按维表补全 %d 个裸符号后缀 (样例: %s)",
+                    len(resolved), list(resolved.items())[:5])
+
+    out: list[str] = []
+    seen: set[str] = set()
+    skipped_set = set(skipped)
+    for s in symbols:
+        if s in skipped_set:
+            continue  # 无法补全的裸符号不发请求 (上游必然 200 空 payload)
+        full = resolved.get(s, s)
+        if full not in seen:
+            seen.add(full)
+            out.append(full)
+    return out, resolved, skipped
+
+
 def sync_daily_batch(symbols: list[str],
                      count: int | None = None,
                      batch_size: int | None = None,
@@ -164,14 +231,41 @@ def sync_and_persist_daily_batch(
     start_date: datetime | None = None,
     end_date: datetime | None = None,
     on_chunk_done: Callable[[int, int], None] | None = None,
+    zero_row_out: list[str] | None = None,
 ) -> int:
     """批量同步日 K 并落到 Parquet。返回写入的行数。
 
     start_date/end_date: 外部传入的时间范围(由 pipeline 根据已有数据计算)。
     未传入时默认拉最近 1 年。
+    zero_row_out: 可选出参。本轮实际入库 0 行的标的(含无法识别被跳过的裸符号、
+                  上游 200 空 payload 的静默 0 行)按调用方原始写法追加进该 list,
+                  供上层 fail-loud 展示, 不再"显示成功实则全空" (#302)。
     """
+    seen: set[str] = set()
+    failed_syms: list[str] = []
+    skipped_bare: list[str] = []
+
+    def _finalize(written: int) -> int:
+        zero_full = [s for s in requested if s not in seen and s not in failed_syms]
+        if zero_full or skipped_bare:
+            zero = [orig_by_full.get(s, s) for s in zero_full] + skipped_bare
+            logger.warning(
+                "日K批量同步: %d/%d 标的本轮入库 0 行 (符号无法识别/停牌/窗口内无数据; 样例: %s)",
+                len(zero), len(requested) + len(skipped_bare), zero[:10])
+            if zero_row_out is not None:
+                zero_row_out.extend(zero)
+        return written
+
+    original = list(symbols)
+    symbols, resolved, skipped_bare = _normalize_bare_symbols(symbols, repo.store.data_dir)
+    requested = symbols
+    # 完整符号 → 调用方原始写法: 0 行回报用用户输入的形式, 裸符号也能对上
+    orig_by_full: dict[str, str] = {}
+    for orig in original:
+        orig_by_full.setdefault(resolved.get(orig, orig), orig)
+
     if not symbols:
-        return 0
+        return _finalize(0)
 
     provider_name = preferences.get_daily_data_provider()
     if provider_name != "tickflow":
@@ -183,15 +277,22 @@ def sync_and_persist_daily_batch(
             start_time = start_date or (end_time - timedelta(days=days))
             iter_daily = getattr(provider, "iter_daily", None)
             if callable(iter_daily):
-                return _persist_daily_chunks(
-                    iter_daily(
+
+                def _tracking_chunks(chunks):
+                    for df in chunks:
+                        if not df.is_empty() and "symbol" in df.columns:
+                            seen.update(df["symbol"].cast(pl.Utf8).unique().to_list())
+                        yield df
+
+                return _finalize(_persist_daily_chunks(
+                    _tracking_chunks(iter_daily(
                         symbols,
                         start_time=start_time,
                         end_time=end_time,
                         on_chunk_done=on_chunk_done,
-                    ),
+                    )),
                     repo,
-                )
+                ))
             df = provider.get_daily(
                 symbols,
                 start_time=start_time,
@@ -199,7 +300,9 @@ def sync_and_persist_daily_batch(
                 on_chunk_done=on_chunk_done,
             )
             if df.is_empty():
-                return 0
+                return _finalize(0)
+            if "symbol" in df.columns:
+                seen.update(df["symbol"].cast(pl.Utf8).unique().to_list())
             repo.append_daily(df)
             try:
                 d = repo.store.data_dir.as_posix()
@@ -209,11 +312,11 @@ def sync_and_persist_daily_batch(
                 )
             except Exception as e:  # noqa: BLE001
                 logger.warning("refresh view failed: %s", e)
-            return df.height
+            return _finalize(df.height)
         # 自定义源未配置 daily → 回退 TickFlow
 
     if not capset.has(Cap.KLINE_DAILY_BATCH):
-        return 0
+        return _finalize(0)
 
     limit = resolve_limit(capset, Cap.KLINE_DAILY_BATCH, default_batch=100)
 
@@ -224,10 +327,14 @@ def sync_and_persist_daily_batch(
         symbols, count=count, batch_size=limit.batch, rpm=limit.rpm,
         start_time=start_time, end_time=end_time,
         on_chunk_done=on_chunk_done,
+        failed_out=failed_syms,
     )
 
+    if not df.is_empty() and "symbol" in df.columns:
+        seen.update(df["symbol"].cast(pl.Utf8).unique().to_list())
+
     if df.is_empty():
-        return 0
+        return _finalize(0)
 
     repo.append_daily(df)
 
@@ -240,7 +347,7 @@ def sync_and_persist_daily_batch(
     except Exception as e:  # noqa: BLE001
         logger.warning("refresh view failed: %s", e)
 
-    return df.height
+    return _finalize(df.height)
 
 
 def _persist_daily_chunks(chunks, repo: KlineRepository) -> int:

backend/tests/test_kline_sync_bare_symbols.py

@@ -0,0 +1,150 @@
+"""#302: 裸符号日K同步不再静默 0 行。
+
+上游对无交易所后缀的符号返回 200 空 payload, 本地写 0 行且流程"成功"。
+修复后: 裸符号按 instruments 维表补全后缀, 无法识别的显式跳过并告警;
+批量同步结束统计 0 行标的 (zero_row_out / API zero_row_symbols)。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime
+from types import SimpleNamespace
+
+import polars as pl
+
+from app.services import kline_sync
+from app.tickflow.repository import DataStore, KlineRepository
+
+
+def _write_instruments(data_dir, symbols: list[str]) -> None:
+    out = data_dir / "instruments" / "instruments.parquet"
+    out.parent.mkdir(parents=True, exist_ok=True)
+    pl.DataFrame({"symbol": symbols}).write_parquet(out)
+
+
+def _daily(symbol: str, day: date) -> pl.DataFrame:
+    return pl.DataFrame({
+        "symbol": [symbol], "date": [day], "open": [10.0], "high": [11.0],
+        "low": [9.0], "close": [10.5], "volume": [100.0], "amount": [1050.0],
+    })
+
+
+class _Provider:
+    """自定义源: 只返回 000001.SZ 的数据 (600000 上游 200 空 payload 场景)。"""
+
+    def __init__(self, chunks):
+        self._chunks = chunks
+        self.requested_symbols: list[str] | None = None
+
+    def iter_daily(self, symbols, *args, **kwargs):
+        self.requested_symbols = list(symbols)
+        yield from self._chunks
+
+
+def _route_provider(monkeypatch, provider):
+    monkeypatch.setattr(kline_sync.preferences, "get_daily_data_provider", lambda: "fuyao")
+    from app.data_providers import custom
+    monkeypatch.setattr(
+        custom, "provider_has_dataset", lambda name, dataset: name == "fuyao" and dataset == "daily"
+    )
+    monkeypatch.setattr(custom, "get_provider", lambda name: provider)
+
+
+def test_normalize_maps_bare_via_instruments_and_dedups(tmp_path):
+    _write_instruments(tmp_path, ["600000.SH", "000001.SZ"])
+
+    out, resolved, skipped = kline_sync._normalize_bare_symbols(
+        ["600000", "000001.SZ", "600000.SH"], tmp_path
+    )
+
+    assert out == ["600000.SH", "000001.SZ"]
+    assert resolved == {"600000": "600000.SH"}
+    assert skipped == []
+
+
+def test_normalize_without_instruments_skips_all_bare(tmp_path, caplog):
+    out, resolved, skipped = kline_sync._normalize_bare_symbols(["600000", "000001.SZ"], tmp_path)
+
+    assert out == ["000001.SZ"]
+    assert resolved == {}
+    assert skipped == ["600000"]
+    assert any("维表不可用" in r.message for r in caplog.records)
+
+
+def test_normalize_skips_ambiguous_bare(tmp_path):
+    _write_instruments(tmp_path, ["830001.BJ", "830001.SH", "600000.SH"])
+
+    out, resolved, skipped = kline_sync._normalize_bare_symbols(["830001", "600000"], tmp_path)
+
+    assert out == ["600000.SH"]
+    assert resolved == {"600000": "600000.SH"}
+    assert skipped == ["830001"]
+
+
+def test_sync_reports_zero_rows_in_caller_spelling(monkeypatch, tmp_path, caplog):
+    """裸符号无维表可补 → 跳过; 带后缀但上游空 payload → 0 行; 均按原始写法回报。"""
+    caplog.set_level(logging.WARNING, logger="app.services.kline_sync")
+    repo = KlineRepository(DataStore(tmp_path))
+    provider = _Provider([_daily("000001.SZ", date(2026, 9, 10))])
+    _route_provider(monkeypatch, provider)
+    zero: list[str] = []
+
+    written = kline_sync.sync_and_persist_daily_batch(
+        ["600000", "000001.SZ", "999999.SZ"], repo, object(),
+        start_date=datetime(2026, 9, 1), end_date=datetime(2026, 9, 10),
+        zero_row_out=zero,
+    )
+
+    assert written == 1
+    assert zero == ["999999.SZ", "600000"]  # 0 行标的 + 被跳过的裸符号
+    assert any("入库 0 行" in r.message for r in caplog.records)
+
+
+def test_sync_tickflow_path_normalizes_and_reports_zero(monkeypatch, tmp_path):
+    _write_instruments(tmp_path, ["600000.SH"])
+    repo = KlineRepository(DataStore(tmp_path))
+    captured: dict = {}
+
+    def _fake_sync_daily_batch(symbols, **kwargs):
+        captured["symbols"] = list(symbols)
+        return pl.DataFrame()  # 上游 200 空 payload → 全部 0 行
+
+    monkeypatch.setattr(kline_sync, "sync_daily_batch", _fake_sync_daily_batch)
+    monkeypatch.setattr(
+        kline_sync.preferences, "get_daily_data_provider", lambda: "tickflow"
+    )
+    capset = SimpleNamespace(has=lambda cap: True, limits=lambda cap: None)
+    zero: list[str] = []
+
+    written = kline_sync.sync_and_persist_daily_batch(
+        ["600000", "000001.SH"], repo, capset,
+        start_date=datetime(2026, 9, 1), end_date=datetime(2026, 9, 10),
+        zero_row_out=zero,
+    )
+
+    assert written == 0
+    # 发往上游前已完成规范化: 600000 → 600000.SH
+    assert captured["symbols"] == ["600000.SH", "000001.SH"]
+    # 回报用调用方原始写法
+    assert zero == ["600000", "000001.SH"]
+
+
+def test_api_sync_batch_returns_zero_row_symbols(monkeypatch, tmp_path):
+    from app.api import kline as kline_api
+
+    _write_instruments(tmp_path, ["600000.SH"])
+    repo = KlineRepository(DataStore(tmp_path))
+    monkeypatch.setattr(
+        kline_sync, "sync_daily_batch", lambda symbols, **kwargs: pl.DataFrame()
+    )
+    monkeypatch.setattr(
+        kline_sync.preferences, "get_daily_data_provider", lambda: "tickflow"
+    )
+    request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(
+        repo=repo, capabilities=SimpleNamespace(has=lambda cap: True, limits=lambda cap: None),
+    )))
+
+    resp = kline_api.sync_batch(request, symbols=["600000", "000001.SH"], days=250)
+
+    assert resp["rows_written"] == 0
+    assert resp["zero_row_symbols"] == ["600000", "000001.SH"]