67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""审计 AlphaX 技术/消息触发时效，防止旧形态冒充当下交易机会。"""
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
from app.db.schema import get_conn, init_db
|
|
init_db()
|
|
conn = get_conn()
|
|
errors = []
|
|
warnings = []
|
|
|
|
|
|
def parse_json(v):
|
|
try:
|
|
if isinstance(v, dict):
|
|
return v
|
|
if v:
|
|
return json.loads(v)
|
|
except Exception:
|
|
pass
|
|
return {}
|
|
|
|
|
|
def add(kind, rule, row):
|
|
(errors if kind == "error" else warnings).append({"rule": rule, "row": row})
|
|
|
|
rows = conn.execute("""
|
|
SELECT id,symbol,rec_state,action_status,execution_status,display_bucket,signals,market_context_json,sector_context_json,entry_plan_json,rec_time
|
|
FROM recommendation
|
|
WHERE status='active' AND COALESCE(display_bucket,'watch_pool')!='history'
|
|
ORDER BY id DESC
|
|
""").fetchall()
|
|
|
|
for r in rows:
|
|
d = dict(r)
|
|
market = parse_json(d.get("market_context_json"))
|
|
sector = parse_json(d.get("sector_context_json"))
|
|
ep = parse_json(d.get("entry_plan_json"))
|
|
tc = market.get("trigger_context") or sector.get("trigger_context") or ep.get("trigger_context") or {}
|
|
sig_text = d.get("signals") or ""
|
|
if d.get("execution_status") == "buy_now":
|
|
if not tc:
|
|
add("warning", "buy_now_missing_trigger_context", {k:d.get(k) for k in ("id","symbol","rec_time","action_status")})
|
|
elif tc.get("trigger_status") == "stale_background_only":
|
|
add("error", "buy_now_from_stale_background", {"id": d["id"], "symbol": d["symbol"], "trigger_context": tc})
|
|
if "历史" in sig_text and "已过期" in sig_text and d.get("execution_status") == "buy_now":
|
|
current = (tc.get("current_triggers") or []) if isinstance(tc, dict) else []
|
|
if not current:
|
|
add("error", "stale_signal_buy_now_without_current_trigger", {"id": d["id"], "symbol": d["symbol"], "signals": sig_text[:300]})
|
|
if "event_title" in sector or "event_source" in market:
|
|
if not tc and not (sector.get("trigger_context") or market.get("trigger_context")):
|
|
add("warning", "event_recommendation_missing_event_marker", {"id": d["id"], "symbol": d["symbol"], "sector_context": sector, "market_context": market})
|
|
|
|
summary = {
|
|
"active_checked": len(rows),
|
|
"errors": errors,
|
|
"warnings": warnings,
|
|
}
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
conn.close()
|
|
if errors:
|
|
raise SystemExit(1)
|