#!/usr/bin/env python3 """Estimate every component the engine currently assumes, and write it down. python3 worker/fit_model.py [--backtest 1500] Produces /calibration.json holding: hour_weights measured hour-of-day curve (was invented constants) weekday_weights measured day-of-week curve (was invented constants) exponents b1,b2,b3 on TVU / share / affinity, with standard errors (were all fixed at 1.0) sigma_by_tier residual log sigma per tier from a hold-out backtest (were fixed constants) correction_by_tier shrunk multiplicative bias correction (was never computed) Everything is stamped with the row counts it came from, so a reader can judge each number rather than take it. Re-run whenever the reference snapshot is rebuilt; the engine falls back to the old constants if this file is absent. """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path BASE = Path(__file__).resolve().parent.parent sys.path.insert(0, str(BASE)) _vendor = BASE / "vendor" if _vendor.is_dir(): sys.path.insert(0, str(_vendor)) import numpy as np # noqa: E402 import pandas as pd # noqa: E402 from engine.calibrate import (fit_exponents, measure_hour_curve, # noqa: E402 measure_weekday_curve, residual_stats) from engine.core import estimate_row, index_evidence # noqa: E402 from engine.reference import ReferenceData # noqa: E402 def load_ref(cfg: dict) -> ReferenceData: if cfg.get("reference_mode") == "fixtures" and cfg.get("fixtures_dir"): return ReferenceData.from_fixtures(cfg["fixtures_dir"]) return ReferenceData.from_mysql(cfg) def backtest(ref: ReferenceData, n_rows: int, seed: int = 7) -> tuple[dict, dict, dict]: """Estimate rows whose true audience we already know, and score the error. Each sampled row is removed from its own evidence before being estimated, so the engine cannot retrieve the answer it is being asked for. Without that exclusion every tier-1 estimate would trivially reproduce its target and the intervals would come out absurdly tight. """ ev = ref.evidence() live = ev[(ev["ama_000"] > 0) & (ev["telecast_type"] == "LIVE")] rng = np.random.default_rng(seed) idx = rng.choice(live.index.to_numpy(), size=min(n_rows, len(live)), replace=False) targets = live.loc[idx] # Leave-k-out rather than leave-one-out: every sampled row is removed from # the evidence in one pass, and the index is built once. Removing 1,200 of # 677,000 LIVE rows changes the evidence base negligibly, no target can # retrieve itself, and it runs in one index build instead of 1,200. held = ReferenceData(frames={**ref.frames, "global_sports": ev.drop(index=idx)}, calibration=ref.calibration) sub = index_evidence(held) preds, truths, tiers, depth = [], [], [], [] t0 = time.time() for i, (ridx, r) in enumerate(targets.iterrows(), 1): row = {"event_name": r["event_name"], "country_name": r["country"], "channel_name": r["channel"], "telecast_type": r["telecast_type"], "hour": r["hour"], "weekday": r["weekday"], "sports_teams": r.get("teams", "")} try: res = estimate_row(row, ref, sub) except Exception: # noqa: BLE001 continue if res["strategy_tier"] and np.isfinite(res["estimated_ama_000"] or np.nan): preds.append(res["estimated_ama_000"]) truths.append(float(r["ama_000"])) tiers.append(int(res["strategy_tier"])) depth.append(int(res["evidence_rows"])) if i % 100 == 0: print(f" backtest {i}/{len(targets)} ({time.time()-t0:.0f}s)", flush=True) p, t, k = np.array(preds), np.array(truths), np.array(tiers) corr, sig = residual_stats(p, t, k) ape = np.abs(p - t) / t score = {"rows_scored": int(len(p)), "WAPE": float(np.abs(p - t).sum() / t.sum()) if len(p) else None, "median_APE": float(np.median(ape)) if len(p) else None, "hit_within_50pct": float((ape <= 0.5).mean()) if len(p) else None, "bias_log_mean": float(np.mean(np.log(t / p))) if len(p) else None, "per_tier": {int(x): int((k == x).sum()) for x in np.unique(k)}} return corr, sig, score def backtest_cross_market(ref: ReferenceData, n_pairs: int, seed: int = 11) -> dict: """Measure the error of the projection the product actually sells. Sampling single rows is dominated by tier 1: a random LIVE broadcast almost always has same-channel history, so 1,108 of 1,200 rows never exercise the cross-market path at all. To measure THAT, remove every row for an (event, country) pair and estimate it from the rest of the world -- which is exactly the position a market with no history for an event is in. This is leave-one-market-out cross-validation, and it is the honest test of the claim "we can estimate a market that has never shown this event". """ ev = ref.evidence() live = ev[(ev["ama_000"] > 0) & (ev["telecast_type"] == "LIVE")] pairs = (live.groupby(["event_name", "country"]).size() .reset_index(name="n").query("n >= 5")) if pairs.empty: return {"pairs_scored": 0} rng = np.random.default_rng(seed) take = pairs.iloc[rng.choice(len(pairs), size=min(n_pairs, len(pairs)), replace=False)] preds, truths, tiers = [], [], [] t0 = time.time() for i, p in enumerate(take.itertuples(index=False), 1): mask = (live["event_name"] == p.event_name) & (live["country"] == p.country) target_rows = live[mask] held = ReferenceData(frames={**ref.frames, "global_sports": ev[~ev.index.isin(target_rows.index)]}, calibration=ref.calibration) sub = index_evidence(held) truth = float(target_rows["ama_000"].median()) r0 = target_rows.iloc[0] row = {"event_name": p.event_name, "country_name": p.country, "channel_name": r0["channel"], "telecast_type": "LIVE", "hour": r0["hour"], "weekday": r0["weekday"], "sports_teams": ""} try: res = estimate_row(row, ref, sub) except Exception: # noqa: BLE001 continue if res["strategy_tier"] and np.isfinite(res["estimated_ama_000"] or np.nan) and truth > 0: preds.append(res["estimated_ama_000"]); truths.append(truth) tiers.append(int(res["strategy_tier"])) if i % 25 == 0: print(f" cross-market {i}/{len(take)} ({time.time()-t0:.0f}s)", flush=True) if not preds: return {"pairs_scored": 0} p_, t_, k_ = np.array(preds), np.array(truths), np.array(tiers) lr = np.log(t_ / p_) ape = np.abs(p_ - t_) / t_ out = {"pairs_scored": int(len(p_)), "WAPE": float(np.abs(p_ - t_).sum() / t_.sum()), "median_APE": float(np.median(ape)), "hit_within_50pct": float((ape <= 0.5).mean()), "hit_within_2x": float((np.abs(lr) <= np.log(2)).mean()), "bias_log_mean": float(np.mean(lr)), "sigma_log": float(np.std(lr, ddof=1)), "tier_mix": {int(x): int((k_ == x).sum()) for x in np.unique(k_)}} corr, sig = residual_stats(p_, t_, k_) out["sigma_by_tier"] = sig out["correction_by_tier"] = corr return out def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--backtest", type=int, default=1200, help="rows to hold out and score (0 to skip)") ap.add_argument("--cross-market", type=int, default=120, help="(event,country) pairs to hold out entirely (0 to skip)") args = ap.parse_args() cfg = json.loads((BASE / "worker" / "config.json").read_text()) out_dir = Path(cfg.get("fixtures_dir") or (BASE / "reference_cache")) print(f"[fit] loading reference …", flush=True) ref = load_ref(cfg) gs = ref.evidence() print(f"[fit] {len(gs):,} evidence rows", flush=True) result: dict = {"generated_from_rows": int(len(gs))} print("[fit] measuring hour curve …", flush=True) hr = measure_hour_curve(gs) if not hr.empty: result["hour_weights"] = hr[["hour", "weight"]].to_dict("records") result["hour_groups"] = int(hr["n_groups"].min()) print(f" {len(hr)} hours, min {hr['n_groups'].min()} groups each", flush=True) print("[fit] measuring weekday curve …", flush=True) wd = measure_weekday_curve(gs) if not wd.empty: result["weekday_weights"] = wd[["weekday", "weight"]].to_dict("records") print(f" {len(wd)} days", flush=True) print("[fit] fitting exponents …", flush=True) tvu = {c: ref.tvu(c) for c in gs["country"].unique() if ref.tvu(c)} share, flagship = {}, {} glr = ref.frames["gl_ratings"] for r in glr.dropna(subset=["ti_total_day_share"]).itertuples(index=False): for ch in {getattr(r, "channel", None), getattr(r, "real_channel", None)}: if ch and isinstance(ch, str): key = (r.country, ch) share[key] = max(share.get(key, 0.0), float(r.ti_total_day_share)) for (c, _ch), s in share.items(): flagship[c] = max(flagship.get(c, 0.0), s) def affinity_of(country, event): sport = ref.sport_of(event) return ref.affinity(country, sport) if sport else None exp = fit_exponents(gs, tvu, share, flagship, affinity_of) result["exponents"] = exp if exp.get("ok"): print(f" tvu {exp['tvu']:+.3f} (se {exp['se_tvu']:.3f}) · " f"share {exp['share']:+.3f} (se {exp['se_share']:.3f}) · " f"affinity {exp['affinity']:+.3f} (se {exp['se_affinity']:.3f})", flush=True) print(f" {exp['n_cells']:,} cells · {exp['n_events']} events · " f"within-R2 {exp['r2_within']:.3f}", flush=True) else: print(" could not fit:", exp.get("reason"), flush=True) if args.backtest: print(f"[fit] backtesting {args.backtest} held-out rows …", flush=True) corr, sig, score = backtest(ref, args.backtest) result["correction_by_tier"] = corr result["sigma_by_tier"] = sig result["backtest"] = score print(f" scored {score['rows_scored']:,} · WAPE {score['WAPE']:.3f} · " f"median APE {score['median_APE']:.3f}", flush=True) print(f" sigma by tier: {sig}", flush=True) print(f" correction : { {k: round(v,3) for k,v in corr.items()} }", flush=True) if args.cross_market: print(f"[fit] leave-one-market-out on {args.cross_market} event-country pairs …", flush=True) cm = backtest_cross_market(ref, args.cross_market) result["cross_market_backtest"] = cm if cm.get("pairs_scored"): print(f" {cm['pairs_scored']} pairs · median APE {cm['median_APE']:.2f} · " f"within 2x {cm['hit_within_2x']:.0%} · sigma(log) {cm['sigma_log']:.2f}", flush=True) print(f" tier mix {cm['tier_mix']} bias(log) {cm['bias_log_mean']:+.3f}", flush=True) # The cross-market sigma is the one tier 4 should actually carry. if cm["sigma_by_tier"].get(4): result.setdefault("sigma_by_tier", {})["4"] = cm["sigma_by_tier"][4] out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "calibration.json").write_text(json.dumps(result, indent=1)) print(f"[fit] wrote {out_dir/'calibration.json'}", flush=True) return 0 if __name__ == "__main__": raise SystemExit(main())