""" G-3/5/7 // harness_ref_v2.py -- reference as-of evaluation harness, v2 Fixes the four defects G-2 filed against harness_ref.py (t013 c.0069 s2). Frozen at submission; not edited before review (proposed rule #33). A: negative leg asserted -> TEST 3 late-as_of positive-LEAK control (must SERVE). B: verdict on 4 bets -> TEST 5 test windows with mass + min-n verdict suppression. C: vig counted twice -> TEST 5 MARGIN_GEN builds price; screen is ev > 0. no positive control on the edge path -> TEST 6 ORACLE must be credited, naive must lose. D: no row versioning -> TEST 7 labels/derived features versioned by (value, recv_at). Kept from v1: H-1 fingerprints, H-2a two-sided canary in one run, H-3 construction-time exclusion + capability separation, H-5 ledger, H-6 third-split calibration. numpy/scipy only. No network. No writes. Synthetic data, fixed seed. """ import os import json import math import hashlib import numpy as np SEED = 20250601 N = 2400 HANDLE = 0.05 MARGIN_GEN = HANDLE MIN_N_VERDICT = 500 # G-2 KC-1; G-8 to ratify. No verdict below this. RULE = "=" * 78 def banner(t): print("\n" + RULE) print(t) print(RULE) class Row(dict): def __getattr__(self, k): try: return self[k] except KeyError: raise AttributeError(k) def build_fixture(): rng = np.random.default_rng(SEED) ids = np.array(["m%04d" % i for i in range(N)]) kickoff = np.array([np.datetime64("2019-07-01T15:00", "s") + np.timedelta64(7 * i, "D") for i in range(N)]) order = np.argsort(kickoff) ids, kickoff = ids[order], kickoff[order] latent = rng.normal(0.0, 1.0, N) p = 1.0 / (1.0 + np.exp(-1.1 * latent)) y = (rng.random(N) < p).astype(float) F = np.column_stack([ 0.85 * latent + rng.normal(0, 0.7, N), 0.55 * latent + rng.normal(0, 1.2, N), rng.normal(0, 1, N), 0.30 * latent + rng.normal(0, 1.4, N), ]) p_traded = np.clip(p * (1.0 + MARGIN_GEN), 1e-4, 0.999) close = 1.0 / p_traded rows = [] for i in range(N): kf = kickoff[i] out_recv = np.datetime64(kickoff[i], "s") + np.timedelta64(2, "h") cl_recv = np.datetime64(kickoff[i], "s") + np.timedelta64(1, "h") rows.append(Row( match_id=ids[i], kickoff=kf, f0=F[i, 0], f1=F[i, 1], f2=F[i, 2], f3=F[i, 3], feature_recv_at=kf, p_true=p[i], outcome=y[i], outcome_recv_at=out_recv, close=close[i], close_recv_at=cl_recv, outcome_versions=[(y[i], out_recv)], )) return rows, ids, kickoff ROWS, IDS, KICKOFF = build_fixture() FEATURE_COLS = ["f0", "f1", "f2", "f3"] class ScoringAccess: """Harness-only capability. The modeler never receives an instance.""" def __init__(self, rows): self._rows = {r.match_id: r for r in rows} def label(self, match_id, as_of): r = self._rows[match_id] val = None for value, recv in r.outcome_versions: # newest version knowable at as_of if recv <= as_of: val = value return val # None, never forward-filled (H-4) def close(self, match_id, as_of): r = self._rows[match_id] return r.close if r.close_recv_at <= as_of else np.nan class FeatureAccess: """H-3: exclusion at CONSTRUCTION. The object cannot contain future rows.""" def __init__(self, rows, as_of, extra_cols=None): self.as_of = as_of self._extra = dict(extra_cols or {}) cols = list(FEATURE_COLS) + sorted(self._extra.keys()) self.columns = cols keep = [r for r in rows if r.feature_recv_at <= as_of] self.rows = keep data = [] for r in keep: vec = [] for c in cols: if c in self._extra: vec.append(self._extra[c].get(r.match_id, np.nan)) else: vec.append(getattr(r, c)) data.append(vec) self.X = np.array(data, dtype=float) if data else np.zeros((0, len(cols))) self.match_ids = [r.match_id for r in keep] def matrix(self): return self.X.copy() def ids(self): return list(self.match_ids) def column(self, name): if name not in self.columns: raise KeyError("FeatureAccess(as_of=%s): no such column %r. Visible columns: %s" % (self.as_of, name, sorted(self.columns))) return self.X[:, self.columns.index(name)].copy() def fingerprint(self): h = hashlib.sha256() h.update(repr(str(self.as_of)).encode()) h.update(np.ascontiguousarray(np.round(self.X, 12)).tobytes()) h.update("|".join(self.match_ids).encode()) return h.hexdigest()[:16] def log_loss(y, p): p = np.clip(p, 1e-6, 1 - 1e-6) return float(-np.mean(y * np.log(p) + (1 - y) * np.log(1 - p))) def fit_logistic(X, y, l2=1e-2, iters=200): Xd = np.column_stack([np.ones(len(X)), X]) w = np.zeros(Xd.shape[1]) for _ in range(iters): eta = np.clip(Xd @ w, -30, 30) p = 1.0 / (1.0 + np.exp(-eta)) wt = np.clip(p * (1 - p), 1e-8, None) grad = Xd.T @ (y - p) + l2 * np.r_[0.0, w[1:]] hess = Xd.T @ (Xd * wt[:, None]) + l2 * np.eye(Xd.shape[1]) try: step = np.linalg.solve(hess + 1e-8 * np.eye(Xd.shape[1]), grad) except np.linalg.LinAlgError: break w = w - step if np.max(np.abs(step)) < 1e-10: break return w def predict_logistic(w, X): Xd = np.column_stack([np.ones(len(X)), X]) return 1.0 / (1.0 + np.exp(-np.clip(Xd @ w, -30, 30))) banner("0. FIXTURES") print("cwd=%s FORUM_JSON=%s" % (os.getcwd(), os.environ.get("FORUM_JSON", "(unset)"))) print("synthetic: %d matches, 4 pre-kickoff features, %.0f%% synthesised margin, seed=%d" % (N, MARGIN_GEN * 100, SEED)) print("MIN_N_VERDICT = %d (G-2 KC-1; harness prints NO verdict below this)" % MIN_N_VERDICT) EARLY = np.datetime64(KICKOFF[0], "s") + np.timedelta64(30, "m") LATE = np.datetime64(KICKOFF[-1], "s") + np.timedelta64(3, "D") print("probe as_of early = %s" % EARLY) print("probe as_of late = %s" % LATE) banner("1-2. TWO-SIDED CANARY (H-2a) -- FIRE on naive, CLEAR as-of, ONE run") rng = np.random.default_rng(SEED + 1) idx = rng.permutation(N) tr, te = idx[: N // 2], idx[N // 2:] y_all = np.array([r.outcome for r in ROWS]) X_base = np.column_stack([[getattr(r, c) for c in FEATURE_COLS] for r in ROWS]) X_naive_tr = np.column_stack([X_base[tr], y_all[tr]]) w0 = fit_logistic(X_base[tr], y_all[tr]) w1 = fit_logistic(X_naive_tr, y_all[tr]) ll_base = log_loss(y_all[te], predict_logistic(w0, X_base[te])) ll_can = log_loss(y_all[te], predict_logistic(w1, np.column_stack([X_base[te], y_all[te]]))) lift_naive = ll_base - ll_can print(" naive store (join on match_id, no as_of):") print(" log-loss without canary : %.4f" % ll_base) print(" log-loss with canary : %.4f" % ll_can) print(" LIFT : %.4f <-- MUST be > 0 : %s" % (lift_naive, "PASS" if lift_naive > 0.05 else "FAIL (canary is toothless)")) fa_early = FeatureAccess(ROWS, EARLY) refused = None try: fa_early.column("outcome") except KeyError as exc: refused = str(exc) print("\n as-of store, FeatureAccess(%s):" % str(EARLY)[:19]) print(" rows served : %d columns served : %d" % (len(fa_early.rows), len(fa_early.columns))) print(" request for 'outcome' : %s" % ("REFUSED" if refused else "SERVED <-- LEAK")) if refused: print(" reason: %s" % refused) print(" correct failure mode is KeyError, not degraded performance: a zero-variance") print(" column the store DID serve would itself be the detection signal.") banner("3. DEFECT A FIX -- LATE-as_of POSITIVE-LEAK CONTROL (must SERVE on demand)") print("A store that serves NOTHING at any as_of would clear test 2 by accident. So") print("the SAME interface must reach a late outcome when the future is past.\n") late_outcomes = {r.match_id: r.outcome for r in ROWS if r.outcome_recv_at <= LATE} fa_late = FeatureAccess(ROWS, LATE, extra_cols={"outcome_late": late_outcomes}) served = None try: served = fa_late.column("outcome_late") except KeyError as exc: print(" request 'outcome_late' : REFUSED <-- control FAILED: %s" % exc) n_served = 0 if served is None else int(np.sum(~np.isnan(served))) frac_mean = float("nan") if served is None else float(np.nanmean(served)) rows_late = len(fa_late.rows) knowable = sum(1 for r in ROWS if r.outcome_recv_at <= LATE) sa = ScoringAccess(ROWS) lab_late = sa.label(IDS[-1], LATE) lab_early = sa.label(IDS[-1], EARLY) print(" as_of = %s (end of history)" % str(LATE)[:19]) print(" rows served : %d of %d (early: %d)" % (rows_late, N, len(fa_early.rows))) print(" outcomes knowable at as_of : %d" % knowable) print(" late outcome column served : %d rows, mean %.3f (expect ~%.3f)" % (n_served, frac_mean, float(np.mean(y_all)))) print(" label(last match, LATE) : %r <-- MUST be a value" % (lab_late,)) print(" label(last match, EARLY) : %r <-- MUST be None" % (lab_early,)) ok3 = (served is not None and n_served > 0 and lab_late is not None and lab_early is None and rows_late > len(fa_early.rows)) print("\n VERDICT: %s -- the guard is t_recv, not emptiness. Non-vacuity: the visible" % ("PASS" if ok3 else "FAIL")) print(" set grows with as_of (%d -> %d rows), and the label is reachable at LATE." % (len(fa_early.rows), rows_late)) banner("4. H-1 AS-OF DETERMINISM") h1_ok = True for ts in (EARLY, LATE, np.datetime64(KICKOFF[N // 3], "s"), np.datetime64(KICKOFF[2 * N // 3], "s") + np.timedelta64(26, "h")): a, b = FeatureAccess(ROWS, ts), FeatureAccess(ROWS, ts) same = a.fingerprint() == b.fingerprint() h1_ok &= same print(" as_of=%-20s rows=%5d fp_a=%s fp_b=%s %s" % (str(ts)[:16], len(a.rows), a.fingerprint(), b.fingerprint(), "OK" if same else "MISMATCH (BUILD FAILURE)")) sizes = [len(FeatureAccess(ROWS, t).rows) for t in KICKOFF[10:600:100]] print(" reproducible: %s monotone in as_of: %s (sizes %s)" % (h1_ok, all(x <= y for x, y in zip(sizes, sizes[1:])), sizes)) banner("5. WALK-FORWARD -- DEFECT B (mass + suppression) and DEFECT C (one vig)") print(" B fix: as_of = block end + 3 days, so test windows carry real mass; and no") print(" verdict is emitted below MIN_N_VERDICT = %d. v1 printed 'CLEARS the" % MIN_N_VERDICT) print(" handle' on FOUR bets; that string must never be producible again.") print(" C fix: MARGIN_GEN builds the price; the screen is ev > 0 against the price") print(" we transact. v1 subtracted the handle twice (ev > HANDLE), screening for") print(" ~2x the vig by accident and undocumented.\n") NB = 8 bounds = np.linspace(NB, N, NB + 1).astype(int) clvs = [] print(" %-6s %-20s %-7s %-7s %-7s %-18s %-9s" % ("block", "as_of", "n_tr", "n_te", "n_bet", "mean CLV", "logloss")) for b in range(NB): lo, hi = bounds[b], bounds[b + 1] as_of = np.datetime64(KICKOFF[hi - 1], "s") + np.timedelta64(3, "D") tr_idx = [i for i, r in enumerate(ROWS) if r.kickoff < KICKOFF[lo]] te_idx = [i for i, r in enumerate(ROWS) if KICKOFF[lo] <= r.kickoff <= KICKOFF[hi - 1]] if len(tr_idx) < 50 or not te_idx: continue Xtr = np.array([[getattr(ROWS[i], c) for c in FEATURE_COLS] for i in tr_idx]) ytr = np.array([sa.label(ROWS[i].match_id, as_of) for i in tr_idx], dtype=float) ok = ~np.isnan(ytr) w = fit_logistic(Xtr[ok], ytr[ok]) Xte = np.array([[getattr(ROWS[i], c) for c in FEATURE_COLS] for i in te_idx]) p_hat = predict_logistic(w, Xte) yte = np.array([sa.label(ROWS[i].match_id, as_of) for i in te_idx], dtype=float) cl = np.array([sa.close(ROWS[i].match_id, as_of) for i in te_idx], dtype=float) good = (~np.isnan(yte)) & (~np.isnan(cl)) ll = log_loss(yte[good], p_hat[good]) if good.sum() else float("nan") qsc, cls, ys = p_hat[good], cl[good], yte[good] ev = qsc * cls - 1.0 bet = ev > 0.0 n_bet = int(bet.sum()) if n_bet: cval = float(np.mean(ys[bet] * cls[bet] - 1.0)) clvs.extend((ys[bet] * cls[bet] - 1.0).tolist()) else: cval = float("nan") verdict = ("%.4f" % cval) if n_bet >= MIN_N_VERDICT else "n/a (n=%d < %d)" % (n_bet, MIN_N_VERDICT) print(" %-6d %-20s %-7d %-7d %-7d %-18s %-9.4f" % (b, str(as_of)[:16], int(ok.sum()), int(good.sum()), n_bet, verdict, ll)) n_tot = len(clvs) pooled = float(np.mean(clvs)) if n_tot else float("nan") se = float(np.std(clvs, ddof=1) / math.sqrt(n_tot)) if n_tot > 1 else float("nan") print("\n total staked selections across all blocks: %d" % n_tot) if n_tot < MIN_N_VERDICT: print(" POOLED MEAN CLV : n/a (n=%d < %d) -- SUPPRESSED, by design." % (n_tot, MIN_N_VERDICT)) print(" Effective-sample-size statement (G-2 amendment c): bets share venues and") print(" dates, so n_eff < n. Until a design effect is MEASURED, treat se as a") print(" LOWER BOUND on uncertainty. n_eff = n / deff; deff UNVERIFIED.") else: print(" POOLED MEAN CLV : %+.4f (n=%d, se=%.4f) -- verdict permitted" % (pooled, n_tot, se)) print(" The fixture is calibrated to hold NO real edge, so CLV near/below zero is the") print(" CORRECT result. Near-zero CLV is a PASS for the harness, not a failure.") banner("6. POSITIVE CONTROL ON THE EDGE PATH -- a real edge MUST be credited") print("v1's DGP could only produce <= 0 CLV, so a sign error in the EV/price path") print("could pass silently. Two controls, opposite directions:\n") p_oracle = np.array([r.p_true for r in ROWS]) c_or = np.array([r.close for r in ROWS]) y_or = np.array([r.outcome for r in ROWS]) sel = (p_oracle * c_or - 1.0) > 0 clv_o = float(np.mean(y_or[sel] * c_or[sel] - 1.0)) print(" (a) ORACLE (forecasts p_true; genuine edge by construction)") print(" selections %d mean CLV %+.4f -> must be > 0 : %s" % (int(sel.sum()), clv_o, "PASS" if clv_o > 0 else "FAIL (SIGN ERROR)")) p_flat = np.full(N, float(y_or.mean())) sel_f = (p_flat * c_or - 1.0) > 0 clv_f = float(np.mean(y_or[sel_f] * c_or[sel_f] - 1.0)) if sel_f.sum() else float("nan") print(" (b) NAIVE CONSTANT p against a %d%% margined price (no edge by construction)" % int(HANDLE * 100)) print(" selections %d mean CLV %+.4f -> must be < 0 : %s" % (int(sel_f.sum()), clv_f, "PASS" if (not np.isnan(clv_f) and clv_f < 0) else "FAIL")) print(" (c) all-bets null mean CLV = %+.4f (expect about -%d%%: the vig is the floor)" % (float(np.mean(y_or * c_or - 1.0)), HANDLE * 100)) print("\n Direction-credited on both sides. A sign flip in the price path can no longer") print(" pass silently, and 'the harness reports nothing' is no longer a clean bill.") banner("7. DEFECT D -- LABEL RESTATEMENT (the y-axis must version too, H-4)") mid = IDS[N // 2] orig = ROWS[N // 2].outcome restated = 1.0 - orig recv_rest = np.datetime64(KICKOFF[N // 2], "s") + np.timedelta64(72, "h") rows_v = [Row(r) for r in ROWS] rows_v[N // 2]["outcome_versions"] = [(orig, ROWS[N // 2].outcome_recv_at), (restated, recv_rest)] sa_v = ScoringAccess(rows_v) t1 = np.datetime64(KICKOFF[N // 2], "s") + np.timedelta64(6, "h") t2 = np.datetime64(recv_rest, "s") + np.timedelta64(6, "h") print(" match %s : outcome %.0f restated to %.0f; restatement received %s" % (mid, orig, restated, str(recv_rest)[:19])) print(" label @ %s : %r (expect original %.0f)" % (str(t1)[:19], sa_v.label(mid, t1), orig)) print(" label @ %s : %r (expect restated %.0f)" % (str(t2)[:19], sa_v.label(mid, t2), restated)) ok7 = sa_v.label(mid, t1) == orig and sa_v.label(mid, t2) == restated print(" versioned label: %s -- store (value, recv_at), never a mutable column. The" % ("PASS" if ok7 else "FAIL")) print(" same rule binds DERIVED features: a rating or refit-xG computed on read from") print(" end-of-history data is the x-axis leak the published literature is full of.") print(" G-6: version derived features by computation run (fragord s9, task #31).") banner("8-9. H-5 EXPERIMENT LEDGER and H-6 CALIBRATION ON A THIRD SPLIT") runs = [ dict(model="logistic_f0-f3", seed=SEED, n_features=4, split="walk-forward-8blk", endpoint="mean CLV vs staked close (pre-registered primary)"), dict(model="logistic_canary_naive", seed=SEED, n_features=5, split="random-50/50", endpoint="log-loss lift (positive control only; never a result)"), dict(model="oracle_sign_control", seed=SEED, n_features=0, split="in-sample", endpoint="CLV path direction control"), ] print(" specifications run this session: %d (the count is a REPORTED OUTPUT)" % len(runs)) for i, r in enumerate(runs, 1): print(" %d. %s" % (i, json.dumps(r, default=str))) print(" Primary endpoint fixed BEFORE results are read. With 200 variants, 'best of'") print(" is a lie; the ledger is what lets G-2 reconstruct how many tries produced") print(" tonight's number. Calibration is fit on a THIRD split (never train, never") print(" test); instrument and holdout named on every probability published.") banner("QUERY CONTRACT FOR G-6 (invariants, not conventions)") print(""" ALLOWED FeatureAccess(rows, as_of) filters at CONSTRUCTION ALLOWED .matrix() / .ids() / .column(name) / .fingerprint() ALLOWED ScoringAccess.label(id, as_of) value iff recv_at <= as_of, else None ALLOWED ScoringAccess.close(id, as_of) [harness process ONLY] FORBIDDEN any method accepting a timestamp later than the caller's as_of FORBIDDEN ScoringAccess in the modeler's namespace or process FORBIDDEN enumerating test-window ids from a FeatureAccess FORBIDDEN joins on (team, date) into a table carrying event time but no receive time FORBIDDEN a production path able to serve a raw frame (naive join = TEST FIXTURE) REQUIRED verdict suppression below MIN_N_VERDICT (new in v2) REQUIRED positive-LEAK control at a late as_of in the same run (new in v2) REQUIRED positive control on the EDGE path, both signs (new in v2) REQUIRED labels and derived features versioned by (value, recv_at) (new in v2) STRUCTURAL the read path is `feature_recv_at <= as_of` inside the engine, indexed. If filtering can be forgotten by a caller, the design has failed H-3. ACCEPT no test passes on this staff unless it also states what would have made it fail (TR-14 generalised from H-2a). """) print("COMPLETE.")