#!/usr/bin/env python3 """因果链数据验证脚本 — 每月 cron 自动跑 (2026-08-27 P2) 对 kpi_causality 每条链： 取 source/target KPI 的 kpi_values 历史值 → Pearson 相关系数 + 方向一致性 + 滞后对齐(lag_months) → 更新 verify_status: data_verified / disputed / pending → 输出验证报告 JSON + 控制台摘要 用法: python3 scripts/correlation-check.py # 全部企业，写库 python3 scripts/correlation-check.py --entity-id 1 # 指定企业 python3 scripts/correlation-check.py --dry-run # 只算不写库 月度 cron: 0 9 1 * * cd /root/cma-management/backend && python3 scripts/correlation-check.py """ import argparse import json import logging import sys from datetime import datetime from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from sqlalchemy import text # noqa: E402 from app.database import get_engine # noqa: E402 from app.services.causality_verification import ( # noqa: E402 STATUS_DATA_VERIFIED, STATUS_DISPUTED, STATUS_HUMAN_VERIFIED, STATUS_PENDING, VERIFIER_SCRIPT, apply_state_machine, evaluate_chain, summarize, ) logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") logger = logging.getLogger("correlation-check") REPORT_DIR = Path(__file__).resolve().parent / "reports" def load_chains(engine, entity_id: int = None) -> list: """加载因果链 + 两端KPI信息。""" q = """ SELECT c.id, c.entity_id, c.source_kpi_id, c.target_kpi_id, c.strength, c.lag_months, c.direction, c.source_type, c.verify_status, s.kpi_code AS src_code, s.kpi_name AS src_name, t.kpi_code AS tgt_code, t.kpi_name AS tgt_name FROM kpi_causality c JOIN kpi_definitions s ON s.id = c.source_kpi_id JOIN kpi_definitions t ON t.id = c.target_kpi_id """ if entity_id is not None: q += " WHERE c.entity_id = :eid" with engine.connect() as conn: rows = conn.execute(text(q), {"eid": entity_id} if entity_id is not None else {}).mappings().all() return [dict(r) for r in rows] def load_values(engine, kpi_ids: list) -> dict: """加载 KPI 历史值: {kpi_id: [(period, actual_value), ...]}""" if not kpi_ids: return {} ids = list(set(int(i) for i in kpi_ids)) q = """ SELECT kpi_id, period, actual_value FROM kpi_values WHERE kpi_id IN :ids AND actual_value IS NOT NULL ORDER BY period """ with engine.connect() as conn: rows = conn.execute(text(q).bindparams(ids=ids), {"ids": ids}).mappings().all() result = {} for r in rows: result.setdefault(r["kpi_id"], []).append((r["period"], r["actual_value"])) return result def main(): ap = argparse.ArgumentParser(description="因果链数据验证") ap.add_argument("--entity-id", type=int, default=None, help="只验证指定企业（默认全部）") ap.add_argument("--dry-run", action="store_true", help="只计算不写库") args = ap.parse_args() engine = get_engine() chains = load_chains(engine, args.entity_id) if not chains: logger.info("无因果链，退出") return 0 kpi_ids = [c["source_kpi_id"] for c in chains] + [c["target_kpi_id"] for c in chains] values = load_values(engine, kpi_ids) now = datetime.now() results = [] updated = {"data_verified": 0, "disputed": 0, "unchanged": 0} notes = [] with engine.begin() as conn: for c in chains: src_vals = values.get(c["source_kpi_id"], []) tgt_vals = values.get(c["target_kpi_id"], []) ev = evaluate_chain( src_vals, tgt_vals, lag_months=c["lag_months"] or 0, direction=c["direction"] or "positive", ) new_status, note = apply_state_machine(c["verify_status"], ev["status"], respect_human=True) if note: notes.append({"causality_id": c["id"], "note": note}) changed = new_status != c["verify_status"] if changed: updated[new_status if new_status in updated else "unchanged"] = \ updated.get(new_status if new_status in updated else "unchanged", 0) + 1 else: updated["unchanged"] += 1 if not args.dry_run: conn.execute(text( "UPDATE kpi_causality SET verify_status = :st, verified_at = :va, verified_by = :vb " "WHERE id = :cid" ), { "st": new_status, "va": now, "vb": VERIFIER_SCRIPT, "cid": c["id"], }) results.append({ "causality_id": c["id"], "source": f'{c["src_code"]}({c["src_name"]})', "target": f'{c["tgt_code"]}({c["tgt_name"]})', "direction": c["direction"], "lag_months": c["lag_months"], "strength": c["strength"], "granularity": ev["granularity"], "n_points": ev["n"], "r": round(ev["r"], 4) if ev["r"] is not None else None, "direction_consistent": ev["direction_consistent"], "old_status": c["verify_status"], "new_status": new_status, "reason": ev["reason"], }) summary = summarize([{"status": r["new_status"]} for r in results]) report = { "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"), "script": VERIFIER_SCRIPT, "dry_run": args.dry_run, "entity_id": args.entity_id, "summary": summary, "updated": updated, "human_verified_notes": notes, "chains": results, } REPORT_DIR.mkdir(exist_ok=True) report_path = REPORT_DIR / f"causality_verification_{now.strftime('%Y%m%d_%H%M%S')}.json" report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") # 控制台摘要（cron 输出即消息） lines = [ f"因果链数据验证{'[dry-run]' if args.dry_run else ''} {now.strftime('%Y-%m-%d %H:%M')}", f"总数: {summary['total']} | 数据证实: {summary['by_status'][STATUS_DATA_VERIFIED]} | " f"存疑: {summary['by_status'][STATUS_DISPUTED]} | 待检(数据不足): {summary['by_status'][STATUS_PENDING]} | " f"人工确认: {summary['by_status'][STATUS_HUMAN_VERIFIED]}", f"本次更新: data_verified={updated['data_verified']} disputed={updated['disputed']} unchanged={updated['unchanged']}", ] verified = [r for r in results if r["new_status"] == STATUS_DATA_VERIFIED] disputed = [r for r in results if r["new_status"] == STATUS_DISPUTED] if verified: lines.append("── 数据证实 ──") for r in verified: lines.append(f" #{r['causality_id']} {r['source']}→{r['target']} r={r['r']} n={r['n_points']}") if disputed: lines.append("── 数据存疑 ──") for r in disputed: lines.append(f" #{r['causality_id']} {r['source']}→{r['target']} r={r['r']} n={r['n_points']} ({r['reason']})") if notes: lines.append("── 人工确认链的数据警示 ──") for n in notes: lines.append(f" #{n['causality_id']}: {n['note']}") lines.append(f"报告: {report_path}") print("\n".join(lines)) logger.info("报告已写入 %s", report_path) return 0 if __name__ == "__main__": sys.exit(main())