open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 1 Activity
portfolio-optimiser/tests/test_ledger_loadbearing.py

234 lines
8.6 KiB
Python
Raw Blame History

"""Load-bearing seams for the savings ledger (Fase 1 — SC4/SC5/SC6/SC8).
Step 5 (SC4) — the dimension-free dedup gate: the same underlying candidate (same codes+measure+
amount) realized under two dimensions counts ONCE in the portfolio sum — double-counting across
dimensions is the exact error SC4 warns against — while the overlap is FLAGGED (never silently
merged). RED if the dimension-free sum-key dedup is detached (the sum doubles). Inverse control:
two DISTINCT realizations with the same codes+measure but a different amount stay SEPARATE (a
magnitude-free identity would collide them).
(Step 6 extends this file with the fail-closed ``realize`` gate + øre-boundary tests; Step 8 adds
the goal-stop detach.)
Pattern: ``tests/test_step8_promotion_loadbearing.py:77`` (fail-closed gate trio).
"""
from __future__ import annotations
import pytest
from portfolio_optimiser.contracts import GoalConfig, GoalContract
from portfolio_optimiser.ledger import (
LedgerEntry,
RealizationRefused,
SavingsLedger,
_candidate_identity,
realize,
stamp,
)
from portfolio_optimiser.run import run_portfolio
from portfolio_optimiser.verdicts import ProposalFeatures, Verdict
_TS = "2026-07-06T00:00:00Z"
_CODES = frozenset({"ENERGI-TOTAL-EL"})
_MEASURE = "energy_efficiency"
def _identity(amount_ore: int) -> str:
return _candidate_identity(affected_codes=_CODES, measure_type=_MEASURE, amount_ore=amount_ore)
def _entry(dimension: str, *, amount_ore: int) -> LedgerEntry:
return LedgerEntry(
project_id="P1",
dimension=dimension,
candidate_identity=_identity(amount_ore),
amount_ore=amount_ore,
verdict_id="v1",
provenance=stamp(approver="ekspert", experiment="sim", timestamp=_TS),
)
def test_same_candidate_two_dimensions_counts_once_and_flags_overlap() -> None:
"""LOAD-BEARING (SC4): the same underlying candidate realized under two dimensions contributes
ONCE to the portfolio sum, and the overlap is flagged. RED if the dimension-free sum-key dedup
is detached in ``_dedup_amount`` (the sum then doubles to 10000)."""
ledger = SavingsLedger()
assert ledger.add_realized(_entry("energi", amount_ore=5000)) is True
# Same candidate identity (same codes+measure+amount), a DIFFERENT dimension -> distinct full
# key -> stored, so the overlap is visible; the sum must still count it once.
assert ledger.add_realized(_entry("asfalt", amount_ore=5000)) is True
assert ledger.portfolio_total() == 5000, (
"double-counted the same candidate across dimensions (SC4)"
)
assert ledger.per_project_total("P1") == 5000
assert ledger.overlaps() == [("P1", _identity(5000))]
def test_distinct_amounts_same_codes_measure_are_not_collided() -> None:
"""INVERSE CONTROL (fixes the magnitude-free collision): two DISTINCT realizations with the same
codes+measure but a different amount are separate candidates -> both counted, no overlap."""
ledger = SavingsLedger()
assert ledger.add_realized(_entry("energi", amount_ore=5000)) is True
assert ledger.add_realized(_entry("energi", amount_ore=7000)) is True # distinct identity
assert ledger.portfolio_total() == 12000 # both counted (no collision)
assert ledger.overlaps() == [] # single dimension -> no cross-dimension overlap
def test_exact_full_key_duplicate_is_not_stored_twice() -> None:
"""An exact (project, dimension, candidate) re-add is a no-op — storage keys on the full key."""
ledger = SavingsLedger()
assert ledger.add_realized(_entry("energi", amount_ore=5000)) is True
assert ledger.add_realized(_entry("energi", amount_ore=5000)) is False # exact duplicate
assert ledger.portfolio_total() == 5000
# --- Step 6: fail-closed realize gate + deterministic øre conversion (SC5/SC8) --------------------
def _features(nok: float = 5000.0) -> ProposalFeatures:
return ProposalFeatures(affected_codes=_CODES, measure_type=_MEASURE, claimed_saving_nok=nok)
def _verdict(decision: str) -> Verdict:
return Verdict(
id="v-abc", proposal_features=_features(), decision=decision, rationale="expert prose"
)
def test_realize_fail_closed_refuses_non_approved() -> None:
"""LOAD-BEARING (SC5): a non-approved verdict — OTHERWISE fully valid (correct amount, valid
provenance, so detaching the gate WOULD add it) — raises ``RealizationRefused`` and the ledger is
unchanged. RED the moment the ``raise`` detaches: the raw entry then appears (self-contamination)."""
ledger = SavingsLedger()
with pytest.raises(RealizationRefused):
realize(
ledger,
_features(),
_verdict("rejected"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
timestamp=_TS,
)
assert ledger.entries == [] # nothing written
assert ledger.portfolio_total() == 0
def test_realize_approved_reaches_ledger() -> None:
"""CAUSALITY: a HITL-approved verdict reaches the ledger (5000 NOK -> 500000 øre)."""
ledger = SavingsLedger()
entry = realize(
ledger,
_features(),
_verdict("approved"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
timestamp=_TS,
)
assert entry in ledger.entries
assert entry.verdict_id == "v-abc"
assert ledger.portfolio_total() == 500000
def test_realize_accepts_approved_with_adjustment() -> None:
"""The gate uses the PROMOTION set: ``approved_with_adjustment`` is admitted (not the binary
run-path FeedbackContract — H6)."""
ledger = SavingsLedger()
realize(
ledger,
_features(),
_verdict("approved_with_adjustment"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
timestamp=_TS,
)
assert ledger.portfolio_total() == 500000
def test_realize_requires_timestamp_keyword() -> None:
"""SC8: ``timestamp`` is a required keyword — no wall-clock default. Omitting it raises TypeError."""
ledger = SavingsLedger()
with pytest.raises(TypeError):
realize(
ledger,
_features(),
_verdict("approved"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
) # no timestamp
def test_realize_ore_conversion_is_exact() -> None:
"""øre boundary: 12345.67 NOK -> 1234567 øre exactly, no binary-float ...66.9999 drift."""
ledger = SavingsLedger()
entry = realize(
ledger,
_features(12345.67),
_verdict("approved"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
timestamp=_TS,
)
assert entry.amount_ore == 1234567
# --- Step 8: goal-stop is load-bearing (SC6) -----------------------------------------------------
_PORTFOLIO_IDS = ["FV42-GSV-E1", "RV13-RAS-TP", "BRU-LAKS-REHAB"]
def _portfolio_ledger(amount_ore: int) -> SavingsLedger:
led = SavingsLedger()
led.add_realized(
LedgerEntry(
project_id="FV42-GSV-E1",
dimension="energi",
candidate_identity="prior-hitl",
amount_ore=amount_ore,
verdict_id="v-prior",
provenance=stamp(approver="ekspert", experiment="earlier", timestamp=_TS),
)
)
return led
async def test_goal_stop_is_load_bearing(make_portfolio_client_factory) -> None:
"""LOAD-BEARING (SC6): a reached HARD portfolio goal stops the pass (no project runs); a
below-goal ledger runs the FULL pass. RED if the goal-stop check is detached in run_portfolio
(a reached goal then runs past the target). The control (below goal -> full pass) proves the
stop is CAUSED by the goal being reached, not by the fixture."""
goals = GoalConfig(portfolio=GoalContract(absolute_ore=1000))
stopped = await run_portfolio(
_PORTFOLIO_IDS,
"local",
ledger=_portfolio_ledger(1000), # == goal (>=)
goals=goals,
client_factory=make_portfolio_client_factory({}),
max_rounds=1,
)
assert stopped.stopped_early is True
assert stopped.runs == () # detach -> this becomes 3 (runs past the reached goal)
below = await run_portfolio(
_PORTFOLIO_IDS,
"local",
ledger=_portfolio_ledger(999), # below the goal
goals=goals,
client_factory=make_portfolio_client_factory({}),
max_rounds=1,
)
assert below.stopped_early is False # control: goal not reached -> full pass
assert len(below.runs) == 3
Reference in a new issue View git blame Copy permalink