feat(fase1): dimension-free dedup key + cross-dimension overlap flag (F1)

This commit is contained in:
Kjell Tore Guttormsen 2026-07-07 07:57:22 +02:00
commit e0778d2230
2 changed files with 149 additions and 14 deletions

View file

@ -15,6 +15,7 @@ overlap flagging) and the ``_candidate_identity`` helper.
from __future__ import annotations
import hashlib
import json
from pathlib import Path
@ -40,37 +41,93 @@ def stamp(*, approver: str, experiment: str, timestamp: str) -> str:
return f"godkjent av {approver}; eksperiment {experiment}; {timestamp}"
class SavingsLedger(BaseModel):
"""A typed store of realized ``LedgerEntry`` records.
def _candidate_identity(
*, affected_codes: frozenset[str], measure_type: str, amount_ore: int
) -> str:
"""Stable content-hash identity for a realized candidate — a canonical-JSON sha256 over
``sorted(affected_codes) + measure_type + amount_ore`` (mirrors ``verdicts._mint_id``'s canonical
form). Integer *øre* is IN the identity, so it is (a) deterministic no ``30000`` vs ``30000.0``
divergence after a JSON round-trip AND (b) collision-free: two genuinely distinct realizations
with the same codes+measure but a different amount stay SEPARATE (a magnitude-free identity would
under-report by merging them). Named module-level so Step-5 tests and Step-6 ``realize`` construct
entries with the same identity."""
canonical = json.dumps(
{
"affected_codes": sorted(affected_codes),
"measure_type": measure_type,
"amount_ore": amount_ore,
},
sort_keys=True,
separators=(",", ":"),
)
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
``add_realized`` gates accumulation on the dimension-free key ``(project_id, candidate_identity)``
(C1): the same underlying candidate, realized under two dimensions, is counted ONCE. Totals sum
integer øre and are therefore order-independent; the sorted iteration in ``save`` is only for
byte-deterministic serialization, not for sum correctness.
class SavingsLedger(BaseModel):
"""A typed store of realized ``LedgerEntry`` records — the two-part key (C1).
``add_realized`` stores on the FULL key ``(project_id, dimension, candidate_identity)`` (so the
same candidate under two dimensions is kept for overlap flagging); the totals dedup on the
dimension-free key ``(project_id, candidate_identity)`` so that underlying saving is counted
ONCE. Totals sum integer øre and are therefore order-independent; the sorted iteration in
``save`` is only for byte-deterministic serialization, not for sum correctness.
"""
entries: list[LedgerEntry] = Field(default_factory=list)
@staticmethod
def _dedup_key(entry: LedgerEntry) -> tuple[str, str]:
def _sum_key(entry: LedgerEntry) -> tuple[str, str]:
"""The dimension-FREE key: sum/dedup on this so the same candidate realized under two
dimensions is counted ONCE (C1). ``candidate_identity`` embeds the amount, so two entries
sharing this key share the amount summing one representative is well-defined."""
return (entry.project_id, entry.candidate_identity)
@staticmethod
def _storage_key(entry: LedgerEntry) -> tuple[str, str, str]:
"""The FULL key: storage + overlap-flagging ONLY, never the sum. The same candidate under a
different dimension is a distinct full key, so both are stored (the overlap can be flagged)."""
return (entry.project_id, entry.dimension, entry.candidate_identity)
def add_realized(self, entry: LedgerEntry) -> bool:
"""Append ``entry`` unless its dimension-free key is already present. Returns ``True`` if
added, ``False`` if it was a duplicate (already counted never double-summed)."""
seen = {self._dedup_key(e) for e in self.entries}
if self._dedup_key(entry) in seen:
"""Store ``entry`` unless its FULL key is already present. The same candidate under a
DIFFERENT dimension is a distinct full key -> both are stored (so ``overlaps`` can flag it),
while the totals still count the underlying saving ONCE (dimension-free key). Returns ``True``
if stored, ``False`` if it was an exact (full-key) duplicate."""
seen = {self._storage_key(e) for e in self.entries}
if self._storage_key(entry) in seen:
return False
self.entries.append(entry)
return True
def _dedup_amount(self, entries: list[LedgerEntry]) -> int:
"""Sum ``amount_ore`` over UNIQUE dimension-free keys — each underlying candidate counted
once, so a cross-dimension overlap is never double-summed (SC4)."""
seen: set[tuple[str, str]] = set()
total = 0
for e in entries:
key = self._sum_key(e)
if key in seen:
continue
seen.add(key)
total += e.amount_ore
return total
def per_project_total(self, project_id: str) -> int:
"""Total realized øre for one project (order-independent integer sum)."""
return sum(e.amount_ore for e in self.entries if e.project_id == project_id)
"""Total realized øre for one project — dimension-free-deduped, order-independent."""
return self._dedup_amount([e for e in self.entries if e.project_id == project_id])
def portfolio_total(self) -> int:
"""Total realized øre across the whole portfolio (order-independent integer sum)."""
return sum(e.amount_ore for e in self.entries)
"""Total realized øre across the portfolio — dimension-free-deduped, order-independent."""
return self._dedup_amount(self.entries)
def overlaps(self) -> list[tuple[str, str]]:
"""The dimension-free keys registered under MORE THAN ONE dimension — flagged, never
double-summed. Returns the sorted ``(project_id, candidate_identity)`` keys whose stored
entries span >1 dimension."""
dims: dict[tuple[str, str], set[str]] = {}
for e in self.entries:
dims.setdefault(self._sum_key(e), set()).add(e.dimension)
return sorted(key for key, ds in dims.items() if len(ds) > 1)
def save(self, path: str) -> None:
"""Serialize deterministically: entries sorted by their full key, then JSON with

View file

@ -0,0 +1,78 @@
"""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
from portfolio_optimiser.ledger import (
LedgerEntry,
SavingsLedger,
_candidate_identity,
stamp,
)
_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