184 lines
7.1 KiB
Python
184 lines
7.1 KiB
Python
"""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"
|