diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index 85dd9558..c15c2d64 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -301,6 +301,7 @@ def sync_and_persist_daily_batch( end_date: datetime | None = None, on_chunk_done: Callable[[int, int], None] | None = None, zero_row_out: list[str] | None = None, + failed_out: list[str] | None = None, ) -> int: """批量同步日 K 并落到 Parquet。返回写入的行数。 @@ -309,12 +310,17 @@ def sync_and_persist_daily_batch( zero_row_out: 可选出参。本轮实际入库 0 行的标的(含无法识别被跳过的裸符号、 上游 200 空 payload 的静默 0 行)按调用方原始写法追加进该 list, 供上层 fail-loud 展示, 不再"显示成功实则全空" (#302)。 + failed_out: 可选出参。分块拉取失败(超时/异常)的标的, 供上层把单票失败 + 上浮为 stage_errors 而非静默当成功 (#514)。 """ seen: set[str] = set() failed_syms: list[str] = [] skipped_bare: list[str] = [] def _finalize(written: int) -> int: + if failed_out is not None and failed_syms: + # 单票失败上浮给调用方 (daily_pipeline 汇入 stage_errors, #514) + failed_out.extend(failed_syms) 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 diff --git a/backend/tests/test_kline_sync_failed_out.py b/backend/tests/test_kline_sync_failed_out.py new file mode 100644 index 00000000..5fc0cc5f --- /dev/null +++ b/backend/tests/test_kline_sync_failed_out.py @@ -0,0 +1,47 @@ +"""回归: sync_and_persist_daily_batch 必须接受 failed_out 并透出失败标的。 + +daily_pipeline (#514) 的修复/定向分支传 failed_out=repair_failed_symbols, +但本函数签名漏接该参数 → TypeError, 数据修正/boot 自检修复路径必炸 +(v0.3.21/v0.3.22 用户同步失败截图: "got an unexpected keyword argument +'failed_out'")。 +""" +from __future__ import annotations + +from types import SimpleNamespace + +import polars as pl +import pytest + +from app.services import kline_sync +from app.tickflow.repository import DataStore, KlineRepository + + +@pytest.fixture +def repo(tmp_path): + return KlineRepository(DataStore(tmp_path)) + + +def test_accepts_failed_out_and_surfaces_failed_symbols(monkeypatch, repo): + def fake_sync_daily_batch(symbols, **kwargs): + # 模拟分块拉取失败: 内部经由 failed_out 出参上报失败标的 + kwargs["failed_out"].extend(["600519.SH", "000001.SZ"]) + return pl.DataFrame() + + monkeypatch.setattr(kline_sync, "sync_daily_batch", fake_sync_daily_batch) + capset = SimpleNamespace(has=lambda cap: True, limits=lambda cap: None) + failed: list[str] = [] + + written = kline_sync.sync_and_persist_daily_batch( + ["600519.SH", "000001.SZ"], repo, capset, failed_out=failed) + + assert written == 0 + assert failed == ["600519.SH", "000001.SZ"] + + +def test_failed_out_omitted_still_works(monkeypatch, repo): + monkeypatch.setattr( + kline_sync, "sync_daily_batch", lambda symbols, **kw: pl.DataFrame()) + capset = SimpleNamespace(has=lambda cap: True, limits=lambda cap: None) + + assert kline_sync.sync_and_persist_daily_batch( + ["600519.SH"], repo, capset) == 0