844 lines
43 KiB
Python
844 lines
43 KiB
Python
"""Pricing & Profitability Analyst — web UI.
|
||
|
||
Three workflows: one SKU, a product line, or a catalog-wide money-at-risk scan.
|
||
Verdicts come from deterministic rules over COSMOS's actual profit; the language
|
||
model only writes the explanation. Read-only — nothing is ever written back.
|
||
|
||
Run: streamlit run app.py
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
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)
|
||
|
||
import logging
|
||
|
||
import streamlit as st
|
||
|
||
from config.settings import get_settings
|
||
from pricing_agent.analyze import (
|
||
AnalysisResult,
|
||
_build_service,
|
||
analyze_price,
|
||
analyze_sku,
|
||
generate_analysis_narrative,
|
||
)
|
||
from pricing_agent.fmt import usd
|
||
from pricing_agent.logging_config import ListHandler, setup_logging
|
||
|
||
setup_logging() # console logs (visible in the terminal running streamlit)
|
||
|
||
st.set_page_config(page_title="Pricing & Profitability Analyst", page_icon="◧",
|
||
layout="wide", initial_sidebar_state="collapsed")
|
||
|
||
PRODUCT_LINE_LIMIT = 15 # default number of SKUs analyzed per product-line request
|
||
|
||
TONE = {"GOOD": "ok", "CAUTION": "warn", "POOR": "bad"}
|
||
|
||
# Ordered exactly like the FBA Profitability Calculator.
|
||
FEE_ORDER = [
|
||
("itemPrice", "Item price"), ("referralFee", "Amazon referral fee"),
|
||
("fbaFee", "FBA fee"), ("lowInventoryFee", "Low inventory fee"),
|
||
("inboundPlacementFee", "Inbound placement fee"), ("returnFee", "Return fee"),
|
||
("eprFee", "EPR fee"), ("vat", "VAT"), ("costSubtotal", "Cost subtotal"),
|
||
("marginImpact", "Margin impact"), ("logisticsCost", "Logistics cost"),
|
||
("dutyAmount", "Duty amount"), ("costPerUnit", "Cost per unit"),
|
||
("orderHandling", "Order handling"), ("weightHandling", "Weight handling"),
|
||
("warehouseExpense", "Warehouse expense"),
|
||
]
|
||
# Ordered like the FBA Bulk Calculator.
|
||
BULK_ORDER = [
|
||
("itemPrice", "Item price"), ("saleUnits", "Sales units"), ("marketing", "Marketing"),
|
||
("inventory", "Inventory"), ("storageCharges", "Storage charges"),
|
||
("takeHome", "Take home"), ("takeHomePerUnit", "Take home per unit"),
|
||
("referralFee", "Referral fee"), ("fbaFee", "FBA fee"),
|
||
("lowInventoryFee", "Low inventory fee"), ("inboundPlacementFee", "Inbound placement fee"),
|
||
("returnFee", "Return fee"), ("costPerUnit", "Cost per unit"), ("vat", "VAT"),
|
||
]
|
||
_UNIT_FIELDS = {"saleUnits", "inventory"}
|
||
|
||
CSS = """
|
||
<style>
|
||
:root{
|
||
--ink:#111827; --ink-2:#374151; --muted:#6b7280; --faint:#9ca3af;
|
||
--line:#e5e7eb; --line-2:#f0f1f3; --paper:#ffffff; --soft:#f8f9fb;
|
||
--accent:#1d4ed8;
|
||
--ok:#15803d; --ok-bg:#f0fdf4; --ok-line:#bbf7d0;
|
||
--warn:#b45309; --warn-bg:#fffbeb; --warn-line:#fde68a;
|
||
--bad:#b91c1c; --bad-bg:#fef2f2; --bad-line:#fecaca;
|
||
}
|
||
.block-container{max-width:1180px; padding-top:2.4rem; padding-bottom:5rem;}
|
||
html,body,[data-testid="stAppViewContainer"]{
|
||
font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Inter,Roboto,Helvetica,Arial,sans-serif;
|
||
color:var(--ink);
|
||
}
|
||
#MainMenu,footer{visibility:hidden;}
|
||
|
||
/* ── Masthead ── */
|
||
.masthead{border-bottom:2px solid var(--ink); padding-bottom:.7rem; margin-bottom:.6rem;}
|
||
.mh-title{font-size:1.5rem; font-weight:640; letter-spacing:-.022em; line-height:1.2;}
|
||
.mh-sub{font-size:.82rem; color:var(--muted); margin-top:.25rem;}
|
||
|
||
/* ── Section label ── */
|
||
.sec{font-size:.7rem; font-weight:680; letter-spacing:.1em; text-transform:uppercase;
|
||
color:var(--faint); margin:1.9rem 0 .55rem; padding-bottom:.35rem;
|
||
border-bottom:1px solid var(--line);}
|
||
.sec:first-child{margin-top:.4rem;}
|
||
|
||
/* ── Status pill ── */
|
||
.pill{display:inline-block; font-size:.68rem; font-weight:700; letter-spacing:.075em;
|
||
text-transform:uppercase; padding:.22rem .55rem; border-radius:3px; border:1px solid;
|
||
vertical-align:.14em;}
|
||
.pill.ok{color:var(--ok); background:var(--ok-bg); border-color:var(--ok-line);}
|
||
.pill.warn{color:var(--warn); background:var(--warn-bg); border-color:var(--warn-line);}
|
||
.pill.bad{color:var(--bad); background:var(--bad-bg); border-color:var(--bad-line);}
|
||
.pill.mute{color:var(--muted); background:var(--soft); border-color:var(--line);}
|
||
|
||
/* ── Record header (SKU) ── */
|
||
.rec{display:flex; align-items:baseline; gap:.7rem; flex-wrap:wrap; margin-bottom:.15rem;}
|
||
.rec .sku{font-size:1.12rem; font-weight:640; letter-spacing:-.01em;
|
||
font-family:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;}
|
||
.rec-meta{font-size:.76rem; color:var(--faint); margin-bottom:1rem;}
|
||
|
||
/* ── Notes / banners ── */
|
||
.note{border:1px solid var(--line); border-left:3px solid var(--faint); background:var(--soft);
|
||
padding:.7rem .9rem; border-radius:0 4px 4px 0; font-size:.88rem; margin:.5rem 0 .9rem;
|
||
line-height:1.55;}
|
||
.note.bad{border-left-color:var(--bad); background:var(--bad-bg); border-color:var(--bad-line);}
|
||
.note.warn{border-left-color:var(--warn); background:var(--warn-bg); border-color:var(--warn-line);}
|
||
.note.ok{border-left-color:var(--ok); background:var(--ok-bg); border-color:var(--ok-line);}
|
||
.note.verdict{border-left-color:var(--ink); background:var(--paper); border-color:var(--line);}
|
||
.note b{font-weight:640;}
|
||
|
||
/* ── Stat grid ── */
|
||
.stats{display:grid; gap:0; border:1px solid var(--line); border-radius:5px; overflow:hidden;
|
||
grid-template-columns:repeat(auto-fit,minmax(150px,1fr)); background:var(--line);}
|
||
.stat{background:var(--paper); padding:.75rem .9rem .8rem;}
|
||
.stat .k{font-size:.68rem; font-weight:600; letter-spacing:.055em; text-transform:uppercase;
|
||
color:var(--faint); white-space:nowrap; overflow:hidden; text-overflow:ellipsis;}
|
||
.stat .v{font-size:1.28rem; font-weight:620; letter-spacing:-.018em; margin-top:.28rem;
|
||
font-variant-numeric:tabular-nums;}
|
||
.stat .s{font-size:.72rem; color:var(--muted); margin-top:.15rem; font-variant-numeric:tabular-nums;}
|
||
.stat .v.pos,.stat .s.pos{color:var(--ok);}
|
||
.stat .v.neg,.stat .s.neg{color:var(--bad);}
|
||
|
||
/* ── Ledger (profit bridge, fee cards) ── */
|
||
.ledger{border:1px solid var(--line); border-radius:5px; padding:.15rem .95rem;}
|
||
.lrow{display:flex; justify-content:space-between; align-items:baseline; gap:1rem;
|
||
padding:.5rem 0; border-bottom:1px solid var(--line-2); font-size:.87rem;}
|
||
.lrow:last-child{border-bottom:none;}
|
||
.lrow .lbl{color:var(--ink-2);}
|
||
.lrow .hint{color:var(--faint); font-size:.74rem; margin-left:.35rem;}
|
||
.lrow .val{font-variant-numeric:tabular-nums; font-weight:560; white-space:nowrap;}
|
||
.lrow.sub .lbl,.lrow.sub .val{color:var(--faint); font-size:.74rem;}
|
||
.lrow.total{border-top:1px solid var(--ink); border-bottom:none; margin-top:.1rem;}
|
||
.lrow.total .lbl{font-weight:660; color:var(--ink);}
|
||
.lrow.total .val{font-weight:700; font-size:1.02rem;}
|
||
.lrow.total .val.pos{color:var(--ok);} .lrow.total .val.neg{color:var(--bad);}
|
||
|
||
/* ── Reasons ── */
|
||
.reasons{border:1px solid var(--line); border-radius:5px; padding:.3rem 0; counter-reset:r;}
|
||
.reason{display:flex; gap:.75rem; padding:.55rem .95rem; font-size:.87rem; line-height:1.5;
|
||
border-bottom:1px solid var(--line-2);}
|
||
.reason:last-child{border-bottom:none;}
|
||
.reason .n{counter-increment:r; color:var(--faint); font-variant-numeric:tabular-nums;
|
||
font-weight:640; min-width:1.1rem;}
|
||
|
||
/* ── Summary rows (product line) ── */
|
||
.srow{display:flex; align-items:baseline; gap:.7rem; padding:.6rem .2rem;
|
||
border-bottom:1px solid var(--line-2); font-size:.86rem;}
|
||
.srow .sku{font-family:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace; font-weight:600;
|
||
min-width:16rem;}
|
||
.srow .body{color:var(--ink-2); line-height:1.5;}
|
||
.srow .body b,.srow .body .b{font-weight:640; color:var(--ink);}
|
||
.srow .body .neg{color:var(--bad); font-weight:640;}
|
||
.caption{font-size:.76rem; color:var(--muted); line-height:1.55; margin-top:.45rem;}
|
||
|
||
/* ── Streamlit widget restyle ── */
|
||
[data-testid="stVerticalBlockBorderWrapper"]:has(>div>[data-testid="stVerticalBlock"]){
|
||
border-radius:6px;}
|
||
.stButton>button{border-radius:5px; font-weight:560; font-size:.87rem; border:1px solid var(--line);
|
||
padding:.4rem .9rem; transition:none;}
|
||
.stButton>button:hover{border-color:var(--ink-2); color:var(--ink);}
|
||
.stButton>button[kind="primary"]{background:var(--ink); border-color:var(--ink); color:#fff;}
|
||
.stButton>button[kind="primary"]:hover{background:var(--accent); border-color:var(--accent);}
|
||
.stTextInput input,.stNumberInput input,.stSelectbox div[data-baseweb="select"]>div{
|
||
border-radius:5px; font-size:.88rem;}
|
||
[data-testid="stTable"] table{font-size:.82rem; border-collapse:collapse;}
|
||
[data-testid="stTable"] thead th{font-size:.67rem; font-weight:660; letter-spacing:.06em;
|
||
text-transform:uppercase; color:var(--faint); border-bottom:1px solid var(--ink)!important;
|
||
background:transparent;}
|
||
[data-testid="stTable"] tbody td{border-bottom:1px solid var(--line-2)!important;
|
||
font-variant-numeric:tabular-nums;}
|
||
[data-testid="stTable"] tbody tr:hover{background:var(--soft);}
|
||
[data-testid="stExpander"] summary{font-size:.84rem; font-weight:560;}
|
||
[data-testid="stExpander"] details{border-radius:5px; border-color:var(--line);}
|
||
.stTabs [data-baseweb="tab"]{font-size:.85rem; font-weight:560;}
|
||
[data-testid="stMetricValue"]{font-size:1.3rem; font-weight:620;}
|
||
hr{margin:1.6rem 0; border-color:var(--line);}
|
||
</style>
|
||
"""
|
||
|
||
|
||
# ── Rendering primitives ───────────────────────────────────────────────
|
||
def _html(markup: str) -> None:
|
||
"""Emit raw HTML. '$' is entity-escaped so Streamlit's markdown pass
|
||
doesn't mistake dollar amounts for inline LaTeX."""
|
||
st.markdown(markup.replace("$", "$"), unsafe_allow_html=True)
|
||
|
||
|
||
def _md(s) -> str:
|
||
"""Escape '$' for plain Streamlit markdown (same LaTeX problem)."""
|
||
return str(s).replace("$", "\\$")
|
||
|
||
|
||
_money = usd # accounting style: the sign leads the symbol (-$3.55, never $-3.55)
|
||
|
||
|
||
def _tone(v) -> str:
|
||
try:
|
||
return "neg" if float(v) < 0 else "pos"
|
||
except (TypeError, ValueError):
|
||
return ""
|
||
|
||
|
||
def _sec(title: str) -> None:
|
||
_html(f'<div class="sec">{title}</div>')
|
||
|
||
|
||
def _note(kind: str, body: str) -> None:
|
||
_html(f'<div class="note {kind}">{body}</div>')
|
||
|
||
|
||
def _pill(rating: str) -> str:
|
||
return f'<span class="pill {TONE.get(rating, "mute")}">{rating.title()}</span>'
|
||
|
||
|
||
def _stats(items) -> None:
|
||
"""items: (label, value, sub|None, tone|None)."""
|
||
cells = ""
|
||
for label, value, sub, tone in items:
|
||
t = f" {tone}" if tone else ""
|
||
s = f'<div class="s{t}">{sub}</div>' if sub else ""
|
||
cells += f'<div class="stat"><div class="k">{label}</div><div class="v{t}">{value}</div>{s}</div>'
|
||
_html(f'<div class="stats">{cells}</div>')
|
||
|
||
|
||
def _ledger(rows) -> str:
|
||
"""rows: (label, hint|None, value, is_total)."""
|
||
out = ""
|
||
for label, hint, value, total in rows:
|
||
cls = "lrow total" if total else "lrow"
|
||
vt = f" {_tone(value)}" if total else ""
|
||
h = f'<span class="hint">{hint}</span>' if hint else ""
|
||
out += (f'<div class="{cls}"><span class="lbl">{label}{h}</span>'
|
||
f'<span class="val{vt}">{_money(value)}</span></div>')
|
||
return f'<div class="ledger">{out}</div>'
|
||
|
||
|
||
def _calc_card(detail: dict, order, highlight: str) -> str:
|
||
"""Field → value ledger; the highlighted total is always rendered last."""
|
||
rows = ""
|
||
for key, label in order:
|
||
if key == highlight or key not in detail or detail[key] is None:
|
||
continue
|
||
v = detail[key]
|
||
disp = f"{float(v):,.0f}" if key in _UNIT_FIELDS else _money(v)
|
||
rows += f'<div class="lrow"><span class="lbl">{label}</span><span class="val">{disp}</span></div>'
|
||
cb = detail.get("costBreakdown")
|
||
if isinstance(cb, dict) and cb:
|
||
parts = " · ".join(f"{k} {_money(v)}" for k, v in cb.items())
|
||
rows += (f'<div class="lrow"><span class="lbl">Cost breakdown</span>'
|
||
f'<span class="val">{_money(sum(cb.values()))}</span></div>'
|
||
f'<div class="lrow sub"><span class="lbl">{parts}</span><span class="val"></span></div>')
|
||
if highlight in detail and detail[highlight] is not None:
|
||
label = next((l for k, l in order if k == highlight), "Take home")
|
||
v = detail[highlight]
|
||
rows += (f'<div class="lrow total"><span class="lbl">{label}</span>'
|
||
f'<span class="val {_tone(v)}">{_money(v)}</span></div>')
|
||
return f'<div class="ledger">{rows}</div>'
|
||
|
||
|
||
# ── Services ───────────────────────────────────────────────────────────
|
||
@st.cache_resource
|
||
def get_service():
|
||
return _build_service(get_settings())
|
||
|
||
|
||
@st.cache_data(ttl=3600, show_spinner="Loading product catalog (one-time)…")
|
||
def all_catalog_skus() -> list[str]:
|
||
"""Full catalog SKU list (demand-ordered), cached for the session."""
|
||
return get_service().list_all_skus()
|
||
|
||
|
||
# ── Result view ────────────────────────────────────────────────────────
|
||
def render_result(r: AnalysisResult, defer_narrative: bool = False):
|
||
"""Render one analysed SKU. With `defer_narrative`, the Analysis section reserves a
|
||
slot and returns it, so the caller can render the numbers immediately and fill in the
|
||
(slower) written analysis afterwards."""
|
||
n_months = len(r.price_performance)
|
||
slot = None
|
||
|
||
_html(f'<div class="rec"><span class="sku">{r.sku}</span>{_pill(r.rating.value)}</div>'
|
||
f'<div class="rec-meta">'
|
||
+ (f"ASIN {r.asin} · " if r.asin else "")
|
||
+ f"Evaluated at {_money(r.price)}</div>")
|
||
|
||
# Actual results outrank any modelled margin, so they open the record.
|
||
if r.actual_profit_per_unit is not None and r.actual_profit_per_unit < 0:
|
||
_note("bad", f"<b>Actually losing money — {_money(r.actual_profit_per_unit)} per unit.</b> "
|
||
"Real COSMOS profit over the last 3 months, including storage, refunds, promo "
|
||
"and ads."
|
||
+ (f" Lost money in <b>{r.unprofitable_months} of {n_months}</b> observed months."
|
||
if r.unprofitable_months else ""))
|
||
elif r.unprofitable_months >= 3:
|
||
_note("bad", f"<b>Unprofitable in {r.unprofitable_months} of the last {n_months} months</b> "
|
||
"on actual profit.")
|
||
elif r.is_losing_money:
|
||
_note("bad", f"<b>Losing money after ads.</b> Net {_money(r.net_per_unit_incl_ad)} per unit at "
|
||
f"{_money(r.price)} once {_money(r.ad_cost_per_unit)} of ad spend is counted.")
|
||
|
||
_note("verdict", f"<b>Verdict.</b> {r.headline}")
|
||
|
||
delta = (f"{r.suggested_price - r.current_price:+.2f} vs current"
|
||
if (r.suggested_price and r.current_price) else None)
|
||
_stats([
|
||
("Current price", _money(r.current_price), "Live, from COSMOS", None),
|
||
("Suggested", _money(r.suggested_price), delta or "—", None),
|
||
("Break-even", _money(r.break_even), "Excludes ads", None),
|
||
("Margin", f"{r.margin_pct*100:.1f}%", "Pre-ads", None),
|
||
("Take-home / unit", _money(r.take_home_per_unit), "Pre-ads", None),
|
||
])
|
||
|
||
if r.competitive is not None or r.gate_decision:
|
||
_sec("Buy Box & competitors — live Amazon (Apify)")
|
||
comp = r.competitive
|
||
gate_tone = {"BLOCKED": "neg", "NEEDS_REVIEW": "neg"}.get(r.gate_decision or "")
|
||
stats = []
|
||
if r.gate_decision:
|
||
stats.append(("Pricing gate", r.gate_decision, "Deterministic rules", gate_tone))
|
||
if comp is not None:
|
||
stats.append((
|
||
"Buy Box",
|
||
comp.buy_box_status.value,
|
||
(f"{comp.buy_box_seller_name or '—'} @ {_money(comp.buy_box_price)}"
|
||
if comp.buy_box_price is not None
|
||
else (comp.reason[:80] if comp.reason else "—")),
|
||
"neg" if comp.is_suppressed else None,
|
||
))
|
||
if comp.competitive_median is not None:
|
||
stats.append((
|
||
"Market band",
|
||
_money(comp.competitive_median),
|
||
f"{_money(comp.competitive_low)} – {_money(comp.competitive_high)}",
|
||
None,
|
||
))
|
||
stats.append((
|
||
"Offers",
|
||
str(len(comp.offers)),
|
||
"Named sellers on this ASIN",
|
||
None,
|
||
))
|
||
if stats:
|
||
_stats(stats)
|
||
# COSMOS current vs live Amazon featured — common source of “UI ≠ presentation”
|
||
if comp is not None and comp.buy_box_price is not None:
|
||
ref = r.current_price if r.current_price is not None else r.price
|
||
if ref and abs(comp.buy_box_price - ref) / ref >= 0.05:
|
||
_note(
|
||
"warn",
|
||
f"<b>Price mismatch.</b> COSMOS/analyzed {_money(ref)} but Amazon Buy Box "
|
||
f"is {_money(comp.buy_box_price)} "
|
||
f"({comp.buy_box_seller_name or 'seller'}). "
|
||
f"Margin at the COSMOS price can look fine while the live listing is cheaper. "
|
||
f"Check <b>Test a specific price</b> and re-run at {_money(comp.buy_box_price)} "
|
||
f"(or your candidate, e.g. $24.99) to match presentation-style reasoning."
|
||
)
|
||
if r.gate_reasons:
|
||
for gr in r.gate_reasons:
|
||
_note("warn" if r.gate_decision == "NEEDS_REVIEW"
|
||
else ("bad" if r.gate_decision == "BLOCKED" else "ok"),
|
||
f"<b>Gate.</b> {gr}")
|
||
if comp is not None and comp.offers:
|
||
st.table([{
|
||
"Seller": o.seller_name,
|
||
"Price": _money(o.price) if o.price is not None else "—",
|
||
"Buy Box": "Yes" if o.is_buy_box else "",
|
||
"Condition": o.condition or "",
|
||
} for o in comp.offers])
|
||
_html('<div class="caption">Sellers and prices scraped from the ASIN product page '
|
||
'(Apify). Same listing — other merchants competing for the Buy Box — not rival '
|
||
'product ASINs.</div>')
|
||
elif comp is not None and not comp.offers:
|
||
_note("warn", "Apify returned no named competitor offers this scrape "
|
||
"(Buy Box may be suppressed or Amazon blocked the offers panel).")
|
||
|
||
if r.ad_cost_per_unit is not None:
|
||
_sec("Observed evidence — what actually happened")
|
||
_stats([
|
||
("Best observed price", _money(r.best_observed_price),
|
||
(f"{r.best_observed_price - r.price:+.2f} vs current"
|
||
if (r.best_observed_price and r.price) else "Highest real profit/day"), None),
|
||
("Actual profit / day there", _money(r.best_observed_profit_day), "At that price",
|
||
_tone(r.best_observed_profit_day)),
|
||
("Actual profit / unit", _money(r.actual_profit_per_unit), "Last 3 months",
|
||
_tone(r.actual_profit_per_unit)),
|
||
("Ad cost / unit", _money(r.ad_cost_per_unit), "6-mo spend ÷ units", None),
|
||
])
|
||
|
||
# Reconciles COSMOS take-home with real profit — without it, +$4.30 and
|
||
# −$1.90 look contradictory.
|
||
if r.ad_cost_per_unit is not None and r.actual_profit_per_unit is not None:
|
||
_sec("Profit bridge — why take-home is not real profit")
|
||
_html(_ledger([
|
||
("Take-home / unit", "price − referral − FBA − inbound − returns − COGS",
|
||
r.take_home_per_unit, False),
|
||
("Less ad spend / unit", "not included in take-home", -(r.ad_cost_per_unit or 0), False),
|
||
("Net after ads", "modelled", r.net_per_unit_incl_ad, True),
|
||
("Less refunds, promo, storage, logistics", "also not in take-home",
|
||
-(r.unmodeled_cost_gap or 0), False),
|
||
("Actual profit / unit", "COSMOS's real profit field", r.actual_profit_per_unit, True),
|
||
]))
|
||
if r.unmodeled_cost_gap and r.unmodeled_cost_gap > 0.25:
|
||
_note("warn", f"The modelled margin <b>overstates profit by {_money(r.unmodeled_cost_gap)} "
|
||
"per unit</b> (unmodeled storage, refunds, promo, logistics). Trust the "
|
||
"actual figures.")
|
||
|
||
if r.reasons:
|
||
_sec("Why this verdict — every reason")
|
||
rows = "".join(f'<div class="reason"><span class="n">{i}</span><span>{reason}</span></div>'
|
||
for i, reason in enumerate(r.reasons, 1))
|
||
_html(f'<div class="reasons">{rows}</div>'
|
||
'<div class="caption">These are the exact deterministic rule outputs that produced '
|
||
'the verdict — no AI involved. The AI analysis only explains them.</div>')
|
||
|
||
if r.narrative:
|
||
_sec("Analysis")
|
||
with st.container(border=True):
|
||
st.markdown(_md(r.narrative))
|
||
elif defer_narrative:
|
||
_sec("Analysis")
|
||
slot = st.empty()
|
||
slot.caption("Writing the analysis…")
|
||
|
||
if r.price_performance:
|
||
_sec("Price performance — actual, by month")
|
||
best = max(r.price_performance, key=lambda m: m["actual_profit_per_day"])
|
||
st.table([{
|
||
"Month": m["month"],
|
||
"Price": _money(m["avg_price"]),
|
||
"Units/day": f"{m['units_per_day']:,.0f}",
|
||
"Actual profit/day": _money(m["actual_profit_per_day"]),
|
||
"Profit/unit": _money(m["profit_per_unit"]),
|
||
"Ad/unit": _money(m["ad_per_unit"]),
|
||
"": "◀ best" if m is best else "",
|
||
} for m in r.price_performance])
|
||
_html('<div class="caption">Real COSMOS profit, including storage, refunds, promo and ads. '
|
||
'Months confound seasonality, ad spend, promotions and stock availability — read this '
|
||
'as evidence in context, not proof of causation.</div>')
|
||
|
||
_sec("Demand & inventory")
|
||
_stats([
|
||
("7-day", f"{(r.avg_7d or 0):,.0f}/day", None, None),
|
||
("30-day", f"{(r.avg_30d or 0):,.0f}/day", None, None),
|
||
("6-month", f"{(r.avg_6m or 0):,.0f}/day", r.trend.title(), None),
|
||
("Inventory", f"{r.inventory:,.0f}",
|
||
f"{r.cover_days} days cover" if r.cover_days is not None else None, None),
|
||
])
|
||
if r.monthly_take_home is not None:
|
||
_html(f'<div class="caption">Estimated monthly take-home {_money(r.monthly_take_home)}, '
|
||
f'after {_money(r.storage_charges)} of storage on current stock.</div>')
|
||
|
||
with st.expander("Evidence used — verify the numbers"):
|
||
st.markdown(_md(
|
||
f"- **Analyzed at** {_money(r.price)} · **current price** {_money(r.current_price)}\n"
|
||
f"- **Take-home/unit** (COSMOS, excl. ads) {_money(r.take_home_per_unit)} · "
|
||
f"**margin** {r.margin_pct*100:.1f}% · **break-even** {_money(r.break_even)}\n"
|
||
f"- **Ad cost/unit** {_money(r.ad_cost_per_unit)} · "
|
||
f"**net after ads (modelled)** {_money(r.net_per_unit_incl_ad)}\n"
|
||
f"- **Actual profit/unit**, last 3 months: {_money(r.actual_profit_per_unit)} · "
|
||
f"**unprofitable months** {r.unprofitable_months}/{n_months}\n"
|
||
f"- **Best observed price** {_money(r.best_observed_price)} → "
|
||
f"{_money(r.best_observed_profit_day)}/day actual profit\n"
|
||
f"- **Model overstates profit by** {_money(r.unmodeled_cost_gap)}/unit\n"
|
||
f"- **6-mo demand** {(r.avg_6m or 0):,.0f}/day ({r.trend}, {r.demand_change_pct}% vs 6-mo) · "
|
||
f"**inventory** {r.inventory:,.0f} ({r.cover_days} days cover)"))
|
||
|
||
# The model is secondary to observed evidence, so it stays collapsed.
|
||
if r.demand_curve:
|
||
with st.expander("Model — elasticity & profit curve (directional only)"):
|
||
if r.elasticity and r.elasticity.get("elasticity") is not None:
|
||
e = r.elasticity
|
||
st.caption(f"Elasticity {e['elasticity']} · confidence {e['confidence']} · "
|
||
f"r²={e.get('r2')} over {e['n']} months, price moved "
|
||
f"{e.get('price_spread_pct')}%.")
|
||
st.table([{
|
||
"Price": _money(s["price"]),
|
||
"Exp. units/day": f"{s['units_per_day']:,.0f}",
|
||
"Net/unit after ad": _money(s["net_per_unit"]),
|
||
"Modelled daily profit": _money(s["daily_profit"]),
|
||
"": "◀ optimum" if (r.profit_optimal_price
|
||
and abs(s["price"] - r.profit_optimal_price) < 0.01) else "",
|
||
} for s in r.demand_curve])
|
||
st.caption("A model, not evidence. "
|
||
+ ("The optimum lies beyond the observed price range — treat it as a "
|
||
"direction, never an exact target." if r.extrapolated
|
||
else "The optimum is within the observed price range.")
|
||
+ " The actual-performance table above outranks it.")
|
||
elif r.scenarios:
|
||
_sec("Price scenarios — volume held constant")
|
||
st.table([{"Price": _money(s["price"]), "Margin": f"{s['margin_pct']*100:.1f}%",
|
||
"Take-home/unit": _money(s["take_home_unit"]),
|
||
"Monthly profit": _money(s["monthly_profit_held"]),
|
||
"": "◀ current" if s["is_current"] else ""} for s in r.scenarios])
|
||
_html('<div class="caption">At current volume — elasticity is not available for this SKU.</div>')
|
||
|
||
if r.fees_detail or r.bulk_detail:
|
||
_sec("Source calculations")
|
||
cols = st.columns(2, gap="medium")
|
||
if r.fees_detail:
|
||
with cols[0]:
|
||
st.caption("FBA profitability")
|
||
_html(_calc_card(r.fees_detail, FEE_ORDER, highlight="takeHome"))
|
||
_html('<div class="caption">This “take home” excludes ad spend, refunds, promo and '
|
||
'logistics — see the profit bridge for real per-unit profit.</div>')
|
||
if r.bulk_detail:
|
||
with cols[1]:
|
||
st.caption("Bulk — one month against current stock")
|
||
_html(_calc_card(r.bulk_detail, BULK_ORDER, highlight="takeHome"))
|
||
return slot
|
||
|
||
|
||
# ── Workflows ──────────────────────────────────────────────────────────
|
||
@st.cache_data(ttl=1800, show_spinner=False)
|
||
def sku_history(sku: str):
|
||
"""Six months of daily actuals for one SKU. Cached: the twelve windows behind it are
|
||
the single most expensive call in the app, and re-analysing a SKU is common."""
|
||
return get_service().get_sales_history(sku, days=180)
|
||
|
||
|
||
def run_single(sku: str, price: float | None) -> None:
|
||
"""Full evidence-based analysis of one SKU (at a given price, or its current price).
|
||
|
||
The numbers render as soon as they exist; the written analysis lands afterwards, so
|
||
the page is useful in seconds rather than after the model finishes.
|
||
"""
|
||
settings, svc = get_settings(), get_service()
|
||
at = f"at ${price:.2f}" if price else "at its current price"
|
||
with_competitive = bool(settings.apify_token)
|
||
with st.status(f"Analyzing {sku} {at}…", expanded=True) as status:
|
||
status.write("Six-month actual profit history, ad cost and best observed price")
|
||
history = sku_history(sku)
|
||
status.write(f"Loaded {len(history)} days of history")
|
||
status.write("Fees, landed cost, current price, demand and inventory")
|
||
if with_competitive:
|
||
status.write("Buy Box + competitor offers via Apify (may take ~30–90s)")
|
||
kw = dict(svc=svc, with_narrative=False, with_elasticity=True, history=history,
|
||
with_competitive=with_competitive)
|
||
r = (analyze_price(sku, price, settings, **kw) if price
|
||
else analyze_sku(sku, settings, **kw))
|
||
status.update(label=f"{sku} — {r.rating.value.title()}"
|
||
+ (f" · gate {r.gate_decision}" if r.gate_decision else ""),
|
||
state="complete", expanded=False)
|
||
|
||
slot = render_result(r, defer_narrative=True)
|
||
if slot is not None:
|
||
r.narrative = generate_analysis_narrative(r)
|
||
with slot.container(border=True):
|
||
st.markdown(_md(r.narrative))
|
||
|
||
|
||
def run_catalog_scan(days: int) -> None:
|
||
"""Catalog-wide money-at-risk scan using COSMOS's ACTUAL profit. Read-only."""
|
||
from pricing_agent.performance import fix_suggestion, loss_reason, root_cause_split
|
||
|
||
svc = get_service()
|
||
with st.status(f"Scanning the catalog — last {days} days…", expanded=True) as status:
|
||
status.write("Pulling every SKU's daily actuals from COSMOS (bulk, paginated)")
|
||
status.write("Summing real profit, ad spend and units per SKU")
|
||
status.write("Diagnosing each money-loser: ads or below cost")
|
||
rows = svc.get_catalog_performance(days=days)
|
||
status.update(label=f"Scanned {len(rows):,} SKUs with sales",
|
||
state="complete", expanded=False)
|
||
if not rows:
|
||
_note("warn", "No SKUs had sales in that window.")
|
||
return
|
||
|
||
losers = [r for r in rows if r["actual_profit"] < 0]
|
||
thin = [r for r in rows if 0 <= r["profit_per_unit"] < 0.50]
|
||
ads_cause, cost_cause = root_cause_split(losers)
|
||
bleed = sum(r["actual_profit"] for r in losers)
|
||
gain = sum(r["actual_profit"] for r in rows if r["actual_profit"] > 0)
|
||
|
||
_sec(f"Money at risk — last {days} days, actual COSMOS profit")
|
||
_stats([
|
||
("SKUs with sales", f"{len(rows):,}", None, None),
|
||
("Losing money", f"{len(losers):,}", _money(bleed), "neg"),
|
||
("Thin, under $0.50/unit", f"{len(thin):,}", None, None),
|
||
("Net across catalog", _money(gain + bleed), None, _tone(gain + bleed)),
|
||
("Annualized bleed", _money(bleed * 365 / days), "from the money-losers", "neg"),
|
||
])
|
||
|
||
_sec("Root cause — which lever to pull")
|
||
_stats([
|
||
("Ads are the cause", f"{len(ads_cause):,} SKUs",
|
||
"Profitable without ad spend — cut or retarget ads", None),
|
||
("Below cost before ads", f"{len(cost_cause):,} SKUs",
|
||
"Loses money at zero ad spend — raise price or cut COGS", None),
|
||
])
|
||
|
||
def _table(items, n=100):
|
||
return [{
|
||
"SKU": r["sku"], "Price": _money(r["avg_price"]), "Units": f"{r['units']:,}",
|
||
"Ad/unit": _money(r["ad_per_unit"]), "Profit/unit": _money(r["profit_per_unit"]),
|
||
"Actual profit": _money(r["actual_profit"]),
|
||
"Why it loses money": loss_reason(r["profit_per_unit"], r["ad_per_unit"], r["avg_price"]),
|
||
"How to fix it": fix_suggestion(r["profit_per_unit"], r["ad_per_unit"], r["avg_price"]),
|
||
} for r in items[:n]]
|
||
|
||
t1, t2, t3 = st.tabs([f"Losing money · {len(losers):,}",
|
||
f"Ads are the cause · {len(ads_cause):,}",
|
||
f"Thin margin · {len(thin):,}"])
|
||
with t1:
|
||
st.dataframe(_table(losers), use_container_width=True, hide_index=True)
|
||
_html('<div class="caption">Worst first. Every row states why it loses money and how to fix '
|
||
'it. To verify: if <code>profit/unit + ad/unit > 0</code>, ads caused the loss.</div>')
|
||
with t2:
|
||
st.dataframe(_table(sorted(ads_cause, key=lambda r: r["actual_profit"])),
|
||
use_container_width=True, hide_index=True)
|
||
_note("ok", "These flip to profitable by <b>cutting ad spend alone</b> — no price change, no "
|
||
"customer impact. The “how to fix it” column gives the target ad spend per unit.")
|
||
with t3:
|
||
st.dataframe(_table(sorted(thin, key=lambda r: -r["units"])),
|
||
use_container_width=True, hide_index=True)
|
||
_html('<div class="caption">Read-only — nothing is written anywhere. Open any SKU under '
|
||
'<b>Single product</b> for its full evidence-based report.</div>')
|
||
|
||
|
||
def run_product_line(token: str, total: int, skus: list) -> None:
|
||
"""Analyze a chosen set of a product line's SKUs, then a one-line summary report."""
|
||
settings, svc = get_settings(), get_service()
|
||
subset = total > len(skus)
|
||
_sec(f"Product line “{token}” — {len(skus)} of {total} SKUs, highest demand first")
|
||
rows = []
|
||
prog = st.progress(0.0, text="Loading the line's six-month actual profit history…")
|
||
status = st.status(f"Analyzing {len(skus)} SKUs…", expanded=True)
|
||
|
||
# One shared fetch of the whole line's 6-month daily actuals: `skuPrefix` matches by
|
||
# substring, so this costs ~12 calls TOTAL instead of ~12 per SKU. Every SKU then gets
|
||
# the same full evidence (ad cost, real profit, best observed price) as Single product.
|
||
history = {}
|
||
try:
|
||
status.write("Loading six-month actual profit history for the whole line (shared)")
|
||
prog.progress(0.0, text=f"Fetching the line's history (bounded to the {len(skus)} "
|
||
f"selected SKUs)…")
|
||
history = svc.get_sales_history_bulk(token, days=180, wanted=set(skus))
|
||
status.write(f"History loaded for {len(history)} of {len(skus)} SKUs")
|
||
except Exception as e:
|
||
status.write(f"History unavailable ({e}) — falling back to margin-only analysis")
|
||
|
||
for i, sku in enumerate(skus, 1):
|
||
status.update(label=f"Analyzing {i}/{len(skus)} — {sku}")
|
||
try:
|
||
hist = history.get(sku)
|
||
r = analyze_sku(sku, settings, svc=svc, with_narrative=False,
|
||
with_elasticity=bool(hist), history=hist)
|
||
rows.append(r)
|
||
actual = (f" · actual {_money(r.actual_profit_per_unit)}/unit"
|
||
if r.actual_profit_per_unit is not None else "")
|
||
status.write(f"{sku} — {r.rating.value.title()}{actual}")
|
||
head = (f"{i}. {sku} — {r.rating.value.title()} · now {_money(r.current_price)}, "
|
||
f"suggest {_money(r.suggested_price)}{actual}")
|
||
with st.expander(head, expanded=False):
|
||
render_result(r)
|
||
except Exception as e:
|
||
status.write(f"{sku} — failed: {e}")
|
||
prog.progress(i / len(skus), text=f"{i} of {len(skus)} analyzed")
|
||
prog.empty()
|
||
status.update(label=f"Analyzed {len(rows)} of {len(skus)} SKUs", state="complete",
|
||
expanded=False)
|
||
|
||
if not rows:
|
||
return
|
||
|
||
order = {"POOR": 0, "CAUTION": 1, "GOOD": 2}
|
||
losing = [r for r in rows if r.actual_profit_per_unit is not None
|
||
and r.actual_profit_per_unit < 0]
|
||
counts = {k: sum(1 for r in rows if r.rating.value == k) for k in ("GOOD", "CAUTION", "POOR")}
|
||
|
||
_sec("Summary report")
|
||
_stats([
|
||
("SKUs analyzed", f"{len(rows)}", f"of {total} in the line" if subset else "whole line", None),
|
||
("Good", f"{counts['GOOD']}", None, "pos" if counts["GOOD"] else None),
|
||
("Caution", f"{counts['CAUTION']}", None, None),
|
||
("Poor", f"{counts['POOR']}", None, "neg" if counts["POOR"] else None),
|
||
("Actually losing money", f"{len(losing)}", "on real COSMOS profit",
|
||
"neg" if losing else None),
|
||
])
|
||
|
||
if losing:
|
||
_note("bad", f"<b>{len(losing)} of {len(rows)} SKUs are actually losing money</b> on real "
|
||
"COSMOS profit, which includes ads, refunds, promo and storage.")
|
||
|
||
body = ""
|
||
for r in sorted(rows, key=lambda x: order.get(x.rating.value, 3)):
|
||
move = ""
|
||
if r.current_price and r.suggested_price:
|
||
d = r.suggested_price - r.current_price
|
||
move = f" ({d:+.2f})" if abs(d) >= 0.5 else " (hold)"
|
||
if r.actual_profit_per_unit is not None:
|
||
cls = "neg" if r.actual_profit_per_unit < 0 else "b"
|
||
econ = (f'<span class="{cls}">actual {_money(r.actual_profit_per_unit)}/unit</span> '
|
||
f"(ads {_money(r.ad_cost_per_unit)}/unit) · {r.unprofitable_months} of "
|
||
f"{len(r.price_performance)} months lost money")
|
||
else:
|
||
econ = f"{r.margin_pct*100:.0f}% margin (pre-ads, no history)"
|
||
body += (f'<div class="srow">{_pill(r.rating.value)}'
|
||
f'<span class="sku">{r.sku}</span>'
|
||
f'<span class="body">now <b>{_money(r.current_price)}</b>, suggest '
|
||
f'<b>{_money(r.suggested_price)}</b>{move} · {econ} · '
|
||
f'6-mo {(r.avg_6m or 0):,.0f}/day ({r.trend}) — {r.headline}</span></div>')
|
||
_html(body)
|
||
|
||
|
||
# ── Page ───────────────────────────────────────────────────────────────
|
||
_html(CSS)
|
||
_html('<div class="masthead"><div class="mh-title">Pricing & Profitability Analyst</div>'
|
||
'<div class="mh-sub">Amazon Seller Central · live COSMOS data · read-only. Verdicts come from '
|
||
'deterministic rules over actual profit; the language model only explains them. '
|
||
'Single-SKU analysis also pulls Buy Box / competitor sellers via Apify when '
|
||
'<code>APIFY_TOKEN</code> is set (product-line bulk skips Apify for speed).</div></div>')
|
||
|
||
if not get_settings().openai_api_key:
|
||
_note("warn", "No <code>OPENAI_API_KEY</code> is set — rule-based analysis only, no written "
|
||
"narrative.")
|
||
|
||
if "action" not in st.session_state:
|
||
st.session_state.action = None
|
||
|
||
left, right = st.columns(2, gap="large")
|
||
|
||
with left:
|
||
with st.container(border=True):
|
||
st.markdown("##### Single product")
|
||
st.caption("Full evidence: six-month real profit, ad cost, best observed price, written "
|
||
"analysis.")
|
||
sku_in = st.text_input("SKU", key="single_sku",
|
||
placeholder="UBMICROFIBERGUSSETPILLOWWHITEQUEEN")
|
||
use_price = st.checkbox(
|
||
"Test a specific price instead of the current one",
|
||
key="use_price",
|
||
help="Unchecked = COSMOS current price (can lag Amazon). "
|
||
"Checked = evaluate a candidate (e.g. $24.99 for the presentation case).",
|
||
)
|
||
price_in = None
|
||
if use_price:
|
||
price_in = st.number_input("Price", min_value=0.01, value=24.99, step=0.01,
|
||
key="single_price")
|
||
st.caption("Presentation case: SKU above @ **$24.99** → expect margin floor "
|
||
"**BLOCKED** + Apify competitor table.")
|
||
if st.button("Analyze product", type="primary", use_container_width=True,
|
||
disabled=not sku_in.strip()):
|
||
st.session_state.action = ("single", sku_in.strip(), price_in)
|
||
|
||
with right:
|
||
with st.container(border=True):
|
||
st.markdown("##### Product line")
|
||
st.caption("Every SKU in a line, worst first, with a one-line summary each.")
|
||
line_in = st.text_input("Product line or SKU prefix", key="line_q",
|
||
placeholder="PILLOW · TOWEL · UBCFK · MATTRESSPROTECTOR")
|
||
query = line_in.strip()
|
||
matches, total = [], 0
|
||
if query:
|
||
needle = query.upper()
|
||
matches = [s for s in all_catalog_skus() if needle in s.upper()]
|
||
total = len(matches)
|
||
|
||
# Streamlit keeps a widget's value in session_state across reruns. When the line
|
||
# changes, `total` (the max) changes too — a stale value would exceed the new max
|
||
# and the input would refuse to update. Reset it whenever the line changes, and
|
||
# clamp it if it ever overshoots.
|
||
if st.session_state.get("_last_line") != query:
|
||
st.session_state.pop("line_n", None)
|
||
st.session_state["_last_line"] = query
|
||
elif total and st.session_state.get("line_n", 0) > total:
|
||
st.session_state["line_n"] = min(PRODUCT_LINE_LIMIT, total)
|
||
|
||
if query and total:
|
||
st.caption(f"**{total} SKUs** match this line.")
|
||
# Quick presets — set the value BEFORE the number_input is created.
|
||
presets = [("5", 5), ("15", 15), ("25", 25), (f"All ({total})", total)]
|
||
for col, (label, val) in zip(st.columns(len(presets)), presets):
|
||
if col.button(label, key=f"preset_{label}", use_container_width=True):
|
||
st.session_state["line_n"] = max(1, min(val, total))
|
||
st.rerun()
|
||
|
||
n_in = st.number_input(f"How many to analyze — 1 to {total}, highest demand first",
|
||
min_value=1, max_value=total,
|
||
value=min(PRODUCT_LINE_LIMIT, total), step=1, key="line_n")
|
||
mins = (60 + int(n_in) * 2.0) / 60 # one shared history fetch + per-SKU analysis
|
||
st.caption(f"About {mins:.1f} min — loads the line's six-month actual profit history "
|
||
f"once, then analyzes {int(n_in)} SKUs. Stop takes effect between SKUs; a "
|
||
f"data call already in flight has to finish first.")
|
||
if st.button("Analyze line", type="primary", use_container_width=True):
|
||
st.session_state.action = ("line", query, total, matches[:int(n_in)])
|
||
elif query:
|
||
st.caption("No SKUs match that text. Try `PILLOW`, `TOWEL` or `UBCFK`.")
|
||
st.button("Analyze line", use_container_width=True, disabled=True)
|
||
else:
|
||
st.caption("Type a product line above to see how many SKUs it contains.")
|
||
st.button("Analyze line", use_container_width=True, disabled=True)
|
||
|
||
with st.container(border=True):
|
||
s1, s2, s3 = st.columns([3, 1, 1])
|
||
with s1:
|
||
st.markdown("##### Catalog money-at-risk scan")
|
||
st.caption("Every SKU's actual profit, including storage, refunds, promo and ads. Finds the "
|
||
"money-losers and says whether ads or price is to blame. Read-only.")
|
||
with s2:
|
||
scan_days = st.selectbox("Window", [15, 30, 60], index=1, key="scan_days",
|
||
format_func=lambda d: f"Last {d} days")
|
||
with s3:
|
||
st.write("")
|
||
if st.button("Run scan", use_container_width=True):
|
||
st.session_state.action = ("scan", int(scan_days))
|
||
|
||
st.divider()
|
||
|
||
act = st.session_state.action
|
||
if not act:
|
||
_note("", "Analyze a <b>single product</b> for the full evidence-based report, scan a "
|
||
"<b>product line</b>, or run the <b>catalog money-at-risk scan</b> to find every SKU "
|
||
"that is actually losing money.")
|
||
else:
|
||
# Rendered BEFORE the long run so it stays clickable while the analysis streams. Clicking
|
||
# any widget mid-run makes Streamlit abort the script and rerun — that is the stop.
|
||
stop_col, label_col = st.columns([1, 5])
|
||
if stop_col.button("Stop / clear", use_container_width=True, key="stop_btn"):
|
||
st.session_state.action = None
|
||
st.rerun()
|
||
label_col.caption("Press stop to cancel a running analysis, or to clear these results.")
|
||
|
||
cap = ListHandler()
|
||
logging.getLogger("pricing_agent").addHandler(cap)
|
||
try:
|
||
if act[0] == "single":
|
||
run_single(act[1], act[2])
|
||
elif act[0] == "scan":
|
||
run_catalog_scan(act[1])
|
||
else:
|
||
run_product_line(act[1], act[2], act[3])
|
||
except Exception as e:
|
||
_note("bad", f"Analysis failed: {e}")
|
||
finally:
|
||
logging.getLogger("pricing_agent").removeHandler(cap)
|
||
if cap.records:
|
||
with st.expander(f"Steps and data fetched — {len(cap.records)} calls logged"):
|
||
st.code("\n".join(cap.records), language="log")
|