jonathan/boardroom-map
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6bf0786993b8eb520f8e9db286fc1e44e2d2230c
boardroom-map/orchestrator/tests/test_scoring.py
T
Jonathan KirkwoodandClaude Fable 5 b1d7aed9f4 Implement BDEF v1.1 grading: scoring core, per-deck pipeline, ledger, dashboard, StartOS layer 
- 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>
2026-07-06 14:15:12 -05:00

424 lines
20 KiB
Python
Raw Blame History

"""Tests for scoring.py (pure scorer), validate.py, and the fixture-driven
Q1 -> Q2 end-to-end flow through the ledger and scorecard renderers."""
import json
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import ledger as ledger_mod
import scorecard
import scoring
import validate
FIXTURES = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures")
WEIGHTS = {
"profitabilityKpi": 30, "otherKpi": 20, "forecastIntegrity": 10,
"qualCategoryMax": 5, "redFlagCap": 15, "kpiCreditFloor": 0.5,
"droppedKpiPenalty": 2, "droppedKpiMax": 3, "evidenceFullCredit": 400,
"singleSourceFlagFactor": 0.5,
}
PINNED_CASH = [{"kpi": "cash_balance", "target": 12.0, "unit": "$M",
"direction": "gte", "profitability": True}]
def _fixture(name):
with open(os.path.join(FIXTURES, name), encoding="utf-8") as f:
return json.load(f)
def _kpi(canonical, actual, direction="gte", prof=False, tid=None, name=None, unit=""):
return {"name": name or canonical, "canonical_name": canonical, "actual": actual,
"unit": unit, "period": None, "direction": direction, "profitability": prof,
"target_in_deck": tid, "source": "slide 1", "notes": ""}
def _ft(canonical, target, direction="gte", target_period="2026-Q2", prof=False):
return {"name": canonical, "canonical_name": canonical, "target": target,
"unit": "", "target_period": target_period, "direction": direction,
"profitability": prof, "source": "slide 9"}
def _extraction(kpis=None, forward=None, flags=None, period="2026-Q2"):
return {"schema_version": 1, "deck": {"period": period},
"kpis": kpis or [], "forward_targets": forward or [],
"red_flag_candidates": flags or [],
"narrative": {"summary": "test deck", "asks": []}}
def _grade(grader="grader-a", score=3, quote_chars=0, red_flags=None, overrides=None):
cats = []
for cid in "ABCDEFGH":
s, qc = score, quote_chars
if overrides and cid in overrides:
s, qc = overrides[cid]
ev = [{"quote": "q" * qc, "location": "slide 1"}] if qc else []
cats.append({"id": cid, "score": s, "evidence": ev, "rationale": f"cat {cid}"})
return {"schema_version": 1, "grader": grader, "categories": cats,
"red_flags": red_flags or [], "overall_comment": "ok"}
def _meta(period="2026-Q2", deck_id="d1"):
return {"company": "acme", "period": period, "deck_id": deck_id, "job_id": "job-1",
"graded_at": "2026-07-06T12:00:00Z",
"panel": [{"rid": "grader-a", "model": "grader-a", "valid": True}],
"artifacts": {"extraction": "extraction.json"}}
def _score(extraction, grades=None, pinned=None, prior=None, aliases=None, meta=None):
return scoring.score_deck(extraction, grades if grades is not None else [_grade()],
pinned or [], prior or [], aliases or {}, WEIGHTS,
meta or _meta())
def _flag(rec, code):
return [f for f in rec["penalties"]["flags"] if f["code"] == code]
class TestMatchKpi(unittest.TestCase):
def test_exact(self):
cand, via = scoring.match_kpi("arr", [{"canonical_name": "arr", "name": "ARR"}], {})
self.assertEqual(via, "exact")
self.assertEqual(cand["name"], "ARR")
def test_alias_forward_and_reverse(self):
aliases = {"arr": ["Annual Recurring Revenue", "run_rate_arr"]}
cand, via = scoring.match_kpi(
"arr", [{"canonical_name": "revenue_annualized",
"name": "Annual Recurring Revenue"}], aliases)
self.assertEqual(via, "alias")
cand, via = scoring.match_kpi(
"run_rate_arr", [{"canonical_name": "arr", "name": "ARR"}], aliases)
self.assertEqual(via, "alias")
def test_fuzzy(self):
cand, via = scoring.match_kpi(
"ebitda_margin", [{"canonical_name": "ebitda_margins", "name": "x"}], {})
self.assertEqual(via, "fuzzy")
def test_no_match(self):
self.assertEqual(
scoring.match_kpi("arr", [{"canonical_name": "cash_balance", "name": "Cash"}], {}),
(None, None))
self.assertEqual(scoring.match_kpi("", [{"canonical_name": "arr"}], {}), (None, None))
class TestCredit(unittest.TestCase):
def test_lte_credit(self):
rec = _score(_extraction([_kpi("churn_rate", 6.0, "lte", tid=5.0)]))
self.assertAlmostEqual(rec["kpi_results"][0]["credit"], 0.6667, places=4)
rec = _score(_extraction([_kpi("churn_rate", 4.0, "lte", tid=5.0)]))
self.assertEqual(rec["kpi_results"][0]["credit"], 1.0)
def test_floor(self):
rec = _score(_extraction([_kpi("arr", 4.0, tid=10.0)])) # r=0.4 < floor
self.assertEqual(rec["kpi_results"][0]["credit"], 0.0)
rec = _score(_extraction([_kpi("arr", 7.5, tid=10.0)])) # r=0.75 -> 0.5
self.assertAlmostEqual(rec["kpi_results"][0]["credit"], 0.5, places=4)
def test_guards(self):
self.assertEqual(scoring._credit(5, 0, "gte", 0.5), 1.0) # zero target, passes
self.assertEqual(scoring._credit(-5, 0, "gte", 0.5), 0.0) # zero target, fails
self.assertEqual(scoring._credit(-1, 1, "gte", 0.5), 0.0) # sign mismatch, fails
self.assertEqual(scoring._credit(1, -1, "gte", 0.5), 1.0) # sign mismatch, passes
self.assertEqual(scoring._credit(0, 5, "lte", 0.5), 1.0) # lte zero actual
def test_negative_targets(self):
# EBITDA margin: target -2, actual -3 -> two thirds of the way -> 0.3333
self.assertAlmostEqual(scoring._credit(-3, -2, "gte", 0.5), 1 / 3, places=4)
self.assertEqual(scoring._credit(-1, -2, "gte", 0.5), 1.0)
class TestQuantBuckets(unittest.TestCase):
def test_first_deck_renormalization(self):
# No prior targets -> forecast NA -> its 10 points redistribute 36/24.
rec = _score(_extraction([_kpi("ebitda_margin", 5.0, prof=True, tid=5.0),
_kpi("arr", 10.0, tid=10.0)]))
q = rec["quant"]
self.assertTrue(q["forecast_integrity"]["na"])
self.assertAlmostEqual(q["profitability"]["weight"], 36.0)
self.assertAlmostEqual(q["profitability"]["score"], 36.0)
self.assertAlmostEqual(q["other"]["weight"], 24.0)
self.assertAlmostEqual(q["other"]["score"], 24.0)
self.assertAlmostEqual(q["score"], 60.0)
self.assertEqual(rec["penalties"]["flags"], [])
self.assertAlmostEqual(rec["composite"], 84.0) # 60 quant + 24 qual (all 3s)
def test_forecast_integrity_second_deck(self):
prior = [_ft("arr", 12.0), _ft("churn_rate", 4.0, "lte")]
rec = _score(_extraction([_kpi("arr", 11.0), _kpi("churn_rate", 3.5, "lte"),
_kpi("fcf", 1.0, prof=True, tid=1.0)]),
prior=prior)
fi = rec["quant"]["forecast_integrity"]
self.assertFalse(fi["na"])
self.assertEqual(fi["weight"], 10.0)
self.assertEqual(fi["kpi_count"], 2)
accs = {f["canonical_name"]: f["accuracy"] for f in rec["forecast_results"]}
self.assertAlmostEqual(accs["arr"], 0.9167, places=4) # 1/12 undershoot
self.assertAlmostEqual(accs["churn_rate"], 0.9375, places=4) # overshoot halved
self.assertAlmostEqual(fi["score"], (0.9167 + 0.9375) / 2 * 10, places=3)
def test_no_profitability_flag_and_redistribution(self):
rec = _score(_extraction([_kpi("arr", 10.0, tid=10.0)]))
q = rec["quant"]
self.assertTrue(q["profitability"]["na"])
self.assertTrue(q["forecast_integrity"]["na"])
self.assertAlmostEqual(q["other"]["weight"], 60.0)
self.assertAlmostEqual(q["score"], 60.0)
flags = _flag(rec, "no_profitability_visibility")
self.assertEqual(len(flags), 1)
self.assertEqual(flags[0]["points"], 3.0) # scoring flags never damped
self.assertEqual(flags[0]["sources"], ["scoring"])
def test_profitability_kpis_without_targets_na_no_flag(self):
rec = _score(_extraction([_kpi("ebitda_margin", -5.0, prof=True),
_kpi("arr", 10.0, tid=10.0)]))
self.assertTrue(rec["quant"]["profitability"]["na"])
self.assertEqual(_flag(rec, "no_profitability_visibility"), [])
def test_all_quant_na_scales_qual(self):
rec = _score(_extraction([]))
# qual 24 (all 3s) scaled to 60, minus no_profitability(3) + no_quantitative(4)
self.assertTrue(all(rec["quant"][b]["na"] for b in
("profitability", "other", "forecast_integrity")))
self.assertEqual(len(_flag(rec, "no_quantitative_kpis")), 1)
self.assertAlmostEqual(rec["composite"], 53.0)
class TestTargetPrecedence(unittest.TestCase):
def test_pinned_beats_extracted_beats_in_deck(self):
kpis = [_kpi("arr", 11.0, tid=9.0)]
pinned = [{"kpi": "arr", "target": 10.0, "unit": "$M",
"direction": "gte", "profitability": False}]
prior = [_ft("arr", 12.0)]
r = _score(_extraction(kpis), pinned=pinned, prior=prior)["kpi_results"][0]
self.assertEqual((r["target"], r["target_source"], r["matched_via"]),
(10.0, "pinned", "exact"))
self.assertEqual(r["credit"], 1.0)
r = _score(_extraction(kpis), prior=prior)["kpi_results"][0]
self.assertEqual((r["target"], r["target_source"]), (12.0, "extracted"))
self.assertAlmostEqual(r["credit"], 0.8333, places=4)
r = _score(_extraction(kpis))["kpi_results"][0]
self.assertEqual((r["target"], r["target_source"], r["matched_via"]),
(9.0, "in_deck", None))
def test_untargeted_kpi_reported_with_none(self):
r = _score(_extraction([_kpi("nps", 40.0)]))["kpi_results"][0]
self.assertIsNone(r["target"])
self.assertIsNone(r["credit"])
self.assertIsNone(r["target_source"])
class TestQualitative(unittest.TestCase):
def test_evidence_regression_both_directions(self):
# Median 5 with no quotes regresses to 3; so does median 1.
rec = _score(_extraction([]), grades=[_grade(score=5, quote_chars=0)])
self.assertEqual(rec["qual"]["categories"]["A"]["adjusted"], 3.0)
rec = _score(_extraction([]), grades=[_grade(score=1, quote_chars=0)])
self.assertEqual(rec["qual"]["categories"]["A"]["adjusted"], 3.0)
self.assertAlmostEqual(rec["qual"]["score"], 24.0)
def test_full_evidence_keeps_extreme_scores(self):
# Per-quote chars cap at 200, so full credit (400) needs two quotes.
g = _grade(score=5, quote_chars=200)
for cat in g["categories"]:
cat["evidence"].append({"quote": "q" * 200, "location": "slide 2"})
rec = _score(_extraction([]), grades=[g])
cat = rec["qual"]["categories"]["A"]
self.assertEqual(cat["evidence_quality"], 1.0)
self.assertEqual(cat["adjusted"], 5.0)
self.assertEqual(cat["points"], 5.0)
def test_quote_chars_capped_at_200_each(self):
# One 1000-char quote counts as 200 -> e = 0.5 -> adjusted 4.
rec = _score(_extraction([]), grades=[_grade(score=5, quote_chars=1000)])
self.assertEqual(rec["qual"]["categories"]["A"]["evidence_quality"], 0.5)
self.assertEqual(rec["qual"]["categories"]["A"]["adjusted"], 4.0)
def test_panel_median_and_rationales(self):
grades = [_grade("g1", score=4, quote_chars=400),
_grade("g2", score=4, quote_chars=400),
_grade("g3", score=2, quote_chars=400)]
rec = _score(_extraction([]), grades=grades)
cat = rec["qual"]["categories"]["B"]
self.assertEqual(cat["panel_scores"], [4, 4, 2])
self.assertEqual(cat["median"], 4.0)
self.assertEqual(len(cat["rationales"]), 3)
self.assertEqual(cat["rationales"][0]["grader"], "g1")
class TestPenalties(unittest.TestCase):
def test_single_source_damping(self):
rec = _score(_extraction([], flags=[{"code": "adjusted_metrics",
"description": "d", "severity": 4}]))
f = _flag(rec, "adjusted_metrics")[0]
self.assertEqual(f["points"], 2.0)
self.assertEqual(f["sources"], ["extractor"])
def test_two_sources_full_severity_max_wins(self):
grades = [_grade("g1", red_flags=[{"code": "governance_gap",
"description": "weak", "severity": 2}]),
_grade("g2", red_flags=[{"code": "governance_gap",
"description": "worse", "severity": 3}])]
rec = _score(_extraction([]), grades=grades)
f = _flag(rec, "governance_gap")[0]
self.assertEqual(f["severity"], 3)
self.assertEqual(f["points"], 3.0)
self.assertEqual(f["sources"], ["g1", "g2"])
def test_penalty_cap(self):
codes = ["related_party", "channel_stuffing_risk", "suppressed_dissent",
"metric_redefinition"]
flags = [{"code": c, "description": c, "severity": 5} for c in codes]
rec = _score(_extraction([_kpi("fcf", 1.0, prof=True, tid=1.0)], flags=flags),
grades=[_grade("g1", red_flags=flags)])
self.assertEqual(rec["penalties"]["total"], 15.0) # 4x5=20 capped
def test_dropped_kpi_flags_capped(self):
prior = [_ft(c, 1.0) for c in ("alpha_metric", "beta_metric", "gamma_metric",
"delta_metric", "epsilon_metric")]
rec = _score(_extraction([]), prior=prior)
dropped = _flag(rec, "kpi_dropped")
self.assertEqual(len(dropped), 3) # droppedKpiMax
for f in dropped:
self.assertEqual(f["points"], 2.0) # droppedKpiPenalty, never damped
class TestValidate(unittest.TestCase):
def test_parse_json_text(self):
self.assertEqual(validate.parse_json_text('{"a": 1}'), {"a": 1})
salvaged = validate.parse_json_text(
'Sure! Here is the JSON:\n```json\n{"a": {"b": "}"}}\n```\ntrailing prose')
self.assertEqual(salvaged, {"a": {"b": "}"}})
self.assertIsNone(validate.parse_json_text("no json here"))
self.assertIsNone(validate.parse_json_text("[1, 2, 3]"))
self.assertIsNone(validate.parse_json_text(""))
def test_schemas_load_and_fixtures_validate(self):
self.assertIn("properties", validate.load_schema("extraction"))
self.assertIn("properties", validate.load_schema("grades"))
for name, schema in (("extraction_q1.json", "extraction"),
("extraction_q2.json", "extraction"),
("grade_a.json", "grades"), ("grade_b.json", "grades"),
("grade_c.json", "grades")):
err = validate.validate_obj(_fixture(name), schema)
self.assertIsNone(err, f"{name}: {err}")
def test_validate_obj_rejects_bad(self):
self.assertIsNotNone(validate.validate_obj({"schema_version": 1}, "grades"))
def test_validate_file(self):
obj, err = validate.validate_file(os.path.join(FIXTURES, "grade_a.json"), "grades")
self.assertIsNone(err)
self.assertEqual(obj["grader"], "grader-a")
obj, err = validate.validate_file("/nonexistent.json", "grades")
self.assertIsNone(obj)
self.assertIsNotNone(err)
class TestEndToEnd(unittest.TestCase):
"""Fixture-driven Q1 -> Q2 flow: score, ledger round-trip, rendering."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.ledger = ledger_mod.Ledger(os.path.join(self._tmp.name, "ledger"))
self.grades = [_fixture("grade_a.json"), _fixture("grade_b.json"),
_fixture("grade_c.json")]
self.q1 = _fixture("extraction_q1.json")
self.q2 = _fixture("extraction_q2.json")
def tearDown(self):
self._tmp.cleanup()
def _score_q1(self):
return scoring.score_deck(self.q1, self.grades, PINNED_CASH, [], {}, WEIGHTS,
_meta("2026-Q1", "2026-Q1"))
def test_q1_first_deck(self):
rec = self._score_q1()
q = rec["quant"]
self.assertTrue(q["forecast_integrity"]["na"])
self.assertAlmostEqual(q["score"], 60.0) # every KPI at/above target
# qual: A-G 3.5 pts each (median 4, evidence 0.5), H 2.6667
self.assertAlmostEqual(rec["qual"]["score"], 27.1667, places=3)
self.assertAlmostEqual(rec["qual"]["categories"]["H"]["points"], 2.6667, places=3)
# hockey_stick (extractor only, sev 3 -> 1.5) + governance_gap (2 graders -> 3)
self.assertAlmostEqual(rec["penalties"]["total"], 4.5)
self.assertAlmostEqual(rec["composite"], 82.7)
cash = next(k for k in rec["kpi_results"] if k["canonical_name"] == "cash_balance")
self.assertEqual(cash["target_source"], "pinned")
def test_q2_against_q1_targets(self):
rec1 = self._score_q1()
self.ledger.record_deck("acme", rec1, self.q1["forward_targets"])
prior = self.ledger.prior_targets("acme", "2026-Q2")
self.assertEqual(len(prior), 4)
rec2 = scoring.score_deck(self.q2, self.grades, PINNED_CASH, prior, {}, WEIGHTS,
_meta("2026-Q2", "2026-Q2"))
by_name = {k["canonical_name"]: k for k in rec2["kpi_results"]}
self.assertAlmostEqual(by_name["arr"]["credit"], 0.8333, places=4)
self.assertEqual(by_name["churn_rate"]["credit"], 1.0)
self.assertAlmostEqual(by_name["ebitda_margin"]["credit"], 0.3333, places=4)
self.assertEqual(by_name["cash_balance"]["target_source"], "pinned")
self.assertEqual(by_name["cash_balance"]["credit"], 1.0)
q = rec2["quant"]
self.assertAlmostEqual(q["profitability"]["score"], 20.0, places=2)
self.assertAlmostEqual(q["other"]["score"], 18.333, places=2)
self.assertAlmostEqual(q["forecast_integrity"]["score"], 7.847, places=2)
self.assertEqual(len(rec2["forecast_results"]), 3)
# qualified_pipeline guided in Q1 but not reported in Q2 -> dropped flag
dropped = _flag(rec2, "kpi_dropped")
self.assertEqual(len(dropped), 1)
self.assertIn("qualified_pipeline", dropped[0]["description"])
# adjusted_metrics 1.0 + governance_gap 3.0 + kpi_dropped 2.0
self.assertAlmostEqual(rec2["penalties"]["total"], 6.0)
self.assertAlmostEqual(rec2["composite"], 67.3)
self.assertAlmostEqual(
rec2["composite"],
round(q["score"] + rec2["qual"]["score"] - rec2["penalties"]["total"], 1))
# ledger round-trip + rendering
self.ledger.record_deck("acme", rec2, self.q2["forward_targets"])
records = self.ledger.deck_records("acme")
self.assertEqual([r["period"] for r in records], ["2026-Q1", "2026-Q2"])
report = scorecard.render_deck_report(rec2, self.q2, adjudication_md="Chair memo.")
self.assertIn("67.3", report)
self.assertIn("pinned", report)
self.assertIn("## Panel adjudication", report)
self.assertIn("Chair memo.", report)
self.assertIn("kpi_dropped", report)
card = scorecard.render_scorecard(self.ledger.get_company("acme"), records)
self.assertIn("2026-Q1", card)
self.assertIn("2026-Q2", card)
self.assertIn("↓", card) # composite fell Q1 -> Q2
self.assertIn("KPI hit-rate", card)
self.assertIn("arr", card)
def test_meta_passthrough_and_record_shape(self):
rec = self._score_q1()
self.assertEqual(rec["company"], "acme")
self.assertEqual(rec["deck_id"], "2026-Q1")
self.assertEqual(rec["job_id"], "job-1")
self.assertEqual(rec["panel"][0]["rid"], "grader-a")
self.assertEqual(rec["artifacts"], {"extraction": "extraction.json"})
self.assertEqual(rec["schema_version"], 1)
self.assertIn("summary", rec["narrative"])
for key in ("composite", "quant", "qual", "penalties", "kpi_results",
"forecast_results"):
self.assertIn(key, rec)
# the record must be JSON-serializable as produced
json.dumps(rec)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink