AI-Pricing-Agent/tests/test_pricing_safety.py

493 lines
21 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")
# ------------------------------------------------- best-observed price provenance
# The band key is a $0.50-rounded BUCKET LABEL; `avg_price` is what customers really
# paid. Publishing the key as `best_observed_price` recommends a price the SKU never
# ran, and re-deriving the sample size by matching that key against each band's
# `avg_price` never matched — so the recommendation rationale claimed "0 days" of
# evidence for a price band with weeks behind it.
class _EvidenceResult:
"""The subset of AnalysisResult that `_evidence_target` reads."""
def __init__(self, bop, bday, bands, days=None):
self.best_observed_price = bop
self.best_observed_profit_day = bday
self.best_observed_days = days
self.price_bands = bands
def _hist_days(price, units, profit, ad, n, month):
class _D:
def __init__(s, date, sale_price, units, profit, marketing_cost):
s.date, s.sale_price, s.units = date, sale_price, units
s.profit, s.marketing_cost = profit, marketing_cost
return [_D(f"{month:02d}/{i + 1:02d}/2025", price, units, profit, ad)
for i in range(n)]
class _Stack:
"""The fee stack `_elasticity_block` reads — a cheap SKU, comfortably profitable."""
selling_price = 12.37
referral_pct = 0.15
returns_reserve = 0.25
landed_cost = 4.00
fba_fee = 3.00
storage_alloc = 0.25
profit = 2.00
def test_best_observed_price_is_what_customers_paid_not_the_band_key():
from config.settings import get_rules
from pricing_agent.analyze import _elasticity_block
from pricing_agent.performance import best_observed, price_band_performance
# Six months, two price levels. $12.37 rounds into the $12.50 band and wins on
# actual profit/day; $13.90 rounds into the $14.00 band.
hist = []
for month in (1, 2, 3):
hist += _hist_days(12.37, 10, 30.0, 5.0, 20, month)
for month in (4, 5, 6):
hist += _hist_days(13.90, 8, 20.0, 5.0, 20, month)
bands = price_band_performance(hist, band=0.50, min_days=7)
best = best_observed(bands)
assert best["price_band"] == 12.5 # the bucket label
assert best["avg_price"] == 12.37 # the price actually charged
out = _elasticity_block(None, "TESTSKU", _Stack(), get_rules(), history=hist)
# The published TARGET must be the charged price, never the rounded bucket label.
assert out["best_observed_price"] == 12.37
assert out["best_observed_price"] != best["price_band"]
# ...and it must carry its own sample size, so nothing has to re-derive it.
assert out["best_observed_days"] == 60
def _evidence_from_history():
"""Run the real analysis path, then hand its output to the real decision helper.
Feeding `_evidence_target` a hand-picked price would test nothing: the bug was that
`_elasticity_block` published the BAND KEY, which `_evidence_target` then could not
match back to any band. Only the two together reproduce it.
"""
from config.settings import get_rules
from pricing_agent.analyze import _elasticity_block
hist = []
for month in (1, 2, 3):
hist += _hist_days(12.37, 10, 30.0, 5.0, 20, month)
for month in (4, 5, 6):
hist += _hist_days(15.90, 6, 6.0, 0.9, 20, month)
return _elasticity_block(None, "TESTSKU", _Stack(), get_rules(), history=hist)
def test_evidence_rationale_quotes_the_real_sample_size():
from dashboard.live_data import _evidence_target
out = _evidence_from_history()
r = _EvidenceResult(bop=out["best_observed_price"],
bday=out["best_observed_profit_day"],
bands=out["price_bands"],
days=out["best_observed_days"])
got = _evidence_target(r, cur_price=15.90, keys={"best_observed"})
assert got is not None
price, why, _gain = got
assert price == 12.37 # the charged price, not the $12.50 band
assert "over 60 days" in why
assert "over 0 days" not in why
def test_evidence_sample_size_falls_back_to_the_bands_when_field_absent():
from dashboard.live_data import _evidence_target
out = _evidence_from_history()
# Older callers that never set the field must still resolve the sample size —
# which works now only because `best_observed_price` IS a band's `avg_price`.
r = _EvidenceResult(bop=out["best_observed_price"],
bday=out["best_observed_profit_day"],
bands=out["price_bands"], days=None)
_price, why, _gain = _evidence_target(r, cur_price=15.90, keys={"best_observed"})
assert "over 60 days" in why
# ------------------------------------------------- elasticity used for projections
# `elasticity_actionable` gated the DECISION but not the scenario projections, which
# took the raw point estimate whenever one existed. A CI spanning zero — or worse, a
# POSITIVE slope — then drove the 30-day impact tile, the portfolio opportunity total
# and the queue sort. A positive slope projects that raising price sells MORE units.
def _fit(elasticity, actionable):
return {"elasticity": elasticity, "actionable": actionable, "r2": 0.1,
"confidence": "none" if not actionable else "high", "why": "test"}
from dashboard.live_data import projection_elasticity as _project_with
@pytest.mark.parametrize("fit,expect_fitted", [
(_fit(-1.8, True), True), # usable fit — its own slope
(_fit(-0.15, False), False), # CI spans zero
(_fit(0.40, False), False), # POSITIVE — would say a raise sells more
(None, False), # no fit at all
({}, False),
])
def test_projections_only_use_a_statistically_usable_elasticity(fit, expect_fitted):
from dashboard.live_data import FALLBACK_ELASTICITY
el, fitted = _project_with(fit)
assert fitted is expect_fitted
assert el < 0, "a projection must never assume raising price sells more"
if not expect_fitted:
assert el == FALLBACK_ELASTICITY
def test_positive_elasticity_never_projects_volume_growth_on_a_raise():
from dashboard.live_data import _scenarios
fp = {"cost": 8.0, "fba": 4.0, "referral_pct": 0.15, "returns_pct": 0.02,
"variable": 0.5}
el, _ = _project_with(_fit(0.40, False))
scen = _scenarios(25.0, 10.0, 5000.0, fp, el, None, None)
by = dict(zip(scen["scenario"], scen["units_day"]))
assert by["up_5"] < by["current"] < by["down_5"]
# ------------------------------------------------- economics quoted beside a headline
# When a move is step-capped, `rec_key` names the DESTINATION rung while `rec_price` is
# what we actually recommend today. Quoting the destination's units/profit/margin beside
# today's price promises what the move does not deliver. Observed live on
# UBMICROFIBERDUVETTWINWHITE: recommend $17.94 (67 units/day, 6.5% margin) on the way to
# $19.90 (58 units/day, 12.8%) — the card showed the $19.90 figures under a $17.94 header.
def _econ_row(key, price, units, net, margin):
return {"key": key, "price": price, "units_day": units, "net_30d": net,
"net_margin_pct": margin}
STEP_CAPPED = {
"rec_key": "best_observed", # the $19.90 destination the cascade chose
"rec_price": 17.94, # what the 5% step cap actually permits today
"rec_units_day": 0, "units_day": 0, "rec_profit_30d": 0, "profit_30d": 0,
"scen_econ": [
_econ_row("current", 18.30, 65, 2482, 6.9),
_econ_row("recommended", 17.94, 67, 2322, 6.5),
_econ_row("best_observed", 19.90, 58, 4462, 12.8),
],
}
def test_card_quotes_the_price_it_recommends_not_the_capped_destination():
from app import econ_rows
cur, rec = econ_rows(STEP_CAPPED)
assert cur["price"] == 18.30
assert rec["price"] == 17.94, "must follow rec_price, not the rec_key rung"
assert (rec["units_day"], rec["net_30d"], rec["net_margin_pct"]) == (67, 2322, 6.5)
# The destination's much rosier figures must NOT be what gets shown.
assert rec["net_margin_pct"] != 12.8
def test_uncapped_move_still_resolves_to_its_rung():
from app import econ_rows
d = dict(STEP_CAPPED, rec_price=19.90)
_cur, rec = econ_rows(d)
assert rec["price"] == 19.90 and rec["net_margin_pct"] == 12.8
def test_econ_rows_degrades_to_empty_dicts_without_scenario_economics():
from app import econ_rows
cur, rec = econ_rows({"rec_key": "x", "rec_price": 10.0, "scen_econ": []})
assert cur == {} and rec == {}
# Callers use .get(..., fallback), so empty must not raise.
assert cur.get("units_day", 42) == 42