"""S3.6 — kostnadssimulering før kjøring (D-I pkt. 3): offline, MAF-free cost estimator. Estimates token/cost for a portfolio run BEFORE running — a what-if over the model-map (models × effort levels) — so run cost becomes an informed decision, not a surprise afterwards (revisjonspakke §0.5 F-INT-5: kjøringskost er en CFO-beslutning). MAF-free by construction (D7-portable; målbilde context/output-layer rule): this module imports ONLY stdlib + pydantic — never ``agent_framework``/``mcp`` nor the MAF-bound framework modules (``backends``/``budget``/``contracts``/``run``). It reads ``data/model_map.json`` as PLAIN DATA via a ``__file__``-relative path, deliberately NOT ``importlib.resources.files("portfolio_optimiser")`` (which would import the MAF-bound package ``__init__`` and defeat MAF-freedom). Registered in ``tests/test_okf.py``'s ``_MAF_FREE_MODULES`` and enforced by ``test_okf_is_maf_free`` (direct-import AST scan) + ``test_costsim_import_is_maf_free`` (transitive import-graph subprocess seam). All money math is INTEGER øre — float NOK is non-deterministic under summation (see ``ledger.py``); kr is formatted only at the display edge. """ from __future__ import annotations import json import sys from pathlib import Path from typing import Any from pydantic import BaseModel, Field _DATA_DIR = Path(__file__).resolve().parent / "data" _MODEL_MAP_FILE = _DATA_DIR / "model_map.json" _PRICING_FILE = _DATA_DIR / "pricing.example.json" # A model-map deployment id that the operator has not yet replaced (data/model_map.json ships the # azure roles as REPLACE-WITH-FOUNDRY-DEPLOYMENT). Such ids are UNPRICED, not an error. PLACEHOLDER_PREFIX = "REPLACE-WITH-" def _load_model_map( profile: str, model_map: dict[str, dict[str, str]] | None = None ) -> dict[str, str]: """Return the ``role -> model`` sub-map for ``profile`` from ``data/model_map.json``. Reads the packaged JSON as plain data via a ``__file__``-relative path — deliberately NOT ``importlib.resources.files("portfolio_optimiser")``, which would import the MAF-bound package ``__init__`` and defeat the MAF-free guarantee. Tests inject ``model_map`` to avoid touching shipped config. Fail-fast (``ValueError``) when the profile is absent or malformed. """ table = ( model_map if model_map is not None else json.loads(_MODEL_MAP_FILE.read_text(encoding="utf-8")) ) entry = table.get(profile) if not isinstance(entry, dict): raise ValueError(f"unknown profile {profile!r} in model map") return dict(entry) class QualityGuidance(BaseModel): """Per-model quality trade-off as GUIDANCE WITH SOURCE (brief Goal 4 / D-I pkt. 3): prose ``note`` + a provenance ``source`` — never a bare measured number. This keeps a quality claim honest (veiledning med kilde), never asserted as measured fact without belegg (målbilde §1).""" note: str = Field(min_length=1) source: str = Field(min_length=1) class ModelPrice(BaseModel): """A model's blended price in INTEGER øre per 1000 total tokens (float NOK is banned — see ``ledger.py``). ``quality_guidance`` is optional sourced prose, never a number.""" ore_per_1k_tokens: int = Field(ge=0) quality_guidance: QualityGuidance | None = None class PricingContract(BaseModel): """Schema-validated pricing config (fail-fast). ``source`` + ``date`` are REQUIRED provenance (kilde + dato) — the framework ships an example; the deployer owns verified vendor prices.""" source: str = Field(min_length=1) date: str = Field(min_length=1) models: dict[str, ModelPrice] def load_pricing(path: str | Path | None = None) -> PricingContract: """Load + validate pricing config, fail-fast. Missing file → ``FileNotFoundError``; malformed / missing ``source``/``date`` → ``pydantic.ValidationError`` (mirrors ``contracts.load_goal_config``). ``None`` loads the packaged ``data/pricing.example.json``. Top-level ``_``-prefixed keys (e.g. ``_note``) are ignored.""" p = Path(path) if path is not None else _PRICING_FILE if not p.is_file(): raise FileNotFoundError(f"pricing config not found: {str(p)!r}") raw = json.loads(p.read_text(encoding="utf-8")) data = {k: v for k, v in raw.items() if not k.startswith("_")} return PricingContract(**data) def _price_ore_per_1k(model: str, pricing: PricingContract) -> int | None: """Return ``model``'s øre-per-1k-tokens rate. Returns ``None`` for a placeholder deployment id (``REPLACE-WITH-*`` — the defined unpriced path; azure profile ships these). Fail-fast (``ValueError``) for a genuine (non-placeholder) model absent from the price map.""" if model.startswith(PLACEHOLDER_PREFIX): return None entry = pricing.models.get(model) if entry is None: raise ValueError(f"missing price for {model}") return entry.ore_per_1k_tokens # Effort is a costsim-LOCAL what-if scenario knob (integer percent multipliers on the token ceiling). # HONESTY (målbilde §1): the runtime does NOT consume an effort setting today — there is no # `effort`/`ChatOptions` counterpart in the model-map or run path — so this is an explicit scenario # dimension for the what-if, never a measured runtime input. EFFORT_FACTORS: dict[str, int] = {"low": 70, "standard": 100, "high": 140} # The hard per-run token ceiling. A run cannot exceed it (mirrors run.py's default budget cap), so it # IS the honest per-run upper bound — rounds are NOT multiplied in (that double-counts the cap). _DEFAULT_MAX_TOKENS = 100_000 def estimate_run_ore( model: str, effort: str, pricing: PricingContract, max_tokens: int = _DEFAULT_MAX_TOKENS ) -> int | None: """Deterministic INTEGER-øre upper-bound estimate for ONE run of ``model`` at ``effort``. ``max_tokens`` is the hard per-run ceiling → the honest per-run upper bound (rounds are not multiplied in — that double-counts the cap). ``effort`` scales it by a scenario factor (see ``EFFORT_FACTORS``). Returns ``None`` when ``model`` is an unpriced placeholder. Pure integer math (float NOK is banned — ``ledger.py``); marked as an estimate (øvre grense), not a prediction.""" if effort not in EFFORT_FACTORS: raise ValueError(f"unknown effort {effort!r}; known: {sorted(EFFORT_FACTORS)}") rate = _price_ore_per_1k(model, pricing) if rate is None: return None scaled_tokens = max_tokens * EFFORT_FACTORS[effort] // 100 return scaled_tokens * rate // 1000 def estimate_portfolio_ore( model: str, effort: str, pricing: PricingContract, *, n_projects: int, max_tokens: int = _DEFAULT_MAX_TOKENS, ) -> int | None: """Deterministic portfolio upper bound = ``n_projects × estimate_run_ore``. ``None`` if the model is an unpriced placeholder. Fail-fast on a negative project count.""" if n_projects < 0: raise ValueError(f"n_projects must be >= 0, got {n_projects}") per_run = estimate_run_ore(model, effort, pricing, max_tokens) if per_run is None: return None return n_projects * per_run def _ore_to_kr_str(ore: int) -> str: """Format integer øre as an edge-formatted kr STRING (never a float — preserves the no-float invariant). E.g. 123456 øre → '1 234,56 kr' (Norwegian: space thousands, comma decimals).""" sign = "-" if ore < 0 else "" kr, rest = divmod(abs(ore), 100) kr_str = f"{kr:,}".replace(",", " ") # non-breaking space thousands separator return f"{sign}{kr_str},{rest:02d} kr" def _guidance_for(model: str, pricing: PricingContract) -> dict[str, str] | None: """The model's quality guidance as ``{note, source}`` (guidance with kilde), or ``None``.""" entry = pricing.models.get(model) if entry is None or entry.quality_guidance is None: return None return {"note": entry.quality_guidance.note, "source": entry.quality_guidance.source} def build_estimate_table( profile: str, efforts: list[str], pricing: PricingContract, *, n_projects: int, model_map: dict[str, dict[str, str]] | None = None, max_tokens: int = _DEFAULT_MAX_TOKENS, ) -> dict[str, Any]: """Build the what-if estimate table for ``profile`` — one row per (role × effort), sorted for determinism. Priced rows carry ``estimat_ore`` (int), ``estimat_kr`` (edge string), ``estimat_type: "øvre grense"`` (honesty label) and ``kvalitets_veiledning`` (sourced guidance, brief Goal 4). Placeholder-deployment rows carry ``estimat_ore: None`` + ``estimat_type: "uspesifisert (placeholder-deployment)"`` (no crash). The top-level ``kost_mot_verdi`` field is the S5.4 value-report seam. A genuine (non-placeholder) model absent from the price map fails fast (``ValueError`` via ``_price_ore_per_1k``).""" roles = _load_model_map(profile, model_map) rows: list[dict[str, Any]] = [] total_ore = 0 for role in sorted(roles): model = roles[role] rate = _price_ore_per_1k(model, pricing) # raises for genuine-missing; None for placeholder guidance = _guidance_for(model, pricing) for effort in sorted(efforts): if rate is None: rows.append( { "role": role, "model": model, "effort": effort, "estimat_ore": None, "estimat_kr": None, "estimat_type": "uspesifisert (placeholder-deployment)", "kvalitets_veiledning": guidance, } ) else: est = estimate_portfolio_ore( model, effort, pricing, n_projects=n_projects, max_tokens=max_tokens ) assert est is not None # rate is not None → est is int total_ore += est rows.append( { "role": role, "model": model, "effort": effort, "estimat_ore": est, "estimat_kr": _ore_to_kr_str(est), "estimat_type": "øvre grense", "kvalitets_veiledning": guidance, } ) return { "profile": profile, "n_projects": n_projects, "max_tokens": max_tokens, "pricing_source": pricing.source, "pricing_date": pricing.date, "rows": rows, "kost_mot_verdi": { "kjoring_kostet_ore": total_ore, "kjoring_kostet_kr": _ore_to_kr_str(total_ore), "kvalitetssikret_modellert_besparelse_ore": None, "note": "S5.4 verdirapport-kjernen finnes; kost-mot-verdi-wiring gjenstår", }, } def dump_estimate_table(table: dict[str, Any]) -> str: """Byte-deterministic JSON serialization of an estimate table (mirrors ``outbox._dump``: ``sort_keys``, ``indent=2``, trailing LF).""" return json.dumps(table, sort_keys=True, indent=2) + "\n" def _format_table_text(table: dict[str, Any]) -> str: """Human-readable estimate table for the CLI (honest labels + per-model guidance + kost-mot-verdi + adoption-path footer).""" lines = [ f"Kostnadsestimat (øvre grense, ikke prediksjon) — profil: {table['profile']}, " f"prosjekter: {table['n_projects']}, max_tokens/run: {table['max_tokens']}", f"Prisdata: {table['pricing_source']} ({table['pricing_date']})", "", f"{'rolle':<10} {'modell':<30} {'effort':<10} {'estimat':>16} type", ] for row in table["rows"]: est = row["estimat_kr"] if row["estimat_kr"] is not None else "—" lines.append( f"{row['role']:<10} {row['model']:<30} {row['effort']:<10} {est:>16} " f"{row['estimat_type']}" ) guidance = row["kvalitets_veiledning"] if guidance is not None: lines.append(f" veiledning (kilde: {guidance['source']}): {guidance['note']}") kmv = table["kost_mot_verdi"] lines += [ "", f"kost-mot-verdi: kjøringen kostet ~{kmv['kjoring_kostet_kr']} (modellert øvre grense); " "kvalitetssikret modellert besparelse: S5.4 verdirapport-kjernen finnes, " "kost-mot-verdi-wiring gjenstår", "", "Adopsjonssti: start liten (én dimensjon, ett prosjekt, lavt tak) → eskaler med tilliten.", ] return "\n".join(lines) def main(argv: list[str] | None = None) -> int: """CLI entry: ``python -m portfolio_optimiser.costsim`` — offline what-if estimate table over the model-map (models × effort levels). No network; fail-fast (rc 1) on missing/invalid config.""" import argparse parser = argparse.ArgumentParser( prog="portfolio_optimiser.costsim", description="Offline kostnadssimulering før kjøring (S3.6) — estimerer kjøringskost per " "modell × effortnivå fra skjema-validert prisdata. Ingen modellkall, ingen nettverk.", ) parser.add_argument("--profile", default="local", help="model-map profil (local/azure)") parser.add_argument("--projects", type=int, default=4, help="antall prosjekter i porteføljen") parser.add_argument( "--efforts", default="low,standard,high", help="komma-liste av effortnivåer" ) parser.add_argument("--pricing", default=None, help="sti til prisdata-konfig (default: pakket)") args = parser.parse_args(argv) efforts = [e.strip() for e in args.efforts.split(",") if e.strip()] try: pricing = load_pricing(args.pricing) table = build_estimate_table(args.profile, efforts, pricing, n_projects=args.projects) except (FileNotFoundError, ValueError) as exc: print(f"costsim: {exc}", file=sys.stderr) return 1 print(_format_table_text(table)) return 0 if __name__ == "__main__": # pragma: no cover - console entry raise SystemExit(main())