218 lines
8.4 KiB
Python
218 lines
8.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""integrity_gate.py — 数据底座完整性门(spec §4.8 前置纪律②, P4-3 F6).
|
|
|
|
三层(UZI 实勘口径, 坐标=调研报告 §3): L1 存在性(profile 域目录+≥1 dt 分区;
|
|
新鲜度归数据监控不归门——两职责不叠); L2 字段级四元组 (dim, path, label,
|
|
critical) 自定清单(窗口内行级有效率≥VALID_RATE_MIN); L3 充实度(enrich 维
|
|
空率>ENRICH_MAX_NULL_RATE → warn)。
|
|
|
|
profile-aware 分母: 仅 --profile 启用域内的 L2 项进 coverage 分母(未启用
|
|
域不进——防新数据域未全启用时结构性误杀, UZI lite 3/18=17% 前车之鉴)。
|
|
|
|
判级: critical 项败或 coverage<40% → crit; coverage<60% 或 optional 项败或
|
|
L3 败 → warning; 否则绿。
|
|
出口: crit → exit 1; warning → exit 2; 绿 → exit 0。
|
|
SANGUO_SKIP_INTEGRITY=1 唯一调试后门(exit 0 + 报告记 skipped)。
|
|
每 check 独立 try/except, 异常=该 check fail(不炸全局); 报告记 checks_run。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
import pandas as pd
|
|
|
|
_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 scripts.data_platform import funnel_lexicon as lex # noqa: E402
|
|
from sanguo_data import partition_reader as pr # noqa: E402
|
|
|
|
log = logging.getLogger("integrity")
|
|
|
|
PROFILES: dict[str, tuple[str, ...]] = {
|
|
"default": ("news_meta", "ann_meta", "flash_meta", "events_in",
|
|
"events_llm"),
|
|
"events": ("events_in", "events_llm"),
|
|
}
|
|
|
|
# L2 字段级四元组 (dim, path, label, critical) — v1 12 项 6 critical
|
|
# (结构照 UZI 18 项 6 critical 比例起步; 扩充=加行)
|
|
CRITICAL_CHECKS: list[tuple[str, str, str, bool]] = [
|
|
("events_llm", "event_id", "事件唯一 id 非空", True),
|
|
("events_llm", "content_hash", "输入快照外键非空", True),
|
|
("events_llm", "event_type", "事件类型枚举合法", True),
|
|
("events_llm", "stock_code", "股票代码六位合法", True),
|
|
("events_llm", "event_date", "事件日期 ISO 合法", True),
|
|
("events_in", "content_hash", "输入切片唯一键非空", True),
|
|
("events_llm", "direction", "方向枚举合法", False),
|
|
("events_llm", "confidence", "置信度 0~1 合法", False),
|
|
("events_llm", "model", "判读来源标记非空", False),
|
|
("events_in", "src_id", "源行 id 非空", False),
|
|
("events_in", "title", "输入标题非空", False),
|
|
("events_in", "text_snippet", "LLM 输入文本非空", False),
|
|
]
|
|
|
|
# L3 充实度维 (domain, field, label) — 空率超阈 → warn
|
|
ENRICH_DIMS: list[tuple[str, str, str]] = [
|
|
("events_llm", "confidence", "判读置信度"),
|
|
("events_llm", "direction", "方向标注"),
|
|
("events_in", "text_snippet", "输入摘要"),
|
|
]
|
|
|
|
VALID_RATE_MIN = 0.95 # L2 行级有效率下限
|
|
ENRICH_MAX_NULL_RATE = 0.50 # L3 充实度空率上限
|
|
CRIT_PCT = 40.0 # coverage < 此 → 强制 critical
|
|
WARN_PCT = 60.0 # coverage < 此 → warning
|
|
WINDOW_DAYS = 7 # 检查窗口(近 N 日历日分区)
|
|
|
|
_CODE_RE = re.compile(r"^\d{6}$")
|
|
_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}")
|
|
|
|
|
|
def _valid_rate(df, dim, path) -> float:
|
|
"""窗口内行级有效率(缺列=0.0, 空表=1.0 无样本不误判; df=None=读取失败
|
|
抛错由调用方记 error——单查隔离)。"""
|
|
if df is None:
|
|
raise RuntimeError("域窗口读取失败(corrupt/缺权限)")
|
|
if path not in df.columns:
|
|
return 0.0
|
|
if df.empty:
|
|
return 1.0
|
|
s = df[path]
|
|
if path == "event_type":
|
|
ok = s.isin(lex.EVENT_TYPES.keys())
|
|
elif path == "direction":
|
|
ok = s.isin(("pos", "neg", "neutral"))
|
|
elif path == "confidence":
|
|
ok = pd.to_numeric(s, errors="coerce").between(0.0, 1.0)
|
|
elif path == "stock_code":
|
|
ok = s.astype(str).str.match(_CODE_RE)
|
|
elif path == "event_date":
|
|
ok = s.astype(str).str.match(_DATE_RE)
|
|
else:
|
|
ok = s.notna() & (s.astype(str).str.len() > 0)
|
|
return float(ok.sum()) / len(s)
|
|
|
|
|
|
def _read_window(root, domain, days):
|
|
"""窗口读取; 失败返 None(由逐 check try/except 记 error, 不炸全局)。"""
|
|
since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
|
|
try:
|
|
return pr.read_partitions(root, domain, start=since)
|
|
except Exception as e: # noqa: BLE001 corrupt 隔离
|
|
log.warning("域 %s 窗口读取失败: %s", domain, e)
|
|
return None
|
|
|
|
|
|
def run_gate(root, profile="default", days=WINDOW_DAYS) -> dict:
|
|
"""跑三层 → report dict。crit 判定只返字段不 raise(CLI 负责 exit 映射,
|
|
库内消费方(夜批收尾)按 verdict 自决——raise 形态挂在 CLI 层)。"""
|
|
enabled = PROFILES.get(profile)
|
|
if enabled is None:
|
|
raise ValueError(f"未知 profile: {profile}")
|
|
checks: list[dict] = []
|
|
crit = warn = 0
|
|
|
|
# ---------- L1 存在性 ----------
|
|
for domain in enabled:
|
|
try:
|
|
dates = pr.list_dates(root, domain)
|
|
ok = bool(dates)
|
|
detail = f"partitions={len(dates)}"
|
|
except Exception as e: # noqa: BLE001 单查隔离
|
|
ok, detail = False, f"error={e}"
|
|
if not ok:
|
|
crit += 1
|
|
checks.append({"layer": "L1", "name": f"{domain}.exists",
|
|
"ok": ok, "critical": True, "detail": detail})
|
|
|
|
# ---------- L2 字段级 ----------
|
|
frames = {d: _read_window(root, d, days) for d in enabled}
|
|
passed = total = 0
|
|
for dim, path, label, critical in CRITICAL_CHECKS:
|
|
if dim not in enabled:
|
|
continue # profile-aware 分母
|
|
total += 1
|
|
try:
|
|
rate = _valid_rate(frames[dim], dim, path)
|
|
ok = rate >= VALID_RATE_MIN
|
|
detail = f"valid_rate={rate:.3f}"
|
|
except Exception as e: # noqa: BLE001
|
|
ok, detail = False, f"error={e}"
|
|
if ok:
|
|
passed += 1
|
|
elif critical:
|
|
crit += 1
|
|
else:
|
|
warn += 1
|
|
checks.append({"layer": "L2", "name": f"{dim}.{path}", "ok": ok,
|
|
"critical": critical, "detail": detail})
|
|
|
|
# ---------- L3 充实度 ----------
|
|
for domain, field, label in ENRICH_DIMS:
|
|
if domain not in enabled:
|
|
continue
|
|
try:
|
|
df = frames[domain]
|
|
rate = (float(df[field].isna().sum()) / len(df)
|
|
if (field in df.columns and len(df)) else 0.0)
|
|
ok = rate <= ENRICH_MAX_NULL_RATE
|
|
detail = f"null_rate={rate:.3f}"
|
|
except Exception as e: # noqa: BLE001
|
|
ok, detail = False, f"error={e}"
|
|
if not ok:
|
|
warn += 1
|
|
checks.append({"layer": "L3", "name": f"{domain}.{field}", "ok": ok,
|
|
"critical": False, "detail": detail})
|
|
|
|
pct = round(passed / total * 100.0, 1) if total else 100.0
|
|
if pct < CRIT_PCT:
|
|
crit += 1
|
|
elif pct < WARN_PCT:
|
|
warn += 1
|
|
verdict = "critical" if crit else ("warning" if warn else "green")
|
|
return {"profile": profile, "window_days": days,
|
|
"coverage_pct": pct, "crit": crit, "warn": warn,
|
|
"verdict": verdict, "checks_run": checks}
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--root", required=True, help="corpus 族根(如 /volume1/stock/corpus)")
|
|
ap.add_argument("--profile", default="default",
|
|
choices=sorted(PROFILES), help="启用域集(分母来源)")
|
|
ap.add_argument("--days", type=int, default=WINDOW_DAYS)
|
|
ap.add_argument("--out", help="报告 JSON 落盘路径(可选)")
|
|
args = ap.parse_args(argv)
|
|
|
|
if os.environ.get("SANGUO_SKIP_INTEGRITY") == "1":
|
|
report = {"skipped": True, "profile": args.profile}
|
|
print(json.dumps(report, ensure_ascii=False))
|
|
return 0
|
|
report = run_gate(args.root, profile=args.profile, days=args.days)
|
|
text = json.dumps(report, ensure_ascii=False)
|
|
print(text)
|
|
if args.out:
|
|
with open(args.out, "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
if report["crit"]:
|
|
log.error("完整性门 critical: crit=%d warn=%d coverage=%s%%",
|
|
report["crit"], report["warn"], report["coverage_pct"])
|
|
return 1
|
|
if report["warn"]:
|
|
return 2
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|