291 lines
12 KiB
Python
291 lines
12 KiB
Python
"""Safety and statistical-honesty invariants for the pricing engine.
|
|
|
|
Each test here corresponds to a way the engine was previously able to publish a
|
|
number it could not justify.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from pricing_agent.elasticity import (
|
|
estimate_elasticity, optimize_price, unconstrained_optimum,
|
|
)
|
|
from pricing_agent.performance import (
|
|
ad_cost_model, ad_per_unit_at, empirical_break_even, observed_price_range,
|
|
weighted_fit,
|
|
)
|
|
|
|
from dashboard.live_data import _order_ladder, _scenario_economics
|
|
|
|
|
|
# ---------------------------------------------------------------- fixtures
|
|
def _band(price, units, ppu, ad, days, enough=True):
|
|
return {"price_band": price, "avg_price": price, "units_per_day": units,
|
|
"actual_profit_per_day": round(ppu * units, 2), "profit_per_unit": ppu,
|
|
"ad_per_unit": ad, "days": days, "enough_data": enough}
|
|
|
|
|
|
# Shape taken from a real SKU: losing money low, profitable high, ad/unit rising.
|
|
BANDS = [
|
|
_band(23.38, 2224, -0.66, 1.27, 56),
|
|
_band(26.05, 1758, 0.64, 2.00, 13),
|
|
_band(26.61, 1198, 1.41, 1.89, 19),
|
|
_band(27.43, 1618, 1.68, 2.07, 20),
|
|
_band(28.01, 1342, 2.94, 1.64, 12),
|
|
_band(28.87, 1448, 2.15, 2.37, 54),
|
|
_band(29.41, 1398, 4.07, 1.67, 2, enough=False), # thin — must be ignored
|
|
]
|
|
|
|
|
|
def _months(pairs):
|
|
return [{"month": f"2026-{i+1:02d}", "avg_price": p, "units_per_day": u,
|
|
"ad_per_unit": 1.5, "days": d} for i, (p, u, d) in enumerate(pairs)]
|
|
|
|
|
|
# ---------------------------------------------------------------- weighted fit
|
|
def test_weighted_fit_respects_days():
|
|
"""A 2-day observation must not sway the line like a 50-day one."""
|
|
pts_equal = [(1.0, 1.0, 1), (2.0, 2.0, 1), (3.0, 30.0, 1)]
|
|
pts_weighted = [(1.0, 1.0, 50), (2.0, 2.0, 50), (3.0, 30.0, 1)]
|
|
assert weighted_fit(pts_weighted)["slope"] < weighted_fit(pts_equal)["slope"]
|
|
|
|
|
|
def test_weighted_fit_needs_three_points():
|
|
assert weighted_fit([(1.0, 1.0, 5), (2.0, 2.0, 5)]) is None
|
|
|
|
|
|
# ---------------------------------------------------------------- break-even
|
|
def test_empirical_break_even_sits_between_loss_and_profit_bands():
|
|
be = empirical_break_even(BANDS)
|
|
assert be is not None
|
|
assert 23.38 < be["price"] < 26.61, be
|
|
assert be["r2"] > 0.5
|
|
|
|
|
|
def test_empirical_break_even_excludes_thin_bands():
|
|
"""The 2-day band is the most profitable one; it must not move the fit."""
|
|
with_thin = empirical_break_even(BANDS)
|
|
without = empirical_break_even([b for b in BANDS if b["enough_data"]])
|
|
assert with_thin == without
|
|
|
|
|
|
def test_empirical_break_even_none_without_spread():
|
|
flat = [_band(20.0, 100, 1.0, 1.0, 30) for _ in range(3)]
|
|
assert empirical_break_even(flat) is None
|
|
|
|
|
|
# ---------------------------------------------------------------- ad model
|
|
def test_ad_cost_rises_with_price():
|
|
m = ad_cost_model(BANDS)
|
|
assert m is not None and m["slope"] > 0, m
|
|
|
|
|
|
def test_ad_prediction_never_negative_and_falls_back_when_fit_is_weak():
|
|
noisy = [_band(20 + i, 100, 1.0, 2.0 if i % 2 else 0.2, 30) for i in range(5)]
|
|
weak = ad_cost_model(noisy)
|
|
assert ad_per_unit_at(25.0, weak, fallback=1.75) == 1.75
|
|
assert ad_per_unit_at(0.01, ad_cost_model(BANDS), fallback=1.0) >= 0.0
|
|
|
|
|
|
def test_observed_range_uses_only_sampled_bands():
|
|
lo, hi = observed_price_range(BANDS)
|
|
assert (lo, hi) == (23.38, 28.87) # the 2-day $29.41 band is excluded
|
|
|
|
|
|
# ---------------------------------------------------------------- elasticity
|
|
def test_elasticity_reports_a_confidence_interval():
|
|
el = estimate_elasticity(_months([(24.94, 1704, 28), (27.89, 1174, 31),
|
|
(25.26, 1845, 30), (25.52, 2013, 31),
|
|
(28.04, 1821, 30), (24.96, 1715, 27)]))
|
|
assert el is not None
|
|
assert {"std_err", "t_stat", "ci_low", "ci_high", "actionable"} <= set(el)
|
|
assert el["ci_low"] <= el["elasticity"] <= el["ci_high"]
|
|
|
|
|
|
def test_noisy_elasticity_is_not_actionable():
|
|
"""The real SKU's fit: a slope whose CI spans zero must not drive price."""
|
|
el = estimate_elasticity(_months([(24.94, 1704, 28), (27.89, 1174, 31),
|
|
(25.26, 1845, 30), (25.52, 2013, 31),
|
|
(28.04, 1821, 30), (24.96, 1715, 27)]))
|
|
assert el["actionable"] is False
|
|
assert el["ci_low"] < 0 < el["ci_high"]
|
|
assert "why" in el
|
|
|
|
|
|
def test_clean_elasticity_is_actionable():
|
|
clean = _months([(20.0, 2000, 30), (22.0, 1500, 30), (24.0, 1180, 30),
|
|
(26.0, 950, 30), (28.0, 790, 30), (30.0, 670, 30)])
|
|
el = estimate_elasticity(clean)
|
|
assert el["actionable"] is True
|
|
assert el["ci_high"] < 0 # the whole interval is negative
|
|
assert el["elasticity"] < -1
|
|
|
|
|
|
def test_short_stub_periods_are_dropped():
|
|
"""A 3-day calendar stub must not be weighted like a full month."""
|
|
full = [(24.94, 1704, 28), (27.89, 1174, 31), (25.26, 1845, 30),
|
|
(25.52, 2013, 31), (28.04, 1821, 30), (24.96, 1715, 27)]
|
|
with_stub = estimate_elasticity(_months([(26.63, 1092, 3)] + full))
|
|
assert with_stub["n"] == 6 # the 3-day point is gone
|
|
assert with_stub["days"] == sum(d for _, _, d in full)
|
|
|
|
|
|
# ---------------------------------------------------------------- optimizer
|
|
def test_corner_solution_is_flagged():
|
|
"""Profit rising to the last grid point is not a maximum."""
|
|
best, _ = optimize_price(floor=22.45, ceiling=31.45, ref_price=24.96,
|
|
ref_units=1715, elasticity=-2.22,
|
|
net_profit_fn=lambda p: p * 0.8308 - 20.01, step=1.0)
|
|
assert best["price"] == 31.45
|
|
assert best["is_corner"] is True
|
|
|
|
|
|
def test_interior_maximum_is_not_flagged_as_corner():
|
|
"""Given room to find a real peak, the sweep stops short of its ceiling."""
|
|
best, table = optimize_price(floor=20.0, ceiling=60.0, ref_price=24.96,
|
|
ref_units=1715, elasticity=-2.22,
|
|
net_profit_fn=lambda p: p * 0.8308 - 20.01, step=1.0)
|
|
assert best["is_corner"] is False
|
|
assert best["price"] < table[-1]["price"]
|
|
|
|
|
|
def test_closed_form_agrees_with_the_sweeps_argmax():
|
|
"""The sweep returns a 3%-tiebreak price (cheapest within 3% of the peak), so
|
|
compare against its raw argmax — that is what the closed form solves for."""
|
|
p_star = unconstrained_optimum(elasticity=-2.22, margin_rate=0.8308,
|
|
fixed_per_unit=20.01)
|
|
_, table = optimize_price(floor=20.0, ceiling=80.0, ref_price=24.96,
|
|
ref_units=1715, elasticity=-2.22,
|
|
net_profit_fn=lambda p: p * 0.8308 - 20.01, step=0.25)
|
|
argmax = max(table, key=lambda r: r["daily_profit"])
|
|
assert abs(p_star - argmax["price"]) < 0.5
|
|
|
|
|
|
def test_tiebreak_price_stays_within_3pct_of_peak_profit():
|
|
best, table = optimize_price(floor=20.0, ceiling=80.0, ref_price=24.96,
|
|
ref_units=1715, elasticity=-2.22,
|
|
net_profit_fn=lambda p: p * 0.8308 - 20.01, step=0.25)
|
|
peak = max(r["daily_profit"] for r in table)
|
|
assert best["daily_profit"] >= peak * 0.97
|
|
assert best["price"] <= unconstrained_optimum(
|
|
elasticity=-2.22, margin_rate=0.8308, fixed_per_unit=20.01)
|
|
|
|
|
|
def test_no_interior_optimum_when_demand_is_inelastic():
|
|
assert unconstrained_optimum(elasticity=-0.9, margin_rate=0.83,
|
|
fixed_per_unit=20.0) is None
|
|
|
|
|
|
def test_optimizer_survives_a_sku_that_loses_money_at_every_price():
|
|
"""`peak * 0.97` raises the bar ABOVE a negative peak, emptying the tiebreak
|
|
candidate list. That crash took the whole evidence block down with it."""
|
|
best, table = optimize_price(floor=14.0, ceiling=17.0, ref_price=15.99,
|
|
ref_units=624, elasticity=-1.3,
|
|
net_profit_fn=lambda p: p * 0.83 - 20.0, step=0.5)
|
|
assert all(r["daily_profit"] < 0 for r in table)
|
|
assert best is not None
|
|
assert best["daily_profit"] == max(r["daily_profit"] for r in table)
|
|
|
|
|
|
def test_tiebreak_still_prefers_the_cheaper_price_on_a_genuine_tie():
|
|
"""Flat profit per unit and flat demand → every price ties, so take the lowest."""
|
|
best, table = optimize_price(floor=20.0, ceiling=24.0, ref_price=20.0,
|
|
ref_units=100, elasticity=0.0,
|
|
net_profit_fn=lambda p: 1.0, step=1.0)
|
|
assert len({r["daily_profit"] for r in table}) == 1 # a real tie
|
|
assert best["price"] == 20.0
|
|
assert best["is_corner"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------- scenarios
|
|
FP = {"referral_pct": 0.15, "returns_pct": 0.0192, "cost": 9.28, "fba": 8.97,
|
|
"variable": 0.11}
|
|
|
|
|
|
def _econ(**kw):
|
|
grid = {"current": 26.07, "up_2": 26.59, "up_5": 27.37, "down_5": 24.77}
|
|
defaults = dict(cur_price=26.07, units_day=1720.6, fp=FP, elasticity=-2.22,
|
|
tacos_frac=0.0631, storage_30d=1003.0, grid=grid,
|
|
actual_30d=14300.0, actual_rev=1293993.0, actual_ad=81698.0,
|
|
actual_units=51618.0, realized_price=25.07)
|
|
defaults.update(kw)
|
|
return _scenario_economics(**defaults)
|
|
|
|
|
|
def test_every_row_reconciles_revenue_to_units_times_price():
|
|
rows, _ = _econ()
|
|
for x in rows:
|
|
if x["is_actual"]:
|
|
continue
|
|
assert x["revenue_30d"] == pytest.approx(x["units_30d"] * x["price"], rel=0.01), x
|
|
|
|
|
|
def test_modelled_revenue_is_monotone_in_price_when_elastic():
|
|
"""With e < -1, revenue must FALL as price rises. Mixing an actual current row
|
|
with list-anchored projections used to make this flip."""
|
|
rows, _ = _econ()
|
|
proj = sorted((x for x in rows if not x["is_actual"]), key=lambda x: x["price"])
|
|
revs = [x["revenue_30d"] for x in proj]
|
|
assert revs == sorted(revs, reverse=True), proj
|
|
|
|
|
|
def test_current_row_keeps_both_actual_and_modelled_readings():
|
|
rows, meta = _econ()
|
|
cur = next(x for x in rows if x["key"] == "current")
|
|
assert cur["net_30d"] == 14300 # COSMOS fact
|
|
assert "modelled_net_30d" in cur # and its like-for-like twin
|
|
assert cur["price"] == pytest.approx(25.07) # what buyers actually paid
|
|
assert cur["list_price"] == pytest.approx(26.07)
|
|
assert meta["anchor_price"] == pytest.approx(25.07)
|
|
|
|
|
|
def test_no_realization_factor_is_applied():
|
|
_, meta = _econ()
|
|
assert meta["factor"] == 1.0 and meta["calibrated"] is False
|
|
|
|
|
|
def test_ad_spend_rises_with_price_when_the_model_says_so():
|
|
rows, _ = _econ(ad_model=ad_cost_model(BANDS))
|
|
proj = sorted((x for x in rows if not x["is_actual"]), key=lambda x: x["price"])
|
|
per_unit = [x["ad_per_unit"] for x in proj]
|
|
assert per_unit == sorted(per_unit), proj
|
|
|
|
|
|
def test_observed_prices_are_marked_as_evidence():
|
|
grid = {"current": 26.07, "tested": 28.87, "untested": 31.45}
|
|
rows, _ = _econ(grid=grid, bands=BANDS)
|
|
by = {x["key"]: x for x in rows}
|
|
assert by["tested"]["observed_days"] == 54
|
|
assert by["tested"]["observed_net_30d"] == round(2.15 * 1448 * 30)
|
|
assert by["untested"]["observed_days"] == 0
|
|
assert by["untested"]["observed_net_30d"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------- ladder
|
|
LADDER_SCEN = pd.DataFrame([
|
|
{"scenario": "current", "price": 26.07}, {"scenario": "up_2", "price": 26.59},
|
|
{"scenario": "up_5", "price": 27.37}, {"scenario": "down_2", "price": 25.55},
|
|
{"scenario": "down_5", "price": 24.77}, {"scenario": "max_profit", "price": 31.45},
|
|
])
|
|
|
|
|
|
@pytest.mark.parametrize("action,rec,cons,aggr", [
|
|
("Increase", "max_profit", "up_2", "up_5"), # the real regression
|
|
("Increase", "up_5", "up_2", "up_5"),
|
|
("Decrease", "down_5", "down_2", "down_5"),
|
|
])
|
|
def test_ladder_is_always_ordered(action, rec, cons, aggr):
|
|
p = dict(zip(LADDER_SCEN.scenario, LADDER_SCEN.price))
|
|
c, a = _order_ladder(action, rec, cons, aggr, LADDER_SCEN, 26.07)
|
|
order = [p[c], p[rec], p[a]]
|
|
assert order == (sorted(order) if action == "Increase"
|
|
else sorted(order, reverse=True)), order
|
|
|
|
|
|
def test_ladder_untouched_for_hold_actions():
|
|
assert _order_ladder("Maintain", "current", "current", "up_2",
|
|
LADDER_SCEN, 26.07) == ("current", "up_2")
|