#!/usr/bin/env python3 """Forward verdict ledger — the only place belief comes from. Each run re-scores the FROZEN studies on the last RESCORE_DAYS UTC days of tape and appends one row per (study, day) to forward_ledger.jsonl. Days are recomputed on later runs so pending (unresolved-at-the-time) triggers resolve into their day's row; readers keep the newest computed_at per key. Studies: flow frozen params from params/study_flow.json — informed set as-of each day's 00:00 UTC, scored at p50 lag, first- AND worst-print fills; plus 3 FIXED control seeds (identity-lift tracking). oracle params/study_oracle.json grid — ALL edge levels tracked until one accumulates >= 30 forward fills (then the selection rule in the pre-registration applies). Skips days without tick coverage. Verdicts are pre-registered in the study issues; this script only reports. """ import json import os import time import tape import sim as simmod import study_flow as sf import study_oracle as so HERE = os.path.dirname(os.path.abspath(__file__)) LEDGER = os.path.join(HERE, "forward_ledger.jsonl") RESCORE_DAYS = 3 CONTROL_SEEDS = (1, 2, 3) def day_bounds(d): lo = time.mktime(time.strptime(d, "%Y-%m-%d")) - time.timezone return lo, lo + 86400 def payouts_for(db, assets): """asset -> payout via tape proxy FIRST, then CTF chain truth for the rest. THE 2026-07-22 SCORER BUG: 'pending' was treated as ignorable, but tape-resolution timing is win-biased — a LOSS keeps its winning sibling trading (sibling-veto holds the market open), so losses hid in pending while wins scored. Jul-21 audit: tape-resolved fills hit 81%; chain-resolving the 'pending' bucket hit 26% (n=329) — combined 53%. The surge paper book (chain-graded from day one) was right; this scorer was flattering every arm. Chain overlay is now mandatory.""" out, missing = {}, [] for a in set(assets): r = db.execute("SELECT payout::DOUBLE FROM res_tok WHERE asset=?", [a]).fetchone() if r: out[a] = r[0] else: missing.append(a) conds = {} for a in missing: c = db.execute("SELECT any_value(cond) FROM trades WHERE asset=?", [a]).fetchone()[0] if c: conds[a] = c if conds: tr = tape.chain_overlay([(c, a) for a, c in conds.items()]) for a, c in conds.items(): v = tr.get((c, a)) if v is not None: out[a] = v # 1.0 / 0.0 / 0.5 (refund) return out def score_flow(db, fz, d, hold_s): lo, hi = day_bounds(d) t_max = db.execute("SELECT max(ts) FROM trades").fetchone()[0] hi = min(hi, t_max) S = sf.informed_set(db, lo, fz["top_n"]) tape.build_resolved(db) trig = sf.signals(db, S, lo, hi, fz["window_s"], fz["flow_usd"]) row = {"triggers": len(trig), "set_size": len(S)} pays = payouts_for(db, [t["asset"] for t in trig]) for mode in ("first", "worst"): s = simmod.Sim(db, lag_s=simmod.LAG_P50, hold_s=hold_s, fill=mode) agg = dict(fills=0, misses=0, pending=0, refunds=0, pnl=0.0, wins=0) for t in trig: pay = pays.get(t["asset"]) r = s.try_buy(t["asset"], t["ts"], t["p_ref"], stake_usd=sf.STAKE) if not r["filled"]: agg["misses"] += 1 elif pay is None: # truly unresolved (chain included) agg["pending"] += 1 elif pay == 0.5: agg["refunds"] += 1 agg["pnl"] += r["shares"] * 0.5 - r["cost"] - r["fee"] else: agg["fills"] += 1 agg["pnl"] += r["shares"] * (pay - r["price"]) - r["fee"] agg["wins"] += pay == 1.0 agg["pnl"] = round(agg["pnl"], 2) if agg["fills"]: agg["ev_per_fill"] = round(agg["pnl"] / agg["fills"], 2) agg["hit"] = round(agg["wins"] / agg["fills"], 3) row[mode] = agg ctl = [] for seed in CONTROL_SEEDS: C = sf.matched_random_set(db, lo, fz["top_n"], seed) ctrig = sf.signals(db, C, lo, hi, fz["window_s"], fz["flow_usd"]) cpays = payouts_for(db, [t["asset"] for t in ctrig]) s = simmod.Sim(db, lag_s=simmod.LAG_P50, hold_s=hold_s, fill="worst") fills = 0 pnl = 0.0 for t in ctrig: pay = cpays.get(t["asset"]) r = s.try_buy(t["asset"], t["ts"], t["p_ref"], stake_usd=sf.STAKE) if r["filled"] and pay is not None and pay != 0.5: fills += 1 pnl += r["shares"] * (pay - r["price"]) - r["fee"] ctl.append({"seed": seed, "fills": fills, "pnl": round(pnl, 2)}) row["controls_worst"] = ctl return row def score_oracle(db, P, d, hold_s): lo, hi = day_bounds(d) series = {s: so.TickSeries(tape.load_ticks(db, s)) for s in ("btcusdt", "ethusdt", "solusdt", "xrpusdt", "bnbusdt", "dogeusdt")} have = [s for s in series.values() if s.ts and s.ts[0] < hi and s.ts[-1] > lo] if not have: return {"skipped": "no tick coverage"} outcomes = so.outcome_map(db) tape.build_resolved(db) uni = so.crypto_universe(db, outcomes, series) sim = simmod.Sim(db, hold_s=hold_s) evs = [] # (asset, edge, sim result) — score after for u in uni: prints = db.execute("""SELECT ts, price FROM trades WHERE asset=? AND ts > ? AND ts <= ? ORDER BY ts""", [u["asset"], lo, hi]).fetchall() s = series[u["mkt"]["sym"]] last_ev = 0.0 for ts, px in prints: if ts - last_ev < so.COOLDOWN_S: continue f = so.fair_value(u["mkt"], u["up"], s.at(ts), s.vol_1s(ts), ts) if f is None: continue edge = f - float(px) if edge < min(so.EDGE_GRID): continue last_ev = ts evs.append((u["asset"], edge, sim.try_buy(u["asset"], ts, float(px), stake_usd=so.STAKE))) # chain-overlay payouts (same 2026-07-22 scorer fix as score_flow) pays = payouts_for(db, [a for a, _, r in evs if r["filled"]]) row = {} for asset, edge, r in evs: for E in so.EDGE_GRID: if edge < E: continue g = row.setdefault(str(E), {"events": 0, "fills": 0, "pending": 0, "refunds": 0, "pnl": 0.0, "wins": 0}) g["events"] += 1 if not r["filled"]: continue pay = pays.get(asset) if pay is None: g["pending"] += 1 elif pay == 0.5: g["refunds"] += 1 g["pnl"] += r["shares"] * 0.5 - r["cost"] - r["fee"] else: g["fills"] += 1 g["pnl"] += r["shares"] * (pay - r["price"]) - r["fee"] g["wins"] += pay == 1.0 for g in row.values(): g["pnl"] = round(g["pnl"], 2) if g["fills"]: g["ev_per_fill"] = round(g["pnl"] / g["fills"], 2) g["hit"] = round(g["wins"] / g["fills"], 3) return row def score_lean(db, d): """Study C (maker inventory-lean, pre-registered): walk-forward day row. Screen as-of day start (maker_lean.screen_asof — tape strictly before the day), frozen trigger (maker_lean.day_leans), chain truth via payouts_for. Arms: follow_small ($150-500) · fade_whale ($2k+) · mid bucket report-only. Scored at last-print entry (stated optimism — a PASS graduates to a real-execution paper arm, never to money).""" import maker_lean as ml lo, hi = day_bounds(d) t_max = db.execute( "SELECT max(ts) FROM aux WHERE type='orders_matched'").fetchone()[0] hi = min(hi, t_max or 0) if hi <= lo: return {"skipped": "no maker-stream coverage"} tape.build_resolved(db) sharps = ml.screen_asof(db, lo) if not sharps: return {"skipped": "no screened wallets as-of day"} leans = ml.day_leans(db, lo, hi, sharps) pays = payouts_for(db, [t["a"] for t in leans]) row = {"screened": len(sharps), "leans": len(leans)} arms = {"follow_small": ("follow", lambda u: u < 500), "mid_report": ("follow", lambda u: 500 <= u < 2000), "fade_whale": ("fade", lambda u: u >= 2000)} for name, (direction, sel) in arms.items(): n = pend = wins = 0 pnl = 0.0 for t in leans: if not sel(t["lean_usd"]): continue p = pays.get(t["a"]) if p is None: pend += 1 continue if p == 0.5: continue lean_pay = p if t["side"] > 0 else 1 - p px, pay = ((t["lean_px"], lean_pay) if direction == "follow" else (1 - t["lean_px"], 1 - lean_pay)) if not (0.05 <= px <= 0.95): continue n += 1 pnl += 100.0 / px * (pay - px) wins += pay == 1.0 row[name] = {"n": n, "pending": pend, "pnl": round(pnl, 2), "ev_per_lean": round(pnl / n, 2) if n else None, "hit": round(wins / n, 3) if n else None} return row def main(): db = tape.connect() cal = json.load(open(os.path.join(HERE, "params", "sim_calibration.json"))) flow_p = json.load(open(os.path.join(HERE, "params", "study_flow.json"))) fz = flow_p["frozen"] frozen_at = flow_p["frozen_at"] t_min, t_max = db.execute("SELECT min(ts), max(ts) FROM trades").fetchone() days = {time.strftime("%Y-%m-%d", time.gmtime(t_max - i * 86400)) for i in range(RESCORE_DAYS)} # offline-proofing (2026-07-22): a Mac gap > RESCORE_DAYS must not leave # permanent holes in the verdict evidence — also score any tape-covered # day the ledger has never seen (additive; scoring method unchanged) have = set() try: for ln in open(LEDGER): try: r_ = json.loads(ln) if r_.get("study") == "flow": have.add(r_["day"]) except Exception: pass except FileNotFoundError: pass t = t_min while t < t_max: d_ = time.strftime("%Y-%m-%d", time.gmtime(t)) if d_ not in have: days.add(d_) t += 86400 days = sorted(days) now = time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime()) with open(LEDGER, "a") as fh: for d in days: r1 = score_flow(db, fz, d, cal["hold_s"]) fh.write(json.dumps({"study": "flow", "day": d, "computed_at": now, "frozen_at": frozen_at, **r1}, default=float) + "\n") print(f"flow {d}: trig {r1['triggers']} " f"worst {r1['worst'].get('ev_per_fill')} " f"({r1['worst']['fills']} fills, {r1['worst']['pending']} pend)") r2 = score_oracle(db, None, d, cal["hold_s"]) fh.write(json.dumps({"study": "oracle", "day": d, "computed_at": now, **r2}, default=float) + "\n") print(f"oracle {d}: " + (r2.get("skipped") or " ".join(f"E{E}:{g.get('ev_per_fill')}({g['fills']}f)" for E, g in sorted(r2.items())))) # EXPLORATORY arm (2026-07-21, user ask): the same surge signal # on sub-5c longshots, all niches. NOT pre-registered — 20-50x # lottery payoffs mean one hit flips a small sample (first scan: # 31 resolved fills, 4 wins, sign set entirely by two esports # winners), and the sim has no depth model ($100 at 2c = 5,000 # shares a longshot book won't hold). Accumulates here until # ~100+ resolved fills exist; only then is a pre-registration # (or a kill) worth writing. band0, nich0 = sf.PRICE_BAND, sf.NICHES try: sf.PRICE_BAND = (0.005, 0.05) sf.NICHES = {"sports", "esports", "tennis", "crypto", "politics", "geo", "other"} r3 = score_flow(db, {**fz, "flow_usd": 300}, d, cal["hold_s"]) finally: sf.PRICE_BAND, sf.NICHES = band0, nich0 fh.write(json.dumps({"study": "flow_sub5c_EXPLORATORY", "day": d, "computed_at": now, **r3}, default=float) + "\n") print(f"sub5c {d}: trig {r3['triggers']} " f"worst {r3['worst'].get('ev_per_fill')} " f"({r3['worst']['fills']} fills, {r3['worst']['pending']} pend)") r5 = score_lean(db, d) fh.write(json.dumps({"study": "lean", "day": d, "computed_at": now, **r5}, default=float) + "\n") print(f"lean {d}: " + (r5.get("skipped") or f"{r5['leans']} leans · follow_small " f"{r5['follow_small'].get('ev_per_lean')} " f"({r5['follow_small']['n']}n) · fade_whale " f"{r5['fade_whale'].get('ev_per_lean')} " f"({r5['fade_whale']['n']}n)")) if __name__ == "__main__": main()