"""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 = """
"""
# ── 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'{title}
')
def _note(kind: str, body: str) -> None:
_html(f'{body}
')
def _pill(rating: str) -> str:
return f'{rating.title()}'
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'{sub}
' if sub else ""
cells += f''
_html(f'{cells}
')
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'{hint}' if hint else ""
out += (f'{label}{h}'
f'{_money(value)}
')
return f'{out}
'
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'{label}{disp}
'
cb = detail.get("costBreakdown")
if isinstance(cb, dict) and cb:
parts = " · ".join(f"{k} {_money(v)}" for k, v in cb.items())
rows += (f'Cost breakdown'
f'{_money(sum(cb.values()))}
'
f'{parts}
')
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'{label}'
f'{_money(v)}
')
return f'{rows}
'
# ── 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'{r.sku}{_pill(r.rating.value)}
'
f''
+ (f"ASIN {r.asin} · " if r.asin else "")
+ f"Evaluated at {_money(r.price)}
")
# 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"Actually losing money — {_money(r.actual_profit_per_unit)} per unit. "
"Real COSMOS profit over the last 3 months, including storage, refunds, promo "
"and ads."
+ (f" Lost money in {r.unprofitable_months} of {n_months} observed months."
if r.unprofitable_months else ""))
elif r.unprofitable_months >= 3:
_note("bad", f"Unprofitable in {r.unprofitable_months} of the last {n_months} months "
"on actual profit.")
elif r.is_losing_money:
_note("bad", f"Losing money after ads. 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"Verdict. {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"Price mismatch. 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 Test a specific price 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"Gate. {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('Sellers and prices scraped from the ASIN product page '
'(Apify). Same listing — other merchants competing for the Buy Box — not rival '
'product ASINs.
')
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 overstates profit by {_money(r.unmodeled_cost_gap)} "
"per unit (unmodeled storage, refunds, promo, logistics). Trust the "
"actual figures.")
if r.reasons:
_sec("Why this verdict — every reason")
rows = "".join(f'{i}{reason}
'
for i, reason in enumerate(r.reasons, 1))
_html(f'{rows}
'
'These are the exact deterministic rule outputs that produced '
'the verdict — no AI involved. The AI analysis only explains them.
')
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('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.
')
_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'Estimated monthly take-home {_money(r.monthly_take_home)}, '
f'after {_money(r.storage_charges)} of storage on current stock.
')
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('At current volume — elasticity is not available for this SKU.
')
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('This “take home” excludes ad spend, refunds, promo and '
'logistics — see the profit bridge for real per-unit profit.
')
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('Worst first. Every row states why it loses money and how to fix '
'it. To verify: if profit/unit + ad/unit > 0, ads caused the loss.
')
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 cutting ad spend alone — 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('Read-only — nothing is written anywhere. Open any SKU under '
'Single product for its full evidence-based report.
')
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"{len(losing)} of {len(rows)} SKUs are actually losing money 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'actual {_money(r.actual_profit_per_unit)}/unit '
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'{_pill(r.rating.value)}'
f'{r.sku}'
f'now {_money(r.current_price)}, suggest '
f'{_money(r.suggested_price)}{move} · {econ} · '
f'6-mo {(r.avg_6m or 0):,.0f}/day ({r.trend}) — {r.headline}
')
_html(body)
# ── Page ───────────────────────────────────────────────────────────────
_html(CSS)
_html('')
if not get_settings().openai_api_key:
_note("warn", "No OPENAI_API_KEY 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 single product for the full evidence-based report, scan a "
"product line, or run the catalog money-at-risk scan 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")