"""Core estimation: evidence retrieval by tier, projection, factors, caps, confidence, and empirical prediction intervals. Evidence tiers (Crystal's hierarchy, preserved): 1 exact same event + country + channel, LIVE history 2 same_teams same normalized teams (fuzzy >= .90), same country, LIVE 3 same_event_country same event + country, other channel (share-adjusted) 4 cross_market same event, other country (TVU + share projected) 5 slot_stats same country+channel time-slot statistics Estimate = tier evidence de-conditioned to a LIVE/reference base, then re-conditioned to the target row via curves/multipliers/factors (log-space). """ from __future__ import annotations import re from difflib import SequenceMatcher import numpy as np import pandas as pd from .reference import ReferenceData TIER_NAMES = {1: "Same event on this channel", 2: "Same teams in this market", 3: "Same event in this market", 4: "Same event in another market", 5: "This channel's time-slot history"} TIER_CONF = {1: 92.0, 2: 88.0, 3: 82.0, 4: 72.0, 5: 58.0} CLIP_SHARE = (0.30, 3.00) CLIP_TVU = (1e-6, 20.0) # Affinity ratios span a genuinely wide range (cricket: ~22% in New Zealand # vs ~90%+ in the Caribbean), so this clip is looser than the share clip -- # but still bounded, because a raw ratio from two sparsely-measured countries # can be extreme without being real. CLIP_AFFINITY = (0.10, 10.0) HARD_MAX_TVU_SHARE = 0.18 FLAG_TVU_SHARE = 0.15 REVIEW_MIN_CONF = 65.0 def norm_teams(s: str) -> str: toks = sorted({re.sub(r"[^a-z0-9]", "", t.lower()) for t in str(s).split("|")} - {""}) return "|".join(toks) def teams_match(a: str, b: str, thresh: float = 0.90) -> bool: return a == b or SequenceMatcher(None, a, b).ratio() >= thresh def _debase(ev: pd.DataFrame, ref: ReferenceData) -> pd.Series: """Strip each observation to a comparable base: base = ama / (hour_w * weekday_w * telecast_mult * team_w * host_f). Evidence-side behavioural factors MUST be removed here, or re-applying them to the target double-counts them (the exact compounding failure documented in the Crystal audit).""" hw = ev["hour"].map(lambda h: ref.hour_w(int(h))) ww = ev["weekday"].map(lambda w: ref.weekday_w(int(w))) tm = ev["telecast_type"].map(ref.telecast_mult) def _tw(teams): ts = [t.strip() for t in str(teams).split("|") if t.strip()] import numpy as _np return float(_np.mean([ref.team_w(t) for t in ts])) if ts else 1.0 tw = ev["teams"].map(_tw) hf = ev.apply(lambda r: ref.host_f(r["event_name"], r["country"]), axis=1) return ev["ama_000"] / (hw * ww * tm * tw * hf).clip(lower=1e-9) MIN_SOURCE_ROWS = 20 # below this a market's de-based median is too noisy to project from # How much evidence a tier must hold before it is allowed to answer. Previously # tiers 1 and 3 accepted a single row, so one Pakistani broadcast became the # basis for 394 estimates and scored 92 confidence, while India -- with 5,791 # rows -- came out 15x lower because the median of a deep, mixed pool sits far # below any one marquee broadcast. Thin evidence must fall through to the next # tier, not masquerade as the strongest one. MIN_TIER_ROWS = {1: 3, 2: 3, 3: 5, 4: 2, 5: 3} def _nearest_channel(chans: dict, country: str, channel: str, ref) -> str | None: """Pick the in-market channel closest in day-share to the target. `chans` is {channel: (median_base, n)} precomputed by fastindex. Returns None when no channel is deep enough or no share is known, and the caller then falls back to the whole-market median. """ sh_t = ref.share(country, channel) if not sh_t: return None scored = [] for ch, (_m, n) in chans.items(): if n < MIN_TIER_ROWS[3]: continue s = ref.share(country, ch) if s and s > 0: scored.append((abs(np.log(sh_t / s)), ch)) return min(scored)[1] if scored else None def _depth_adjust(conf: float, n: int) -> float: """Move confidence with the weight of evidence, not by a token amount. The old bonus was min(6, 1.5*log10 n): one row scored 92 and five thousand scored 98. Six points across a 5,000x difference in evidence is not a confidence score, it is decoration. This rewards depth on the same scale it penalises thinness -- roughly -14 at one row, neutral at ten, +6 at a hundred or more. """ depth = float(np.log10(max(int(n), 1))) return conf + 3.0 * min(depth, 2.0) - 14.0 * max(0.0, 1.0 - depth) def _pick_source_fast(cands: dict, target: str, event: str, ref) -> str: """Choose the analog market to project this event from. `cands` is {country: (median_base, n)}. Markets with at least MIN_SOURCE_ROWS observations are ranked by ReferenceData.similarity and the closest wins; the row floor is dropped rather than the estimate refused if nothing clears it. Falls back to the most-rows market only when similarity cannot be measured for any candidate. """ eligible = [c for c, (_m, n) in cands.items() if n >= MIN_SOURCE_ROWS and c != target] or \ [c for c in cands if c != target] or list(cands) scored = [(ref.similarity(target, c, event), c) for c in eligible] scored = [(d, c) for d, c in scored if d is not None] if scored: return min(scored)[1] return max(eligible, key=lambda c: cands[c][1]) def estimate_row(row: dict, ref: ReferenceData, ev_indexed: dict, correction_by_tier: dict | None = None) -> dict: event, country, channel = row["event_name"], row["country_name"], row["channel_name"] telecast = str(row["telecast_type"]).strip().upper().replace(" ", "_") hour = int(row.get("hour", 19)); wd = int(row.get("weekday", 5)) tkey = norm_teams(row.get("sports_teams", "")) ev = ev_indexed # dict of pre-filtered frames from .fastindex import match_tkey tier, base, n_ev, src_channel = None, None, 0, None hit = ev["ecc"].get((event, country, channel)) if hit and hit[1] >= MIN_TIER_ROWS[1]: tier, base, n_ev = 1, hit[0], hit[1] if tier is None and tkey: hit = match_tkey(ev, country, tkey) if hit and hit[1] >= MIN_TIER_ROWS[2]: tier, base, n_ev = 2, hit[0], hit[1] if tier is None: chans = ev["ec_channels"].get((event, country)) if chans: total = sum(v[1] for v in chans.values()) if total >= MIN_TIER_ROWS[3]: # One comparable channel beats a median across all of them. src_channel = _nearest_channel(chans, country, channel, ref) if src_channel: base, n_ev = chans[src_channel] else: whole = ev["ec"].get((event, country)) base, n_ev = whole tier = 3 if tier is None: cands = ev["e_countries"].get(event) if cands and len(cands) >= 2: tier = 4 if tier is None: hit = ev["cc"].get((country, channel)) if hit and hit[1] >= MIN_TIER_ROWS[5]: tier, base, n_ev = 5, hit[0], hit[1] out = {"strategy_tier": tier, "strategy": TIER_NAMES.get(tier, "No evidence"), "evidence_rows": int(n_ev), "estimated_ama_000": np.nan, "confidence": np.nan, "flag": "", "cap_applied": "", "formula": "", "source_channel": "", "source_country": ""} if tier is None: out["flag"] = "NO_EVIDENCE" return out # cross-market projection for tier 4: rescale by TVU, share and affinity proj_note = "" if tier == 4: cands = ev["e_countries"][event] src_country = _pick_source_fast(cands, country, event, ref) base, n_ev = cands[src_country] out["evidence_rows"] = int(n_ev) # The analog market is the single most useful thing to show an analyst # who wants to sanity-check a projected number. out["source_country"] = src_country tvu_t, tvu_s = ref.tvu(country), ref.tvu(src_country) if tvu_t is None or tvu_s is None: out["flag"] = "MISSING_TVU_REFUSED" # Crystal's honest refusal, kept return out beta = ref.exponents() # Exponents are fitted from evidence where a calibration file exists, # and 1.0 otherwise. Measured on 5.9M rows they come out well below 1: # doubling a market's TV universe does not double its audience. r_tvu = float(np.clip((tvu_t / tvu_s) ** beta["tvu"], *CLIP_TVU)) # shares must be RELATIVE TO EACH MARKET'S FLAGSHIP: absolute shares # across markets leave a stray flagship_t/flagship_s factor. fl_t, fl_s = ref.flagship_share(country), ref.flagship_share(src_country) sh_t = ref.share(country, channel) or fl_t rel_t = (sh_t or 1.0) / (fl_t or 1.0) rel_s = ev["ec_rel_share"].get((event, src_country)) or 1.0 r_sh = float(np.clip((rel_t / rel_s) ** beta["share"], *CLIP_SHARE)) base *= r_tvu * r_sh proj_note = f" * tvu_ratio({r_tvu:.3f}) * share_ratio({r_sh:.3f})" # Sport affinity. TVU says how many people COULD watch; affinity says # how many would care. Skipped (not assumed to be 1.0) when either # country's affinity is unknown, so an unmeasurable projection stays # flagged rather than silently confident. r_aff = ref.affinity_ratio(event, country, src_country) if r_aff is not None: r_aff = float(np.clip(r_aff ** beta["affinity"], *CLIP_AFFINITY)) base *= r_aff proj_note += f" * affinity_ratio({r_aff:.3f})" else: out["flag"] = "NO_AFFINITY_DATA" elif tier in (2, 3, 5): # channel adjustment inside the market when the evidence came off a # different channel. With tier 3 now pooled on ONE channel this ratio is # small by construction, which is the point: a big correction is where # the clip and the linear-scaling assumption did the damage. sh_t = ref.share(country, channel) sh_pool = ref.share(country, src_channel) if src_channel else None if sh_t and sh_pool: r = float(np.clip(sh_t / sh_pool, *CLIP_SHARE)) base *= r proj_note = f" * share_ratio({r:.3f})" # re-condition to the target slot/type + behavioural factors tm = ref.telecast_mult(telecast) hw, ww = ref.hour_w(hour), ref.weekday_w(wd) teams = [t.strip() for t in str(row.get("sports_teams", "")).split("|") if t.strip()] tw = float(np.mean([ref.team_w(t) for t in teams])) if teams else 1.0 hf = ref.host_f(event, country) est = base * hw * ww * tm * tw * hf corr = float((correction_by_tier or {}).get(tier, 1.0)) corr = float(np.clip(corr, 0.5, 1.8)) # Crystal's calibrator clamp, kept est *= corr # constraints (alarm + clamp, Crystal-compatible; audit preserved) tvu_t = ref.tvu(country) if tvu_t: share_of_tvu = est / tvu_t if share_of_tvu > HARD_MAX_TVU_SHARE: out["cap_applied"] = f"HARD_MAX_18PCT (was {est:.1f})" est = HARD_MAX_TVU_SHARE * tvu_t elif share_of_tvu > FLAG_TVU_SHARE: out["flag"] = "REVIEW_ABOVE_15PCT_TVU" conf = _depth_adjust(TIER_CONF[tier], n_ev) conf -= 6.0 if telecast not in ("LIVE",) and tier >= 4 else 0 # weak+converted conf = float(np.clip(conf, 5, 99)) if n_ev < 10 and not out["flag"]: out["flag"] = "THIN_EVIDENCE" if conf < REVIEW_MIN_CONF and not out["flag"]: out["flag"] = "REVIEW_LOW_CONFIDENCE" if src_channel: out["source_channel"] = src_channel out["evidence_rows"] = int(n_ev) out.update({ "estimated_ama_000": round(float(est), 2), "confidence": round(conf, 1), "formula": (f"median_debased_evidence({base:.1f}){proj_note}" f" * hour_w({hw:.2f}) * weekday_w({ww:.2f})" f" * telecast({tm:.2f}) * team_w({tw:.2f}) * host({hf:.2f})" + (f" * calib({corr:.3f})" if corr != 1.0 else "")), }) return out def index_evidence(ref: ReferenceData, cache_dir=None) -> dict: """Precompute every group median once. See engine/fastindex.""" from .fastindex import build_cached return build_cached(ref, cache_dir or getattr(ref, "source_dir", None)) def attach_intervals(df: pd.DataFrame, residual_sigma_by_tier: dict | None) -> pd.DataFrame: """80% interval from per-tier log-residual sigma (empirical when supplied, conservative defaults otherwise). z(80%) = 1.2816.""" default = {1: 0.18, 2: 0.22, 3: 0.28, 4: 0.40, 5: 0.50} sig = {**default, **(residual_sigma_by_tier or {})} z = 1.2816 lo, hi = [], [] for _, r in df.iterrows(): s, e = sig.get(r["strategy_tier"], 0.5), r["estimated_ama_000"] if pd.isna(e): lo.append(np.nan); hi.append(np.nan) else: lo.append(round(e * float(np.exp(-z * s)), 2)) hi.append(round(e * float(np.exp(z * s)), 2)) df["interval80_low"], df["interval80_high"] = lo, hi return df