1922 lines
100 KiB
Python
1922 lines
100 KiB
Python
"""Live COSMOS adapter — builds the exact data structure the dashboard renders.
|
||
|
||
Produces a ``{"summary": DataFrame, "details": {sku: {...}}}`` structure where
|
||
every number comes from the real pipeline:
|
||
|
||
* fees / costs / break-even . takehome-calculator (via analyze_price)
|
||
* sales trend + inventory .... invp-insight
|
||
* 6-month daily history ...... sales-insight (price, units, ad spend, profit)
|
||
* elasticity + optimal price . pricing_agent.elasticity (log-log OLS fit)
|
||
* competitors (optional) ..... Apify Buy Box scrape when APIFY_TOKEN is set
|
||
|
||
Fields the real APIs cannot provide are explicitly None (the UI shows "—"):
|
||
ad-attributed sales / ACoS (no ads API), purchase orders, competitor history.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import math
|
||
from datetime import date, datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
logger = logging.getLogger("pricing_agent.dashboard")
|
||
|
||
HISTORY_DAYS = 180
|
||
FORECAST_DAYS = 30
|
||
DOW = np.array([1.00, 0.94, 0.90, 0.93, 1.02, 1.22, 1.18]) # Mon..Sun
|
||
FALLBACK_ELASTICITY = -1.3
|
||
REFERRAL_PCT = 0.15 # fallback referral fraction when a detail lacks fee params
|
||
VARIABLE = 0.55 # fallback per-unit variable expense
|
||
|
||
# --- safety policy -----------------------------------------------------------
|
||
# One price move may not exceed this. The UI has always claimed a 5% cap (bulk
|
||
# approve, the Modify warning); now the engine enforces it, so a "destination"
|
||
# price is reached in steps that can each be read before the next one.
|
||
MAX_STEP_PCT = 0.05
|
||
# A recommendation may not exceed the highest price with a real sample behind it
|
||
# by more than this. Beyond it there is no evidence, only extrapolation.
|
||
OBSERVED_HEADROOM = 1.05
|
||
# Margin held above the true (ad-inclusive) break-even by the guardrail floor.
|
||
FLOOR_SAFETY = 1.02
|
||
# Default averaging window. 30 days is one month of noise; 90 spans enough price
|
||
# variation for the current-price baseline to be representative.
|
||
DEFAULT_WINDOW_DAYS = 90
|
||
# Inventory thresholds, shared with the dashboard tiles so the two can never
|
||
# disagree. Both are inclusive: a SKU sitting exactly on the line counts as at
|
||
# risk, because a strict `<` once filed a SKU with 26 days of on-hand cover as
|
||
# "no stockout risk" on the strength of a projection that read 35.
|
||
#
|
||
# Read from pricing_rules.yaml because they are policy, not code — and because they have to be
|
||
# reconcilable with the COSMOS Inventory Planning bands that colour the same numbers on screen.
|
||
# The two are deliberately different: COSMOS's pink "high" band at 70 days is a REPLENISHMENT
|
||
# warning, while crossing `HIGH_COVER_DAYS` here spends real margin on a 5% discount. See the
|
||
# measured impact table in pricing_rules.yaml.
|
||
def _cover_rules() -> tuple[int, int]:
|
||
try:
|
||
from config.settings import get_rules
|
||
r = get_rules()
|
||
return int(r.low_cover_days), int(r.high_cover_days)
|
||
except Exception: # noqa: BLE001 — config must never stop the module importing
|
||
return 35, 90
|
||
|
||
|
||
LOW_COVER_DAYS, HIGH_COVER_DAYS = _cover_rules()
|
||
|
||
REASON_LABELS = {
|
||
"EXCESS_STOCK": "Excess inventory",
|
||
"INBOUND_PO": "Incoming stock",
|
||
"LOW_STOCK": "Low stock",
|
||
"STOCKOUT_BEFORE_ARRIVAL": "Stockout before PO arrival",
|
||
"COMPETITOR_PROMO": "Competitor promotion",
|
||
"VELOCITY_DROP": "Weak sales velocity",
|
||
"PREMIUM_HEADROOM": "Premium headroom",
|
||
"BELOW_BREAK_EVEN": "Selling below break-even",
|
||
"SEASONAL_LOW": "Off-season demand",
|
||
"NEW_LAUNCH": "New launch ramp",
|
||
"COMPETITOR_UNDERCUT": "Competitor undercut",
|
||
"NO_SIGNALS": "Stable — no action needed",
|
||
"UNEXPLAINED_DROP": "Unexplained sales drop",
|
||
"NO_COST_DATA": "Missing cost data (COSMOS)",
|
||
"PROFIT_OPTIMAL": "Profit-optimal price differs",
|
||
"LOSING_MONEY": "Losing money after ads",
|
||
"BEST_OBSERVED": "Better price already proven",
|
||
"AD_SPIRAL": "Ad cost scales with price — price is not the lever",
|
||
"BUYBOX_SUPPRESSED": "Buy Box suppressed — sells nothing at any price",
|
||
"COMPETITOR_PREMIUM": "Priced above the rival field",
|
||
}
|
||
|
||
|
||
def _sold_badge(units_month: float) -> str:
|
||
for floor, label in [(2000, "2K+"), (1000, "1K+"), (500, "500+"), (300, "300+"),
|
||
(100, "100+"), (50, "50+")]:
|
||
if units_month >= floor:
|
||
return f"{label} sold in last month"
|
||
return "—"
|
||
|
||
CONF_BAND = {"High": 0.18, "Medium": 0.28, "Low": 0.40, "Insufficient": 0.50}
|
||
|
||
RISKS = {
|
||
"Increase": [("Volume may drop more than the elasticity model expects",
|
||
"Small step; re-evaluate after 14 days"),
|
||
("Raising price can slow organic ranking",
|
||
"Monitor sessions/share after the change")],
|
||
"Decrease": [("Demand may not respond at the modeled elasticity",
|
||
"Small step; re-evaluate after 14 days"),
|
||
("Margin given up is certain, extra volume is not",
|
||
"Floor at min-safe; weekly checkpoint")],
|
||
"Maintain": [("Situation can drift while holding",
|
||
"Re-evaluated automatically on next refresh")],
|
||
"Investigate": [("No confirmed cause — a price cut could be the wrong lever",
|
||
"Human review; check listing health & traffic manually")],
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------- fee model
|
||
def _fee_params(fees: dict, price: float) -> dict:
|
||
"""Per-unit fee model parameters from the raw takehome-calculator response."""
|
||
if not fees or price <= 0:
|
||
return {"referral_pct": 0.15, "returns_pct": 0.0, "cost": 0.0, "fba": 0.0,
|
||
"variable": 0.0}
|
||
other = sum(float(fees.get(k) or 0.0) for k in (
|
||
"lowInventoryFee", "inboundPlacementFee", "eprFee", "vat",
|
||
"orderHandling", "weightHandling", "warehouseExpense"))
|
||
cost = float(fees.get("costPerUnit") or 0.0)
|
||
if cost <= 0 and isinstance(fees.get("costBreakdown"), dict):
|
||
cb = fees["costBreakdown"]
|
||
cost = sum(float(cb.get(k) or 0.0) for k in ("purchasePrice", "freight", "duty"))
|
||
return {
|
||
"referral_pct": float(fees.get("referralFee") or 0.0) / price,
|
||
"returns_pct": float(fees.get("returnFee") or 0.0) / price,
|
||
"cost": cost,
|
||
"fba": float(fees.get("fbaFee") or 0.0),
|
||
"variable": other,
|
||
}
|
||
|
||
|
||
def _take_home(price: float, fp: dict) -> float:
|
||
return price * (1 - fp["referral_pct"] - fp["returns_pct"]) - fp["cost"] - fp["fba"] - fp["variable"]
|
||
|
||
|
||
# ---------------------------------------------------------------- history
|
||
def _hist_frame(history, current_price: float, today: date) -> pd.DataFrame:
|
||
"""Daily SalesDay points → the chart frame (full 180-day daily index)."""
|
||
idx = pd.date_range(pd.Timestamp(today) - pd.Timedelta(days=HISTORY_DAYS - 1),
|
||
periods=HISTORY_DAYS)
|
||
rows = {}
|
||
for p in history or []:
|
||
try:
|
||
dt = pd.Timestamp(f"{p.date[6:]}-{p.date[:2]}-{p.date[3:5]}")
|
||
except Exception:
|
||
continue
|
||
rows[dt] = {"price": p.sale_price, "units": p.units or 0.0,
|
||
"ad_spend": (p.marketing_cost or 0.0) + (p.promotion_spend or 0.0),
|
||
"revenue": p.revenue or 0.0, "profit": p.profit or 0.0,
|
||
# Daily inventory — the ONLY historical stock series we get, and
|
||
# what lets a stockout be told apart from a collapse in demand.
|
||
"inventory": p.inventory}
|
||
df = pd.DataFrame.from_dict(rows, orient="index").reindex(idx)
|
||
if df.empty or "price" not in df:
|
||
df = pd.DataFrame(index=idx, columns=["price", "units", "ad_spend",
|
||
"revenue", "profit", "inventory"],
|
||
dtype=float)
|
||
df["units"] = df["units"].fillna(0.0)
|
||
df["ad_spend"] = df["ad_spend"].fillna(0.0)
|
||
df["revenue"] = df["revenue"].fillna(0.0)
|
||
df["profit"] = df["profit"].fillna(0.0)
|
||
df["price"] = df["price"].ffill().bfill().fillna(current_price)
|
||
# inventory is deliberately NOT filled: a missing reading is "unknown", not
|
||
# "empty shelf". Filling it with 0 would manufacture phantom stockouts.
|
||
df["comp_median"] = np.nan # no historical competitor series in COSMOS
|
||
out = df.reset_index().rename(columns={"index": "date"})
|
||
return out[["date", "price", "comp_median", "units", "ad_spend", "revenue",
|
||
"profit", "inventory"]]
|
||
|
||
|
||
def _change_dates(hist: pd.DataFrame) -> list:
|
||
"""Dates where our price moved ≥1% vs the previous day (max 8 markers)."""
|
||
p = hist["price"].to_numpy(dtype=float)
|
||
out = []
|
||
for i in range(1, len(p)):
|
||
if p[i - 1] > 0 and abs(p[i] - p[i - 1]) / p[i - 1] >= 0.01:
|
||
out.append(hist["date"].iloc[i])
|
||
return out[-8:]
|
||
|
||
|
||
def _forecast(hist: pd.DataFrame, tier: str, today: date) -> pd.DataFrame:
|
||
recent = float(hist["units"].tail(28).mean()) if len(hist) else 0.0
|
||
dates = pd.date_range(pd.Timestamp(today) + pd.Timedelta(days=1), periods=FORECAST_DAYS)
|
||
p50 = recent * DOW[[d.weekday() for d in dates]]
|
||
band = CONF_BAND.get(tier, 0.4)
|
||
return pd.DataFrame({"date": dates, "p50": np.round(p50, 1),
|
||
"p10": np.round(p50 * (1 - band), 1),
|
||
"p90": np.round(p50 * (1 + band), 1)})
|
||
|
||
|
||
# ---------------------------------------------------------------- inventory outlook
|
||
# An arrival smaller than this many days of sales is noise, not relief.
|
||
MATERIAL_ARRIVAL_DAYS = 3
|
||
|
||
|
||
def _proj_date(value) -> date | None:
|
||
"""Parse a projection date — COSMOS mixes ISO `dataDate` with MM/DD/YYYY keys."""
|
||
s = str(value or "").strip()
|
||
for fmt in ("%Y-%m-%d", "%m/%d/%Y"):
|
||
try:
|
||
return datetime.strptime(s[:10], fmt).date()
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _inventory_outlook(projections: list, units_day: float,
|
||
today: date | None = None) -> dict:
|
||
"""When does this SKU run out, and when does the next stock actually land?
|
||
|
||
COSMOS already projects inventory forward in weekly steps and those snapshots
|
||
net off incoming arrivals, so the projection is the better answer whenever the
|
||
curve reaches zero inside the horizon. When it never does, we fall back to
|
||
on-hand ÷ velocity and SAY SO via `source` — a projected date and a
|
||
straight-line extrapolation are not the same claim and must not look alike.
|
||
|
||
`horizon_end` exists so "no stockout" can be reported as "not within N weeks"
|
||
rather than implying "never".
|
||
"""
|
||
today = today or date.today()
|
||
out = {"stockout_date": None, "stockout_source": None, "days_to_stockout": None,
|
||
"next_arrival_date": None, "next_arrival_units": 0.0,
|
||
"horizon_end": None, "horizon_days": None}
|
||
rows = [(d, p) for p in (projections or [])
|
||
if (d := _proj_date(p.get("date"))) is not None]
|
||
rows.sort(key=lambda dp: dp[0])
|
||
|
||
if rows:
|
||
out["horizon_end"] = rows[-1][0]
|
||
out["horizon_days"] = (rows[-1][0] - today).days
|
||
for d, p in rows:
|
||
if float(p.get("inventory") or 0) <= 0:
|
||
out["stockout_date"] = d
|
||
out["stockout_source"] = "projection"
|
||
break
|
||
# First arrival big enough to matter — a handful of units is not relief.
|
||
floor = max(units_day * MATERIAL_ARRIVAL_DAYS, 1.0)
|
||
for d, p in rows:
|
||
if float(p.get("warehouse_arrival") or 0) >= floor:
|
||
out["next_arrival_date"] = d
|
||
out["next_arrival_units"] = float(p["warehouse_arrival"])
|
||
break
|
||
|
||
# Projection never empties. Only extrapolate when there is NOTHING inbound —
|
||
# a straight line from on-hand ÷ velocity cannot see arrivals, so using it on
|
||
# a SKU with stock landing would invent a stockout the projection disproves.
|
||
# (Observed: a pillow whose projected inventory RISES 52k→63k on a 10,830-unit
|
||
# arrival was being reported as "out on 29 Aug".) With stock inbound, the
|
||
# honest answer is "not within the horizon".
|
||
if (out["stockout_date"] is None and units_day > 0 and rows
|
||
and not any(float(p.get("warehouse_arrival") or 0) > 0 for _, p in rows)):
|
||
on_hand = float(rows[0][1].get("inventory") or 0)
|
||
if on_hand > 0:
|
||
out["stockout_date"] = today + timedelta(days=int(on_hand / units_day))
|
||
out["stockout_source"] = "velocity"
|
||
|
||
if out["stockout_date"]:
|
||
out["days_to_stockout"] = (out["stockout_date"] - today).days
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- scenarios
|
||
def _scenarios(cur_price: float, units_day: float, inventory: float, fp: dict,
|
||
elasticity: float, comp_median: float | None,
|
||
extra_keys: dict | None = None) -> pd.DataFrame:
|
||
grid = {
|
||
"current": cur_price,
|
||
"up_2": round(cur_price * 1.02, 2), "up_5": round(cur_price * 1.05, 2),
|
||
"down_2": round(cur_price * 0.98, 2), "down_5": round(cur_price * 0.95, 2),
|
||
}
|
||
if comp_median and comp_median > 0:
|
||
grid["match_median"] = round(comp_median, 2)
|
||
for k, v in (extra_keys or {}).items():
|
||
if v and v > 0:
|
||
grid[k] = round(v, 2)
|
||
|
||
rows = []
|
||
for key, p in grid.items():
|
||
if p <= 0:
|
||
continue
|
||
units = units_day * (p / cur_price) ** elasticity if cur_price > 0 else units_day
|
||
th = _take_home(p, fp)
|
||
rows.append({
|
||
"scenario": key, "price": p, "units_day": round(units, 1),
|
||
"revenue_30d": round(units * 30 * p), "profit_30d": round(units * 30 * th),
|
||
"margin_pct": round(th / p * 100, 1) if p else 0.0,
|
||
"cover_days": round(inventory / max(units, 0.1)),
|
||
})
|
||
return pd.DataFrame(rows)
|
||
|
||
|
||
# ---------------------------------------------------------------- decision
|
||
def _order_ladder(action: str, rec_key: str, cons_key: str, aggr_key: str,
|
||
scen: pd.DataFrame, cur_price: float) -> tuple[str, str]:
|
||
"""Force Conservative ≤ Recommended ≤ Aggressive (mirrored for a decrease).
|
||
|
||
Each branch of `_decide` picks the three rungs independently, so a
|
||
recommendation landing on `max_profit` / `min_safe` could sit BEYOND the rung
|
||
labelled "aggressive" — the ladder then read Conservative $26.59 ·
|
||
Recommended $31.45 · Aggressive $27.37, i.e. the aggressive option was the
|
||
*least* aggressive of the three. Rebuild the two flanking rungs from the
|
||
candidate prices so the ladder always reads in order.
|
||
|
||
When no candidate is bolder than the recommendation, the aggressive rung
|
||
collapses onto it; the UI merges rungs that share a price into one row.
|
||
"""
|
||
if action not in ("Increase", "Decrease") or not len(scen):
|
||
return cons_key, aggr_key
|
||
prices = dict(zip(scen["scenario"], scen["price"]))
|
||
rec_p = prices.get(rec_key)
|
||
if rec_p is None:
|
||
return cons_key, aggr_key
|
||
sign = 1 if action == "Increase" else -1
|
||
# candidates that move in the action's direction, ordered mildest → boldest
|
||
side = sorted(((k, p) for k, p in prices.items() if sign * (p - cur_price) > 0),
|
||
key=lambda kp: sign * kp[1])
|
||
if not side:
|
||
return rec_key, rec_key
|
||
milder = [k for k, p in side if sign * (p - rec_p) <= 0]
|
||
bolder = [k for k, p in side if sign * (p - rec_p) >= 0]
|
||
return (milder[0] if milder else rec_key), (bolder[-1] if bolder else rec_key)
|
||
|
||
|
||
def price_floor(r, cur_price: float) -> dict:
|
||
"""The lowest price this SKU may be set to, and why.
|
||
|
||
The fee-stack break-even carries no advertising, so it sits well below the
|
||
price at which a unit actually stops losing money. Four floors are computed
|
||
and the highest wins:
|
||
|
||
accounting fees + COGS only (what the API returns)
|
||
with_ads + advertising held flat (fixed + ad) / margin_rate
|
||
ad_curve + advertising rising with price (fixed + a) / (margin − slope)
|
||
empirical fitted to ACTUAL booked profit (everything COSMOS books)
|
||
"""
|
||
floors = {"accounting": r.break_even or 0.0,
|
||
"with_ads": r.break_even_with_ads or 0.0,
|
||
"empirical": r.break_even_empirical or 0.0}
|
||
# The ad-curve break-even only acts as a floor when it is actually reachable.
|
||
# When it isn't, it is a diagnosis (price can't fix this SKU), and using it
|
||
# would put a price 4x anything ever charged on screen as a "safe floor".
|
||
if r.break_even_ad_curve and not r.ad_curve_unrecoverable:
|
||
floors["ad_curve"] = r.break_even_ad_curve
|
||
binding = max(floors, key=lambda k: floors[k])
|
||
value = floors[binding]
|
||
return {"floors": floors, "binding": binding,
|
||
"floor": round(value * FLOOR_SAFETY, 2) if value else 0.0,
|
||
"unrecoverable": bool(r.ad_curve_unrecoverable),
|
||
"contribution_per_dollar": r.contribution_per_dollar}
|
||
|
||
|
||
MATERIAL_GAIN = 1.10 # observed profit must beat today's by this much to act
|
||
|
||
# A day counts as "effectively out of stock" below this many days of cover.
|
||
# NOT zero: Amazon's feed never reports a clean 0 — the audited SKU had 90/90 days
|
||
# of inventory readings and not one at zero — so an `inventory <= 0` test would
|
||
# never fire. One day of cover is deliberately conservative; no SKU examined so far
|
||
# contains an in-sample stockout to calibrate against, so this is a judgement call
|
||
# and should be revisited once one does.
|
||
STOCKOUT_COVER_DAYS = 1.0
|
||
# Share of the drop window that must be stock-constrained before a velocity drop
|
||
# is explained by supply rather than demand.
|
||
STOCKOUT_SHARE = 0.20
|
||
|
||
|
||
def _stockout_days(hist: pd.DataFrame, window: int = 30) -> tuple[int, int]:
|
||
"""(days effectively out of stock, days with a reading) over the last `window`.
|
||
|
||
Returns the denominator too, because "0 of 0" and "0 of 30" are very different
|
||
claims and the caller must not treat a data gap as proof of healthy stock.
|
||
"""
|
||
if "inventory" not in hist or not len(hist):
|
||
return 0, 0
|
||
tail = hist.tail(window)
|
||
inv, units = tail["inventory"], tail["units"]
|
||
known = inv.notna()
|
||
if not known.any():
|
||
return 0, 0
|
||
# Threshold per row against that row's own demand, so a slow SKU is not judged
|
||
# by a fast SKU's yardstick.
|
||
floor = (units * STOCKOUT_COVER_DAYS).clip(lower=1.0)
|
||
return int((known & (inv <= floor)).sum()), int(known.sum())
|
||
|
||
|
||
def _evidence_target(r, cur_price: float, keys: set) -> tuple | None:
|
||
"""The price this SKU has ACTUALLY run that booked the most profit per day.
|
||
|
||
Returns (key, why, gain_per_day) or None. ``best_observed_price`` has always
|
||
been computed and displayed; it was never allowed to drive a recommendation,
|
||
which is how an untested extrapolation could outrank a price with 54 days of
|
||
booked profit behind it.
|
||
|
||
Acting requires the observed winner to beat today's own observed performance
|
||
by a clear margin — otherwise the difference is noise between price bands.
|
||
"""
|
||
from pricing_agent.performance import observed_profit_at
|
||
|
||
bop, bday = r.best_observed_price, r.best_observed_profit_day
|
||
if not bop or bday is None or not cur_price or "best_observed" not in keys:
|
||
return None
|
||
if abs(bop - cur_price) / cur_price < 0.02:
|
||
return None # already there
|
||
cur_band = observed_profit_at(cur_price, r.price_bands or [])
|
||
cur_day = cur_band["actual_profit_per_day"] if cur_band else None
|
||
if cur_day is not None and bday <= max(cur_day * MATERIAL_GAIN, cur_day + 1):
|
||
return None # not materially better than today
|
||
if bday <= 0:
|
||
return None # every observed price lost money
|
||
days = next((b["days"] for b in (r.price_bands or [])
|
||
if abs(b["avg_price"] - bop) < 0.01), 0)
|
||
gain = bday - cur_day if cur_day is not None else bday
|
||
why = (f"${bop:.2f} booked ${bday:,.0f}/day of actual profit over {days} days"
|
||
+ (f", against ${cur_day:,.0f}/day at today's price" if cur_day is not None
|
||
else "")
|
||
+ " — the best of any price this product has run")
|
||
return (bop, why, gain)
|
||
|
||
|
||
def _decide(r, cur_price: float, hist: pd.DataFrame, fp: dict, scen: pd.DataFrame,
|
||
outlook: dict | None = None, comp=None, cover_days: int | None = None):
|
||
"""Deterministic action from real signals, with an ordered strategy ladder.
|
||
|
||
Two gates sit on top of the branch logic:
|
||
|
||
1. The model may only set price when its elasticity is statistically usable
|
||
AND its optimum is a genuine interior maximum.
|
||
2. When the model is gated out — or when nothing else fires — observed
|
||
evidence gets its turn, because a price with weeks of booked profit behind
|
||
it outranks any projection.
|
||
|
||
`comp` is an optional `CompetitiveState`. It may only ever ADD a verdict: when it
|
||
is absent, stale or unusable every branch below behaves exactly as it did before
|
||
competitor data existed. Nothing waits on it and nothing is blocked by it.
|
||
|
||
Returns (action, rec_key, cons_key, aggr_key, reasons, root_cause, objective).
|
||
"""
|
||
action, rec, cons, aggr, reasons, root, objective = _decide_raw(
|
||
r, cur_price, hist, fp, scen, outlook, comp, cover_days)
|
||
keys = set(scen["scenario"]) if len(scen) else {"current"}
|
||
|
||
# Price is not always the lever. When ad cost per unit climbs almost as fast
|
||
# as price, each extra $1 of price buys only cents of contribution, so no
|
||
# price reaches break-even — recommending a raise would cost volume and fix
|
||
# nothing. The lever is ad efficiency or COGS.
|
||
# Both this and BUYBOX_SUPPRESSED return Investigate-and-hold, so neither moves a price
|
||
# either way — but if the Buy Box is suppressed that is the thing to go and fix, and it
|
||
# must not be relabelled as an advertising problem on the way out.
|
||
if r.ad_curve_unrecoverable and "BUYBOX_SUPPRESSED" not in reasons:
|
||
cpd = r.contribution_per_dollar
|
||
detail = (f"each extra $1 of price yields only ${cpd:.2f} of contribution"
|
||
if cpd is not None else "ad cost rises faster than price")
|
||
if r.break_even_ad_curve and r.observed_price_max:
|
||
detail += (f"; break-even against that curve is ${r.break_even_ad_curve:.2f}, "
|
||
f"{r.break_even_ad_curve / r.observed_price_max:.1f}x the highest "
|
||
f"price ever charged (${r.observed_price_max:.2f})")
|
||
return ("Investigate", "current", "current", "current", ["AD_SPIRAL"],
|
||
root + [("Ad cost vs price",
|
||
detail + " — no reachable price breaks even. Fix ad "
|
||
"efficiency or COGS before touching price.")],
|
||
"fix_ad_efficiency")
|
||
|
||
model_ok = bool(r.elasticity_actionable and r.profit_optimal_price
|
||
and not r.profit_optimal_is_corner)
|
||
if rec == "max_profit" and not model_ok:
|
||
rec, action, reasons = "current", "Maintain", ["NO_SIGNALS"]
|
||
root = root + [("Model gate", "elasticity not usable for pricing — "
|
||
+ ((r.elasticity or {}).get("why") or "corner solution"))]
|
||
|
||
# Evidence branch: only where the cascade found nothing to act on. Inventory
|
||
# and break-even signals keep priority — they are about risk, not optimisation.
|
||
if rec == "current" and action in ("Maintain",):
|
||
ev = _evidence_target(r, cur_price, keys)
|
||
if ev:
|
||
bop, why, gain = ev
|
||
rec = "best_observed"
|
||
action = "Increase" if bop > cur_price else "Decrease"
|
||
reasons = ["BEST_OBSERVED"]
|
||
objective = "observed_best"
|
||
root = root + [("Observed evidence", why)]
|
||
|
||
cons, aggr = _order_ladder(action, rec, cons, aggr, scen, cur_price)
|
||
return action, rec, cons, aggr, reasons, root, objective
|
||
|
||
|
||
def _decide_raw(r, cur_price: float, hist: pd.DataFrame, fp: dict, scen: pd.DataFrame,
|
||
outlook: dict | None = None, comp=None, cover_days: int | None = None):
|
||
"""Branch logic: which action and which scenario key each rung starts from."""
|
||
keys = set(scen["scenario"]) if len(scen) else {"current"}
|
||
be = r.break_even or 0.0
|
||
# The RECONCILED cover the dashboard tile shows, not COSMOS's raw field.
|
||
#
|
||
# COSMOS returns `cover_days: 0` on some low-velocity SKUs that are in fact sitting on
|
||
# months of stock, and 0 satisfies NEITHER inventory rule: `0 < cover <= 35` is false and
|
||
# `cover >= 90` is false, so both go silent and the SKU falls through to whatever rung is
|
||
# next. Measured on UBMICROFIBERBS4PCFULLGREY — 167 units on hand, 15 sales in six months,
|
||
# 334 days of real cover — COSMOS reported 0 and the SKU was filed COMPETITOR_UNDERCUT
|
||
# instead of EXCESS_STOCK, quietly bypassing "inventory outranks competitor position".
|
||
#
|
||
# `build_live_sku` already reconciles this for the tile (inventory / velocity when COSMOS's
|
||
# figure is missing). It just did it AFTER the decision. Passing the same number in is what
|
||
# makes the cascade and the tile agree by construction rather than by coincidence; the
|
||
# thresholds themselves are untouched.
|
||
cover = cover_days if cover_days is not None else r.cover_days
|
||
th_now = _take_home(cur_price, fp)
|
||
outlook = outlook or {}
|
||
|
||
# real root-cause checks
|
||
recent_prices = hist["price"].tail(30)
|
||
own_change = bool(len(recent_prices) > 1 and recent_prices.iloc[0] > 0 and
|
||
abs(recent_prices.iloc[-1] - recent_prices.iloc[0]) / recent_prices.iloc[0] > 0.01)
|
||
prior_ad = float(hist["ad_spend"].iloc[-44:-14].mean()) if len(hist) >= 44 else 0.0
|
||
last_ad = float(hist["ad_spend"].tail(14).mean())
|
||
ad_collapse = bool(prior_ad > 1 and last_ad < prior_ad * 0.5)
|
||
drop = bool(r.avg_30d and r.avg_6m and r.avg_30d < 0.70 * r.avg_6m)
|
||
|
||
# A velocity drop caused by empty shelves is a SUPPLY problem, and cutting
|
||
# price to fix it pays margin for nothing. Until now the two were
|
||
# indistinguishable and every such SKU was filed as UNEXPLAINED_DROP.
|
||
oos_days, known_days = _stockout_days(hist)
|
||
stock_explained = bool(known_days and oos_days >= known_days * STOCKOUT_SHARE)
|
||
|
||
# Inventory outlook — dated urgency rather than a bare threshold crossing.
|
||
stockout_on = outlook.get("stockout_date")
|
||
arrival_on = outlook.get("next_arrival_date")
|
||
# True when we empty the shelf before relief lands, or no relief is scheduled.
|
||
stockout_first = bool(stockout_on and (arrival_on is None or stockout_on < arrival_on))
|
||
inbound = bool(arrival_on)
|
||
|
||
no_cost = fp["cost"] <= 0 and fp["fba"] <= 0
|
||
root = []
|
||
if no_cost:
|
||
root.append(("Cost data in COSMOS", "missing — costPerUnit/fbaFee are 0; "
|
||
"fill COGS in COSMOS to price this SKU"))
|
||
else:
|
||
root.append(("Take-home at current price",
|
||
f"{'confirmed negative' if th_now < 0 else 'positive'} — "
|
||
f"${th_now:.2f}/unit before ads"))
|
||
if r.ad_cost_per_unit is not None:
|
||
net = th_now - r.ad_cost_per_unit
|
||
root.append(("Net after ad cost",
|
||
f"${net:.2f}/unit (ads ${r.ad_cost_per_unit:.2f}/unit)"))
|
||
root.append(("Own price change (30d)", "detected" if own_change else "not found"))
|
||
root.append(("Ad-spend collapse", "detected — spend halved" if ad_collapse else "not found"))
|
||
if cover is not None:
|
||
root.append(("Inventory cover", f"{cover} days on hand"))
|
||
if known_days:
|
||
root.append(("Days effectively out of stock (30d)",
|
||
f"{oos_days} of {known_days} days with a reading"
|
||
+ (" — enough to explain the velocity drop"
|
||
if stock_explained else "")))
|
||
if stockout_on:
|
||
src = ("COSMOS projection" if outlook.get("stockout_source") == "projection"
|
||
else "extrapolated from on-hand ÷ velocity")
|
||
line = f"stock out around {stockout_on:%d %b} ({src})"
|
||
if arrival_on:
|
||
gap = (arrival_on - stockout_on).days
|
||
line += (f"; next arrival {arrival_on:%d %b}"
|
||
+ (f" — {gap} days uncovered" if gap > 0 else " — arrives in time"))
|
||
else:
|
||
line += f"; no arrival scheduled within {outlook.get('horizon_days') or 0} days"
|
||
root.append(("Stockout outlook", line))
|
||
if r.elasticity and r.elasticity.get("elasticity"):
|
||
root.append(("Elasticity fit",
|
||
f"e={r.elasticity['elasticity']} (r²={r.elasticity.get('r2')}, "
|
||
f"{r.elasticity.get('confidence')} confidence)"))
|
||
else:
|
||
root.append(("Elasticity fit", "not measurable — price barely moved in 6 months"))
|
||
|
||
prices = dict(zip(scen["scenario"], scen["price"])) if len(scen) else {}
|
||
|
||
def pick(key, fallback="current"):
|
||
return key if key in keys else fallback
|
||
|
||
def raise_target():
|
||
"""Highest of min_safe / up_5 — a raise must never land below current."""
|
||
cands = [k for k in ("min_safe", "up_5") if k in keys
|
||
and prices.get(k, 0) > cur_price]
|
||
return max(cands, key=lambda k: prices[k]) if cands else pick("up_5")
|
||
|
||
# ── competitor signals ───────────────────────────────────────────────────
|
||
# Read ONCE, here, so every branch below sees the same facts. `comp_usable` is the only
|
||
# gate: absent, stale, failed and "ownership unknown" all collapse to False and the
|
||
# cascade then behaves exactly as it did before competitor data existed. A competitor
|
||
# rule can only ever ADD a verdict, never withhold one.
|
||
from config.settings import get_rules
|
||
rules = get_rules()
|
||
# Scoped kill switch. Rather than branching around the two competitor rules — which would
|
||
# be a SECOND "pretend competitor data isn't there" path to keep in step with the first —
|
||
# the flag routes through the EXISTING fail-safe: the state is replaced with an explicit
|
||
# unavailable one, so `comp_usable` goes False and every branch below behaves exactly as it
|
||
# does for missing or stale data. One path, already tested.
|
||
if comp is not None and not rules.competitor_rules_enabled:
|
||
from pricing_agent.competitive_state import unavailable
|
||
comp = unavailable(
|
||
"competitor rules are switched off in pricing_rules.yaml "
|
||
"(competitor_rules_enabled: false) — this verdict is competitor-blind")
|
||
comp_usable = bool(comp is not None and getattr(comp, "usable", False))
|
||
comp_suppressed = comp_undercut = False
|
||
comp_price = comp_gap = None
|
||
if comp_usable:
|
||
from pricing_agent.competitive_state import BASIS_SAME_ASIN, BASIS_SHEET
|
||
from pricing_agent.schemas import BuyBoxStatus
|
||
comp_suppressed = comp.status is BuyBoxStatus.SUPPRESSED
|
||
comp_gap = comp.undercut_pct
|
||
comp_price = comp.competitor_min
|
||
# What counts as "a cheaper competitor" depends on WHAT the cheaper price is a price
|
||
# of, and the two cases are not interchangeable:
|
||
#
|
||
# * SAME ASIN (Apify): only LOST_PRICE qualifies. A rival holding the Buy Box at or
|
||
# ABOVE our price is LOST_ELIGIBILITY — cutting fixes nothing and donates margin.
|
||
# * LIKE-FOR-LIKE SHEET: a rival BRAND's equivalent product is cheaper. This is a
|
||
# competitive-position signal and holds regardless of who owns our Buy Box — we can
|
||
# hold our own perfectly well while a different product undercuts us. It is exactly
|
||
# what the comparison workbook exists to report, and the sibling tool's own action
|
||
# classifier already treats it this way.
|
||
basis_qualifies = (
|
||
(comp.basis == BASIS_SAME_ASIN and comp.status is BuyBoxStatus.LOST_PRICE)
|
||
or comp.basis == BASIS_SHEET
|
||
)
|
||
comp_undercut = bool(
|
||
basis_qualifies
|
||
and comp_gap is not None
|
||
and comp_gap >= rules.competitor_undercut_material_pct
|
||
and comp_price
|
||
)
|
||
root.append(("Competitor position",
|
||
f"{comp.status.value} ({comp.source}, basis={comp.basis}"
|
||
+ (f", {comp.rivals} rival(s)" if comp.rivals else "")
|
||
+ ")"
|
||
+ (f" — cheapest comparable rival ${comp_price:.2f}, "
|
||
f"{comp_gap:.1%} below us" if comp_gap is not None and comp_price
|
||
else "")))
|
||
# We HOLD the Buy Box and sit well above the rival field. Deliberately a NOTE and
|
||
# nothing else: holding the Buy Box at a premium is usually a position worth keeping,
|
||
# and the elasticity fit is the thing with actual evidence behind it. If the model
|
||
# already wants a raise it says so on its own rung; if it does not, a premium is not
|
||
# grounds to overrule it. So this can only ever SUPPORT a raise a human is already
|
||
# considering — it never sets a price and never changes an action.
|
||
if (comp.status is BuyBoxStatus.WON and comp_price and cur_price
|
||
and cur_price > comp_price * (1 + rules.competitor_premium_material_pct)):
|
||
over = (cur_price - comp_price) / comp_price
|
||
model_wants_raise = bool(
|
||
r.profit_optimal_price and r.profit_optimal_price > cur_price * 1.02)
|
||
root.append((
|
||
"Competitor premium",
|
||
f"we hold the Buy Box at ${cur_price:.2f}, {over:.0%} above the cheapest "
|
||
f"rival (${comp_price:.2f})"
|
||
+ (" — and the elasticity fit independently points higher, so the raise "
|
||
"already has evidence behind it"
|
||
if model_wants_raise else
|
||
" — the elasticity fit does NOT point higher, so this is context for a "
|
||
"human, not a reason to raise: no price change is recommended from it")))
|
||
for note in getattr(comp, "notes", []) or []:
|
||
root.append(("Competitor data note", note))
|
||
elif comp is not None:
|
||
# Say WHY it was ignored. "No competitor data" and "competitor data 9h old" look
|
||
# identical in a blank cell, and only one of them is worth re-running a scrape for.
|
||
root.append(("Competitor Buy Box",
|
||
f"not used — {getattr(comp, 'unusable_because', 'unavailable')}"))
|
||
|
||
if no_cost:
|
||
return ("Investigate", "current", "current", "current",
|
||
["NO_COST_DATA"], root, "fix_data_first")
|
||
|
||
# A suppressed Buy Box sells NOTHING at any price, so every downstream number on this
|
||
# SKU — margin, elasticity, profit-optimal price — describes a listing nobody can buy
|
||
# from. It therefore sits directly under NO_COST_DATA: those are the only two states
|
||
# where the answer is "go and fix something", not "set a price".
|
||
#
|
||
# This does NOT reorder inventory risk against profit optimisation; those two keep their
|
||
# existing order below. It also matches the sibling competitor tool, whose action
|
||
# classifier already ranks a suppressed Buy Box above every pricing question.
|
||
if comp_suppressed:
|
||
why = comp.reason or "Amazon is not showing our offer"
|
||
return ("Investigate", "current", "current", "current",
|
||
["BUYBOX_SUPPRESSED"],
|
||
root + [("Buy Box", f"SUPPRESSED — {why}. No price change can help until "
|
||
f"the listing is reinstated.")],
|
||
"fix_buybox_first")
|
||
|
||
if be and cur_price < be:
|
||
rec = raise_target()
|
||
return ("Increase", rec, pick("up_2"), rec,
|
||
["BELOW_BREAK_EVEN"], root, "margin_protection")
|
||
|
||
if r.is_losing_money:
|
||
rec = raise_target()
|
||
return ("Increase", rec, pick("up_2"), rec,
|
||
["LOSING_MONEY"] + (["VELOCITY_DROP"] if drop else []),
|
||
root, "margin_protection")
|
||
|
||
if cover is not None and 0 < cover <= LOW_COVER_DAYS:
|
||
# STOCKOUT_BEFORE_ARRIVAL / INBOUND_PO are ADDITIVE — they explain, they do
|
||
# not escalate. This branch already recommends +5%, which IS the step cap
|
||
# (MAX_STEP_PCT), so there is nothing stronger to escalate to. Do not read
|
||
# their absence from the action as an oversight.
|
||
return ("Increase", pick("up_5"), pick("up_2"), pick("up_5"),
|
||
["LOW_STOCK"]
|
||
+ (["STOCKOUT_BEFORE_ARRIVAL"] if stockout_first else [])
|
||
+ (["INBOUND_PO"] if inbound else []),
|
||
root, "low_stock_protection")
|
||
|
||
if drop and not own_change and not ad_collapse:
|
||
# An empty shelf is not a demand mystery. Naming the cause keeps this off
|
||
# the Investigate pile, where it would have sat waiting for a human to
|
||
# rediscover that the SKU had simply run out.
|
||
if stock_explained:
|
||
return ("Maintain", "current", "current", pick("up_2"),
|
||
["STOCKOUT_BEFORE_ARRIVAL", "VELOCITY_DROP"], root,
|
||
"low_stock_protection")
|
||
# A material competitor undercut is a THIRD known cause, alongside our own price
|
||
# change and an ad collapse. Once we hold that explanation, filing the drop as
|
||
# UNEXPLAINED is simply inaccurate — it sends someone off to rediscover a rival price
|
||
# already sitting in the root cause above. Same precedent as `stock_explained`
|
||
# directly above: a named cause converts an Investigate into an actionable verdict.
|
||
#
|
||
# So fall THROUGH rather than returning, and let the competitor branch below decide.
|
||
# It is the branch that knows what to do about a rival price, and its guardrails still
|
||
# apply. Observed on real SKUs: UBMICROFIBERDUVETKINGPURPLE (rival 15.6% below) and
|
||
# UBMICROFIBERBS4PCFULLGREY (35.2% below) were both filed UNEXPLAINED_DROP while the
|
||
# sheet held the explanation.
|
||
if not comp_undercut:
|
||
return ("Investigate", "current", "current", "current",
|
||
["UNEXPLAINED_DROP", "VELOCITY_DROP"], root, "max_profit")
|
||
|
||
if cover is not None and cover >= HIGH_COVER_DAYS:
|
||
# Same rule: additive. −5% is already the cap in the other direction.
|
||
return ("Decrease", pick("down_5"), pick("down_2"), pick("down_5"),
|
||
["EXCESS_STOCK"] + (["VELOCITY_DROP"] if drop else [])
|
||
+ (["INBOUND_PO"] if inbound else []),
|
||
root, "overstock_reduction")
|
||
|
||
# A rival holds the Buy Box and is materially cheaper. Sits BELOW every inventory rule
|
||
# (a stockout costs more than a lost Buy Box — chasing a rival down while the shelf is
|
||
# emptying pays margin to sell out faster) and ABOVE profit-optimal, because a price the
|
||
# market is actively beating us on is a stronger signal than a modelled optimum.
|
||
#
|
||
# The recommendation is only ever a TARGET. It is floored at break-even x1.02, capped at
|
||
# +-5% per step and clamped to observed headroom by the guardrails downstream, exactly
|
||
# like every other branch — so "match the rival" can never be executed below break-even
|
||
# no matter what the rival charges.
|
||
if comp_undercut and "comp_match" in keys:
|
||
return ("Decrease", pick("comp_match", pick("down_2")), pick("down_2"),
|
||
pick("down_5"),
|
||
["COMPETITOR_UNDERCUT"] + (["VELOCITY_DROP"] if drop else []),
|
||
root + [("Undercut basis",
|
||
f"{comp.basis} — "
|
||
+ ("a rival brand's like-for-like variant is cheaper; we may still "
|
||
"hold our own Buy Box"
|
||
if comp.basis == BASIS_SHEET else
|
||
"we lost the Buy Box on our own listing to a cheaper offer"))],
|
||
"competitive_position")
|
||
|
||
if r.profit_optimal_price and cur_price > 0:
|
||
gap = (r.profit_optimal_price - cur_price) / cur_price
|
||
if gap >= 0.02:
|
||
return ("Increase", pick("max_profit", pick("up_2")), pick("up_2"),
|
||
pick("up_5"), ["PROFIT_OPTIMAL", "PREMIUM_HEADROOM"], root, "max_profit")
|
||
if gap <= -0.02:
|
||
return ("Decrease", pick("max_profit", pick("down_2")), pick("down_2"),
|
||
pick("down_5"), ["PROFIT_OPTIMAL"], root, "max_profit")
|
||
|
||
return ("Maintain", "current", "current", pick("up_2"), ["NO_SIGNALS"], root, "max_profit")
|
||
|
||
|
||
def _confidence(action: str, hist: pd.DataFrame, r) -> tuple[str, int]:
|
||
days_with_sales = int((hist["units"] > 0).sum()) if len(hist) else 0
|
||
el = (r.elasticity or {}).get("confidence")
|
||
if days_with_sales >= 120 and el in ("high", "medium"):
|
||
tier, score = "High", 82
|
||
elif days_with_sales >= 90:
|
||
tier, score = "Medium", 62
|
||
elif days_with_sales >= 30:
|
||
tier, score = "Low", 45
|
||
else:
|
||
tier, score = "Insufficient", 25
|
||
if action == "Investigate" and tier in ("High", "Medium"):
|
||
tier, score = "Low", 40
|
||
return tier, score
|
||
|
||
|
||
_SHEET_CACHE: dict = {}
|
||
|
||
|
||
def _load_competitor_sheet(force: bool = False):
|
||
"""The competitor comparison workbook, loaded ONCE per process.
|
||
|
||
Cached because a product line is 60+ SKUs and re-reading a 100 KB workbook per SKU would
|
||
turn one file read into sixty. Keyed on the resolved path and its mtime, so replacing the
|
||
sheet on disk is picked up on the next run without a restart.
|
||
|
||
Returns None whenever there is no usable sheet — not configured, not found, unreadable,
|
||
wrong shape. That is a normal state, and every caller renders it as N/A.
|
||
"""
|
||
from config.settings import get_settings
|
||
from pricing_agent.competitor_sheet import CompetitorSheet
|
||
|
||
s = get_settings()
|
||
try:
|
||
if s.competitor_sheet_path:
|
||
path = Path(s.competitor_sheet_path)
|
||
if not path.is_file():
|
||
logger.warning("COMPETITOR_SHEET_PATH is set to %s but there is no file "
|
||
"there — competitor data will read N/A", path)
|
||
return None
|
||
key = (str(path), path.stat().st_mtime)
|
||
if force or _SHEET_CACHE.get("key") != key:
|
||
_SHEET_CACHE.clear()
|
||
_SHEET_CACHE.update(key=key, sheet=CompetitorSheet.load(path))
|
||
return _SHEET_CACHE.get("sheet")
|
||
|
||
dirs = [d for d in (s.competitor_sheet_dirs or "").split(";") if d.strip()]
|
||
key = ("discover", tuple(dirs))
|
||
if force or _SHEET_CACHE.get("key") != key:
|
||
_SHEET_CACHE.clear()
|
||
_SHEET_CACHE.update(key=key, sheet=CompetitorSheet.discover(*dirs))
|
||
return _SHEET_CACHE.get("sheet")
|
||
except Exception: # noqa: BLE001 — a bad sheet must never break a pricing run
|
||
logger.warning("competitor sheet could not be loaded; competitor data reads N/A",
|
||
exc_info=True)
|
||
return None
|
||
|
||
|
||
def _competitors(sku: str, asin: str | None, cur_price: float, units_month: float,
|
||
snap, sheet_row=None, sheet=None) -> tuple[pd.DataFrame, float | None, dict]:
|
||
"""Our row + competitor rivals when available.
|
||
|
||
Returns (frame, competitor_median, meta). `meta` carries the Buy Box status and
|
||
a `rivals` count so the UI can tell apart "not scraped", "scraped — no third-party
|
||
sellers", and "scraped — rivals found".
|
||
|
||
TWO KINDS OF RIVAL, and the tab must not blur them:
|
||
|
||
* `snap` (Apify) holds other sellers' offers on **our own ASIN** — same product, same
|
||
page, and a cheaper one means we lost the Buy Box.
|
||
* `sheet_row` holds a rival BRAND's equivalent variant, matched like-for-like on size and
|
||
colour. Different ASIN, different listing, different product.
|
||
|
||
The sheet rows were previously read for the DECISION but never for this tab, so a SKU the
|
||
workbook covered — with two exactly-matched rivals priced — still rendered "Competitor
|
||
prices not scraped". That is the workbook and the recommendation disagreeing in front of a
|
||
stakeholder, which is the one thing the competitor wiring exists to prevent.
|
||
"""
|
||
rows = [{
|
||
"seller": f"★ {sku} (us)", "asin": asin or "—", "is_us": True,
|
||
"effective_price": round(cur_price, 2), "coupon": 0.0, "badge": "—",
|
||
"sold_badge": _sold_badge(units_month), "rating": None, "reviews": None,
|
||
"bsr": None, "delivery_days": None, "in_stock": True,
|
||
"freshness": "live (COSMOS)",
|
||
}]
|
||
comp_median, rivals = None, 0
|
||
meta = {"scraped": snap is not None, "buy_box_status": None,
|
||
"buy_box_price": None, "buy_box_seller": None, "own_offers": 0,
|
||
# Which of the two sources the rival rows below actually came from, so the UI can
|
||
# word "offers on this ASIN" vs "a rival brand's own listing" correctly instead of
|
||
# calling both the same thing.
|
||
"source": None, "exact_rivals": 0, "sheet_as_of": None}
|
||
if snap is not None:
|
||
comp_median = snap.competitive_median
|
||
meta["buy_box_status"] = getattr(snap.buy_box_status, "value", None)
|
||
meta["buy_box_price"] = snap.buy_box_price
|
||
meta["buy_box_seller"] = snap.buy_box_seller_name
|
||
meta["own_offers"] = sum(1 for o in snap.offers if o.is_ours)
|
||
for o in snap.offers:
|
||
if o.is_ours or o.price is None:
|
||
continue
|
||
rivals += 1
|
||
rows.append({
|
||
"seller": o.seller_name or "Unknown seller", "asin": snap.asin or asin or "—",
|
||
"is_us": False, "effective_price": round(o.price, 2), "coupon": 0.0,
|
||
"badge": "Featured Offer" if o.is_buy_box else "—", "sold_badge": "—",
|
||
"rating": None, "reviews": None, "bsr": None, "delivery_days": None,
|
||
"in_stock": True, "freshness": "competitor scrape",
|
||
})
|
||
# Like-for-like rivals from the comparison workbook. Added when the per-ASIN scrape found
|
||
# no third-party offers — which is the normal case here, because we are the brand owner and
|
||
# sell our own ASINs alone. Our real competition is another brand's listing, and that is
|
||
# exactly what the sheet holds.
|
||
if sheet_row is not None and rivals == 0:
|
||
meta["source"] = "comparison-sheet"
|
||
meta["sheet_as_of"] = getattr(sheet, "as_of", None)
|
||
prices = []
|
||
for brand, price in sorted(sheet_row.rivals.items()):
|
||
fuzzy = brand in sheet_row.fuzzy_rivals
|
||
supp = brand in sheet_row.suppressed_rivals
|
||
# Both exclusions are shown rather than hidden: a reader deciding whether to act
|
||
# needs to see the number AND why it did not move the price.
|
||
if fuzzy:
|
||
badge = "NEAR colour match — verify, does not price"
|
||
elif supp:
|
||
badge = "Buy Box suppressed — not buyable"
|
||
else:
|
||
badge = "exact match"
|
||
prices.append(price)
|
||
rows.append({
|
||
"seller": brand, "asin": sheet_row.rival_asins.get(brand) or "—",
|
||
"is_us": False, "effective_price": round(price, 2), "coupon": 0.0,
|
||
"badge": badge, "sold_badge": "—", "rating": None, "reviews": None,
|
||
"bsr": None, "delivery_days": None, "in_stock": not supp,
|
||
"freshness": ("comparison sheet"
|
||
+ (f" · {sheet.as_of:%d %b}" if getattr(sheet, "as_of", None)
|
||
else "")),
|
||
})
|
||
rivals += 1
|
||
if prices:
|
||
# A real median, not the upper of a pair: with two rivals at $19.99 and $24.99,
|
||
# `prices[len//2]` reports $24.99, which is simply the wrong number to print next
|
||
# to the word "median".
|
||
import statistics
|
||
comp_median = round(float(statistics.median(prices)), 2)
|
||
meta["exact_rivals"] = len(prices)
|
||
elif rivals:
|
||
meta["source"] = "apify"
|
||
|
||
meta["rivals"] = rivals
|
||
return pd.DataFrame(rows), comp_median, meta
|
||
|
||
|
||
def _explain(sku, action, cur_price, rec_row, cur_row, reasons, tier, r) -> str:
|
||
labels = ", ".join(REASON_LABELS.get(c, c).lower() for c in reasons)
|
||
if "NO_COST_DATA" in reasons:
|
||
return (f"COSMOS has **no cost data** for {sku} (costPerUnit and fbaFee came back 0), "
|
||
f"so break-even and margins cannot be computed. Fix the product's COGS in "
|
||
f"COSMOS/Inventory Planner first — any price recommendation would be a guess.")
|
||
if "AD_SPIRAL" in reasons:
|
||
cpd = getattr(r, "contribution_per_dollar", None)
|
||
be_c = getattr(r, "break_even_ad_curve", None)
|
||
return (
|
||
f"**Price is not the lever on this SKU.** Advertising cost per unit rises "
|
||
f"with price almost as fast as price itself"
|
||
+ (f", so each extra $1 of price leaves only **${cpd:.2f}** of contribution"
|
||
if cpd is not None else "")
|
||
+ (f". Break-even against that curve is **${be_c:.2f}** — far beyond any price "
|
||
f"this product has ever run" if be_c else ".")
|
||
+ f" Raising from ${cur_price:.2f} would shed volume without reaching profit. "
|
||
f"The fix is ad efficiency (bid/placement/targeting) or landed cost; "
|
||
f"re-run pricing once ad cost per unit is under control."
|
||
)
|
||
if action == "Investigate":
|
||
return (f"Sales are running well below the 6-month baseline but **no checked cause "
|
||
f"explains it** — no price change on our side and no ad-spend collapse. "
|
||
f"Cutting price without a confirmed cause risks paying margin for a problem "
|
||
f"price didn't create. Recommend holding ${cur_price:.2f} and reviewing "
|
||
f"listing health and traffic manually.")
|
||
if action == "Maintain":
|
||
return (f"Signals: {labels}. Best response is to hold at **${cur_price:.2f}** — "
|
||
f"expected ≈ ${cur_row['profit_30d']:,.0f}/30d at "
|
||
f"{cur_row['units_day']:.0f} units/day. Re-evaluated on every refresh.")
|
||
delta = rec_row["price"] - cur_price
|
||
extra = ""
|
||
if r.actual_profit_per_unit is not None:
|
||
extra = (f" COSMOS actuals show {'a loss' if r.actual_profit_per_unit < 0 else 'profit'} of "
|
||
f"${abs(r.actual_profit_per_unit):.2f}/unit over the last 3 months (ads included).")
|
||
return (f"Driven by {labels}: move from ${cur_price:.2f} to **${rec_row['price']:.2f}** "
|
||
f"({'+' if delta >= 0 else '−'}${abs(delta):.2f}). Expected: "
|
||
f"{rec_row['units_day']:.0f} units/day, profit ${rec_row['profit_30d']:,.0f}/30d "
|
||
f"(from ${cur_row['profit_30d']:,.0f}), margin {rec_row['margin_pct']:.1f}%, "
|
||
f"cover {rec_row['cover_days']:.0f} days. Confidence is {tier.lower()}; "
|
||
f"guardrails enforced on every option.{extra}")
|
||
|
||
|
||
# ---------------------------------------------------------------- bulk economics
|
||
SCEN_LABELS = {
|
||
"current": "Current", "up_2": "+2%", "up_5": "+5%", "down_2": "−2%",
|
||
"down_5": "−5%", "match_median": "Match competitor", "min_safe": "Safe floor",
|
||
"max_profit": "Profit-optimal", "best_observed": "Best observed",
|
||
}
|
||
|
||
|
||
def _storage_30d(svc, sku: str, cur_price: float, units_30d: float,
|
||
inventory: float, days: int = 30) -> float:
|
||
"""One bulk-calculator call → the storage charge over `days` (fixed across prices).
|
||
|
||
**The API returns a PER-DAY charge.** This is not documented and is easy to
|
||
reintroduce, so: `storageCharges` scales linearly with `inventory` and is
|
||
completely independent of `saleUnits` — identical at 100, 1,000, 10,000 and
|
||
43,007 units. Backing the rate out of item volume gives **$0.7800/cu ft/month**
|
||
on two independent SKUs to four decimal places, which is Amazon's exact
|
||
standard-size Jan–Sep FBA *monthly* storage rate. Consuming it as a monthly
|
||
figure understated storage 30x — $971 vs $29,127 per 30 days on one SKU,
|
||
about $0.55/unit against a $1.36/unit unexplained profit gap.
|
||
|
||
Bulk take-home per unit is linear in price and equals our fee model, so a
|
||
single call at the current price gives the rate; every scenario's economics
|
||
then reconcile to what the bulk calculator would return.
|
||
"""
|
||
try:
|
||
b = svc.bulk_quote(sku, cur_price, sale_units=max(units_30d, 1),
|
||
inventory=max(inventory, 1), marketing=0.0)
|
||
return float(b.storage_charges or 0.0) * days
|
||
except Exception:
|
||
logger.warning("bulk storage lookup failed for %s", sku)
|
||
return 0.0
|
||
|
||
|
||
def _scenario_economics(cur_price: float, units_day: float, fp: dict,
|
||
elasticity: float, tacos_frac: float, storage_30d: float,
|
||
grid: dict, actual_30d: float | None = None,
|
||
actual_rev: float | None = None,
|
||
actual_ad: float | None = None,
|
||
actual_units: float | None = None,
|
||
realized_price: float | None = None,
|
||
ad_model: dict | None = None,
|
||
bands: list | None = None,
|
||
gap_per_unit: float = 0.0,
|
||
) -> tuple[list[dict], dict]:
|
||
"""Economics per candidate price, on ONE consistent price basis.
|
||
|
||
Two rules make this table honest, and both were once broken:
|
||
|
||
1. **Anchor demand at the price customers actually paid**, not the list price.
|
||
Units come from a window in which the SKU sold at ``realized_price``
|
||
(revenue ÷ units), which sits below list whenever there are promos or
|
||
coupons. Anchoring the elasticity ratio at list while taking units from
|
||
that window inflates every projection and makes modelled revenue
|
||
non-monotonic in price — impossible under ``p^(1+e)``.
|
||
|
||
2. **Ad spend per unit rises with price**, because a higher price converts
|
||
worse. Holding TACoS constant implies ad spend *falls* when you raise
|
||
price, which is backwards. ``ad_model`` (fitted on observed price bands)
|
||
supplies the real relationship; without it we hold ad $/unit flat.
|
||
|
||
The current row carries both readings: ``*_actual`` are COSMOS's booked facts,
|
||
while the modelled fields put it on the same basis as every other row so the
|
||
two are comparable. There is deliberately no calibration factor — a
|
||
multiplicative correction fitted at one price also scales the slope, which
|
||
flattens the very price/profit relationship this table exists to measure.
|
||
"""
|
||
from pricing_agent.performance import ad_per_unit_at
|
||
|
||
anchor = realized_price if (realized_price or 0) > 0 else cur_price
|
||
ad_flat = (actual_ad / actual_units) if (actual_ad and actual_units) else (
|
||
tacos_frac * anchor)
|
||
|
||
rows = []
|
||
for key, p in grid.items():
|
||
if p <= 0:
|
||
continue
|
||
ratio = (p / anchor) ** elasticity if anchor > 0 else 1.0
|
||
u_day = max(units_day * ratio, 0.0)
|
||
u_30d = u_day * 30
|
||
revenue = u_30d * p
|
||
th_pu = _take_home(p, fp) # == bulk take-home/unit
|
||
gross = th_pu * u_30d - storage_30d # bulk total take-home
|
||
ad_pu = ad_per_unit_at(p, ad_model, ad_flat) # $/unit, price-dependent
|
||
ad = ad_pu * u_30d
|
||
# Costs COSMOS books that the fee model cannot see, as a constant $/unit.
|
||
# Constant, not proportional: it corrects the level without bending the
|
||
# slope between prices, and the backtest decides whether to apply it.
|
||
gross -= gap_per_unit * u_30d
|
||
obs = _observed_band(p, bands)
|
||
rows.append({
|
||
"key": key, "label": SCEN_LABELS.get(key, key), "price": round(p, 2),
|
||
"delta_pct": (p - cur_price) / cur_price * 100 if cur_price else 0.0,
|
||
"units_day": int(round(u_day)), "units_30d": int(round(u_30d)),
|
||
"revenue_30d": round(revenue), "ad_30d": round(ad),
|
||
"ad_per_unit": round(ad_pu, 2),
|
||
"gross_30d": round(gross), "net_30d": round(gross - ad),
|
||
"th_pu": round(th_pu, 2),
|
||
# provenance: has this price ever actually been run?
|
||
"observed_days": obs["days"] if obs else 0,
|
||
"observed_net_30d": (round(obs["actual_profit_per_day"] * 30)
|
||
if obs else None),
|
||
"is_actual": False,
|
||
})
|
||
cur = next((x for x in rows if x["key"] == "current"), rows[0] if rows else None)
|
||
|
||
# The current row keeps COSMOS's booked facts ALONGSIDE its modelled twin, so
|
||
# the reader can see the model's own baseline instead of a silent basis change.
|
||
if cur:
|
||
cur["is_actual"] = True
|
||
# Snapshot the model's own reading BEFORE the booked figures overwrite it. Every
|
||
# economic field gets a twin, not just some of them: a caller that finds
|
||
# `modelled_net_30d` but no `modelled_price` is the one that ends up comparing a
|
||
# modelled profit against a booked price. Read them through `modelled()`.
|
||
cur["modelled_price"] = cur["price"]
|
||
cur["modelled_revenue_30d"] = cur["revenue_30d"]
|
||
cur["modelled_net_30d"] = cur["net_30d"]
|
||
cur["modelled_units_30d"] = cur["units_30d"]
|
||
cur["modelled_units_day"] = cur["units_day"]
|
||
cur["modelled_ad_30d"] = cur["ad_30d"]
|
||
if actual_units:
|
||
cur["units_30d"] = int(round(actual_units))
|
||
cur["units_day"] = int(round(actual_units / 30))
|
||
if actual_rev:
|
||
cur["revenue_30d"] = round(actual_rev)
|
||
if actual_ad is not None:
|
||
cur["ad_30d"] = round(actual_ad)
|
||
if actual_30d is not None:
|
||
cur["net_30d"] = round(actual_30d)
|
||
cur["price"] = round(anchor, 2) # what buyers actually paid
|
||
cur["list_price"] = round(cur_price, 2)
|
||
|
||
if cur:
|
||
# The modelled margin needs the modelled numerator AND denominator. Computing it here,
|
||
# once, is what stops a caller pairing a booked margin with a projected one.
|
||
cur["modelled_net_margin_pct"] = (
|
||
round(cur["modelled_net_30d"] / cur["modelled_revenue_30d"] * 100, 1)
|
||
if cur.get("modelled_revenue_30d") else 0.0)
|
||
|
||
for x in rows:
|
||
x["net_margin_pct"] = round(x["net_30d"] / x["revenue_30d"] * 100, 1) \
|
||
if x["revenue_30d"] else 0.0
|
||
x["net_vs_current"] = round(x["net_30d"] - (cur["net_30d"] if cur else 0))
|
||
x["ad_vs_current"] = round(x["ad_30d"] - (cur["ad_30d"] if cur else 0))
|
||
x["advice"] = _scenario_advice(x, cur)
|
||
best = max(rows, key=lambda z: z["net_30d"]) if rows else None
|
||
summary = _scenario_summary(cur, best, tacos_frac, False)
|
||
return rows, {"best_key": best["key"] if best else None, "summary": summary,
|
||
"calibrated": False, "factor": 1.0, "anchor_price": round(anchor, 2)}
|
||
|
||
|
||
# The economic fields the `current` row carries twice — once booked, once modelled.
|
||
_TWINNED = ("price", "units_30d", "units_day", "revenue_30d", "ad_30d", "net_30d",
|
||
"net_margin_pct")
|
||
|
||
|
||
def modelled(row: dict | None) -> dict:
|
||
"""A scenario row as THE MODEL sees it — safe to compare against any other row.
|
||
|
||
The `current` row deliberately carries two readings of the same product:
|
||
|
||
* **booked** — COSMOS's actual units, revenue, ad spend and profit over the window,
|
||
priced at the average customers really paid (revenue / units);
|
||
* **modelled** — the same model every other row uses, evaluated at TODAY's price.
|
||
|
||
Both are correct and they answer different questions. Booked says what this SKU earned;
|
||
modelled is the only one that can be differenced against another price, because a delta
|
||
has to subtract like from like.
|
||
|
||
Getting that wrong put three wrong numbers on screen before this function existed —
|
||
an impact of -$78 on a +5% raise (so the headline tile read "Profit opportunity $0"),
|
||
a strategy table whose "Conservative" option looked like a price cut, and a margin tile
|
||
claiming 6.7% -> 6.6% for a move that actually doubles margin. Each was a separate
|
||
hand-written ``row.get("modelled_x", row["x"])`` that covered some fields and not others.
|
||
|
||
Every other row is already modelled, so this returns it unchanged. Use it for ANY
|
||
cross-row arithmetic; reach for the raw row only to report what was actually booked.
|
||
"""
|
||
if not row:
|
||
return {}
|
||
out = dict(row)
|
||
for k in _TWINNED:
|
||
twin = f"modelled_{k}"
|
||
if twin in row:
|
||
out[k] = row[twin]
|
||
return out
|
||
|
||
|
||
def booked(row: dict | None) -> dict | None:
|
||
"""A row as COSMOS BOOKED it, or None when the row is a projection rather than a fact.
|
||
|
||
The counterpart to `modelled()`, so "what did this actually earn?" has an answer that is
|
||
equally hard to get wrong. Only the `current` row is ever a fact.
|
||
"""
|
||
if not row or not row.get("is_actual"):
|
||
return None
|
||
return dict(row)
|
||
|
||
|
||
def modelled_net_30d(row: dict | None, fallback: float = 0.0) -> float:
|
||
"""The model's 30-day net for a row. Thin wrapper over `modelled()`."""
|
||
if not row:
|
||
return fallback
|
||
return float(modelled(row).get("net_30d", fallback))
|
||
|
||
|
||
def reproject_inventory(projections: list, units_day_now: float, units_day_new: float,
|
||
) -> list[dict]:
|
||
"""COSMOS's weekly stock projection, re-run at a different rate of sale.
|
||
|
||
Answers the one question the Inventory tab cannot: COSMOS projects the shelf at TODAY's
|
||
price, so a recommendation to raise or cut is shown with no consequence for stock. Raising
|
||
slows the burn and stretches cover; cutting empties the shelf sooner. On a SKU already
|
||
close to a stockout that is the difference between a sound price move and an expensive one.
|
||
|
||
This is OUR model, not COSMOS's, and it is labelled that way wherever it is shown. Only
|
||
two things are re-derived, and only from what the scenario table already computed:
|
||
|
||
* **units sold per week** scale by `units_day_new / units_day_now` — the same elasticity
|
||
response the Scenarios tab uses, so the two can never disagree;
|
||
* **cover days** are recomputed against the new rate.
|
||
|
||
Everything else is carried through UNCHANGED and deliberately:
|
||
|
||
* **arrivals** — a purchase order already placed does not move because we re-priced;
|
||
* **unit value** — taken from COSMOS's own `inventoryValue / inventory` for that week,
|
||
so the value column stays on COSMOS's cost basis rather than being re-derived from
|
||
our COGS and quietly disagreeing with the tab above.
|
||
|
||
Stock is floored at zero and the week is flagged, because a negative shelf is not a
|
||
forecast. Returns [] when the inputs cannot support the arithmetic, so the caller shows
|
||
nothing rather than a fabricated table.
|
||
"""
|
||
if not projections or units_day_now <= 0 or units_day_new <= 0:
|
||
return []
|
||
|
||
ratio = units_day_new / units_day_now
|
||
rows: list[dict] = []
|
||
# Walk the weeks forward carrying our own running stock level: each week's DRAW is what
|
||
# scales with price, not the stock level itself. Rescaling the level directly would also
|
||
# rescale the arrivals baked into it and quietly inflate the shelf.
|
||
prev_units: float | None = None
|
||
running: float | None = None
|
||
for i, p in enumerate(projections):
|
||
units = float(p.get("inventory") or 0.0)
|
||
arrival = float(p.get("warehouse_arrival") or 0.0)
|
||
value = float(p.get("inventory_value") or 0.0)
|
||
per_unit = (value / units) if units else 0.0
|
||
|
||
if running is None: # week 0 is today's shelf — unchanged by price
|
||
running = units
|
||
else:
|
||
# COSMOS's own week-on-week movement, split into the draw and the arrival.
|
||
drawn = max((prev_units or 0.0) + arrival - units, 0.0)
|
||
running = running + arrival - drawn * ratio
|
||
|
||
stockout = running <= 0
|
||
running = max(running, 0.0)
|
||
# BOTH covers are computed here, on OUR velocities — deliberately not COSMOS's
|
||
# `coverDays` for the "now" column.
|
||
#
|
||
# COSMOS divides by its own 7-day average (64/day on the audited SKU) while the new
|
||
# column divides by the modelled rate at the new price (66/day). Putting those side by
|
||
# side made a 5% RAISE look like it SHRANK cover, 78 -> 76, which is backwards: the
|
||
# whole difference was the divisor, not the price. One basis, so the delta between the
|
||
# two columns is the price effect and nothing else.
|
||
#
|
||
# The consequence is that this column can differ by a day or two from the Inventory
|
||
# tab, which shows COSMOS's figure verbatim. That is stated in the caption.
|
||
# FLOORED, not rounded, because that is what COSMOS does: 5,029 units over a 64/day
|
||
# rate is 78.58 days and its grid prints 78. Rounding put every week exactly one day
|
||
# above COSMOS's own column, which reads as a discrepancy rather than a convention.
|
||
# Flooring is also the safer direction for a stockout figure.
|
||
cover_now = (units / units_day_now) if units_day_now > 0 else None
|
||
cover_new = (running / units_day_new) if units_day_new > 0 else None
|
||
rows.append({
|
||
"date": p.get("date"),
|
||
"units_now": int(round(units)),
|
||
"units_new": int(round(running)),
|
||
"delta_units": int(round(running - units)),
|
||
"value_new": round(running * per_unit),
|
||
"cover_now": int(cover_now) if cover_now is not None else None,
|
||
"cover_new": int(cover_new) if cover_new is not None else None,
|
||
# COSMOS's own figure, carried alongside rather than mixed into the comparison.
|
||
"cover_cosmos": (int(p["cover_days"]) if p.get("cover_days") is not None else None),
|
||
"arrival": arrival,
|
||
"stockout": stockout,
|
||
"week": i,
|
||
})
|
||
prev_units = units
|
||
return rows
|
||
|
||
|
||
def _observed_band(price: float, bands: list | None, tol: float = 0.02):
|
||
"""The observed price band matching `price`, when this price has really run."""
|
||
near = [b for b in (bands or [])
|
||
if b.get("enough_data") and abs(b["avg_price"] - price) <= tol * price]
|
||
return max(near, key=lambda b: b["days"]) if near else None
|
||
|
||
|
||
def _scenario_advice(x: dict, cur: dict | None) -> str:
|
||
"""One-line marketing/pricing advice for a single price point.
|
||
|
||
Compares MODELLED against MODELLED. The current row also carries COSMOS's
|
||
booked facts, and comparing a projection against them mixes two price bases —
|
||
that is how this column once managed to say ad spend was both higher at +2%
|
||
and lower at +5%.
|
||
"""
|
||
if not cur or x["key"] == "current":
|
||
return "Baseline — today's actual price, volume and ad spend."
|
||
if x.get("observed_days"):
|
||
obs = x.get("observed_net_30d")
|
||
return (f"OBSERVED: this price ran {x['observed_days']} days and booked "
|
||
f"${obs:,.0f}/30d." if obs is not None else
|
||
f"OBSERVED: this price ran {x['observed_days']} days.")
|
||
_base = modelled(cur)
|
||
base_net, base_ad = _base["net_30d"], _base["ad_30d"]
|
||
up = x["price"] > cur.get("list_price", cur["price"])
|
||
better = x["net_30d"] > base_net
|
||
ad_delta = x["ad_30d"] - base_ad
|
||
ad_note = (f"ad spend ~${abs(ad_delta):,.0f} {'higher' if ad_delta > 0 else 'lower'}/30d"
|
||
if abs(ad_delta) >= 50 else "ad spend ~flat")
|
||
if up and better:
|
||
return f"Raise: fatter margin per unit and {ad_note} — net profit improves."
|
||
if up and not better:
|
||
return f"Raise: better unit margin but volume falls; {ad_note} — net dips."
|
||
if not up and better:
|
||
return f"Cut: volume + revenue rise ({ad_note}) yet net still improves — worth testing."
|
||
return f"Cut: cheaper than optimal, {ad_note} — erodes net profit."
|
||
|
||
|
||
def _scenario_summary(cur: dict | None, best: dict | None, tacos_frac: float,
|
||
calibrated: bool = False) -> str:
|
||
if not cur or not best:
|
||
return ""
|
||
tacos = tacos_frac * 100
|
||
grounded = " (calibrated to your actual booked profit)" if calibrated else ""
|
||
base = (f"At **${cur['price']:.2f}** you net **${cur['net_30d']:,.0f}/30d**{grounded} on "
|
||
f"{cur['units_30d']:,} units after **${cur['ad_30d']:,.0f}** ad spend "
|
||
f"(TACoS {tacos:.1f}%).")
|
||
if best["key"] == "current":
|
||
return base + " Current price is already the profit-optimal point in this range."
|
||
direction = "raising" if best["price"] > cur["price"] else "lowering"
|
||
gain = best["net_30d"] - cur["net_30d"]
|
||
ad_move = best["ad_30d"] - cur["ad_30d"]
|
||
ad_phrase = (f"ad spend would rise ~${ad_move:,.0f}" if ad_move > 0
|
||
else f"ad spend would ease ~${abs(ad_move):,.0f}")
|
||
return (base + f" Net profit peaks at **${best['net_30d']:,.0f}/30d** by {direction} "
|
||
f"to **${best['price']:.2f}** (**+${gain:,.0f}/30d**); {ad_phrase}/30d as volume "
|
||
f"shifts.")
|
||
|
||
|
||
# ---------------------------------------------------------------- build one SKU
|
||
def build_live_sku(svc, sku: str, with_competitive: bool = False,
|
||
today: date | None = None, on_stage=None) -> dict:
|
||
from pricing_agent.analyze import analyze_price
|
||
|
||
def stage(frac: float, label: str):
|
||
if on_stage:
|
||
on_stage(frac, label)
|
||
|
||
today = today or date.today()
|
||
stage(0.03, "reading current price")
|
||
cur_price = svc.get_current_price(sku)
|
||
if not cur_price or cur_price <= 0:
|
||
# "No selling price" is a SYMPTOM, and on the SKUs that matter most it has a known
|
||
# cause: a suppressed Buy Box sells nothing, so COSMOS records no sales, so there is
|
||
# no realized price to read. Reporting only the symptom buries the single most
|
||
# actionable finding in the competitor data behind a shrug — and these are precisely
|
||
# the rows a human needs to see. So if the sheet knows why, say why.
|
||
#
|
||
# No verdict is fabricated: without a price there is genuinely nothing to price, and
|
||
# this SKU still returns an error row rather than a recommendation. It just returns
|
||
# one that names the cause.
|
||
_sheet = _load_competitor_sheet()
|
||
if _sheet is not None and _sheet.covers(sku):
|
||
_row = _sheet.rows[sku.strip().upper()]
|
||
if _row.is_suppressed:
|
||
raise ValueError(
|
||
f"{sku}: Buy Box SUPPRESSED ({_row.our_buybox}) — Amazon is not showing "
|
||
f"our offer, which is WHY COSMOS has no selling price for it. It sells "
|
||
f"nothing at any price; reinstate the listing before pricing it.")
|
||
raise ValueError(f"{sku}: no current selling price found in COSMOS")
|
||
|
||
stage(0.10, "fetching 6-month sales history")
|
||
history = svc.get_sales_history(sku, days=HISTORY_DAYS)
|
||
stage(0.45, "fees, margin, trend & elasticity")
|
||
# Analysis NEVER depends on competitors — the recommendation, scenarios and verdict
|
||
# are driven purely by our own economics. Competitor prices are fetched separately
|
||
# below, for the Competitors tab only.
|
||
r = analyze_price(sku, cur_price, svc=svc, with_elasticity=True, history=history,
|
||
with_competitive=False, current_price=cur_price)
|
||
stage(0.82, "bulk economics & scenarios")
|
||
|
||
# COSMOS's own forward inventory projection (real weekly snapshots: units, value,
|
||
# cover days, incoming arrivals) — the authoritative INVP outlook.
|
||
inv_projection, min_po, invp_title = [], None, None
|
||
# COSMOS's OWN headline figures, carried through verbatim. The Inventory Planning grid
|
||
# prints these next to the same weekly projection, so recomputing our own from
|
||
# sales-insight history put two different daily-sale numbers on two screens for one SKU
|
||
# (ours 56 / 62 / 65 against COSMOS's 64 / 130 / 67). The weekly cells always agreed
|
||
# because they come straight from the dateMap; only the header was re-derived.
|
||
invp_stats: dict = {}
|
||
try:
|
||
invp = svc.get_invp(sku)
|
||
if invp:
|
||
inv_projection = invp.projections
|
||
min_po = {"qty": invp.min_po_quantity, "days": invp.min_po_days}
|
||
invp_title = invp.parent_asin_title
|
||
invp_stats = {
|
||
# `averageSale` is the figure COSMOS shows large; the 7-day average tracks it.
|
||
"avg_sale": invp.avg_sale,
|
||
"avg_7d": invp.avg_7d,
|
||
"avg_14d": invp.avg_14d,
|
||
"avg_30d": invp.avg_30d,
|
||
"avg_6m": invp.avg_6m,
|
||
"potential_daily_sale": invp.potential_daily_sale,
|
||
# On-hand as COSMOS counts it, kept apart from the first projected week (which
|
||
# already includes that week's arrival).
|
||
"inventory": invp.inventory,
|
||
"limbo_inventory": invp.limbo_inventory,
|
||
}
|
||
except Exception:
|
||
logger.warning("invp projection lookup failed for %s", sku)
|
||
|
||
hist = _hist_frame(history, cur_price, today)
|
||
fp = _fee_params(r.fees_detail or {}, cur_price)
|
||
|
||
# Daily actuals over the last 90 data-days (units, revenue, ad, profit). ALL current
|
||
# facts derive from ONE window on this series so units/day and revenue reconcile:
|
||
# revenue/day ÷ units/day == the actual average selling price.
|
||
daily = [{
|
||
"units": float(p.units or 0),
|
||
"revenue": float(p.revenue or ((p.sale_price or 0) * (p.units or 0))),
|
||
"ad": float((p.marketing_cost or 0) + (p.promotion_spend or 0)),
|
||
"profit": float(p.profit or 0),
|
||
} for p in (history or [])][-180:] # keep the full 6 months for the price suggestion
|
||
|
||
def window_agg(days: int) -> dict:
|
||
w = daily[-days:] if daily else []
|
||
n = len(w) or 1
|
||
u = sum(x["units"] for x in w)
|
||
rev = sum(x["revenue"] for x in w)
|
||
ad = sum(x["ad"] for x in w)
|
||
prof = sum(x["profit"] for x in w)
|
||
return {"days": n, "units_day": u / n, "revenue_day": rev / n,
|
||
"ad_day": ad / n, "profit_day": prof / n,
|
||
"units": u, "revenue": rev, "ad": ad, "profit": prof}
|
||
|
||
w30 = window_agg(30)
|
||
# Canonical velocity = actual last-30-day average (reconciles with revenue), with a
|
||
# COSMOS-average fallback only when there is no recent sales history at all.
|
||
units_day = w30["units_day"] or float(r.avg_30d or r.avg_6m or 0.0)
|
||
units_day = max(units_day, 0.0)
|
||
inventory = float(r.inventory or 0.0)
|
||
units_month = w30["units"]
|
||
|
||
# Competitor prices. Fetched in their own try/except because a scrape is best-effort by
|
||
# nature: blocked, timed out or never run are all normal, and none of them may stop a
|
||
# verdict being produced. As of the competitor wiring these DO reach the decision — but
|
||
# only ever to ADD a verdict (a suppressed Buy Box, a material undercut), never to
|
||
# withhold one. With no token set, a failed scrape, or data past its age limit, the
|
||
# verdict is byte-identical to the competitor-blind one.
|
||
comp_snap = None
|
||
if with_competitive:
|
||
stage(0.70, "competitor Buy Box + rival prices")
|
||
try:
|
||
from config.settings import get_settings
|
||
from pricing_agent.providers import get_provider
|
||
from pricing_agent.schemas import SkuInput
|
||
prov = get_provider(get_settings())
|
||
comp_snap = prov.competitive(SkuInput(
|
||
sku=sku, asin=r.asin, landed_cost=fp["cost"], target_price=cur_price))
|
||
except Exception:
|
||
logger.warning("competitor scrape failed for %s", sku)
|
||
# Loaded BEFORE the tab is built so the workbook can populate it. It was previously read
|
||
# only further down, for the decision, which is how a SKU the sheet covered still rendered
|
||
# "Competitor prices not scraped".
|
||
_sheet = _load_competitor_sheet()
|
||
_sheet_row = _sheet.rows.get(sku.strip().upper()) if _sheet is not None else None
|
||
comp_df, comp_median, comp_meta = _competitors(
|
||
sku, r.asin, cur_price, units_month, comp_snap,
|
||
sheet_row=_sheet_row, sheet=_sheet)
|
||
comp_meta["median"] = comp_median
|
||
|
||
# The single competitive fact the cascade is allowed to read. Built through
|
||
# competitive_state so that the Apify path and the Playwright workbook path cannot
|
||
# silently disagree — see that module's docstring for which is authoritative and why.
|
||
from config.settings import get_rules, get_settings
|
||
from pricing_agent.competitive_state import (
|
||
from_apify, from_playwright_cache, reconcile, unavailable,
|
||
)
|
||
_rules, _st = get_rules(), get_settings()
|
||
max_age = _rules.competitor_state_max_age_hours
|
||
|
||
# ── the comparison sheet comes first, and it GATES ────────────────────────
|
||
# The workbook currently covers one product line. A SKU inside it has real like-for-like
|
||
# rival prices; a SKU outside it has nothing, and those two must not read the same. So a
|
||
# covered SKU is priced from the sheet, and an uncovered one gets an explicit N/A naming
|
||
# what the sheet does cover — rather than falling through to a per-ASIN scrape that would
|
||
# quietly price a SKU the sheet was supposed to govern (`competitor_sheet_only`).
|
||
#
|
||
# Same object the Competitors tab was built from a few lines above: the tab and the verdict
|
||
# must never be able to read different rows.
|
||
sheet = _sheet
|
||
comp_state = None
|
||
if sheet is not None:
|
||
if sheet.covers(sku):
|
||
comp_state = sheet.state_for(
|
||
sku, our_price=cur_price,
|
||
max_age_hours=_rules.competitor_sheet_max_age_hours)
|
||
elif _st.competitor_sheet_only:
|
||
comp_state = unavailable(
|
||
f"N/A — {sku} is not covered by the competitor sheet; "
|
||
f"{sheet.coverage_note()}")
|
||
|
||
if comp_state is None:
|
||
if comp_snap is not None:
|
||
comp_state = from_apify(comp_snap, our_price=cur_price, max_age_hours=max_age)
|
||
else:
|
||
comp_state = unavailable(
|
||
"competitor scrape not run for this SKU"
|
||
if not with_competitive else "competitor scrape failed or returned nothing")
|
||
# Cross-check against the sibling workbook tool's own scrape of the same ASIN. Apify
|
||
# stays authoritative (it is the only source carrying a seller id, so the only one that
|
||
# can tell WON from LOST_PRICE) — this exists so that when the two disagree, the
|
||
# disagreement is logged and carried on the state instead of surfacing later as a
|
||
# workbook and a recommendation that contradict each other.
|
||
# Skipped when the state already came from the sheet: the sheet IS the Playwright run's
|
||
# output, so cross-checking it against that same scraper's cache compares a source to
|
||
# itself and can only ever agree.
|
||
from pricing_agent.competitive_state import SOURCE_SHEET
|
||
if comp_state.source != SOURCE_SHEET:
|
||
try:
|
||
other = from_playwright_cache(
|
||
r.asin, our_price=cur_price, zip_code=_st.apify_zip_code,
|
||
max_age_hours=max_age)
|
||
if other is not None:
|
||
comp_state = reconcile(comp_state, other, sku=sku,
|
||
material_pct=_rules.competitor_undercut_material_pct)
|
||
except Exception: # a cross-check must never be able to fail a verdict
|
||
logger.debug("competitor cross-check skipped for %s", sku, exc_info=True)
|
||
comp_meta["state"] = comp_state.status.value
|
||
comp_meta["state_usable"] = comp_state.usable
|
||
comp_meta["state_source"] = comp_state.source
|
||
comp_meta["state_basis"] = comp_state.basis
|
||
comp_meta["state_why_unused"] = comp_state.unusable_because
|
||
# So the UI can say "N/A — not in the covered line" rather than a bare dash.
|
||
comp_meta["sheet_covers_sku"] = bool(sheet is not None and sheet.covers(sku))
|
||
comp_meta["sheet_line"] = sheet.line if sheet is not None else None
|
||
comp_meta["sheet_covered_skus"] = len(sheet.rows) if sheet is not None else 0
|
||
|
||
el = (r.elasticity or {}).get("elasticity") or FALLBACK_ELASTICITY
|
||
be = r.break_even or 0.0
|
||
# The floor that actually matters: the highest of accounting / ad-inclusive /
|
||
# empirical break-even. Everything below it loses money on every unit.
|
||
floor_info = price_floor(r, cur_price)
|
||
safe_floor = floor_info["floor"] or (be * 1.12 if be > 0 else 0.0)
|
||
extra = {}
|
||
if safe_floor > 0:
|
||
extra["min_safe"] = safe_floor
|
||
if r.profit_optimal_price:
|
||
extra["max_profit"] = r.profit_optimal_price
|
||
if r.best_observed_price:
|
||
extra["best_observed"] = r.best_observed_price
|
||
# A price that matches the cheapest RIVAL, offered as a candidate only when competitor
|
||
# state is usable. It is a candidate, not a decision: the guardrails downstream floor it
|
||
# at break-even x1.02 and cap the move at ±5%, so this key can never execute a price
|
||
# below break-even however cheap the rival is.
|
||
if comp_state.usable and comp_state.competitor_min:
|
||
extra["comp_match"] = comp_state.competitor_min
|
||
# comp_median deliberately NOT passed — the scenario ECONOMICS (units, revenue, profit)
|
||
# are still computed from our own fee stack and elasticity alone. A competitor price can
|
||
# add a candidate price to evaluate; it never alters the maths applied to one.
|
||
scen = _scenarios(cur_price, max(units_day, 0.1), inventory, fp, el, None, extra)
|
||
|
||
# Dated inventory urgency — computed before the decision so the branches can
|
||
# say "out on the 12th, restock on the 26th" instead of just "cover is low".
|
||
outlook = _inventory_outlook(inv_projection, units_day, today)
|
||
|
||
# Cover days, reconciled ONCE, here — before the decision that depends on it.
|
||
#
|
||
# COSMOS's own `coverDays` IS the figure, by decision: the Inventory Planning grid is the
|
||
# number the business reads and plans against, so this dashboard must not quote a different
|
||
# one for the same SKU. It counts inbound stock against a 7-day velocity, which is why it
|
||
# runs longer than what is physically on the shelf — for UBMICROFIBERDUVETTWINWHITE, 78
|
||
# days on (3,999 on hand + 1,030 inbound) / 64 a day, against 63 on-hand-only.
|
||
#
|
||
# The on-hand figure is still computed, still shown beside it, and is still the FALLBACK,
|
||
# because COSMOS returns `coverDays: 0` on some very low-velocity SKUs that are in fact
|
||
# sitting on months of stock (UBMICROFIBERBS4PCFULLGREY: 167 units, 334 real days, COSMOS
|
||
# said 0). A zero with stock on hand is a missing answer, not a full shelf, and taking it
|
||
# literally silences both inventory rules.
|
||
#
|
||
# Trade-off, stated: matching COSMOS means a stockout risk is judged on stock that has not
|
||
# landed yet. A shipment that slips leaves this reading optimistic.
|
||
cosmos_cover = r.cover_days
|
||
onhand_cover = int(round(inventory / units_day)) if units_day > 0 else 0
|
||
projected_cover = int(cosmos_cover) if cosmos_cover else onhand_cover
|
||
cur_cover = projected_cover
|
||
if cosmos_cover in (0, None) and onhand_cover:
|
||
logger.info("%s: COSMOS reported cover_days=%r with %.0f units on hand — falling back "
|
||
"to the on-hand figure of %dd", sku, cosmos_cover, inventory, cur_cover)
|
||
|
||
action, rec_key, cons_key, aggr_key, reasons, root, objective = _decide(
|
||
r, cur_price, hist, fp, scen, outlook, comp_state, cur_cover)
|
||
# Every SKU the competitor rules touched, named, with the rule that fired. This is the
|
||
# audit trail for the change: without it, "did the competitor wiring move this price?" is
|
||
# only answerable by re-running with the feature off.
|
||
comp_reasons = [c for c in reasons
|
||
if c in ("BUYBOX_SUPPRESSED", "COMPETITOR_UNDERCUT")]
|
||
if comp_reasons:
|
||
logger.info("competitor rule fired for %s: %s (state=%s source=%s, action=%s)",
|
||
sku, ",".join(comp_reasons), comp_state.status.value,
|
||
comp_state.source, action)
|
||
|
||
# Score the profit model against this SKU's own held-out history. Cheap — it
|
||
# reuses the 180 days already fetched — and it is the only thing that turns
|
||
# "the engine says X" into "the engine has been within Y% on this SKU".
|
||
from pricing_agent.backtest import trust_score, walk_forward
|
||
storage_30d = _storage_30d(svc, sku, cur_price, units_day * 30, inventory)
|
||
try:
|
||
bt = walk_forward(history, fp, storage_30d=storage_30d, list_price=cur_price)
|
||
except Exception:
|
||
logger.warning("backtest failed for %s", sku)
|
||
bt = None
|
||
trust = trust_score(bt, r.elasticity, r.price_bands,
|
||
has_costs=not (fp["cost"] <= 0 and fp["fba"] <= 0))
|
||
tier, score = _confidence(action, hist, r)
|
||
|
||
s = scen.set_index("scenario")
|
||
cur_row, rec_row = s.loc["current"], s.loc[rec_key]
|
||
|
||
# ── SAFETY: clamp the target, then step toward it ────────────────────────
|
||
# `target_price` is where the evidence says to end up; `rec_price` is the most
|
||
# this move may travel today. Splitting them keeps every change small enough
|
||
# to read a real demand response before the next one — which is also the only
|
||
# way to earn the price variation a trustworthy elasticity needs.
|
||
obs_max = r.observed_price_max
|
||
clamps = []
|
||
stepped = False
|
||
|
||
if action == "Investigate":
|
||
# The root-cause gate said price is not the lever. Lifting the price to a
|
||
# floor here would contradict that verdict, so hold and report the breach.
|
||
target_price = rec_price_final = cur_price
|
||
if safe_floor and cur_price < safe_floor:
|
||
clamps.append(f"selling ${safe_floor - cur_price:.2f} BELOW the "
|
||
f"${safe_floor:.2f} floor ({floor_info['binding']} "
|
||
f"break-even) — but see the verdict: price is not the "
|
||
f"lever here")
|
||
else:
|
||
target_price = float(rec_row["price"])
|
||
if obs_max and target_price > obs_max * OBSERVED_HEADROOM:
|
||
target_price = round(obs_max * OBSERVED_HEADROOM, 2)
|
||
clamps.append(f"capped at ${target_price:.2f} — {OBSERVED_HEADROOM:.0%} of "
|
||
f"the highest price with a real sample behind it "
|
||
f"(${obs_max:.2f})")
|
||
if safe_floor and target_price < safe_floor:
|
||
target_price = round(safe_floor, 2)
|
||
clamps.append(f"lifted to the ${safe_floor:.2f} floor "
|
||
f"({floor_info['binding']} break-even + safety)")
|
||
step_cap = cur_price * MAX_STEP_PCT
|
||
rec_price_final = target_price
|
||
if abs(target_price - cur_price) > step_cap + 0.005:
|
||
direction = math.copysign(1.0, target_price - cur_price)
|
||
rec_price_final = round(cur_price + direction * step_cap, 2)
|
||
# Rounding to cents must never push the move back over the cap.
|
||
if abs(rec_price_final - cur_price) > step_cap:
|
||
rec_price_final = round(rec_price_final - direction * 0.01, 2)
|
||
stepped = abs(rec_price_final - target_price) > 0.005
|
||
if stepped:
|
||
clamps.append(f"moving ${rec_price_final:.2f} now ({MAX_STEP_PCT:.0%} step "
|
||
f"cap); re-evaluate in 14 days on the way to "
|
||
f"${target_price:.2f}")
|
||
if abs(rec_price_final - cur_price) / cur_price < 0.005:
|
||
action, rec_price_final = "Maintain", cur_price
|
||
|
||
delta = float(rec_price_final - cur_price)
|
||
delta_pct = delta / cur_price * 100 if cur_price else 0.0
|
||
|
||
# 30-day actuals, all from the SAME window aggregate so they reconcile.
|
||
spend_30d = w30["ad"]
|
||
revenue_30d = w30["revenue"]
|
||
actual_profit_30d = w30["profit"]
|
||
actual_units_30d = w30["units"]
|
||
tacos_frac = spend_30d / revenue_30d if revenue_30d else 0.0
|
||
# Amazon Advertising metrics, proxied by COSMOS. `sales-insight` knows only
|
||
# total marketing spend (hence TACoS); this endpoint carries the attributed
|
||
# side — ad sales, ACoS, ROAS, CPC and the campaign budget.
|
||
stage(0.90, "advertising metrics")
|
||
try:
|
||
mk = svc.get_marketing(sku, days=30)
|
||
except Exception:
|
||
logger.warning("marketing metrics failed for %s", sku)
|
||
mk = None
|
||
ppc = dict(
|
||
spend_30d=round(spend_30d), spend_day=round(w30["ad_day"], 2),
|
||
revenue_30d=round(revenue_30d), units_30d=int(round(actual_units_30d)),
|
||
tacos=round(tacos_frac * 100, 1),
|
||
# attributed side — None only when the SKU has no campaigns
|
||
ad_spend_30d=round(mk.ad_spend) if mk else None,
|
||
ad_sales_30d=round(mk.ppc_sales) if mk else None,
|
||
ad_units_30d=int(round(mk.units)) if mk else None,
|
||
acos=round(mk.acos * 100, 1) if mk else None,
|
||
roas=round(mk.roas, 2) if mk else None,
|
||
cpc=round(mk.cpc, 2) if mk else None,
|
||
clicks=int(round(mk.clicks)) if mk else None,
|
||
impressions=int(round(mk.impressions)) if mk else None,
|
||
conversion=round(mk.conversion * 100, 2) if mk else None,
|
||
daily_budget=round(mk.daily_budget) if mk else None,
|
||
budget_utilization=round(mk.budget_utilization * 100, 1) if mk else None,
|
||
# share of units the ads can actually claim — the rest is organic
|
||
ad_share_pct=(round(mk.units / actual_units_30d * 100, 1)
|
||
if mk and actual_units_30d else None),
|
||
)
|
||
|
||
# Scenario economics, anchored to the price customers actually paid.
|
||
# The baseline uses a 90-day window, not 30: a single month is mostly noise
|
||
# (this SKU's months ranged $-0.54 to $+2.75 profit/unit), and the baseline
|
||
# is what every projection is compared against.
|
||
wb = window_agg(DEFAULT_WINDOW_DAYS)
|
||
scale = 30.0 / (wb["days"] or 30)
|
||
grid = {row["scenario"]: row["price"] for _, row in scen.iterrows()}
|
||
# Only add the recommendation as its own row when no candidate already sits at
|
||
# that price — otherwise the table shows the same price twice.
|
||
if not any(abs(p - rec_price_final) < 0.005 for p in grid.values()):
|
||
grid["recommended"] = rec_price_final
|
||
rec_econ_key = "recommended"
|
||
else:
|
||
rec_econ_key = next(k for k, p in grid.items()
|
||
if abs(p - rec_price_final) < 0.005)
|
||
realized = (wb["revenue"] / wb["units"]) if wb["units"] else cur_price
|
||
# The backtest chose the calibration; apply exactly that one here, so the
|
||
# table on screen is the model that was actually scored.
|
||
gap_pu = 0.0
|
||
if bt and bt["selected"] == "anchored_gap" and wb["units"]:
|
||
from pricing_agent.backtest import unmodeled_gap_per_unit
|
||
gap_pu = unmodeled_gap_per_unit(
|
||
{"units": wb["units"], "price": realized, "ad": wb["ad"],
|
||
"profit": wb["profit"], "days": wb["days"],
|
||
"units_day": wb["units_day"]},
|
||
fp, storage_30d * wb["days"] / 30.0, r.ad_cost_model)
|
||
scen_econ, scen_meta = _scenario_economics(
|
||
cur_price, max(wb["units_day"] or units_day, 0.1), fp, el,
|
||
(wb["ad"] / wb["revenue"] if wb["revenue"] else 0.0), storage_30d, grid,
|
||
actual_30d=wb["profit"] * scale, actual_rev=wb["revenue"] * scale,
|
||
actual_ad=wb["ad"] * scale, actual_units=wb["units"] * scale,
|
||
realized_price=realized, ad_model=r.ad_cost_model, bands=r.price_bands,
|
||
gap_per_unit=gap_pu)
|
||
|
||
# Headline profit/impact comes from the same ad-inclusive net as the tables, so
|
||
# the tiles, queue sort and decision card all agree with the Scenarios tab.
|
||
#
|
||
# BOTH SIDES OF THIS SUBTRACTION MUST BE THE SAME MODEL AT TWO PRICES.
|
||
#
|
||
# The current row deliberately carries two readings: COSMOS's BOOKED profit, priced at the
|
||
# average customers actually paid over the window, and its MODELLED twin at today's price.
|
||
# Subtracting a projection from the booked figure mixes those bases and measures the wrong
|
||
# move. Observed on UBMICROFIBERDUVETTWINWHITE: today $17.09, recommendation $17.94, but
|
||
# the booked row sits at $18.18 — so `rec - booked` priced a CUT from $18.18 and returned
|
||
# -$78 on a +5% raise. The tile then showed "Profit opportunity $0" next to a live RAISE
|
||
# recommendation, and the queue sorted on that number.
|
||
#
|
||
# Against the modelled baseline the same move is +$1,207, which is the actual effect of
|
||
# going from $17.09 to $17.94. The booked figure keeps its own job — anchoring the
|
||
# Scenarios tab to reality — it is just not a baseline for a price DELTA.
|
||
_econ = {x["key"]: x for x in scen_econ}
|
||
_cur_econ = _econ.get("current", {})
|
||
cur_net_30d = modelled_net_30d(_cur_econ, float(cur_row["profit_30d"]))
|
||
# Applied to the recommendation too, so a "Maintain" that lands on the current row
|
||
# subtracts a value from itself and correctly reports zero rather than the booked/modelled
|
||
# difference.
|
||
rec_net_30d = modelled_net_30d(_econ.get(rec_econ_key, {}), cur_net_30d)
|
||
impact_30d = rec_net_30d - cur_net_30d
|
||
# The booked figure is still what the card reports as "profit today".
|
||
cur_booked_30d = float(_cur_econ.get("net_30d", cur_row["profit_30d"]))
|
||
|
||
stage(0.94, "finalizing")
|
||
# Prefer the real Amazon listing title (from INVP); fall back to brand, then SKU.
|
||
# Never append the SKU — it's already shown next to the title in the UI.
|
||
marketplace, brand = "AMAZON_USA", None
|
||
try:
|
||
product = svc.get_product(sku)
|
||
if product:
|
||
marketplace = product.marketplace or marketplace
|
||
brand = product.brand or product.manufacturer
|
||
except Exception:
|
||
pass
|
||
title = (invp_title or "").strip() or brand or sku
|
||
|
||
cfg = dict(
|
||
sku=sku, title=title, marketplace=marketplace, objective=objective,
|
||
base=cur_price, cost=fp["cost"], fba=fp["fba"],
|
||
velocity=units_day, elasticity=el, inventory=inventory, comp_offset=1.0,
|
||
rating=None, reviews=None, inv_class="alpha", bsr=None,
|
||
)
|
||
|
||
explanation = _explain(sku, action, cur_price, rec_row, cur_row, reasons, tier, r)
|
||
|
||
# `cosmos_cover` / `onhand_cover` / `projected_cover` / `cur_cover` were reconciled above,
|
||
# before `_decide`, so the tile below and the rule that fired read the same number. Do not
|
||
# recompute them here — that is precisely how the two came to disagree.
|
||
rec_units_day = float(_econ.get(rec_econ_key, {}).get("units_day") or units_day)
|
||
if rec_units_day > 0 and units_day > 0:
|
||
rec_cover = int(round(cur_cover * units_day / rec_units_day))
|
||
else:
|
||
rec_cover = cur_cover
|
||
|
||
return dict(
|
||
cfg=cfg, hist=hist, change_dates=_change_dates(hist), scenarios=scen,
|
||
forecast=_forecast(hist, tier, today), competitors=comp_df,
|
||
action=action, rec_key=rec_key,
|
||
strategy_keys={"Conservative": cons_key, "Recommended": rec_key,
|
||
"Aggressive": aggr_key},
|
||
reasons=reasons, root_cause=root, risks=RISKS[action],
|
||
confidence=(tier, score),
|
||
current_price=float(cur_price), rec_price=float(rec_price_final),
|
||
# where the evidence says to end up, vs how far this single move goes
|
||
target_price=float(target_price), stepped=stepped, clamps=clamps,
|
||
delta=delta, delta_pct=float(delta_pct), impact_30d=impact_30d,
|
||
cover_days=cur_cover, rec_cover_days=rec_cover,
|
||
cover_days_onhand=onhand_cover, cover_days_projected=projected_cover,
|
||
outlook=outlook, stockout_days_30d=_stockout_days(hist),
|
||
break_even=round(be, 2),
|
||
# Floors: the guardrail may never sit below the price at which a unit
|
||
# actually stops losing money once advertising is counted.
|
||
break_even_with_ads=r.break_even_with_ads,
|
||
break_even_empirical=r.break_even_empirical,
|
||
break_even_ad_curve=r.break_even_ad_curve,
|
||
ad_curve_unrecoverable=r.ad_curve_unrecoverable,
|
||
contribution_per_dollar=r.contribution_per_dollar,
|
||
floor_info=floor_info,
|
||
min_price=round(max(safe_floor, be * 1.05, 0.01), 2),
|
||
max_price=round(max(cur_price, safe_floor, be * 1.05) * 1.25, 2),
|
||
observed_price_min=r.observed_price_min, observed_price_max=r.observed_price_max,
|
||
elasticity_actionable=r.elasticity_actionable,
|
||
elasticity_detail=r.elasticity,
|
||
profit_optimal_blocked_reason=r.profit_optimal_blocked_reason,
|
||
profit_optimal_unconstrained=r.profit_optimal_unconstrained,
|
||
ad_cost_model=r.ad_cost_model, price_bands=r.price_bands,
|
||
# how wrong the model has been on THIS SKU, and what that permits
|
||
backtest=bt, trust=trust, gap_per_unit=round(gap_pu, 3),
|
||
units_day=float(units_day),
|
||
rec_units_day=rec_units_day,
|
||
profit_30d=cur_net_30d, rec_profit_30d=rec_net_30d,
|
||
po=None, explanation=explanation, ppc=ppc,
|
||
asin=r.asin, comp_median_now=None, comp_meta=comp_meta, # chart stays competitor-free
|
||
referral_pct=fp["referral_pct"], returns_pct=fp["returns_pct"],
|
||
variable=fp["variable"],
|
||
# bulk-reconciled, ad-inclusive scenario economics
|
||
scen_econ=scen_econ, scen_best_key=scen_meta["best_key"],
|
||
scen_summary=scen_meta["summary"], storage_30d=round(storage_30d),
|
||
actual_profit_30d=round(actual_profit_30d),
|
||
scen_calibrated=scen_meta["calibrated"], scen_factor=scen_meta["factor"],
|
||
# daily actuals + params so the UI window filter (7/14/30/90d) can recompute
|
||
# everything from one consistent averaging window
|
||
daily=daily, scen_window=DEFAULT_WINDOW_DAYS,
|
||
recompute=dict(cur_price=float(cur_price), fp=fp, el=el,
|
||
storage_30d=float(storage_30d), grid=grid,
|
||
ad_model=r.ad_cost_model, bands=r.price_bands,
|
||
gap_per_unit=gap_pu),
|
||
# real COSMOS INVP forward projection (weekly units/value/cover/arrivals)
|
||
inv_projection=inv_projection, min_po=min_po, invp_stats=invp_stats,
|
||
# extra live evidence for the AI-reasoning tab
|
||
evidence=dict(
|
||
actual_profit_per_unit=r.actual_profit_per_unit,
|
||
unprofitable_months=r.unprofitable_months,
|
||
best_observed_price=r.best_observed_price,
|
||
best_observed_profit_day=r.best_observed_profit_day,
|
||
unmodeled_cost_gap=r.unmodeled_cost_gap,
|
||
suggested_price=r.suggested_price,
|
||
trend=r.trend,
|
||
),
|
||
)
|
||
|
||
|
||
def scenarios_for_window(d: dict, days: int) -> tuple[list[dict], dict, dict]:
|
||
"""Recompute scenario economics using the last `days` of daily actuals as the
|
||
averaging window. Everything (velocity, revenue, ad, profit) derives from this one
|
||
window, so units/day and revenue always reconcile. Returns (econ, meta, facts)."""
|
||
daily = d.get("daily") or []
|
||
rp = d.get("recompute") or {}
|
||
w = daily[-days:] if daily else []
|
||
n = len(w) or 1
|
||
u = sum(x["units"] for x in w)
|
||
rev = sum(x["revenue"] for x in w)
|
||
ad = sum(x["ad"] for x in w)
|
||
prof = sum(x["profit"] for x in w)
|
||
facts = {"days": n, "units_day": round(u / n), "revenue_day": round(rev / n),
|
||
"ad_day": round(ad / n), "profit_day": round(prof / n),
|
||
"avg_price": round(rev / u, 2) if u else 0.0}
|
||
# Normalise the window rate to a 30-day basis so the /30d columns stay comparable.
|
||
scale = 30.0 / n
|
||
tacos_frac = ad / rev if rev else 0.0
|
||
econ, meta = _scenario_economics(
|
||
rp.get("cur_price", d["current_price"]), max(u / n, 0.1), rp["fp"], rp["el"],
|
||
tacos_frac, rp.get("storage_30d", 0.0), rp["grid"],
|
||
actual_30d=prof * scale, actual_rev=rev * scale, actual_ad=ad * scale,
|
||
actual_units=u * scale,
|
||
realized_price=(rev / u) if u else None,
|
||
ad_model=rp.get("ad_model"), bands=rp.get("bands"),
|
||
gap_per_unit=rp.get("gap_per_unit", 0.0))
|
||
return econ, meta, facts
|
||
|
||
|
||
# ---------------------------------------------------------------- entry point
|
||
def get_live_data(skus: tuple[str, ...], with_competitive: bool = False,
|
||
progress_cb=None) -> dict:
|
||
"""Analyze `skus` against live COSMOS. Failed SKUs land in ``errors``."""
|
||
from config.settings import get_settings
|
||
from pricing_agent.analyze import _build_service
|
||
|
||
svc = _build_service(get_settings())
|
||
details: dict[str, dict] = {}
|
||
errors: dict[str, str] = {}
|
||
total = len(skus)
|
||
for i, sku in enumerate(skus):
|
||
def on_stage(frac: float, label: str, i=i, sku=sku):
|
||
# Map per-SKU stage fraction onto the overall bar.
|
||
if progress_cb:
|
||
overall = (i + frac) / total if total else 1.0
|
||
prefix = f"{sku}" + (f" ({i + 1}/{total})" if total > 1 else "")
|
||
progress_cb(overall, f"{prefix} — {label}")
|
||
try:
|
||
details[sku] = build_live_sku(svc, sku, with_competitive=with_competitive,
|
||
on_stage=on_stage)
|
||
except Exception as e: # one bad SKU must not sink the load
|
||
logger.exception("live build failed for %s", sku)
|
||
errors[sku] = str(e)
|
||
if progress_cb:
|
||
progress_cb(1.0, f"Done — {total} product{'s' if total != 1 else ''} analyzed")
|
||
|
||
rows = []
|
||
for sku, d in details.items():
|
||
rows.append({
|
||
"sku": sku, "title": d["cfg"]["title"], "marketplace": d["cfg"]["marketplace"],
|
||
"objective": d["cfg"]["objective"], "action": d["action"],
|
||
"current_price": d["current_price"], "rec_price": d["rec_price"],
|
||
"delta": d["delta"], "delta_pct": d["delta_pct"], "impact_30d": d["impact_30d"],
|
||
"cover_days": d["cover_days"], "confidence_tier": d["confidence"][0],
|
||
"confidence_score": d["confidence"][1],
|
||
"trust_tier": (d.get("trust") or {}).get("tier", "None"),
|
||
"primary_reason": REASON_LABELS.get(d["reasons"][0], d["reasons"][0]),
|
||
})
|
||
summary = (pd.DataFrame(rows).sort_values("impact_30d", ascending=False,
|
||
key=lambda sr: sr.abs())
|
||
if rows else pd.DataFrame(columns=[
|
||
"sku", "title", "marketplace", "objective", "action", "current_price",
|
||
"rec_price", "delta", "delta_pct", "impact_30d", "cover_days",
|
||
"confidence_tier", "confidence_score", "trust_tier",
|
||
"primary_reason"]))
|
||
return {"summary": summary, "details": details, "errors": errors}
|