Files
boardroom-map/orchestrator/progress.py
T
Jonathan KirkwoodandClaude Fable 5 f13d044a70 progress: only track KPIs that ever carried a target
Tested against the live Strike ledger (9 decks): its decks mint ad-hoc KPI
names every quarter, so ~70 one-off unscored mentions flooded the watch list,
trajectory table, and dropped-KPIs gap. A KPI now only counts as trackable/
vanished if it was held to a target at least once; the rest collapse to a
one-line count. Scoring-derived flags (no_profitability_visibility,
no_quantitative_kpis, kpi_dropped) no longer duplicate their dedicated gaps
as recurring-flag requests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-31 20:53:14 -05:00

536 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Longitudinal progress review + deck-quality guidance — pure, stdlib only.
Where scorecard.py answers "how did the latest deck score", this module answers
two portfolio-owner questions across the whole ledger:
1. Is the company actually progressing? Composite / BDEF-category / KPI
trajectories across every graded deck, recurring vs resolved red flags.
2. Is the material good enough to judge them by? A deterministic gap engine
inspects the latest record for what the deck is NOT showing (no
profitability visibility, untargeted KPIs, no forward guidance, thin
evidence per BDEF category, silently dropped KPIs, no board asks) and
turns each gap into a concrete request for the next deck.
Everything numeric is derived from scoring records the deterministic pipeline
already produced — no model calls, so the dashboard can compute this live.
`analyze(company, records)` returns the JSON shape; `render_progress_md`
renders it as PROGRESS.md.
"""
from __future__ import annotations
import decks as decks_mod
_CATEGORIES = "ABCDEFGH"
_CATEGORY_TITLES = {
"A": "Incentive Alignment & Skin in the Game",
"B": "Inversion Discipline & Margin of Safety",
"C": "Circle of Competence & Rational Learning",
"D": "Capital Allocation Quality",
"E": "Moat Durability & Competitive Reality",
"F": "Psychological & Cultural Health",
"G": "Simplicity, Clarity & Decision Velocity",
"H": "Board Value-Add & Governance Quality",
}
# What a deck that scores well on each BDEF category actually shows (bdef.md's
# "5" anchors, phrased as material to request). Used when a category's evidence
# is thin: the graders had nothing to quote, so the deck isn't showing this.
_CATEGORY_MATERIAL = {
"A": "comp structure and promotion criteria, management/founder ownership and "
"option-pool detail, and which metrics bonuses actually pay on",
"B": "a downside scenario for the plan (what breaks the thesis), cash "
"buffer/runway under that scenario, and pre-committed 'we won't do X' "
"boundaries",
"C": "explicit 'we know / we don't know' boundaries, postmortems on this "
"period's misses, and the criteria used before entering adjacencies",
"D": "expected return on each major use of cash compared against the "
"alternatives — including doing nothing or returning capital",
"E": "specific, testable moat metrics (retention/churn, switching costs, "
"pricing power, win rates) plus named competitive threats and the "
"response to each",
"F": "what went wrong this period surfaced by management first, regrettable "
"attrition, and culture signals beyond an engagement score",
"G": "a one-page summary of the 23 decisions that matter, with KPI "
"definitions and format kept identical to the prior deck",
"H": "explicit asks for the board, each with a recommendation and what "
"would make that recommendation wrong",
}
# Latest-deck evidence_quality below this = the panel scored the category on
# thin air (scores regress to 3 anyway) — the deck isn't showing the material.
THIN_EVIDENCE = 0.35
# |net composite change| within this band counts as flat.
FLAT_BAND = 1.0
# Category adjusted-score moves within this band count as flat.
CAT_BAND = 0.3
def _f(x) -> float | None:
return float(x) if isinstance(x, (int, float)) and not isinstance(x, bool) else None
def _cats(rec: dict) -> dict:
return ((rec.get("qual") or {}).get("categories")) or {}
# ---------------------------------------------------------------- trajectory
def _trend(records: list[dict]) -> dict:
series = []
prev_comp = None
for r in records:
comp = _f(r.get("composite"))
entry = {
"period": r.get("period"),
"composite": comp,
"quant": _f((r.get("quant") or {}).get("score")),
"qual": _f((r.get("qual") or {}).get("score")),
"penalties": _f((r.get("penalties") or {}).get("total")),
"delta": (round(comp - prev_comp, 1)
if comp is not None and prev_comp is not None else None),
}
series.append(entry)
prev_comp = comp if comp is not None else prev_comp
scored = [s for s in series if s["composite"] is not None]
out = {"series": series, "direction": "insufficient", "net_change": None,
"best": None, "worst": None}
if not scored:
return out
out["best"] = max(scored, key=lambda s: s["composite"])
out["worst"] = min(scored, key=lambda s: s["composite"])
if len(scored) < 2:
return out
net = round(scored[-1]["composite"] - scored[0]["composite"], 1)
out["net_change"] = net
deltas = [s["delta"] for s in scored[1:] if s["delta"] is not None]
if abs(net) <= FLAT_BAND:
out["direction"] = "flat"
elif net > 0:
out["direction"] = "improving" if all(d >= -FLAT_BAND for d in deltas) else "mixed"
else:
out["direction"] = "declining" if all(d <= FLAT_BAND for d in deltas) else "mixed"
return out
def _category_trends(records: list[dict]) -> dict:
out: dict[str, dict] = {}
for cid in _CATEGORIES:
series = []
for r in records:
c = _cats(r).get(cid) or {}
series.append({"period": r.get("period"),
"adjusted": _f(c.get("adjusted")),
"evidence_quality": _f(c.get("evidence_quality"))})
scored = [s for s in series if s["adjusted"] is not None]
if not scored:
continue
latest = scored[-1]
net = (round(latest["adjusted"] - scored[0]["adjusted"], 2)
if len(scored) >= 2 else None)
direction = ("flat" if net is None or abs(net) <= CAT_BAND
else "improving" if net > 0 else "declining")
evid = [s["evidence_quality"] for s in scored if s["evidence_quality"] is not None]
out[cid] = {
"title": _CATEGORY_TITLES[cid],
"series": series,
"latest": latest["adjusted"],
"net_change": net,
"direction": direction if len(scored) >= 2 else "insufficient",
"latest_evidence": latest["evidence_quality"],
"avg_evidence": round(sum(evid) / len(evid), 4) if evid else None,
}
return out
def _kpi_trends(records: list[dict]) -> dict:
"""Per canonical KPI across records: credit series + a trajectory status."""
out: dict[str, dict] = {}
order: list[str] = []
for r in records:
for k in r.get("kpi_results") or []:
cn = (k.get("canonical_name") or k.get("name") or "").strip()
if not cn:
continue
s = out.get(cn)
if s is None:
s = out[cn] = {"name": k.get("name") or cn, "profitability": False,
"series": [], "attempts": 0, "hits": 0,
"last_credit": None, "status": "untargeted",
"last_period": None}
order.append(cn)
if k.get("name"):
s["name"] = k["name"]
s["profitability"] = bool(k.get("profitability", s["profitability"]))
credit = _f(k.get("credit"))
s["series"].append({"period": r.get("period"), "credit": credit,
"target_source": k.get("target_source")})
s["last_period"] = r.get("period")
if credit is not None:
s["attempts"] += 1
s["last_credit"] = credit
if credit >= 0.999:
s["hits"] += 1
latest_period = records[-1].get("period") if records else None
for s in out.values():
credits = [p["credit"] for p in s["series"] if p["credit"] is not None]
# Accountable = the KPI was actually held to a target at least once.
# One-off unscored mentions (ad-hoc deck metrics) are extraction noise,
# not a trackable KPI — they'd otherwise flood the review.
s["accountable"] = bool(credits)
if s["last_period"] != latest_period:
s["status"] = "not-reported" # vanished from the latest deck
elif not credits:
s["status"] = "untargeted" # reported, but never had a target
elif credits[-1] >= 0.999:
s["status"] = ("recovered" if len(credits) >= 2 and credits[-2] < 0.999
else "on-track")
elif len(credits) >= 2 and all(c < 0.999 for c in credits[-2:]):
s["status"] = "missing-repeatedly"
elif len(credits) >= 2 and credits[-2] >= 0.999:
s["status"] = "slipped"
else:
s["status"] = "missed-latest"
return {cn: out[cn] for cn in order}
def _flag_history(records: list[dict]) -> dict:
"""Recurring / new-in-latest / resolved flags by code across records."""
seen: dict[str, dict] = {}
for r in records:
for f in (r.get("penalties") or {}).get("flags") or []:
code = f.get("code") or "flag"
s = seen.setdefault(code, {"code": code, "description": f.get("description") or "",
"severity": 0, "periods": []})
s["severity"] = max(s["severity"], int(f.get("severity") or 0))
s["description"] = f.get("description") or s["description"]
if r.get("period") not in s["periods"]:
s["periods"].append(r.get("period"))
latest_codes = {f.get("code") for f in
((records[-1].get("penalties") or {}).get("flags") or [])} if records else set()
prior_codes = set(seen) - latest_codes
recurring = sorted((s for s in seen.values() if len(s["periods"]) >= 2),
key=lambda s: (-len(s["periods"]), -s["severity"]))
new = sorted((s for s in seen.values()
if s["code"] in latest_codes and len(s["periods"]) == 1),
key=lambda s: -s["severity"])
resolved = sorted((s for s in seen.values() if s["code"] in prior_codes),
key=lambda s: -s["severity"])
return {"recurring": recurring, "new_in_latest": new, "resolved": resolved}
# ---------------------------------------------------------------- gap engine
def _gaps(company: dict, records: list[dict], categories: dict,
kpis: dict, flags: dict) -> list[dict]:
"""Deck-quality findings on the LATEST record: what the material isn't
showing, each with a concrete request for the next deck."""
if not records:
return []
latest = records[-1]
out: list[dict] = []
def gap(code, severity, title, detail, ask):
out.append({"code": code, "severity": severity, "title": title,
"detail": detail, "ask": ask})
q = latest.get("quant") or {}
if (q.get("profitability") or {}).get("na"):
gap("no_profitability_kpis", 3, "No profitability visibility",
"The latest deck reports no profit, margin, or cash KPI the pipeline "
"could score — the heaviest bucket (30/60 quant points) sat empty.",
"Report gross margin, EBITDA (or net burn) and cash runway as "
"first-class KPIs every period, each with a stated target.")
untargeted = [s["name"] for s in kpis.values()
if s["status"] == "untargeted" and s["last_period"] == latest.get("period")]
if untargeted:
gap("untargeted_kpis", 2,
f"KPIs reported without targets ({len(untargeted)})",
"These KPIs appear in the deck but carry no target from any source "
f"(pinned, prior guidance, or in-deck): {', '.join(sorted(untargeted))}. "
"Untargeted KPIs earn no quant credit and can't be held to account.",
"State a target next to every KPI — or set pinned targets in "
"Configure Companies for the ones the board owns.")
# Forward guidance: did the latest deck leave targets for any FUTURE period?
latest_key = decks_mod.period_sort_key(latest.get("period"))
extracted = company.get("extracted_targets") or {}
has_forward = any(decks_mod.period_sort_key(p) > latest_key for p in extracted)
if not has_forward:
gap("no_forward_guidance", 3, "No forward guidance for next period",
"No forward targets for a future period were extracted from the "
"latest deck, so the next deck's forecast-integrity bucket will be "
"unscorable and management can't be graded against its own plan.",
"Include an explicit next-period target for each headline KPI "
"(the number, the period, and the direction).")
if len(records) >= 2 and (q.get("forecast_integrity") or {}).get("na"):
gap("forecast_chain_broken", 2, "Forecast chain broken",
"This deck's actuals could not be matched against the prior deck's "
"targets — either no prior guidance existed or the KPI names changed "
"between periods.",
"Keep KPI names identical period to period (or record aliases in "
"Configure Companies so renamed KPIs still chain).")
dropped = sorted(s["name"] for s in kpis.values()
if s["status"] == "not-reported" and s["accountable"])
if dropped:
shown = ", ".join(dropped[:12]) + \
(f", … and {len(dropped) - 12} more" if len(dropped) > 12 else "")
gap("dropped_kpis", 3,
f"Previously targeted KPIs vanished ({len(dropped)})",
f"KPIs that were held to a target in an earlier deck are no longer "
f"reported as of {latest.get('period')}: {shown}. Silently dropped "
"KPIs are a classic way bad news leaves a deck.",
"Restore these KPIs — or retire them explicitly with one line on why.")
thin = [(cid, c) for cid, c in categories.items()
if c.get("latest") is not None
and (c.get("latest_evidence") or 0.0) < THIN_EVIDENCE]
for cid, c in sorted(thin):
gap(f"thin_evidence_{cid}", 2,
f"{cid}. {c['title']} — scored on thin evidence",
f"The panel found almost nothing to quote for this category "
f"(evidence quality {c.get('latest_evidence') or 0.0:.0%}; scores "
"regress to neutral without support). The deck isn't showing this "
"dimension at all.",
f"Add material on {_CATEGORY_MATERIAL[cid]}.")
asks = ((latest.get("narrative") or {}).get("asks")) or []
if not asks:
gap("no_board_asks", 2, "No asks for the board",
"The deck asks the board for nothing — the board is positioned to "
"preside rather than govern (BDEF category H).",
"End the deck with the 23 decisions that matter, each with a "
"recommendation and what would make it wrong.")
# Recurring flags only demand action while still open in the latest deck —
# a flag that recurred historically but is clear now shows under "resolved".
# Scoring-derived codes already have a dedicated gap above; repeating them
# here would duplicate the request.
covered = {"no_profitability_visibility", "no_quantitative_kpis", "kpi_dropped"}
latest_codes = {f.get("code") for f in
(latest.get("penalties") or {}).get("flags") or []}
for f in flags.get("recurring") or []:
if f["code"] in latest_codes and f["code"] not in covered:
gap(f"recurring_{f['code']}", min(3, f["severity"]),
f"Recurring flag: {f['code']}",
f"Raised in {len(f['periods'])} decks ({', '.join(str(p) for p in f['periods'])}) "
f"and still open: {f['description']}",
f"Address `{f['code']}` head-on in the next deck — acknowledge it "
"and show the fix, so the penalty stops repeating.")
out.sort(key=lambda g: (-g["severity"], g["code"]))
return out
# ---------------------------------------------------------------- entry point
def analyze(company: dict, records: list[dict]) -> dict:
"""Full progress analysis for one company over its period-sorted records."""
company = company or {}
records = records or []
categories = _category_trends(records)
kpis = _kpi_trends(records)
flags = _flag_history(records)
return {
"schema_version": 1,
"slug": company.get("slug"),
"name": company.get("name") or company.get("slug"),
"deck_count": len(records),
"trend": _trend(records),
"categories": categories,
"kpis": kpis,
"flags": flags,
"gaps": _gaps(company, records, categories, kpis, flags),
}
# ---------------------------------------------------------------- markdown
def _fmt(x, digits: int = 1) -> str:
return "—" if x is None else (f"{x:.{digits}f}" if isinstance(x, float) else str(x))
def _signed(x, digits: int = 1) -> str:
if x is None:
return "—"
return f"{'+' if x > 0 else ''}{x:.{digits}f}"
_STATUS_LABEL = {
"on-track": "on track", "recovered": "recovered", "slipped": "slipped",
"missed-latest": "missed latest", "missing-repeatedly": "missing repeatedly",
"untargeted": "no target", "not-reported": "NOT REPORTED",
}
def render_progress_md(analysis: dict) -> str:
"""PROGRESS.md — the analyze() result as a readable review."""
a = analysis
lines = [f"# Progress review — {a.get('name') or '?'}", ""]
n = a.get("deck_count") or 0
if not n:
lines.append("No graded decks yet.")
return "\n".join(lines) + "\n"
t = a.get("trend") or {}
series = t.get("series") or []
scored = [s for s in series if s.get("composite") is not None]
# --- verdict
lines.append("## Verdict")
lines.append("")
if len(scored) >= 2:
first, last = scored[0], scored[-1]
lines.append(
f"Trajectory over {n} graded decks: **{t.get('direction')}** — composite "
f"{_fmt(first['composite'])} ({first.get('period')}) → "
f"{_fmt(last['composite'])} ({last.get('period')}), "
f"net {_signed(t.get('net_change'))}. "
f"Best {_fmt((t.get('best') or {}).get('composite'))} "
f"({(t.get('best') or {}).get('period')}); "
f"worst {_fmt((t.get('worst') or {}).get('composite'))} "
f"({(t.get('worst') or {}).get('period')}).")
else:
lines.append(f"Only one graded deck ({scored[0].get('period') if scored else '?'}) — "
"trajectory starts with the next deck.")
gaps = a.get("gaps") or []
if gaps:
lines.append("")
lines.append(f"Deck quality: **{len(gaps)} gap(s)** in what the material shows "
"— see “What the materials aren't showing” below.")
lines.append("")
# --- composite table
lines.append("## Composite progress")
lines.append("")
lines.append("| Period | Composite | Δ | Quant | Qual | Penalties |")
lines.append("|---|---|---|---|---|---|")
for s in series:
lines.append(f"| {s.get('period') or '?'} | {_fmt(s.get('composite'))} "
f"| {_signed(s.get('delta'))} | {_fmt(s.get('quant'))} "
f"| {_fmt(s.get('qual'))} | {_fmt(s.get('penalties'))} |")
lines.append("")
# --- what's moving
cats = a.get("categories") or {}
kpis = a.get("kpis") or {}
up = [(cid, c) for cid, c in cats.items() if c.get("direction") == "improving"]
down = [(cid, c) for cid, c in cats.items() if c.get("direction") == "declining"]
tracked = [s for s in kpis.values() if s["accountable"]]
untracked = len(kpis) - len(tracked)
kpi_good = [s for s in tracked if s["status"] in ("on-track", "recovered")]
kpi_bad = [s for s in tracked
if s["status"] in ("slipped", "missing-repeatedly", "missed-latest",
"not-reported")]
lines.append("## What's moving")
lines.append("")
lines.append("### Improving")
lines.append("")
for cid, c in sorted(up, key=lambda x: -(x[1].get("net_change") or 0)):
lines.append(f"- **{cid}. {c['title']}** — {_fmt(c.get('latest'), 2)} "
f"({_signed(c.get('net_change'), 2)} since first deck)")
for s in sorted(kpi_good, key=lambda s: s["name"].lower()):
label = _STATUS_LABEL[s["status"]]
lines.append(f"- KPI **{s['name']}** — {label} "
f"({s['hits']}/{s['attempts']} targets hit)")
if not up and not kpi_good:
lines.append("- Nothing improving yet.")
lines.append("")
lines.append("### Declining / watch")
lines.append("")
for cid, c in sorted(down, key=lambda x: (x[1].get("net_change") or 0)):
lines.append(f"- **{cid}. {c['title']}** — {_fmt(c.get('latest'), 2)} "
f"({_signed(c.get('net_change'), 2)} since first deck)")
for s in sorted(kpi_bad, key=lambda s: s["name"].lower()):
lines.append(f"- KPI **{s['name']}** — {_STATUS_LABEL[s['status']]}"
+ (f" (last credit {_fmt(s['last_credit'], 2)})"
if s["last_credit"] is not None else ""))
if not down and not kpi_bad:
lines.append("- Nothing declining.")
lines.append("")
# --- KPI trajectory (only KPIs that were ever held to a target)
if tracked:
lines.append("## KPI trajectory")
lines.append("")
lines.append("| KPI | Profit | Trend (oldest → latest) | Hits | Status |")
lines.append("|---|---|---|---|---|")
ordered = sorted(tracked,
key=lambda s: (not s["profitability"], s["name"].lower()))
for s in ordered:
marks = []
for p in s["series"]:
c = p.get("credit")
marks.append("·" if c is None else "✓" if c >= 0.999 else "✗")
lines.append(f"| {s['name']} | {'yes' if s['profitability'] else ''} "
f"| {' '.join(marks)} | {s['hits']}/{s['attempts']} "
f"| {_STATUS_LABEL[s['status']]} |")
lines.append("")
lines.append("✓ target hit · ✗ missed · `·` no target that period")
if untracked:
lines.append(f"({untracked} more KPI(s) were mentioned in decks without "
"ever carrying a target — not tracked here.)")
lines.append("")
elif untracked:
lines.append("## KPI trajectory")
lines.append("")
lines.append(f"No KPI has ever carried a target — {untracked} KPI(s) were "
"mentioned across the decks but none can be held to account.")
lines.append("")
# --- flags
fl = a.get("flags") or {}
lines.append("## Flag history")
lines.append("")
rec = fl.get("recurring") or []
if rec:
lines.append("| Recurring flag | Severity | Seen in |")
lines.append("|---|---|---|")
for f in rec:
lines.append(f"| `{f['code']}` | {f['severity']} "
f"| {', '.join(str(p) for p in f['periods'])} |")
lines.append("")
new = fl.get("new_in_latest") or []
if new:
lines.append("New in the latest deck: " +
", ".join(f"`{f['code']}`" for f in new))
lines.append("")
res = fl.get("resolved") or []
if res:
lines.append("Resolved (raised earlier, clear in the latest deck): " +
", ".join(f"`{f['code']}`" for f in res))
lines.append("")
if not rec and not new and not res:
lines.append("No red flags raised on any graded deck.")
lines.append("")
# --- deck quality gaps
lines.append("## What the materials aren't showing")
lines.append("")
if gaps:
for g in gaps:
lines.append(f"### {g['title']}")
lines.append("")
lines.append(g["detail"])
lines.append("")
else:
lines.append("No material gaps detected — the deck gives the pipeline "
"everything it needs to grade this company.")
lines.append("")
# --- the ask list
if gaps:
lines.append("## Requests for the next deck")
lines.append("")
seen: set[str] = set()
i = 0
for g in gaps:
ask = g["ask"].strip()
if ask.lower() in seen:
continue
seen.add(ask.lower())
i += 1
lines.append(f"{i}. {ask}")
lines.append("")
lines.append("*Paste this list into the board follow-up — each item maps "
"to a gap above.*")
return "\n".join(lines).rstrip() + "\n"