173 lines
6.9 KiB
Python
173 lines
6.9 KiB
Python
"""Deterministic, blocking hybrid validator (B1) — the obligatory, non-optional gate.
|
|
|
|
Pure module: **NO** ``agent_framework`` import (D7-portable core). Three stages over the
|
|
typed IR (``ir.py``):
|
|
|
|
1. **Pydantic IR** invariants already ran at construction (``ir.SavingsProposal``).
|
|
2. **PuLP solver-in-the-loop** — a real CBC solve bounds the maximum feasible saving (R2).
|
|
CBC ships in PuLP's wheel; if it is genuinely absent the step **escalates**
|
|
(``CbcUnavailable``) — no silent LP-relaxation fallback.
|
|
3. **Monte Carlo** — stdlib ``random`` (seeded ``_MC_SEED``) + ``statistics.quantiles`` over
|
|
uncertain unit-costs give P10/P50/P90 of the feasible saving.
|
|
|
|
The structural block (stage 4) returns a ``Rejection`` that is a *different type* from
|
|
``ValidatedProposal`` and carries no percentiles, so it can never be consumed as validated.
|
|
|
|
Promoted verbatim from ``spikes/c_validator.py``. The one deliberate change vs the spike:
|
|
``self_repair`` no longer borrows the harness ``Budget`` (that pulls ``agent_framework`` via
|
|
``spikes/_harness``) — it loops directly on ``max_attempts``; the token-budget bound is
|
|
layered on by the Step 10 generate loop, keeping THIS module pure.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import random
|
|
import statistics
|
|
import warnings
|
|
from collections.abc import Callable
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
|
|
import pulp
|
|
|
|
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
|
|
from portfolio_optimiser.reference_domain import Project
|
|
|
|
MAX_SAVING_FRACTION = 0.30
|
|
"""Policy cap: at most 30% of an affected item's cost is realistically recoverable as a
|
|
saving. The LP bounds the feasible saving by this fraction."""
|
|
|
|
_MC_SAMPLES = 512
|
|
_MC_SEED = 20260624
|
|
|
|
|
|
class CbcUnavailable(RuntimeError):
|
|
"""PuLP's bundled CBC solver is not available — escalate (no silent fallback)."""
|
|
|
|
|
|
@contextmanager
|
|
def _quiet_pulp():
|
|
"""Silence PuLP 3.x's ``PULP_CBC_CMD`` DeprecationWarning. The bundled CBC is only
|
|
reachable via ``PULP_CBC_CMD``; PuLP 4.0 will require ``pip install pulp[cbc]`` +
|
|
COIN_CMD (a migration note). The warning is cosmetic here."""
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
yield
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ValidatedProposal:
|
|
"""A proposal that passed every stage. Carries the Monte Carlo percentiles."""
|
|
|
|
proposal: SavingsProposal
|
|
p10: float
|
|
p50: float
|
|
p90: float
|
|
nominal_feasible: float
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Rejection:
|
|
"""A structurally-blocked proposal. Distinct type, no percentiles — it can never be
|
|
consumed as a ``ValidatedProposal``."""
|
|
|
|
proposal: SavingsProposal
|
|
reason: str
|
|
|
|
|
|
def _solve_max_feasible(items: list[AffectedItem], fraction: float) -> float:
|
|
"""Real CBC solve: maximize total saving subject to a per-item upper bound and a
|
|
global fraction cap. Raises ``CbcUnavailable`` if CBC is genuinely missing."""
|
|
with _quiet_pulp():
|
|
solver = pulp.PULP_CBC_CMD(msg=False)
|
|
if not solver.available():
|
|
raise CbcUnavailable("PuLP's bundled CBC solver is not available on this platform")
|
|
prob = pulp.LpProblem("max_feasible_saving", pulp.LpMaximize)
|
|
xs = [pulp.LpVariable(f"x_{i}", lowBound=0, upBound=it.total) for i, it in enumerate(items)]
|
|
prob += pulp.lpSum(xs)
|
|
prob += pulp.lpSum(xs) <= fraction * sum(it.total for it in items)
|
|
status = prob.solve(solver)
|
|
if pulp.LpStatus[status] != "Optimal":
|
|
raise CbcUnavailable(
|
|
f"CBC did not reach an optimal solution (status={pulp.LpStatus[status]})"
|
|
)
|
|
return float(pulp.value(prob.objective))
|
|
|
|
|
|
def _monte_carlo(
|
|
proposal: SavingsProposal, *, fraction: float = MAX_SAVING_FRACTION
|
|
) -> tuple[float, float, float]:
|
|
"""Vary uncertain unit-costs (seeded) and return (P10, P50, P90) of the feasible
|
|
saving. Uses the LP's closed-form optimum (= fraction x sum of sampled totals), which
|
|
is exact here, so we do NOT spawn a CBC subprocess per sample (D6)."""
|
|
rng = random.Random(_MC_SEED)
|
|
feasibles: list[float] = []
|
|
for _ in range(_MC_SAMPLES):
|
|
total = 0.0
|
|
for item in proposal.affected_items:
|
|
rng_range = proposal.assumptions.get(item.code)
|
|
unit_cost = rng.uniform(*rng_range) if rng_range else item.unit_cost
|
|
total += item.quantity * unit_cost
|
|
feasibles.append(fraction * total)
|
|
deciles = statistics.quantiles(feasibles, n=10, method="inclusive")
|
|
return deciles[0], deciles[4], deciles[8] # P10, P50, P90
|
|
|
|
|
|
def validate_proposal(proposal: SavingsProposal) -> ValidatedProposal | Rejection:
|
|
"""Deterministic blocking validation. Returns a ``ValidatedProposal`` only when the
|
|
claim is feasible; otherwise a ``Rejection`` that cannot be consumed as validated."""
|
|
# Stage 1 (Pydantic) already ran at construction. Stage 2: real CBC solve.
|
|
nominal = _solve_max_feasible(proposal.affected_items, MAX_SAVING_FRACTION)
|
|
# Stage 3: Monte Carlo percentiles of the feasible saving.
|
|
p10, p50, p90 = _monte_carlo(proposal)
|
|
# Stage 4: structural block — a claim above the optimistic feasible (P90) is out of range.
|
|
if proposal.claimed_saving_nok > p90:
|
|
return Rejection(
|
|
proposal=proposal,
|
|
reason=f"claimed saving {proposal.claimed_saving_nok:.0f} exceeds P90 feasible {p90:.0f}",
|
|
)
|
|
return ValidatedProposal(proposal=proposal, p10=p10, p50=p50, p90=p90, nominal_feasible=nominal)
|
|
|
|
|
|
def self_repair(
|
|
generate: Callable[[int], SavingsProposal],
|
|
*,
|
|
max_attempts: int = 3,
|
|
) -> ValidatedProposal | Rejection:
|
|
"""Call ``generate(attempt)`` and validate; retry on rejection up to ``max_attempts``,
|
|
then hard-stop and return the last rejection. Attempts-bounded — never loops forever
|
|
(B4). The token-budget bound is layered on by the Step 10 generate loop, not here (this
|
|
module stays pure: no ``agent_framework`` import)."""
|
|
if max_attempts <= 0:
|
|
raise ValueError(f"max_attempts must be positive, got {max_attempts}")
|
|
last: Rejection | None = None
|
|
for attempt in range(1, max_attempts + 1):
|
|
result = validate_proposal(generate(attempt))
|
|
if isinstance(result, ValidatedProposal):
|
|
return result
|
|
last = result
|
|
assert last is not None
|
|
return last
|
|
|
|
|
|
def proposal_for(
|
|
project: Project,
|
|
codes: list[str],
|
|
*,
|
|
claimed_saving_nok: float,
|
|
measure: str = "Reduce scope on selected cost codes",
|
|
assumptions: dict[str, tuple[float, float]] | None = None,
|
|
) -> SavingsProposal:
|
|
"""Build a ``SavingsProposal`` from a real reference project's cost items (helper)."""
|
|
items = [
|
|
AffectedItem(code=ci.code, quantity=ci.quantity, unit_cost=ci.unit_cost)
|
|
for ci in project.cost_items
|
|
if ci.code in codes
|
|
]
|
|
return SavingsProposal(
|
|
project_id=project.id,
|
|
measure=measure,
|
|
affected_items=items,
|
|
claimed_saving_nok=claimed_saving_nok,
|
|
assumptions=assumptions or {},
|
|
)
|