# -*- coding: utf-8 -*- """P3-S.21.1 — FEATURE CONTRIBUTION AUDIT (frozen Logistic, walk-forward folds). Descriptive attribution of the P3-S20 pre-registered walk-forward LOGISTIC result to the FROZEN 12-causal-feature schema. This is a DESCRIPTIVE audit of the EXISTING fitted decision function ONLY: - reproduces the exact frozen P3-S20 pipeline per fold (StandardScaler fit on train only + LogisticRegression(C=1.0, max_iter=5000, seed 42)); - VERIFIES exact reproduction against the committed P3-S20 OOS predictions CSV and summary JSON before trusting any coefficient; - reports raw + standardized coefficients, signs, ranks per fold; - groups the 12 features into PREDEFINED semantic families; - aggregates group-level contribution (deterministic); - records descriptive per-fold feature distributions; - re-verifies H4/M30/direction structural collinearity (known result); - audits zone/ATR geometry semantics; - classifies the contribution pattern per a deterministic rule. NOT ALLOWED / NOT DONE here: calibration analysis, new models (tree/boost/MLP/ LSTM/Informer/regime), feature selection/removal, ablation, TP/SL/horizon/label changes, Candidate Setup/MQL5/FEATURE_CONTRACT changes, external data, hyperparameter tuning. Nothing is causal; language is "contribution to the fitted linear decision function". Guard discipline: this file is NOT named spec_tests_* and must avoid the frozen parity-absence guard tokens (the zone-type runtime identifier family frozen by the P3-S.4/P3-S.5 parity-absence guards). It does not reference any such runtime identifier or detector name; the concrete provenance lives in the report document only. """ import csv import datetime as dt import hashlib import json import os import subprocess import sys import numpy as np from sklearn.linear_model import LogisticRegression from sklearn.metrics import roc_auc_score from sklearn.preprocessing import StandardScaler 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 import walk_forward as WF # noqa: E402 SEED = 42 # reproduction match tolerance for OOS predicted probabilities vs committed CSV REPRO_TOL = 1e-9 # deterministic classification thresholds (documented; NOT tuned to results) NONZERO_EPS = 1e-12 # |std coef| below this is treated as zero MAG_THRESH = 0.05 # a feature with mean|std coef| < this -> negligible RANK_FOCUS = 4 # consider top-N by |std coef| for "focus" features # --------------------------------------------------------------------------- # PREDEFINED SEMANTIC FAMILIES (frozen per section 9, assigned by semantics per # the actual feature definitions in prepare_dataset.FEATURE_COLS; NOT invented # from observed results). # # CONTEXT : chain/gate direction (H4/M30 context gates). # LIQUIDITY_STRUCTURE: sweep / CHoCH event timing. # ZONE : zone-type, zone age, zone geometry (width, entry # position, signed distance to zone center). # SCALE : ATR scale at entry (price-scaled magnitude). # ENTRY_MOMENTUM : EMPTY in the frozen 12 (no M15 displacement / entry- # condition column among the frozen features). # MICRO : EMPTY in the frozen 12 (no M3 confirmation column). # --------------------------------------------------------------------------- FEATURE_FAMILIES = { "direction": "CONTEXT", "h4_gate": "CONTEXT", "m30_gate": "CONTEXT", "sweep_age_bars": "LIQUIDITY_STRUCTURE", "choch_age_bars": "LIQUIDITY_STRUCTURE", "choch_latency_bars": "LIQUIDITY_STRUCTURE", "zone_type_code": "ZONE", "zone_age_bars": "ZONE", "zone_width_atr": "ZONE", "price_in_zone_offset": "ZONE", "dist_to_zone_center_atr": "ZONE", "atr_at_entry": "SCALE", } FAMILY_ORDER = ["CONTEXT", "LIQUIDITY_STRUCTURE", "ZONE", "SCALE", "ENTRY_MOMENTUM", "MICRO"] # metadata for the inventory (static, semantic) FEATURE_META = { "direction": ("frozen chain direction", "M15", "signed", "+1/-1"), "h4_gate": ("H4 narrative gate at entry", "H4", "signed", "+1/-1/0"), "m30_gate": ("M30 context gate at entry", "M30", "signed", "+1/-1/0"), "sweep_age_bars": ("bars since sweep onset", "M15", "bars", "int >=0"), "choch_age_bars": ("bars since CHoCH onset", "M15", "bars", "int >=0"), "choch_latency_bars": ("CHoCH onset - sweep onset", "M15", "bars", "int >=0"), "zone_type_code": ("zone type code (1=OB,0=gap)", "M15", "code", "0/1"), "zone_age_bars": ("bars since zone formation", "M15", "bars", "int >=0"), "zone_width_atr": ("zone height / ATR at entry", "M15", "ratio", "real >=0"), "price_in_zone_offset": ("(entry-bot)/width within zone", "M15", "ratio", "[0,1]"), "dist_to_zone_center_atr":("(entry-mid)/width signed dist", "M15", "ratio", "real"), "atr_at_entry": ("ATR(14) at entry", "M15", "price", "real >0"), } def _sha16(obj): return hashlib.sha256(json.dumps(obj, sort_keys=True, default=str).encode()).hexdigest()[:16] def git_head(): try: return subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() except Exception: # noqa: BLE001 return "unknown" def _stat(vals): a = np.asarray(vals, dtype=float) n = a.size q = np.nanpercentile(a, [5, 25, 50, 75, 95]) if n else [np.nan] * 5 return { "mean": float(np.mean(a)) if n else float("nan"), "median": float(np.median(a)) if n else float("nan"), "std": float(np.std(a)) if n else float("nan"), "min": float(np.min(a)) if n else float("nan"), "max": float(np.max(a)) if n else float("nan"), "q05": float(q[0]), "q25": float(q[1]), "q50": float(q[2]), "q75": float(q[3]), "q95": float(q[4]), "n": int(n), "missing": int(np.isnan(a).sum()), } def load_binary(): """Chronologically sorted 571 WIN/LOSS binary lead rows (frozen).""" ctx, rows, bin_rows = WF.load_binary() return ctx, rows, bin_rows def fit_logistic(Xtr, ytr): """Exact frozen P3-S20 configuration (standalone refit for coefficients).""" scaler = StandardScaler().fit(Xtr) # train only Xtr_s = scaler.transform(Xtr) clf = LogisticRegression(C=1.0, max_iter=5000, random_state=SEED) clf.fit(Xtr_s, ytr) return scaler, clf def verify_reproduction(bin_rows, oos_prob, folds, setup_ids, outcome): """Compare freshly fitted OOS WIN-probabilities to the committed P3-S20 CSV. Returns (ok, max_abs_diff). Order must match the committed file (the same fold-by-fold OOS row order produced by walk_forward.py).""" with open(os.path.join(OUT, "p3_s20_oos_predictions.csv"), encoding="utf-8") as f: ref = list(csv.DictReader(f)) assert len(ref) == len(oos_prob), "committed OOS row count mismatch" maxdiff = 0.0 for row, r_ in zip(ref, zip(folds, setup_ids, outcome, oos_prob)): f_, sid, out, p = r_ maxdiff = max(maxdiff, round(abs(float(row["pred_logistic_win_prob"]) - p), 15)) if int(float(row["fold"])) != int(f_) or int(row["setup_id"]) != int(sid): return False, maxdiff if row["outcome"] != out: return False, maxdiff return maxdiff <= REPRO_TOL, maxdiff def feature_inventory(ctx, rows, bin_rows): """Frozen 12-feature inventory + population statistics over binary rows.""" inv = [] # (n_rows, n_features): one row per setup, one column per feature X = np.asarray([[r["feature_" + k] for k in PD.FEATURE_COLS] for r in bin_rows], dtype=float) for j, k in enumerate(PD.FEATURE_COLS): src, tfm, dt_, uniq_desc = FEATURE_META[k] vals = X[:, j] s = _stat(vals) uniq = int(len(np.unique(vals))) const = bool(uniq == 1) inv.append({ "feature": k, "source": src, "timeframe": tfm, "data_type": dt_, "semantic_family": FEATURE_FAMILIES[k], "mean": s["mean"], "median": s["median"], "std": s["std"], "min": s["min"], "max": s["max"], "missing_rate": round(s["missing"] / max(s["n"], 1), 8), "unique_count": uniq, "constant": const, "units": uniq_desc, }) return inv def fold_coefficients(bin_rows, X, y): """Per-fold frozen-logistic raw/std coefficients, signs, ranks.""" results = [] parts = WF.fold_parts(bin_rows) oos_prob = [] oos_fold, oos_sid, oos_out = [], [], [] for fi, (train_idx, oos_idx, _gap) in enumerate(parts): Xtr, ytr = X[np.asarray(train_idx)], y[np.asarray(train_idx)] Xoos = X[np.asarray(oos_idx)] scaler, clf = fit_logistic(Xtr, ytr) poos = clf.predict_proba(scaler.transform(Xoos))[:, 1] oos_prob.extend(float(p) for p in poos) for g in oos_idx: oos_fold.append(fi + 1) oos_sid.append(int(bin_rows[g]["setup_id"])) oos_out.append(bin_rows[g]["outcome"]) std = clf.coef_[0] # coefficients on standardized input raw = std / scaler.scale_ # back-transform to original units for j, k in enumerate(PD.FEATURE_COLS): results.append({ "fold": fi + 1, "feature": k, "raw_coef": float(raw[j]), "std_coef": float(std[j]), "sign": 1.0 if std[j] > NONZERO_EPS else ( -1.0 if std[j] < -NONZERO_EPS else 0.0), "abs_std_coef": float(abs(std[j])), "family": FEATURE_FAMILIES[k], }) # rank within fold by |std coef| for j, k in enumerate(PD.FEATURE_COLS): for r_ in results: if r_["fold"] == fi + 1 and r_["feature"] == k: r_["rank_in_fold"] = 1 + sum( 1 for rr in results if rr["fold"] == fi + 1 and rr["abs_std_coef"] > r_["abs_std_coef"]) return results, oos_prob, oos_fold, oos_sid, oos_out, parts def temporal_class(feature_results): """Deterministic per-feature temporal-consistency classification. Uses the std coefficient across the 3 P3-S20 folds: INCONCLUSIVE : mean|std coef| < MAG_THRESH (negligible contribution). FOLD-SPECIFIC : sign flips across folds (has both + and - non-zero). VARIABLE : same sign in all folds but large magnitude spread (max/min |std| > 2.5). STABLE : same sign in all folds and magnitude spread <= 2.5. """ names = [k for k in PD.FEATURE_COLS] out = {} for k in names: cs = [r_["std_coef"] for r_ in feature_results if r_["feature"] == k] cs = [c for c in cs if abs(c) > NONZERO_EPS] mean_abs = float(np.mean([abs(c) for c in cs])) if cs else 0.0 if not cs or mean_abs < MAG_THRESH: out[k] = "INCONCLUSIVE" continue signs = set((1 if c > 0 else -1) for c in cs) if len(signs) > 1: out[k] = "FOLD-SPECIFIC" continue mn, mx = min(abs(c) for c in cs), max(abs(c) for c in cs) spread = (mx / mn) if mn > 0 else float("inf") out[k] = ("VARIABLE" if spread > 2.5 else "STABLE") return out def _fold_rank(feature_results, fold, feature): """Rank (1 = largest |std coef|) of a feature within one fold.""" rows = [r_ for r_ in feature_results if r_["fold"] == fold and r_["feature"] == feature] if not rows: return None return 1 + sum(1 for rr in feature_results if rr["fold"] == fold and rr["abs_std_coef"] > rows[0]["abs_std_coef"]) def focus_features(feature_results): """Features that are FOCUS: mean|std coef| >= MAG_THRESH AND same sign in all 3 folds AND ranked in the top RANK_FOCUS by |std coef| in >= 2 folds.""" focus = [] for k in PD.FEATURE_COLS: cs = [r_["std_coef"] for r_ in feature_results if r_["feature"] == k] csnz = [c for c in cs if abs(c) > NONZERO_EPS] if len(csnz) != 3: continue if float(np.mean([abs(c) for c in csnz])) < MAG_THRESH: continue signs = set((1 if c > 0 else -1) for c in csnz) if len(signs) != 1: continue top_in_folds = sum( 1 for fi_ in (1, 2, 3) if (_fold_rank(feature_results, fi_, k) or 99) <= RANK_FOCUS) if top_in_folds >= 2: focus.append(k) return focus def _mean_abs_std(feature_results, k): """Mean |std coef| over the 3 folds for one feature (0.0 when none).""" vals = [abs(r_["std_coef"]) for r_ in feature_results if r_["feature"] == k and abs(r_["std_coef"]) > NONZERO_EPS] return float(np.mean(vals)) if vals else 0.0 def _group_dominant(feature_results, members): """Dominant member = largest mean |std coef| (NaN-safe).""" return max(members, key=lambda k_: _mean_abs_std(feature_results, k_)) def group_contributions(feature_results): """Deterministic group-level aggregation. Combined = mean over folds( sum of |std coef| over the group's features ). Sign consistency = all non-zero member coefficients share one sign per (fold-independent) pooled view AND the dominant member's sign is stable. """ groups = {g: [] for g in FAMILY_ORDER} for k in PD.FEATURE_COLS: groups.setdefault(FEATURE_FAMILIES[k], []).append(k) rows = [] for g in FAMILY_ORDER: members = groups.get(g, []) if not members: rows.append({"group": g, "n_members": 0, "members": "", "mean_combined_abs_std": 0.0, "sum_abs_std_f1": 0.0, "sum_abs_std_f2": 0.0, "sum_abs_std_f3": 0.0, "sign_stable": None, "dominant_feature": None, "fold_consistency": "n/a"}) continue sums = {fi_: 0.0 for fi_ in (1, 2, 3)} for fi_ in (1, 2, 3): for k in members: sums[fi_] += abs(next(r_["std_coef"] for r_ in feature_results if r_["fold"] == fi_ and r_["feature"] == k)) mean_c = float(np.mean(list(sums.values()))) # dominant member = largest mean |std coef| (NaN-safe) dom = _group_dominant(feature_results, members) dom_signs = set(r_["sign"] for r_ in feature_results if r_["feature"] == dom and r_["sign"] != 0) sign_stable = bool(len(dom_signs) == 1) if dom_signs else None # fold consistency: spread of the per-fold summed |std coef| spread = (max(sums.values()) / min(sums.values())) if min(list(sums.values())) > 0 else float("inf") fold_cons = "STABLE" if spread <= 2.5 else ("VARIABLE" if spread < float("inf") else "FOLD-SPECIFIC") if dom_signs: fold_cons = "STABLE" if (sign_stable and spread <= 2.5) else ( "VARIABLE" if sign_stable else "FOLD-SPECIFIC") rows.append({ "group": g, "n_members": len(members), "members": ";".join(members), "mean_combined_abs_std": round(mean_c, 6), "sum_abs_std_f1": round(sums[1], 6), "sum_abs_std_f2": round(sums[2], 6), "sum_abs_std_f3": round(sums[3], 6), "sign_stable": sign_stable, "dominant_feature": dom, "fold_consistency": fold_cons, }) return rows def structural_redundancy(bin_rows): """Descriptive audit of by-construction / exact structural redundancy. Reports features that are EXACTLY equal or affine in the verified population (P3-S19 known result + affine zone-offset identity). This is a COMPOSITION/STRUCTURE fact, NOT a bug and NOT a basis for removal here. """ n = len(bin_rows) d = [r["feature_direction"] for r in bin_rows] h = [r["feature_h4_gate"] for r in bin_rows] m = [r["feature_m30_gate"] for r in bin_rows] poz = np.asarray([r["feature_price_in_zone_offset"] for r in bin_rows], dtype=float) dist = np.asarray([r["feature_dist_to_zone_center_atr"] for r in bin_rows], dtype=float) sw = np.asarray([r["feature_sweep_age_bars"] for r in bin_rows], dtype=float) ch = np.asarray([r["feature_choch_age_bars"] for r in bin_rows], dtype=float) lat = np.asarray([r["feature_choch_latency_bars"] for r in bin_rows], dtype=float) ztype = [r["feature_zone_type_code"] for r in bin_rows] # exact equality: direction == h4 == m30 on every row gates_eq = all(a == b == c for a, b, c in zip(d, h, m)) # affine identity: dist = poz - 0.5 (zone-width centering) -> r=1 exactly affine_ok = bool(np.allclose(dist, poz - 0.5, rtol=0, atol=1e-9)) # functional identity: latency = sweep_age - choch_age (by CHoCH-after-sweep) func_ok = bool(np.allclose(lat, sw - ch, rtol=0, atol=1e-9)) ztype_const = len(set(ztype)) == 1 return { "n": n, "direction_equals_h4_equals_m30_every_row": gates_eq, "dist_eq_poz_minus_half_every_row": bool(affine_ok), "choch_latency_eq_sweep_minus_choch_every_row": bool(func_ok), "zone_type_code_constant": bool(ztype_const), "readiness_note": "These are EXACT structural/affine identities from " "the verified canonical chain (F3 gates must be " "direction-compatible; zone-center offset = in-zone " "offset - 0.5 width; CHoCH latency = sweep age - " "CHoCH age). They explain identical standardized " "columns/coefficients; they do NOT make any feature " "leaky or invalid, and this audit does NOT remove " "any feature.", } def fold_feature_stats(bin_rows, parts): """Descriptive per-fold (training rows) feature distributions.""" stat_rows = [] for fi_, (train_idx, _oos, _g) in enumerate(parts): for k in PD.FEATURE_COLS: vals = np.asarray([bin_rows[i]["feature_" + k] for i in train_idx], dtype=float) s = _stat(vals) stat_rows.append({"fold": fi_ + 1, "feature": k, "mean": s["mean"], "median": s["median"], "std": s["std"], "min": s["min"], "max": s["max"], "q05": s["q05"], "q25": s["q25"], "q50": s["q50"], "q75": s["q75"], "q95": s["q95"], "missing": s["missing"], "n_train": s["n"]}) return stat_rows def collinearity_audit(bin_rows): """Verify h4_gate == m30_gate == direction structurally (all binary rows).""" d = [r["feature_direction"] for r in bin_rows] h = [r["feature_h4_gate"] for r in bin_rows] m = [r["feature_m30_gate"] for r in bin_rows] eq_h = all(a == b for a, b in zip(d, h)) eq_m = all(a == b for a, b in zip(d, m)) return { "n": len(bin_rows), "h4_equals_direction": eq_h, "m30_equals_direction": eq_m, "direction_unique": sorted(set(d)), "unique_combinations": sorted({(a, b, c) for a, b, c in zip(d, h, m)}), "note": "STRUCTURAL redundancy by construction (context gates must be " "direction-compatible for any Candidate Setup); NOT a bug.", } def zone_atr_mechanism(): """Exact zone/ATR feature semantics from prepare_dataset (availability).""" return { "zone_width_atr": { "definition": "(zone_top - zone_bot) / ATR(entry); 0 if degenerate", "units": "ATR-ratio (width in ATR multiples)", "availability": "entry"}, "price_in_zone_offset": { "definition": "(entry_price - zone_bot) / zone_width in [0,1]", "units": "unit interval (position within zone)", "availability": "entry"}, "dist_to_zone_center_atr": { "definition": "(entry_price - (top+bot)/2) / zone_width (signed)", "units": "zone-width multiples", "availability": "entry"}, "atr_at_entry": { "definition": "rolling 14-bar M15 ATR ending at the entry bar", "units": "price units", "availability": "entry"}, "dependence_on_atr_is_not_leakage": True, "scale_note": "ATR dependence is contextual scaling, not target " "leakage: ATR is known at entry; features carry no " "post-entry or outcome-derived value (P3-S19).", } def classify_final(group_rows, feature_results, coll, focus): """Deterministic final classification (A..G) from the audit evidence.""" by = {r_["group"]: r_ for r_ in group_rows} c = by.get("CONTEXT", {}) liq = by.get("LIQUIDITY_STRUCTURE", {}) zone = by.get("ZONE", {}) sc = by.get("SCALE", {}) # contribution mass by family masses = {} for g in ("CONTEXT", "LIQUIDITY_STRUCTURE", "ZONE", "SCALE"): masses[g] = by[g]["mean_combined_abs_std"] if g in by else 0.0 # focus counts per family foc = {g: 0 for g in masses} for k in focus: foc[FEATURE_FAMILIES[k]] += 1 focus_total = sum(foc.values()) verdicts = [] if focus_total == 0: # none cleared the focus bar -> still check for any consistent sign mass stable_fams = [g for g in masses if by[g].get("sign_stable") is True] if not stable_fams: return "F_NO_CONSISTENT_FEATURE_CONTRIBUTION", masses, foc if zone.get("sign_stable") and zone["mean_combined_abs_std"] > 0 \ and zone["mean_combined_abs_std"] >= max( [masses[g] for g in masses]): return "A_STABLE_ZONE_SCALE_CONTRIBUTION", masses, foc return "G_INCONCLUSIVE", masses, foc dominant = max(masses, key=lambda g: masses[g]) # require sign stability of at least the dominant family to call it STABLE dom_stable = by[dominant].get("sign_stable") is True multiple = sum(1 for g in masses if foc[g] > 0) > 1 fold_ok = all(by[g]["fold_consistency"] != "FOLD-SPECIFIC" for g in masses if foc[g] > 0) if not dom_stable or not fold_ok: return "E_FOLD_SPECIFIC_CONTRIBUTION", masses, foc if multiple and dominant in ("ZONE", "SCALE"): return "D_MIXED_CONTRIBUTION", masses, foc if dominant == "ZONE" or dominant == "SCALE": return "A_STABLE_ZONE_SCALE_CONTRIBUTION", masses, foc if dominant == "LIQUIDITY_STRUCTURE": return "B_STABLE_EVENT_TIMING_CONTRIBUTION", masses, foc if dominant == "CONTEXT": return "C_STABLE_CONTEXT_CONTRIBUTION", masses, foc return "D_MIXED_CONTRIBUTION", masses, foc def _sign_pattern(feature_results, k): """Human-readable sign pattern across folds (all +; all -; mixed; zero).""" signs = set(r_["sign"] for r_ in feature_results if r_["feature"] == k and r_["sign"] != 0) if not signs: return "zero" if signs == {1.0}: return "all_positive" if signs == {-1.0}: return "all_negative" return "mixed" def write_csv(fn, rows, fieldnames): with open(os.path.join(OUT, fn), "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") w.writeheader() for r_ in rows: w.writerow(r_) def main(): os.makedirs(OUT, exist_ok=True) ctx, rows, bin_rows = load_binary() bars = np.asarray([r["creation_bar"] for r in bin_rows], dtype=int) y = np.asarray([1.0 if r["outcome"] == "WIN" else 0.0 for r in bin_rows], dtype=float) X = np.asarray([[r["feature_" + k] for k in PD.FEATURE_COLS] for r in bin_rows], dtype=float) # ---- coefficient refit (exact frozen config) + reproduction verify ---- feat_results, oos_prob, oos_fold, oos_sid, oos_out, parts = ( fold_coefficients(bin_rows, X, y)) repro_ok, maxdiff = verify_reproduction(bin_rows, oos_prob, oos_fold, oos_sid, oos_out) # ---- pooled OOS ROC-AUC check vs committed summary ---- with open(os.path.join(OUT, "p3_s20_summary.json"), encoding="utf-8") as f: s20 = json.load(f) pooled_ref = s20["pooled_oos_logistic"]["roc_auc"] # recompute pooled ROC-AUC from our reproduction probabilities pooled_ours = float(roc_auc_score(np.asarray( [1.0 if o_ == "WIN" else 0.0 for o_ in oos_out]), np.asarray(oos_prob))) inventory = feature_inventory(ctx, rows, bin_rows) tclass = temporal_class(feat_results) focus = focus_features(feat_results) coll = collinearity_audit(bin_rows) redun = structural_redundancy(bin_rows) gm = zone_atr_mechanism() grp = group_contributions(feat_results) fstats = fold_feature_stats(bin_rows, parts) final, masses, foc = classify_final(grp, feat_results, coll, focus) # scale dependence on the scaler fit (for reporting standardization) scaler_note = ("StandardScaler z=(x-mean)/std; mean/std fit on THAT fold's " "TRAINING rows only; std_coef = coefficient on the " "standardized input; raw_coef = std_coef / scaler.scale_ " "(back-transformed to original feature units).") feats_all = [[r["feature_" + k] for k in PD.FEATURE_COLS] for r in rows] labels = [r["outcome"] for r in rows] summary = { "session": "P3-S21.1 feature contribution audit", "model": {"primary": "LogisticRegression(C=1.0, L2, max_iter=5000)", "scaler": "StandardScaler fit on train only per fold", "seed": SEED}, "reproduction": { "verified_against": "ml/p3/baseline/output/p3_s20_oos_predictions.csv", "exact": bool(repro_ok), "max_abs_prob_diff": maxdiff, "tol": REPRO_TOL, "pooled_oos_roc_recomputed": pooled_ours, "pooled_oos_roc_committed": pooled_ref, }, "standardization": scaler_note, "population": { "all_setup": len(rows), "leads": int(sum(1 for r in rows if r["lead"])), "binary_fit": len(bin_rows), "win": int((y == 1).sum()), "loss": int((y == 0).sum()), }, "hashes": { "feature_sha16": _sha16([PD.FEATURE_COLS, feats_all]), "label_sha16": _sha16(labels), "schema_sha16": _sha16([PD.CONTRACT_VERSION, PD.FEATURE_VERSION, PD.FEATURE_COLS]), }, "folds": WF.FOLD_WINDOWS, "purge_gap_bars": [int(parts[i][2]) for i in range(len(parts))], "coefficient_method": ("refit frozen logistic per fold; verified OOS " "probabilities == committed P3-S20 predictions " "within %g" % REPRO_TOL), "temporal_classification": tclass, "focus_features": focus, "family_contribution_mass": masses, "focus_by_family": foc, "collinearity": coll, "structural_redundancy": redun, "zone_atr_semantics": gm, "final_classification": final, "git_commit": git_head(), "generated_utc": dt.datetime.now(dt.timezone.utc).isoformat(), } # ---- outputs ---- with open(os.path.join(OUT, "p3_s21_1_summary.json"), "w", encoding="utf-8") as f: json.dump(summary, f, indent=2, default=str) write_csv("p3_s21_1_feature_inventory.csv", inventory, ["feature", "source", "timeframe", "data_type", "semantic_family", "mean", "median", "std", "min", "max", "missing_rate", "unique_count", "constant", "units"]) write_csv("p3_s21_1_feature_coefficients.csv", [{"feature": k, "family": FEATURE_FAMILIES[k], "std_coef_f1": float(next(r_["std_coef"] for r_ in feat_results if r_["feature"] == k and r_["fold"] == 1)), "std_coef_f2": float(next(r_["std_coef"] for r_ in feat_results if r_["feature"] == k and r_["fold"] == 2)), "std_coef_f3": float(next(r_["std_coef"] for r_ in feat_results if r_["feature"] == k and r_["fold"] == 3)), "raw_coef_f1": float(next(r_["raw_coef"] for r_ in feat_results if r_["feature"] == k and r_["fold"] == 1)), "mean_abs_std": round(float(np.mean([ abs(next(r_["std_coef"] for r_ in feat_results if r_["feature"] == k and r_["fold"] == fi)) for fi in (1, 2, 3)])), 6), "sign_pattern": _sign_pattern(feat_results, k), "temporal_class": tclass[k]} for k in PD.FEATURE_COLS], ["feature", "family", "std_coef_f1", "std_coef_f2", "std_coef_f3", "raw_coef_f1", "mean_abs_std", "sign_pattern", "temporal_class"]) write_csv("p3_s21_1_fold_coefficients.csv", [{"fold": r_["fold"], "feature": r_["feature"], "family": r_["family"], "raw_coef": r_["raw_coef"], "std_coef": r_["std_coef"], "sign": r_["sign"], "abs_std_coef": r_["abs_std_coef"], "rank_in_fold": r_["rank_in_fold"]} for r_ in feat_results], ["fold", "feature", "family", "raw_coef", "std_coef", "sign", "abs_std_coef", "rank_in_fold"]) write_csv("p3_s21_1_group_contributions.csv", grp, ["group", "n_members", "members", "mean_combined_abs_std", "sum_abs_std_f1", "sum_abs_std_f2", "sum_abs_std_f3", "sign_stable", "dominant_feature", "fold_consistency"]) write_csv("p3_s21_1_fold_feature_stats.csv", fstats, ["fold", "feature", "mean", "median", "std", "min", "max", "q05", "q25", "q50", "q75", "q95", "missing", "n_train"]) print("P3-S21.1 feature audit:") print(" reproduction exact=%s max_abs_prob_diff=%.2e" % (repro_ok, maxdiff)) print(" pooled ROC ours=%.4f committed=%.4f" % (pooled_ours, pooled_ref)) print(" feature_sha16=%s" % summary["hashes"]["feature_sha16"]) print(" final classification: %s" % final) print(" focus features: %s" % (focus or "none")) print(" structural: gates_eq=%s dist_eq_poz_minus_half=%s " "latency_eq=%s zone_type_const=%s" % ( redun["direction_equals_h4_equals_m30_every_row"], redun["dist_eq_poz_minus_half_every_row"], redun["choch_latency_eq_sweep_minus_choch_every_row"], redun["zone_type_code_constant"])) for g in grp: if g["n_members"]: print(" %-22s mean_abs=%.4f sign_stable=%s dom=%s" % ( g["group"], g["mean_combined_abs_std"], g["sign_stable"], g["dominant_feature"])) print("[saved] p3_s21_1_summary.json + 5 CSVs") return 0 if __name__ == "__main__": sys.exit(main())