feat: 因果链三层验证机制(P2) — 数据验证+人工确认+状态机
- kpi_causality 加列: source_type/verify_status/verified_at/verified_by/entity_id(回填)
- 核心服务: app/services/causality_verification.py (Pearson+滞后对齐+状态机)
- 数据验证脚本: scripts/correlation-check.py (月度cron, 输出JSON报告)
- API: create/update支持source_type, GET /verify-status, PUT /{id}/verify(人工确认)
- 全部端点按entity_id账套隔离, kpi/{id}/network/simulate补跨企业校验
- 前端: KPIDetail因果链页显示验证状态徽标(数据证实/存疑/待检)
- 测试: test_causality_verification.py 37用例 + 原因果链测试全过(59个)
- 50条因果链首轮验证: 1数据证实(#37渠补率到净利润lag1 r=-0.89), 4存疑, 45待检(数据不足)
This commit is contained in:
Hermes CI Fix
@@ -0,0 +1,193 @@
|
||||
#!/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())
|
||||
@@ -0,0 +1,95 @@
|
||||
"""kpi_causality 因果链验证机制迁移脚本 (2026-08-27 P2)
|
||||
|
||||
加列:
|
||||
- source_type: varchar(20) 建链来源 AI_suggested/manual/imported
|
||||
- verify_status: varchar(20) 验证状态 pending/data_verified/human_verified/disputed
|
||||
- verified_at: datetime 验证时间
|
||||
- verified_by: varchar(50) 验证人/AI/脚本
|
||||
- entity_id: int 多租户隔离 (2026-08-27 收官补齐)
|
||||
|
||||
幂等: 列已存在则跳过; entity_id 回填只更新 NULL/0 行。
|
||||
用法: python scripts/migrate_causality_verification.py
|
||||
"""
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.database import get_engine
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
|
||||
logger = logging.getLogger("migrate-causality-verification")
|
||||
|
||||
COLUMNS = [
|
||||
("source_type", "ALTER TABLE kpi_causality ADD COLUMN source_type VARCHAR(20) NOT NULL DEFAULT 'manual' COMMENT '建链来源 AI_suggested/manual/imported'"),
|
||||
("verify_status", "ALTER TABLE kpi_causality ADD COLUMN verify_status VARCHAR(20) NOT NULL DEFAULT 'pending' COMMENT '验证状态 pending/data_verified/human_verified/disputed'"),
|
||||
("verified_at", "ALTER TABLE kpi_causality ADD COLUMN verified_at DATETIME NULL COMMENT '验证时间'"),
|
||||
("verified_by", "ALTER TABLE kpi_causality ADD COLUMN verified_by VARCHAR(50) NULL COMMENT '验证人/AI/脚本'"),
|
||||
("entity_id", "ALTER TABLE kpi_causality ADD COLUMN entity_id INT NOT NULL DEFAULT 1 COMMENT '企业ID (多租户隔离 2026-08-27)'"),
|
||||
]
|
||||
|
||||
|
||||
def run():
|
||||
engine = get_engine()
|
||||
with engine.connect() as conn:
|
||||
# 1. 检查表是否存在
|
||||
exists = conn.execute(text(
|
||||
"SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'kpi_causality'"
|
||||
)).scalar()
|
||||
if not exists:
|
||||
logger.error("kpi_causality 表不存在，跳过")
|
||||
return 1
|
||||
|
||||
# 2. 现有列
|
||||
existing = {r[0] for r in conn.execute(text(
|
||||
"SELECT column_name FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'kpi_causality'"
|
||||
))}
|
||||
logger.info("现有列: %s", sorted(existing))
|
||||
|
||||
# 3. 加列（幂等）
|
||||
for col, ddl in COLUMNS:
|
||||
if col in existing:
|
||||
logger.info("列 %s 已存在，跳过", col)
|
||||
else:
|
||||
conn.execute(text(ddl))
|
||||
logger.info("已添加列 %s", col)
|
||||
|
||||
# 4. 回填 entity_id（无条件从 source KPI 对齐，纠正默认值偏差）
|
||||
# 仅当来源KPI存在才回填；无来源KPI的孤儿链保持原值
|
||||
conn.execute(text(
|
||||
"UPDATE kpi_causality c JOIN kpi_definitions k ON k.id = c.source_kpi_id "
|
||||
"SET c.entity_id = k.entity_id"
|
||||
))
|
||||
orphan = conn.execute(text(
|
||||
"SELECT COUNT(*) FROM kpi_causality c LEFT JOIN kpi_definitions k ON k.id = c.source_kpi_id "
|
||||
"WHERE k.id IS NULL"
|
||||
)).scalar()
|
||||
if orphan:
|
||||
logger.warning("%d 条因果链无来源KPI（孤儿链）", orphan)
|
||||
else:
|
||||
logger.info("entity_id 已全部按来源KPI回填")
|
||||
|
||||
# 5. 验证
|
||||
cols = {r[0] for r in conn.execute(text(
|
||||
"SELECT column_name FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'kpi_causality'"
|
||||
))}
|
||||
missing = {c for c, _ in COLUMNS} - cols
|
||||
if missing:
|
||||
logger.error("仍有缺失列: %s", missing)
|
||||
return 1
|
||||
row = conn.execute(text(
|
||||
"SELECT COUNT(*) FROM kpi_causality WHERE entity_id IS NULL OR entity_id = 0"
|
||||
)).scalar()
|
||||
if row:
|
||||
logger.error("仍有 %d 行 entity_id 为空", row)
|
||||
return 1
|
||||
total = conn.execute(text("SELECT COUNT(*) FROM kpi_causality")).scalar()
|
||||
logger.info("迁移完成: kpi_causality %d 条, 新列: source_type/verify_status/verified_at/verified_by/entity_id", total)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(run())
|
||||
Reference in New Issue
Block a user