数据新鲜度防御体系(致命错误防御)

根因:今下午报告用周五多周期缓存(multi_tf_cache)作周一操作建议,
中芯国际H浮盈+10%被错报破止损。

修改:
1. price_monitor 新增 live_prices.json 写入(每2分钟刷新所有实时价)
2. 新增 data_freshness.py — data_freshness check function
3. intraday_health_check price_monitor检测从10min收紧到5min
4. 新增 midday MTF cache refresh (11:00+14:00)
5. cron-report-format pre-flight checklist 新增数据新鲜度检查项

所有报告产出前必须先跑 data_freshness,过期则禁止出操作建议
This commit is contained in:
知微
2026-06-29 15:23:32 +08:00
parent 6a97d93018
commit aa4f013ee5
7 changed files with 9016 additions and 905 deletions
+80
View File
@@ -0,0 +1,80 @@
#!/usr/bin/env python3
"""data_freshness.py — 数据新鲜度校验
所有报告管道在生成输出前必须调用 check_fresh()。
返回 (pass: bool, details: str),如果数据过期则阻止生成操作建议。
用法:
from data_freshness import check_fresh
ok, msg = check_fresh()
if not ok:
print(f"⚠️ 数据过期: {msg}")
sys.exit(0) # 不生成报告
校验规则:
- 盘中 (9:30~15:00)price/live_prices.json 必须在 5 分钟内刷新
- 盘后 (9:30以前/15:00以后):允许最长 120 分钟
- 周末/节假日:跳过校验
"""
import json, os
from datetime import datetime, timedelta
LIVE_PRICES_PATH = "/home/hmo/web-dashboard/data/live_prices.json"
PORTFOLIO_PATH = "/home/hmo/web-dashboard/data/portfolio.json"
def is_market_hours():
now = datetime.now()
if now.weekday() >= 5: # 周六日
return False, "weekend"
t = now.hour * 60 + now.minute
if 9*60+30 <= t <= 15*60:
return True, "trading"
return False, "closed"
def check_fresh():
"""返回 (ok: bool, msg: str)"""
now = datetime.now()
# 先看是不是交易日
in_market, period = is_market_hours()
max_age_min = 5 if in_market else 120
# 主指标:live_prices.json
if os.path.exists(LIVE_PRICES_PATH):
try:
lp = json.load(open(LIVE_PRICES_PATH))
lp_time = lp.get("updated_at", "")
if not lp_time:
return False, "live_prices.json updated_at 为空"
lp_dt = datetime.fromisoformat(lp_time)
age = (now - lp_dt).total_seconds() / 60
if age > max_age_min:
return False, f"live_prices.json 已 {age:.0f} 分钟未更新(阈值 {max_age_min} 分钟)"
return True, f"数据新鲜({age:.0f} 分钟前)"
except Exception as e:
return False, f"live_prices.json 读取失败: {e}"
else:
# fallback: portfolio.json
if os.path.exists(PORTFOLIO_PATH):
try:
pf = json.load(open(PORTFOLIO_PATH))
pf_time = pf.get("updated_at", "")
if not pf_time:
return False, "portfolio.json updated_at 为空"
pf_dt = datetime.fromisoformat(pf_time)
age = (now - pf_dt).total_seconds() / 60
if age > max_age_min:
return False, f"portfolio.json 已 {age:.0f} 分钟未更新(阈值 {max_age_min} 分钟)"
return True, f"数据新鲜(portfolio.json {age:.0f} 分钟前)"
except Exception as e:
return False, f"portfolio.json 读取失败: {e}"
return False, "live_prices.json 和 portfolio.json 均不存在"
if __name__ == "__main__":
ok, msg = check_fresh()
print(f"{'' if ok else ''} {msg}")