open/portfolio-optimiser-claude
1
0
Fork 0
Code Issues Releases 1 Activity

feat(portfolio): K11 — per-run value report, pure projection over the three layers (parity row 25) [skip-docs]

Browse source 
The S5.4 analog: every run/portfolio pass can produce a deterministic value
report — modelled → expert-corrected → realized, goal progress, a quantified
learning effect, cost against value — with no model call, no clock and no new
state. It is a PURE PROJECTION over what is already persisted (K5 outbox pairs,
§4.2 inbox verdicts, K1 ledger), joined on the verdict_id K5 mints.

The honesty rule (§1) sets the shape, not the layout:
- approved      -> the claim stands        (quantified)
- rejected      -> the claim is void, 0    (quantified — an earned zero)
- approved_with_adjustment -> §4.2 carries NO adjusted amount, so the corrected
  value is UNQUANTIFIED, never back-filled with the claim
- no verdict    -> realized is UNMARKED, never zero-that-reads-as-judged and
  never the modelled figure
Partial quantification is counted in the output (2 of 4 …, 2 UNMARKED) rather
than summed into a full-looking total. Learning is measured, not asserted: a
rising approval share is reported only alongside the modelled→corrected gap that
shrank behind it, over cohorts split by run_id order. Cost (USD, a K6 upper
bound) and value (NOK) sit side by side and are never divided — no sourced FX
rate exists here, and a ratio would invent one.

Surfaces: standalone CLI (valuereport) and an opt-in --value-report on run.py,
which requires --outbox and is refused BEFORE any spend without one; the report
is written on both run outcomes and never rewrites the run's exit code (a budget
stop stays a budget stop).

Six seams detach-proven RED: honesty boundary (corrected + realized), gap
arithmetic, projection purity, run-seam wiring, pre-spend fail-fast, both-outcome
reporting. Fixtures are COMMITTED and generated with the real primitives, so the
id-join under test is genuine.

Note on the purity test: it was green-but-dead in its first form. Snapshotting
the committed fixture tree in place let an earlier test's stray write pre-seed
the 'before' snapshot, so the detached write reproduced it byte-for-byte. Every
test now projects from a per-test copy, and a pinned file-set test guards the
committed tree. Found by running the detach proof — which is what §11 is for.

portfolio.py is deliberately NOT wired: run_portfolio persists nothing, so there
is nothing for a projection to read. Its docstring now says that instead of
promising the wiring it did not get.

562 -> 584 tests green; ruff + mypy --strict clean over 27 src files.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MQu2xxwedckjU56byu1aUG
This commit is contained in:
Kjell Tore Guttormsen 2026-07-25 06:25:02 +02:00
commit 4dcdd8017a
18 changed files with 1477 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/portfolio_optimiser_claude/portfolio.py
View file

@ -17,9 +17,17 @@ project still composes its OWN context inside the loop, never a hoisted, shared
Failure policy is a STACK-LOCAL choice until D-D: the default RAISES (today
everything is thrown — a budget stop or any run error propagates and the portfolio
stops). K18 flips this to a collect-and-continue wave model when the D-D fasit
lands. Goals/ledger are out of scope here beyond being accepted as optional
arguments in a later session (K11); this module wires the run path and the
portfolio learning store, nothing more.
lands. This module wires the run path and the portfolio learning store, nothing
more.
Goals/ledger are still NOT wired here, and K11 (the value report) deliberately
did not change that: the report is a projection over PERSISTED state, and this
path persists nothing — it returns typed results and leaves filing to the caller.
Wiring ``--value-report`` here would first require portfolio-level outbox
persistence, which is its own decision (the outbox names pairs by ``run_id``, and
a portfolio pass has no run id of its own). Until that lands, the report is
produced per run (``run.py --value-report``) or standalone over an outbox that
already holds the pairs.
"""
from __future__ import annotations

64
src/portfolio_optimiser_claude/run.py
View file

@ -58,6 +58,7 @@ from portfolio_optimiser_claude.notify import (
notify_config_from_args,
)
from portfolio_optimiser_claude.validator import Rejection
from portfolio_optimiser_claude.valuereport import build_value_report, report_to_json
_PROPOSER_ROLE = "proposer"
_CHECKER_ROLE = "checker"
@ -240,6 +241,35 @@ def execute_run(
return 0
def write_value_report(
*,
outbox_dir: Path,
inbox_dir: Path | None,
ledger_path: Path | None,
destination: Path,
) -> bool:
"""K11 opt-in: project the persisted layers into a value report AFTER the run.
The report is a projection over what is already on disk — including the pair
this run just filed — so it runs after the run, never before, and it can
never change what the run itself decided. A malformed layer is reported here
and makes the COMMAND non-zero (the operator asked for a report and did not
get one), but a budget stop stays a budget stop: reporting never rewrites a
run's own verdict. Returns whether the report was written.
"""
try:
report = build_value_report(
outbox_dir=outbox_dir, inbox_dir=inbox_dir, ledger_path=ledger_path
)
except (OSError, TypeError, ValueError) as exc:
print(f"VALUE REPORT FAILED — refusing to project a malformed layer (§10): {exc}")
return False
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_text(report_to_json(report), encoding="utf-8", newline="\n")
print(f"value-report: {destination}")
return True
def build_dry_run_config(
contracts: Contracts,
*,
@ -357,6 +387,19 @@ def main(
parser.add_argument("--max-debate-rounds", type=int, default=3)
parser.add_argument("--max-attempts", type=int, default=3)
parser.add_argument("--top-k", type=int, default=3)
parser.add_argument(
"--value-report",
type=Path,
default=None,
help="after the run, project the outbox (+ --inbox, + --ledger) into a "
"deterministic value report written as JSON to this path (K11; no model call).",
)
parser.add_argument(
"--ledger",
type=Path,
default=None,
help="realized-savings ledger read by --value-report; absent = an empty book.",
)
parser.add_argument(
"--live-dry-run",
action="store_true",
@ -378,6 +421,11 @@ def main(
)
elif args.outbox is not None and not (args.run_id or "").strip():
parser.error("--outbox requires --run-id (no wall-clock default)")
# K11: the value report PROJECTS the outbox — without one there is nothing to
# project. Rejected here, before any spend, so a run never completes only to
# find it cannot produce the report the operator asked for.
if args.value_report is not None and args.outbox is None:
parser.error("--value-report requires --outbox (the report projects the outbox pairs)")
# K10 (§8): build the notify sinks BEFORE any spend — a --notify-webhook
# without --allow-webhook-egress is refused fail-fast here, so no run rides
@ -442,7 +490,7 @@ def main(
refusals=refusals,
)
return execute_run(
code = execute_run(
client,
composed,
contracts=contracts,
@ -453,6 +501,20 @@ def main(
run_id=args.run_id,
notifiers=notifiers,
)
# K11: the value report is produced on BOTH outcomes — a budget-stopped run
# still has a value picture worth reporting (the accumulated outbox, inbox
# and ledger are what it projects, not this one run's success).
if args.value_report is not None:
assert args.outbox is not None # narrowed by the fail-fast above
written = write_value_report(
outbox_dir=args.outbox,
inbox_dir=args.inbox,
ledger_path=args.ledger,
destination=args.value_report,
)
if not written and code == 0:
code = 1
return code
if __name__ == "__main__":

615
src/portfolio_optimiser_claude/valuereport.py Normal file
View file

@ -0,0 +1,615 @@
"""Per-run value report (method-spec §1/§4/§5 analog; S5.4; paritetsrad 25; K11).
What did the loop actually deliver? This module answers that as a PURE
PROJECTION over state that already exists — the K5 outbox pairs, the §4.2 inbox
verdicts, and the K1 ledger — with no model call, no clock, and no new state.
Three stages, each measured from its OWN layer and never copied from the one
before it:
* **modelled** — what the system claimed (``claimed_saving_nok`` in the outbox
proposal). The cheapest figure to produce and the easiest to over-trust.
* **expert-corrected** — what the expert's §4.2 verdict makes of that claim:
``approved`` leaves it standing, ``rejected`` writes it to zero. The third
vocabulary member, ``approved_with_adjustment``, says the amount changes but
the §4.2 shape carries NO adjusted figure — so the corrected value is
UNQUANTIFIED, reported as such and never silently back-filled with the claim.
* **realized** — what passed the expert gate into the K1 book. A project with no
ledger entry and no settled verdict is UNMARKED (``None``), never zero-that-
looks-judged and never the modelled figure (§1: the report may not claim more
than the layers carry).
The learning effect is quantified rather than asserted: settled proposals are
split by ``run_id`` order into an earlier and a later cohort, and BOTH the
approval share and the modelled→corrected gap are compared across them. A rising
approval share alone is not evidence of learning — the report pairs it with the
gap that shrank, or it reports neither.
Cost against value is reported side by side WITHOUT a ratio: the cost estimate is
USD (K6, itself an upper bound) and the realized value is NOK, and this repo has
no sourced exchange rate. Dividing them would manufacture a number no source
backs (§1).
Run: uv run python -m portfolio_optimiser_claude.valuereport --outbox <dir> \\
--inbox <dir> [--ledger <file>] [--goal-nok N] \\
[--estimated-cost-usd X] [--json <file>]
"""
from __future__ import annotations
import argparse
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Sequence
from pydantic import ValidationError
from portfolio_optimiser_claude.goals import GoalContract
from portfolio_optimiser_claude.hitl import load_outbox_proposals
from portfolio_optimiser_claude.inbox import load_inbox
from portfolio_optimiser_claude.ledger import SavingsLedger
# §4.2 decisions that keep the measure alive (the expert did not throw it out).
_ACCEPTING = frozenset({"approved", "approved_with_adjustment"})
# The decision that voids the claim outright — a QUANTIFIED correction to zero.
_REJECTED = "rejected"
# The decision that changes the amount without carrying one (§4.2 has no field
# for an adjusted figure) — quantifying it here would be invention, not reading.
_ADJUSTED = "approved_with_adjustment"
_UNMARKED = "UNMARKED"
_SHARE_DIGITS = 6 # shares are rounded so the JSON bytes never carry float noise
@dataclass(frozen=True)
class ProposalValue:
"""One outbox proposal projected through the expert stage.
``expert_corrected_nok`` is ``None`` for exactly two reasons, both honest:
no verdict has arrived, or the verdict adjusted the amount without stating
it. ``status`` names which.
"""
run_id: str
verdict_id: str
project_id: str
measure: str
modelled_nok: float
decision: str | None
expert_corrected_nok: float | None
status: str
@dataclass(frozen=True)
class ProjectValue:
"""One project's three stages, plus how much of it is quantified at all."""
project_id: str
modelled_nok: float
expert_corrected_nok: float | None
realized_nok: float | None
quantified_proposals: int
unquantified_proposals: int
proposals: list[ProposalValue]
@dataclass(frozen=True)
class Cohort:
"""One half of the run history: how it was judged, and how big its gap was.
``gap_share`` is computed over the QUANTIFIED proposals only — an
unquantified adjustment cannot contribute to a gap measurement — and
``quantified`` says how many that was, so a thin cohort is visible.
"""
settled: int
accepted: int
approval_share: float | None
quantified: int
modelled_nok: float
expert_corrected_nok: float | None
gap_share: float | None
@dataclass(frozen=True)
class LearningEffect:
"""The quantified trend across cohorts — approval share AND the shrinking gap.
``comparable`` is False whenever there are fewer than two settled proposals
to split: one judgment is a data point, not a trend, and the report says so
rather than rendering a confident-looking zero.
"""
comparable: bool
earlier: Cohort | None
later: Cohort | None
earlier_approval_share: float | None
later_approval_share: float | None
approval_share_delta: float | None
earlier_gap_share: float | None
later_gap_share: float | None
gap_share_delta: float | None
gap_shrinking: bool | None
@dataclass(frozen=True)
class GoalProgress:
"""Progress toward the §8-adjacent savings goal, measured on REALIZED value only."""
target_nok: float
realized_nok: float
share: float
reached: bool
@dataclass(frozen=True)
class CostVersusValue:
"""Run cost against realized value — two currencies, deliberately NOT divided."""
estimated_cost_usd: float
realized_value_nok: float
note: str
@dataclass(frozen=True)
class ValueReport:
"""The whole projection: portfolio roll-up, per project, trend, goal, cost."""
modelled_nok: float
expert_corrected_nok: float | None
realized_nok: float
gap_nok: float
gap_share: float | None
n_proposals: int
n_settled: int
n_pending: int
quantified_proposals: int
unquantified_proposals: int
approval_share: float | None
projects: list[ProjectValue]
learning: LearningEffect
goal: GoalProgress | None
cost: CostVersusValue | None
def _share(numerator: float, denominator: float) -> float | None:
"""A share, or ``None`` when there is nothing to divide by (never a fake zero)."""
if denominator == 0:
return None
return round(numerator / denominator, _SHARE_DIGITS)
def _corrected(decision: str | None, modelled_nok: float) -> tuple[float | None, str]:
"""Project one verdict onto the claim — the §1 honesty boundary, in one place.
Detaching this (returning ``modelled_nok`` for the pending or adjusted case)
is precisely the flattering lie the load-bearing tests exist to catch.
"""
if decision is None:
return None, "pending"
if decision == _REJECTED:
return 0.0, "rejected"
if decision == _ADJUSTED:
return None, "adjusted_unquantified"
return modelled_nok, "approved"
def _cohort(proposals: Sequence[ProposalValue]) -> Cohort:
"""Roll one cohort up: approval share over settled, gap share over quantified."""
settled = [p for p in proposals if p.decision is not None]
accepted = [p for p in settled if p.decision in _ACCEPTING]
quantified = [p for p in settled if p.expert_corrected_nok is not None]
modelled = sum(p.modelled_nok for p in quantified)
corrected = sum(p.expert_corrected_nok or 0.0 for p in quantified) if quantified else None
return Cohort(
settled=len(settled),
accepted=len(accepted),
approval_share=_share(len(accepted), len(settled)),
quantified=len(quantified),
modelled_nok=modelled,
expert_corrected_nok=corrected,
gap_share=None if corrected is None else _share(modelled - corrected, modelled),
)
def _learning_effect(proposals: Sequence[ProposalValue]) -> LearningEffect:
"""Split the settled history in two by ``run_id`` order and compare the halves.
The split point is the midpoint of the SETTLED proposals (pending ones carry
no judgment to compare). Fewer than two settled → not comparable.
"""
settled = sorted((p for p in proposals if p.decision is not None), key=lambda p: p.run_id)
if len(settled) < 2:
return LearningEffect(
comparable=False,
earlier=None,
later=None,
earlier_approval_share=None,
later_approval_share=None,
approval_share_delta=None,
earlier_gap_share=None,
later_gap_share=None,
gap_share_delta=None,
gap_shrinking=None,
)
midpoint = len(settled) // 2
earlier, later = _cohort(settled[:midpoint]), _cohort(settled[midpoint:])
approval_delta = (
None
if earlier.approval_share is None or later.approval_share is None
else round(later.approval_share - earlier.approval_share, _SHARE_DIGITS)
)
# The gap arithmetic is load-bearing: without it a rising approval share
# would be reported as "learning" with nothing measured behind it.
gap_delta = (
None
if earlier.gap_share is None or later.gap_share is None
else round(later.gap_share - earlier.gap_share, _SHARE_DIGITS)
)
return LearningEffect(
comparable=True,
earlier=earlier,
later=later,
earlier_approval_share=earlier.approval_share,
later_approval_share=later.approval_share,
approval_share_delta=approval_delta,
earlier_gap_share=earlier.gap_share,
later_gap_share=later.gap_share,
gap_share_delta=gap_delta,
gap_shrinking=None if gap_delta is None else gap_delta < 0,
)
def _project_value(
project_id: str, proposals: Sequence[ProposalValue], realized_by_project: dict[str, float]
) -> ProjectValue:
"""Roll one project up — realized stays UNMARKED until something passed the gate.
A project earns a realized ZERO only when every one of its proposals came
back rejected: that is a judged outcome. Anything else without a ledger entry
is ``None`` — unjudged is not the same as worth nothing, and neither is the
same as the modelled claim.
"""
quantified = [p for p in proposals if p.expert_corrected_nok is not None]
corrected = sum(p.expert_corrected_nok or 0.0 for p in quantified) if quantified else None
if project_id in realized_by_project:
realized: float | None = realized_by_project[project_id]
elif proposals and all(p.decision == _REJECTED for p in proposals):
realized = 0.0 # judged, and judged worthless — an earned zero
else:
realized = None
return ProjectValue(
project_id=project_id,
modelled_nok=sum(p.modelled_nok for p in proposals),
expert_corrected_nok=corrected,
realized_nok=realized,
quantified_proposals=len(quantified),
unquantified_proposals=len(proposals) - len(quantified),
proposals=list(proposals),
)
def load_ledger(ledger_path: Path | None) -> SavingsLedger:
"""Load the book fail-fast (§10) — a wrong-SHAPE ledger never masquerades as empty.
``None`` means no book was supplied (an empty one). A path that exists but
holds valid JSON of the wrong shape (an array, a string, ``{"entries": {}}``)
raises: reading it as an empty book would silently report every realized
saving as unmarked. ``SavingsLedger.load`` unpacks the payload, so a non-object
top level surfaces as ``TypeError`` — normalized here to ``ValueError`` so the
caller has ONE failure type to catch.
"""
if ledger_path is None:
return SavingsLedger()
try:
return SavingsLedger.load(ledger_path)
except TypeError as exc: # non-object JSON: `**` needs a mapping
raise ValueError(f"ledger {ledger_path} is not a JSON object: {exc}") from exc
def build_value_report(
*,
outbox_dir: Path,
inbox_dir: Path | None = None,
ledger_path: Path | None = None,
goal: GoalContract | None = None,
estimated_cost_usd: float | None = None,
) -> ValueReport:
"""Project the three layers into one deterministic report — reads only, writes nothing.
The id-join is K5's: each outbox pair carries the ``verdict_id`` minted over
the candidate features, and an inbox verdict for that id is the expert stage
for that proposal. Ordering everywhere is ``run_id`` / ``project_id`` sorted,
so the same inputs always yield the same report (no clock, no set iteration).
``inbox_dir`` and ``ledger_path`` are optional because a run may legitimately
have neither yet — the result is a report where every expert-stage figure is
UNMARKED, which is the honest picture of a loop whose experts have not spoken.
"""
decisions = (
{}
if inbox_dir is None
else {document.id: document.decision for document in load_inbox(inbox_dir)}
)
ledger = load_ledger(ledger_path)
realized_by_project: dict[str, float] = {}
for entry in ledger.entries():
realized_by_project[entry.project] = realized_by_project.get(entry.project, 0.0) + (
entry.amount_nok
)
proposals: list[ProposalValue] = []
for outbox_proposal in load_outbox_proposals(outbox_dir):
decision = decisions.get(outbox_proposal.verdict_id)
corrected, status = _corrected(decision, outbox_proposal.claimed_saving_nok)
proposals.append(
ProposalValue(
run_id=outbox_proposal.run_id,
verdict_id=outbox_proposal.verdict_id,
project_id=outbox_proposal.project_id,
measure=outbox_proposal.measure,
modelled_nok=outbox_proposal.claimed_saving_nok,
decision=decision,
expert_corrected_nok=corrected,
status=status,
)
)
by_project: dict[str, list[ProposalValue]] = {}
for proposal in proposals:
by_project.setdefault(proposal.project_id, []).append(proposal)
projects = [
_project_value(project_id, by_project[project_id], realized_by_project)
for project_id in sorted(by_project)
]
modelled = sum(p.modelled_nok for p in proposals)
realized = ledger.total_realized_nok()
quantified = [p for p in proposals if p.expert_corrected_nok is not None]
settled = [p for p in proposals if p.decision is not None]
accepted = [p for p in settled if p.decision in _ACCEPTING]
return ValueReport(
modelled_nok=modelled,
expert_corrected_nok=(
sum(p.expert_corrected_nok or 0.0 for p in quantified) if quantified else None
),
realized_nok=realized,
gap_nok=modelled - realized,
gap_share=_share(modelled - realized, modelled),
n_proposals=len(proposals),
n_settled=len(settled),
n_pending=len(proposals) - len(settled),
quantified_proposals=len(quantified),
unquantified_proposals=len(proposals) - len(quantified),
approval_share=_share(len(accepted), len(settled)),
projects=projects,
learning=_learning_effect(proposals),
goal=(
None
if goal is None
else GoalProgress(
target_nok=goal.target_nok,
realized_nok=realized,
share=round(realized / goal.target_nok, _SHARE_DIGITS),
reached=realized >= goal.target_nok,
)
),
cost=(
None
if estimated_cost_usd is None
else CostVersusValue(
estimated_cost_usd=estimated_cost_usd,
realized_value_nok=realized,
note=(
"ESTIMAT (K6 upper bound) in USD against realized value in NOK — "
"reported side by side, NOT divided: this repo carries no sourced "
"NOK/USD rate, and a ratio would invent one (§1)."
),
)
),
)
def _cohort_payload(cohort: Cohort | None) -> dict[str, Any] | None:
if cohort is None:
return None
return {
"accepted": cohort.accepted,
"approval_share": cohort.approval_share,
"expert_corrected_nok": cohort.expert_corrected_nok,
"gap_share": cohort.gap_share,
"modelled_nok": cohort.modelled_nok,
"quantified": cohort.quantified,
"settled": cohort.settled,
}
def report_payload(report: ValueReport) -> dict[str, Any]:
"""The report as plain data — the one place the JSON shape is defined."""
return {
"approval_share": report.approval_share,
"cost": (
None
if report.cost is None
else {
"estimated_cost_usd": report.cost.estimated_cost_usd,
"note": report.cost.note,
"realized_value_nok": report.cost.realized_value_nok,
}
),
"expert_corrected_nok": report.expert_corrected_nok,
"gap_nok": report.gap_nok,
"gap_share": report.gap_share,
"goal": (
None
if report.goal is None
else {
"reached": report.goal.reached,
"realized_nok": report.goal.realized_nok,
"share": report.goal.share,
"target_nok": report.goal.target_nok,
}
),
"learning": {
"approval_share_delta": report.learning.approval_share_delta,
"comparable": report.learning.comparable,
"earlier": _cohort_payload(report.learning.earlier),
"earlier_approval_share": report.learning.earlier_approval_share,
"earlier_gap_share": report.learning.earlier_gap_share,
"gap_share_delta": report.learning.gap_share_delta,
"gap_shrinking": report.learning.gap_shrinking,
"later": _cohort_payload(report.learning.later),
"later_approval_share": report.learning.later_approval_share,
"later_gap_share": report.learning.later_gap_share,
},
"modelled_nok": report.modelled_nok,
"n_pending": report.n_pending,
"n_proposals": report.n_proposals,
"n_settled": report.n_settled,
"projects": [
{
"expert_corrected_nok": project.expert_corrected_nok,
"modelled_nok": project.modelled_nok,
"project_id": project.project_id,
"proposals": [
{
"decision": proposal.decision,
"expert_corrected_nok": proposal.expert_corrected_nok,
"measure": proposal.measure,
"modelled_nok": proposal.modelled_nok,
"run_id": proposal.run_id,
"status": proposal.status,
"verdict_id": proposal.verdict_id,
}
for proposal in project.proposals
],
"quantified_proposals": project.quantified_proposals,
"realized_nok": project.realized_nok,
"unquantified_proposals": project.unquantified_proposals,
}
for project in report.projects
],
"quantified_proposals": report.quantified_proposals,
"realized_nok": report.realized_nok,
"unquantified_proposals": report.unquantified_proposals,
}
def report_to_json(report: ValueReport) -> str:
"""Deterministic house JSON: sorted keys, 2-space indent, trailing LF."""
return json.dumps(report_payload(report), sort_keys=True, indent=2, ensure_ascii=False) + "\n"
def _nok(value: float | None) -> str:
"""Render a figure, or the UNMARKED label — never a blank that reads as zero."""
return _UNMARKED if value is None else f"{value:,.0f} NOK".replace(",", " ")
def _pct(share: float | None) -> str:
return _UNMARKED if share is None else f"{share * 100:.1f}%"
def render_report(report: ValueReport) -> str:
"""Render the report as text — every missing figure is LABELLED, never blank."""
lines = [
"VERDIRAPPORT — modellert → ekspert-korrigert → realisert",
(
f" modellert {_nok(report.modelled_nok)}\n"
f" ekspert-korrigert {_nok(report.expert_corrected_nok)} "
f"({report.quantified_proposals} of {report.n_proposals} proposals quantified, "
f"{report.unquantified_proposals} {_UNMARKED})\n"
f" realisert {_nok(report.realized_nok)} "
f"(gap vs modellert: {_nok(report.gap_nok)}, {_pct(report.gap_share)})"
),
(
f" {report.n_proposals} proposal(s): {report.n_settled} settled, "
f"{report.n_pending} awaiting a verdict, approval share "
f"{_pct(report.approval_share)}"
),
"",
"PER PROSJEKT",
]
for project in report.projects:
lines.append(
f" {project.project_id:<12} modellert {_nok(project.modelled_nok):>15} "
f"korrigert {_nok(project.expert_corrected_nok):>15} "
f"realisert {_nok(project.realized_nok):>15}"
)
lines.append("")
lines.append("LÆRINGSEFFEKT (tidligere → senere kjøringer, delt på run_id-rekkefølge)")
learning = report.learning
if not learning.comparable:
lines.append(
" not comparable — fewer than two settled proposals; one judgment is a "
"data point, not a trend"
)
else:
lines.append(
f" godkjenningsandel {_pct(learning.earlier_approval_share)} → "
f"{_pct(learning.later_approval_share)} "
f"(delta {_pct(learning.approval_share_delta)})"
)
lines.append(
f" gap modellert→korrigert {_pct(learning.earlier_gap_share)} → "
f"{_pct(learning.later_gap_share)} "
f"(delta {_pct(learning.gap_share_delta)}, "
f"shrinking={learning.gap_shrinking})"
)
if report.goal is not None:
lines.append("")
lines.append(
f"MÅLPROGRESJON {_nok(report.goal.realized_nok)} of "
f"{_nok(report.goal.target_nok)} ({_pct(report.goal.share)}), "
f"reached={report.goal.reached}"
)
if report.cost is not None:
lines.append("")
lines.append(
f"KOST MOT VERDI ~${report.cost.estimated_cost_usd:.6f} (ESTIMAT, K6 upper bound) "
f"vs realisert {_nok(report.cost.realized_value_nok)}"
)
lines.append(f" {report.cost.note}")
return "\n".join(lines)
def main(argv: list[str] | None = None) -> int:
"""The thin CLI: load the three layers fail-fast (§10) → project → print → optionally write."""
parser = argparse.ArgumentParser(
description=(
"Project the outbox, the inbox and the ledger into one deterministic "
"value report — modelled → expert-corrected → realized, goal progress, "
"learning effect, cost against value. Reads only; no model call."
)
)
parser.add_argument("--outbox", required=True, help="outbox dir (run_id-named pairs, K5)")
parser.add_argument("--inbox", required=True, help="inbox dir (expert verdict files, §4.2)")
parser.add_argument("--ledger", default=None, help="ledger file (K1); absent = empty book")
parser.add_argument("--goal-nok", type=float, default=None, help="absolute savings target")
parser.add_argument(
"--estimated-cost-usd", type=float, default=None, help="K6 upper-bound run cost estimate"
)
parser.add_argument("--json", default=None, help="also write the report as JSON to this path")
args = parser.parse_args(argv)
try:
report = build_value_report(
outbox_dir=Path(args.outbox),
inbox_dir=Path(args.inbox),
ledger_path=None if args.ledger is None else Path(args.ledger),
goal=None
if args.goal_nok is None
else GoalContract(target_nok=args.goal_nok, mode="soft"),
estimated_cost_usd=args.estimated_cost_usd,
)
except (OSError, ValueError, ValidationError) as exc:
print(f"VALUE REPORT FAILED — refusing to project a malformed layer (§10): {exc}")
return 1
print(render_report(report))
if args.json is not None:
Path(args.json).write_text(report_to_json(report), encoding="utf-8", newline="\n")
return 0
if __name__ == "__main__":
raise SystemExit(main())