"""Synthetic fixture world with PLANTED ground truth. We generate reference tables (tvuniverse, gl_ratings, curves, factors) and a history of observed broadcasts (global_sports) from a known generative model: log AMA = log(session_base) + log(share_ratio_vs_flagship) + log(hour_w) + log(weekday_w) + log(telecast_mult) + Normal(0, sigma) Because WE own the truth, the engine's job is measurable: recover it. """ from __future__ import annotations import numpy as np import pandas as pd TELECAST_MULT = {"LIVE": 1.00, "DELAYED": 0.40, "ON-DEMAND": 0.35, "PRE_POST_ANALYSIS": 0.28, "OTHERS": 0.15, "HIGHLIGHTS": 0.14, "REPEAT": 0.10, "ARCHIVE": 0.04} WEEKDAY_W = {0: 0.90, 1: 0.88, 2: 0.90, 3: 0.98, 4: 1.10, 5: 1.15, 6: 1.20} HOUR_W = [0.35, 0.25, 0.18, 0.12, 0.10, 0.15, 0.30, 0.45, 0.50, 0.40, 0.45, 0.48, 0.55, 0.60, 0.48, 0.55, 0.55, 0.80, 0.85, 1.00, 0.95, 0.90, 0.80, 0.65] COUNTRIES = { # name -> (tvu_000, n_channels) "India": (890_000, 4), "United Kingdom": (62_000, 3), "Australia": (23_000, 3), "South Africa": (28_000, 2), "UAE": (8_500, 2), "Singapore": (4_200, 2), } EVENTS = { # event -> (base draw as fraction of TVU at flagship/LIVE/1900, team pool) "T20 Super League": (0.055, ["Mumbai Kings", "Chennai Titans", "Delhi Chargers", "Punjab Lions", "Bengal Riders", "Hyderabad Hawks"]), "World Test Series": (0.030, ["India", "Australia", "England", "South Africa"]), "Premier Kabaddi": (0.018, ["Patna Pirates", "U Mumba", "Jaipur Panthers", "Bengal Warriors"]), } TEAM_WEIGHT = {"India": 1.30, "Australia": 1.10, "England": 1.10, "Mumbai Kings": 1.15, "Chennai Titans": 1.15} # everyone else 1.00 (planted) HOST = {"T20 Super League": "India", "World Test Series": "Australia", "Premier Kabaddi": "India"} HOST_UPLIFT = 1.10 NOISE_SIGMA = 0.15 def build_world(seed: int = 7, n_history_days: int = 240): rng = np.random.default_rng(seed) # channels & shares: flagship share drawn per country, others decay channels, gl_rows, tvu_rows = {}, [], [] for country, (tvu, nch) in COUNTRIES.items(): tvu_rows.append({"country": country, "tvuniverse_000": tvu}) shares = np.sort(rng.uniform(0.02, 0.14, nch))[::-1] chs = [f"{country.split()[0]} Sports {i+1}" for i in range(nch)] channels[country] = list(zip(chs, shares)) for ch, sh in channels[country]: gl_rows.append({"country": country, "channel": ch, "ti_total_day_share": round(float(sh), 4)}) tvuniverse = pd.DataFrame(tvu_rows) gl_ratings = pd.DataFrame(gl_rows) def true_log_ama(event, country, channel_share, flagship_share, hour, wd, telecast, teams): base_frac, _ = EVENTS[event] tvu = COUNTRIES[country][0] session = base_frac * tvu tw = float(np.mean([TEAM_WEIGHT.get(t, 1.0) for t in teams])) host = HOST_UPLIFT if HOST[event] == country else 1.0 return (np.log(session) + np.log(channel_share / flagship_share) + np.log(HOUR_W[hour]) + np.log(WEEKDAY_W[wd]) + np.log(TELECAST_MULT[telecast]) + np.log(tw) + np.log(host)) # history: LIVE-heavy observation log rows = [] start = pd.Timestamp("2025-10-01") for d in range(n_history_days): date = start + pd.Timedelta(days=d) for event, (_, pool) in EVENTS.items(): if rng.random() < 0.55: # event active this day teams = list(rng.choice(pool, 2, replace=False)) for country, chs in channels.items(): # planted gaps: no history for (Premier Kabaddi, Singapore) or # (World Test Series, UAE) -> forces cross-market tier; # history only on the top-2 channels -> weaker channels # force same-event/same-teams tiers. if (event, country) in {("Premier Kabaddi", "Singapore"), ("World Test Series", "UAE")}: continue if rng.random() < 0.7: ch, sh = chs[int(rng.integers(0, min(2, len(chs))))] flagship = chs[0][1] hour = int(rng.choice([14, 17, 18, 19, 20], p=[.1, .15, .2, .35, .2])) telecast = "LIVE" if rng.random() < 0.8 else \ str(rng.choice(["REPEAT", "HIGHLIGHTS", "DELAYED"])) mu = true_log_ama(event, country, sh, flagship, hour, date.weekday(), telecast, teams) ama = float(np.exp(mu + rng.normal(0, NOISE_SIGMA))) rows.append({"event_name": event, "country": country, "channel": ch, "teams": " | ".join(teams), "date": date.date().isoformat(), "weekday": date.weekday(), "hour": hour, "telecast_type": telecast, "ama_000": round(ama, 2)}) global_sports = pd.DataFrame(rows) ref = { "tvuniverse": tvuniverse, "gl_ratings": gl_ratings, "global_sports": global_sports, "telecast_multipliers": pd.DataFrame( [{"telecast_type": k, "multiplier": v} for k, v in TELECAST_MULT.items()]), "weekday_weights": pd.DataFrame( [{"weekday": k, "weight": v} for k, v in WEEKDAY_W.items()]), "hour_weights": pd.DataFrame( [{"hour": h, "weight": w} for h, w in enumerate(HOUR_W)]), "team_weights": pd.DataFrame( [{"team": t, "weight": w} for t, w in TEAM_WEIGHT.items()]), "host_factor": pd.DataFrame( [{"event_name": e, "host_country": c, "factor": HOST_UPLIFT} for e, c in HOST.items()]), } def truth_fn(row) -> float: chs = channels[row["country"]] share = dict(chs)[row["channel"]] teams = [t.strip() for t in str(row["teams"]).split("|")] return float(np.exp(true_log_ama(row["event_name"], row["country"], share, chs[0][1], int(row["hour"]), int(row["weekday"]), row["telecast_type"], teams))) return ref, truth_fn def build_schedule(ref, seed: int = 21, n_rows: int = 400) -> pd.DataFrame: """A future schedule to estimate (dates AFTER the history window).""" rng = np.random.default_rng(seed) gs = ref["global_sports"] channels_by_country = (ref["gl_ratings"].groupby("country")["channel"] .apply(list).to_dict()) rows, start = [], pd.Timestamp("2026-06-01") events = list(EVENTS.items()) for i in range(n_rows): event, (_, pool) = events[int(rng.integers(0, len(events)))] country = str(rng.choice(list(COUNTRIES))) ch = str(rng.choice(channels_by_country[country])) date = start + pd.Timedelta(days=int(rng.integers(0, 45))) hour = int(rng.choice([13, 14, 17, 18, 19, 20, 21])) telecast = str(rng.choice(list(TELECAST_MULT), p=[.55, .08, .05, .05, .05, .08, .1, .04])) teams = list(rng.choice(pool, 2, replace=False)) rows.append({"event_name": event, "country_name": country, "channel_name": ch, "programme_name": f"{teams[0]} vs {teams[1]}", "sports_teams": " | ".join(teams), "broadcast_date": date.date().isoformat(), "weekday": date.weekday(), "hour": hour, "telecast_type": telecast}) return pd.DataFrame(rows)