fix(screener): 列表涨跌幅改用实时行情, 与个股详情对齐

策略结果缓存只记录「计算那一刻」的价格。盘中跑完 run_all 后, 列表里的
close/change_pct 会一直停在那时, 而点开个股详情走的是实时 enriched 缓存,
同一只票在两处显示不同涨幅。实测 002543.SZ: 13:27 扫描时 2.49%, 收盘
4.48%, 缓存里 279 行中有 271 行与实时值不一致。

后端:
- /api/screener/cached 与 /cached-result 返回前, 用当前实时快照覆盖纯展示
  字段 (open/high/low/close/prev_close/volume/amount/change_pct/
  change_amount/turnover_rate/raw_*/quote_ts)
- 指标与信号列 (ma20/score/signal_* 等) 保持计算时点值: 用新价格重算会让
  「为什么入选」不可解释
- 仅覆盖同日结果 (优先结果级 as_of, 行级 date 兜底), 历史回看不被今日行情
  污染; 实时缓存为空时原样返回, 不伪造价格
- 覆盖只发生在响应层, 不写回 strategy_cache.json (写回会让 mtime 指纹失效,
  等于每轮触发一次全量重读)

前端:
- 明细查询在交易时段 6s 重取、收盘后 60s 兜底, 让覆盖真正反映到页面

验证: 新增 4 条回归 (实时覆盖生效且指标保留/历史日期不覆盖/实时缓存空时
不变/cached-result 同口径); 后端全量 2926 passed; pnpm build 通过;
运行实例上两接口均返回实时价, 缓存文件仍保存原值。
This commit is contained in:
shy3130
2026-10-20 16:06:43 +08:00
parent 2c6f03dad7
commit 5f6444277d
3 changed files with 294 additions and 5 deletions
+139 -3
View File
@@ -282,6 +282,136 @@ def _cache_payload_with_ext(
return payload
# 被实时行情覆盖的展示字段。
#
# 策略结果缓存只记录「计算那一刻」的价格: 盘中运行 run_all 后, 列表里的
# close/change_pct 会一直停在那时, 而点开详情走的是实时 enriched 缓存 ——
# 同一个票在两个地方显示不同涨幅 (实测 13:27 的扫描结果到收盘差了 2 个点)。
# 这里在返回前用当前实时快照覆盖这些纯展示字段; 指标列 (ma20/score/signal_*)
# 保持计算时点值不动, 否则等于用新价格重算了一个过期策略结果。
_LIVE_QUOTE_FIELDS: tuple[str, ...] = (
"open", "high", "low", "close",
"prev_close", "volume", "amount",
"change_pct", "change_amount", "turnover_rate",
"raw_close", "raw_high", "raw_low", "quote_ts",
)
def _live_quote_overlay(repo) -> tuple[dict[str, dict[str, Any]], str | None]:
"""当前实时 enriched 快照的展示字段: ({symbol: row}, 日期字符串)。
缓存为空 (启动预热中) 或 repo 不支持该接口时返回空映射, 调用方原样返回
策略结果 —— 拿不到实时数据时不能伪造新价格。
"""
loader = getattr(repo, "get_enriched_latest", None)
if not callable(loader):
return {}, None
try:
frame, cache_date = loader()
except Exception: # noqa: BLE001 — 展示层增强绝不打断缓存读取
logger.debug("live quote overlay unavailable", exc_info=True)
return {}, None
if frame is None or frame.is_empty() or cache_date is None:
return {}, None
columns = [c for c in ("symbol", *_LIVE_QUOTE_FIELDS) if c in frame.columns]
if "symbol" not in columns:
return {}, None
return {
str(row["symbol"]): row
for row in frame.select(columns).to_dicts()
}, str(cache_date)
def _quote_date_matches(row: dict, live_date: str | None, result_as_of: Any = None) -> bool:
"""结果日期与实时快照同日才可覆盖。
历史日期结果 (例如回看昨天/上周) 必须保留当时价格, 不能用今天的行情覆盖。
优先用结果级 as_of (策略结果整体属于哪一天), 行级 date 作兜底 —— 个别
策略输出行不带 date 字段, 只看行日期会误覆盖历史结果。
"""
if live_date is None:
return False
for candidate in (result_as_of, row.get("date")):
if candidate is None:
continue
return str(candidate)[:10] == live_date
# 结果与行都没有日期: 视为「最新」(与 as_of=None 的缓存语义一致)
return True
def _rows_with_live_quotes(
rows: list[dict],
overlay: dict[str, dict[str, Any]],
live_date: str | None,
result_as_of: Any = None,
) -> list[dict]:
if not overlay:
return rows
merged: list[dict] = []
for row in rows:
if not isinstance(row, dict):
merged.append(row)
continue
live = overlay.get(str(row.get("symbol") or ""))
if live is None or not _quote_date_matches(row, live_date, result_as_of):
merged.append(row)
continue
next_row = dict(row)
for key, value in live.items():
if key == "symbol":
continue
next_row[key] = value
merged.append(next_row)
return merged
def _cached_with_live_quotes(request: Request) -> dict:
"""盘后缓存 + 监控实时结果 + 当前行情展示字段覆盖。
与 `_cached_with_realtime` 的区别: 后者只覆盖「加入监控的策略」(每轮行情
重算结果), 未加监控的策略仍显示计算时点价格。这里对所有策略只覆盖价格类
展示字段, 行情轮询期间列表与详情保持一致。
"""
cached = _cached_with_realtime(request)
results = cached.get("results") or {}
ever_rows = cached.get("today_ever_rows")
if not results and not ever_rows:
return cached
overlay, live_date = _live_quote_overlay(request.app.state.repo)
if not overlay:
return cached
payload = dict(cached)
payload["results"] = {
sid: (
{
**result,
"rows": _rows_with_live_quotes(
result.get("rows") or [], overlay, live_date, result.get("as_of"),
),
}
if isinstance(result, dict) else result
)
for sid, result in results.items()
}
if isinstance(ever_rows, dict):
# 「今日曾命中」行是逐行的历史快照 (各自带 date), 一律按行日期判断
payload["today_ever_rows"] = {
sid: (
{
symbol: _rows_with_live_quotes([row], overlay, live_date)[0]
for symbol, row in sym_map.items()
if isinstance(row, dict)
}
if isinstance(sym_map, dict) else sym_map
)
for sid, sym_map in ever_rows.items()
}
return payload
def _update_cache_strategy(data_dir, as_of: str, strategy_id: str, safe_data: dict) -> None:
"""单跑后更新缓存中该策略的结果,保持缓存与最新计算一致。"""
from app.services import strategy_cache
@@ -449,8 +579,10 @@ def get_cached(
- 监控引擎内存结果 (latest_strategy_results): 实时行情每轮对「加入监控的策略」算出,
不落盘 (避免与 read_cache 的 mtime 校验冲突), 在此直接叠加覆盖盘后结果。
被监控的策略拿到新鲜数据, 非监控策略仍用盘后缓存。
- 价格类展示字段统一用当前实时快照覆盖 (同日才覆盖), 避免列表显示扫描时刻
价格、详情显示实时价格造成的"同一只票两个涨幅"。
"""
cached = _cached_with_realtime(request)
cached = _cached_with_live_quotes(request)
# 无任何数据 (盘后缓存空 + 无实时结果) → 返回空标记, 前端据此提示
if not cached.get("results") and cached.get("as_of") is None:
@@ -529,8 +661,12 @@ def get_cached_result(
request: Request,
ext_columns: Optional[str] = Query(None, description="逗号分隔: config_id.field_name"),
):
"""按需返回单个策略的完整明细及其今日失效行。"""
cached = _cached_with_realtime(request)
"""按需返回单个策略的完整明细及其今日失效行。
与 /cached 一致: 价格类展示字段用当前实时快照覆盖 (同日才覆盖), 保证
详情里的涨跌幅与列表、以及个股详情弹窗同源。
"""
cached = _cached_with_live_quotes(request)
raw_result = (cached.get("results") or {}).get(strategy_id)
if not isinstance(raw_result, dict):
return {
+145 -1
View File
@@ -2,6 +2,8 @@ from __future__ import annotations
from types import SimpleNamespace
import polars as pl
from app.api import screener as screener_api
@@ -13,8 +15,10 @@ class _MonitorEngine:
return self.results
def _request(tmp_path, monitor_results=None):
def _request(tmp_path, monitor_results=None, live=None, live_date=None):
repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path))
if live is not None:
repo.get_enriched_latest = lambda: (live, live_date)
state = SimpleNamespace(repo=repo, monitor_engine=_MonitorEngine(monitor_results))
return SimpleNamespace(app=SimpleNamespace(state=state))
@@ -95,3 +99,143 @@ def test_cached_result_returns_only_requested_rows_with_ext_and_strategy_members
assert payload["result"]["rows"] == [{"symbol": "000001.SZ", "concept.concept": "银行"}]
assert payload["today_ever_rows"]["000002.SZ"]["concept.concept"] == "科技"
assert payload["strategy_ids_by_symbol"] == {"000001.SZ": ["strategy_a", "strategy_b"]}
# ---------------------------------------------------------------------------
# 列表价格与实时行情对齐: 缓存只记录策略计算时刻的价格
# ---------------------------------------------------------------------------
def _live_frame() -> pl.DataFrame:
return pl.DataFrame({
"symbol": ["002543.SZ", "600000.SH"],
"date": ["2026-10-20", "2026-10-20"],
"open": [7.8, 10.0],
"high": [8.44, 10.5],
"low": [7.77, 9.9],
"close": [8.39, 10.2],
"prev_close": [8.03, 10.0],
"volume": [449688.0, 1000.0],
"amount": [3.63e8, 1.0e7],
"change_pct": [0.04483, 0.02],
"change_amount": [0.36, 0.2],
"turnover_rate": [6.8, 1.0],
"raw_close": [8.39, 10.2],
"raw_high": [8.44, 10.5],
"raw_low": [7.77, 9.9],
"quote_ts": [1792479600000, 1792479600000],
})
def _cached_row() -> dict:
return {
"symbol": "002543.SZ",
"name": "万和电气",
"date": "2026-10-20",
"close": 8.23,
"prev_close": 8.03,
"change_pct": 0.0249,
"change_amount": 0.20,
"amount": 1.84e8,
"score": 50.0,
"ma20": 7.5,
"vol_ratio_5d": 1.32,
}
def _cached_payload(rows=None, as_of="2026-10-20") -> dict:
return {
"as_of": as_of,
"results": {
"everbloom_qibao": {
"as_of": as_of,
"total": 1,
"rows": rows if rows is not None else [_cached_row()],
},
},
"today_ever_rows": {
"everbloom_qibao": {"002543.SZ": dict(_cached_row())},
},
"updated_at": 1,
}
def test_cached_rows_take_live_quote_fields_but_keep_indicator_values(monkeypatch, tmp_path):
"""列表的 close/change_pct 用实时值; 指标列仍是策略计算时点值。
实测现场: 13:27 扫描时涨幅 2.49%, 收盘 4.48%, 详情弹窗读实时行情,
同一只票在两处显示不同涨幅。
"""
monkeypatch.setattr(
screener_api.strategy_cache, "read_cache", lambda *_a: _cached_payload()
)
payload = screener_api.get_cached(
_request(tmp_path, live=_live_frame(), live_date="2026-10-20"),
ext_columns=None,
)
row = payload["results"]["everbloom_qibao"]["rows"][0]
assert row["close"] == 8.39
assert row["change_pct"] == 0.04483
assert row["change_amount"] == 0.36
assert row["amount"] == 3.63e8
# 指标/评分不覆盖: 它们属于「计算时点」的策略结果
assert row["score"] == 50.0
assert row["ma20"] == 7.5
assert row["vol_ratio_5d"] == 1.32
ever = payload["today_ever_rows"]["everbloom_qibao"]["002543.SZ"]
assert ever["close"] == 8.39
assert ever["score"] == 50.0
def test_cached_rows_keep_historical_prices_for_other_dates(monkeypatch, tmp_path):
"""历史日期的结果不能被今天的实时价覆盖 (回看昨日/上周必须显示当时价格)。"""
monkeypatch.setattr(
screener_api.strategy_cache,
"read_cache",
lambda *_a: _cached_payload(as_of="2026-10-17"),
)
payload = screener_api.get_cached(
_request(tmp_path, live=_live_frame(), live_date="2026-10-20"),
ext_columns=None,
)
row = payload["results"]["everbloom_qibao"]["rows"][0]
assert row["close"] == 8.23
assert row["change_pct"] == 0.0249
def test_cached_rows_unchanged_when_live_cache_is_empty(monkeypatch, tmp_path):
"""实时缓存为空 (启动预热中) 时保持原样, 不能伪造价格。"""
monkeypatch.setattr(
screener_api.strategy_cache, "read_cache", lambda *_a: _cached_payload()
)
payload = screener_api.get_cached(
_request(tmp_path, live=pl.DataFrame(), live_date=None),
ext_columns=None,
)
assert payload["results"]["everbloom_qibao"]["rows"][0]["close"] == 8.23
def test_cached_result_overlays_live_quotes(monkeypatch, tmp_path):
"""/cached-result (详情接口) 与列表使用同一套实时覆盖。"""
monkeypatch.setattr(
screener_api.strategy_cache, "read_cache", lambda *_a: _cached_payload()
)
monkeypatch.setattr(screener_api, "_load_ext_value_maps", lambda *_a, **_k: {})
payload = screener_api.get_cached_result(
"everbloom_qibao",
_request(tmp_path, live=_live_frame(), live_date="2026-10-20"),
ext_columns=None,
)
row = payload["result"]["rows"][0]
assert row["close"] == 8.39
assert row["change_pct"] == 0.04483
assert row["score"] == 50.0
assert payload["today_ever_rows"]["002543.SZ"]["close"] == 8.39
+10 -1
View File
@@ -170,6 +170,12 @@ export function Screener() {
// 卡片首屏只读取轻量摘要;明细在点击策略或“全部”时按需加载。
// 摘要只覆盖日线缓存; 分钟策略命中数来自手动单跑。
// run_all 渐进式返回后后台仍在算 → 轮询摘要, 算完的策略逐个点亮。
//
// 后端会用当前实时行情覆盖结果里的价格类字段 (close/change_pct 等), 所以
// 交易时段内需要定期重取, 否则列表一直停在首次加载那一刻的涨跌幅, 与详情
// 弹窗的实时价对不上。摘要没有价格字段, 仍只在 run_all 未完成时轮询。
const quoteStatus = useQuoteStatus()
const inTradingSession = quoteStatus.data?.is_trading_hours !== false
const summaryQuery = useQuery({
queryKey: QK.screenerCachedSummary,
queryFn: api.screenerCachedSummary,
@@ -181,6 +187,8 @@ export function Screener() {
queryKey: QK.screenerCached(asOf, extColumnsParam),
queryFn: () => api.screenerCached(extColumnsParam || undefined),
enabled: assetType === 'stock' && tfFilter !== '1m' && showAll,
// 交易时段跟随行情轮询; 收盘后价格不再变, 降为 60s 仅兜底数据修正
refetchInterval: inTradingSession ? 6_000 : 60_000,
})
const singleCachedQuery = useQuery({
@@ -191,6 +199,8 @@ export function Screener() {
&& !showAll
&& !!activeStrategy
&& summaryQuery.data?.results[activeStrategy]?.as_of === asOf,
// 与 fullCachedQuery 同口径: 列表与详情共享同一份实时覆盖后的价格
refetchInterval: inTradingSession ? 6_000 : 60_000,
})
const dataStatus = useDataStatus({ staleTime: 0 })
@@ -578,7 +588,6 @@ export function Screener() {
// - 刷新: minute_intraday_refresh 偏好开启时按用户设定间隔轮询; 否则仅首次加载,
// 用户可点表头刷新按钮手动更新
const minuteBatchCap = caps.data?.capabilities?.['kline.minute.batch']?.batch ?? 100
const quoteStatus = useQuoteStatus()
const realtimeRunning = quoteStatus.data?.running ?? false
const intradayRefreshEnabled = prefs?.minute_intraday_refresh ?? false
const intradayRefreshInterval = prefs?.minute_intraday_refresh_interval ?? 6