"""Utopia Pricing Agent โ one-page dashboard (CRAI design system).
Utopia Brands look (crai.utopiabrands.com): off-white cards on warm cream paper,
teal primary, coral accent, navy chrome. Sidebar with filters, stat tiles, action
pills, and a recommendation queue where clicking a SKU row drops down the advanced
analysis (price & demand, inventory, scenarios, competitors, PPC, costs, AI
reasoning). All numbers come from the live COSMOS pipeline โ fees, 6-month
history, elasticity, actual profit.
Run: streamlit run app.py
"""
from __future__ import annotations
import hmac
import os
import sys
from pathlib import Path
# Make src/ + project root importable when launched via `streamlit run`.
ROOT = Path(__file__).resolve().parent
for p in (str(ROOT / "src"), str(ROOT)):
if p not in sys.path:
sys.path.insert(0, p)
from datetime import date, datetime, timedelta
import pandas as pd
import streamlit as st
from plotly.subplots import make_subplots
import plotly.graph_objects as go
from dashboard import theme
# Ranking rules live in the data layer so they can be tested without booting
# Streamlit โ importing app.py would execute the whole page.
from dashboard.line import LINE_SORTS, line_sort_key as _line_sort_key
from dashboard.live_data import (
FALLBACK_ELASTICITY, REASON_LABELS, REFERRAL_PCT, VARIABLE, modelled,
reproject_inventory,
scenarios_for_window,
)
TODAY = date.today()
st.set_page_config(page_title="Utopia Pricing Agent", page_icon="๐งญ",
layout="wide", initial_sidebar_state="auto")
theme.register_template()
def _require_password() -> None:
"""Shared-password gate, active only when APP_PASSWORD is set.
The app authenticates to COSMOS with the credentials in .env and shows live
cost, margin and sales data, so on a shared network the URL alone is enough
for anyone to read it. Set APP_PASSWORD before binding to 0.0.0.0. Left unset
(the default, and the local-only case) nothing changes.
This is a doorlock, not an identity system: one shared secret, no accounts,
no audit of who looked. For anything beyond an internal LAN, put it behind a
real reverse proxy with SSO.
"""
expected = os.environ.get("APP_PASSWORD")
if not expected or st.session_state.get("_authed"):
return
st.markdown(
'๐ Utopia Pricing Agent
'
'
This dashboard shows live cost and margin data. '
'Enter the shared password to continue.
'
f'๐ก Recommendation: '
f'{body}
',
unsafe_allow_html=True,
)
# Averaging-window filter โ everything below derives from this one window.
wl = st.segmented_control(
"Averaging window", list(WINDOW_OPTS), key=f"win_{key_prefix}",
default=DEFAULT_WINDOW, help="Units/day, revenue, ad spend and profit are all "
"averaged over this window, so they always reconcile.")
days = WINDOW_OPTS.get(wl or DEFAULT_WINDOW, 90)
if days == d.get("scen_window", 90):
econ = base_econ
summary = d.get("scen_summary")
best_key = d.get("scen_best_key")
facts = None
else:
econ, meta, facts = scenarios_for_window(d, days)
summary = meta.get("summary")
best_key = meta.get("best_key")
cur = next((x for x in econ if x["key"] == "current"), econ[0])
price_check = (cur["revenue_30d"] / cur["units_30d"]) if cur["units_30d"] else 0.0
best = next((x for x in econ if x["key"] == best_key), None)
# ONE caveat block, not four scattered ones. Every projection on this tab
# carries the same two qualifications, so they are stated once, together.
el = d.get("elasticity_detail") or {}
caveats = []
if not d.get("elasticity_actionable"):
caveats.append("**Elasticity is not usable for pricing on this SKU** โ "
+ (el.get("why") or "the fit does not establish a price/volume link")
+ ". The recommendation comes from prices this product has "
"actually run, not from the curve below.")
if d.get("profit_optimal_blocked_reason"):
line = f"**No profit-optimal price published** โ {d['profit_optimal_blocked_reason']}."
if d.get("profit_optimal_unconstrained"):
line += (f" Taken literally the fit implies pricing at "
f"${d['profit_optimal_unconstrained']:.2f}, which is itself evidence "
f"the fit is unusable.")
caveats.append(line)
if caveats:
st.warning(esc("โ ๏ธ " + "\n\n".join(caveats)))
a1, a2, a3 = st.columns(3)
a1.metric(f"Units/day (last {days}d)", f"{cur['units_day']:,}",
f"avg sold ${price_check:,.2f}/unit", delta_color="off")
a2.metric("Net profit /30d", f"${cur['net_30d']:,.0f}",
f"on ${cur['revenue_30d']:,.0f} revenue", delta_color="off")
if best:
lbl = "Best price = hold" if best["key"] == "current" else "Best price (projected)"
a3.metric(lbl, f"${best['price']:.2f}",
f"{best['net_vs_current']:+,.0f}/30d vs now", delta_color="off")
rows_html = []
for x in econ:
is_best = x["key"] == best_key
is_cur = x["key"] == "current"
emoji = "๐ฐ" if is_cur else ("๐" if x["price"] > cur["price"] else "๐")
row_bg = "#eef6f3" if is_best else ("#fbf6ee" if is_cur else "transparent")
star = " โญ" if is_best else ""
net_color = "#0a6f65" if x["net_30d"] >= 0 else "#c9442b"
# Current row: show the AVG PRICE SOLD (revenue รท units) so price ร units ร 30 =
# revenue reconciles. It sits below list when there are promos, and varies by
# window because the avg selling price differed period to period.
if is_cur and x["units_30d"]:
eff = x["revenue_30d"] / x["units_30d"]
price_cell = (f'${eff:,.2f}'
f'avg sold ยท '
f'{days}d
'
f'list '
f'${x.get("list_price", x["price"]):.2f}
')
move_cell = 'actual'
else:
# Provenance: a price this SKU has really run is evidence; one it has
# never run is extrapolation, and the table must not look the same.
obs = x.get("observed_days") or 0
tag = (f''
f'โ ran {obs}d
' if obs else
'never tested
')
price_cell = f'${x["price"]:.2f}{tag}'
move_cell = _price_move_html(x["delta_pct"])
rows_html.append(
f''
f'| {emoji} {esc(x["label"])}{star} | '
f'{price_cell} | '
f'{move_cell} | '
f'{x["units_day"]:,} | '
f'${x["revenue_30d"]:,.0f} | '
f'${x["ad_30d"]:,.0f} | '
f'${x["net_30d"]:,.0f} | '
f'{x["net_margin_pct"]:.1f}% | '
# No esc() inside the raw HTML table โ Streamlit does not run LaTeX on
# these cells, so escaping would render a literal backslash.
+ ("" if compact else
f'{x["advice"]} | ')
+ '
'
)
# Column headers with hover tooltips explaining how each number is computed.
elas = d["cfg"].get("elasticity", -1.3)
eld = d.get("elasticity_detail") or {}
el_note = (f"elasticity {elas:.2f}"
+ (f" (95% CI {eld['ci_low']} to {eld['ci_high']}, "
f"{'usable' if d.get('elasticity_actionable') else 'NOT statistically usable'})"
if eld.get("ci_low") is not None else ""))
anchor_p = cur.get("price", d["current_price"])
tacos_now = cur["ad_30d"] / cur["revenue_30d"] * 100 if cur["revenue_30d"] else 0.0
def th(label, tip, left=False):
align = "left" if left else "center"
return (f''
f'{label}'
f' โ | ')
headers = (
th("Scenario", f"Current = your actual last-{days}-day facts. Every other row is "
f"a what-if. 'โ ran Nd' means this price really has been live for "
f"N days โ that row is evidence, not a projection.", left=True)
+ th("Price", "Current = your AVERAGE price sold over the window (revenue รท "
"units); your list price is shown beneath it. Other rows = the "
"price you would set.")
+ th("Move", "Percentage change vs your current list price.")
+ th("Units/day", f"Projected demand = current units ร (new price รท the price "
f"customers actually paid, ${anchor_p:,.2f}) ^ elasticity. "
f"Anchoring on the paid price โ not list โ is what keeps every "
f"row on one basis. {el_note}.")
+ th("Revenue/30d", "Units/day ร 30 ร price.")
+ th("Ad spend/30d", f"Ad cost PER UNIT ร units, with per-unit cost fitted "
f"against price from your own history (a higher price "
f"converts worse, so each sale costs more). Current TACoS "
f"{tacos_now:.1f}%.")
+ th("Net profit/30d", "Take-home โ storage โ ad spend. No realization factor: "
"a single multiplier fitted at one price also distorts the "
"difference between prices, which is the whole question.")
+ th("Net margin", "Net profit รท revenue.")
+ ("" if compact else th("Marketing advice",
"Plain-language read of the price / volume / ad-spend "
"trade-off for this row.", left=True))
)
st.markdown(
''
'' + headers + '
'
+ "".join(rows_html) + '
{pd.Timestamp(r["date"]):%m/%d/%Y} | ' for r in rows)
at_new, delta = [], []
for r in rows:
cover = float(r["cover_new"] or 0)
at_new.append(
f''
f' {r["units_new"]:,.0f} '
f'${r["value_new"]:,.0f} '
f'{cover:,.0f}'
f'{r["arrival"]:,.0f} | '
)
# The delta row is intentionally UNSHADED: colour here would compete with the cover
# bands above and imply a second, unrelated scale.
du, dc = r["delta_units"], (r["cover_new"] or 0) - (r["cover_now"] or 0)
colour = (theme.DELTA_GOOD if du > 0 else
("#c9442b" if du < 0 else theme.MUTED))
delta.append(
f'{du:+,.0f} '
f'{dc:+,.0f} d | '
)
return (
''
f'| At the recommended price | Daily Sale | {heads}
|---|
'
""
# --- row 1: the shelf at the new price
""
f'${d["rec_price"]:,.2f} '
f'was ${d["current_price"]:,.2f} ยท {d["delta_pct"]:+.1f}% '
f'our projection ยท class {cls.capitalize()} | '
f'{u_new:,.0f} '
f'was {u_now:,.0f}/day '
f'{(u_new / u_now - 1) if u_now else 0:+.1%} demand | '
f'{"".join(at_new)}'
"
"
# --- row 2: the difference against COSMOS's own projection
""
'Change '
'vs today\'s price '
'units ยท cover days | '
' | '
f'{"".join(delta)}'
"
"
"
%{y:,.0f} unitstoday's price"))
fig.add_trace(go.Scatter(
x=dates, y=new, name=f"At ${d['rec_price']:,.2f} (recommended)",
mode="lines+markers", line=dict(color=theme.SERIES["alt"], width=2, dash="dash"),
hovertemplate="%{x|%d %b}
%{y:,.0f} unitsrecommended price"))
fig.add_hline(y=0, line_color=theme.STATUS["critical"], line_dash="dot", line_width=1)
# Arrivals move the shelf on BOTH lines, so they are annotated once, on the axis.
for r, dt in zip(rows, dates):
if r["arrival"] > 0:
fig.add_annotation(x=dt, y=0, yref="y", text=f"๐ +{r['arrival']:,.0f}",
showarrow=False, yshift=-14,
font=dict(size=10, color=theme.INK2))
# The first week the shelf empties under the new price โ the whole point of the chart.
out = next((r for r in rows if r["stockout"]), None)
if out:
fig.add_vline(x=pd.Timestamp(out["date"]), line_color=theme.STATUS["critical"],
line_dash="dot", line_width=1)
fig.add_annotation(x=pd.Timestamp(out["date"]), y=max(now + new),
text="empty at the new price", showarrow=False,
font=dict(size=10, color=theme.STATUS["critical"]))
fig.update_yaxes(rangemode="tozero", title_text="units")
fig.update_layout(height=300, legend=dict(orientation="h", y=1.12, x=0))
return fig
# Cover-day color bands (Alpha/Beta scheme, matched to the planning sheet)
def _bucket_bg(cover: float, inv_class: str) -> str:
"""Cell shading by projected cover days, per the SKU's Alpha/Beta band scheme.
Thresholds and colours both live in `theme.COVER_BANDS` โ one definition, so the
inventory grid, the cover tile and anything added later cannot drift into shading the
same number three different ways.
"""
return theme.cover_band(cover, inv_class)[0]
def ppc_chart(d) -> go.Figure:
"""Daily PPC spend (thin) with 7-day moving average (bold), last 90 days."""
h = d["hist"].tail(90)
ma = h["ad_spend"].rolling(7, min_periods=1).mean()
fig = go.Figure()
fig.add_trace(go.Scatter(x=h["date"], y=h["ad_spend"], name="Daily spend",
line=dict(color=theme.MUTED, width=1)))
fig.add_trace(go.Scatter(x=h["date"], y=ma, name="7-day average",
line=dict(color=theme.SERIES["us"], width=2)))
fig.update_yaxes(rangemode="tozero", title_text="spend ($/day)")
fig.update_layout(height=280)
return fig
def inventory_table_html(d) -> str:
"""COSMOS INVP weekly projection table โ REAL data from invp-insight dateMap:
projected units, inventory value, cover days (blue), and incoming arrivals."""
proj = d.get("inv_projection") or []
cfg, hist = d["cfg"], d["hist"]
asin = d.get("asin") or f"B0{abs(hash(cfg['sku'])) % 10**8:08d}"
asin_url = _amazon_url(asin, cfg.get("marketplace", "AMAZON_USA"))
asin_html = (f'{asin}' if asin_url else asin)
# COSMOS's OWN daily-sale figures, not ours re-derived from sales-insight history. The
# weekly cells below come straight from the INVP dateMap and always matched; the header did
# not, so one SKU showed 56 / 62 / 65 here against 64 / 130 / 67 in Inventory Planning.
#
# Falls back to the history means only where COSMOS returns nothing, so a gap still shows a
# number rather than a dash โ but the two tools agree wherever COSMOS has an answer.
_st = d.get("invp_stats") or {}
d7 = _st.get("avg_sale") or _st.get("avg_7d") or hist["units"].tail(7).mean()
# The middle figure in COSMOS's header is POTENTIAL daily sale โ its own demand estimate,
# not a trailing average. It is a different quantity from the 30-day mean that used to sit
# here, so it is labelled for what it is.
d_pot = _st.get("potential_daily_sale")
d6m = _st.get("avg_6m") or hist["units"].tail(90).mean()
reviews = f" ({cfg['reviews']:,})" if cfg.get("reviews") else ""
if not proj:
return (''
'No INVP projection available for this product from COSMOS.
')
# Match the COSMOS Inventory Planning header exactly.
#
# `proj[0].inventory` is the FIRST PROJECTED WEEK, which already has that
# week's arrival folded in (5,029 = 3,999 on hand + 1,030 landing), so showing
# it as stock overstated the shelf by the inbound load. COSMOS shows them apart.
#
# Inbound is likewise the IMMEDIATE arrival, not every arrival on the horizon:
# summing the lot gave 1,770 against COSMOS's 1,030 by sweeping in +90 and +650
# that land weeks later. Those are real, so they are reported separately.
on_hand = float((d.get("invp_stats") or {}).get("inventory")
or proj[0]["inventory"] or 0)
total_inbound = float(proj[0]["warehouse_arrival"] or 0)
later_inbound = sum(float(p["warehouse_arrival"] or 0) for p in proj[1:])
heads = "".join(
f'{pd.Timestamp(p["date"]):%m/%d/%Y} | ' for p in proj
)
cells = []
for p in proj:
units = float(p["inventory"] or 0)
value = float(p["inventory_value"] or (units * cfg["cost"]))
cover = float(p["cover_days"]) if p["cover_days"] is not None else 0.0
arr = float(p["warehouse_arrival"] or 0)
cells.append(
f''
f' {units:,.0f} '
f'${value:,.0f} '
f'{cover:,.0f}'
f'{arr:,.0f} | '
)
return (
''
f'| Product Description | Daily Sale | {heads}
|---|
'
""
f'{cfg["title"]} '
f'{cfg["sku"]} ยท {asin_html}{reviews} '
f'๐ฆ {on_hand:,.0f} on hand ยท ๐ {total_inbound:,.0f} inbound'
f'{f" (+{later_inbound:,.0f} later)" if later_inbound else ""} ยท '
f'class {cfg["inv_class"].capitalize()} | '
f'{d7:,.0f} '
+ (f'{d_pot:,.0f} ยท potential ' if d_pot else '')
+ f'{d6m:,.0f} ยท 6mo avg | '
f'{"".join(cells)}'
"
U
'
'
Utopia Pricing
'
'
AI Pricing Agent
โ๏ธ Analyzing against COSMOS
{n} product{'s' if n > 1 else ''} ยท live fees, 6-month
history, elasticity & bulk economics
{bar}
{pct}%
{esc(message)}
""",
unsafe_allow_html=True,
)
except Exception:
pass
_on_progress(0.0, "Connecting to COSMOSโฆ")
data = _load_live(st.session_state.live_skus,
st.session_state.get("live_comp", False), progress=_on_progress)
_loader.empty()
if data["errors"]:
st.warning("Some SKUs could not be analyzed: " + esc(
" ยท ".join(f"**{s}** ({m})" for s, m in data["errors"].items())))
summary, details = data["summary"], data["details"]
for sku in details: # keep statuses across reruns; init the new ones
st.session_state.status.setdefault(sku, "Pending")
if summary.empty:
st.info("No SKUs produced an analysis. Check the warnings above (typical causes: "
"wrong SKU code, or the SKU has no sales/price data in COSMOS).")
st.stop()
# ---------------------------------------------------------------- sidebar filters (cont.)
with st.sidebar:
f_mkt = st.selectbox("Marketplace", ["All"] + sorted(summary["marketplace"].unique()),
format_func=lambda m: "๐ All" if m == "All" else mkt_label(m))
f_conf = st.selectbox(
"Model trust", ["All", "High", "Medium", "Low", "None"],
help="How far this SKU's model may be trusted: backtest accuracy, breadth "
"of price evidence, elasticity usability and data completeness. Only "
"High may be bulk-approved.")
f_stat = st.selectbox("Status", ["All", "Pending", "Approved", "Rejected"])
st.divider()
st.caption("CONTROLS")
kill = st.toggle("๐ Global kill switch",
help="Emergency stop: pauses all approvals.")
st.button("โบ Reset workspace", on_click=reset_workspace, use_container_width=True)
st.divider()
st.caption("DATA STATUS")
st.caption(f"Prices as of {TODAY:%b %d, %Y}\n\n"
f"{len(details)} SKUs ยท 6-month COSMOS history\n\n"
+ ("Competitor prices: included\n\n" if st.session_state.get("live_comp")
else "Competitor prices: not included\n\n")
+ "Engine v1 ยท live COSMOS data")
# ---------------------------------------------------------------- header
hour = datetime.now().hour
greet = "Good morning" if hour < 12 else ("Good afternoon" if hour < 18 else "Good evening")
pending = [s for s, v in st.session_state.status.items()
if v == "Pending" and s in details]
# Bulk approval requires a model that has been SCORED on that SKU's own history,
# not merely a confident-looking label. `trust.auto_approve` is High only when the
# backtest, the evidence base and the data completeness all pass.
bulk_eligible = [s for s in pending
if (details[s].get("trust") or {}).get("auto_approve")
and details[s]["action"] in ("Increase", "Decrease")
and abs(details[s]["delta_pct"]) <= 5]
h1, h2 = st.columns([3, 1.3])
with h1:
# The headline slot goes to the thing the reviewer is here to do, not to a
# greeting. The greeting keeps the warmth, at caption size where it belongs.
n_pend = len(pending)
if len(details) == 1:
# Lead with the product and the verdict. "1 recommendation to review" is a queue
# heading, and on a single product there is no queue โ the reader already knows what
# they opened and wants to know what to do about it.
_hs = next(iter(details))
_hd = details[_hs]
_title = (_hd.get("cfg") or {}).get("title") or _hs
_act = _hd["action"]
_verb = {"Increase": "Raise", "Decrease": "Lower",
"Maintain": "Hold", "Investigate": "Investigate"}.get(_act, _act)
if _act in ("Increase", "Decrease"):
_hl = (f"{_verb} to ${_hd['rec_price']:,.2f} "
f"({_hd['delta_pct']:+.1f}% from ${_hd['current_price']:,.2f})")
elif _act == "Maintain":
_hl = f"Hold at ${_hd['current_price']:,.2f}"
else:
_hl = f"Investigate โ hold at ${_hd['current_price']:,.2f}"
st.markdown(f"## {esc(_hl)}")
_why = ", ".join(REASON_LABELS.get(c, c) for c in _hd["reasons"]) or "no signals"
st.caption(esc(f"{_title} ยท {_hs} ยท {_why}"))
else:
st.markdown(f"## {n_pend} recommendation{'s' if n_pend != 1 else ''} to review")
st.caption(f"{greet}, Utopia ๐ ยท sorted by expected profit impact")
with h2:
st.markdown("")
# Explain a control that cannot fire, rather than leaving a dead button.
if not bulk_eligible and not kill:
low_trust = [s for s in pending
if not (details[s].get("trust") or {}).get("auto_approve")]
why_blocked = (f"{len(low_trust)} of {n_pend} below High trust"
if low_trust else "nothing eligible")
else:
why_blocked = ""
st.button(f"โ
Bulk approve ยท {len(bulk_eligible)}",
disabled=kill or not bulk_eligible, use_container_width=True,
on_click=bulk_approve, args=(bulk_eligible,),
help=("Bulk approval needs a model scored on that SKU's own history. "
+ (f"Blocked: {why_blocked}." if why_blocked else
"Approves pending High-trust recommendations within the 5% step cap.")))
if why_blocked:
st.caption(esc(f"Needs High trust ยท {why_blocked}"))
if kill:
st.error("๐ **Kill switch active** โ approvals and price pushes are paused across all "
"marketplaces. Toggle it off in the sidebar to resume.")
# ---------------------------------------------------------------- stat tiles
needing_action = [s for s in pending if details[s]["action"] != "Maintain"]
opportunity = sum(max(details[s]["impact_30d"], 0) for s in pending)
# Same thresholds the decision logic uses, imported rather than restated so the
# tiles and the recommendations can never drift apart.
from dashboard.live_data import HIGH_COVER_DAYS, LOW_COVER_DAYS
stockout_risk = sum(1 for s in details if 0 < details[s]["cover_days"] <= LOW_COVER_DAYS)
overstock_risk = sum(1 for s in details if details[s]["cover_days"] >= HIGH_COVER_DAYS)
# The opportunity figure inherits the model's measured error, so it is shown as a
# band. A precise-looking $32,974 from a model backtested at ยฑ58% reads as a promise.
_errs = [e for s in pending if (e := model_error_pct(details[s])) is not None]
_err = max(_errs) if _errs else None
# ONE SKU is a different question from a portfolio, so it gets different tiles.
#
# The counts above answer "where in my catalogue should I look?" โ they are triage. On a
# single product that question is already answered, and the same tiles degenerate into
# tautologies: "1 SKU needing action" out of one, "0 stockout-risk SKUs", "1 pending
# approval". Nothing there tells a seller whether to take the price.
#
# So on a single product the header answers the seller's questions instead: what does this
# move earn, what margin does it leave, how much room is there before it loses money, will
# stock last, is it selling, and who else is on the listing.
if len(details) == 1:
_sku = next(iter(details))
_d = details[_sku]
_e1 = model_error_pct(_d)
_econ = {x["key"]: x for x in (_d.get("scen_econ") or [])}
_cur_e = _econ.get("current") or {}
# The row at the price we are RECOMMENDING TODAY, not the rung the cascade named. When a
# move is step-capped those differ: `rec_key` here is the $20.00 destination while the
# headline recommends $17.94, and quoting the destination's margin beside a $17.94 headline
# would promise 13.4% for a move that delivers 6.6%.
_rec_e = next((x for x in (_d.get("scen_econ") or [])
if abs(x["price"] - _d["rec_price"]) < 0.011),
_econ.get(_d.get("rec_key")) or {})
_fi = _d.get("floor_info") or {}
_floor = _fi.get("floor")
_cover = _d.get("cover_days")
_out = _d.get("outlook") or {}
_cm = _d.get("comp_meta") or {}
_ppc = _d.get("ppc") or {}
# 1. What the move is worth. Held to the same error band as the portfolio figure.
_impact = _d.get("impact_30d") or 0.0
_t_impact = ("๐ฐ", money_range(_impact, _e1) if _impact else "โ",
"Impact ยท 30d if approved",
(f"net of ads + storage ยท ยฑ{_e1 * 100:.0f}% model error" if _e1
else "net of advertising and storage"))
# 2. Margin now vs after. The number a seller judges a price by โ so both halves have to
# be the same model at two prices. The current row's headline margin is computed from
# BOOKED profit at the average price customers paid; comparing that to a projection makes
# the move look better or worse than it is. On this SKU booked reads 6.7% while the model
# at today's price reads 3.3%, so "6.7% โ 6.6%" said the move achieved nothing when it
# actually roughly doubles margin.
_m_rec = modelled(_rec_e).get("net_margin_pct")
_m_now = modelled(_cur_e).get("net_margin_pct")
_t_margin = ("๐",
f"{_m_rec:.1f}%" if _m_rec is not None else "โ",
"Net margin after the move",
(f"now {_m_now:.1f}% ยท same model, both prices" if _m_now is not None
else "after ads and storage"))
# 3. Headroom to the floor โ how much room before the price stops paying.
if _floor and _d.get("current_price"):
_head = (_d["current_price"] - _floor) / _d["current_price"] * 100
_t_floor = ("๐", f"{_head:+.1f}%", "Headroom above the floor",
f"floor ${_floor:,.2f} ({_fi.get('binding', 'break-even')}) ยท "
f"today ${_d['current_price']:,.2f}")
else:
_t_floor = ("๐", "โ", "Headroom above the floor", "no floor โ cost data missing")
# 4. Will the stock last? A date beats a threshold count on a single SKU.
#
# Both readings are shown, because COSMOS's Inventory Planning grid prints the LONGER
# one and a single unexplained number here reads as a mismatch with their tool. They are
# different questions: ours is what is on the shelf now, COSMOS's counts stock still in
# transit. Neither is wrong; the pricing rules use the shorter one deliberately, because
# a SKU cannot sell a unit that has not landed.
_inv_cls = ((_d.get("cfg") or {}).get("inv_class") or "alpha").lower()
_onhand = _d.get("cover_days_onhand")
_band_hex, _band_word = theme.cover_band(_cover, _inv_cls)
_so, _arr = _out.get("stockout_date"), _out.get("next_arrival_date")
if _so and _arr:
_cover_note = (f"{_band_word} ยท out ~{_so:%d %b} ยท restock {_arr:%d %b}"
+ (" โ gap" if _so < _arr else " โ covered"))
elif _so:
_cover_note = f"{_band_word} ยท out ~{_so:%d %b} ยท no restock scheduled"
else:
_cover_note = f"{_band_word} ยท {_inv_cls.capitalize()}-class band"
# Matches the COSMOS Inventory Planning grid. On-hand is shown beside it because the two
# differ by stock still in transit, and a stockout judged on undelivered units is worth
# being able to see.
if _onhand is not None and _cover is not None and _onhand != _cover:
_cover_note += f" ยท {_onhand:,} d on hand, rest inbound"
# A COSMOS band and a pricing trigger are different things, and the colour is the loudest
# thing on the tile โ so where they disagree, say so. Without this a pink 82-day SKU reads
# as "the engine is about to discount this", when the cut rule does not fire until 90.
if _cover is not None and _band_word in ("high", "low") and not (
0 < _cover <= LOW_COVER_DAYS or _cover >= HIGH_COVER_DAYS):
_edge = HIGH_COVER_DAYS if _band_word == "high" else LOW_COVER_DAYS
_cover_note += (f" ยท COSMOS band only โ no price move until {_edge} d")
_t_cover = ("๐ฆ", f"{_cover:,} d" if _cover is not None else "โ",
"Inventory cover", _cover_note, _band_hex)
# 5. Is it actually selling?
_u = _d.get("units_day")
_t_vel = ("โก", f"{_u:,.0f}/day" if _u else "โ", "Sales velocity",
(f"TACoS {_ppc['tacos']:.1f}%" if _ppc.get("tacos") is not None
else "30-day average"))
# 6. Who else is on this listing โ the one tile that can invalidate all the others.
if _cm.get("state_usable"):
_bb = str(_cm.get("state") or "โ").replace("_", " ").title()
_riv = _cm.get("rivals") or 0
_t_comp = ("๐", _bb, "Buy Box",
(f"{_riv} rival(s) ยท median ${_cm['median']:,.2f}"
if _cm.get("median") else f"{_riv} rival(s)"))
else:
_t_comp = ("๐", "N/A", "Buy Box",
"not covered by the competitor sheet"
if not _cm.get("sheet_covers_sku") else "no usable competitor data")
tiles = [_t_impact, _t_margin, _t_floor, _t_cover, _t_vel, _t_comp]
else:
tiles = [
("โก", str(len(needing_action)), "SKUs needing action", "queue below, ranked by impact"),
("๐ฐ", money_range(opportunity, _err), "Profit opportunity ยท 30d",
(f"open positive impacts ยท ยฑ{_err * 100:.0f}% model error" if _err
else "sum of open positive impacts")),
("โณ", str(len(pending)), "Pending approvals", f"{len(bulk_eligible)} bulk-eligible"),
("๐", str(stockout_risk), "Stockout-risk SKUs", f"cover โค {LOW_COVER_DAYS} days"),
("๐ฆ", str(overstock_risk), "Overstock-risk SKUs", f"cover โฅ {HIGH_COVER_DAYS} days"),
]
st.markdown(''
# Tiles are (icon, value, label, note) with an OPTIONAL 5th accent colour, so a
# banded reading can tint its icon without every other tile growing a field.
+ "".join(theme.tile(*t) for t in tiles)
+ '
', unsafe_allow_html=True)
st.markdown("")
# ---------------------------------------------------------------- action pills
counts = summary["action"].value_counts().to_dict()
pills = [("All", "All", len(summary)),
("Increase", "โ Raise", counts.get("Increase", 0)),
("Decrease", "โ Lower", counts.get("Decrease", 0)),
("Maintain", "โ Hold", counts.get("Maintain", 0)),
("Investigate", "๐ Check", counts.get("Investigate", 0))]
with st.container(key="pillrow"): # keyed so the CSS can keep these pills tight
pcols = st.columns([0.8, 1, 1, 1, 1, 2])
for col, (a, lbl, n) in zip(pcols, pills):
col.button(f"{lbl} ยท {n}", key=f"pill_{a}",
type="primary" if st.session_state.action_sel == a else "secondary",
use_container_width=True, on_click=set_action, args=(a,),
help=f"Show {a.lower()} recommendations" if a != "All" else "Show all")
# ---------------------------------------------------------------- queue
rows = summary.copy()
if f_mkt != "All":
rows = rows[rows["marketplace"] == f_mkt]
if f_conf != "All":
rows = rows[rows["trust_tier"] == f_conf]
if f_stat != "All":
rows = rows[[st.session_state.status[s] == f_stat for s in rows["sku"]]]
if st.session_state.action_sel != "All":
rows = rows[rows["action"] == st.session_state.action_sel]
if q:
mask = (rows["sku"].str.contains(q, case=False, regex=False)
| rows["title"].str.contains(q, case=False, regex=False))
rows = rows[mask]
if rows.empty:
st.info("No recommendations match the current filters.")
for _, r in rows.iterrows():
sku = r["sku"]
d = details[sku]
status = st.session_state.status[sku]
modified_price = st.session_state.modified.get(sku)
tier, score = d["confidence"]
stat_icon = theme.REC_STATUS[status][1]
glyph = ACTION_GLYPH[d["action"]]
# Header: SKU ยท action+price ยท reason. ASIN, marketplace and inventory band moved
# into the card โ six competing facts in one line had no hierarchy and wrapped.
asin = d.get("asin") or d["competitors"].iloc[0]["asin"]
inv_cls = d["cfg"]["inv_class"].lower()
trust = d.get("trust") or {}
target = modified_price if modified_price is not None else d["rec_price"]
mod_tag = " (modified)" if modified_price is not None else ""
if d["action"] == "Increase":
act = (f":green[**โ RAISE ${d['current_price']:.2f} โ ${target:.2f} "
f"(+{abs(d['delta_pct']):.1f}%){mod_tag}**]")
elif d["action"] == "Decrease":
act = (f":red[**โ LOWER ${d['current_price']:.2f} โ ${target:.2f} "
f"(โ{abs(d['delta_pct']):.1f}%){mod_tag}**]")
elif d["action"] == "Investigate":
act = f":orange[**๐ INVESTIGATE ยท hold ${d['current_price']:.2f}**]"
else:
act = f":gray[**โ HOLD ${d['current_price']:.2f}**]"
# Trust rides in the row header, so it is visible before the row is opened.
trust_mark = {"High": "", "Medium": " ยท โ ๏ธ medium trust",
"Low": " ยท โ ๏ธ low trust", "None": " ยท โ data incomplete"}.get(
trust.get("tier", ""), "")
label = esc(f"{stat_icon} **{sku}** ยท {act} ยท {r['primary_reason']}{trust_mark}")
with st.expander(label):
# ---------- decision card ----------
# Keyed container so the responsive CSS can stack these three columns on a
# narrow window instead of squeezing the metrics.
with st.container(key=f"deccard_{sku}"):
c1, c2, c3 = st.columns([2.4, 2.2, 1.4])
with c1:
asin_url = _amazon_url(asin, r["marketplace"])
asin_md = f"[{asin}]({asin_url})" if asin_url else esc(str(asin))
st.caption(f"{r['title']} ยท ASIN {asin_md} ยท {mkt_label(r['marketplace'])}"
f" ยท {INV_CLASS_LABELS.get(inv_cls, inv_cls)}")
st.markdown(esc(f"#### {glyph} {d['action']}" + (
f" to **${d['rec_price']:.2f}**" if d["action"] in ("Increase", "Decrease") else "")))
st.markdown(esc(f"**Current price ${d['current_price']:.2f}**"))
chips = "".join(theme.chip(REASON_LABELS.get(c, c), theme.BRAND)
for c in d["reasons"])
err = model_error_pct(d)
trust_chip = ""
if trust:
tcol = {"High": theme.STATUS["good"], "Medium": theme.STATUS["warning"],
"Low": theme.STATUS["serious"],
"None": theme.STATUS["critical"]}[trust["tier"]]
tlabel = f"๐ฏ {trust['tier']} trust"
if err:
tlabel += f" ยท ยฑ{err * 100:.0f}% backtested"
trust_chip = theme.chip(tlabel, tcol)
# Only ONE reliability badge. The old confidence tier was derived
# from days-with-sales and happily read "Medium (62)" for a SKU
# whose elasticity could not be told apart from zero โ shown next
# to "Low trust" it just contradicted it. The tier is still used
# internally to width the forecast band.
st.markdown(trust_chip + theme.status_badge(status) + chips,
unsafe_allow_html=True)
if trust.get("tier") in ("Low", "None"):
st.caption(esc(f"โ {trust['permits']} ยท see **๐ฏ Track record**"))
if modified_price is not None:
st.caption(esc(f"โ๏ธ Approved at modified price ${modified_price:.2f} "
f"(engine recommended ${d['rec_price']:.2f})"))
for c in (d.get("clamps") or []):
st.caption(esc(f"๐ฏ {c}"))
# Guardrail detail is reference, not a decision input โ one line of
# it belongs on the card, the rest behind a disclosure.
fl = d.get("floor_info") or {}
with st.popover("๐ก๏ธ Guardrails & break-even", use_container_width=True):
st.markdown(esc(
f"**Allowed range ${d['min_price']:.2f}โ${d['max_price']:.2f}** ยท "
f"objective: {OBJECTIVE_LABELS.get(r['objective'], r['objective'])}"))
be_rows = [("Fees + COGS only", d["break_even"], "ignores advertising")]
if d.get("break_even_with_ads"):
be_rows.append(("Including advertising", d["break_even_with_ads"],
"at current ad cost per unit"))
if d.get("break_even_empirical"):
be_rows.append(("From actual booked profit", d["break_even_empirical"],
"carries every cost COSMOS books"))
st.dataframe(
pd.DataFrame(be_rows, columns=["Break-even", "Price", "Basis"]),
hide_index=True, use_container_width=True,
column_config={"Price": st.column_config.NumberColumn(format="$%.2f")})
if fl.get("binding") and fl["binding"] != "accounting":
st.caption(esc(
f"Floor is set by the **{fl['binding']}** figure "
f"(${fl['floors'][fl['binding']]:.2f}). The fees-only number "
f"(${fl['floors']['accounting']:.2f}) ignores advertising, so "
f"pricing to it loses money on every unit."))
with c2:
# Calibrated net profit for the headline metric (matches the tables).
_econ = {x["key"]: x for x in (d.get("scen_econ") or [])}
_rec_e = _econ.get(d["rec_key"])
_cur_e = _econ.get("current")
rec_units = _rec_e["units_day"] if _rec_e else round(d["rec_units_day"])
cur_units = _cur_e["units_day"] if _cur_e else round(d["units_day"])
rec_net = _rec_e["net_30d"] if _rec_e else d["rec_profit_30d"]
cur_net = _cur_e["net_30d"] if _cur_e else d["profit_30d"]
m1, m2, m3, m4 = st.columns(4)
m1.metric("Units/day", f"{rec_units:,}",
f"{rec_units - cur_units:+,} vs now", delta_color="off")
# Point estimate stays the headline so it can be scanned; the range
# sits underneath. A range in the value slot has no break
# opportunity and shatters character-by-character in a narrow column.
# Point estimate as the headline, uncertainty as a short suffix. The
# delta cell is ~110px wide, so the full range belongs on the
# Scenarios tab, not here.
m2.metric("Net profit/30d", compact_money(rec_net),
(f"ยฑ{err * 100:.0f}% model error" if err else
f"{'+' if rec_net >= cur_net else 'โ'}"
f"{compact_money(abs(rec_net - cur_net))} vs now"),
delta_color="off")
# Cover in days, with the DATE it runs out โ "26 days" is abstract,
# "out on 24 Aug" is a diary entry someone can act on.
ol = d.get("outlook") or {}
so = ol.get("stockout_date")
# "Cover after the move", not "Cover". This is cover at the RECOMMENDED
# price, while the header tile and the Inventory grid both show cover at
# TODAY's price โ 75 d against 78 d on one screen, under the same word, with
# nothing saying which was which. Both were right; the label was the bug.
m3.metric("Cover after the move", f"{d['rec_cover_days']} d",
(f"out ~{so:%d %b}" if so else
f"{d['rec_cover_days'] - d['cover_days']:+d} d vs today"),
delta_color="off")
m4.metric("PPC/day", f"${d['ppc']['spend_day']:,.0f}",
f"TACoS {d['ppc']['tacos']:.1f}%", delta_color="off")
with c3:
# The primary action states what the evidence actually supports. At
# Low trust a bold "Approve" invites a commitment the model cannot
# back โ the same click, honestly labelled, is a test.
tier = trust.get("tier", "Medium")
if tier == "None":
st.button("โ Blocked โ fix data", key=f"ap_{sku}", disabled=True,
use_container_width=True,
help="; ".join(trust.get("reasons", [])) or "data incomplete")
elif tier == "Low":
st.button(f"๐งช Start test at ${d['rec_price']:.2f}", key=f"ap_{sku}",
type="primary", disabled=kill or status == "Approved",
on_click=set_status, args=(sku, "Approved"),
use_container_width=True,
help="Backtest error is high on this SKU, so treat the move "
"as an experiment: run it, then re-read the response.")
else:
st.button("โ
Approve", key=f"ap_{sku}", type="primary",
disabled=kill or status == "Approved",
on_click=set_status, args=(sku, "Approved"),
use_container_width=True)
with st.popover("โ๏ธ Modify", use_container_width=True, disabled=kill):
st.number_input("New price ($)", key=f"mod_{sku}",
min_value=float(d["min_price"]), max_value=float(d["max_price"]),
value=float(min(max(d["rec_price"], d["min_price"]),
d["max_price"])),
step=0.05, format="%.2f")
st.caption(esc(f"Allowed ${d['min_price']:.2f}โ${d['max_price']:.2f} ยท "
">5% from current needs elevated approval"))
st.button("Apply & approve", key=f"modb_{sku}",
on_click=apply_modify, args=(sku, details), use_container_width=True)
st.button("โ๏ธ Reject", key=f"rj_{sku}", disabled=kill or status == "Rejected",
on_click=set_status, args=(sku, "Rejected"), use_container_width=True)
if status != "Pending":
st.button("โฉ๏ธ Undo", key=f"un_{sku}",
on_click=set_status, args=(sku, "Pending"), use_container_width=True)
# ---------- strategy options ----------
scen = d["scenarios"].set_index("scenario")
if d["action"] == "Investigate":
st.info("No price options proposed โ investigate first. A change without a "
"confirmed cause is blocked by the root-cause gate.")
else:
# Calibrated economics keyed by scenario, so this table matches the
# Scenarios tab (net profit anchored to actual booked profit).
econ_by_key = {x["key"]: x for x in (d.get("scen_econ") or [])}
cover_by_key = {k: scen.loc[k]["cover_days"] for k in scen.index}
def opt_row(label, key):
e = econ_by_key.get(key)
if not e:
return None
return {
"Option": label, "Price": e["price"], "Units/day": e["units_day"],
"Profit/30d": e["net_30d"], "Margin %": e["net_margin_pct"],
"Cover (days)": int(cover_by_key.get(key, 0)),
}
# One row per distinct price. Rungs can legitimately share a price (e.g.
# nothing in the grid is bolder than the recommendation), so merge their
# labels instead of rendering the same row twice.
#
# Two things this table used to say that were not true:
#
# 1. It called the first row "Current" at a price that is NOT the current price.
# That row is real booked history, so its price is the AVERAGE SOLD price over
# the window (revenue / units) โ deliberately, so revenue reconciles. But the
# rungs below it are built from today's price, so on one SKU the table read
# Current $18.18 / Conservative $17.43 and the "conservative" option looked like
# a cut on an Increase recommendation. Same column, two different bases, one
# label. Naming the basis is what makes the column readable.
# 2. It starred "Recommended" against the TARGET price while the header recommended
# the step-capped price for today. A row labelled Recommended has to be the
# thing being recommended.
win = d.get("scen_window") or 90
rec_now, tgt = float(d["rec_price"]), float(d["target_price"])
merged: dict[str, list[str]] = {}
for name, key in [(f"Current ยท avg sold ({win}d)", "current")] \
+ list(d["strategy_keys"].items()):
merged.setdefault(key, []).append(name)
# Which grid key are we actually moving to today?
rec_key_now = next(
(k for k, e in econ_by_key.items() if abs(e["price"] - rec_now) < 0.011), None)
opt_rows = []
for key, names in merged.items():
e = econ_by_key.get(key)
label = " ยท ".join(names)
# This rung is the destination, not this week's move โ said once, over the
# whole merged label, not once per name it merged.
if d.get("stepped") and e is not None and abs(e["price"] - tgt) < 0.011:
label = f"Target (~14d) ยท {label}"
star = (rec_key_now is None and "Recommended" in names) or key == rec_key_now
opt_rows.append(opt_row(("โ
" if star else "") + label, key))
# The step-capped price may not be one of the named rungs. If it is a grid row that
# nothing else labelled, surface it โ otherwise the price in the header appears
# nowhere in the table underneath it.
if rec_key_now is not None and rec_key_now not in merged:
row = opt_row("โ
Recommended now (5% step cap)", rec_key_now)
if row:
opt_rows.append(row)
opt_rows = [x for x in opt_rows if x]
opt_rows.sort(key=lambda x: x["Price"])
with st.container(key=f"optrow_{sku}"):
oc1, oc2 = st.columns([4, 1])
with oc1:
st.dataframe(
pd.DataFrame(opt_rows), hide_index=True, use_container_width=True,
column_config={
"Price": st.column_config.NumberColumn(format="$%.2f"),
"Units/day": st.column_config.NumberColumn(format="%d"),
"Profit/30d": st.column_config.NumberColumn(format="$%d"),
"Margin %": st.column_config.NumberColumn(format="%.1f%%"),
},
)
cur_econ = econ_by_key.get("current") or {}
cur_sold = cur_econ.get("price")
cap = ("Profit/30d is net of advertising and storage โ same "
"basis as the Scenarios tab.")
# Reconcile the two prices ON SCREEN. They are both correct and they are
# different things; without this the reader is left to conclude one is a bug.
if cur_sold and abs(cur_sold - d["current_price"]) >= 0.01:
cap += (f" The Current row is **${cur_sold:,.2f}** โ the average price "
f"actually *sold* over the last {win} days (revenue รท units), "
f"which is what its profit is computed from. Today's price is "
f"**${d['current_price']:,.2f}**, and every option below is "
f"calculated from that, so an option can sit below the Current "
f"row without being a price cut.")
st.caption(cap)
with oc2:
st.button("๐ View more", key=f"vm_{sku}", use_container_width=True,
on_click=goto_view, args=(sku, VIEW_SCENARIOS),
help="Jump to the full Scenarios breakdown")
# ---------- section switcher (jumpable, unlike native tabs) ----------
view_key = f"view_{sku}"
if view_key not in st.session_state:
st.session_state[view_key] = VIEW_KEYS[0]
sel = st.segmented_control("Section", VIEW_KEYS, key=view_key,
format_func=lambda k: VIEW_LABELS[k],
label_visibility="collapsed")
if sel is None: # ignore an accidental deselect
sel = st.session_state[view_key] or VIEW_KEYS[0]
if sel == "price":
st.plotly_chart(price_demand_chart(d), use_container_width=True,
config={"displayModeBar": False}, key=f"pd_{sku}")
with st.popover("View as table"):
st.dataframe(d["hist"].tail(30), hide_index=True, use_container_width=True)
elif sel == "inventory":
# Dated outlook up top: when it empties, when relief lands, and how
# many days of the last month were spent effectively out of stock.
ol = d.get("outlook") or {}
oos, known = d.get("stockout_days_30d") or (0, 0)
i1, i2, i3 = st.columns(3)
so, arr = ol.get("stockout_date"), ol.get("next_arrival_date")
if so:
src = ("COSMOS projection" if ol.get("stockout_source") == "projection"
else "extrapolated from velocity")
i1.metric("Runs out", f"{so:%d %b}",
f"{ol.get('days_to_stockout')} days ยท {src}", delta_color="off")
else:
i1.metric("Runs out", "not in horizon",
f"within {ol.get('horizon_days') or 0} days", delta_color="off")
if arr:
gap = (arr - so).days if so else None
i2.metric("Next arrival", f"{arr:%d %b}",
(f"{gap:+d} days vs stockout" if gap is not None
else f"{ol.get('next_arrival_units', 0):,.0f} units"),
delta_color="off")
else:
i2.metric("Next arrival", "none scheduled",
f"within {ol.get('horizon_days') or 0} days", delta_color="off")
i3.metric("Days out of stock", f"{oos}",
f"of {known} days with a reading" if known
else "no inventory readings", delta_color="off")
if so and arr and (arr - so).days > 0:
st.warning(esc(
f"โ ๏ธ Stock runs out around **{so:%d %b}** but the next arrival is "
f"**{arr:%d %b}** โ **{(arr - so).days} days uncovered.** Raising "
f"price slows the burn; it does not create stock."))
st.markdown(inventory_table_html(d), unsafe_allow_html=True)
st.caption(
"Cell: big = projected units ยท purple = inventory value ยท blue = days of "
"cover ยท right = incoming units. Shading per this SKU's "
f"{d['cfg']['inv_class'].capitalize()}-class cover bands."
)
mp = d.get("min_po") or {}
if mp.get("qty"):
po_note = (f"COSMOS suggests a min PO of **{mp['qty']:,.0f} units**"
+ (f" within **{mp['days']:,.0f} days**" if mp.get("days") else "")
+ " to avoid stockout.")
else:
po_note = "No minimum-PO recommendation from COSMOS."
st.caption(f"**Real COSMOS INVP projection** (invp-insight dateMap โ projected "
f"units, value, cover days & warehouse arrivals). {po_note}")
st.plotly_chart(inventory_chart(d), use_container_width=True,
config={"displayModeBar": False}, key=f"inv_{sku}")
# ---- and the same shelf at the recommended price, directly below ----
# The projection above is COSMOS's, at TODAY's price, so the price move is
# otherwise shown with no stock consequence at all. Raising slows the burn;
# cutting empties the shelf sooner, which on a SKU near a stockout is the
# difference between a sound move and an expensive one.
_ed = d.get("elasticity_detail") or {}
_el = _ed.get("elasticity") or FALLBACK_ELASTICITY
# COSMOS's OWN daily sale, the same figure the grid above prints โ not our 30-day
# mean from sales history. They differ (64 vs 63.3 on the audited SKU), and using
# ours put two different "daily sale" numbers in adjacent tables AND shifted every
# cover day: 5,029 / 63.3 = 79 where COSMOS says 78.
#
# Driving both off COSMOS's rate makes the cover column reproduce COSMOS's own
# coverDays exactly, so the two grids reconcile line for line.
_u_now = float((d.get("invp_stats") or {}).get("avg_sale")
or d.get("units_day") or 0.0)
_ratio = ((d["rec_price"] / d["current_price"]) ** _el
if d.get("current_price") else 1.0)
_rows = reproject_inventory(d.get("inv_projection") or [],
_u_now, _u_now * _ratio)
_moved = abs(d["rec_price"] - d["current_price"]) >= 0.005
st.markdown("---")
if not _moved:
st.caption("**No price change is recommended**, so the shelf above is the "
"outlook. Nothing to re-project.")
elif not _rows:
st.caption(esc(
"Cannot re-project the shelf at the new price for this SKU โ that needs "
"COSMOS's weekly projection and a usable demand response, and one of "
"them is missing."))
else:
_last = _rows[-1]
st.markdown(esc(
f"**At the recommended ${d['rec_price']:,.2f}** "
f"({d['delta_pct']:+.1f}%), demand moves {_ratio - 1:+.1%} โ "
f"**{_last['units_new']:,} units** left at the end of the horizon "
f"against {_last['units_now']:,} at today's price "
f"(**{_last['delta_units']:+,}**)."))
# The one line worth interrupting for: the move empties the shelf sooner than
# COSMOS projects. Stated in words as well as drawn on the chart.
_out = next((r for r in _rows if r["stockout"]), None)
if _out:
st.error(esc(
f"At ${d['rec_price']:,.2f} the shelf empties by "
f"**{pd.Timestamp(_out['date']):%d %b}** โ sooner than COSMOS "
f"projects at today's price. A cut sells the remaining stock faster; "
f"it does not create any."))
if all(r["delta_units"] == 0 for r in _rows):
st.caption(esc(
f"COSMOS projects no depletion for this SKU โ the same "
f"{_rows[0]['units_now']:,} units in every snapshot โ so there is no "
f"draw to re-scale and the two lines coincide. Only cover days move."))
# The same grid as above, at the new price, so the two read cell for cell.
st.markdown(
inventory_reprice_table_html(d, _rows, _u_now, _u_now * _ratio),
unsafe_allow_html=True)
st.caption(
"Same layout as the COSMOS grid above and the same Alpha/Beta cover "
"shading, so the two compare cell for cell. **Change** is this projection "
"minus COSMOS's โ green = more stock left, red = burns through faster.")
st.plotly_chart(inventory_reprice_chart(d, _rows),
use_container_width=True,
config={"displayModeBar": False}, key=f"invrp_{sku}")
st.caption(
"Solid = COSMOS's projection at today's price. Dashed = **our** model of "
"the same shelf at the recommended price, scaling each week's SALES by "
"the elasticity the Scenarios tab uses. Arrivals are unchanged โ a PO "
"already placed does not move because we re-priced โ and stock is floored "
"at zero.")
elif sel == "scenarios":
render_scenarios(d, key_prefix=f"sc_{sku}")
elif sel == "competitors":
comp = d["competitors"]
meta = d.get("comp_meta") or {}
rivals = comp[~comp["is_us"]] # third-party sellers only
if len(rivals) == 0:
# Nothing to compare against โ show a message, NOT your own listing.
if meta.get("sheet_covers_sku"):
# The workbook covers this SKU but produced no usable rival: every match
# was fuzzy, or the only rivals had their own Buy Box suppressed. Say which,
# because "no data" and "data we refused to price against" are different.
st.warning(
"๐ **The comparison sheet covers this SKU but has no priceable "
"rival.** Every matched rival was either a NEAR colour match "
"(unverified) or had its own Buy Box suppressed, so none may move a "
"price. Open the workbook to see the raw pairings.")
elif not meta.get("scraped"):
st.info("๐ **Competitor prices not scraped.** Tick **Include "
"competitor prices** in the sidebar and reload to pull Buy "
"Box + rival offers for this ASIN."
+ (f" This SKU is also outside the comparison sheet, which "
f"covers {meta['sheet_covered_skus']} SKU(s) of "
f"{meta['sheet_line']}."
if meta.get("sheet_line") else ""))
else:
bb = meta.get("buy_box_status")
bb_price = meta.get("buy_box_price")
own = meta.get("own_offers", 0)
detail = ""
if bb == "WON" and bb_price:
detail = (f" You hold the **Buy Box at ${bb_price:.2f}**"
+ (f" across {own} of your own offers" if own > 1 else "")
+ ".")
st.success("โ
**No competitors on this ASIN.** Utopia Brands is the "
f"only seller โ no third-party offers were found.{detail} "
"Pricing is driven by your own economics, not rivals.")
else:
# Real competitors exist โ show ONLY the rival sellers.
#
# The two sources describe different things and must not share a sentence: an
# Apify row is another seller on OUR listing, a sheet row is a rival brand's own
# listing. Calling the latter "an offer on this ASIN" would be simply untrue.
if meta.get("source") == "comparison-sheet":
as_of = meta.get("sheet_as_of")
st.caption(
f"**{len(rivals)} like-for-like rival(s)** from the comparison sheet โ "
f"each is a *different brand's own listing* matched on size + colour, "
f"not an offer on your ASIN"
+ (f" ยท {meta['exact_rivals']} exact match(es) priceable"
if meta.get("exact_rivals") is not None else "")
+ (f" ยท median ${meta['median']:,.2f}" if meta.get("median") else "")
+ f" ยท your price ${d['current_price']:.2f}"
+ (f" ยท sheet dated {as_of:%d %b %Y}" if as_of else "") + ".")
else:
st.caption(f"**{len(rivals)} competitor offer(s)** on this ASIN"
+ (f" ยท competitor median ${meta['median']:,.2f}"
if meta.get("median") else "")
+ f" ยท your price ${d['current_price']:.2f}.")
disp = pd.DataFrame({
"Seller": rivals["seller"],
"ASIN": rivals["asin"].map(lambda a: _amazon_url(a, r["marketplace"])),
"ASIN code": rivals["asin"],
"Effective price": rivals["effective_price"].map(lambda v: f"${v:,.2f}"),
"Coupon": rivals["coupon"].map(lambda v: f"${v:,.2f}" if v else "โ"),
"Badge": rivals["badge"],
"In stock": rivals["in_stock"].map({True: "Yes", False: "No"}),
"Scraped": rivals["freshness"],
})
st.dataframe(
disp, hide_index=True, use_container_width=True,
column_config={
"ASIN": st.column_config.LinkColumn(
"ASIN", display_text=r"/dp/([A-Z0-9]+)", help="Open on Amazon"),
"ASIN code": None,
},
)
st.caption("Third-party sellers on the same ASIN. Your own offers are "
"excluded. Prices are the effective (after-coupon) offer price.")
elif sel == "ppc":
p = d["ppc"]
st.markdown("**Whole business โ from sales-insight (last 30 days)**")
b1, b2, b3, b4 = st.columns(4)
b1.metric("Total revenue/30d", dash(p.get("revenue_30d"), "${:,.0f}"),
f"{p.get('units_30d', 0):,} units")
b2.metric("Ad budget spent/30d", dash(p["spend_30d"], "${:,.0f}"),
"marketing + promotion")
b3.metric("Ad spend/day", dash(p["spend_day"], "${:,.0f}"))
b4.metric("TACoS", dash(p["tacos"], "{:.1f}%"), "ad spend รท TOTAL revenue")
if p.get("acos") is not None:
st.markdown("**Advertising only โ from the Amazon Ads data COSMOS "
"proxies (`marketing/dashboard`)**")
a1, a2, a3, a4 = st.columns(4)
a1.metric("ACoS", dash(p["acos"], "{:.1f}%"),
"ad spend รท ad-attributed sales")
a2.metric("Ad-attributed sales", dash(p["ad_sales_30d"], "${:,.0f}"),
f"{p['ad_units_30d']:,} units" if p.get("ad_units_30d")
else "")
a3.metric("ROAS", dash(p["roas"], "{:.2f}ร"), "sales per $1 of ad spend")
a4.metric("Assigned budget", dash(p["daily_budget"], "${:,.0f}"),
dash(p["budget_utilization"], "{:.0f}% used"))
e1, e2, e3, e4 = st.columns(4)
e1.metric("CPC", dash(p["cpc"], "${:.2f}"))
e2.metric("Clicks", dash(p["clicks"], "{:,}"))
e3.metric("Impressions", dash(p["impressions"], "{:,}"))
e4.metric("Conversion", dash(p["conversion"], "{:.2f}%"),
"orders รท clicks")
if p.get("ad_share_pct") is not None:
st.info(esc(
f"๐ฃ Ads can claim **{p['ad_share_pct']:.0f}%** of units "
f"({p['ad_units_30d']:,} of {p['units_30d']:,}); the rest is "
f"organic. ACoS **{p['acos']:.1f}%** is the cost of the "
f"attributed portion โ TACoS **{p['tacos']:.1f}%** spreads the "
f"same spend over all revenue. Judge campaigns on ACoS; judge "
f"the SKU's price on TACoS."))
else:
st.info("No advertising campaigns found for this SKU in the last 30 "
"days, so ACoS / ad-attributed sales are not applicable.")
st.plotly_chart(ppc_chart(d), use_container_width=True,
config={"displayModeBar": False}, key=f"ppc_{sku}")
st.caption(
"Two different denominators, both real: **TACoS** = total ad spend รท "
"**total** revenue (from `sales-insight`, includes promotions); "
"**ACoS** = ad spend รท **ad-attributed** sales (from the Ads data). "
"The pricing model uses total ad cost per unit, because a price change "
"moves all units โ not only the ad-attributed ones.")
elif sel == "costs":
cur_p, rec_p = d["current_price"], d["rec_price"]
rp = d.get("referral_pct", REFERRAL_PCT)
ret = d.get("returns_pct", 0.0)
var = d.get("variable", VARIABLE)
cost_rows = [
("Selling price", cur_p, rec_p),
(f"Referral fee ({rp * 100:.1f}%)", -cur_p * rp, -rec_p * rp),
("FBA fee", -d["cfg"]["fba"], -d["cfg"]["fba"]),
("Landed cost", -d["cfg"]["cost"], -d["cfg"]["cost"]),
]
if ret:
cost_rows.append((f"Returns reserve ({ret * 100:.1f}%)",
-cur_p * ret, -rec_p * ret))
cost_rows += [
("Other fees / variable", -var, -var),
("Take-home / unit", unit_take_home(cur_p, d), unit_take_home(rec_p, d)),
]
cdf = pd.DataFrame(cost_rows, columns=["Line", "At current price", "At recommended"])
st.dataframe(cdf, hide_index=True, use_container_width=True,
column_config={
"At current price": st.column_config.NumberColumn(format="$%.2f"),
"At recommended": st.column_config.NumberColumn(format="$%.2f"),
})
st.caption(esc(f"Break-even ${d['break_even']:.2f} ยท guardrail floor "
f"${d['min_price']:.2f} ยท ceiling ${d['max_price']:.2f}"))
if d["cfg"]["cost"] <= 0 and d["cfg"]["fba"] <= 0:
st.warning("โ ๏ธ COSMOS returned **no cost data** for this SKU โ every line "
"above that depends on COGS/FBA is wrong until the product's "
"costs are filled in COSMOS/Inventory Planner.")
elif sel == "track":
# How wrong has this model been on THIS SKU? Previously buried at the
# bottom of the last tab, under the narration it is meant to qualify.
tr, bt = d.get("trust") or {}, d.get("backtest")
if tr:
icon = {"High": "๐ข", "Medium": "๐ก", "Low": "๐ ", "None": "โ"}[tr["tier"]]
st.markdown(f"### {icon} {tr['tier']} trust")
st.markdown(esc(f"**What this permits:** {tr['permits']}"))
if tr["reasons"]:
st.markdown("**Why it isn't higher**")
for why in tr["reasons"]:
st.markdown(esc(f"- {why}"))
st.divider()
if bt:
sc = bt["scores"]
st.markdown(esc(
f"**Backtest** โ fitted on older history, then asked to predict the "
f"profit actually booked over {bt['n_folds']} held-out "
f"{bt['holdout_days']}-day window(s), at the price really charged:"))
names = {"anchored": "Anchored (no calibration)",
"anchored_gap": "Anchored + per-unit gap",
"persistence": "Assume last month repeats (baseline)",
"legacy": "Previous engine (TACoS + factor)"}
st.dataframe(
pd.DataFrame([
{"Method": ("โ " if m == bt["selected"] else "") + names[m],
"Avg error $/30d": sc[m]["mae"],
"Error % of actual": sc[m]["mape"],
"Bias $/30d": sc[m]["bias"]}
for m in ("anchored", "anchored_gap", "persistence", "legacy")]),
hide_index=True, use_container_width=True,
column_config={
"Avg error $/30d": st.column_config.NumberColumn(format="$%d"),
"Error % of actual": st.column_config.NumberColumn(format="%.1f%%"),
"Bias $/30d": st.column_config.NumberColumn(format="$%d"),
})
gp = d.get("gap_per_unit") or 0.0
st.caption(esc(
f"โ marks the method actually used for this SKU's projections, "
f"chosen by this table"
+ (f" (applying a **${gp:.2f}/unit** unmodeled cost)" if gp else "")
+ ". Lower is better. **Bias** shows the direction of the miss: "
"positive means the model flatters this SKU. Beating *'assume last "
"month repeats'* is the minimum bar โ a model that can't has added "
"nothing, and drops this SKU to Low trust."))
with st.popover("Per-fold detail"):
st.dataframe(pd.DataFrame([
{"Train days": f["train_days"], "Held-out days": f["test_days"],
"Price charged": f["test_price"],
"Actual $/30d": f["actual_net_30d"],
"Selected": f["pred"][bt["selected"]],
"Persistence": f["pred"]["persistence"],
"Previous": f["pred"]["legacy"]} for f in bt["folds"]]),
hide_index=True, use_container_width=True)
else:
st.info("Not enough history to hold any out โ this SKU's model is "
"unscored, so recommendations stay human-approved.")
elif sel == "why":
st.markdown(esc(d["explanation"]))
ev = d.get("evidence") or {}
if any(v is not None for v in ev.values()):
st.markdown("**Evidence from COSMOS actuals (6 months)**")
lines = []
if ev.get("actual_profit_per_unit") is not None:
lines.append(f"- Actual profit/unit (ads incl.): "
f"**${ev['actual_profit_per_unit']:.2f}**")
if ev.get("unprofitable_months"):
lines.append(f"- Months that lost money: **{ev['unprofitable_months']}**")
if ev.get("best_observed_price") is not None:
lines.append(f"- Best observed price: **${ev['best_observed_price']:.2f}** "
f"(${ev.get('best_observed_profit_day') or 0:,.0f}/day actual)")
if ev.get("unmodeled_cost_gap") is not None:
lines.append(f"- Model vs actual gap: **${ev['unmodeled_cost_gap']:.2f}/unit**")
if ev.get("suggested_price") is not None:
lines.append(f"- Margin-floor suggested price: "
f"**${ev['suggested_price']:.2f}**")
if ev.get("trend"):
lines.append(f"- Sales trend: **{ev['trend']}**")
st.markdown(esc("\n".join(lines)))
st.markdown("**Root-cause checks**")
for check, result in d["root_cause"]:
ok = result.startswith(("confirmed", "positive", "detected"))
icon = "โ
" if ok else ("โ ๏ธ" if ("missing" in result or "not feasible" in result)
else "โซ๏ธ")
st.markdown(esc(f"{icon} {check} โ *{result}*"))
st.markdown("**Risks**")
for risk, mit in d["risks"]:
st.markdown(f"- {risk} โ *{mit}*")
st.caption("Every number here comes from the deterministic engine over "
"COSMOS data โ see **๐ฏ Track record** for how accurate it has "
"been on this SKU.")
st.divider()
st.caption("Utopia Pricing Agent ยท live COSMOS data ยท read-only (no prices are "
"written back) ยท legacy analyst UI: `streamlit run legacy_app.py`")