"""Verdict backtest: what do the competitor rules actually change, on real SKUs? Same shape as the storage-fix backtest (61.9% -> 45.2%), but the quantity under test is not an error percentage -- it is the VERDICT. So this reports, per SKU, the action and the reason code with the competitor rules blind vs. live, and names every SKU that moved and why. Method ------ The real pipeline runs ONCE per SKU. `dashboard.live_data._decide` is wrapped so the exact arguments it was handed in production -- the real AnalysisResult, the real 180-day history, the real fee stack, the real scenario grid, the real inventory outlook -- are captured. Every arm below then re-runs that same captured input through the same cascade, varying ONLY the competitor state. Nothing is reconstructed by hand, so no arm can drift from what the live dashboard would actually compute. Four arms: BLIND comp=None -- the behaviour before this change REAL the actual competitor state -- what production would do right now CF_UNDERCUT counterfactual: a rival holds the Buy Box `--undercut` below us CF_SUPPRESSED counterfactual: our Buy Box is suppressed The REAL arm is built from the Apify data ALREADY ON DISK (`data/apify_cache.json`) rather than by scraping. The actor is pay-per-run, and a backtest is not a reason to spend: the cached payloads are real responses for real ASINs, and reading them exercises the same `from_apify` adapter production uses. Pass `--scrape` to fetch fresh data instead, which does cost money -- it is opt-in for that reason, never the default. The counterfactuals exist because the real arm can only exercise the states our data happens to be in today. They are labelled as counterfactual everywhere and are never presented as observed fact. python scripts/backtest_competitor_rules.py --skus SKU1,SKU2 --undercut 0.08 """ from __future__ import annotations import argparse import sys from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] for p in (ROOT / "src", ROOT): if str(p) not in sys.path: sys.path.insert(0, str(p)) import dashboard.live_data as L # noqa: E402 from pricing_agent.competitive_state import CompetitiveState # noqa: E402 from pricing_agent.schemas import BuyBoxStatus # noqa: E402 NOW = datetime.now(timezone.utc) # Defaults: the two ASINs with real cached competitive data, plus a slice of the duvet line # the workbook tool was last run against. DEFAULT_SKUS = [ "UBMICROFIBERGUSSETPILLOWWHITEQUEEN", "UBCFKFITTEDSHEETWHITECALKING", "UBMICROFIBERDUVETQUEENWHITE", ] def capture_decide(): """Wrap _decide so we keep the real arguments it was called with.""" calls: list[dict] = [] original = L._decide def spy(r, cur_price, hist, fp, scen, outlook=None, comp=None): calls.append({"r": r, "cur_price": cur_price, "hist": hist, "fp": fp, "scen": scen, "outlook": outlook, "comp": comp}) return original(r, cur_price, hist, fp, scen, outlook, comp) L._decide = spy return calls, original def with_comp_match(call: dict, rival: float | None): """The captured scenario grid, plus the `comp_match` candidate build_live_sku would add. Without this the counterfactual arms are untestable: the undercut branch requires a priced `comp_match` candidate to move toward, and build_live_sku only adds one when competitor state was usable AT PIPELINE TIME. Reusing the captured grid therefore holds the rule permanently off and the arm would report "no change" for the wrong reason. The row is computed the same way `_scenarios` computes every other row -- same elasticity, same fee stack -- so the candidate is priced consistently with its neighbours rather than pasted in. """ scen = call["scen"] if not rival or rival <= 0: return scen cur = call["cur_price"] row_cur = scen.set_index("scenario").loc["current"] el = ((call["r"].elasticity or {}).get("elasticity")) or L.FALLBACK_ELASTICITY units_day = float(row_cur["units_day"]) inventory = float(row_cur["cover_days"]) * max(units_day, 0.1) units = units_day * (rival / cur) ** el if cur > 0 else units_day th = L._take_home(rival, call["fp"]) extra = { "scenario": "comp_match", "price": round(rival, 2), "units_day": round(units, 1), "revenue_30d": round(units * 30 * rival), "profit_30d": round(units * 30 * th), "margin_pct": round(th / rival * 100, 1) if rival else 0.0, "cover_days": round(inventory / max(units, 0.1)), } import pandas as pd return pd.concat([scen, pd.DataFrame([extra])], ignore_index=True) def rerun(call: dict, comp, *, rival: float | None = None) -> tuple[str, str, str, float]: """Re-run the captured input through the real cascade with a different comp state. Returns the SHIPPABLE price too, i.e. after the same guardrails build_live_sku applies: lifted to the safe floor, then capped at one MAX_STEP_PCT move. Reporting the raw target would let a "match the rival at $8" recommendation look like it shipped when the floor would in fact have stopped it. """ scen = with_comp_match(call, rival) if rival else call["scen"] action, rec, _cons, _aggr, reasons, _root, objective = L._decide( call["r"], call["cur_price"], call["hist"], call["fp"], scen, call["outlook"], comp, ) cur = call["cur_price"] target = float(scen.set_index("scenario").loc[rec, "price"]) floor = L.price_floor(call["r"], cur).get("floor") or 0.0 if action == "Investigate": shipped = cur else: shipped = max(target, floor) if floor else target step = cur * L.MAX_STEP_PCT if abs(shipped - cur) > step + 0.005: shipped = cur + (step if shipped > cur else -step) return action, ",".join(reasons), objective, round(shipped, 2) def state_from_disk(asin: str | None, our_price: float): """The real competitor state for this ASIN from the Apify cache — no network, no spend. Uses the same `item_to_competitive` -> `from_apify` path production uses, so the state is gated (age, unknown ownership) exactly as it would be live. Returns None when the cache holds nothing for this ASIN, which is itself a real and common condition. """ import json from config.settings import get_rules, get_settings from pricing_agent.competitive_state import from_apify from pricing_agent.tools.amazon.apify import item_to_competitive s, rules = get_settings(), get_rules() try: raw = json.loads(Path(s.apify_cache_path).read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None entry = raw.get((asin or "").upper()) if not isinstance(entry, dict) or "item" not in entry: return None snap = item_to_competitive(entry["item"], asin=asin, our_price=our_price, our_seller_id=s.apify_seller_id) return from_apify( snap, our_price=our_price, as_of=datetime.fromtimestamp(float(entry["ts"]), tz=timezone.utc), max_age_hours=rules.competitor_state_max_age_hours, now=NOW) def cf(status: BuyBoxStatus, our_price: float, rival: float | None) -> CompetitiveState: """A COUNTERFACTUAL state — fresh by construction, so the age gate cannot mask the rule.""" return CompetitiveState( status=status, our_price=our_price, buy_box_price=rival, competitor_min=rival, competitor_median=rival, rivals=1 if rival else 0, source="counterfactual", as_of=NOW, reason="counterfactual injected by the verdict backtest", ) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--skus", default=",".join(DEFAULT_SKUS)) ap.add_argument("--undercut", type=float, default=0.08, help="counterfactual rival price, as a fraction below ours") ap.add_argument("--scrape", action="store_true", help="fetch FRESH competitor data (COSTS MONEY — the Apify actor is " "pay-per-run). Off by default; the cached payloads on disk are real.") args = ap.parse_args() skus = tuple(s.strip() for s in args.skus.split(",") if s.strip()) calls, original = capture_decide() try: print(f"running the real pipeline for {len(skus)} SKU(s) — this fetches live COSMOS " f"data and may take a minute each") print("competitor data: " + ("FRESH SCRAPE (paid)" if args.scrape else "read from data/apify_cache.json (free)") + "\n") data = L.get_live_data(skus, with_competitive=args.scrape) finally: L._decide = original details, errors = data.get("details") or {}, data.get("errors") or {} for sku, err in errors.items(): print(f"!! {sku}: {err}") # Pair each captured call with its SKU by current price — build_live_sku calls _decide # exactly once per SKU. by_price = {round(c["cur_price"], 4): c for c in calls} rows = [] for sku in skus: d = details.get(sku) if not d: continue cur = round(float(d.get("current_price") or 0), 4) call = by_price.get(cur) if call is None: print(f"!! {sku}: could not pair a captured cascade call (price {cur})") continue # With --scrape the pipeline already built a real state; otherwise read the real # cached payload off disk. Either way the REAL arm is real Apify data. comp = call["comp"] if not args.scrape or comp is None or not comp.usable: from_disk = state_from_disk(d.get("asin"), cur) if from_disk is not None: comp = from_disk rival = round(cur * (1 - args.undercut), 2) arms = { "BLIND": rerun(call, None), "REAL": rerun(call, comp), "CF_UNDERCUT": rerun(call, cf(BuyBoxStatus.LOST_PRICE, cur, rival), rival=rival), "CF_SUPPRESSED": rerun(call, cf(BuyBoxStatus.SUPPRESSED, cur, None)), } rows.append((sku, cur, comp, arms)) print("\n" + "=" * 100) print("VERDICT BACKTEST — competitor rules blind vs live, on real SKUs") print("=" * 100) for sku, cur, comp, arms in rows: print(f"\n{sku} current ${cur:.2f}") st = (f"{comp.status.value} via {comp.source}" if comp is not None else "none") usable = "usable" if (comp is not None and comp.usable) else ( f"NOT usable — {comp.unusable_because}" if comp is not None else "n/a") print(f" real competitor state : {st}") print(f" : {usable}") if comp is not None and comp.competitor_min: print(f" : cheapest rival ${comp.competitor_min:.2f} " f"({comp.rivals} rival offer(s))") floor = L.price_floor(call["r"], cur).get("floor") or 0.0 print(f" break-even floor : ${floor:.2f}" f" (counterfactual rival ${rival:.2f})") for arm, (action, reasons, obj, shipped) in arms.items(): moved = "" if arms[arm] == arms["BLIND"] else " <<< CHANGED" tag = " (counterfactual)" if arm.startswith("CF_") else "" flag = "" if floor and shipped < floor - 0.005: flag = " *** BELOW FLOOR — INVARIANT VIOLATED ***" print(f" {arm:14} {action:12} ${shipped:>7.2f} {reasons:42} " f"{obj}{moved}{tag}{flag}") print("\n" + "=" * 100) real_changed = [s for s, _c, _st, a in rows if a["REAL"] != a["BLIND"]] cf_u_changed = [s for s, _c, _st, a in rows if a["CF_UNDERCUT"] != a["BLIND"]] cf_s_changed = [s for s, _c, _st, a in rows if a["CF_SUPPRESSED"] != a["BLIND"]] print(f"SKUs analysed : {len(rows)}") print(f"Verdicts changed by REAL competitor data : {len(real_changed)} " f"{real_changed}") print(f"Verdicts that WOULD change on a {args.undercut:.0%} undercut : " f"{len(cf_u_changed)} {cf_u_changed}") print(f"Verdicts that WOULD change if suppressed : {len(cf_s_changed)} " f"{cf_s_changed}") # The invariant that matters most: no arm, however cheap the counterfactual rival, may # ship a price under break-even. violations = [] for sku, cur, _comp, arms in rows: floor = L.price_floor(by_price[round(cur, 4)]["r"], cur).get("floor") or 0.0 if not floor: continue for arm, (_action, _reasons, _obj, shipped) in arms.items(): if shipped < floor - 0.005: violations.append(f"{sku}/{arm} ${shipped:.2f} < ${floor:.2f}") print(f"Prices shipped BELOW break-even (any arm) : {len(violations)} {violations}") print("\nA REAL delta of 0 is the fail-safe working, not the feature missing: with no " "usable\ncompetitor state every verdict is byte-identical to the competitor-blind " "one.\nThe counterfactual arms show the rules do fire once state IS usable.") return 0 if __name__ == "__main__": raise SystemExit(main())