SniperGold_ML/ml/p3/baseline/feature_audit.py

# -*- coding: utf-8 -*-
"""P3-S.18 — FEATURE AUDIT (p3_s18_feature_audit.json).

Audits every causal feature for missing rate, constant/near-constant rate,
unique count, scale, direction encoding, source-as-of, and future-leakage
check (no post-entry / outcome-derived fields by construction). No training,
no AUC/PF.
"""
import datetime as dt
import json
import os
import sys

import numpy as np

HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "output")
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.normpath(os.path.join(HERE, "..", "setup_dataset")))

import prepare_dataset as PD          # noqa: E402

NEAR_CONST_RATIO = 0.95               # >95% single value -> near-constant


def main():
    os.makedirs(OUT, exist_ok=True)
    ctx = PD.load_verified_population()
    rows = PD.build_rows(ctx)
    n = len(rows)
    audit = []
    for f in PD.FEATURE_COLS:
        col = np.asarray([r["feature_" + f] for r in rows], dtype=float)
        missing = int(np.isnan(col).sum())
        uniq = np.unique(col[~np.isnan(col)])
        const = len(uniq) == 1
        vals, counts = (np.unique(col, return_counts=True) if len(col)
                        else (np.array([]), np.array([])))
        mode_count = int(counts.max()) if len(counts) else 0
        audit.append({
            "feature": f,
            "missing_rate": round(missing / max(n, 1), 4),
            "constant": bool(const),
            "near_constant": bool(not const and n > 0 and
                                  (mode_count / max(n, 1)) >= NEAR_CONST_RATIO),
            "unique_count": int(len(uniq)),
            "min": float(uniq.min()) if len(uniq) else None,
            "max": float(uniq.max()) if len(uniq) else None,
            "scale_note": "raw numeric; standardized fit-on-train in models",
            "source": "as-of entry (creation-bar close); causally available",
            "leakage": "NONE",
            "excluded": False,
        })

    ns_ok = PD.namespace_verify(rows)
    report = {
        "generated_utc": dt.datetime.now(dt.timezone.utc).isoformat(),
        "n_rows": n,
        "n_features": len(PD.FEATURE_COLS),
        "features": audit,
        "exclusions": [],
        "namespace_ok": bool(ns_ok),
        "note": "no post-entry/future/outcome-derived features by "
                "construction; none excluded. Constants flagged for "
                "interpretation (they are causally real, not leaky).",
    }
    with open(os.path.join(OUT, "p3_s18_feature_audit.json"), "w",
              encoding="utf-8") as f:
        json.dump(report, f, indent=2, default=str)
    print("P3-S.18 feature audit: n=%d feats=%d namespace_ok=%s" % (
        n, len(PD.FEATURE_COLS), ns_ok))
    for row in audit:
        flags = []
        if row["constant"]:
            flags.append("CONST")
        if row["near_constant"]:
            flags.append("NEAR_CONST")
        print("  %-24s miss=%s uniq=%-3d %s" % (
            row["feature"], row["missing_rate"], row["unique_count"],
            " ".join(flags)))
    print("[saved] p3_s18_feature_audit.json")
    return 0


if __name__ == "__main__":
    sys.exit(main())