feat(screener): 数据不足时选股显式提示, 盘后管道薄库 WARN

空库首跑/仅当日实时覆写 1 天的库, 指标暖机不足, 选股静默全 0 且
无任何信号 (#303)。

- enriched_history_days: 按 date=* 分区目录计数本地覆盖交易日数
  (目录列举 O(天数), 不读 parquet 内容, 不进指标计算路径)
- ScreenerService.coverage_warnings: 覆盖低于暖机需求时生成人话警告;
  策略运行用 engine.required_history_bars 精确值, 自定义 SQL 用通用
  30 天窗口
- /run /run_preset /run_all 响应附带 warnings 字段 (1d 周期),
  run_preset/run_all 同时写入盘后缓存随 get_cached 读出
- 盘后管道收尾检测 enriched 覆盖过薄 (<30 天) 时打 WARN 引导全量回填,
  result 新增 enriched_total_days

Closes #303
This commit is contained in:
shy3130 committed 2026-09-14 14:02:18 +08:00
1 parent 6063fb8eb1
commit cf915c82c4
4 files changed
+291 -2

No files matched your search

+49 -2
View File
@@ -55,6 +55,28 @@ def _safe(result_dict: dict) -> dict:
return result_dict
def _coverage_warnings(
svc, as_of, *, engine=None, strategy_ids=None, params_map=None, overrides_map=None,
) -> list[str]:
"""数据充足性提示 (#303): enriched 覆盖低于暖机需求时给出人话警告。
advisory 元数据: 任何计算失败都静默返回 [], 绝不影响选股主流程。
引擎/服务无该方法时 (测试 Fake) 同样跳过, 与 build_shared_matrix 的
可选能力探测同风格。
"""
try:
required: int | None = None
rhb = getattr(engine, "required_history_bars", None)
if engine is not None and strategy_ids and callable(rhb):
required = rhb(strategy_ids, params_map=params_map, overrides_map=overrides_map)
cw = getattr(svc, "coverage_warnings", None)
if callable(cw):
return cw(as_of, required_bars=required)
except Exception:
logger.debug("coverage warnings unavailable", exc_info=True)
return []
def _one_word_limit_expr(status_main: str, columns: list[str]) -> Any:
required = {"open", "high", "low", "close", "status"}
if not required.issubset(columns):
@@ -273,6 +295,9 @@ def run_custom(req: CustomRequest, request: Request):
pool=req.pool,
)
safe_data = _safe(asdict(result))
warnings = _coverage_warnings(svc, as_of)
if warnings:
safe_data["warnings"] = warnings
ext_values = _load_ext_value_maps(repo, req.ext_columns)
return _result_with_ext(safe_data, ext_values)
@@ -319,9 +344,17 @@ def run_preset(req: PresetRequest, request: Request):
raise HTTPException(status_code=status_code, detail=str(e)) from e
safe_data = _safe(asdict(result))
# 分钟周期结果不写入盘后缓存 (strategy_cache 是日线语义, as_of/updated_at
# 混入分钟结果会污染页面秒加载路径)。
if req.timeframe == "1d":
# 数据不足提示随结果返回并写入盘后缓存 (#303); 分钟周期结果不写入盘后缓存
# (strategy_cache 是日线语义, as_of/updated_at 混入分钟结果会污染页面秒加载路径),
# 分钟策略数据源也与日线 enriched 历史无关, 不提示。
warnings = _coverage_warnings(
svc, as_of, engine=engine, strategy_ids=[req.strategy_id],
params_map={req.strategy_id: params},
overrides_map={req.strategy_id: overrides or {}},
)
if warnings:
safe_data["warnings"] = warnings
_update_cache_strategy(data_dir, str(as_of), req.strategy_id, safe_data)
return _result_with_ext(safe_data, ext_values)
@@ -575,6 +608,13 @@ def _run_all_progressive(
"rows": _safe(asdict(result)).get("rows", []),
"computed_at": int(time.time() * 1000),
}
if timeframe == "1d":
w = _coverage_warnings(
svc, as_of, engine=engine, strategy_ids=[sid],
params_map=params_map, overrides_map=overrides_map,
)
if w:
payload["warnings"] = w
all_results[sid] = payload
elapsed_map[sid] = (time.perf_counter() - t0) * 1000
# 逐策略增量落盘 (write_cache 同日按 sid 合并), 前端轮询即可逐个看到
@@ -729,6 +769,13 @@ def run_all(request: Request, body: Optional[dict] = None):
"as_of": str(as_of),
"rows": safe_rows,
}
if timeframe == "1d":
w = _coverage_warnings(
svc, as_of, engine=engine, strategy_ids=[sid],
params_map=params_map, overrides_map=overrides_map,
)
if w:
results[sid]["warnings"] = w
elapsed = (time.perf_counter() - t_total) * 1000
logger.info("run_all: total took %.1fms (%d strategies)", elapsed, len(all_ids))
+22
View File
@@ -511,6 +511,10 @@ def run_now(
_refresh_single_view(repo, "kline_enriched")
_invalidate("enriched")
# Step 2.1: 数据充足性可见化 (#303) — 空库首跑/仅当日实时覆写 1 天的库,
# 均线/动量/量比等指标暖机不足, 管道各 stage 都"成功"但选股会静默全 0。
enriched_total_days = warn_if_enriched_too_thin(repo.store.data_dir)
# Step 2.3: 指数 / ETF 同步 — 物理分开存储;ETF 可复权,指数不复权。
written_index_daily = 0
written_etf_daily = 0
@@ -746,6 +750,7 @@ def run_now(
"regime_days": regime_days,
"mainline_rows": mainline_rows,
"lagging_symbols": len(lagging_symbols),
"enriched_total_days": enriched_total_days,
"integrity_repair_from": repair_start.isoformat() if repair_start else None,
"integrity_issues": len(integrity_issues),
"skipped_stages": skipped,
@@ -760,6 +765,23 @@ def run_now(
return result
def warn_if_enriched_too_thin(data_dir: Path) -> int:
"""enriched 总覆盖天数; 低于常见指标暖机窗口时 WARN 引导全量回填 (#303)。
返回天数供管道 result 上报。目录列举 O(天数), 不在热路径。
"""
from app.services.screener import MIN_INDICATOR_WARMUP_DAYS, enriched_history_days
days = enriched_history_days(data_dir)
if days < MIN_INDICATOR_WARMUP_DAYS:
logger.warning(
"enriched 仅覆盖 %d 个交易日 (<%d): 指标暖机不足, 选股可能全部 0 命中且无提示 — "
"建议全量回填 (同步标的维表 → 日K批量同步(带后缀符号) → 重算 enriched)",
days, MIN_INDICATOR_WARMUP_DAYS,
)
return days
def _refresh_views(repo: KlineRepository) -> None:
"""刷新所有 DuckDB 视图 —— 委托给 repository 的唯一权威实现 rebuild_views()。"""
repo.rebuild_views()
+45
View File
@@ -11,6 +11,7 @@ import logging
import time
from dataclasses import dataclass, field
from datetime import date, timedelta
from pathlib import Path
import polars as pl
@@ -19,6 +20,30 @@ from app.tickflow.repository import KlineRepository
logger = logging.getLogger(__name__)
# 常用指标 (MA5/10/20、BOLL(20)、量比等) 的近似最小暖机窗口。
# 用于没有精确回看需求的场景 (自定义 SQL 选股、盘后管道) 的数据充足性提示 (#303);
# 策略运行用 engine.required_history_bars 的精确值。
MIN_INDICATOR_WARMUP_DAYS = 30
def enriched_history_days(data_dir, asset_type: str = "stock", as_of: date | None = None) -> int:
"""本地 enriched 在 as_of 及之前覆盖的交易日数 (#303)。
按 date=* 分区目录名计数 (目录列举 O(天数)), 不读 parquet 内容 —
只做数据充足性提示, 不进入指标计算路径。
"""
from app.tickflow.repository import enriched_dirname
root = Path(data_dir) / enriched_dirname(asset_type)
if not root.exists():
return 0
days = [d.name[5:] for d in root.glob("date=*") if d.is_dir() and len(d.name) > 5]
if as_of is not None:
as_of_s = as_of.isoformat()
days = [d for d in days if d <= as_of_s]
return len(days)
# ── 进程级历史数据缓存 (避免 run_all 每次重新扫描 parquet + 计算指标) ──
_history_cache: dict[tuple[str, date, int], tuple[float, pl.DataFrame]] = {}
_HISTORY_CACHE_TTL = 120.0 # 秒
@@ -503,3 +528,23 @@ class ScreenerService:
except Exception: # noqa: BLE001
return None
return None
def coverage_warnings(self, as_of: date, *, required_bars: int | None = None) -> list[str]:
"""数据充足性提示 (#303): enriched 覆盖不足时返回用户可读警告, 充足返回 []。
空库首跑只拉到 1 个交易日时, 均线/动量/量比等指标暖机不足, 选股会静默
全 0 — 这里把"数据不够"显式说出来。required_bars 缺省用通用暖机窗口
(自定义 SQL 选股); 策略运行传 engine.required_history_bars 的精确值。
available 为 0 时 enriched 为空, 上层 latest_date 已 400, 不重复提示。
"""
available = enriched_history_days(self.repo.store.data_dir, self.asset_type, as_of)
if available == 0:
return []
need = required_bars if required_bars and required_bars > 0 else MIN_INDICATOR_WARMUP_DAYS
if available >= need:
return []
return [
f"本地数据仅覆盖 {available} 个交易日, 低于本次计算所需约 {need} 天暖机窗口 — "
"指标可能失真或全部落空 (选股 0 命中)。建议先全量回填日K并重算指标 "
"(数据页「日K批量同步」, 符号需带交易所后缀)"
]
@@ -0,0 +1,175 @@
"""#303: 空库/薄库数据不足时的选股可见化。
增量管道只拉到 1 个交易日时, 指标暖机不足, 选股静默全 0。修复后:
- enriched_history_days 按 date=* 分区目录计数 (不读 parquet 内容);
- ScreenerService.coverage_warnings 生成人话警告, run/run_preset/run_all
响应附带 warnings 字段;
- 盘后管道在 enriched 覆盖过薄时打 WARN (warn_if_enriched_too_thin)。
"""
from __future__ import annotations
import logging
from datetime import date
from types import SimpleNamespace
from app.api import screener as screener_api
from app.jobs import daily_pipeline
from app.services.screener import (
MIN_INDICATOR_WARMUP_DAYS,
ScreenerService,
enriched_history_days,
)
def _make_partitions(root, dirname: str, dates: list[str]) -> None:
for ds in dates:
(root / dirname / f"date={ds}").mkdir(parents=True, exist_ok=True)
def test_enriched_history_days_counts_and_filters_as_of(tmp_path):
_make_partitions(tmp_path, "kline_daily_enriched",
["2026-09-01", "2026-09-02", "2026-09-03"])
assert enriched_history_days(tmp_path) == 3
assert enriched_history_days(tmp_path, as_of=date(2026, 9, 2)) == 2
assert enriched_history_days(tmp_path / "nonexistent") == 0
def test_enriched_history_days_routes_etf_directory(tmp_path):
_make_partitions(tmp_path, "kline_etf_enriched", ["2026-09-01"])
assert enriched_history_days(tmp_path, asset_type="etf") == 1
assert enriched_history_days(tmp_path, asset_type="stock") == 0
def _svc(tmp_path) -> ScreenerService:
return ScreenerService(SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)))
def test_coverage_warnings_fire_on_thin_history(tmp_path):
_make_partitions(tmp_path, "kline_daily_enriched", ["2026-09-01"])
warnings = _svc(tmp_path).coverage_warnings(date(2026, 9, 1))
assert len(warnings) == 1
assert "1 个交易日" in warnings[0]
def test_coverage_warnings_respect_strategy_required_bars(tmp_path):
# 36 天: 通用窗口 (30) 足够, 策略精确需求 60 天不够
_make_partitions(
tmp_path, "kline_daily_enriched",
[f"2026-{m:02d}-{d:02d}" for m in range(1, 10) for d in (1, 15)]
+ [f"2026-10-{d:02d}" for d in range(1, 19)],
)
assert _svc(tmp_path).coverage_warnings(date(2026, 10, 18)) == []
thin = _svc(tmp_path).coverage_warnings(date(2026, 10, 18), required_bars=60)
assert len(thin) == 1 and "60" in thin[0]
def test_coverage_warnings_silent_when_sufficient(tmp_path):
# 18 天 (1~9 月每月两日) + 12 天 (10 月) = 30, 恰好达标
_make_partitions(
tmp_path, "kline_daily_enriched",
[f"2026-{m:02d}-{d:02d}" for m in range(1, 10) for d in (1, 15)]
+ [f"2026-10-{d:02d}" for d in range(1, 13)],
)
assert _svc(tmp_path).coverage_warnings(date(2026, 10, 12)) == []
class _CoverageSvc:
"""run_preset 接线测试用: 只实现 coverage_warnings/latest_date/build_strategy_context。"""
def __init__(self, warnings):
self._warnings = warnings
def latest_date(self):
return date(2026, 9, 10)
def coverage_warnings(self, as_of, *, required_bars=None):
return self._warnings
def build_strategy_context(self, engine, as_of, strategy_ids, **kwargs):
return SimpleNamespace(as_of=as_of)
class _MinimalEngine:
def __init__(self):
self.run_called = False
def has(self, strategy_id):
return strategy_id == "s1"
def get(self, strategy_id):
return SimpleNamespace(meta={"id": strategy_id})
def run(self, strategy_id, context, **kwargs):
self.run_called = True
from app.services.screener import ScreenerResult
return ScreenerResult(as_of=context.as_of, strategy=strategy_id)
def test_run_preset_response_carries_warnings(monkeypatch, tmp_path):
svc = _CoverageSvc(["数据不足提示"])
engine = _MinimalEngine()
request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(
repo=SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)),
strategy_engine=engine,
)))
monkeypatch.setattr(screener_api, "ScreenerService", lambda *a, **k: svc)
monkeypatch.setattr(screener_api, "_load_ext_value_maps", lambda *a, **k: {})
monkeypatch.setattr(screener_api, "_update_cache_strategy", lambda *a, **k: None)
monkeypatch.setattr(screener_api.strategy_config, "load_override", lambda *a: {})
resp = screener_api.run_preset(
screener_api.PresetRequest(strategy_id="s1", as_of=date(2026, 9, 10)),
request,
)
assert engine.run_called
assert resp["warnings"] == ["数据不足提示"]
def test_run_preset_no_warnings_key_when_sufficient(monkeypatch, tmp_path):
svc = _CoverageSvc([])
engine = _MinimalEngine()
request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(
repo=SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)),
strategy_engine=engine,
)))
monkeypatch.setattr(screener_api, "ScreenerService", lambda *a, **k: svc)
monkeypatch.setattr(screener_api, "_load_ext_value_maps", lambda *a, **k: {})
monkeypatch.setattr(screener_api, "_update_cache_strategy", lambda *a, **k: None)
monkeypatch.setattr(screener_api.strategy_config, "load_override", lambda *a: {})
resp = screener_api.run_preset(
screener_api.PresetRequest(strategy_id="s1", as_of=date(2026, 9, 10)),
request,
)
assert "warnings" not in resp
def test_pipeline_warns_when_enriched_too_thin(tmp_path, caplog):
caplog.set_level(logging.WARNING, logger="app.jobs.daily_pipeline")
_make_partitions(tmp_path, "kline_daily_enriched", ["2026-09-01", "2026-09-02"])
days = daily_pipeline.warn_if_enriched_too_thin(tmp_path)
assert days == 2
assert any("指标暖机不足" in r.message for r in caplog.records)
def test_pipeline_silent_when_coverage_sufficient(tmp_path, caplog):
caplog.set_level(logging.WARNING, logger="app.jobs.daily_pipeline")
_make_partitions(
tmp_path, "kline_daily_enriched",
[f"2026-{m:02d}-15" for m in range(1, 10)] + [f"2026-10-{d:02d}" for d in range(1, 25)],
)
days = daily_pipeline.warn_if_enriched_too_thin(tmp_path)
assert days >= MIN_INDICATOR_WARMUP_DAYS
assert not any("指标暖机不足" in r.message for r in caplog.records)