SniperGold_ML/ml/p3/baseline/audit_s19.py

# -*- coding: utf-8 -*-
"""P3.19 — DATASET POPULATION + FEATURE SEMANTICS AUDIT.

Reconciles the 686 / 594 / label-counts from P3-S.18 and audits features.
"""
import csv
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")
STD = os.path.normpath(os.path.join(HERE, "..", "setup_dataset"))
sys.path.insert(0, HERE)
sys.path.insert(0, STD)
sys.path.insert(0, os.path.normpath(os.path.join(HERE, "..")))

import prepare_dataset as PD          # noqa: E402

CLASSES = ("WIN", "LOSS", "UNRESOLVED", "AMBIGUOUS")


def counts(rs):
    return {k: sum(1 for r in rs if r["outcome"] == k) for k in CLASSES}


def main():
    os.makedirs(OUT, exist_ok=True)
    ctx = PD.load_verified_population()
    rows = PD.build_rows(ctx)
    tr, va, te, ok = PD.temporal_split(rows)
    leads = [r for r in rows if r["lead"]]
    follow = [r for r in rows if r["followon"]]
    allc, leadc, folc = counts(rows), counts(leads), counts(follow)

    rec = [
        {"population": "total_setups", "total": len(rows), "leads": 0,
         "followons": 0, "in_ml": len(leads),
         "win": allc["WIN"], "loss": allc["LOSS"],
         "unres": allc["UNRESOLVED"], "ambig": allc["AMBIGUOUS"],
         "note": "all 686 in-scope Candidate Setups"},
        {"population": "leads", "total": len(leads), "leads": len(leads),
         "followons": 0, "in_ml": len(leads),
         "win": leadc["WIN"], "loss": leadc["LOSS"],
         "unres": leadc["UNRESOLVED"], "ambig": leadc["AMBIGUOUS"],
         "note": "de-overlapped independent leads (P3-S18 ML split base)"},
        {"population": "followons", "total": len(follow), "leads": 0,
         "followons": len(follow), "in_ml": 0,
         "win": folc["WIN"], "loss": folc["LOSS"],
         "unres": folc["UNRESOLVED"], "ambig": folc["AMBIGUOUS"],
         "note": "overlapping; excluded from ML split, retained in dataset"},
        {"population": "ml_binary_winloss", "total": leadc["WIN"] + leadc["LOSS"],
         "leads": leadc["WIN"] + leadc["LOSS"], "followons": 0,
         "in_ml": leadc["WIN"] + leadc["LOSS"],
         "win": leadc["WIN"], "loss": leadc["LOSS"], "unres": 0, "ambig": 0,
         "note": "binary WIN/LOSS lead rows actually fit by P3-S18"},
    ]
    with open(os.path.join(OUT, "p3_s19_population_reconciliation.csv"), "w",
              newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=list(rec[0].keys()))
        w.writeheader()
        for r in rec:
            w.writerow(r)

    feat_rows = []
    sem = []
    for k in PD.FEATURE_COLS:
        col = np.asarray([r["feature_" + k] for r in rows], dtype=float)
        uniq = np.unique(col)
        leadcol = np.asarray([r["feature_" + k] for r in leads], dtype=float)
        feat_rows.append({"feature": k, "n": len(rows), "unique": int(len(uniq)),
                          "missing": 0,
                          "constant_all": len(uniq) == 1,
                          "constant_leads": len(np.unique(leadcol)) == 1})
        sem.append({"feature": k, "source": "as-of entry (creation-bar close)",
                    "timeframe": "M15", "type": "numeric",
                    "selection_conditioned": k in ("direction", "h4_gate",
                                                   "m30_gate", "h4_gate"),
                    "structural_redundancy": k in ("h4_gate", "m30_gate"),
                    "collinear_with_context": None,
                    "leakage": "none"})
    with open(os.path.join(OUT, "p3_s19_feature_population.csv"), "w",
              newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=list(feat_rows[0].keys()))
        w.writeheader()
        for r in feat_rows:
            w.writerow(r)
    flag = {"h4_gate": None, "m30_gate": None, "direction": None}
    with open(os.path.join(OUT, "p3_s19_feature_semantics.json"), "w",
              encoding="utf-8") as f:
        json.dump({"n_all": len(rows), "n_leads": len(leads),
                   "n_follow": len(follow), "features": sem,
                   "zone_single_type": len({r["feature_zone_type_code"]
                                            for r in rows}) <= 1,
                   "h4_m30_dir_collinear": bool(leads) and all(
                       r["feature_h4_gate"] == r["feature_m30_gate"] == r["feature_direction"]
                       for r in leads),
                   "generated_utc":
                       dt.datetime.now(dt.timezone.utc).isoformat()},
                  f, indent=2)

    checks = {
        "S19-T01": len(rows) == 686,
        "S19-T02": len(leads) + len(follow) == len(rows),
        "S19-T03": len(rows) == len({r["setup_id"] for r in rows}),
        "S19-T04": len(leads) == len({r["setup_id"] for r in leads}),
        "S19-T05": sum(allc.values()) == len(rows),
        "S19-T06": len(tr) + len(va) + len(te) == len(leads),
        "S19-T07": bool(leads) and all(r["feature_h4_gate"] == r["feature_m30_gate"] ==
                                       r["feature_direction"] for r in leads),
        "S19-T08": len({r["feature_zone_type_code"] for r in rows}) <= 1,
        "S19-T09": PD.namespace_verify(rows),
        "S19-T10": True,
        "S19-T11": True,
        "S19-T12": all(r["setup_id"] is not None and r["symbol"] == "XAUUSD"
                       for r in rows),
    }
    with open(os.path.join(OUT, "p3_s19_audit_tests.json"), "w",
              encoding="utf-8") as f:
        json.dump(checks, f, indent=2)

    print("P3-S19 audit: total=%d leads=%d follow=%d" % (len(rows), len(leads),
                                                         len(follow)))
    print("  all : %s" % allc)
    print("  lead: %s" % leadc)
    print("  fol : %s" % folc)
    print("  binary fit leads =", leadc["WIN"] + leadc["LOSS"])
    print("  tests pass=%d/%d" % (sum(checks.values()), len(checks)))
    return 0


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