feat(portfolio): K6 — pre-run cost simulation, priced what-if (parity row 18) [skip-docs]

Before ANY spend the operator sees a deterministic UPPER-BOUND USD estimate
for a (portfolio-)run — a what-if over the models in model_map.json (Claude
models) × effort levels (S3.6-analog, D-I pkt. 3 MUST-krav). No network, no
model call, no key: pure config arithmetic (bound by an import-purity test,
mirroring okf.py).

- contracts.py: ModelPriceContract (usd_per_mtok > 0 + REQUIRED source +
  source_date so a stale rate is visible, never silent, §1) + PricingContract
  (non-empty; no hardcoded fallback rate) + load_pricing/_bundled_pricing.
- data/pricing.example.json: per-Mtok rate per model id, each with source+date.
  Example rates are Anthropic's OUTPUT price (the higher rate) so the whole cap
  billed at that single rate can only overstate — the figure is marked ESTIMAT.
  Covers the model model_map configures, so the default path runs green.
- costsim.py: estimate_costs (n_projects × cap × effort_factor tokens at the
  per-Mtok rate; a model with no price fails fast "missing price for <id>",
  never a guess) + render_estimate + `python -m …costsim`. Effort factors are
  a coarse modeling weight (not prices) — max effort = full cap = the true
  upper bound. No price literal anywhere (grep-guard proves it).
- tests/test_costsim.py: schema fail-fast, missing-price fail-fast, scales with
  model × effort + reproducible, grep-guard, import purity, bundled-example +
  CLI offline smoke. Three seams detach-proven RED (effort factor, price guard,
  price literal).

462→478 green, golden byte-exact, full gate clean (ruff+format+mypy strict,
23 src files), run_s10.py/runs/ byte-untouched. README test-count sync ×2 +
costsim.py module note. CLI run-total-cap wiring stays out of scope (planen
lists 4 files); the mechanism is complete and proven load-bearing.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RiTwaKLesgcwXx2mDviqpt
This commit is contained in:
Kjell Tore Guttormsen 2026-07-23 23:01:01 +02:00
commit 1600c188b4
5 changed files with 495 additions and 2 deletions

View file

@ -13,7 +13,7 @@ human-in-the-loop, and the system learns from the verdicts.
> **Status:** the D7 build (S5S10) is complete, and the deterministic **ingest layer**
> (CSV and SQL source types) has since been added in front of the loop. The deterministic
> backbone, the agentic loop, the learning loop, and the ingest connectors are wired seam by
> seam, each proven by load-bearing tests (462 tests, all running offline without an API
> seam, each proven by load-bearing tests (478 tests, all running offline without an API
> key). The programme's single budgeted **live model run has been executed and validated**
> its artifacts are committed under [`runs/s10/`](runs/s10/) (see below).
@ -110,6 +110,14 @@ description, never from its code)
portfolio-level expert inbox the system reads before each fold. The default failure policy
raises (a stack-local choice until D-D flips it to collect-and-continue).
- `run_s10.py` — the programme's ONE live run (cost discipline D6); run-path only.
- `costsim.py` — pre-run cost simulation (**offline** — the one Run-layer module that never
touches the network): a deterministic UPPER-BOUND USD estimate for a (portfolio-)run
*before* any spend, a what-if over the models in `model_map.json` × effort levels. Pricing
is schema-validated config (`data/pricing.example.json`): a per-Mtok rate per model, each
with a required source + date so a stale rate is visible, never silent. A model configured
with no price fails fast — there is no hardcoded rate anywhere (a grep-guard proves it), and
the figure is marked `ESTIMAT` (the whole cap billed at the rate is an upper bound; real runs
cost less). `uv run python -m portfolio_optimiser_claude.costsim`.
### Load-bearing tests (§11)
@ -188,7 +196,7 @@ Python ≥3.10 · [`claude-agent-sdk`](https://pypi.org/project/claude-agent-sdk
```bash
uv sync # install dependencies
uv run pytest # 462 tests — run without any API key and without network
uv run pytest # 478 tests — run without any API key and without network
uv run ruff check . && uv run ruff format --check .
uv run mypy src # strict
```

View file

@ -29,6 +29,7 @@ from pydantic import BaseModel, Field, model_validator
_MODEL_MAP_RESOURCE = "data/model_map.json"
_REFERENCE_PROJECTS_RESOURCE = "data/reference_projects.json"
_PRICING_RESOURCE = "data/pricing.example.json"
# C2.6: an empty-string model id is a startup schema error, never a client-layer one.
_ModelId = Annotated[str, Field(min_length=1)]
@ -87,6 +88,32 @@ class ReferenceProjectsContract(BaseModel):
projects: list[ReferenceProjectContract] = Field(min_length=1)
class ModelPriceContract(BaseModel):
"""One model's per-Mtok USD rate (K6 cost sim) — provenance is REQUIRED.
``usd_per_mtok`` must be positive (a non-positive rate is a startup schema
error, never a guessed price). ``source`` + ``source_date`` are mandatory so
a stale rate is always VISIBLE (§1 honesty) a price with no provenance
would let a silently-outdated figure through, exactly what the ESTIMAT label
must never hide.
"""
usd_per_mtok: float = Field(gt=0)
source: str = Field(min_length=1)
source_date: str = Field(min_length=1)
class PricingContract(BaseModel):
"""Model-id -> per-Mtok price (validates data/pricing.example.json, K6).
Non-empty by construction (an empty price book is a startup error). There is
NO hardcoded fallback rate anywhere on the cost-sim path a missing price is
fail-fast, never a guess (method-spec §10, D-I pkt. 3).
"""
prices: dict[str, ModelPriceContract] = Field(min_length=1)
class Contracts(BaseModel):
"""The validated bundle of all startup contracts."""
@ -127,6 +154,25 @@ def _bundled_reference_projects() -> dict[str, Any]:
return raw
def _bundled_pricing() -> dict[str, Any]:
raw: dict[str, Any] = json.loads(
files("portfolio_optimiser_claude").joinpath(_PRICING_RESOURCE).read_text(encoding="utf-8")
)
return raw
def load_pricing(raw: dict[str, Any] | None = None) -> PricingContract:
"""Validate the cost-sim pricing config (K6, fail-fast before any estimate).
Raises ``pydantic.ValidationError`` on the first malformed/missing price
(§10). ``raw`` defaults to the bundled ``data/pricing.example.json`` EXAMPLE,
whose SHAPE is validated (not its rates those carry source+date so a stale
figure is visible, never silent, §1). The suite never spends against it.
"""
data = _bundled_pricing() if raw is None else raw
return PricingContract(**data)
def load_reference_projects(raw: dict[str, Any] | None = None) -> ReferenceProjectsContract:
"""Validate the portfolio config at startup (§10, fail-fast before any run).

View file

@ -0,0 +1,195 @@
"""Pre-run cost simulation (method-spec §8/§10 analog; S3.6; paritetsrad 18; K6).
Before ANY spend, the operator can see an ESTIMATED upper-bound USD cost for a
(portfolio-)run a what-if over the models configured in ``model_map.json``
(Claude models) × effort levels. Pricing is schema-validated config
(``data/pricing.example.json``): a per-Mtok USD rate per model id, each carrying
a REQUIRED source + date so a stale price is visible, never silent (§1 honesty).
A model configured in ``model_map`` with no price fails FAST ("missing price for
<id>") — the estimate is never guessed from a hardcoded rate (there is no price
literal anywhere in this module; the grep-guard test proves it).
The estimate is a deterministic UPPER BOUND, not a forecast: it bills the whole
token cap (portfolio shape × per-project cap × an effort weight) at the model's
per-Mtok rate. The example rates are Anthropic's OUTPUT price (the higher of the
two published rates), so billing the whole cap at that single rate can only
overstate, never understate the figure is marked ESTIMAT in the output. Real
runs cost less (input is cheaper and often cached, output is small). No network,
no model call, no key: pure config arithmetic (offline invariant; bound by the
import-purity test, mirroring ``okf.py``).
Run: uv run python -m portfolio_optimiser_claude.costsim [--projects N]
[--token-cap T] [--pricing FILE]
"""
from __future__ import annotations
import argparse
import json
from dataclasses import dataclass
from pathlib import Path
from portfolio_optimiser_claude.contracts import (
ModelMapContract,
PricingContract,
_bundled_model_map,
load_pricing,
load_reference_projects,
)
# Coarse deterministic weighting of expected token consumption as a FRACTION of
# the cap, per Claude effort level (low -> max). NOT measured: a modeling weight
# for the ESTIMATE, where ``max`` effort is modeled as consuming the full cap
# (the true upper bound) and lower efforts proportionally less. These are effort
# weights, not prices — the grep-guard forbids only price literals here.
_EFFORT_FACTORS: dict[str, float] = {
"low": 0.2,
"medium": 0.4,
"high": 0.6,
"xhigh": 0.8,
"max": 1.0,
}
_DEFAULT_EFFORTS: tuple[str, ...] = ("low", "medium", "high", "xhigh", "max")
_TOKENS_PER_MTOK = 1_000_000
_DEFAULT_TOKEN_CAP = 150_000 # mirrors run.py's --max-tokens default (the §8 cap)
@dataclass(frozen=True)
class EffortEstimate:
"""One (model, effort) cell of the what-if grid — tokens + upper-bound USD."""
effort: str
estimated_tokens: int
cost_usd: float
@dataclass(frozen=True)
class ModelEstimate:
"""One model's row: its sourced rate + the per-effort upper-bound estimates."""
model_id: str
usd_per_mtok: float
source: str
source_date: str
efforts: list[EffortEstimate]
@dataclass(frozen=True)
class CostEstimate:
"""The full deterministic what-if: portfolio shape × cap × (model × effort)."""
n_projects: int
token_cap_per_project: int
models: list[ModelEstimate]
def _configured_model_ids(model_map: ModelMapContract) -> list[str]:
"""The DISTINCT model ids configured anywhere in the map, sorted (determinism)."""
ids: set[str] = set()
for mapping in model_map.profiles.values():
ids.update(mapping.values())
return sorted(ids)
def estimate_costs(
model_map: ModelMapContract,
pricing: PricingContract,
*,
n_projects: int,
token_cap_per_project: int,
efforts: tuple[str, ...] = _DEFAULT_EFFORTS,
) -> CostEstimate:
"""Deterministic upper-bound cost estimate over (model_map models × efforts).
Every configured model REQUIRES a price a missing one raises
``ValueError("missing price for <id>")`` (fail-fast, never a guessed rate).
The per-cell figure is ``n_projects × token_cap_per_project × effort_factor``
tokens billed at the model's per-Mtok rate — an UPPER BOUND (the whole cap at
the sourced rate), reproducible from the inputs alone (no clock, no random).
"""
if n_projects <= 0:
raise ValueError(f"n_projects must be positive, got {n_projects}")
if token_cap_per_project <= 0:
raise ValueError(f"token_cap_per_project must be positive, got {token_cap_per_project}")
models: list[ModelEstimate] = []
for model_id in _configured_model_ids(model_map):
price = pricing.prices.get(model_id)
if price is None:
raise ValueError(f"missing price for {model_id}")
cells: list[EffortEstimate] = []
for effort in efforts:
factor = _EFFORT_FACTORS.get(effort)
if factor is None:
raise ValueError(f"unknown effort level {effort!r}")
estimated_tokens = int(n_projects * token_cap_per_project * factor)
cost = round(estimated_tokens * price.usd_per_mtok / _TOKENS_PER_MTOK, 6)
cells.append(
EffortEstimate(effort=effort, estimated_tokens=estimated_tokens, cost_usd=cost)
)
models.append(
ModelEstimate(
model_id=model_id,
usd_per_mtok=price.usd_per_mtok,
source=price.source,
source_date=price.source_date,
efforts=cells,
)
)
return CostEstimate(
n_projects=n_projects, token_cap_per_project=token_cap_per_project, models=models
)
def render_estimate(estimate: CostEstimate) -> str:
"""Render the what-if as an ESTIMAT-marked table (the honesty label is load-bearing)."""
lines = [
f"ESTIMAT (deterministic upper bound) — cost for a run of "
f"{estimate.n_projects} project(s), token cap "
f"{estimate.token_cap_per_project}/project.",
"NB: upper bound — the whole cap is billed at each model's per-Mtok rate; "
"real runs cost less (input is cheaper and often cached, output is small).",
]
for model in estimate.models:
lines.append(
f"model {model.model_id} (${model.usd_per_mtok}/Mtok "
f"source={model.source} {model.source_date})"
)
for cell in model.efforts:
lines.append(
f" {cell.effort:<7} ~{cell.estimated_tokens:>12} tok ~${cell.cost_usd:.6f}"
)
return "\n".join(lines)
def main(argv: list[str] | None = None) -> int:
"""The thin CLI: schema-validate pricing (§10) → estimate (offline) → print."""
parser = argparse.ArgumentParser(
description=(
"Estimate the upper-bound USD cost for a (portfolio-)run BEFORE any "
"spend — a what-if over model_map models x effort levels (offline, ESTIMAT)."
)
)
parser.add_argument("--projects", type=int, default=None)
parser.add_argument("--token-cap", type=int, default=_DEFAULT_TOKEN_CAP)
parser.add_argument("--pricing", type=Path, default=None)
args = parser.parse_args(argv)
# §10: pricing is schema-validated BEFORE any estimate; a bad price fails fast.
pricing = load_pricing(
json.loads(args.pricing.read_text(encoding="utf-8")) if args.pricing is not None else None
)
model_map = ModelMapContract(**_bundled_model_map())
n_projects = (
args.projects
if args.projects is not None
else len(load_reference_projects().projects) # portfolio shape from the config
)
estimate = estimate_costs(
model_map, pricing, n_projects=n_projects, token_cap_per_project=args.token_cap
)
print(render_estimate(estimate))
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,25 @@
{
"_note": "EXAMPLE pricing config for the K6 cost sim (contracts.PricingContract). Schema-validated fail-fast at startup (§10); the suite validates only its SHAPE and never spends against it (offline invariant). usd_per_mtok is the SINGLE per-Mtok rate applied to the whole token cap for a deterministic UPPER-BOUND estimate: each example rate is Anthropic's published OUTPUT price (the higher of the two rates), so billing the entire cap at it can only overstate, never understate — the figure is marked ESTIMAT in the output and real runs cost less (input is cheaper and often cached, output is small). source + source_date are REQUIRED so a stale rate is visible, never silent (honesty rule, method-spec §1); operators re-source against the official page before relying on the numbers. Prices verified against the claude-api reference (cached 2026-06-24) which cites platform.claude.com/docs/en/pricing; NOT independently verified beyond that date.",
"prices": {
"claude-haiku-4-5-20251001": {
"usd_per_mtok": 5.0,
"source": "https://platform.claude.com/docs/en/pricing",
"source_date": "2026-06-24"
},
"claude-sonnet-5": {
"usd_per_mtok": 15.0,
"source": "https://platform.claude.com/docs/en/pricing",
"source_date": "2026-06-24"
},
"claude-opus-4-8": {
"usd_per_mtok": 25.0,
"source": "https://platform.claude.com/docs/en/pricing",
"source_date": "2026-06-24"
},
"claude-fable-5": {
"usd_per_mtok": 50.0,
"source": "https://platform.claude.com/docs/en/pricing",
"source_date": "2026-06-24"
}
}
}

219
tests/test_costsim.py Normal file
View file

@ -0,0 +1,219 @@
"""Pre-run cost simulation (K6, S3.6-analog, D-I pkt. 3; paritetsrad 18).
The operator sees an ESTIMATED upper-bound USD cost for a (portfolio-)run BEFORE
any spend a what-if over the models in model_map.json × effort levels. Every
seam here is load-bearing (§11): the RED-when-detached notes on each test name
the mutation that makes it fail, so a green-but-dead test can't hide.
Nothing here touches the network or a key (offline invariant): the estimate is
pure config arithmetic over schema-validated pricing config.
"""
from __future__ import annotations
import ast
import json
from pathlib import Path
import pytest
from pydantic import ValidationError
from portfolio_optimiser_claude.contracts import (
ModelMapContract,
ModelPriceContract,
PricingContract,
load_pricing,
)
from portfolio_optimiser_claude.costsim import estimate_costs, main
SRC_PKG = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser_claude"
def _model_map(*model_ids: str) -> ModelMapContract:
# Every profile needs a 'default' (§10); role split is irrelevant to the
# what-if, which iterates the DISTINCT model ids configured anywhere.
mapping = {"default": model_ids[0]}
for i, mid in enumerate(model_ids[1:], start=1):
mapping[f"role{i}"] = mid
return ModelMapContract(profiles={"anthropic": mapping})
def _pricing(**rates: float) -> PricingContract:
return PricingContract(
prices={
mid: ModelPriceContract(usd_per_mtok=rate, source="src", source_date="2026-01-01")
for mid, rate in rates.items()
}
)
class TestPricingSchemaFailsFast:
"""§10: a malformed/missing price is a startup schema error, never a guessed rate."""
def test_non_positive_price_is_rejected(self) -> None:
with pytest.raises(ValidationError):
ModelPriceContract(usd_per_mtok=0.0, source="src", source_date="2026-01-01")
def test_a_price_entry_without_a_rate_is_rejected(self) -> None:
with pytest.raises(ValidationError):
PricingContract(prices={"m": {"source": "src", "source_date": "2026-01-01"}})
def test_source_and_date_are_required(self) -> None:
# A price with no provenance would let a stale rate pass silently (§1).
with pytest.raises(ValidationError):
PricingContract(prices={"m": {"usd_per_mtok": 5.0}})
def test_empty_pricing_is_rejected(self) -> None:
with pytest.raises(ValidationError):
PricingContract(prices={})
class TestMissingPriceFailsFast:
"""RED (detach the price lookup guard → KeyError not ValueError): a model in
model_map with NO price stops the estimate cold never a guessed rate."""
def test_a_configured_model_without_a_price_raises_named(self) -> None:
model_map = _model_map("m-cheap", "m-unpriced")
pricing = _pricing(**{"m-cheap": 1.0}) # m-unpriced deliberately absent
with pytest.raises(ValueError, match="missing price for m-unpriced"):
estimate_costs(model_map, pricing, n_projects=1, token_cap_per_project=1000)
class TestEstimateScalesAndIsDeterministic:
"""RED (detach the effort factor → efforts identical): the estimate scales
with BOTH model price and effort, and is byte-reproducible (no clock/random)."""
def test_cost_scales_linearly_with_model_price(self) -> None:
model_map = _model_map("m-cheap", "m-pricey")
pricing = _pricing(**{"m-cheap": 1.0, "m-pricey": 10.0})
est = estimate_costs(model_map, pricing, n_projects=3, token_cap_per_project=1000)
by_id = {m.model_id: m for m in est.models}
# Same portfolio, same effort → 10x price ⇒ 10x cost, per effort level.
for effort in ("low", "high", "max"):
cheap = next(e.cost_usd for e in by_id["m-cheap"].efforts if e.effort == effort)
pricey = next(e.cost_usd for e in by_id["m-pricey"].efforts if e.effort == effort)
assert pricey == pytest.approx(cheap * 10.0)
def test_cost_scales_strictly_with_effort(self) -> None:
# DETACH POINT: make every _EFFORT_FACTORS value equal (e.g. all 1.0) and
# this strict-increase assertion goes RED — the effort axis is dead.
model_map = _model_map("m")
pricing = _pricing(**{"m": 5.0})
est = estimate_costs(model_map, pricing, n_projects=2, token_cap_per_project=1000)
costs = [e.cost_usd for e in est.models[0].efforts]
assert costs == sorted(costs)
assert costs[0] < costs[-1]
def test_estimate_is_reproducible(self) -> None:
model_map = _model_map("m")
pricing = _pricing(**{"m": 7.0})
a = estimate_costs(model_map, pricing, n_projects=4, token_cap_per_project=1234)
b = estimate_costs(model_map, pricing, n_projects=4, token_cap_per_project=1234)
assert [(m.model_id, [(e.effort, e.cost_usd) for e in m.efforts]) for m in a.models] == [
(m.model_id, [(e.effort, e.cost_usd) for e in m.efforts]) for m in b.models
]
def test_non_positive_shape_is_rejected(self) -> None:
pricing = _pricing(**{"m": 5.0})
with pytest.raises(ValueError):
estimate_costs(_model_map("m"), pricing, n_projects=0, token_cap_per_project=1000)
with pytest.raises(ValueError):
estimate_costs(_model_map("m"), pricing, n_projects=1, token_cap_per_project=0)
class TestNoPriceLiteralInSource:
"""RED (hardcode any example rate in costsim.py): the grep-guard proves the
price MUST come from config no rate literal lives in the module source."""
def test_no_example_price_appears_as_a_literal(self) -> None:
source = (SRC_PKG / "costsim.py").read_text(encoding="utf-8")
pricing = load_pricing() # the bundled data/pricing.example.json
for model_id, price in pricing.prices.items():
assert str(price.usd_per_mtok) not in source, (
f"price for {model_id} is hardcoded in costsim.py — must come from config"
)
def _imported_module_names(module_path: Path) -> set[str]:
tree = ast.parse(module_path.read_text(encoding="utf-8"))
names: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
names.update(alias.name.split(".")[0] for alias in node.names)
elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
names.add(node.module.split(".")[0])
return names
class TestCostsimPurity:
"""LOAD-BEARING (§11): the cost sim imports no agent toolkit — offline by design."""
def test_costsim_never_imports_an_agent_toolkit(self) -> None:
names = _imported_module_names(SRC_PKG / "costsim.py")
assert not names & {"claude_agent_sdk", "anthropic"}
class TestBundledExampleAndCli:
"""Nøkkelantakelse: the bundled example prices COVER the model model_map
configures, so the offline default path runs green and the CLI is offline."""
def test_bundled_pricing_validates(self) -> None:
pricing = load_pricing()
assert pricing.prices # non-empty, schema-valid
for price in pricing.prices.values():
assert price.usd_per_mtok > 0
assert price.source and price.source_date
def test_the_default_estimate_runs_on_bundled_config(self) -> None:
# The bundled model_map's model must have a bundled price — else the
# default CLI would fail-fast. This binds the two example files together.
from portfolio_optimiser_claude.contracts import _bundled_model_map
est = estimate_costs(
ModelMapContract(**_bundled_model_map()),
load_pricing(),
n_projects=1,
token_cap_per_project=150_000,
)
assert est.models
assert all(m.efforts for m in est.models)
def test_cli_prints_an_estimate_marked_estimat(
self, capsys: pytest.CaptureFixture[str]
) -> None:
rc = main([])
out = capsys.readouterr().out
assert rc == 0
assert "ESTIMAT" in out
def test_cli_honours_explicit_pricing_file(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
from portfolio_optimiser_claude.contracts import _bundled_model_map
model_id = ModelMapContract(**_bundled_model_map()).profiles["anthropic"]["default"]
pricing_file = tmp_path / "pricing.json"
pricing_file.write_text(
json.dumps(
{
"prices": {
model_id: {
"usd_per_mtok": 3.0,
"source": "src",
"source_date": "2026-01-01",
}
}
}
),
encoding="utf-8",
)
rc = main(["--projects", "2", "--token-cap", "1000", "--pricing", str(pricing_file)])
out = capsys.readouterr().out
assert rc == 0
assert model_id in out
def test_no_network_import_in_costsim() -> None:
# Belt on the offline invariant: not even the stdlib socket/urllib slip in.
names = _imported_module_names(SRC_PKG / "costsim.py")
assert not names & {"socket", "urllib", "http", "requests", "httpx"}