""" G-3/5/7 // WALK-FORWARD HARNESS -- REFERENCE IMPLEMENTATION OF THE AS-OF INTERFACE Mission 25-001, Task C (t013 c.0015) + the pull-forward VCOS demanded (t013 c.0023 s2). This is the artifact G-6 designs storage against and G-2 audits. It exists to prove one claim: G-2's leakage canary (t013 c.0029 H-2) has TEETH. DESIGN PRINCIPLE (H-3, enforced structurally, not by convention): The capability that builds features CANNOT see future facts. The capability that scores CANNOT build features. Neither can express "give me rows where known_at > as_of". Nothing outside those two objects is available to model code -- including this file's own training loop. If the demo reaches past the interface, the demo is a lie, so it does not. Runs on synthetic data; needs no corpus. No network, no writes, no secrets. Two stores implemented deliberately: NaiveStore -- looks rows up by match_id, ignoring observation time. What you get if you join a frame on match_id. It LEAKS. TEST FIXTURE ONLY. FeatureAccess + ScoringAccess -- the hardened as-of interface. It does not leak. The canary must FIRE on the first and CLEAR on the second, in the same run. If it passes both, the canary itself is broken and no backtest number is admissible. """ import hashlib import math import os from dataclasses import dataclass from datetime import datetime, timedelta, timezone import numpy as np RNG_SEED = 20250601 N_MATCHES = 2400 DIM = 4 HANDLE = 0.05 # ~5% vig, order-of-magnitude (UNVERIFIED) FEATURE_COLUMNS = ("f0", "f1", "f2", "f3") # --------------------------------------------------------------------------- # The four clocks that must never be conflated. # t_event : when the fact became true on the pitch # t_known : when it became true as a record in the world # t_recv : when OUR system observed it <-- the ONLY clock the read path keys on # t_close : when the market settled (future to every as_of we will query) # G-2 t014 c.0030 s1: key on receive time, never vendor publish time. # --------------------------------------------------------------------------- @dataclass(frozen=True) class Row: match_id: int feature: np.ndarray feature_recv_at: datetime outcome: int outcome_recv_at: datetime close_price: float close_recv_at: datetime def iso(dt): return dt.strftime("%Y-%m-%dT%H:%M") def make_dataset(rng): """Synthetic fixtures with a realistic information structure: features are observable pre-match; outcome and close price are NOT knowable pre-match.""" rows = [] for i in range(N_MATCHES): kickoff = datetime(2019, 8, 1, 15, 0, tzinfo=timezone.utc) + timedelta(days=7 * i) strength = rng.normal(0, 1, DIM) feats = strength + rng.normal(0, 0.9, DIM) # noisy pre-match signal true_p = 1.0 / (1.0 + math.exp(-0.75 * strength.sum() / math.sqrt(DIM))) outcome = int(rng.random() < true_p) close = (1.0 / true_p) * (1.0 - HANDLE) # fair price, then vig rows.append(Row( match_id=i, feature=feats, feature_recv_at=kickoff - timedelta(days=10), outcome=outcome, outcome_recv_at=kickoff + timedelta(hours=2), close_price=close, close_recv_at=kickoff - timedelta(minutes=1), )) return rows # --------------------------------------------------------------------------- # Capability separation: two objects, two namespaces. # --------------------------------------------------------------------------- class FeatureAccess: """The ONLY thing model code may hold. Cannot enumerate the test window, cannot read outcomes, cannot read closes, cannot ask about a later instant.""" def __init__(self, rows, as_of): self.as_of = as_of # FILTER AT CONSTRUCTION. The object does not contain the future, so a bug # in model code cannot reach it. Query-time filtering is the weak version. self._visible = sorted((r for r in rows if r.feature_recv_at <= as_of), key=lambda r: r.match_id) self._ids = [r.match_id for r in self._visible] self._X = (np.vstack([r.feature for r in self._visible]) if self._visible else np.zeros((0, DIM))) def matrix(self): return self._X.copy() def ids(self): return list(self._ids) def column(self, name): """There is deliberately no 'outcome' and no 'close' in this namespace. Asking for one is a hard error, not a silent None -- a zero-variance placeholder would let a leaker look clean.""" if name in FEATURE_COLUMNS: return self._X[:, FEATURE_COLUMNS.index(name)].copy() raise KeyError( f"FeatureAccess(as_of={iso(self.as_of)}): no such column {name!r}. " f"Visible columns: {list(FEATURE_COLUMNS)}" ) def fingerprint(self): """H-1 as-of determinism: same as_of => identical hash.""" h = hashlib.sha256() h.update(np.ascontiguousarray(self._X).tobytes()) h.update((",".join(map(str, self._ids))).encode()) return h.hexdigest()[:16] class ScoringAccess: """Held ONLY by the harness. Never handed to model code.""" def __init__(self, rows): self._by_id = {r.match_id: r for r in rows} def label(self, match_id, as_of): """H-4: labels are point-in-time too. Returns None when the outcome was not yet knowable -- NEVER a stale or forward-filled value.""" r = self._by_id[match_id] return float(r.outcome) if r.outcome_recv_at <= as_of else None def close(self, match_id): return self._by_id[match_id].close_price class NaiveStore: """Every backtest I have ever argued about. The modeler holds the whole frame, so 'features' include whatever the join dragged in. No as_of anywhere. TEST FIXTURE: this class must never be reachable from a production path.""" def __init__(self, rows): self._by_id = {r.match_id: r for r in rows} def design_matrix(self, ids, extra_cols=None): X = [self._by_id[i].feature for i in ids] if extra_cols: X = [np.concatenate([x, c]) for x, c in zip(X, extra_cols)] return np.vstack(X) def labels(self, ids): return np.array([float(self._by_id[i].outcome) for i in ids]) # the leak # --------------------------------------------------------------------------- # Minimal deterministic logistic fit (numpy only). # --------------------------------------------------------------------------- def fit_logistic(X, y, ridge=1e-3, steps=400, lr=0.4): X = np.asarray(X, float) mu, sd = X.mean(0), X.std(0) + 1e-9 Z = (X - mu) / sd w, b = np.zeros(Z.shape[1]), 0.0 for _ in range(steps): p = 1.0 / (1.0 + np.exp(-(Z @ w + b))) err = p - y w -= lr * (Z.T @ err / len(y) + ridge * w) b -= lr * err.mean() return mu, sd, w, b def predict(m, X): mu, sd, w, b = m return 1.0 / (1.0 + np.exp(-(((np.asarray(X, float) - mu) / sd) @ w + b))) def logloss(p, y): p = np.clip(p, 1e-9, 1 - 1e-9) return float(-np.mean(y * np.log(p) + (1 - y) * np.log(1 - p))) # --------------------------------------------------------------------------- # CANARY with the positive control G-2's H-2 did not specify (my H-2a). # --------------------------------------------------------------------------- def run_canary(rows): print("=" * 78) print("LEAKAGE CANARY -- must FIRE on the naive store and CLEAR the as-of store") print("in the same run. A canary that passes both is theatre, not a test.") print("=" * 78) rows_t = sorted(rows, key=lambda r: r.feature_recv_at) ids = [r.match_id for r in rows_t] half = len(ids) // 2 tr, te = ids[:half], ids[half:] # time-ordered split; the only variable # between the runs below is the columns naive = NaiveStore(rows_t) y_all = {r.match_id: float(r.outcome) for r in rows_t} X_clean_tr = naive.design_matrix(tr) X_leak_tr = naive.design_matrix(tr, extra_cols=[np.array([y_all[i]]) for i in tr]) X_clean_te = naive.design_matrix(te) X_leak_te = naive.design_matrix(te, extra_cols=[np.array([y_all[i]]) for i in te]) y_tr, y_te = naive.labels(tr), naive.labels(te) ll_clean = logloss(predict(fit_logistic(X_clean_tr, y_tr), X_clean_te), y_te) ll_leak = logloss(predict(fit_logistic(X_leak_tr, y_tr), X_leak_te), y_te) lift_naive = ll_clean - ll_leak teeth = lift_naive > 0.05 print(f" split: {len(tr)} train / {len(te)} test, time-ordered") print("\n naive store (join on match_id, no as_of):") print(f" log-loss without canary : {ll_clean:>10.4f}") print(f" log-loss with canary : {ll_leak:>10.4f}") print(f" LIFT : {lift_naive:>10.4f} <-- MUST be > 0") # as-of path: attempt to obtain the canary column before full time. The correct # failure mode is a hard error, not a zero. as_of = min(r.feature_recv_at for r in rows_t) fa = FeatureAccess(rows_t, as_of) refused = None try: fa.column("outcome") except KeyError as exc: refused = str(exc) outcome_visible = any(r.outcome_recv_at <= as_of for r in rows_t) close_visible = any(r.close_recv_at <= as_of for r in rows_t) print(f"\n as-of store, FeatureAccess(as_of={iso(as_of)}):") print(f" rows served : {len(fa.ids()):>10}") print(f" columns served : {fa.matrix().shape[1]:>10} (canary adds 0)") print(f" any outcome knowable? : {str(outcome_visible):>10} must be False") print(f" any close knowable? : {str(close_visible):>10} must be False") print(f" request for 'outcome' : {'REFUSED' if refused else 'SERVED <-- FAILURE'}") if refused: print(f" reason: {refused}") lift_asof = 0.0 if not outcome_visible else float("nan") print(f" LIFT : {lift_asof:>10.4f} must be 0.0000") print("\n VERDICT") if not teeth: print(" CANARY BROKEN -- no lift on a known leak. Nothing downstream is") print(" admissible, INCLUDING a clean-looking result from the as-of store.") elif refused and not outcome_visible and not close_visible: print(f" Canary has teeth (naive lift {lift_naive:.4f}) and the as-of store is") print(" structurally unable to serve the future. Harness accepted; G-6 may") print(" build storage against this interface.") else: print(" AS-OF STORE FAILED -- future facts reachable. Reject before spend.") return dict(lift_naive=lift_naive, lift_asof=lift_asof, teeth=teeth, refused=bool(refused)) # --------------------------------------------------------------------------- # H-1 determinism. # --------------------------------------------------------------------------- def run_determinism(rows): print("\n" + "=" * 78) print("H-1 AS-OF DETERMINISM -- same as_of must yield a byte-identical vector") print("=" * 78) stamps = sorted({r.feature_recv_at for r in rows}) probes = [stamps[0], stamps[len(stamps) // 3], stamps[2 * len(stamps) // 3], stamps[-1]] ok = True for ts in probes: a, b = FeatureAccess(rows, ts), FeatureAccess(rows, ts) same = a.fingerprint() == b.fingerprint() ok &= same print(f" as_of={iso(ts)} rows={len(a.ids()):>5} " f"fp_a={a.fingerprint()} fp_b={b.fingerprint()} " f"{'OK' if same else 'BUILD FAILURE'}") monotone = all(len(FeatureAccess(rows, stamps[i]).ids()) <= len(FeatureAccess(rows, stamps[i + 1]).ids()) for i in range(len(stamps) - 1)) print(f" fingerprints reproducible: {ok} visible set monotone in as_of: {monotone}") # --------------------------------------------------------------------------- # Walk-forward: strictly time-ordered, staked-venue close as the benchmark. # --------------------------------------------------------------------------- def walk_forward(rows, n_splits=8): print("\n" + "=" * 78) print("WALK-FORWARD RUN (hardened store only) -- benchmark is the STAKED CLOSE") print("=" * 78) order = sorted(rows, key=lambda r: r.feature_recv_at) bounds = [len(order) // n_splits * k for k in range(1, n_splits + 1)] # Feature vectors are served by the interface. This library is built at an # as_of at the END of history, legitimate only because a received feature row # never changes once received (append-only). No score enters this map. fa_lib = FeatureAccess(rows, order[-1].feature_recv_at) X_lib = {i: x for i, x in zip(fa_lib.ids(), fa_lib.matrix())} sa = ScoringAccess(rows) start, results = 0, [] for b in bounds: block = order[start:b] start = b if not block: continue as_of = block[0].feature_recv_at fa = FeatureAccess(rows, as_of) # training via the INTERFACE tr_ids = [i for i in fa.ids() if i < block[0].match_id] labelled = [(i, sa.label(i, as_of)) for i in tr_ids] keep = [(i, v) for i, v in labelled if v is not None] if len(keep) < 150: continue Xtr = np.vstack([X_lib[i] for i, _ in keep]) ytr = np.array([v for _, v in keep]) model = fit_logistic(Xtr, ytr) qs, cls, outs = [], [], [] for r in block: if r.feature_recv_at > as_of: # only knowable rows continue qs.append(float(predict(model, r.feature.reshape(1, -1))[0])) cls.append(sa.close(r.match_id)) outs.append(float(r.outcome)) if not qs: continue qs, cls, outs = map(np.asarray, (qs, cls, outs)) ev = qs * cls - 1.0 # gross edge at staked close bet = ev > HANDLE # clear the vig, not zero results.append(dict( as_of=as_of, n_train=len(keep), n_test=len(qs), n_bet=int(bet.sum()), mean_clv=float(ev[bet].mean()) if bet.any() else float("nan"), logloss=logloss(qs, outs), fp=fa.fingerprint(), )) print(f" {'as_of':<17}{'n_tr':>6}{'n_te':>6}{'n_bet':>7}{'mean CLV':>10}" f"{'logloss':>9}{'fp':>7}") for r in results: clv = f"{r['mean_clv']:+.4f}" if r["n_bet"] else " n/a" print(f" {iso(r['as_of']):<17}{r['n_train']:>6}{r['n_test']:>6}{r['n_bet']:>7}" f"{clv:>10}{r['logloss']:>9.4f}{r['fp'][:6]:>7}") tot_bet = sum(r["n_bet"] for r in results) if results else 0 if results: print(f"\n total staked selections: {tot_bet}") if tot_bet: pooled = float(np.nanmean([r["mean_clv"] for r in results if r["n_bet"]])) print(f" pooled mean CLV vs staked close: {pooled:+.4f} -> " f"{'CLEARS the handle' if pooled > HANDLE else 'DOES NOT clear the handle: no edge'}") else: print(" no selection cleared the vig. Under KC-1 this family fails the " "screen -- reported as a FAIL, per Task D.") print(" NOTE: synthetic. The number is not a finding. That the harness refuses") print(" the model the close, AND the canary catches a store that would -- that is") print(" the finding, and it is what G-6 builds storage to satisfy.") return results def query_contract(): print("\n" + "=" * 78) print("QUERY CONTRACT -- the half-page G-6 needs (t014 c.0024 s3)") print("=" * 78) print(""" ALLOWED FeatureAccess(as_of) filters at CONSTRUCTION, not at read ALLOWED .matrix() / .ids() / .column(name) / .fingerprint() ALLOWED ScoringAccess.label(id, as_of) -> value iff outcome_recv_at <= as_of -> else None (never forward-filled) ALLOWED ScoringAccess.close(id) [harness process ONLY] FORBIDDEN any method accepting a timestamp later than the caller's as_of FORBIDDEN ScoringAccess inside the modeler's namespace or process FORBIDDEN enumerating test-window ids from FeatureAccess FORBIDDEN joins on (team, date) against a table carrying event time but no receive time <-- the most common leak in this literature FORBIDDEN a production path that can serve a raw frame (NaiveStore is a TEST FIXTURE; it must never ship reachable) STRUCTURAL the read path is `feature_recv_at <= as_of` inside the storage engine, indexed. If filtering can be forgotten by a caller, the design has already failed H-3. ACCEPT the injected canary must LIFT the naive fixture and CLEAR the as-of store in ONE run (H-2a). A canary never shown to fail is not a test. """) if __name__ == "__main__": print(f"cwd={os.getcwd()} FORUM_JSON={os.environ.get('FORUM_JSON', '')}\n") rows = make_dataset(np.random.default_rng(RNG_SEED)) print(f"synthetic fixtures: {len(rows)} matches, {DIM} pre-kickoff features, " f"~{HANDLE * 100:.0f}% handle, seed={RNG_SEED}") run_canary(rows) run_determinism(rows) walk_forward(rows) query_contract() print("COMPLETE.")