"""Walk-forward backtest + trust scoring (pure, no network).""" from __future__ import annotations import pytest from pricing_agent.backtest import METHODS, trust_score, walk_forward from pricing_agent.cosmos.models import SalesDay FP = {"referral_pct": 0.15, "returns_pct": 0.02, "cost": 9.28, "fba": 8.97, "variable": 0.11} def _hist(days, price_fn, units_fn, profit_fn, ad_fn=lambda i: 100.0): """Synthetic daily history, oldest first, over `days` days of 2026.""" out = [] for i in range(days): month, day = divmod(i, 28) out.append(SalesDay( date=f"{month + 1:02d}/{day + 1:02d}/2026", salePrice=price_fn(i), unitOrders=units_fn(i), profit=profit_fn(i), marketingCost=ad_fn(i), revenue=price_fn(i) * units_fn(i))) return out def _flat(days=180, price=26.0, units=100.0, profit=200.0): return _hist(days, lambda i: price, lambda i: units, lambda i: profit) # ---------------------------------------------------------------- shape def test_returns_none_without_enough_history(): assert walk_forward(_flat(days=60), FP) is None def test_produces_folds_and_scores_for_every_method(): bt = walk_forward(_flat(), FP, list_price=26.0) assert bt is not None assert bt["n_folds"] >= 1 assert set(bt["scores"]) == set(METHODS) for m in METHODS: assert bt["scores"][m]["n"] == bt["n_folds"] assert bt["scores"][m]["mae"] >= 0 def test_folds_never_train_on_the_window_they_predict(): bt = walk_forward(_flat(), FP, list_price=26.0) for f in bt["folds"]: assert f["train_days"] == f["cut_day"] assert f["test_days"] <= bt["holdout_days"] def test_persistence_is_exact_on_a_perfectly_stable_sku(): """Constant profit every day → 'assume last month repeats' cannot be wrong. The gap-corrected model is also exact here (it is fitted to reproduce the training level), so the winner just has to be one of the zero-error methods.""" bt = walk_forward(_flat(profit=200.0), FP, list_price=26.0) assert bt["scores"]["persistence"]["mae"] == 0 assert bt["scores"][bt["best"]]["mae"] == 0 def test_gap_correction_fixes_the_level_on_a_stable_sku(): """A SKU booking far less than the fee model implies: the constant $/unit gap should close almost all of it, where the raw model cannot.""" hist = _hist(180, lambda i: 26.0, lambda i: 100.0, lambda i: 1.0) bt = walk_forward(hist, FP, list_price=26.0) assert bt["scores"]["anchored_gap"]["mae"] < bt["scores"]["anchored"]["mae"] def test_selection_never_picks_a_non_pricing_model(): """Persistence can score best, but it has no price response, so it must never become the model the engine prices with.""" bt = walk_forward(_flat(profit=200.0), FP, list_price=26.0) assert bt["scores"]["persistence"]["mae"] == 0 assert bt["selected"] in ("anchored", "anchored_gap") def test_selection_keeps_the_default_without_a_material_margin(): """A rival must be clearly better, not marginally, before the engine switches.""" from pricing_agent.backtest import DEFAULT_METHOD bt = walk_forward(_flat(), FP, list_price=26.0) s = bt["scores"] if s["anchored"]["mae"] >= s[DEFAULT_METHOD]["mae"] * 0.85: assert bt["selected"] == DEFAULT_METHOD def test_beats_baseline_flag_tracks_the_selected_method(): bt = walk_forward(_flat(), FP, list_price=26.0) sel, base = bt["scores"][bt["selected"]]["mae"], bt["scores"]["persistence"]["mae"] assert bt["beats_baseline"] == (sel < base) def test_predicting_at_the_charged_price_scores_the_profit_model(): """The held-out price is the one really charged, not one the engine picked.""" prices = lambda i: 24.0 if i < 120 else 28.0 bt = walk_forward( _hist(180, prices, lambda i: 100.0, lambda i: 200.0), FP, list_price=26.0) last = bt["folds"][-1] assert last["test_price"] == pytest.approx(28.0) assert last["train_price"] < last["test_price"] def test_bias_sign_reports_direction_of_the_miss(): """A SKU whose booked profit is far below the fee model → model over-states.""" bt = walk_forward( _hist(180, lambda i: 26.0, lambda i: 100.0, lambda i: 1.0), FP, list_price=26.0) assert bt["scores"]["anchored"]["bias"] > 0 # predicted higher than actual # ---------------------------------------------------------------- trust def _bands(n, days=30): return [{"avg_price": 20.0 + i, "price_band": 20.0 + i, "units_per_day": 100, "actual_profit_per_day": 100.0, "profit_per_unit": 1.0, "ad_per_unit": 1.0, "days": days, "enough_data": True} for i in range(n)] GOOD_EL = {"actionable": True, "elasticity": -1.5} BAD_EL = {"actionable": False, "elasticity": -2.2} def _bt(mape, mae=1000, persistence_mae=5000, selected="anchored_gap"): scores = {m: {"mae": mae, "mape": mape, "bias": 0, "n": 3} for m in ("anchored", "anchored_gap")} scores["persistence"] = {"mae": persistence_mae, "mape": 50, "bias": 0, "n": 3} scores["legacy"] = {"mae": 9000, "mape": 90, "bias": 0, "n": 3} return {"scores": scores, "best": selected, "selected": selected, "beats_baseline": mae < persistence_mae, "n_folds": 3, "folds": [], "holdout_days": 30} def test_high_trust_requires_everything_to_pass(): t = trust_score(_bt(12), GOOD_EL, _bands(5), storage_verified=True) assert t["tier"] == "High" and t["auto_approve"] is True def test_missing_costs_blocks_all_pricing_action(): t = trust_score(_bt(5), GOOD_EL, _bands(5), has_costs=False, storage_verified=True) assert t["tier"] == "None" and t["auto_approve"] is False def test_unusable_elasticity_demotes_below_auto_approve(): t = trust_score(_bt(10), BAD_EL, _bands(5), storage_verified=True) assert t["auto_approve"] is False assert any("elasticity" in w for w in t["reasons"]) def test_thin_evidence_base_demotes(): t = trust_score(_bt(10), GOOD_EL, _bands(1), storage_verified=True) assert t["tier"] == "Low" def test_large_backtest_error_demotes_to_low(): t = trust_score(_bt(80), GOOD_EL, _bands(5), storage_verified=True) assert t["tier"] == "Low" assert any("does not predict" in w for w in t["reasons"]) def test_losing_to_persistence_demotes_even_when_error_looks_small(): t = trust_score(_bt(10, mae=6000, persistence_mae=1000), GOOD_EL, _bands(5), storage_verified=True) assert t["tier"] == "Low" assert any("repeats" in w for w in t["reasons"]) def test_unverified_storage_caps_trust_at_medium(): t = trust_score(_bt(5), GOOD_EL, _bands(5), storage_verified=False) assert t["tier"] == "Medium" assert any("storage" in w for w in t["reasons"]) def test_unscored_sku_is_never_auto_approved(): t = trust_score(None, GOOD_EL, _bands(5), storage_verified=True) assert t["auto_approve"] is False assert any("unscored" in w for w in t["reasons"]) def test_tier_is_the_worst_signal_not_an_average(): """One broken input is enough to make a recommendation wrong.""" t = trust_score(_bt(5), GOOD_EL, _bands(1), has_costs=False, storage_verified=True) assert t["tier"] == "None"