"""Precomputed evidence index. The original index handed `estimate_row` a dict of pandas DataFrames and let it do the arithmetic per target row. That cost two things: * `_debase` ran per row per pool, with `.map(lambda)` and `.apply(axis=1)` over thousands of evidence rows -- Python-level work repeated for every target that happened to share a pool. * tier 2 ran `SequenceMatcher` against EVERY live row in the market. India has 228,000 of them, so a single target row cost 228,000 fuzzy string comparisons. Profiling a 10-row schedule showed 458,062 calls and 16.6 of 18 seconds spent inside difflib. Neither is necessary. The de-based value of an evidence row does not depend on the target, and the median of a group does not either, so both are computed once here. Fuzzy team matching only ever needs to consider the DISTINCT team strings in a market -- a few thousand, not a few hundred thousand -- and only those sharing a token with the target. What `estimate_row` receives is therefore plain dicts of floats. No pandas touches the hot path. """ from __future__ import annotations from collections import defaultdict import numpy as np import pandas as pd def debase_column(ev: pd.DataFrame, ref) -> np.ndarray: """De-base every evidence row at once. base = ama / (hour_w * weekday_w * telecast_mult * team_w * host_f) Identical arithmetic to the row-wise version, but each factor is resolved over the DISTINCT values present and then mapped, so the Python-level work is proportional to the number of distinct hours / teams / events rather than to the 5.9M rows. """ n = len(ev) hw = ev["hour"].astype(int).map( {h: ref.hour_w(h) for h in ev["hour"].astype(int).unique()}).to_numpy(float) ww = ev["weekday"].astype(int).map( {w: ref.weekday_w(w) for w in ev["weekday"].astype(int).unique()}).to_numpy(float) tm = ev["telecast_type"].map( {t: ref.telecast_mult(t) for t in ev["telecast_type"].unique()}).to_numpy(float) def team_mean(s): parts = [p.strip() for p in str(s).split("|") if p.strip()] return float(np.mean([ref.team_w(p) for p in parts])) if parts else 1.0 teams = ev["teams"].fillna("") tw = teams.map({t: team_mean(t) for t in teams.unique()}).to_numpy(float) pairs = list(zip(ev["event_name"], ev["country"])) hf_map = {p: ref.host_f(p[0], p[1]) for p in set(pairs)} hf = np.fromiter((hf_map[p] for p in pairs), dtype=float, count=n) denom = np.clip(hw * ww * tm * tw * hf, 1e-9, None) return ev["ama_000"].to_numpy(float) / denom def _med_size(df: pd.DataFrame, keys, col="base") -> dict: """{key: (median, count)} for one grouping, as plain Python floats.""" g = df.groupby(keys, sort=False)[col].agg(["median", "size"]) return {k: (float(m), int(s)) for k, m, s in zip(g.index, g["median"].to_numpy(), g["size"].to_numpy())} CACHE_NAME = "evidence_index.pkl" def build_cached(ref, cache_dir=None) -> dict: """Load the prebuilt index if one exists, else build and save it. The index is a pure function of the reference snapshot, so it only has to be rebuilt when the snapshot is. Building it takes ~14s over 5.9M rows; loading it takes about a second, and every run paid the 14s before. """ import pickle from pathlib import Path if cache_dir: f = Path(cache_dir) / CACHE_NAME if f.exists(): try: with f.open("rb") as fh: idx = pickle.load(fh) idx["_fuzzy_cache"] = {} # memo is per-process, never persisted return idx except Exception: # noqa: BLE001 — a stale pickle must not stop a run pass idx = build(ref) if cache_dir: try: f = Path(cache_dir) / CACHE_NAME tmp = f.with_suffix(".tmp") keep = {k: v for k, v in idx.items() if k != "_fuzzy_cache"} with tmp.open("wb") as fh: pickle.dump(keep, fh, protocol=pickle.HIGHEST_PROTOCOL) tmp.replace(f) except Exception: # noqa: BLE001 pass return idx def build(ref) -> dict: """Everything `estimate_row` needs, precomputed once.""" ev = ref.evidence().copy() ev["base"] = debase_column(ev, ref) ev = ev[np.isfinite(ev["base"]) & (ev["base"] > 0)] from .core import norm_teams live = ev[ev["telecast_type"] == "LIVE"].copy() live["tkey"] = live["teams"].fillna("").map( {t: norm_teams(t) for t in live["teams"].fillna("").unique()}) idx: dict = { # tier 1 / 3 / 4 / 5 group medians "ecc": _med_size(live, ["event_name", "country", "channel"]), "ec": _med_size(live, ["event_name", "country"]), "cc": _med_size(ev, ["country", "channel"]), } # tier 3 needs each channel's own level inside the market, so it can pick # the channel nearest the target rather than averaging all of them. by_ec_ch: dict = defaultdict(dict) for (e, c, ch), v in idx["ecc"].items(): by_ec_ch[(e, c)][ch] = v idx["ec_channels"] = dict(by_ec_ch) # tier 4 needs, per event, each candidate source market's level and the # median relative share of the channels that carried it. by_e_c: dict = defaultdict(dict) for (e, c), v in idx["ec"].items(): by_e_c[e][c] = v idx["e_countries"] = dict(by_e_c) rel: dict = {} for (e, c), chans in by_ec_ch.items(): fl = ref.flagship_share(c) if not fl: continue vals = [ (ref.share(c, ch) or 0) / fl for ch in chans if ref.share(c, ch) ] if vals: rel[(e, c)] = float(np.median(vals)) idx["ec_rel_share"] = rel # tier 2: distinct team strings per market, plus a token index so fuzzy # matching only ever looks at strings that share a word with the target. tk_med: dict = defaultdict(dict) for (c, tk), v in _med_size(live[live["tkey"] != ""], ["country", "tkey"]).items(): tk_med[c][tk] = v idx["tkeys"] = dict(tk_med) tok: dict = defaultdict(lambda: defaultdict(set)) for c, d in tk_med.items(): for tk in d: for t in tk.split("|"): if t: tok[c][t].add(tk) idx["tkey_tokens"] = {c: dict(v) for c, v in tok.items()} idx["_fuzzy_cache"] = {} return idx def match_tkey(idx: dict, country: str, tkey: str, thresh: float = 0.90): """Team pool for a market, exact first and fuzzy only over plausible strings. Returns (median, count) or None. Preserves the original semantics -- exact equality, else SequenceMatcher ratio >= thresh -- but compares against the market's DISTINCT team strings that share at least one token with the target, instead of every row in the market. Results are memoised, because a schedule asks the same question many times. """ if not tkey: return None per_country = idx["tkeys"].get(country) if not per_country: return None hit = per_country.get(tkey) if hit is not None: return hit ck = (country, tkey) cache = idx["_fuzzy_cache"] if ck in cache: return cache[ck] from .core import teams_match toks = [t for t in tkey.split("|") if t] tokmap = idx["tkey_tokens"].get(country, {}) cands: set = set() for t in toks: cands |= tokmap.get(t, set()) best = None for cand in cands: if teams_match(cand, tkey, thresh): m, n = per_country[cand] if best is None or n > best[1]: best = (m, n) cache[ck] = best return best