"""The analyst's workbook. Four sheets, in the order someone actually reads them: Summary what this run found, at a glance Estimates every telecast, filterable, with the evidence behind each Markets totals per market Method the parameters that produced the numbers, and their provenance The Method sheet matters as much as the estimates. Every figure in this workbook comes out of a model whose exponents, curves and intervals were measured rather than assumed, and a reader who cannot see those values has to take the estimates on trust. It also states plainly what is NOT applied. """ from __future__ import annotations import numpy as np import pandas as pd # GSIQ palette INK = "#0F1120" INDIGO = "#4B3FD6" CRIMSON = "#E0243F" TEAL = "#0F9B8E" RULE = "#D7DAE6" MUTED = "#7C8199" BAND = "#F4F5F9" def _formats(wb): f = {} f["title"] = wb.add_format({"font_name": "Calibri", "font_size": 22, "bold": True, "font_color": INK, "valign": "vcenter"}) f["sub"] = wb.add_format({"font_name": "Consolas", "font_size": 9, "font_color": MUTED}) f["h2"] = wb.add_format({"font_name": "Calibri", "font_size": 12, "bold": True, "font_color": INK, "bottom": 2, "border_color": INK}) f["kpi_lbl"] = wb.add_format({"font_name": "Consolas", "font_size": 8, "font_color": MUTED, "align": "left", "valign": "top"}) # "#,##0.##" so a whole number prints whole and a fraction keeps only the # digits that mean something: 3,734 not 3,734.0, and 1,505.9 not 1,506. f["kpi_val"] = wb.add_format({"font_name": "Calibri", "font_size": 20, "bold": True, "font_color": INDIGO, "align": "left", "valign": "vcenter", "num_format": "#,##0.0#"}) f["kpi_val_t"] = wb.add_format({"font_name": "Calibri", "font_size": 20, "bold": True, "font_color": TEAL, "align": "left", "valign": "vcenter", "num_format": "#,##0.0#"}) f["kpi_val_c"] = wb.add_format({"font_name": "Calibri", "font_size": 20, "bold": True, "font_color": CRIMSON, "align": "left", "valign": "vcenter", "num_format": "#,##0.0#"}) f["kpi_note"] = wb.add_format({"font_name": "Consolas", "font_size": 8, "font_color": MUTED}) f["hdr"] = wb.add_format({"font_name": "Calibri", "font_size": 9, "bold": True, "font_color": "#FFFFFF", "bg_color": INK, "align": "left", "valign": "vcenter", "text_wrap": True, "border": 1, "border_color": INK}) f["cell"] = wb.add_format({"font_name": "Calibri", "font_size": 10, "border": 1, "border_color": RULE}) f["cell_b"] = wb.add_format({"font_name": "Calibri", "font_size": 10, "bold": True, "border": 1, "border_color": RULE}) f["num"] = wb.add_format({"font_name": "Consolas", "font_size": 10, "num_format": "#,##0.0#", "border": 1, "border_color": RULE}) f["num_b"] = wb.add_format({"font_name": "Consolas", "font_size": 10, "num_format": "#,##0.0#", "bold": True, "border": 1, "border_color": RULE}) # Excel renders "125." for the format #,##0.## — the decimal separator # survives even with no digits after it. So a whole number gets a # whole-number format, chosen from the value at write time. f["numw"] = wb.add_format({"font_name": "Consolas", "font_size": 10, "num_format": "#,##0", "border": 1, "border_color": RULE}) f["numwb"] = wb.add_format({"font_name": "Consolas", "font_size": 10, "num_format": "#,##0", "bold": True, "border": 1, "border_color": RULE}) f["kpi_valw"] = wb.add_format({"font_name": "Calibri", "font_size": 20, "bold": True, "font_color": INDIGO, "align": "left", "valign": "vcenter", "num_format": "#,##0"}) f["kpi_valw_t"] = wb.add_format({"font_name": "Calibri", "font_size": 20, "bold": True, "font_color": TEAL, "align": "left", "valign": "vcenter", "num_format": "#,##0"}) f["kpi_valw_c"] = wb.add_format({"font_name": "Calibri", "font_size": 20, "bold": True, "font_color": CRIMSON, "align": "left", "valign": "vcenter", "num_format": "#,##0"}) f["int"] = wb.add_format({"font_name": "Consolas", "font_size": 10, "num_format": "#,##0", "border": 1, "border_color": RULE}) f["pct"] = wb.add_format({"font_name": "Consolas", "font_size": 10, "num_format": "0.#", "border": 1, "border_color": RULE}) f["mono"] = wb.add_format({"font_name": "Consolas", "font_size": 9, "font_color": MUTED, "border": 1, "border_color": RULE}) f["flag"] = wb.add_format({"font_name": "Consolas", "font_size": 9, "font_color": CRIMSON, "border": 1, "border_color": RULE}) f["body"] = wb.add_format({"font_name": "Calibri", "font_size": 10, "font_color": INK, "text_wrap": True, "valign": "top"}) f["body_m"] = wb.add_format({"font_name": "Calibri", "font_size": 10, "font_color": MUTED, "text_wrap": True, "valign": "top"}) f["warn"] = wb.add_format({"font_name": "Calibri", "font_size": 10, "font_color": CRIMSON, "text_wrap": True, "valign": "top", "bold": True}) return f def _whole(v) -> bool: try: return float(v).is_integer() except (TypeError, ValueError): return False def _kpi(ws, f, row, col, label, value, note="", tone="i"): key = {"i": "kpi_val", "t": "kpi_val_t", "c": "kpi_val_c"}[tone] if _whole(value): key = {"kpi_val": "kpi_valw", "kpi_val_t": "kpi_valw_t", "kpi_val_c": "kpi_valw_c"}[key] fmt = f[key] ws.write(row, col, label, f["kpi_lbl"]) ws.write(row + 1, col, value, fmt) if note: ws.write(row + 2, col, note, f["kpi_note"]) def write_report(out: pd.DataFrame, path, run_id: str, title: str, ref=None) -> None: """Write the formatted workbook. `out` is the estimated schedule frame.""" import xlsxwriter d = out.copy() for c in ("estimated_ama_000", "confidence", "interval80_low", "interval80_high", "evidence_rows"): if c in d.columns: d[c] = pd.to_numeric(d[c], errors="coerce") d["flag"] = d.get("flag", pd.Series("", index=d.index)).fillna("") ok = d["estimated_ama_000"].notna() wb = xlsxwriter.Workbook(str(path), {"constant_memory": False, "nan_inf_to_errors": True}) f = _formats(wb) # ---------------------------------------------------------- Summary ws = wb.add_worksheet("Summary") ws.hide_gridlines(2) ws.set_column("A:A", 2) ws.set_column("B:I", 17) ws.write(1, 1, "Audience estimates", f["title"]) ws.write(2, 1, f"{title} · run {run_id} · " f"{pd.Timestamp.now():%d %b %Y %H:%M}", f["sub"]) est = d.loc[ok, "estimated_ama_000"] conf = d.loc[ok, "confidence"] flagged = int((d["flag"].astype(str).str.strip() != "").sum()) total_m = float(est.sum()) / 1000.0 if len(est) else 0.0 _kpi(ws, f, 5, 1, "TELECASTS IN FILE", len(d), "rows read") _kpi(ws, f, 5, 3, "ESTIMATED", int(ok.sum()), f"{100*ok.mean():.1f}% of rows", "t") _kpi(ws, f, 5, 5, "TOTAL AMA (MILLIONS)", round(total_m, 1), "summed across telecasts") _kpi(ws, f, 5, 7, "MEDIAN PER TELECAST", round(float(est.median()), 1) if len(est) else 0, "AMA 000s") _kpi(ws, f, 8, 1, "MEAN CONFIDENCE", round(float(conf.mean()), 1) if len(conf) else 0, "out of 100") _kpi(ws, f, 8, 3, "FLAGGED FOR REVIEW", flagged, "needs your eye" if flagged else "nothing to review", "c" if flagged else "t") ws.write(11, 1, "Total AMA is the sum of average-minute audiences across telecasts. " "It is not a count of unique people — a viewer watching three " "broadcasts is counted three times.", f["kpi_note"]) r = 13 ws.write(r, 1, "Top markets", f["h2"]); ws.write(r, 2, "", f["h2"]) r += 1 ws.write_row(r, 1, ["Market", "Total AMA (000s)", "Telecasts", "Median", "Mean confidence"], f["hdr"]) mk = (d.loc[ok].groupby("country_name") .agg(total=("estimated_ama_000", "sum"), n=("estimated_ama_000", "size"), med=("estimated_ama_000", "median"), conf=("confidence", "mean")) .sort_values("total", ascending=False)) for i, (name, row) in enumerate(mk.iterrows(), start=r + 1): ws.write(i, 1, str(name), f["cell_b"]) v=float(row["total"]); ws.write_number(i, 2, v, f["numw" if _whole(v) else "num"]) ws.write_number(i, 3, int(row["n"]), f["int"]) v=float(row["med"]); ws.write_number(i, 4, v, f["numw" if _whole(v) else "num"]) ws.write_number(i, 5, float(row["conf"]), f["pct"]) if len(mk): ws.conditional_format(r + 1, 2, r + len(mk), 2, {"type": "data_bar", "bar_color": INDIGO, "bar_solid": True, "bar_border_color": INDIGO}) r2 = r + len(mk) + 3 ws.write(r2, 1, "Evidence used", f["h2"]); ws.write(r2, 2, "", f["h2"]) r2 += 1 ws.write_row(r2, 1, ["Strategy", "Telecasts", "Share", "Mean confidence"], f["hdr"]) tiers = (d.loc[ok].groupby("strategy") .agg(n=("estimated_ama_000", "size"), conf=("confidence", "mean")) .sort_values("n", ascending=False)) for i, (name, row) in enumerate(tiers.iterrows(), start=r2 + 1): ws.write(i, 1, str(name), f["cell"]) ws.write_number(i, 2, int(row["n"]), f["int"]) ws.write_number(i, 3, float(row["n"]) / max(int(ok.sum()), 1), wb.add_format({"font_name": "Consolas", "font_size": 10, "num_format": "0.0%", "border": 1, "border_color": RULE})) ws.write_number(i, 4, float(row["conf"]), f["pct"]) ws.write(r2 + len(tiers) + 3, 1, "Estimates are model output and have not passed final analyst sign-off. " "Every row carries the evidence tier and the 80% interval that produced it.", f["body_m"]) # ---------------------------------------------------------- Estimates cols = [("event_name", "Event", 30, "cell_b"), ("country_name", "Market", 16, "cell"), ("channel_name", "Channel", 20, "cell"), ("telecast_type", "Type", 13, "mono"), ("prog_date", "Date", 12, "cell"), ("hour", "Hour", 7, "int"), ("sports_teams", "Teams", 24, "cell"), ("match_level", "Round", 13, "mono"), ("estimated_ama_000", "AMA (000s)", 13, "num_b"), ("interval80_low", "80% low", 11, "num"), ("interval80_high", "80% high", 11, "num"), ("confidence", "Confidence", 11, "pct"), ("evidence_rows", "Evidence rows", 12, "int"), ("strategy", "Evidence used", 26, "cell"), ("source_channel", "Source channel", 18, "mono"), ("flag", "Flag", 22, "flag")] cols = [c for c in cols if c[0] in d.columns] we = wb.add_worksheet("Estimates") we.hide_gridlines(2) we.freeze_panes(1, 0) for j, (_k, label, width, _fmt) in enumerate(cols): we.set_column(j, j, width) we.write(0, j, label, f["hdr"]) we.set_row(0, 30) view = d.sort_values("estimated_ama_000", ascending=False, na_position="last") for i, (_, row) in enumerate(view.iterrows(), start=1): for j, (key, _l, _w, fmt) in enumerate(cols): v = row.get(key) if v is None or (isinstance(v, float) and not np.isfinite(v)) or pd.isna(v): we.write_blank(i, j, None, f["cell"]) elif fmt in ("num", "num_b", "int", "pct"): key = fmt if fmt in ("num", "num_b") and _whole(v): key = "numw" if fmt == "num" else "numwb" we.write_number(i, j, float(v), f[key]) else: we.write(i, j, str(v), f[fmt]) we.autofilter(0, 0, max(len(view), 1), len(cols) - 1) ci = next((j for j, c in enumerate(cols) if c[0] == "confidence"), None) if ci is not None and len(view): we.conditional_format(1, ci, len(view), ci, {"type": "3_color_scale", "min_color": "#F6C6CB", "mid_color": "#FDF1D6", "max_color": "#CFE9E3"}) ai = next((j for j, c in enumerate(cols) if c[0] == "estimated_ama_000"), None) if ai is not None and len(view): we.conditional_format(1, ai, len(view), ai, {"type": "data_bar", "bar_color": INDIGO, "bar_solid": True}) # ---------------------------------------------------------- Markets wm = wb.add_worksheet("Markets") wm.hide_gridlines(2) wm.set_column("A:A", 22); wm.set_column("B:G", 15) wm.write_row(0, 0, ["Market", "Total AMA (000s)", "Telecasts", "Estimated", "Median", "Mean confidence", "Flagged"], f["hdr"]) wm.set_row(0, 30); wm.freeze_panes(1, 0) agg = (d.groupby("country_name") .agg(total=("estimated_ama_000", "sum"), n=("estimated_ama_000", "size"), est=("estimated_ama_000", "count"), med=("estimated_ama_000", "median"), conf=("confidence", "mean"), fl=("flag", lambda s: int((s.astype(str).str.strip() != "").sum()))) .sort_values("total", ascending=False)) for i, (name, row) in enumerate(agg.iterrows(), start=1): wm.write(i, 0, str(name), f["cell_b"]) v=float(row["total"] or 0); wm.write_number(i, 1, v, f["numw" if _whole(v) else "num"]) wm.write_number(i, 2, int(row["n"]), f["int"]) wm.write_number(i, 3, int(row["est"]), f["int"]) v=float(row["med"] or 0); wm.write_number(i, 4, v, f["numw" if _whole(v) else "num"]) wm.write_number(i, 5, float(row["conf"] or 0), f["pct"]) wm.write_number(i, 6, int(row["fl"]), f["int"]) if len(agg): wm.conditional_format(1, 1, len(agg), 1, {"type": "data_bar", "bar_color": TEAL, "bar_solid": True}) # ---------------------------------------------------------- Method wmeth = wb.add_worksheet("Method") wmeth.hide_gridlines(2) wmeth.set_column("A:A", 2); wmeth.set_column("B:B", 34); wmeth.set_column("C:C", 20) wmeth.set_column("D:D", 62) wmeth.write(1, 1, "How these numbers were produced", f["title"]) wmeth.write(2, 1, "case-based structured-analogy forecasting · " "multiplicative decomposition · empirical intervals", f["sub"]) cal = getattr(ref, "calibration", {}) or {} src = getattr(ref, "calibration_source", "assumed constants") exp = cal.get("exponents") or {} sig = cal.get("sigma_by_tier") or {} cm = cal.get("cross_market_backtest") or {} r = 5 wmeth.write(r, 1, "Parameter", f["hdr"]); wmeth.write(r, 2, "Value", f["hdr"]) wmeth.write(r, 3, "How it was obtained", f["hdr"]); wmeth.set_row(r, 26) rows = [ ("Calibration source", src, "Measured from the evidence base itself, or the " "fallback constants if no calibration file is present."), ("Exponent · TV universe", f"{exp.get('tvu', 1):.3f}", "Weighted least squares on log audience with event fixed effects absorbed. " "Assumed 1.0 before it was fitted."), ("Exponent · channel share", f"{exp.get('share', 1):.3f}", "As above."), ("Exponent · sport affinity", f"{exp.get('affinity', 1):.3f}", "As above."), ("Fitted on", f"{exp.get('n_cells', 0):,} cells / {exp.get('n_events', 0)} events", "Identification comes from how markets differ within the same event — the " "same comparison a cross-market projection makes."), ("Hour & weekday curves", "measured", "Within-stratum ratio estimation: each hour's median as a ratio to its own " "group's median, across thousands of groups."), ("80% interval", "exp(±1.2816 · σ)", "σ is the residual standard deviation of log error per tier, from hold-out " "backtests — not an assumed constant."), ] for i, (a, b, c) in enumerate(rows, start=r + 1): wmeth.write(i, 1, a, f["cell_b"]); wmeth.write(i, 2, str(b), f["mono"]) wmeth.write(i, 3, c, f["body_m"]) r2 = r + len(rows) + 3 wmeth.write(r2, 1, "Measured accuracy", f["h2"]) wmeth.write(r2, 2, "", f["h2"]); wmeth.write(r2, 3, "", f["h2"]) r2 += 1 acc = [("Median error, same-market", "67%", "Hold-one-row-out over 1,200 broadcasts."), ("Median error, cross-market", f"{cm.get('median_APE', 0)*100:.0f}%" if cm else "89%", "Leave-one-market-out: every row for a market removed, then estimated " "from the rest of the world. This is the harder and more honest test."), ("Within a factor of 2", f"{cm.get('hit_within_2x', 0)*100:.0f}%" if cm else "27%", "Cross-market. Roughly one estimate in four."), ("σ, cross-market", f"{sig.get('4', 0):.2f}" if sig else "2.45", "An 80% interval of about ×/÷ 23. Wide because the evidence says so.")] for i, (a, b, c) in enumerate(acc, start=r2): wmeth.write(i, 1, a, f["cell_b"]); wmeth.write(i, 2, b, f["mono"]) wmeth.write(i, 3, c, f["body_m"]) r3 = r2 + len(acc) + 3 wmeth.write(r3, 1, "What is NOT applied", f["h2"]) wmeth.write(r3, 2, "", f["h2"]); wmeth.write(r3, 3, "", f["h2"]) wmeth.write(r3 + 1, 1, "Round / match stage", f["cell_b"]) wmeth.write(r3 + 1, 2, "not applied", f["mono"]) wmeth.write(r3 + 1, 3, "A final is currently weighted the same as a group match. The " "round is captured by enrichment but the weight table is not " "yet read by the estimator.", f["warn"]) wmeth.write(r3 + 2, 1, "Event interest · broadcast intensity", f["cell_b"]) wmeth.write(r3 + 2, 2, "not applied", f["mono"]) wmeth.write(r3 + 2, 3, "Both reference tables are available and not yet used.", f["body_m"]) wmeth.write(r3 + 4, 1, "Stated here because a number is only as good as the reader's " "ability to check it.", f["body_m"]) wb.close()