"""ReferenceData: one interface, two sources. fixtures mode : reads parquet files from a directory (sandbox / testing / CI) mysql mode : reads the live tvviewers tables on the server (same frames) The engine only ever sees the frames, so swapping source is a config change. """ from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path import numpy as np import pandas as pd FRAMES = ["tvuniverse", "gl_ratings", "global_sports", "telecast_multipliers", "weekday_weights", "hour_weights", "team_weights", "host_factor"] # Real telecast_type values in global_sports include case variants # ("Magazine/Studio" vs "STUDIO"), spelling variants ("ARCHIVED"/"ARCHIVE") and # several ways of saying "not recorded" (NULL, '', 'Not Supplied', 'UNKNOWN', # '-none-'). Left unnormalised these become distinct categories, none of which # match the multiplier table, so every one of them silently takes the unknown # fallback. Both the evidence pool and the target row are canonicalised through # this map so they agree. _TELECAST_CANON = { "": "UNKNOWN", "NONE": "UNKNOWN", "-NONE-": "UNKNOWN", "NAN": "UNKNOWN", "NOT_SUPPLIED": "UNKNOWN", "UNKNOWN": "UNKNOWN", "ARCHIVED": "ARCHIVE", "ARCHIVE": "ARCHIVE", # The production EPG export spells archive material RECORDED_ARCHIVE and # calls a rebroadcast REPEAT; both fell through to the UNKNOWN multiplier. "RECORDED_ARCHIVE": "ARCHIVE", "ARCHIVE_RECORDED": "ARCHIVE", "MAGAZINE/STUDIO": "STUDIO", "STUDIO": "STUDIO", "NON_LIVE": "RECORDED", "RECORDED": "RECORDED", "DELAYED": "RECORDED", "REPEAT": "RECORDED", "RERUN": "RECORDED", "LIVE-LINK": "LIVE_LINK", "LIVE_LINK": "LIVE_LINK", "OTHERS": "UNKNOWN", "OTHER": "UNKNOWN", } def canonical_telecast(t) -> str: """Fold a raw telecast_type into the canonical vocabulary.""" s = str(t).strip().upper().replace(" ", "_") if s in ("NONE", "NAN", "NAT", ""): return "UNKNOWN" return _TELECAST_CANON.get(s, s) def derive_telecast_multipliers(gs: pd.DataFrame) -> pd.DataFrame: """Measure telecast-type multipliers from the evidence pool itself. Within each (country, channel, event_name) group containing LIVE rows, take the median ama_000 per type and express it as a ratio to that group's LIVE median. Comparing only inside a group controls for channel size and event pull, so what survives is the telecast-type effect. The median across groups is the multiplier. This replaces the hardcoded constants in world.py, which were invented for the synthetic world and understated every non-live category (ARCHIVE was 0.04 against a measured 0.36; NON_EVENT/STUDIO/UNKNOWN were absent entirely and fell through to a 0.15 catch-all). """ d = gs[gs["ama_000"] > 0].copy() d["tt"] = d["telecast_type"].map(canonical_telecast) keys = ["country", "channel", "event_name"] med = d.groupby(keys + ["tt"])["ama_000"].median().reset_index() live = (med[med["tt"] == "LIVE"][keys + ["ama_000"]] .rename(columns={"ama_000": "live_ama"})) j = med.merge(live, on=keys, how="inner") j = j[j["live_ama"] > 0] if j.empty: return pd.DataFrame(columns=["telecast_type", "multiplier"]) j["ratio"] = j["ama_000"] / j["live_ama"] agg = j.groupby("tt")["ratio"].agg(["median", "size"]) # Require a few comparable groups before trusting a category. agg = agg[agg["size"] >= 5] return pd.DataFrame({"telecast_type": agg.index, "multiplier": agg["median"].astype(float).values}) def reference_engine(cfg: dict): """SQLAlchemy engine for the read-only tvviewers reference database. Credentials are URL-encoded: a password containing '@', '/', '#' or ':' otherwise corrupts the DSN (an '@' makes SQLAlchemy read the tail of the password as the hostname). The read-only tvviewers reference DB and the read-write crystal application schema are usually different servers, so explicit ref_db_* keys win; db_* is the fallback for single-server setups. """ from urllib.parse import quote_plus from sqlalchemy import create_engine user = quote_plus(str(cfg.get("ref_db_user") or cfg["db_user"])) pwd = quote_plus(str(cfg.get("ref_db_pass") or cfg["db_pass"])) host = cfg.get("ref_db_host") or cfg["db_host"] port = int(cfg.get("ref_db_port") or cfg.get("db_port", 3306)) return create_engine( f"mysql+pymysql://{user}:{pwd}@{host}:{port}/tvviewers", pool_pre_ping=True) @dataclass class ReferenceData: frames: dict = field(default_factory=dict) calibration: dict = field(default_factory=dict) calibration_source: str = "assumed constants (world.py)" source_dir: str | None = None # ---- constructors ------------------------------------------------- @classmethod def from_frames(cls, frames: dict) -> "ReferenceData": return cls(frames={k: v.copy() for k, v in frames.items()}) @classmethod def from_fixtures(cls, fixtures_dir: str | Path) -> "ReferenceData": """Read every parquet in the directory, not just the core eight. FRAMES lists what the estimator cannot run without. The snapshot also carries sport_interest, event_category, event_country_interest_factor and the rest; iterating FRAMES alone silently dropped them, which disabled affinity — so every cross-market projection came back NO_AFFINITY_DATA in fixtures mode while working in mysql mode. """ p = Path(fixtures_dir) frames = {f.stem: pd.read_parquet(f) for f in sorted(p.glob("*.parquet"))} missing = [n for n in FRAMES if n not in frames] if missing: raise FileNotFoundError(f"Fixtures missing: {missing} in {p}") obj = cls(frames=frames) obj.source_dir = str(p) obj.load_calibration(p / "calibration.json") return obj # ---- measured parameters ------------------------------------------ def load_calibration(self, path: str | Path) -> bool: """Replace assumed constants with quantities measured from evidence. worker/fit_model.py estimates the hour and weekday curves, the three exponents and the per-tier residual sigma from global_sports itself. Where that file exists its values win; where it does not, the engine falls back to the constants in world.py and says so through `calibration_source`. """ p = Path(path) if not p.exists(): self.calibration = {} self.calibration_source = "assumed constants (world.py)" return False cal = json.loads(p.read_text()) self.calibration = cal src = f"measured from {cal.get('generated_from_rows', 0):,} rows" if cal.get("stale"): src += " (STALE — snapshot rebuilt since; re-run fit_model.py)" self.calibration_source = src if cal.get("hour_weights"): self.frames["hour_weights"] = pd.DataFrame(cal["hour_weights"]) if cal.get("weekday_weights"): self.frames["weekday_weights"] = pd.DataFrame(cal["weekday_weights"]) self.__dict__.pop("_lookups", None) # drop caches built on the old values return True def exponents(self) -> dict: """Fitted exponents for the cross-market ratios. Defaults to 1.0 each -- the original assumption that audience scales proportionally with market size, channel share and appetite -- so the engine behaves exactly as before when no calibration is present. """ e = getattr(self, "calibration", {}).get("exponents") or {} if not e.get("ok"): return {"tvu": 1.0, "share": 1.0, "affinity": 1.0, "fitted": False} return {"tvu": float(e["tvu"]), "share": float(e["share"]), "affinity": float(e["affinity"]), "fitted": True} def sigma_by_tier(self) -> dict | None: s = getattr(self, "calibration", {}).get("sigma_by_tier") return {int(k): float(v) for k, v in s.items()} if s else None def correction_by_tier(self) -> dict | None: c = getattr(self, "calibration", {}).get("correction_by_tier") return {int(k): float(v) for k, v in c.items()} if c else None @classmethod def from_mysql(cls, cfg: dict) -> "ReferenceData": """Server mode. Column names below are NOT guesses — each is taken verbatim from the original Crystal source (estimateuploadfile), cited per table. This still needs Claude Code's Task 1 (SHOW CREATE TABLE) to confirm data types / nullability / indexes before trusting it at scale, but the column names themselves should already be correct, not placeholders.""" eng = reference_engine(cfg) q = lambda sql: pd.read_sql(sql, eng) # noqa: E731 def q_opt(sql): """For frames the ESTIMATOR does not need. ott_multipliers and the pan tables feed the viewership report only. If one of them is missing or renamed the report should lose a sheet, not the whole run lose its reference data. """ try: return q(sql) except Exception as exc: # noqa: BLE001 print(f"[reference] optional frame unavailable: " f"{type(exc).__name__}: {exc}"[:160], flush=True) return pd.DataFrame() frames = { # Source: utils/metrics/global_sports/strategies/real_channel_overlap_core.py # (lines ~188-193) — exact SELECT list used by the original strategy. # VERIFIED against SHOW CREATE TABLE 2026-08-16: `real_channel` is # NOT a column on global_sports -- it lives on gl_ratings, which is # the bridge between GSIQ channel naming ("Star Sports 1 IN") and # broadcaster naming ("STAR Sports 1"). Removed here; join via # gl_ratings when the mapped name is needed. "global_sports": q(""" SELECT global_id, country, channel, sub_genre, event_name, sports_teams, match_level, prog_date, telecast_type, ama_000, start_time, end_time FROM global_sports WHERE ama_000 IS NOT NULL AND ama_000 > 0 """).rename(columns={"prog_date": "date"}), # Source: utils/metrics/real_mapping.py (line ~14-15) — gl_ratings join # keys are gsiq_country / gsiq_channel, NOT plain country/channel. "gl_ratings": q(""" SELECT gl_id, gsiq_country AS country, gsiq_channel AS channel, real_country, real_channel, ti_total_day_share, prefer_gl_ratings_ama, mean_ama_000, max_ama_000 FROM gl_ratings WHERE gsiq_channel IS NOT NULL AND TRIM(gsiq_channel) <> '' """), # Source: utils/metrics/global_sports/strategies/ama_sanity_caps.py # (line ~11-12) — territory/total_individuals_2025 are the real # column names, NOT "country"/"tvuniverse_000". # world_sub_region / gdp_billion_usd are pulled for analog-market # selection (core.py picks the source market for a cross-market # projection by resemblance, not by row count). Both are well # populated: 93% and 74% of the 274 territories. "tvuniverse": q(""" SELECT territory AS country, total_individuals_2025 AS tvuniverse_000, missing_channel_weight, world_sub_region, gdp_billion_usd FROM tvuniverse """), # Source: utils/metrics/factors/sports_team_weight.py # (TABLE_NAME, line 24) — column names confirmed from the module's # own COL_* constants. # VERIFIED 2026-08-16: real weight column is `ama_team_weight`. # NOTE: team_name is NOT unique -- 28,102 rows / 8,062 teams, keyed # by (event_name, event_season_key, team_name). "France" alone has # 491 rows spanning 90 distinct weights (0.53-1.80). team_w() below # takes the first match, so event/season are carried here to let a # future accessor disambiguate. See handoff/schema_diff.md. "team_weights": q(""" SELECT team_name AS team, event_name, event_season, ama_team_weight AS weight FROM sports_team_weight WHERE ama_team_weight IS NOT NULL """), # Source: utils/metrics/factors/host_country_factor.py # (TABLE_NAME + COL_* constants, lines 17-57) # VERIFIED 2026-08-16: real column is `host_country_factor`, and the # table carries an active_flag that must be honoured. "host_factor": q(""" SELECT event_name, host_country_name AS host_country, host_country_factor AS factor FROM event_host_country_factor WHERE active_flag = 1 """), # Source: utils/metrics/global_sports/strategies/ama_sanity_caps.py # (lines 538-548) — exact SELECT list for the cap-band benchmark table. "event_country_tvuniverse_factor": q(""" SELECT id, event_name, event_season, country, event_ama_000, event_country_tvuniverse_percentage, global_event_ama_000, country_perc_of_global_event_ama FROM event_country_tvuniverse_factor WHERE event_country_tvuniverse_percentage IS NOT NULL AND event_country_tvuniverse_percentage > 0 """), # Source: utils/metrics/global_sports/strategies/non_global_sports_fallback.py # (lines 312-317) — real column is event_country_interest_factor. "event_country_interest_factor": q(""" SELECT event_name, country, event_country_interest_factor AS factor FROM event_country_interest_factor WHERE event_country_interest_factor IS NOT NULL AND event_country_interest_factor > 0 """), # Source: utils/metrics/factors/match_level_interest.py # (TABLE_NAME line 31, COL_* constants 42-44) "match_level_weights": q(""" SELECT event_name, season_name, sport, match_level, match_level_weight AS factor FROM match_level_weight_by_event """), # Source: utils/metrics/global_sports/strategies/post_estimation_features.py # (lines 158-165) — real columns country/channel/cap_ama_000. "global_sports_ama_caps": q(""" SELECT country, channel, cap_ama_000 FROM global_sports_ama_caps WHERE cap_ama_000 IS NOT NULL AND cap_ama_000 > 0 """), # How much each country actually cares about each sport. 273 # countries x 48 categories -- BROADER coverage than global_sports # has broadcast history for (142 countries), which is what makes # cross-market projection possible at all: TV-universe size alone # says how many people could watch, not how many would. # Not in table_inventory.md, so it was never wired before. "sport_interest": q(""" SELECT country, sports_category, follow_percentage, interest_lineartv_avg_percentage FROM national_sport_interest WHERE follow_percentage IS NOT NULL AND follow_percentage > 0 """), # Platform multipliers per market: linear TV -> OTT, and out-of-home # viewing. 273 countries, so full coverage of the no-history set. # Needed for the platform sheets in the viewership report. "ott_multipliers": q_opt(""" SELECT country, tv_to_ott_multiplier, ooh_multiplier, household_min_size, household_max_size FROM ott_multipliers WHERE tv_to_ott_multiplier IS NOT NULL """), # Which channels are pan-regional, and the (country, channel) pairs # that resolve to one. A pan feed covers many markets from a single # signal, so it must never be summed into a single-market total -- # and guessing that from a channel name misfiles national channels. "pan_channels": q_opt(""" SELECT channel_name, pan_region, pan_countries FROM pan_channels """), "pan_mapping": q_opt(""" SELECT country_name, channel_name, pan_channel_name, pan_region FROM pan_mapping WHERE pan_channel_name IS NOT NULL """), # event_name -> sport category, the bridge into sport_interest. "event_category": q(""" SELECT event_name, sub_genre AS sports_category FROM event_category_map WHERE sub_genre IS NOT NULL AND TRIM(sub_genre) <> '' """), } # engine/core.py indexes the evidence pool on `hour`, `weekday` and # `teams`. The real global_sports table stores `start_time`, `prog_date` # and `sports_teams` instead, so derive them here rather than reshape # core.py (Task 3 rule: wire data in, don't change engine logic). gs = frames["global_sports"] # MySQL TIME permits values past 24:00 for broadcast-day scheduling # (6.47% of rows; max observed 30:xx) and arrives as a timedelta, so # 24:37 reads as "1 days 00:37". Modulo 24 maps it to the real # clock hour rather than letting it fall through to hour_w()'s default. st = pd.to_timedelta(gs["start_time"], errors="coerce") gs["hour"] = (st.dt.total_seconds() // 3600 % 24).fillna(19).astype(int) gs["weekday"] = (pd.to_datetime(gs["date"], errors="coerce") .dt.weekday.fillna(5).astype(int)) # core.py expects a single `teams` column; empty string means "unknown" # (85.5% of real rows have no sports_teams value). gs["teams"] = gs["sports_teams"].fillna("") # Canonicalise the evidence pool's telecast_type so that case and # spelling variants ("Magazine/Studio" vs "STUDIO", "ARCHIVED" vs # "ARCHIVE") and the several spellings of "not recorded" collapse into # one vocabulary. Without this each variant is its own category and # silently takes the unknown fallback. gs["telecast_type"] = gs["telecast_type"].map(canonical_telecast) # weekday/hour curves stay as Crystal's hardcoded constants for now. from .world import WEEKDAY_W, HOUR_W # Telecast multipliers are MEASURED from the evidence, not inherited # from world.py's synthetic guesses. Falls back to the constants only # if the pool is too thin to measure. derived = derive_telecast_multipliers(gs) if len(derived) >= 3: frames["telecast_multipliers"] = derived else: from .world import TELECAST_MULT frames["telecast_multipliers"] = pd.DataFrame( [{"telecast_type": k, "multiplier": v} for k, v in TELECAST_MULT.items()]) frames["weekday_weights"] = pd.DataFrame( [{"weekday": k, "weight": v} for k, v in WEEKDAY_W.items()]) frames["hour_weights"] = pd.DataFrame( [{"hour": h, "weight": w} for h, w in enumerate(HOUR_W)]) return cls(frames=frames) def save_fixtures(self, fixtures_dir: str | Path) -> None: p = Path(fixtures_dir); p.mkdir(parents=True, exist_ok=True) for name, df in self.frames.items(): df.to_parquet(p / f"{name}.parquet", index=False) # ---- lookup caches ------------------------------------------------- # Each accessor below used to run a full DataFrame scan per call, and # core.py::_debase calls them once per row of the evidence pool. On real # data (pools of 100k+ rows) that made a single estimate take minutes. # The frames never change after construction, so each lookup is built once # into a dict. Semantics are preserved exactly, including first-match-wins # on duplicate keys and the per-accessor defaults. def _cache(self, name: str, build): c = self.__dict__.setdefault("_lookups", {}) if name not in c: c[name] = build() return c[name] @staticmethod def _first_wins(df, keys, value): d = df.dropna(subset=[value]) d = d.drop_duplicates(subset=keys, keep="first") idx = d[keys[0]] if len(keys) == 1 else list(zip(*(d[k] for k in keys))) return dict(zip(idx, d[value].astype(float))) # ---- typed accessors the estimator uses --------------------------- def tvu(self, country: str) -> float | None: return self._cache("tvu", lambda: self._first_wins( self.frames["tvuniverse"], ["country"], "tvuniverse_000")).get(country) def share(self, country: str, channel: str) -> float | None: return self._cache("share", lambda: self._first_wins( self.frames["gl_ratings"], ["country", "channel"], "ti_total_day_share")).get((country, channel)) def telecast_mult(self, t: str) -> float: # The target row's raw value is canonicalised the same way the evidence # pool was, so "Magazine/Studio" and "STUDIO" resolve to one multiplier # instead of the caller silently taking the unknown fallback. d = self._cache("telecast", lambda: self._first_wins( self.frames["telecast_multipliers"], ["telecast_type"], "multiplier")) key = canonical_telecast(t) if key in d: return d[key] # Unknown category: prefer the measured UNKNOWN multiplier over the # legacy 0.15 constant, which measurement showed to be ~3x too low. return d.get("UNKNOWN", 0.15) def weekday_w(self, wd: int) -> float: return self._cache("weekday", lambda: self._first_wins( self.frames["weekday_weights"], ["weekday"], "weight")).get(int(wd), 1.0) def hour_w(self, h: int) -> float: return self._cache("hour", lambda: self._first_wins( self.frames["hour_weights"], ["hour"], "weight")).get(int(h), 0.5) def team_w(self, team: str) -> float: # NOTE: team_name is not unique in the real table (28,102 rows / 8,062 # teams, keyed by event+season). First-match-wins is preserved here to # match previous behaviour -- see handoff/schema_diff.md for the open # question about disambiguating by event. return self._cache("team", lambda: self._first_wins( self.frames["team_weights"], ["team"], "weight")).get(team, 1.0) def host_f(self, event: str, country: str) -> float: return self._cache("host", lambda: self._first_wins( self.frames["host_factor"], ["event_name", "host_country"], "factor")).get((event, country), 1.0) # ---- sport affinity (cross-market projection) --------------------- def sport_of(self, event: str) -> str | None: """Which sport category an event belongs to.""" if "event_category" not in self.frames: return None d = self._cache("evcat", lambda: dict( self.frames["event_category"] .drop_duplicates(subset=["event_name"], keep="first") [["event_name", "sports_category"]].itertuples(index=False, name=None))) return d.get(event) def affinity(self, country: str, sport: str) -> float | None: """Share of a country's population that follows this sport (percent). Prefers the linear-TV figure where present — it measures watching on TV rather than merely following the sport — and falls back to overall follow_percentage, which is populated far more widely. Duplicate (country, sport) rows exist (e.g. New Zealand lists Rugby at both 18% and 8.2%), so take the MAX rather than an arbitrary first match: the higher figure is the broader definition of following. """ if "sport_interest" not in self.frames: return None def build(): df = self.frames["sport_interest"].copy() df["v"] = (df["interest_lineartv_avg_percentage"] .fillna(df["follow_percentage"]).astype(float)) g = df.groupby(["country", "sports_category"])["v"].max() return {k: float(v) for k, v in g.items()} return self._cache("affinity", build).get((country, sport)) def affinity_ratio(self, event: str, target_country: str, source_country: str) -> float | None: """Target-vs-source appetite for this event's sport. Returns None when either side is unknown, so the caller can decline to project rather than silently assuming parity. """ sport = self.sport_of(event) if not sport: return None a_t = self.affinity(target_country, sport) a_s = self.affinity(source_country, sport) if not a_t or not a_s: return None return a_t / a_s # ---- analog-market similarity ------------------------------------- def sub_region(self, country: str) -> str | None: def build(): df = self.frames["tvuniverse"] if "world_sub_region" not in df.columns: return {} d = df.dropna(subset=["world_sub_region"]).drop_duplicates("country") return dict(zip(d["country"], d["world_sub_region"])) return self._cache("subregion", build).get(country) def gdp(self, country: str) -> float | None: def build(): df = self.frames["tvuniverse"] if "gdp_billion_usd" not in df.columns: return {} return self._first_wins(df, ["country"], "gdp_billion_usd") return self._cache("gdp", build).get(country) def similarity(self, target: str, source: str, event: str) -> float | None: """Distance between two markets for projecting THIS event. Lower is closer. Three terms, all in log space so ratios either way cost the same: appetite — how differently the two markets follow the sport scale — how differently sized their economies are, which stands in for TV spend, channel count and production values geography — a flat penalty for leaving the sub-region Returns None when appetite cannot be measured, so the caller can fall back rather than rank on scale alone. """ sport = self.sport_of(event) a_t = self.affinity(target, sport) if sport else None a_s = self.affinity(source, sport) if sport else None if not a_t or not a_s: return None d = abs(np.log(a_t / a_s)) g_t, g_s = self.gdp(target), self.gdp(source) if g_t and g_s: d += 0.5 * abs(np.log(g_t / g_s)) r_t, r_s = self.sub_region(target), self.sub_region(source) if r_t and r_s and r_t != r_s: d += 0.25 return float(d) def ott(self, country: str) -> tuple[float, float] | None: """(tv_to_ott, out_of_home) multipliers for a market, or None.""" def build(): df = self.frames.get("ott_multipliers") if df is None or df.empty: return {} d = df.dropna(subset=["tv_to_ott_multiplier"]).drop_duplicates("country") return {r.country: (float(r.tv_to_ott_multiplier), float(r.ooh_multiplier) if pd.notna(r.ooh_multiplier) else 0.0) for r in d.itertuples(index=False)} return self._cache("ott", build).get(country) def evidence(self) -> pd.DataFrame: return self.frames["global_sports"] def flagship_share(self, country: str) -> float | None: def build(): df = self.frames["gl_ratings"].dropna(subset=["ti_total_day_share"]) return df.groupby("country")["ti_total_day_share"].max().astype(float).to_dict() return self._cache("flagship", build).get(country) def validate_schedule(df: pd.DataFrame) -> list[str]: """Plain-language problems the dashboard can show. Empty list = fine.""" problems = [] required = ["event_name", "country_name", "channel_name", "telecast_type"] for c in required: if c not in df.columns: problems.append(f"The schedule is missing a '{c}' column.") if "hour" not in df.columns and "broadcast_time" not in df.columns: problems.append("Add an 'hour' (0-23) or 'broadcast_time' column.") if len(df) == 0: problems.append("The schedule has no rows.") return problems