mirror of
https://github.com/shy3130/tick-stock-panel.git
synced 2026-11-02 10:26:13 +00:00
fix(kline): 裸符号按维表补全后缀, 批量同步 0 行标的 fail-loud
上游对无交易所后缀的裸符号返回 200 空 payload, 本地静默写 0 行且流程 显示成功, 新环境初始化极易踩坑 (#302)。 - 请求前按 instruments 维表把 600000 补全为 600000.SH; 维表缺失或 无唯一匹配的裸符号显式跳过并告警, 不再带已知无效符号发请求 - 批量同步收尾统计实际入库 0 行的标的(含被跳过的裸符号), 聚合一条 WARNING 并通过 zero_row_out 出参回报, /sync 与 /sync_batch 响应 附带 zero_row_symbols (按调用方原始写法) - 拉取失败分块与本统计互斥(failed_out 已单独可见), 不重复计入 Closes #302
This commit is contained in:
shy3130
@@ -1099,14 +1099,18 @@ def get_minute(
|
||||
@router.post("/sync")
|
||||
def sync_symbol(
|
||||
request: Request,
|
||||
symbol: str = Query(...),
|
||||
symbol: str,
|
||||
days: int = Query(250, ge=10, le=2000),
|
||||
):
|
||||
"""手动触发单股同步(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")
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user