Jonathan KirkwoodandClaude Fable 5
b1d7aed9f4
- Deterministic scoring.py (quant 60 / qual 40 / flags -15, profitability heaviest) - Per-company JSON ledger with forecast-target chaining deck N-1 -> N - Single-shot sandbox agent with guided-JSON fallback ladder (no tool loop) - Portfolio dashboard with sparklines, KPI hit rates, BDEF category bars - 48 unit tests green; endpoints smoke-tested; npm check+build green Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
328 lines
14 KiB
Python
328 lines
14 KiB
Python
"""Deterministic BDEF scoring — pure functions, stdlib only, no I/O.
|
|
|
|
Everything numeric happens here, never in a model: the panel supplies 1-5
|
|
category scores with verbatim evidence, the extractor supplies KPI actuals and
|
|
targets, and this module turns them into the 0-100 composite:
|
|
|
|
composite = quant (60) + qualitative (40) - red-flag penalties (cap 15)
|
|
|
|
All knobs come from the `weights` dict (bm_config.WEIGHTS_DEFAULTS shape) so
|
|
the operator can retune without a rebuild.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import re
|
|
import statistics
|
|
|
|
FUZZY_THRESHOLD = 0.85
|
|
_CATEGORIES = "ABCDEFGH"
|
|
|
|
|
|
# ---------------------------------------------------------------- KPI matching
|
|
def _norm(name: str) -> str:
|
|
return re.sub(r"[^a-z0-9]+", " ", (name or "").lower()).strip()
|
|
|
|
|
|
def match_kpi(canonical: str, candidates: list[dict], aliases: dict) -> tuple[dict | None, str | None]:
|
|
"""Match a canonical KPI name against candidate dicts ({canonical_name, name}).
|
|
|
|
Exact canonical -> alias map (either direction) -> fuzzy ratio >= 0.85.
|
|
Returns (candidate, via) with via in exact|alias|fuzzy, or (None, None)."""
|
|
canon = (canonical or "").strip().lower()
|
|
if not canon:
|
|
return None, None
|
|
for c in candidates:
|
|
if (c.get("canonical_name") or "").strip().lower() == canon:
|
|
return c, "exact"
|
|
amap = {(k or "").strip().lower(): {(a or "").strip().lower() for a in (v or [])}
|
|
for k, v in (aliases or {}).items()}
|
|
ours = amap.get(canon, set())
|
|
for c in candidates:
|
|
cn = (c.get("canonical_name") or "").strip().lower()
|
|
nm = (c.get("name") or "").strip().lower()
|
|
if cn in ours or nm in ours or canon in amap.get(cn, set()):
|
|
return c, "alias"
|
|
best, best_r = None, 0.0
|
|
for c in candidates:
|
|
for other in (c.get("canonical_name") or "", c.get("name") or ""):
|
|
r = difflib.SequenceMatcher(None, _norm(canon), _norm(other)).ratio()
|
|
if r > best_r:
|
|
best, best_r = c, r
|
|
if best is not None and best_r >= FUZZY_THRESHOLD:
|
|
return best, "fuzzy"
|
|
return None, None
|
|
|
|
|
|
# ---------------------------------------------------------------- KPI credit
|
|
def _passes(actual: float, target: float, direction: str) -> bool:
|
|
return actual <= target if direction == "lte" else actual >= target
|
|
|
|
|
|
def _credit(actual: float, target: float, direction: str, floor: float) -> float:
|
|
"""Partial credit for a targeted KPI: 0 below floor, linear to 1 at target."""
|
|
if target == 0 or (actual < 0) != (target < 0) or (direction == "lte" and actual == 0):
|
|
return 1.0 if _passes(actual, target, direction) else 0.0
|
|
r = target / actual if direction == "lte" else actual / target
|
|
if actual < 0 and target < 0:
|
|
# Both negative (EBITDA margin target -2, actual -3): the plain ratio
|
|
# inverts the ordering, so flip it back.
|
|
r = 1.0 / r
|
|
if r >= 1:
|
|
return 1.0
|
|
if floor >= 1 or r < floor:
|
|
return 0.0
|
|
return max(0.0, min(1.0, (r - floor) / (1 - floor)))
|
|
|
|
|
|
def _resolve_target(kpi: dict, pinned_targets: list[dict], prior_targets: list[dict],
|
|
aliases: dict) -> tuple[dict | None, str | None, str | None]:
|
|
"""(target dict {target, direction}, target_source, matched_via) for one KPI.
|
|
|
|
Precedence: pinned config target > prior deck's extracted forward target
|
|
for this period > target printed in the deck itself."""
|
|
canon = kpi.get("canonical_name") or ""
|
|
pinned_cands = [{"canonical_name": p.get("kpi"), "name": p.get("kpi"), "_src": p}
|
|
for p in (pinned_targets or [])]
|
|
cand, via = match_kpi(canon, pinned_cands, aliases)
|
|
if cand is not None:
|
|
p = cand["_src"]
|
|
return {"target": p.get("target"), "direction": p.get("direction") or kpi.get("direction")}, "pinned", via
|
|
cand, via = match_kpi(canon, prior_targets or [], aliases)
|
|
if cand is not None:
|
|
return {"target": cand.get("target"),
|
|
"direction": cand.get("direction") or kpi.get("direction")}, "extracted", via
|
|
if kpi.get("target_in_deck") is not None:
|
|
return {"target": kpi["target_in_deck"], "direction": kpi.get("direction")}, "in_deck", None
|
|
return None, None, None
|
|
|
|
|
|
# ---------------------------------------------------------------- scoring
|
|
def _bucket(results: list[dict], weight: float) -> dict:
|
|
if not results:
|
|
return {"score": 0.0, "weight": 0.0, "na": True, "kpi_count": 0}
|
|
mean = sum(r["credit"] for r in results) / len(results)
|
|
return {"score": mean, "weight": weight, "na": False, "kpi_count": len(results)}
|
|
|
|
|
|
def _qual(grades: list[dict], weights: dict) -> tuple[float, dict]:
|
|
"""(qual score, per-category detail). Evidence regresses medians toward 3."""
|
|
full_credit = max(1, int(weights.get("evidenceFullCredit", 400)))
|
|
cat_max = float(weights.get("qualCategoryMax", 5))
|
|
out: dict[str, dict] = {}
|
|
total = 0.0
|
|
for cid in _CATEGORIES:
|
|
scores: list[int] = []
|
|
equalities: list[float] = []
|
|
rationales: list[dict] = []
|
|
for g in grades or []:
|
|
cat = next((c for c in (g.get("categories") or []) if c.get("id") == cid), None)
|
|
if cat is None:
|
|
continue
|
|
evidence = cat.get("evidence") or []
|
|
quote_chars = sum(min(len(ev.get("quote") or ""), 200) for ev in evidence)
|
|
scores.append(int(cat.get("score", 3)))
|
|
equalities.append(min(1.0, quote_chars / full_credit))
|
|
rationales.append({
|
|
"grader": g.get("grader") or "grader",
|
|
"rationale": cat.get("rationale") or "",
|
|
"evidence": [{"quote": ev.get("quote") or "", "location": ev.get("location") or ""}
|
|
for ev in evidence],
|
|
})
|
|
if scores:
|
|
median = float(statistics.median(scores))
|
|
e_mean = sum(equalities) / len(equalities)
|
|
else:
|
|
median, e_mean = 3.0, 0.0
|
|
adjusted = 3.0 + (median - 3.0) * e_mean
|
|
points = adjusted * cat_max / 5.0
|
|
total += points
|
|
out[cid] = {
|
|
"panel_scores": scores,
|
|
"median": round(median, 2),
|
|
"evidence_quality": round(e_mean, 4),
|
|
"adjusted": round(adjusted, 4),
|
|
"points": round(points, 4),
|
|
"rationales": rationales,
|
|
}
|
|
return total, out
|
|
|
|
|
|
def score_deck(extraction: dict, grades: list[dict], pinned_targets: list[dict],
|
|
prior_targets: list[dict], kpi_aliases: dict, weights: dict,
|
|
meta: dict) -> dict:
|
|
"""Score one graded deck into the canonical ledger record. Pure."""
|
|
kpis = extraction.get("kpis") or []
|
|
floor = float(weights.get("kpiCreditFloor", 0.5))
|
|
scoring_flags: list[dict] = []
|
|
|
|
# --- per-KPI target resolution + credit
|
|
kpi_results: list[dict] = []
|
|
for k in kpis:
|
|
tgt, source, via = _resolve_target(k, pinned_targets, prior_targets, kpi_aliases)
|
|
credit = None
|
|
target = None
|
|
if tgt is not None and tgt.get("target") is not None:
|
|
target = float(tgt["target"])
|
|
credit = _credit(float(k.get("actual", 0)), target,
|
|
tgt.get("direction") or k.get("direction") or "gte", floor)
|
|
kpi_results.append({
|
|
"canonical_name": k.get("canonical_name"), "name": k.get("name"),
|
|
"actual": k.get("actual"), "unit": k.get("unit") or "",
|
|
"direction": k.get("direction"), "profitability": bool(k.get("profitability")),
|
|
"target": target, "target_source": source, "matched_via": via,
|
|
"credit": None if credit is None else round(credit, 4),
|
|
})
|
|
|
|
prof_all = [r for r in kpi_results if r["profitability"]]
|
|
prof_hit = [r for r in prof_all if r["credit"] is not None]
|
|
other_hit = [r for r in kpi_results if not r["profitability"] and r["credit"] is not None]
|
|
|
|
# --- forecast integrity: this deck's actuals vs the prior deck's targets
|
|
forecast_results: list[dict] = []
|
|
dropped: list[str] = []
|
|
for pt in prior_targets or []:
|
|
cand, _via = match_kpi(pt.get("canonical_name") or "", kpis, kpi_aliases)
|
|
if cand is None:
|
|
dropped.append(pt.get("canonical_name") or pt.get("name") or "kpi")
|
|
continue
|
|
actual = float(cand.get("actual", 0))
|
|
target = float(pt.get("target", 0))
|
|
direction = pt.get("direction") or "gte"
|
|
if target == 0:
|
|
acc = 1.0 if _passes(actual, target, direction) else 0.0
|
|
else:
|
|
err = (actual - target) / abs(target)
|
|
if direction == "lte":
|
|
err = -err
|
|
e = abs(err) if err < 0 else abs(err) / 2.0 # overshoot penalized half
|
|
acc = 1.0 - min(1.0, e)
|
|
forecast_results.append({
|
|
"canonical_name": pt.get("canonical_name"), "target": target,
|
|
"actual": actual, "accuracy": round(acc, 4),
|
|
})
|
|
|
|
# Pinned targets that no reported actual matches count as dropped too.
|
|
for p in pinned_targets or []:
|
|
cand, _via = match_kpi(p.get("kpi") or "", kpis, kpi_aliases)
|
|
if cand is None:
|
|
dropped.append(p.get("kpi") or "kpi")
|
|
seen: set[str] = set()
|
|
dropped_unique = [d for d in dropped
|
|
if not (d.strip().lower() in seen or seen.add(d.strip().lower()))]
|
|
for name in dropped_unique[: int(weights.get("droppedKpiMax", 3))]:
|
|
scoring_flags.append({
|
|
"code": "kpi_dropped",
|
|
"description": f"previously targeted KPI '{name}' is not reported this period",
|
|
"severity": int(weights.get("droppedKpiPenalty", 2)),
|
|
})
|
|
|
|
# --- quant buckets + renormalization (NA weight redistributes pro-rata)
|
|
prof = _bucket(prof_hit, float(weights.get("profitabilityKpi", 30)))
|
|
other = _bucket(other_hit, float(weights.get("otherKpi", 20)))
|
|
fmean = (sum(f["accuracy"] for f in forecast_results) / len(forecast_results)
|
|
if forecast_results else 0.0)
|
|
if forecast_results:
|
|
forecast = {"score": fmean, "weight": float(weights.get("forecastIntegrity", 10)),
|
|
"na": False, "kpi_count": len(forecast_results)}
|
|
else:
|
|
forecast = {"score": 0.0, "weight": 0.0, "na": True, "kpi_count": 0}
|
|
|
|
if not prof_all:
|
|
prof["na"] = True
|
|
prof["weight"] = 0.0
|
|
scoring_flags.append({
|
|
"code": "no_profitability_visibility",
|
|
"description": "no profit/margin/cash KPI reported at all",
|
|
"severity": 3,
|
|
})
|
|
|
|
total_quant_w = (float(weights.get("profitabilityKpi", 30))
|
|
+ float(weights.get("otherKpi", 20))
|
|
+ float(weights.get("forecastIntegrity", 10)))
|
|
present = [b for b in (prof, other, forecast) if not b["na"]]
|
|
if present:
|
|
scale = total_quant_w / sum(b["weight"] for b in present)
|
|
for b in present:
|
|
b["weight"] = round(b["weight"] * scale, 4)
|
|
b["score"] = round(b["score"] * b["weight"], 4)
|
|
quant_score = sum(b["score"] for b in present)
|
|
all_quant_na = False
|
|
else:
|
|
quant_score = 0.0
|
|
all_quant_na = True
|
|
scoring_flags.append({
|
|
"code": "no_quantitative_kpis",
|
|
"description": "no quantitative bucket could be scored (no targeted KPIs, "
|
|
"no prior targets)",
|
|
"severity": 4,
|
|
})
|
|
|
|
# --- qualitative
|
|
qual_score, categories = _qual(grades, weights)
|
|
qual_max = 8.0 * float(weights.get("qualCategoryMax", 5))
|
|
|
|
# --- red flags: extractor + graders + scoring; dedup, damp single-source
|
|
damp = float(weights.get("singleSourceFlagFactor", 0.5))
|
|
cap = float(weights.get("redFlagCap", 15))
|
|
flag_map: dict[str, dict] = {}
|
|
|
|
def add_flag(f: dict, source: str, scoring_flag: bool = False):
|
|
code = (f.get("code") or "flag").strip().lower()
|
|
key = f"{code}:{f.get('description', '')}" if scoring_flag and code == "kpi_dropped" else code
|
|
sev = int(f.get("severity", 1))
|
|
cur = flag_map.get(key)
|
|
if cur is None:
|
|
flag_map[key] = {"code": code, "description": f.get("description") or "",
|
|
"severity": sev, "sources": {source}, "scoring": scoring_flag}
|
|
else:
|
|
if sev > cur["severity"]:
|
|
cur["severity"] = sev
|
|
cur["description"] = f.get("description") or cur["description"]
|
|
cur["sources"].add(source)
|
|
cur["scoring"] = cur["scoring"] or scoring_flag
|
|
|
|
for f in extraction.get("red_flag_candidates") or []:
|
|
add_flag(f, "extractor")
|
|
for g in grades or []:
|
|
for f in g.get("red_flags") or []:
|
|
add_flag(f, g.get("grader") or "grader")
|
|
for f in scoring_flags:
|
|
add_flag(f, "scoring", scoring_flag=True)
|
|
|
|
flags: list[dict] = []
|
|
for f in flag_map.values():
|
|
full = f["scoring"] or len(f["sources"]) >= 2
|
|
points = float(f["severity"]) if full else float(f["severity"]) * damp
|
|
flags.append({"code": f["code"], "description": f["description"],
|
|
"severity": f["severity"], "points": round(points, 4),
|
|
"sources": sorted(f["sources"])})
|
|
flags.sort(key=lambda f: (-f["points"], f["code"]))
|
|
penalty_total = round(min(cap, sum(f["points"] for f in flags)), 4)
|
|
|
|
# --- composite
|
|
if all_quant_na:
|
|
base = (qual_score / qual_max * 100.0) if qual_max else 0.0
|
|
else:
|
|
base = quant_score + qual_score
|
|
composite = round(max(0.0, min(100.0, base - penalty_total)), 1)
|
|
|
|
return {
|
|
"schema_version": 1,
|
|
"company": meta.get("company"),
|
|
"period": meta.get("period"),
|
|
"deck_id": meta.get("deck_id"),
|
|
"job_id": meta.get("job_id"),
|
|
"graded_at": meta.get("graded_at"),
|
|
"composite": composite,
|
|
"quant": {"score": round(quant_score, 4), "profitability": prof,
|
|
"other": other, "forecast_integrity": forecast},
|
|
"qual": {"score": round(qual_score, 4), "categories": categories},
|
|
"penalties": {"total": penalty_total, "flags": flags},
|
|
"kpi_results": kpi_results,
|
|
"forecast_results": forecast_results,
|
|
"panel": meta.get("panel") or [],
|
|
"artifacts": meta.get("artifacts", {}),
|
|
"narrative": extraction.get("narrative", {}),
|
|
}
|