863 lines
40 KiB
Python
863 lines
40 KiB
Python
"""Competitor-driven pricing rules, and the guarantees that make them safe to ship.
|
|
|
|
The whole risk of wiring competitor data into a deterministic cascade is that a best-effort
|
|
scrape starts deciding prices. So the invariants tested here are, in order of importance:
|
|
|
|
1. **Absent / stale / failed / unknown competitor data changes NOTHING.** The verdict must be
|
|
byte-identical to the competitor-blind one. This is the test that lets the feature ship.
|
|
2. **No competitor rule can price below break-even**, however cheap a rival is.
|
|
3. **A rival above our price never triggers a cut** — that is LOST_ELIGIBILITY, where cutting
|
|
donates margin and fixes nothing.
|
|
4. **Inventory risk still outranks competitor position**, which still outranks profit-optimal.
|
|
5. Every fired rule is traceable to ONE named reason code — no blended scores.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import json
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from pricing_agent.competitive_state import (
|
|
CompetitiveState, from_apify, from_scraper_listing, reconcile, unavailable,
|
|
)
|
|
from pricing_agent.schemas import (
|
|
BuyBoxStatus, CompetitiveSnapshot, CompetitorOffer,
|
|
)
|
|
|
|
from dashboard.live_data import _decide, _decide_raw, _scenarios
|
|
|
|
NOW = datetime(2026, 7, 30, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
# ---------------------------------------------------------------- fixtures
|
|
@dataclass
|
|
class FakeResult:
|
|
"""The subset of AnalysisResult the cascade actually reads."""
|
|
|
|
break_even: float = 15.0
|
|
is_losing_money: bool = False
|
|
cover_days: int | None = 60 # between the 35/90 thresholds: no inventory rule
|
|
avg_30d: float = 10.0
|
|
avg_6m: float = 10.0
|
|
profit_optimal_price: float | None = None
|
|
profit_optimal_is_corner: bool = False
|
|
elasticity_actionable: bool = True
|
|
elasticity: dict = field(default_factory=lambda: {
|
|
"elasticity": -1.3, "r2": 0.8, "confidence": "high"})
|
|
ad_curve_unrecoverable: bool = False
|
|
contribution_per_dollar: float | None = None
|
|
break_even_ad_curve: float | None = None
|
|
observed_price_max: float | None = None
|
|
best_observed_price: float | None = None
|
|
best_observed_profit_day: float | None = None
|
|
price_bands: list = field(default_factory=list)
|
|
ad_cost_per_unit: float | None = None
|
|
inventory: float = 5000.0
|
|
|
|
|
|
# Same shape _fee_params() produces: break-even here is ~$15.06, so $25 clears it and a
|
|
# rival at $9 does not.
|
|
FP = {"cost": 8.0, "fba": 4.0, "referral_pct": 0.15, "returns_pct": 0.02, "variable": 0.5}
|
|
CUR = 25.00
|
|
|
|
|
|
def hist(days: int = 180, price: float = CUR, units: float = 10.0):
|
|
return pd.DataFrame({
|
|
"date": pd.date_range("2026-02-01", periods=days),
|
|
"price": [price] * days,
|
|
"units": [units] * days,
|
|
"ad_spend": [5.0] * days,
|
|
"revenue": [price * units] * days,
|
|
"profit": [20.0] * days,
|
|
"inventory": [5000.0] * days,
|
|
})
|
|
|
|
|
|
def scen(comp_match: float | None = None):
|
|
extra = {"min_safe": 16.0}
|
|
if comp_match:
|
|
extra["comp_match"] = comp_match
|
|
return _scenarios(CUR, 10.0, 5000.0, FP, -1.3, None, extra)
|
|
|
|
|
|
def state(status: BuyBoxStatus, *, rival: float | None = None, our: float = CUR,
|
|
age_h: float = 0.0, rivals: int = 1, max_age_hours: float = 6.0
|
|
) -> CompetitiveState:
|
|
"""A state built through the REAL adapter + gate, not hand-assembled.
|
|
|
|
Hand-setting the fields would leave `unusable_because` None on a state that is hours past
|
|
its age limit, so a staleness test would pass while never exercising the gate that makes
|
|
staleness safe. Build it the way production does.
|
|
"""
|
|
snap = CompetitiveSnapshot(
|
|
buy_box_status=status, buy_box_price=rival, competitive_low=rival,
|
|
competitive_median=rival, reason="test",
|
|
offers=([CompetitorOffer(seller_name="them", price=rival, is_ours=False)]
|
|
* rivals if rival else []),
|
|
)
|
|
return from_apify(snap, our_price=our, as_of=NOW - timedelta(hours=age_h),
|
|
max_age_hours=max_age_hours, now=NOW)
|
|
|
|
|
|
def verdict(comp=None, r=None, comp_match=None):
|
|
"""(action, rec_key, reasons) — what the cascade decided."""
|
|
r = r or FakeResult()
|
|
a, rec, _cons, _aggr, reasons, _root, _obj = _decide(
|
|
r, CUR, hist(), FP, scen(comp_match), None, comp)
|
|
return a, rec, reasons
|
|
|
|
|
|
# ============================================================ 1. FAIL-SAFE
|
|
# The invariant the whole feature rests on: competitor data may only ADD a verdict.
|
|
@pytest.mark.parametrize("comp,label", [
|
|
(None, "no competitor object at all"),
|
|
(unavailable("scrape not run"), "explicitly unavailable"),
|
|
(unavailable("scrape failed"), "scrape failed"),
|
|
(state(BuyBoxStatus.UNKNOWN, rival=20.0), "ownership unknown"),
|
|
(state(BuyBoxStatus.LOST_PRICE, rival=20.0, age_h=9.0), "9h old, past the 6h limit"),
|
|
])
|
|
def test_unusable_competitor_data_changes_nothing(comp, label):
|
|
"""Every one of these must reproduce the competitor-blind verdict exactly."""
|
|
blind = verdict(None)
|
|
assert verdict(comp) == blind, f"{label} changed the verdict"
|
|
|
|
|
|
def test_stale_data_is_reported_not_silently_dropped():
|
|
"""'No data' and '9h-old data' look identical in a blank cell; only one is re-runnable."""
|
|
stale = state(BuyBoxStatus.LOST_PRICE, rival=20.0, age_h=9.0)
|
|
assert not stale.usable
|
|
assert "9.0h old" in stale.unusable_because
|
|
# A 20% undercut, which WOULD fire the rule if it were fresh — proving the gate, not the
|
|
# absence of a signal, is what stops it.
|
|
fresh = state(BuyBoxStatus.LOST_PRICE, rival=20.0, age_h=0.0)
|
|
assert verdict(fresh, comp_match=20.0)[2] == ["COMPETITOR_UNDERCUT"]
|
|
assert verdict(stale, comp_match=20.0)[2] == ["NO_SIGNALS"]
|
|
# ...and the reason it was ignored is on the record, not a blank.
|
|
root = _decide_raw(FakeResult(), CUR, hist(), FP, scen(20.0), None, stale)[5]
|
|
assert any("not used" in str(v) for _k, v in root)
|
|
|
|
|
|
# ============================================================ 1b. SCOPED KILL SWITCH
|
|
@pytest.fixture()
|
|
def rules_off(monkeypatch):
|
|
"""Flip competitor_rules_enabled off the way config does, cache included."""
|
|
from config.settings import get_rules
|
|
get_rules.cache_clear()
|
|
real = get_rules()
|
|
patched = real.model_copy(update={"competitor_rules_enabled": False})
|
|
monkeypatch.setattr("config.settings.get_rules", lambda: patched)
|
|
yield
|
|
get_rules.cache_clear()
|
|
|
|
|
|
def test_the_flag_is_on_by_default():
|
|
from config.settings import get_rules
|
|
assert get_rules().competitor_rules_enabled is True
|
|
|
|
|
|
@pytest.mark.parametrize("comp_state,comp_match,rule", [
|
|
(BuyBoxStatus.SUPPRESSED, None, "BUYBOX_SUPPRESSED"),
|
|
(BuyBoxStatus.LOST_PRICE, 22.0, "COMPETITOR_UNDERCUT"),
|
|
])
|
|
def test_switching_the_flag_off_disables_exactly_the_two_competitor_rules(
|
|
comp_state, comp_match, rule, monkeypatch):
|
|
"""Same SKU, same competitor state, flag on vs off — both arms in one test.
|
|
|
|
Asserting only the OFF arm would pass vacuously if the rule never fired in the first place,
|
|
so the ON arm is measured here rather than assumed from a sibling test.
|
|
"""
|
|
from config.settings import get_rules
|
|
|
|
comp = state(comp_state, rival=(comp_match or 22.0))
|
|
|
|
# ---- flag ON: the rule fires
|
|
get_rules.cache_clear()
|
|
on = verdict(comp, comp_match=comp_match)
|
|
assert on[2] == [rule], f"ON arm did not fire {rule}; the OFF arm would be vacuous"
|
|
|
|
# ---- flag OFF: identical to the competitor-blind verdict
|
|
patched = get_rules().model_copy(update={"competitor_rules_enabled": False})
|
|
monkeypatch.setattr("config.settings.get_rules", lambda: patched)
|
|
off = verdict(comp, comp_match=comp_match)
|
|
assert off == verdict(None, comp_match=comp_match)
|
|
assert rule not in off[2]
|
|
assert off != on # the flag demonstrably did something
|
|
|
|
|
|
def test_flag_off_leaves_every_other_cascade_rule_working(rules_off):
|
|
"""The switch is scoped: only the two competitor rules go quiet."""
|
|
assert verdict(None, r=FakeResult(cover_days=20))[2][0] == "LOW_STOCK"
|
|
assert verdict(None, r=FakeResult(cover_days=120))[2][0] == "EXCESS_STOCK"
|
|
assert verdict(None, r=FakeResult(break_even=26.0))[2] == ["BELOW_BREAK_EVEN"]
|
|
assert verdict(None, r=FakeResult(is_losing_money=True))[2] == ["LOSING_MONEY"]
|
|
assert verdict(None, r=FakeResult(profit_optimal_price=30.0))[2] == [
|
|
"PROFIT_OPTIMAL", "PREMIUM_HEADROOM"]
|
|
no_cost = {**FP, "cost": 0.0, "fba": 0.0}
|
|
assert _decide(FakeResult(), CUR, hist(), no_cost, scen(), None, None)[4] == ["NO_COST_DATA"]
|
|
|
|
|
|
def test_flag_off_says_WHY_rather_than_going_quiet(rules_off):
|
|
"""A silently competitor-blind verdict is indistinguishable from a missing scrape."""
|
|
comp = state(BuyBoxStatus.LOST_PRICE, rival=22.0)
|
|
root = _decide_raw(FakeResult(), CUR, hist(), FP, scen(22.0), None, comp)[5]
|
|
note = next(v for k, v in root if k == "Competitor Buy Box")
|
|
assert "switched off in pricing_rules.yaml" in note
|
|
|
|
|
|
def test_the_flag_uses_the_same_fail_safe_path_as_stale_data(rules_off):
|
|
"""Flag-off and stale-data must produce the IDENTICAL verdict, not merely similar ones.
|
|
|
|
If these ever diverge there are two 'pretend competitor data isn't there' code paths, which
|
|
is the bug risk the single-path requirement exists to prevent.
|
|
"""
|
|
fresh = state(BuyBoxStatus.LOST_PRICE, rival=22.0)
|
|
stale = state(BuyBoxStatus.LOST_PRICE, rival=22.0, age_h=9.0)
|
|
assert verdict(fresh, comp_match=22.0) == verdict(stale, comp_match=22.0)
|
|
|
|
|
|
# ============================================================ 2. SUPPRESSION
|
|
def test_suppressed_buybox_returns_investigate():
|
|
action, rec, reasons = verdict(state(BuyBoxStatus.SUPPRESSED))
|
|
assert action == "Investigate"
|
|
assert reasons == ["BUYBOX_SUPPRESSED"] # ONE named reason, no blend
|
|
assert rec == "current" # holds; no price is moved
|
|
|
|
|
|
def test_suppression_outranks_profit_optimal():
|
|
"""A listing nobody can buy from makes its modelled optimum meaningless."""
|
|
r = FakeResult(profit_optimal_price=30.0) # would otherwise fire PROFIT_OPTIMAL
|
|
assert verdict(None, r=r)[2] == ["PROFIT_OPTIMAL", "PREMIUM_HEADROOM"]
|
|
assert verdict(state(BuyBoxStatus.SUPPRESSED), r=r)[2] == ["BUYBOX_SUPPRESSED"]
|
|
|
|
|
|
def test_missing_cost_data_still_outranks_suppression():
|
|
"""With no COGS nothing is computable at all — that stays the first gate."""
|
|
r = FakeResult()
|
|
assert verdict(state(BuyBoxStatus.SUPPRESSED), r=r,
|
|
)[2] == ["BUYBOX_SUPPRESSED"]
|
|
no_cost = {**FP, "cost": 0.0, "fba": 0.0}
|
|
reasons = _decide(r, CUR, hist(), no_cost, scen(), None,
|
|
state(BuyBoxStatus.SUPPRESSED))[4]
|
|
assert reasons == ["NO_COST_DATA"]
|
|
|
|
|
|
def test_suppression_is_not_relabelled_as_an_ad_problem():
|
|
"""Both hold, but the Buy Box is the thing to go and fix."""
|
|
r = FakeResult(ad_curve_unrecoverable=True, contribution_per_dollar=0.05)
|
|
assert verdict(None, r=r)[2] == ["AD_SPIRAL"]
|
|
assert verdict(state(BuyBoxStatus.SUPPRESSED), r=r)[2] == ["BUYBOX_SUPPRESSED"]
|
|
|
|
|
|
# ============================================================ 3. UNDERCUT
|
|
def test_material_undercut_recommends_a_decrease():
|
|
action, rec, reasons = verdict(state(BuyBoxStatus.LOST_PRICE, rival=22.0),
|
|
comp_match=22.0)
|
|
assert action == "Decrease"
|
|
assert reasons == ["COMPETITOR_UNDERCUT"]
|
|
assert rec == "comp_match"
|
|
|
|
|
|
def test_immaterial_undercut_does_nothing():
|
|
"""1.2% below us is inside the noise our own realized price already swings through."""
|
|
action, _rec, reasons = verdict(state(BuyBoxStatus.LOST_PRICE, rival=24.70),
|
|
comp_match=24.70)
|
|
assert reasons == ["NO_SIGNALS"]
|
|
assert action == "Maintain"
|
|
|
|
|
|
def test_rival_above_us_never_triggers_a_cut():
|
|
"""LOST_ELIGIBILITY: they hold the Buy Box at a HIGHER price. Cutting fixes nothing."""
|
|
action, _rec, reasons = verdict(state(BuyBoxStatus.LOST_ELIGIBILITY, rival=28.0),
|
|
comp_match=28.0)
|
|
assert "COMPETITOR_UNDERCUT" not in reasons
|
|
assert action != "Decrease"
|
|
|
|
|
|
def test_winning_the_buybox_never_triggers_a_cut():
|
|
action, _rec, reasons = verdict(state(BuyBoxStatus.WON, rival=22.0), comp_match=22.0)
|
|
assert "COMPETITOR_UNDERCUT" not in reasons
|
|
|
|
|
|
# ======================= 3b. COSMOS cover_days == 0 must not silence the inventory rules
|
|
# Found while validating a real suggested price. COSMOS returns `cover_days: 0` on some
|
|
# low-velocity SKUs that are sitting on months of stock, and 0 passes NEITHER inventory test
|
|
# (`0 < cover <= 35` is false; `cover >= 90` is false), so both rules go quiet and the SKU falls
|
|
# through to the next rung. On UBMICROFIBERBS4PCFULLGREY — 167 units on hand, 15 sales in six
|
|
# months, 334 days of real cover — that produced COMPETITOR_UNDERCUT instead of EXCESS_STOCK,
|
|
# bypassing the "inventory outranks competitor position" guarantee.
|
|
def test_cosmos_cover_days_zero_does_not_silence_both_inventory_rules():
|
|
r = FakeResult(cover_days=0, inventory=167.0) # exactly what COSMOS returned
|
|
comp = state(BuyBoxStatus.LOST_PRICE, rival=22.0)
|
|
|
|
# The raw COSMOS value alone: neither rule can fire, and the competitor rule wins by
|
|
# default. This is the behaviour being fixed, asserted so the bug cannot creep back.
|
|
assert _decide(r, CUR, hist(), FP, scen(22.0), None, comp)[4] == ["COMPETITOR_UNDERCUT"]
|
|
|
|
# With the reconciled cover passed in — 167 units / 0.5 a day = 334 days — the inventory
|
|
# rule fires and correctly outranks the competitor rule.
|
|
reasons = _decide(r, CUR, hist(), FP, scen(22.0), None, comp, 334)[4]
|
|
assert reasons[0] == "EXCESS_STOCK"
|
|
assert "COMPETITOR_UNDERCUT" not in reasons
|
|
|
|
|
|
def test_reconciled_cover_also_restores_the_low_stock_rule():
|
|
"""The same zero hides an UNDERSTOCKED SKU, which is the more expensive direction."""
|
|
r = FakeResult(cover_days=0, inventory=10.0)
|
|
assert _decide(r, CUR, hist(), FP, scen(), None, None)[4] == ["NO_SIGNALS"]
|
|
assert _decide(r, CUR, hist(), FP, scen(), None, None, 20)[4][0] == "LOW_STOCK"
|
|
|
|
|
|
def test_an_explicit_cover_of_zero_is_still_honoured_when_that_is_the_truth():
|
|
"""A genuinely empty shelf (no stock, no velocity) must not be invented into cover."""
|
|
r = FakeResult(cover_days=0, inventory=0.0)
|
|
# 0 units on hand and 0 velocity -> the reconciled figure is also 0, and neither inventory
|
|
# rule should fire off a fabricated number.
|
|
reasons = _decide(r, CUR, hist(), FP, scen(), None, None, 0)[4]
|
|
assert "EXCESS_STOCK" not in reasons and "LOW_STOCK" not in reasons
|
|
|
|
|
|
# ============================================================ 4. ORDERING
|
|
def test_low_stock_outranks_a_competitor_undercut():
|
|
"""Chasing a rival down while the shelf empties pays margin to sell out faster."""
|
|
r = FakeResult(cover_days=20) # <= 35d
|
|
_a, _rec, reasons = verdict(state(BuyBoxStatus.LOST_PRICE, rival=22.0), r=r,
|
|
comp_match=22.0)
|
|
assert reasons[0] == "LOW_STOCK"
|
|
assert "COMPETITOR_UNDERCUT" not in reasons
|
|
|
|
|
|
def test_excess_stock_outranks_a_competitor_undercut():
|
|
r = FakeResult(cover_days=120) # >= 90d
|
|
_a, _rec, reasons = verdict(state(BuyBoxStatus.LOST_PRICE, rival=22.0), r=r,
|
|
comp_match=22.0)
|
|
assert reasons[0] == "EXCESS_STOCK"
|
|
|
|
|
|
def test_below_break_even_outranks_a_competitor_undercut():
|
|
"""Margin protection first: we do not chase a rival while already under water."""
|
|
r = FakeResult(break_even=26.0) # cur 25.00 < 26.00
|
|
_a, _rec, reasons = verdict(state(BuyBoxStatus.LOST_PRICE, rival=22.0), r=r,
|
|
comp_match=22.0)
|
|
assert reasons == ["BELOW_BREAK_EVEN"]
|
|
|
|
|
|
def test_undercut_outranks_profit_optimal():
|
|
"""A price the market is actively beating us on beats a modelled optimum."""
|
|
r = FakeResult(profit_optimal_price=30.0)
|
|
_a, _rec, reasons = verdict(state(BuyBoxStatus.LOST_PRICE, rival=22.0), r=r,
|
|
comp_match=22.0)
|
|
assert reasons == ["COMPETITOR_UNDERCUT"]
|
|
|
|
|
|
# ============================================================ 5. NEVER BELOW BREAK-EVEN
|
|
def test_comp_match_candidate_is_never_recommended_below_break_even():
|
|
"""A rival at $9 against a $15 break-even must not produce a $9 recommendation.
|
|
|
|
The comp_match KEY is allowed to hold $9 — it is a candidate to evaluate. The floor
|
|
clamp downstream is what guarantees the number that ships is not below break-even.
|
|
"""
|
|
s = scen(9.0)
|
|
assert float(s.set_index("scenario").loc["comp_match", "price"]) == 9.0
|
|
_a, rec, reasons = verdict(state(BuyBoxStatus.LOST_PRICE, rival=9.0), comp_match=9.0)
|
|
assert reasons == ["COMPETITOR_UNDERCUT"]
|
|
assert rec == "comp_match"
|
|
# The floor (min_safe) is a candidate in the same grid and sits above the rival price,
|
|
# which is what the guardrail in build_live_sku lifts the target to.
|
|
assert float(s.set_index("scenario").loc["min_safe", "price"]) > 9.0
|
|
|
|
|
|
def test_no_comp_match_key_means_no_undercut_rule():
|
|
"""The rule cannot fire without a priced candidate to move toward."""
|
|
_a, _rec, reasons = verdict(state(BuyBoxStatus.LOST_PRICE, rival=22.0),
|
|
comp_match=None)
|
|
assert "COMPETITOR_UNDERCUT" not in reasons
|
|
|
|
|
|
# ============================================================ 6. PREMIUM = NOTE ONLY
|
|
def test_premium_while_winning_is_a_note_and_never_an_action():
|
|
r = FakeResult()
|
|
comp = state(BuyBoxStatus.WON, rival=20.0) # we are 25% above the field
|
|
action, rec, reasons = verdict(comp, r=r)
|
|
blind_action, blind_rec, blind_reasons = verdict(None, r=r)
|
|
assert (action, rec, reasons) == (blind_action, blind_rec, blind_reasons)
|
|
root = _decide_raw(r, CUR, hist(), FP, scen(), None, comp)[5]
|
|
note = next(v for k, v in root if k == "Competitor premium")
|
|
assert "not a reason to raise" in note
|
|
|
|
|
|
def test_premium_does_not_override_the_elasticity_optimum():
|
|
"""Where the model wants a LOWER price, a premium must not flip it upward."""
|
|
r = FakeResult(profit_optimal_price=23.0) # model says come down
|
|
comp = state(BuyBoxStatus.WON, rival=20.0) # and we are above the field
|
|
action, _rec, reasons = verdict(comp, r=r)
|
|
assert action == "Decrease"
|
|
assert reasons == ["PROFIT_OPTIMAL"]
|
|
|
|
|
|
# ============================================================ 7. STATE ADAPTERS
|
|
def test_apify_snapshot_excludes_our_own_offers_from_the_band():
|
|
"""Our own offer must not become 'the cheapest rival' — that reports a 0% undercut."""
|
|
snap = CompetitiveSnapshot(
|
|
buy_box_status=BuyBoxStatus.LOST_PRICE, buy_box_price=21.0,
|
|
offers=[CompetitorOffer(seller_name="us", price=25.0, is_ours=True),
|
|
CompetitorOffer(seller_name="them", price=21.0, is_ours=False)])
|
|
st = from_apify(snap, our_price=25.0, as_of=NOW, max_age_hours=6.0, now=NOW)
|
|
assert st.competitor_min == 21.0
|
|
assert st.rivals == 1
|
|
assert st.undercut_pct == pytest.approx(0.16)
|
|
|
|
|
|
def test_used_stock_never_enters_the_rival_band():
|
|
"""Second-hand stock is not a competing price for a new listing.
|
|
|
|
Found on LIVE data, not in review. Two real ASINs from the item-3 validation run:
|
|
* B06X421WJ6 — cheapest offer Amazon Resale $22.53 *Used - Very Good* vs our $22.99 new,
|
|
which scored as us being 2% dearer. The only NEW rival was $30.99: we were $8 CHEAPER.
|
|
* B00U6HREPQ — Amazon Resale $21.37 *Used - Like New* vs our $22.49 scored a 4.98%
|
|
undercut, clearing the 3% action threshold outright.
|
|
"""
|
|
snap = CompetitiveSnapshot(
|
|
buy_box_status=BuyBoxStatus.LOST_PRICE, buy_box_price=22.99,
|
|
offers=[
|
|
CompetitorOffer(seller_name="Utopia Brands", price=22.99, is_ours=True,
|
|
is_buy_box=True, condition="New"),
|
|
CompetitorOffer(seller_name="Amazon Resale", price=22.53, is_ours=False,
|
|
condition="Used - Very Good"),
|
|
CompetitorOffer(seller_name="Amazon Resale", price=21.37, is_ours=False,
|
|
condition="Used - Like New"),
|
|
CompetitorOffer(seller_name="Amazon.com", price=30.99, is_ours=False,
|
|
condition="New"),
|
|
])
|
|
st = from_apify(snap, our_price=22.99, as_of=NOW, max_age_hours=6.0, now=NOW)
|
|
assert st.competitor_min == 30.99 # the NEW rival, not the $21.37 used one
|
|
assert st.rivals == 1
|
|
assert st.undercut_pct < 0 # we are cheaper, not undercut
|
|
# The whole point: a used offer must not be able to cut our price.
|
|
assert "COMPETITOR_UNDERCUT" not in verdict(st, comp_match=21.37)[2]
|
|
|
|
|
|
def test_a_new_condition_rival_still_counts():
|
|
"""The used filter must not silently swallow real competition."""
|
|
snap = CompetitiveSnapshot(
|
|
buy_box_status=BuyBoxStatus.LOST_PRICE, buy_box_price=18.00,
|
|
offers=[
|
|
CompetitorOffer(seller_name="us", price=25.00, is_ours=True, condition="New"),
|
|
CompetitorOffer(seller_name="Rival", price=18.00, is_ours=False, condition="New"),
|
|
])
|
|
st = from_apify(snap, our_price=25.00, as_of=NOW, max_age_hours=6.0, now=NOW)
|
|
assert st.competitor_min == 18.00
|
|
assert verdict(st, comp_match=18.00)[2] == ["COMPETITOR_UNDERCUT"]
|
|
|
|
|
|
def test_playwright_listing_cannot_claim_ownership_without_a_seller_name():
|
|
"""It carries no seller id, so 'a featured price rendered' is not 'we hold it'."""
|
|
|
|
class L:
|
|
buybox, price, error = "buybox", 21.0, None
|
|
fields: dict = {}
|
|
|
|
st = from_scraper_listing(L(), our_price=25.0, as_of=NOW, max_age_hours=6.0, now=NOW)
|
|
assert st.status is BuyBoxStatus.UNKNOWN
|
|
assert not st.usable # and therefore cannot drive any rule
|
|
assert "authoritative" in st.notes[-1]
|
|
|
|
|
|
def test_playwright_suppression_is_a_fact_not_an_error():
|
|
"""This scraper sets `error` on a suppressed listing. That must not read as a failure."""
|
|
|
|
class L:
|
|
buybox, price = "suppressed", None
|
|
error = "Buy Box suppressed; Amazon reason: High price"
|
|
fields = {"Buy Box Status": "SUPPRESSED — High price"}
|
|
|
|
st = from_scraper_listing(L(), our_price=25.0, as_of=NOW, max_age_hours=6.0, now=NOW)
|
|
assert st.status is BuyBoxStatus.SUPPRESSED
|
|
assert st.usable
|
|
|
|
|
|
def test_playwright_reads_ownership_when_the_pdp_names_us():
|
|
class L:
|
|
buybox, price, error = "buybox", 25.0, None
|
|
fields = {"Sold By": "Utopia Deals"}
|
|
|
|
st = from_scraper_listing(L(), our_price=25.0, our_seller_name="Utopia Deals",
|
|
as_of=NOW, max_age_hours=6.0, now=NOW)
|
|
assert st.status is BuyBoxStatus.WON
|
|
|
|
|
|
# ============================================================ 8. RECONCILIATION
|
|
def test_disagreement_on_suppression_is_logged_and_recorded():
|
|
apify = state(BuyBoxStatus.WON, rival=22.0)
|
|
play = CompetitiveState(status=BuyBoxStatus.SUPPRESSED, source="playwright",
|
|
as_of=NOW, our_price=CUR)
|
|
merged = reconcile(apify, play, sku="UBTEST")
|
|
assert merged.source == "apify" # primary always wins the field
|
|
assert any("disagree on Buy Box suppression" in n for n in merged.notes)
|
|
|
|
|
|
def test_disagreement_on_price_is_recorded_with_both_figures():
|
|
apify = state(BuyBoxStatus.LOST_PRICE, rival=22.0)
|
|
play = CompetitiveState(status=BuyBoxStatus.LOST_PRICE, buy_box_price=27.0,
|
|
competitor_min=27.0, source="playwright", as_of=NOW,
|
|
our_price=CUR)
|
|
merged = reconcile(apify, play, sku="UBTEST")
|
|
assert merged.buy_box_price == 22.0 # authoritative source's number ships
|
|
note = next(n for n in merged.notes if "featured price" in n)
|
|
assert "$22.00" in note and "$27.00" in note
|
|
|
|
|
|
def test_unusable_primary_promotes_a_usable_secondary():
|
|
"""A Playwright SUPPRESSED must not be discarded to preserve a tidy hierarchy.
|
|
|
|
Suppression means the variant sells nothing at any price. If Apify did not run, throwing
|
|
that away would be choosing the source hierarchy over the fact.
|
|
"""
|
|
apify = unavailable("scrape not run")
|
|
play = CompetitiveState(status=BuyBoxStatus.SUPPRESSED, source="playwright",
|
|
as_of=NOW, our_price=CUR)
|
|
merged = reconcile(apify, play, sku="UBTEST")
|
|
assert merged.source == "playwright"
|
|
assert merged.usable
|
|
assert any("rests on playwright" in n for n in merged.notes)
|
|
# ...and it drives the rule, exactly as an Apify suppression would.
|
|
assert verdict(merged)[2] == ["BUYBOX_SUPPRESSED"]
|
|
|
|
|
|
def test_playwright_cache_reader_is_best_effort(tmp_path):
|
|
"""Any problem returns None. It is a cross-check, not a dependency."""
|
|
from pricing_agent.competitive_state import from_playwright_cache
|
|
|
|
missing = tmp_path / "nope.json"
|
|
assert from_playwright_cache("B08DTH86Q2", our_price=CUR,
|
|
cache_path=str(missing)) is None
|
|
|
|
bad = tmp_path / "bad.json"
|
|
bad.write_text("{not json", encoding="utf-8")
|
|
assert from_playwright_cache("B08DTH86Q2", our_price=CUR,
|
|
cache_path=str(bad)) is None
|
|
|
|
good = tmp_path / "c.json"
|
|
good.write_text(json.dumps({
|
|
"B08DTH86Q2@10001": {"ts": NOW.timestamp(), "row": {
|
|
"buybox": "suppressed", "price": None,
|
|
"error": "Buy Box suppressed; Amazon reason: High price",
|
|
"fields": {"Buy Box Status": "SUPPRESSED — High price"}}}}), encoding="utf-8")
|
|
st = from_playwright_cache("B08DTH86Q2", our_price=CUR, cache_path=str(good),
|
|
max_age_hours=6.0, now=NOW)
|
|
assert st is not None and st.status is BuyBoxStatus.SUPPRESSED
|
|
assert st.source == "playwright"
|
|
|
|
# A cache written for a DIFFERENT ZIP describes a different marketplace. Ignored, not
|
|
# reinterpreted — the ZIP is what decides the price.
|
|
assert from_playwright_cache("B08DTH86Q2", our_price=CUR, zip_code="90210",
|
|
cache_path=str(good), now=NOW) is None
|
|
assert from_playwright_cache("B0NOTTHERE", our_price=CUR, cache_path=str(good),
|
|
now=NOW) is None
|
|
|
|
|
|
# ============================================================ 9. THE SHEET + COVERAGE GATE
|
|
def _write_sheet(tmp_path, rows, *, name="Competitor_Price_Comparison_UBTESTLINE_2026-07-29_1923.xlsx",
|
|
brands=("Bedsure", "CGK")):
|
|
"""A workbook shaped like the real one — including the footnotes that trip a naive parse."""
|
|
from openpyxl import Workbook
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Price Comparison"
|
|
header = ["Our SKU", "Our ASIN", "Size", "Colour", "Config", "Our Price",
|
|
"Our List Price", "Our Buy Box", "Our BSR", "Our Bought/Mo", "Our Stock"]
|
|
for b in brands:
|
|
header += [f"{b} ASIN", f"{b} Price", f"vs {b} ($)", f"vs {b} (%)", f"{b} BSR",
|
|
f"BSR vs {b}", f"{b} Bought/Mo"]
|
|
header += ["Cheapest", "Notes"]
|
|
ws.append(header)
|
|
for r in rows:
|
|
line = [r.get("sku"), r.get("asin"), r.get("size"), r.get("colour"), "3-pc",
|
|
r.get("our_price"), None, r.get("buybox"), None, None, None]
|
|
for b in brands:
|
|
line += [f"B0RIV{b[:2].upper()}01", r.get(b), None, None, None, None, None]
|
|
line += [r.get("cheapest"), r.get("notes")]
|
|
ws.append(line)
|
|
# The real workbook writes its footnotes into column A under the table. 6 of them.
|
|
ws.append([])
|
|
for note in ("Our Price is scraped LIVE from Amazon on every row…",
|
|
"Our Buy Box is scraped live: COSMOS knows what we charge…",
|
|
"Our Stock is UNITS ON HAND from COSMOS…",
|
|
"A 'NEAR colour match' note means the colours are not identically labelled…",
|
|
"BSR is each listing's OWN sub-category rank…",
|
|
"'BSR vs <brand>' is left BLANK when their listing ranks in a DIFFERENT…"):
|
|
ws.append([note])
|
|
p = tmp_path / name
|
|
wb.save(p)
|
|
wb.close()
|
|
return p
|
|
|
|
|
|
SHEET_ROWS = [
|
|
{"sku": "UBTESTLINEQUEENWHITE", "asin": "B0OURS0001", "size": "QUEEN",
|
|
"colour": "White", "our_price": 28.99, "buybox": "Buy Box",
|
|
"Bedsure": 24.99, "CGK": 31.00, "cheapest": "Bedsure"},
|
|
{"sku": "UBTESTLINEKINGWHITE", "asin": "B0OURS0002", "size": "KING",
|
|
"colour": "White", "our_price": 30.00, "buybox": "Buy Box",
|
|
"Bedsure": 29.70, "CGK": None, "cheapest": "Bedsure"},
|
|
{"sku": "UBTESTLINETWINPINK", "asin": "B0OURS0003", "size": "TWIN",
|
|
"colour": "Pink", "our_price": None, "buybox": "SUPPRESSED — High price",
|
|
"Bedsure": 25.00, "CGK": None, "cheapest": None},
|
|
{"sku": "UBTESTLINEFULLSAGE", "asin": "B0OURS0004", "size": "FULL",
|
|
"colour": "Sage", "our_price": 26.00, "buybox": "Buy Box",
|
|
"Bedsure": 20.00, "CGK": None, "cheapest": "Bedsure",
|
|
"notes": "Bedsure: Buy Box SUPPRESSED — High price; their price is not currently "
|
|
"buyable, do not price against it"},
|
|
# Match quality. These two rows are IDENTICAL apart from the NEAR-colour note, so any
|
|
# difference in verdict is attributable to match quality alone.
|
|
{"sku": "UBTESTLINEKINGFUZZY", "asin": "B0OURS0005", "size": "KING",
|
|
"colour": "Purple", "our_price": 30.00, "buybox": "Buy Box",
|
|
"Bedsure": 24.00, "CGK": None, "cheapest": "Bedsure",
|
|
"notes": "Bedsure: our 'Purple' ≈ their '06 - Dusty Purple (No Comforter)' — NEAR "
|
|
"colour match, verify"},
|
|
{"sku": "UBTESTLINEKINGEXACT", "asin": "B0OURS0006", "size": "KING",
|
|
"colour": "Purple", "our_price": 30.00, "buybox": "Buy Box",
|
|
"Bedsure": 24.00, "CGK": None, "cheapest": "Bedsure"},
|
|
# One fuzzy rival AND one exact rival on the same row. The flag is written per rival, so
|
|
# the exact one must survive while the fuzzy one is dropped.
|
|
{"sku": "UBTESTLINEQUEENMIXED", "asin": "B0OURS0007", "size": "QUEEN",
|
|
"colour": "Grey", "our_price": 30.00, "buybox": "Buy Box",
|
|
"Bedsure": 20.00, "CGK": 27.00, "cheapest": "Bedsure",
|
|
"notes": "Bedsure: our 'Grey' ≈ their '08 - Dark Grey (No Comforter)' — NEAR colour "
|
|
"match, verify"},
|
|
]
|
|
|
|
|
|
def _sheet(tmp_path):
|
|
from pricing_agent.competitor_sheet import CompetitorSheet
|
|
return CompetitorSheet.load(_write_sheet(tmp_path, SHEET_ROWS))
|
|
|
|
|
|
# ---- match quality: a FUZZY row may inform a human, never move a price -------------------
|
|
# Real-data motivation: 18 of 56 rows in the live workbook carry a NEAR-colour rival, and 2 of
|
|
# the 6 material undercuts were being driven by one. UBMICROFIBERDUVETKINGPURPLE would have cut
|
|
# 15.6% off an unverified 'Purple' ~ 'Dusty Purple' pairing — and once the fuzzy match is
|
|
# excluded the EXACT rival turns out to be 5.6% more expensive than us, i.e. the undercut was
|
|
# entirely an artifact of the guess.
|
|
def test_a_fuzzy_match_cannot_trigger_the_undercut_rule(tmp_path):
|
|
s = _sheet(tmp_path)
|
|
row = s.rows["UBTESTLINEKINGFUZZY"]
|
|
assert row.fuzzy_rivals == {"Bedsure"}
|
|
assert row.rivals == {"Bedsure": 24.00} # still READ, for display
|
|
assert row.priceable_rivals == {} # but not priceable
|
|
|
|
st = s.state_for("UBTESTLINEKINGFUZZY", our_price=None)
|
|
assert st.competitor_min is None # a 20% gap that must not reach the cascade
|
|
assert st.undercut_pct is None
|
|
action, _rec, reasons = verdict(st, comp_match=24.00)
|
|
assert "COMPETITOR_UNDERCUT" not in reasons
|
|
assert (action, reasons) == verdict(None)[0::2]
|
|
|
|
|
|
def test_an_otherwise_identical_exact_match_does_trigger_it(tmp_path):
|
|
"""Same prices, same 20% gap — the ONLY difference is the NEAR-colour note."""
|
|
s = _sheet(tmp_path)
|
|
assert s.rows["UBTESTLINEKINGEXACT"].fuzzy_rivals == set()
|
|
st = s.state_for("UBTESTLINEKINGEXACT", our_price=None)
|
|
assert st.competitor_min == 24.00
|
|
assert st.undercut_pct == pytest.approx(0.20)
|
|
action, rec, reasons = verdict(st, comp_match=24.00)
|
|
assert (action, rec, reasons) == ("Decrease", "comp_match", ["COMPETITOR_UNDERCUT"])
|
|
|
|
|
|
def test_fuzzy_exclusion_is_per_rival_not_per_row(tmp_path):
|
|
"""One row, one fuzzy rival, one exact rival: the exact one must survive.
|
|
|
|
The workbook writes the flag per rival ('Bedsure: our X ≈ their Y'), so dropping the whole
|
|
row would discard a perfectly good exact comparison — and keeping the whole row would price
|
|
us against the guess.
|
|
"""
|
|
s = _sheet(tmp_path)
|
|
row = s.rows["UBTESTLINEQUEENMIXED"]
|
|
assert row.fuzzy_rivals == {"Bedsure"}
|
|
assert row.priceable_rivals == {"CGK": 27.00} # Bedsure $20 dropped, CGK $27 kept
|
|
st = s.state_for("UBTESTLINEQUEENMIXED", our_price=None)
|
|
assert st.competitor_min == 27.00 # NOT 20.00
|
|
assert st.undercut_pct == pytest.approx(0.10)
|
|
# It still fires — off the exact rival, at the exact rival's price.
|
|
assert verdict(st, comp_match=27.00)[2] == ["COMPETITOR_UNDERCUT"]
|
|
|
|
|
|
def test_an_excluded_fuzzy_rival_is_still_reported_with_its_price(tmp_path):
|
|
"""The one exclusion a human may want to overrule, so give them the number to act on."""
|
|
s = _sheet(tmp_path)
|
|
st = s.state_for("UBTESTLINEKINGFUZZY", our_price=None)
|
|
note = next(n for n in st.notes if "NEAR colour match" in n and "excluded" in n)
|
|
assert "Bedsure $24.00" in note
|
|
assert "may not move a price on its own" in note
|
|
|
|
|
|
def test_sheet_parsing_rejects_the_footnote_rows(tmp_path):
|
|
"""The footnotes live in column A, so 'Our SKU' alone is not trustworthy.
|
|
|
|
Measured on the real workbook: a naive read yields 62 'SKUs', 6 of which are footnote
|
|
prose ('Our Price is scra…'). A real row carries a real ASIN.
|
|
"""
|
|
s = _sheet(tmp_path)
|
|
# Derived from the fixture, not hardcoded: the point is that the 6 footnote rows written
|
|
# under the table are rejected, whatever the data-row count happens to be.
|
|
assert len(s.rows) == len(SHEET_ROWS), sorted(s.rows)
|
|
assert all(r.asin and r.asin.startswith("B0") for r in s.rows.values())
|
|
assert not any("Our Price is" in k for k in s.rows)
|
|
|
|
|
|
def test_sheet_reads_rivals_from_the_header_not_a_hardcoded_list(tmp_path):
|
|
s = _sheet(tmp_path)
|
|
assert s.brands == ["Bedsure", "CGK"]
|
|
assert s.line == "UBTESTLINE"
|
|
|
|
|
|
def test_sheet_blank_our_price_stays_none(tmp_path):
|
|
"""A suppressed row legitimately has no price. That is a fact, not a parse failure."""
|
|
s = _sheet(tmp_path)
|
|
assert s.rows["UBTESTLINETWINPINK"].our_price is None
|
|
assert s.rows["UBTESTLINEQUEENWHITE"].our_price == 28.99
|
|
|
|
|
|
def test_coverage_gate_returns_na_for_an_uncovered_sku(tmp_path):
|
|
"""The point of the gate: 'not in our sheet' must not read as 'has no competitors'."""
|
|
s = _sheet(tmp_path)
|
|
assert s.covers("UBTESTLINEQUEENWHITE")
|
|
assert not s.covers("UBMICROFIBERGUSSETPILLOWWHITEQUEEN")
|
|
|
|
st = s.state_for("UBMICROFIBERGUSSETPILLOWWHITEQUEEN", our_price=26.07)
|
|
assert not st.usable
|
|
assert st.unusable_because.startswith("N/A")
|
|
# The message has to say what the sheet DOES cover, or a reader cannot tell a coverage
|
|
# hole from a market fact.
|
|
assert "UBTESTLINE" in st.unusable_because
|
|
assert f"{len(SHEET_ROWS)} SKU(s)" in st.unusable_because
|
|
# ...and it changes no verdict.
|
|
assert verdict(st) == verdict(None)
|
|
|
|
|
|
def test_covered_sku_yields_a_like_for_like_sheet_state(tmp_path):
|
|
from pricing_agent.competitive_state import BASIS_SHEET, SOURCE_SHEET
|
|
|
|
s = _sheet(tmp_path)
|
|
st = s.state_for("UBTESTLINEQUEENWHITE", our_price=None)
|
|
assert st.usable and st.source == SOURCE_SHEET and st.basis == BASIS_SHEET
|
|
assert st.our_price == 28.99
|
|
assert st.competitor_min == 24.99 # the cheaper of Bedsure 24.99 / CGK 31.00
|
|
assert st.rivals == 2
|
|
assert st.undercut_pct == pytest.approx(0.1379, abs=1e-3)
|
|
|
|
|
|
def test_sheet_excludes_a_rival_whose_own_buybox_is_suppressed(tmp_path):
|
|
"""Their price is not buyable, so undercutting it donates margin for nothing."""
|
|
s = _sheet(tmp_path)
|
|
row = s.rows["UBTESTLINEFULLSAGE"]
|
|
assert row.suppressed_rivals == {"Bedsure"}
|
|
st = s.state_for("UBTESTLINEFULLSAGE", our_price=None)
|
|
assert st.competitor_min is None # the only rival was excluded
|
|
assert st.rivals == 0
|
|
assert any("not buyable" in n for n in st.notes)
|
|
# The state stays USABLE — 'we hold the Buy Box' is a real, known fact worth reporting.
|
|
# What must not happen is a price move: with no buyable rival there is nothing to undercut,
|
|
# so the rule cannot fire even though a $20.00 figure sits in the sheet.
|
|
assert st.usable and st.status is BuyBoxStatus.WON
|
|
assert st.undercut_pct is None
|
|
assert verdict(st, comp_match=20.00) == verdict(None)
|
|
|
|
|
|
def test_sheet_suppression_drives_the_suppressed_rule(tmp_path):
|
|
s = _sheet(tmp_path)
|
|
st = s.state_for("UBTESTLINETWINPINK", our_price=25.00)
|
|
assert st.status is BuyBoxStatus.SUPPRESSED and st.usable
|
|
assert verdict(st)[2] == ["BUYBOX_SUPPRESSED"]
|
|
|
|
|
|
def test_stale_sheet_is_na_and_names_the_line_to_re_run(tmp_path):
|
|
s = _sheet(tmp_path)
|
|
s.as_of = NOW - timedelta(hours=400)
|
|
st = s.state_for("UBTESTLINEQUEENWHITE", our_price=None, max_age_hours=168.0, now=NOW)
|
|
assert not st.usable
|
|
assert "400h old" in st.unusable_because
|
|
assert "UBTESTLINE" in st.unusable_because
|
|
|
|
|
|
def test_sheet_basis_undercut_fires_without_a_buybox_loss(tmp_path):
|
|
"""A rival BRAND's cheaper product is a real signal even while we hold our own Buy Box.
|
|
|
|
This is the difference the `basis` field exists to record: it is NOT a Buy Box loss, and
|
|
must never be reported as one.
|
|
"""
|
|
from pricing_agent.competitive_state import BASIS_SHEET
|
|
|
|
s = _sheet(tmp_path)
|
|
st = s.state_for("UBTESTLINEQUEENWHITE", our_price=None)
|
|
assert st.status is BuyBoxStatus.WON # we hold it, and it still fires
|
|
assert st.basis == BASIS_SHEET
|
|
action, rec, reasons = verdict(st, comp_match=24.99)
|
|
assert (action, rec, reasons) == ("Decrease", "comp_match", ["COMPETITOR_UNDERCUT"])
|
|
|
|
|
|
def test_sheet_basis_respects_the_same_material_threshold(tmp_path):
|
|
"""KING row: rival 29.70 vs our 30.00 = 1% — inside coupon noise, so nothing fires."""
|
|
s = _sheet(tmp_path)
|
|
st = s.state_for("UBTESTLINEKINGWHITE", our_price=None)
|
|
assert st.undercut_pct == pytest.approx(0.01)
|
|
assert "COMPETITOR_UNDERCUT" not in verdict(st, comp_match=29.70)[2]
|
|
|
|
|
|
def test_a_same_asin_state_still_needs_a_buybox_loss():
|
|
"""The two bases are not interchangeable — WON on the SAME ASIN must not fire."""
|
|
won = state(BuyBoxStatus.WON, rival=22.0)
|
|
assert "COMPETITOR_UNDERCUT" not in verdict(won, comp_match=22.0)[2]
|
|
|
|
|
|
# ==================================== 10. A DROP WITH A COMPETITOR EXPLANATION
|
|
def _dropping():
|
|
"""A SKU whose velocity halved, with no price change and no ad collapse of its own."""
|
|
return FakeResult(avg_30d=4.0, avg_6m=10.0)
|
|
|
|
|
|
def test_a_drop_with_no_competitor_cause_is_still_unexplained():
|
|
assert verdict(None, r=_dropping())[2] == ["UNEXPLAINED_DROP", "VELOCITY_DROP"]
|
|
|
|
|
|
def test_a_material_undercut_explains_a_drop_instead_of_filing_it_as_a_mystery():
|
|
"""Observed on real SKUs: a rival 15.6% / 35.2% below us, filed UNEXPLAINED_DROP.
|
|
|
|
Calling a drop unexplained while holding the explanation sends someone off to rediscover
|
|
a rival price already printed in the root cause. Same precedent as `stock_explained`.
|
|
"""
|
|
r = _dropping()
|
|
comp = state(BuyBoxStatus.LOST_PRICE, rival=22.0)
|
|
_a, _rec, reasons = verdict(comp, r=r, comp_match=22.0)
|
|
assert "UNEXPLAINED_DROP" not in reasons
|
|
assert reasons == ["COMPETITOR_UNDERCUT", "VELOCITY_DROP"]
|
|
|
|
|
|
def test_an_immaterial_rival_does_not_explain_a_drop():
|
|
"""Only a MATERIAL undercut is an explanation; 1% is not."""
|
|
r = _dropping()
|
|
comp = state(BuyBoxStatus.LOST_PRICE, rival=24.70)
|
|
assert verdict(comp, r=r, comp_match=24.70)[2] == ["UNEXPLAINED_DROP", "VELOCITY_DROP"]
|
|
|
|
|
|
def test_a_stockout_still_explains_a_drop_before_a_competitor_does():
|
|
"""The existing stockout branch keeps priority — it is the cheaper explanation to act on."""
|
|
r = _dropping()
|
|
hist_oos = hist()
|
|
hist_oos.loc[hist_oos.index[-25:], "units"] = 0.0
|
|
hist_oos.loc[hist_oos.index[-25:], "inventory"] = 0.0
|
|
comp = state(BuyBoxStatus.LOST_PRICE, rival=22.0)
|
|
reasons = _decide(r, CUR, hist_oos, FP, scen(22.0), None, comp)[4]
|
|
assert reasons[0] == "STOCKOUT_BEFORE_ARRIVAL"
|
|
|
|
|
|
def test_one_source_missing_is_not_a_disagreement():
|
|
apify = state(BuyBoxStatus.WON, rival=22.0)
|
|
assert reconcile(apify, None, sku="X").notes == []
|
|
assert reconcile(None, apify, sku="X").source == "apify"
|
|
assert not reconcile(None, None, sku="X").usable
|