164 lines
6.8 KiB
Python
164 lines
6.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""sanguo-strategy-monitor — VPS 策略域运维监控班次(08:05 morn / 21:30 eve / noon 按需).
|
|
|
|
数据面 data_monitor.py 三件套的策略镜像(轨=spec §20.15): 只读体检不修数据;
|
|
告警写 alerts.db source="strategy"(与 data 域同表分域, 前端类别页直读);
|
|
health 落 {data_root}/monitor/health_strategy_{date}_{shift}.json +
|
|
health_strategy_latest.json——**health_latest.json 是数据域的, 永不写**。
|
|
退出码: 0=绿/黄, 3=有红(wrapper 恒转 0, 红态由 alerts/health 承载)。
|
|
"""
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
_REPO = os.path.dirname(os.path.dirname(_HERE))
|
|
for _p in (_HERE, _REPO):
|
|
if _p not in sys.path:
|
|
sys.path.insert(0, _p)
|
|
|
|
from sanguo_data import alerts_store as store # noqa: E402
|
|
from sanguo_portfolio import pipeline_store # noqa: E402
|
|
import strategy_checks as chk # noqa: E402
|
|
import strategy_registry as reg # noqa: E402
|
|
|
|
_TODAY_OVERRIDE = None # 测试注入口
|
|
_HEALTH_DIR = "monitor"
|
|
_HEALTH_PREFIX = "health_strategy"
|
|
_DEFAULT_MAIN_DB = os.path.join("data", "backtest_results.db")
|
|
|
|
|
|
def _now():
|
|
if _TODAY_OVERRIDE is not None:
|
|
return dt.datetime.combine(_TODAY_OVERRIDE, dt.time(23, 50))
|
|
return dt.datetime.now()
|
|
|
|
|
|
def default_main_db_path():
|
|
"""env SANGUO_DB_PATH 覆盖; 缺省 data/backtest_results.db(相对 cwd)."""
|
|
return os.environ.get("SANGUO_DB_PATH", _DEFAULT_MAIN_DB)
|
|
|
|
|
|
def default_logs_dir():
|
|
return os.path.join(_REPO, "logs")
|
|
|
|
|
|
def calendar_union(confirmed_days, now_date, lookback=15):
|
|
"""结果层日历 = 确认集(dbbardata 实证)∪近 lookback 日工作日假定集(镜像
|
|
data_monitor._calendar_days)。10-03 假黄根治: 假定集剔 EXCHANGE_HOLIDAYS
|
|
休市日(已知假日=零期望, 国庆连休不天天假黄); 表缺年=不剔(黄照旧绝不
|
|
静默转绿); 确认集恒保留(表错不吞真相)。"""
|
|
union = set(confirmed_days)
|
|
d0 = now_date - dt.timedelta(days=lookback)
|
|
while d0 <= now_date:
|
|
hol = reg.EXCHANGE_HOLIDAYS.get(d0.year)
|
|
if d0.weekday() < 5 and (hol is None or d0.strftime("%Y%m%d") not in hol):
|
|
union.add(d0.isoformat())
|
|
d0 += dt.timedelta(days=1)
|
|
return union
|
|
|
|
|
|
def check_entry(e, ctx):
|
|
"""注册表项分发; 查表+探测器自身抛错=黄(单检坏了不阻断全班; S-1:
|
|
旧码查表在 try 外, 注册表 fn 名笔误=KeyError 整班崩+wrapper 恒 0 静默)."""
|
|
try:
|
|
fn = chk.CHECK_FNS[e["fn"]]
|
|
return fn(ctx, e)
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"key": e["key"], "status": "yellow",
|
|
"expected": e.get("title", e["key"]),
|
|
"actual": f"error: {exc!r}", "evidence": {"error": repr(exc)}}
|
|
|
|
|
|
def write_health(data_root, health):
|
|
d = os.path.join(data_root, _HEALTH_DIR)
|
|
os.makedirs(d, exist_ok=True)
|
|
for fn in (f"{_HEALTH_PREFIX}_{health['date']}_{health['shift']}.json",
|
|
f"{_HEALTH_PREFIX}_latest.json"):
|
|
tmp = os.path.join(d, fn + ".tmp")
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(health, f, ensure_ascii=False, indent=1)
|
|
os.replace(tmp, os.path.join(d, fn))
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--shift", choices=["morn", "eve", "noon"], required=True)
|
|
ap.add_argument("--data-root", default="data")
|
|
ap.add_argument("--db", default=store.default_alerts_db_path())
|
|
ap.add_argument("--main-db", default=default_main_db_path())
|
|
ap.add_argument("--pipeline-db",
|
|
default=pipeline_store.default_pipeline_db_path())
|
|
ap.add_argument("--logs-dir", default=default_logs_dir())
|
|
args = ap.parse_args(argv)
|
|
|
|
now = _now()
|
|
today = now.date().isoformat()
|
|
today_compact = today.replace("-", "")
|
|
db_path = args.db
|
|
if args.db == store.default_alerts_db_path() and args.data_root != "data":
|
|
db_path = os.path.join(args.data_root, "alerts.db")
|
|
|
|
# 日历语义逐字镜像 data_monitor: 确认集(dbbardata 000300 实证)∪近 15 日
|
|
# 工作日假定集(剔休市表日)——dbbardata 晚间源滞后/漏日会把全体判错,
|
|
# 假定集兜底。
|
|
days = chk.trading_days(args.main_db, now.date())
|
|
confirmed = {d.replace("-", "") for d in days}
|
|
union = calendar_union(days, now.date())
|
|
is_confirmed = today_compact in confirmed
|
|
|
|
ctx = {"main_db": args.main_db, "pipeline_db": args.pipeline_db,
|
|
"logs_dir": args.logs_dir, "now": now, "today": today,
|
|
"midnight": dt.datetime.combine(now.date(), dt.time.min),
|
|
# data-monitor 健康文件(互为外盯 P1-10): 同 data_root/monitor 约定
|
|
"data_monitor_health": os.path.join(
|
|
args.data_root, "monitor", "health_latest.json")}
|
|
|
|
conn = store.connect(db_path)
|
|
checks, events, passed = [], [], []
|
|
for e in reg.REGISTRY:
|
|
if not e.get("enabled", True) or args.shift not in e["shifts"]:
|
|
continue
|
|
if e.get("trading_day") and today not in union:
|
|
continue # 周末/连休: 交易日语义不评(告警不误开, 周五开的也不消解)
|
|
r = check_entry(e, ctx)
|
|
if r["status"] == "red" and e.get("trading_day") and not is_confirmed:
|
|
# 未确认工作日红级违例降黄(节假日/合并窗/源滞后三态容忍, 同款降级)
|
|
r = dict(r, status="yellow",
|
|
evidence=dict(r.get("evidence", {}), unconfirmed_day=True))
|
|
checks.append(r)
|
|
if r["status"] == "green":
|
|
passed.append(r["key"])
|
|
else:
|
|
events.append({
|
|
"alert_id": f"{e['key']}-{today_compact}",
|
|
"check_key": e["key"], "source": "strategy", "host": "vps",
|
|
"severity": r["status"], "title": e.get("title", e["key"]),
|
|
"detail": f"actual={r['actual']} expected={r['expected']}",
|
|
"evidence": r.get("evidence", {})})
|
|
|
|
for ev in events:
|
|
store.upsert_alert(conn, ev)
|
|
store.resolve_passed(conn, passed) # 本班评估为绿→自动消解(悬死案教训)
|
|
store.purge_expired(conn)
|
|
conn.close()
|
|
|
|
counts = {s: sum(1 for c in checks if c["status"] == s)
|
|
for s in ("green", "yellow", "red")}
|
|
overall = "red" if counts["red"] else (
|
|
"yellow" if counts["yellow"] else "green")
|
|
health = {"date": today, "shift": args.shift, "host": "vps",
|
|
"generated_at": now.isoformat(timespec="seconds"),
|
|
"overall": overall, "counts": counts,
|
|
"alerts_emitted": len(events), "checks": checks}
|
|
write_health(args.data_root, health)
|
|
print(f"[strategy-monitor] {args.shift} overall={overall} "
|
|
f"checks={len(checks)} alerts={len(events)}")
|
|
return 3 if overall == "red" else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|