"""Bid MILP: property tests that the output respects ledger bounds and capacity (ROADMAP M2 acceptance), plus revenue-consistency and infeasibility handling.""" from __future__ import annotations from decimal import Decimal import numpy as np from hypothesis import given, settings from hypothesis import strategies as st from vpp_contracts.bid_optimization_request import BidOptimizationRequest from vpp_skills.bid_milp import optimize_bid from vpp_skills.numeric import quantize from .conftest import REF_A, curve_of DATE = "2026-03-15" def _request(p50: np.ndarray, spread: float, cap_mw: np.ndarray, e_min: float, e_max: float, lam: str = "0.3", k: str = "0.9", min_block: str = "0.5", cost: str = "0") -> BidOptimizationRequest: p10, p90 = p50 * (1 - spread), p50 * (1 + spread) q = lambda arr, s: [quantize(float(v), s) for v in arr] # noqa: E731 return BidOptimizationRequest.model_validate( { "market_date": DATE, "price_forecast": { "id": "fc-price", "kind": "PRICE", "market_date": DATE, "unit": "yuan_per_mwh", "quantiles": {"p10": curve_of(q(p10, 2), DATE), "p50": curve_of(q(p50, 2), DATE), "p90": curve_of(q(p90, 2), DATE)}, "model": {"name": "price-forecast", "version": "1.0.0"}, "features_snapshot_ref": REF_A, "generated_at": "2026-03-14T06:00:00Z", }, "adjustable_capacity_mw": curve_of(q(cap_mw, 3), DATE), "position_bounds": {"ledger_version": 42, "daily_energy_min_mwh": quantize(e_min, 3), "daily_energy_max_mwh": quantize(e_max, 3)}, "risk": {"risk_aversion": lam, "commitment_buffer_k": k, "min_block_mwh": min_block, "marginal_cost_yuan_per_mwh": cost}, } ) def _vals(c) -> np.ndarray: return np.array([float(Decimal(v.root)) for v in c.values]) @settings(max_examples=40, deadline=None) @given( seed=st.integers(0, 10_000), lam=st.sampled_from(["0", "0.25", "0.5", "1"]), k=st.sampled_from(["0.5", "0.8", "1"]), band=st.floats(0.0, 0.3), ) def test_bid_respects_ledger_bounds_and_capacity(seed, lam, k, band): rng = np.random.default_rng(seed) p50 = rng.uniform(200, 800, 96) cap = rng.uniform(0, 40, 96) sellable = float(k) * cap.sum() * 0.25 centre = rng.uniform(0.2, 0.9) * sellable e_min, e_max = centre * (1 - band), centre * (1 + band) req = _request(p50, 0.2, cap, e_min, e_max, lam=lam, k=k) res = optimize_bid(req) assert res.solver.status.value == "OPTIMAL", res.solver qty = _vals(res.quantities_mwh) total = float(Decimal(res.daily_energy_mwh)) assert abs(total - qty.sum()) < 1e-6 # headline figure derived from the curve lo, hi = Decimal(req.position_bounds.daily_energy_min_mwh), Decimal(req.position_bounds.daily_energy_max_mwh) assert lo <= Decimal(res.daily_energy_mwh) <= hi # P7 cascade: hard constraint, exact assert np.all(qty <= float(k) * cap * 0.25 + 1e-3) assert np.all((qty == 0) | (qty >= 0.5 - 1e-3)) # min block honoured assert np.all(qty >= 0) def test_prefers_high_price_intervals(): p50 = np.full(96, 300.0) p50[72:80] = 900.0 # 18:00–20:00 evening peak cap = np.full(96, 40.0) res = optimize_bid(_request(p50, 0.1, cap, 40.0, 60.0, lam="0")) qty = _vals(res.quantities_mwh) assert qty[72:80].sum() > 0.99 * qty.sum() assert "daily_energy_max" in res.binding_constraints def test_revenue_distribution_is_exact_and_ordered(): rng = np.random.default_rng(1) p50 = rng.uniform(300, 600, 96) req = _request(p50, 0.15, np.full(96, 30.0), 200.0, 400.0, lam="0.5") res = optimize_bid(req) d = res.revenue_distribution_yuan assert Decimal(d.p10) <= Decimal(d.p50) <= Decimal(d.p90) assert res.expected_revenue_yuan == d.p50 # Price-taker offer (cost 0) clears everywhere: P50 revenue = Σ p50·q exactly, in Decimal. p50s = [v.root for v in req.price_forecast.quantiles.p50.values] qty = [v.root for v in res.quantities_mwh.values] expected = sum(Decimal(p) * Decimal(q) for p, q in zip(p50s, qty)) assert Decimal(res.expected_revenue_yuan) == expected.quantize(Decimal("0.01")) def test_offer_is_the_marginal_cost_floor(): p50 = np.full(96, 500.0) free = optimize_bid(_request(p50, 0.2, np.full(96, 30.0), 100.0, 200.0, cost="0")) costly = optimize_bid(_request(p50, 0.2, np.full(96, 30.0), 100.0, 200.0, cost="450")) assert set(_vals(free.prices_yuan_per_mwh)) == {0.0} assert set(_vals(costly.prices_yuan_per_mwh)) == {450.0} # At cost 450 the offer no longer clears on the P10 path (400): floor revenue is zero. assert costly.revenue_distribution_yuan.p10 == "0.00" assert Decimal(free.revenue_distribution_yuan.p10) > 0 def test_risk_aversion_tilts_allocation_toward_narrow_bands(): """Two intervals, same P50; one has a wide band. Risk-neutral is indifferent (fills by index order), risk-averse must prefer the narrow band.""" p50 = np.full(96, 100.0) p50[[10, 20]] = 500.0 cap = np.zeros(96) cap[[10, 20]] = 40.0 band = np.full(96, 0.1) band[10] = 0.6 # interval 10: P10 = 200; interval 20: P10 = 450 q = lambda arr, s: [quantize(float(v), s) for v in arr] # noqa: E731 req = _request(p50, 0.1, cap, 5.0, 5.0, lam="1") data = req.model_dump() data["price_forecast"]["quantiles"]["p10"]["values"] = q(p50 * (1 - band), 2) req = BidOptimizationRequest.model_validate(data) res = optimize_bid(req) qty = _vals(res.quantities_mwh) assert qty[20] == 5.0 and qty[10] == 0.0 def test_infeasible_when_position_exceeds_sellable_energy(): res = optimize_bid(_request(np.full(96, 400.0), 0.1, np.full(96, 10.0), 500.0, 600.0, k="1")) assert res.solver.status.value == "INFEASIBLE" assert res.daily_energy_mwh == "0.000" assert res.binding_constraints == ["daily_energy_min exceeds sellable energy"] def test_rejects_out_of_range_risk_params(): import pytest with pytest.raises(ValueError, match="risk_aversion"): optimize_bid(_request(np.full(96, 400.0), 0.1, np.full(96, 10.0), 10.0, 20.0, lam="1.5")) with pytest.raises(ValueError, match="commitment_buffer_k"): optimize_bid(_request(np.full(96, 400.0), 0.1, np.full(96, 10.0), 10.0, 20.0, k="0"))