File: stage-2-08-calculation-script.md
ClearCompare: Kewpie's calculation script
Sample data. Produced by Kewpie at Stage 2 and approved on 21-Oct-26.
"""Kewpie, Stage 2: test Gary's draft measures and calculate every Must Have KPI for ClearCompare.
Usage: python -I kewpie_calculations.py <data folder> <output kpi_results.json>
Sample data only (synthetic). Data cut-off: 30-Sep-26.
Periods: FY26 = Jul-25 to Jun-26 · "last quarter" = Apr-Jun 26 · "this quarter" = Jul-Sep 26.
"""
import csv
import json
import statistics as st
import sys
from collections import defaultdict
from datetime import date, datetime, timedelta
from pathlib import Path
CUTOFF = date(2026, 9, 30)
MATURE = CUTOFF - timedelta(days=90) # sales on or before this date have had 90 days for clawbacks
CLAWBACK_VERTICALS = {"health", "life"}
VERTICALS = ["health", "home_loans", "energy", "life"]
def qtr(d):
return f"{d.year}-Q{(d.month - 1) // 3 + 1}"
def month(d):
return f"{d.year}-{d.month:02d}"
FY26 = {f"2025-{m:02d}" for m in range(7, 13)} | {f"2026-{m:02d}" for m in range(1, 7)}
LAST_Q, THIS_Q = "2026-Q2", "2026-Q3"
LAST3 = ["2026-06", "2026-07", "2026-08", "2026-09"] # run rate = average monthly change Jun to Sep
def load(folder):
rd = lambda n: list(csv.DictReader(open(folder / n, encoding="utf8")))
leads, contacts, sales, spend, complaints = (rd(n) for n in
["leads.csv", "contacts.csv", "sales.csv", "marketing_spend.csv", "complaints.csv"])
for r in leads:
r["dt"] = datetime.fromisoformat(r["created_ts"])
for r in contacts:
r["dt"] = datetime.fromisoformat(r["first_contact_ts"])
for r in sales:
r["d"] = date.fromisoformat(r["sale_date"])
r["gross"] = float(r["gross_commission"])
r["claw"] = float(r["clawback_amount"] or 0)
r["cbd"] = date.fromisoformat(r["clawback_date"]) if r["clawback_date"] else None
for r in spend:
r["d"] = date.fromisoformat(r["date"])
r["cost"] = float(r["cost"])
for r in complaints:
r["d"] = date.fromisoformat(r["date"])
return leads, contacts, sales, spend, complaints
def main():
folder, out = Path(sys.argv[1]), Path(sys.argv[2])
leads, contacts, sales, spend, complaints = load(folder)
lead_by_id = {r["lead_id"]: r for r in leads}
R = {"data_cutoff": str(CUTOFF), "label": "Sample data (synthetic)"}
# ---------- Data checks (confirm Gary's data mapping) ----------
def nulls(rows, f):
return sum(1 for r in rows if not str(r.get(f, "")).strip())
unmatched = [s for s in sales if not s["lead_id"]]
manila_blank = sum(1 for c in contacts if c["agent_team"] == "Manila" and c["dt"].date() < date(2025, 3, 1) and not c["outcome"])
manila_pre = sum(1 for c in contacts if c["agent_team"] == "Manila" and c["dt"].date() < date(2025, 3, 1))
contacted = {c["lead_id"] for c in contacts}
R["data_checks"] = {
"rows": {"leads": len(leads), "contacts": len(contacts), "sales": len(sales), "marketing_spend": len(spend), "complaints": len(complaints)},
"fields_present": True,
"nulls": {"leads.vertical": nulls(leads, "vertical"), "leads.channel": nulls(leads, "channel"),
"sales.gross_commission": nulls(sales, "gross_commission"), "sales.vertical": nulls(sales, "vertical"),
"contacts.first_contact_ts": nulls(contacts, "first_contact_ts")},
"unmatched_sales": len(unmatched), "unmatched_sales_pct": round(100 * len(unmatched) / len(sales), 1),
"manila_blank_outcomes_before_mar25": manila_blank, "manila_contacts_before_mar25": manila_pre,
"leads_never_contacted": sum(1 for r in leads if r["lead_id"] not in contacted),
"clawbacks_only_health_life": all(s["vertical"] in CLAWBACK_VERTICALS for s in sales if s["claw"]),
"clawback_equals_commission": all(abs(s["claw"] - s["gross"]) < 0.01 for s in sales if s["claw"]),
"sales_after_cutoff": sum(1 for s in sales if s["d"] > CUTOFF),
}
# ---------- Building blocks ----------
leads_m, leads_mv, leads_q, leads_qv = defaultdict(int), defaultdict(int), defaultdict(int), defaultdict(int)
for r in leads:
d = r["dt"].date()
leads_m[month(d)] += 1; leads_mv[(month(d), r["vertical"])] += 1
leads_q[qtr(d)] += 1; leads_qv[(qtr(d), r["vertical"])] += 1
net_m, net_q, gross_q = defaultdict(float), defaultdict(float), defaultdict(float)
sales_mv, sales_qv, net_qv = defaultdict(int), defaultdict(int), defaultdict(float)
for s in sales:
m, q = month(s["d"]), qtr(s["d"])
net_m[m] += s["gross"] - s["claw"]; net_q[q] += s["gross"] - s["claw"]; gross_q[q] += s["gross"]
sales_mv[(m, s["vertical"])] += 1; sales_qv[(q, s["vertical"])] += 1; net_qv[(q, s["vertical"])] += s["gross"] - s["claw"]
months = sorted(leads_m)
def pace(current, target, months_left):
return (target - current) / months_left
# ---------- K1 Net revenue per lead ----------
k1_m = {m: net_m[m] / leads_m[m] for m in months}
k1_q = {q: net_q[q] / leads_q[q] for q in sorted(leads_q)}
# Clawback maturity: estimate clawbacks still to land on health and life sales made after MATURE
mature_hl = [s for s in sales if s["vertical"] in CLAWBACK_VERTICALS and date(2026, 4, 1) <= s["d"] <= MATURE]
mature_rate = sum(1 for s in mature_hl if s["claw"]) / len(mature_hl)
immature = [s for s in sales if s["vertical"] in CLAWBACK_VERTICALS and s["d"] > MATURE]
expected = mature_rate * sum(s["gross"] for s in immature)
observed = sum(s["claw"] for s in immature)
extra_this_q = sum(mature_rate * s["gross"] for s in immature if qtr(s["d"]) == THIS_Q) - sum(s["claw"] for s in immature if qtr(s["d"]) == THIS_Q)
k1_this_est = (net_q[THIS_Q] - extra_this_q) / leads_q[THIS_Q]
target_k1 = k1_q[LAST_Q]
run_k1 = (k1_m["2026-09"] - k1_m["2026-06"]) / 3
R["K1"] = {"by_quarter": k1_q, "by_month": k1_m, "baseline": k1_q[THIS_Q], "baseline_clawback_adjusted": k1_this_est,
"target": target_k1, "target_by": "2027-06-30", "months_to_target": 9,
"pace_needed": pace(k1_q[THIS_Q], target_k1, 9), "pace_needed_adjusted": pace(k1_this_est, target_k1, 9),
"run_rate": run_k1, "mature_clawback_rate_apr_jun26": mature_rate,
"clawbacks_still_to_land_estimate": expected - observed,
"by_vertical_this_q": {v: net_qv[(THIS_Q, v)] / leads_qv[(THIS_Q, v)] for v in VERTICALS}}
# ---------- K1.1 Lead mix (energy share) ----------
share_m = {m: leads_mv[(m, "energy")] / leads_m[m] for m in months}
R["K1.1"] = {"energy_share_by_month": share_m, "baseline_month": "2026-09", "baseline": share_m["2026-09"],
"baseline_quarter": leads_qv[(THIS_Q, "energy")] / leads_q[THIS_Q],
"mix_this_q": {v: leads_qv[(THIS_Q, v)] / leads_q[THIS_Q] for v in VERTICALS},
"mix_last_q": {v: leads_qv[(LAST_Q, v)] / leads_q[LAST_Q] for v in VERTICALS},
"target": 0.30, "target_by": "2027-03-31", "months_to_target": 6,
"pace_needed": pace(share_m["2026-09"], 0.30, 6),
"run_rate": (share_m["2026-09"] - share_m["2026-06"]) / 3}
# ---------- K2 Conversion: Gary's draft (matched cohort) vs corrected (all sales by vertical) ----------
sold_leads = {s["lead_id"] for s in sales if s["lead_id"]}
open_cut = datetime(2026, 9, 23, 23, 59) # draft rule: exclude leads under 7 days old
def draft_conv(q, v):
pool = [r for r in leads if qtr(r["dt"].date()) == q and r["vertical"] == v and r["dt"] <= open_cut]
return sum(1 for r in pool if r["lead_id"] in sold_leads) / len(pool)
def corr_conv_q(q, v):
return sales_qv[(q, v)] / leads_qv[(q, v)]
def corr_conv_fy26(v):
return sum(sales_mv[(m, v)] for m in FY26) / sum(leads_mv[(m, v)] for m in FY26)
conv_m = {v: {m: sales_mv[(m, v)] / leads_mv[(m, v)] for m in months} for v in VERTICALS}
R["K2"] = {"draft_matched_this_q": {v: draft_conv(THIS_Q, v) for v in VERTICALS},
"corrected_this_q": {v: corr_conv_q(THIS_Q, v) for v in VERTICALS},
"corrected_last_q": {v: corr_conv_q(LAST_Q, v) for v in VERTICALS},
"fy26_average": {v: corr_conv_fy26(v) for v in VERTICALS},
"by_month": conv_m}
R["K2"]["targets"] = {v: (0.11 if v == "health" else R["K2"]["fy26_average"][v]) for v in VERTICALS}
h = R["K2"]["corrected_this_q"]["health"]
R["K2.1"] = {"baseline": h, "draft_matched_baseline": R["K2"]["draft_matched_this_q"]["health"], "target": 0.11,
"target_by": "2027-06-30", "months_to_target": 9, "pace_needed": pace(h, 0.11, 9),
"run_rate": (conv_m["health"]["2026-09"] - conv_m["health"]["2026-06"]) / 3,
"by_month": conv_m["health"]}
# ---------- K2.2 Complaints per 1,000 sales ----------
comp_q, comp_fy = defaultdict(int), 0
for c in complaints:
comp_q[qtr(c["d"])] += 1
comp_fy += month(c["d"]) in FY26
sales_q = defaultdict(int)
for s in sales:
sales_q[qtr(s["d"])] += 1
fy_sales = sum(1 for s in sales if month(s["d"]) in FY26)
R["K2.2"] = {"by_quarter": {q: 1000 * comp_q[q] / sales_q[q] for q in sorted(sales_q)},
"complaints_by_quarter": dict(comp_q), "fy26_rate": 1000 * comp_fy / fy_sales,
"baseline": 1000 * comp_q[THIS_Q] / sales_q[THIS_Q], "complaints_this_q": comp_q[THIS_Q],
"target": 1000 * comp_fy / fy_sales, "target_by": "2027-06-30"}
r12 = {f"2025-{m:02d}" for m in range(10, 13)} | {f"2026-{m:02d}" for m in range(1, 10)}
c12 = sum(1 for c in complaints if month(c["d"]) in r12)
s12 = sum(1 for s in sales if month(s["d"]) in r12)
R["K2.2"].update({"rolling_12m_to": "2026-09", "rolling_12m_complaints": c12, "rolling_12m_sales": s12,
"rolling_12m_rate": 1000 * c12 / s12})
# ---------- K3 Cost per acquisition by channel ----------
chans = sorted({r["channel"] for r in leads})
spend_qc, spend_fyc = defaultdict(float), defaultdict(float)
for r in spend:
spend_qc[(qtr(r["d"]), r["channel"])] += r["cost"]
if month(r["d"]) in FY26:
spend_fyc[r["channel"]] += r["cost"]
net_sales_qc, net_sales_fyc = defaultdict(int), defaultdict(int)
for s in sales:
if s["lead_id"] and not s["claw"]:
ch = lead_by_id[s["lead_id"]]["channel"]
net_sales_qc[(qtr(s["d"]), ch)] += 1
if month(s["d"]) in FY26:
net_sales_fyc[ch] += 1
netrev_qc, netrev_fyc = defaultdict(float), defaultdict(float)
for s in sales:
if s["lead_id"]:
ch = lead_by_id[s["lead_id"]]["channel"]
netrev_qc[(qtr(s["d"]), ch)] += s["gross"] - s["claw"]
if month(s["d"]) in FY26:
netrev_fyc[ch] += s["gross"] - s["claw"]
R["K3_counter"] = {
"name": "Net revenue per marketing dollar",
"this_q": {c: netrev_qc[(THIS_Q, c)] / spend_qc[(THIS_Q, c)] for c in chans},
"last_q": {c: netrev_qc[(LAST_Q, c)] / spend_qc[(LAST_Q, c)] for c in chans},
"fy26_average": {c: netrev_fyc[c] / spend_fyc[c] for c in chans},
"all_channels_this_q": sum(netrev_qc[(THIS_Q, c)] for c in chans) / sum(spend_qc[(THIS_Q, c)] for c in chans),
"all_channels_last_q": sum(netrev_qc[(LAST_Q, c)] for c in chans) / sum(spend_qc[(LAST_Q, c)] for c in chans),
"note": "Matched sales only (6.1% of sales have no channel)"}
cpa_this = {c: spend_qc[(THIS_Q, c)] / net_sales_qc[(THIS_Q, c)] for c in chans}
cpa_fy = {c: spend_fyc[c] / net_sales_fyc[c] for c in chans}
R["K3"] = {"this_q": cpa_this, "last_q": {c: spend_qc[(LAST_Q, c)] / net_sales_qc[(LAST_Q, c)] for c in chans},
"fy26_average": cpa_fy, "paid_search_target": 0.9 * cpa_fy["paid_search"], "target_by": "2027-06-30",
"pace_needed_paid_search": pace(cpa_this["paid_search"], 0.9 * cpa_fy["paid_search"], 9)}
# ---------- K4 Speed to contact ----------
first = {}
for c in contacts:
if c["lead_id"] not in first or c["dt"] < first[c["lead_id"]][0]:
first[c["lead_id"]] = (c["dt"], c["agent_team"])
mins_q, mins_q_team, mins_q_chan, mins_m = defaultdict(list), defaultdict(list), defaultdict(list), defaultdict(list)
for r in leads:
if r["lead_id"] in first:
mn = (first[r["lead_id"]][0] - r["dt"]).total_seconds() / 60
q = qtr(r["dt"].date())
mins_q[q].append(mn); mins_m[month(r["dt"].date())].append(mn)
mins_q_team[(q, first[r["lead_id"]][1])].append(mn); mins_q_chan[(q, r["channel"])].append(mn)
med_m = {m: st.median(v) for m, v in sorted(mins_m.items())}
R["K4"] = {"by_quarter": {q: st.median(v) for q, v in sorted(mins_q.items())},
"baseline": st.median(mins_q[THIS_Q]),
"by_team_this_q": {t: st.median(mins_q_team[(THIS_Q, t)]) for t in ["Sydney", "Melbourne", "Manila"]},
"by_channel_this_q": {c: st.median(mins_q_chan[(THIS_Q, c)]) for c in chans},
"share_over_15_min_this_q": sum(1 for x in mins_q[THIS_Q] if x > 15) / len(mins_q[THIS_Q]),
"target": 10.0, "target_by": "2027-03-31", "months_to_target": 6,
"pace_needed": pace(st.median(mins_q[THIS_Q]), 10.0, 6),
"run_rate": (med_m["2026-09"] - med_m["2026-06"]) / 3}
# ---------- K5 90-day clawback rate (mature cohorts only) ----------
def claw_rate(rows):
return sum(1 for s in rows if s["claw"] and (s["cbd"] - s["d"]).days <= 90) / len(rows) if rows else None
hl = [s for s in sales if s["vertical"] in CLAWBACK_VERTICALS]
by_q = {}
for q in sorted({qtr(s["d"]) for s in hl}):
rows = [s for s in hl if qtr(s["d"]) == q and s["d"] <= MATURE]
by_q[q] = {"rate": claw_rate(rows), "final": all(s["d"] <= MATURE for s in hl if qtr(s["d"]) == q)}
fy_rows = [s for s in hl if month(s["d"]) in FY26 and s["d"] <= MATURE]
last_mature = [s for s in hl if qtr(s["d"]) == LAST_Q]
R["K5"] = {"by_quarter": by_q, "fy26_average": claw_rate(fy_rows), "baseline_quarter": LAST_Q,
"baseline": claw_rate(last_mature), "provisional_this_q": claw_rate([s for s in hl if qtr(s["d"]) == THIS_Q]),
"by_vertical_last_q": {v: claw_rate([s for s in last_mature if s["vertical"] == v]) for v in CLAWBACK_VERTICALS},
"target": claw_rate(fy_rows), "target_by": "2027-06-30", "months_to_target": 9,
"pace_needed": pace(claw_rate(last_mature), claw_rate(fy_rows), 9)}
# ---------- Revenue bridge: last quarter to this quarter ----------
def comp(q):
L = leads_q[q]
return L, {v: (leads_qv[(q, v)] / L, sales_qv[(q, v)] / leads_qv[(q, v)],
(net_qv[(q, v)] / sales_qv[(q, v)]) if sales_qv[(q, v)] else 0) for v in VERTICALS}
def rev(L, parts):
return sum(L * s * c * n for s, c, n in parts.values())
L0, p0 = comp(LAST_Q)
L1, p1 = comp(THIS_Q)
r0 = rev(L0, p0)
step_vol = rev(L1, p0)
step_mix = rev(L1, {v: (p1[v][0], p0[v][1], p0[v][2]) for v in VERTICALS})
step_conv = rev(L1, {v: (p1[v][0], p1[v][1], p0[v][2]) for v in VERTICALS})
r1 = rev(L1, p1)
R["bridge"] = {"last_q_net_revenue": r0, "this_q_net_revenue": r1,
"volume": step_vol - r0, "mix": step_mix - step_vol, "conversion": step_conv - step_mix,
"revenue_per_sale_incl_clawbacks": r1 - step_conv, "total_change": r1 - r0,
"leads_last_q": L0, "leads_this_q": L1,
"per_lead_last_q": r0 / L0, "per_lead_this_q": r1 / L1}
out.write_text(json.dumps(R, indent=2, default=str), encoding="utf8")
print(json.dumps({k: R[k] for k in ["data_checks"]}, indent=1))
for k in ["K1", "K1.1", "K2.1", "K2.2", "K4", "K5"]:
print(k, {x: (round(y, 4) if isinstance(y, float) else y) for x, y in R[k].items() if not isinstance(y, dict)})
print("K2", {x: {v: round(y, 4) for v, y in R["K2"][x].items()} for x in ["draft_matched_this_q", "corrected_this_q", "fy26_average"]})
print("K3", {x: {c: round(y, 2) for c, y in R["K3"][x].items()} for x in ["this_q", "fy26_average"]}, round(R["K3"]["paid_search_target"], 2))
print("bridge", {k: round(v, 2) for k, v in R["bridge"].items()})
if __name__ == "__main__":
main()