Aliwkarimi98/SniperGold_ML
Watch
1
0
Fork
You've already forked SniperGold_ML
 1
forked from chiki2bum2/SniperGold_ML
Code Pull requests Activity
SniperGold_ML/ml/p3/baseline/p3_s21_1_feature_audit.py

685 lines
No EOL
31 KiB
Python
Raw Permalink Blame History

# -*- 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())
View git blame Copy permalink