File: stage-2-07-calculation-script.md
Kestrel Medical: Kewpie's calculation script
Sample data. Produced by Kewpie at Stage 2.
"""Kestrel Medical (Sample data): Stage 2 calculations. Re-run: python calc_kpis.py (needs pandas; not python -I)
Reads data/*.csv, writes kpi_results.json. Data cutoff = last date in the sales data."""
import json, re, os
import pandas as pd
HERE = os.path.dirname(os.path.abspath(__file__))
D = os.path.join(HERE, "data")
R = {"label": "Sample data"}
TARGET_G = 4.0
s = pd.read_csv(f"{D}/sales_order_lines.csv", low_memory=False, parse_dates=["order_date", "actual_ship_date"])
s["gross"] = s.list_price * s.qty
s["oy"], s["om"] = s.order_date.dt.year, s.order_date.dt.month
s["sy"], s["sm"] = s.actual_ship_date.dt.year, s.actual_ship_date.dt.month
CUT = s.order_date.max()
M = CUT.month # last complete month (Sep)
MONTHS_LEFT = (2027 - CUT.year) * 12 + 12 - CUT.month # Sep-26 to Dec-27 = 15
R["cutoff"] = str(CUT.date()); R["months_left"] = MONTHS_LEFT
def r(x, n=2):
return None if x is None or pd.isna(x) else round(float(x), n)
# ---- K1: cost of sales
cost_cols = [f"{f}.{c}" for f in os.listdir(D) if f.endswith(".csv")
for c in pd.read_csv(f"{D}/{f}", nrows=0).columns if "cost" in c.lower() and "consum" not in c.lower()]
R["K1"] = {"cost_fields_found": cost_cols, "calculable": False, "stated_margin_pct": 41.0, "target": 43.0,
"pace_points_per_month_indicative": r((43 - 41) / MONTHS_LEFT, 3)}
# ---- K3 and sub-KPIs: YTD growth, ship date basis
def ytd(df, y, m, by="sy", bm="sm"):
return df[(df[by] == y) & (df[bm] <= m)].net_value_aud.sum()
def growth_series(df, by="sy", bm="sm"):
out = []
for m in range(1, M + 1):
a, b = ytd(df, CUT.year, m, by, bm), ytd(df, CUT.year - 1, m, by, bm)
out.append((a / b - 1) * 100 if b else None)
return out
def g_block(df, by="sy", bm="sm"):
g = growth_series(df, by, bm); cur = g[-1]
run = (g[-1] - g[-4]) / 3 if g[-4] is not None and cur is not None else None
pace = (TARGET_G - cur) / MONTHS_LEFT if cur is not None else None
return {"baseline_pct": r(cur), "ytd_2026_m": r(ytd(df, CUT.year, M, by, bm) / 1e6),
"ytd_2025_m": r(ytd(df, CUT.year - 1, M, by, bm) / 1e6),
"run_rate_pts_per_month": r(run, 3), "pace_needed_pts_per_month": r(pace, 3),
"series_pct": [r(x) for x in g]}
k3 = g_block(s)
fy = lambda y: s[s.sy == y].net_value_aud.sum()
k3["fy2025_vs_fy2024_pct"] = r((fy(2025) / fy(2024) - 1) * 100)
k3["order_date_basis_pct"] = g_block(s, "oy", "om")["baseline_pct"]
k3["excl_warranty_pct"] = g_block(s[s.order_type != "warranty_replacement"])["baseline_pct"]
q4_25 = s[(s.sy == 2025) & (s.sm >= 10)].net_value_aud.sum()
fy26 = ytd(s, 2026, M) + q4_25 * (1 + k3["baseline_pct"] / 100)
k3["fy2026_estimate_m"] = r(fy26 / 1e6); k3["fy2025_actual_m"] = r(fy(2025) / 1e6)
k3["need_2027_on_2026_base_m"] = r(fy26 * 1.04 / 1e6)
k3["growth_needed_vs_2025_pct"] = r((fy26 * 1.04 / fy(2025) - 1) * 100)
k3["order_types_2025_m"] = {k: r(v / 1e6, 1) for k, v in s[s.sy == 2025].groupby("order_type").net_value_aud.sum().items()}
R["K3"] = k3
R["K3.1"] = {g: g_block(df) for g, df in s.groupby("material_group")}
R["K3.2"] = {g: g_block(df) for g, df in s.groupby("rep_id")}
# ---- K2: discount on big deals (placeholder = order_type tender_contract)
def disc(df):
return (1 - df.net_value_aud.sum() / df.gross.sum()) * 100 if len(df) else None
def ytd_o(df, y, m):
return df[(df.oy == y) & (df.om <= m)]
tc = s[s.order_type == "tender_contract"]
series = [disc(ytd_o(tc, 2026, m)) for m in range(1, M + 1)]
p99 = s.net_value_aud.quantile(0.99)
k2 = {"definition": "order_type = tender_contract, order date", "baseline_2026_ytd_pct": r(series[-1]),
"fy": {str(y): r(disc(tc[tc.oy == y])) for y in range(2021, 2026)},
"ytd_2025_pct": r(disc(ytd_o(tc, 2025, M))), "series_2026_pct": [r(x, 3) for x in series],
"run_rate_pts_per_month": r((series[-1] - series[-4]) / 3, 3), "pace_needed_pts_per_month": 0.0,
"lines_2026_ytd": int(len(ytd_o(tc, 2026, M))),
"net_value_by_year_m": {str(y): r(tc[tc.oy == y].net_value_aud.sum() / 1e6, 1) for y in range(2021, 2027)},
"max_order_total_aud": r(s.net_value_aud.max(), 0), "orders_ge_5m": int((s.net_value_aud >= 5e6).sum()),
"sens_tender_plus_distributor_pct": r(disc(ytd_o(s[s.order_type.isin(["tender_contract", "distributor"])], 2026, M))),
"sens_top1pct_lines_pct": r(disc(ytd_o(s[s.net_value_aud >= p99], 2026, M))), "top1pct_cutoff_aud": r(p99, 0),
"draft_placeholder_contract_id": {"lines": int(s.contract_id.notna().sum()),
"groups": s[s.contract_id.notna()].material_group.unique().tolist(),
"order_types": s[s.contract_id.notna()].order_type.unique().tolist(),
"prefix": sorted(s.contract_id.dropna().str[:2].unique().tolist())},
"tc_net_ytd_2025_m": r(ytd_o(tc, 2025, M).net_value_aud.sum() / 1e6, 1),
"tc_net_ytd_2026_m": r(ytd_o(tc, 2026, M).net_value_aud.sum() / 1e6, 1),
"all_lines_avg_discount_2026_ytd_pct": r(disc(ytd_o(s, 2026, M))),
"all_lines_avg_discount_2025_pct": r(disc(s[s.oy == 2025])),
"calc_vs_field_gap_p999_pts": r((((1 - s.net_value_aud / s.gross) * 100) - s.discount_pct).abs().quantile(0.999), 3),
"avg_discount_by_type_2026": {k: r(disc(ytd_o(df, 2026, M))) for k, df in s.groupby("order_type")}}
R["K2"] = k2
# ---- Tender register
t = pd.read_csv(f"{D}/tender_register.csv")
def money(v):
m = re.match(r"\$([\d.]+)m( pa)?", v.strip())
return float(m.group(1)) * 1e6, bool(m.group(2))
def total_val(row, naive=False):
x, pa = money(row.estimated_value)
return x if (naive or not pa) else x * int(row.contract_length.split()[0])
t["value"] = t.apply(total_val, axis=1); t["value_naive"] = t.apply(lambda q: total_val(q, True), axis=1)
t["our"] = t.our_submitted_price.map(lambda v: money(v)[0])
t["win"] = t.winning_price.map(lambda v: money(v)[0] if pd.notna(v) else None)
t["close"] = pd.to_datetime(t.close_date, format="%d/%m/%Y")
big = t[t.value >= 5e6].sort_values("close"); dec = big[big.outcome.isin(["won", "lost"])]
wins = (dec.outcome == "won").tolist()
need = {}
for tgt in (40, 60, 80, 100):
for k in range(0, 12):
w = (wins + [True] * k)[-5:]
if sum(w) / 5 * 100 >= tgt:
need[str(tgt)] = k; break
def rate(df):
d = df[df.outcome.isin(["won", "lost"])]
return [int((d.outcome == "won").sum()), int(len(d))]
k4 = {"big_tenders": [{"ref": x.tender_reference, "customer": x.customer, "close": x.close_date,
"value_m": r(x.value / 1e6, 2), "outcome": x.outcome, "winner": x.winning_supplier}
for x in big.itertuples()],
"won_of_decided": rate(big), "rate_pct": r(rate(big)[0] / rate(big)[1] * 100, 0),
"same_set_if_pa_ignored": sorted(t[t.value_naive >= 5e6].tender_reference) == sorted(big.tender_reference),
"pa_rows": int(t.estimated_value.str.contains("pa").sum()),
"bid_per_year": {str(y): int((big.close.dt.year == y).sum()) for y in (2024, 2025, 2026)},
"sens": {f">={n}m": rate(t[t.value >= n * 1e6]) for n in (5, 4, 3, 2, 1, 0)},
"partial_cancelled_big": int(big.outcome.isin(["partial", "cancelled"]).sum()),
"consecutive_wins_needed_for_rolling5": need,
"all_outcomes": t.outcome.value_counts().to_dict(),
"name_variants_in_register": sorted({c for c in t.customer if "Southern Metro" in c or "Western Plains" in c})}
R["K4"] = k4
# K4.1 price against winning price
lost = t[t.outcome == "lost"]
lw = lost.dropna(subset=["win"]).copy(); lw["gap"] = (lw.our / lw.win - 1) * 100
bl = big[big.outcome == "lost"]
R["K4.1"] = {"lost_all": len(lost), "lost_with_winning_price": len(lw),
"avg_gap_pct_known": r(lw.gap.mean()), "median_gap_pct_known": r(lw.gap.median()),
"lost_gap_list": [{"ref": x.tender_reference, "gap_pct": r(x.gap), "big": bool(x.value >= 5e6)}
for x in lw.itertuples()],
"big_lost": [{"ref": x.tender_reference, "customer": x.customer,
"gap_pct": r((x.our / x.win - 1) * 100) if pd.notna(x.win) else None}
for x in bl.itertuples()]}
# K4.2 scoring tables (gap in weighted points out of 100)
W = {"price": 30, "clinical": 25, "service": 15, "supply": 10, "training": 10, "sustain": 10}
def area(n):
n = n.lower()
for k in W:
if n.strip().startswith(k) or k in n:
return k
def parse(txt):
out = {}
if "Criterion" in txt:
for ln in txt.splitlines()[1:]:
p = [x.strip() for x in ln.split("|")]
if len(p) == 4:
w, a, b = (float(x.strip("%")) for x in p[1:]); out[area(p[0])] = (b - a) * w / 100
elif "PREF=" in txt:
for m in re.finditer(r"([A-Z&_]+)=(\d+(?:\.\d+)?)/PREF=(\d+(?:\.\d+)?)", txt):
a = area(m.group(1).replace("_", " ").replace("&", " "))
out[a] = (float(m.group(3)) - float(m.group(2))) * W[a] / 100
else:
for m in re.finditer(r"([A-Za-z ]+): ([\d.]+)/(\d+) \(preferred ([\d.]+)/\d+\)", txt):
out[area(m.group(1))] = float(m.group(4)) - float(m.group(2))
return out
rows = []
for x in lost.itertuples():
if isinstance(x.scoring_table, str):
g = parse(x.scoring_table)
if len(g) == 6:
rows.append({"ref": x.tender_reference, "customer": x.customer, "big": bool(x.value >= 5e6),
"gaps": {k: r(v) for k, v in g.items()}, "top": max(g, key=g.get), "price_gap": r(g["price"])})
cnt = pd.Series([z["top"] for z in rows]).value_counts().to_dict()
R["K4.2"] = {"lost": len(lost), "lost_with_scoring": len(rows), "lost_with_debrief": int((lost.debrief_received == "Y").sum()),
"top_gap_counts": cnt, "price_is_top_gap": int(sum(z["top"] == "price" for z in rows)),
"price_ahead_count": int(sum(z["price_gap"] <= 0 for z in rows)), "rows": rows,
"big": [z for z in rows if z["big"]], "weights": W}
g = lost.groupby("winning_supplier").agg(n=("tender_reference", "size"), value_m=("value", lambda v: v.sum() / 1e6))
R["K4.3"] = {"all_lost": {k: {"n": int(v.n), "value_m": r(v.value_m, 1)} for k, v in g.iterrows()},
"big_lost": [{"ref": x.tender_reference, "winner": x.winning_supplier, "value_m": r(x.value / 1e6, 1)}
for x in bl.itertuples()],
"lost_missing_winner": int(lost.winning_supplier.isna().sum())}
# ---- K5
m = pd.read_csv(f"{D}/monthly_reports_index.csv", parse_dates=["last_refreshed"], keep_default_na=False)
cand = {}
for sd in m.sales_definition.unique():
if sd != "n/a":
cand[f"{sd} + parent where known"] = int(((m.sales_definition == sd) & (m.customer_grouping == "parent where known")).sum())
stale = m[m.last_refreshed < CUT - pd.Timedelta(days=30)]
R["K5"] = {"reports": len(m), "sales_def": m.sales_definition.value_counts().to_dict(),
"cust_group": m.customer_grouping.value_counts().to_dict(), "candidate_sets_count": cand,
"both_na": int(((m.sales_definition == "n/a") & (m.customer_grouping == "n/a")).sum()),
"any_na": int(((m.sales_definition == "n/a") | (m.customer_grouping == "n/a")).sum()),
"distinct_combos": int(m.groupby(["sales_definition", "customer_grouping"]).ngroups),
"stale_over_30d": len(stale), "stale_names": stale.workbook.tolist(), "owners": m.owner.value_counts().to_dict()}
# ---- extra checks for notes
sv = s[(s.material_group == "SERVICE") & (s.sy == 2025)]
R["extra"] = {"median_reps_per_sold_to": float(s.groupby("sold_to_id").rep_id.nunique().median()),
"service_jan_jul_share_2025_pct": r(sv[sv.sm.isin([1, 7])].net_value_aud.sum() / sv.net_value_aud.sum() * 100, 0)}
# ---- data quality
R["dq"] = {"sales_lines": len(s), "unshipped_lines": int(s.actual_ship_date.isna().sum()),
"unshipped_pct": r(s.actual_ship_date.isna().mean() * 100), "list_price_missing": int((s.list_price <= 0).sum()),
"negative_discount_lines": int((s.discount_pct < 0).sum()), "duplicate_lines": int(s.duplicated(["order_id", "line_no"]).sum()),
"tender_rows": len(t), "lost_missing_winning_price": int(len(lost) - len(lw)),
"tenders_pa_values": int(t.estimated_value.str.contains("pa").sum()),
"lost_missing_scoring": int(len(lost) - len(rows)), "winner_missing": int(t.winning_supplier.isna().sum())}
# ---- v1.1 (Business Requirements v1.1, approved at the Stage 2 gate): changed targets and measures
# K2 (v1.3): NO target in the first pass (Sharon left it open). The path below is Gary's Phase 2 proposal, kept as information only.
start, end = k2["baseline_2026_ytd_pct"], 14.0
step = (end - start) / MONTHS_LEFT
path = [start + step * i for i in range(MONTHS_LEFT + 1)] # index 0 = Sep-26, 15 = Dec-27
flat_gap = [start - p for p in path] # gap above the path if the discount stays flat
mn = ["Sep-26", "Oct-26", "Nov-26", "Dec-26", "Jan-27", "Feb-27", "Mar-27", "Apr-27", "May-27", "Jun-27", "Jul-27", "Aug-27", "Sep-27", "Oct-27", "Nov-27", "Dec-27"]
k2["v11"] = {"role": "Gary Phase 2 proposal, information only", "start_pct": start, "end_pct": end, "pace_needed_pts_per_month": r(step, 3),
"path_pct": {mn[i]: r(path[i]) for i in (0, 3, 6, 9, 12, 15)},
"status_now": "Not rated (no target in the first pass)",
"flat_first_amber": next(mn[i] for i in range(1, 16) if flat_gap[i] > 0),
"flat_first_red": next(mn[i] for i in range(1, 16) if flat_gap[i] > 1),
"run_rate_pts_per_month": k2["run_rate_pts_per_month"],
"dollars_note_net_ytd_2026_m": k2["tc_net_ytd_2026_m"]}
# K4: count out of the last 5 big decided tenders, calendar-year count alongside; Gary's starting proposal 3 of the last 5.
last5 = dec.tail(5)
need3 = next(k for k in range(0, 12) if sum((wins + [True] * k)[-5:]) >= 3)
cy = {str(y): [int((dec[dec.close.dt.year == y].outcome == "won").sum()), int((dec.close.dt.year == y).sum())] for y in (2024, 2025, 2026)}
k4["v11"] = {"window_won": int(sum(wins[-5:])), "window_size": int(len(wins[-5:])), "calendar_year_won_of_decided": cy,
"proposed_target_won_of_last5": 3, "next_consecutive_wins_needed": need3,
"window_after_next_win": int(sum((wins + [True])[-5:])), "window_after_two_wins": int(sum((wins + [True] * 2)[-5:])),
"big_tenders_per_year": k4["bid_per_year"], "months_to_dec27": MONTHS_LEFT,
"expected_big_tenders_to_dec27": r(2 * MONTHS_LEFT / 12, 1)}
# K3: unchanged measure; target base is now the 2026 full-year actuals (estimate until the year closes).
k3["v11"] = {"base_2026_full_year_estimate_m": k3["fy2026_estimate_m"], "target_2027_m": k3["need_2027_on_2026_base_m"],
"vs_2025_pct": k3["growth_needed_vs_2025_pct"],
"monthly_sales_needed_2027_m": r(k3["need_2027_on_2026_base_m"] / 12, 2), "monthly_sales_2026_ytd_avg_m": r(k3["ytd_2026_m"] / M, 2)}
json.dump(R, open(os.path.join(HERE, "kpi_results.json"), "w"), indent=1, default=str)
print("ok")