portfolio-optimiser/spikes/c_validator.py

"""Spike C — blocking deterministic hybrid validator (B1).

Fully deterministic; **no MAF builders**. Demonstrates that an out-of-range savings
proposal can be **structurally blocked** before it ever reaches an expert, via three
stages over a typed IR:

1. **Pydantic IR** (`SavingsProposal`) — field validators + a cross-field
   `@model_validator` reject malformed proposals at construction (negative quantities,
   a claimed saving exceeding the affected items' own total).
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) + `statistics.quantiles` over uncertain
   unit-costs give P10/P50/P90 of the feasible saving. To honour D6 (no heavy runs) the
   CBC solve runs **once**; the MC reuses that LP's closed-form optimum, which is exact
   for this model.

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.
"""

from __future__ import annotations

import random
import statistics
import warnings
from contextlib import contextmanager
from dataclasses import dataclass

import pulp
from pydantic import BaseModel, Field, model_validator

from portfolio_optimiser.reference_domain import Project
from spikes._harness import Budget, live_local_client_or_skip

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 Fase 2 migration note). The warning is cosmetic here."""
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", DeprecationWarning)
        yield


class AffectedItem(BaseModel):
    """One project cost line a proposal claims to save against."""

    code: str
    quantity: float = Field(ge=0)  # quantities must be >= 0
    unit_cost: float = Field(gt=0)

    @property
    def total(self) -> float:
        return self.quantity * self.unit_cost


class SavingsProposal(BaseModel):
    """Typed IR for a candidate cost-saving measure (B1)."""

    project_id: str
    measure: str
    affected_items: list[AffectedItem] = Field(min_length=1)
    claimed_saving_nok: float = Field(gt=0)
    # code -> (low_unit_cost, high_unit_cost) for the Monte Carlo step; empty = degenerate.
    assumptions: dict[str, tuple[float, float]] = Field(default_factory=dict)

    @model_validator(mode="after")
    def _claim_within_affected_total(self) -> SavingsProposal:
        total = sum(item.total for item in self.affected_items)
        if self.claimed_saving_nok > total:
            raise ValueError(
                f"claimed saving {self.claimed_saving_nok} exceeds affected items' total {total}"
            )
        return self


@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: object,
    *,
    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. Bounded — never loops forever (B4)."""
    if max_attempts <= 0:
        raise ValueError(f"max_attempts must be positive, got {max_attempts}")
    budget = Budget(max_tokens=10**9, max_rounds=max_attempts)  # round cap == attempt cap
    last: Rejection | None = None
    for attempt in range(1, budget.max_rounds + 1):
        result = validate_proposal(generate(attempt))  # type: ignore[operator]
        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 (test 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 {},
    )


def generate_via_llm(project: Project) -> SavingsProposal:
    """Gated LLM-generation entry (live arm). ``skip``s without a LOCAL endpoint — the
    deterministic validator/solver/MC logic is what the gate proves."""
    _client = live_local_client_or_skip()  # pytest.skip if PORTFOLIO_LOCAL_* unset
    raise NotImplementedError(  # pragma: no cover - reached only with a live endpoint
        "live LLM generation is a Fase 2 concern; the spike validates crafted IR directly"
    )