initial commit

.gitignore

@@ -0,0 +1,25 @@
+# Python
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+*.egg-info/
+dist/
+build/
+.venv/
+venv/
+.env
+
+# Streamlit
+.streamlit/
+
+# Data (keep repo clean)
+data/**/*.csv
+data/**/*.xlsx
+data/**/*.xls
+data/**/*.parquet
+data/**/*.json
+
+# OS
+.DS_Store
+Thumbs.db

app.py

@@ -0,0 +1,472 @@
+import numpy as np
+import pandas as pd
+import streamlit as st
+import plotly.graph_objects as go
+from pathlib import Path
+
+from pricing_engine.data import generate_synthetic_sku
+from pricing_engine.data_uci import make_sku_week_panel, eligible_skus
+from pricing_engine.core import (
+    estimate_loglog_elasticity,
+    bootstrap_optimal_price,
+    robust_optimal_price,
+    profit_distribution_at_price,
+)
+
+# ============================
+# Config
+# ============================
+RAW_UCI_PATH = Path("data/Online Retail.xlsx")
+PANEL_PARQUET_PATH = Path("data/processed/uci_sku_week.parquet")
+
+DEFAULT_BOOT_SEED = 42
+DEFAULT_N_BOOT = 300
+
+DEFAULT_MAX_MOVE_FRAC = 0.10      # ±10% around median price
+DEFAULT_LEVERAGE_TH = 0.05        # <5% movement => HOLD
+
+# Absolute downside risk cap as a fraction of baseline median profit (prevents ratio blow-ups)
+DEFAULT_ABS_DOWNSIDE_CAP_FRAC = 0.25  # 25% of baseline median profit (tune via advanced)
+
+# ============================
+# Data utilities
+# ============================
+@st.cache_data(show_spinner=False)
+def load_uci_panel() -> pd.DataFrame:
+    if PANEL_PARQUET_PATH.exists():
+        return pd.read_parquet(PANEL_PARQUET_PATH)
+
+    if not RAW_UCI_PATH.exists():
+        raise FileNotFoundError(
+            f"Missing both parquet and raw file.\n"
+            f"Expected parquet: {PANEL_PARQUET_PATH}\n"
+            f"Fallback raw: {RAW_UCI_PATH}"
+        )
+
+    df_raw = pd.read_excel(RAW_UCI_PATH)
+    panel = make_sku_week_panel(df_raw)
+
+    PANEL_PARQUET_PATH.parent.mkdir(parents=True, exist_ok=True)
+    panel.to_parquet(PANEL_PARQUET_PATH, index=False)
+    return panel
+
+
+def standardize_engine_df(df: pd.DataFrame) -> pd.DataFrame:
+    if "price" not in df.columns:
+        raise KeyError("df must contain 'price'.")
+    if "qty" in df.columns:
+        q_col = "qty"
+    elif "demand" in df.columns:
+        q_col = "demand"
+    else:
+        raise KeyError("df must contain 'qty' or 'demand'.")
+    return df[["price", q_col]].copy().rename(columns={q_col: "qty"})
+
+
+def _week_col(panel: pd.DataFrame) -> str:
+    if "week" in panel.columns:
+        return "week"
+    if "Week" in panel.columns:
+        return "Week"
+    raise KeyError("panel must contain 'week' or 'Week'.")
+
+# ============================
+# Decision engine
+# ============================
+def compute_bundle(
+    df_engine: pd.DataFrame,
+    cost: float,
+    n_boot: int,
+    risk_lambda: float,
+    downside_q: float,
+    seed: int = DEFAULT_BOOT_SEED,
+    n_grid: int = 250,
+    max_move_frac: float = DEFAULT_MAX_MOVE_FRAC,
+    leverage_threshold: float = DEFAULT_LEVERAGE_TH,
+    # relative cap
+    max_downside_frac: float = 0.05,
+    # absolute cap (fraction of baseline median profit)
+    abs_downside_cap_frac: float = DEFAULT_ABS_DOWNSIDE_CAP_FRAC,
+) -> dict:
+    # Point estimate exists but will not be shown (bootstrap summary will be shown instead)
+    _a_hat, b_hat = estimate_loglog_elasticity(df_engine)
+
+    boot = bootstrap_optimal_price(df_engine, cost=float(cost), n_boot=int(n_boot), seed=int(seed))
+    params = boot[["intercept", "elasticity"]]
+
+    # price grid around median within allowed move window and observed min/max
+    p_center = float(df_engine["price"].median())
+    p_min = max(float(df_engine["price"].min()), p_center * (1 - max_move_frac))
+    p_max = min(float(df_engine["price"].max()), p_center * (1 + max_move_frac))
+    p_grid = np.linspace(p_min, p_max, int(n_grid)).astype(float)
+
+    def solve(lmbda: float) -> dict:
+        sol = robust_optimal_price(
+            boot_params=params,
+            cost=float(cost),
+            price_grid=p_grid,
+            risk_lambda=float(lmbda),
+            downside_quantile=float(downside_q),
+        )
+        price = float(sol["price"])
+        stats = profit_distribution_at_price(params, float(cost), price, q=float(downside_q))
+        return {"price": price, "stats": stats}
+
+    naive = solve(0.0)
+    rob = naive if risk_lambda == 0.0 else solve(risk_lambda)
+
+    # profit bands across grid (for proof plot)
+    A = np.exp(params["intercept"].values)
+    beta = params["elasticity"].values
+    profit_mat = (p_grid[None, :] - float(cost)) * (A[:, None] * (p_grid[None, :] ** beta[:, None]))
+    med_profit = np.median(profit_mat, axis=0)
+    q_low = np.quantile(profit_mat, float(downside_q), axis=0)
+    q_high = np.quantile(profit_mat, 1.0 - float(downside_q), axis=0)
+
+    # Feasibility: existence of at least one price with positive median AND positive downside
+    feasible_mask = (med_profit > 0) & (q_low > 0)
+    any_feasible = bool(np.any(feasible_mask))
+
+    # Leverage: how much median profit moves across the grid (scale-safe)
+    profit_range = float(np.max(med_profit) - np.min(med_profit))
+    profit_scale = float(max(np.max(np.abs(med_profit)), 1e-9))
+    leverage_frac = profit_range / profit_scale
+
+    # Baseline = status quo (median observed price)
+    baseline_price = p_center
+    baseline_stats = profit_distribution_at_price(params, float(cost), float(baseline_price), q=float(downside_q))
+    baseline_med = float(baseline_stats["median_profit"])
+
+    # Downside definitions: relative + absolute
+    downside_risk = float(rob["stats"]["downside_risk"])
+    # Relative downside normalized by baseline (not recommended) to avoid blow-ups near 0
+    downside_fraction = float(downside_risk / max(abs(baseline_med), 1e-9))
+    # Absolute downside cap tied to baseline median profit magnitude (>=0)
+    abs_downside_cap = float(abs_downside_cap_frac * max(abs(baseline_med), 0.0))
+
+    # Decide
+    if not any_feasible:
+        decision = "NO-GO"
+        rec_price, rec_stats = rob["price"], rob["stats"]
+        rationale = "No feasible price yields positive median and positive downside profit (q-down) under uncertainty. Do not deploy."
+        tone = "error"
+
+    elif risk_lambda == 0.0:
+        decision = "OPTIMIZE"
+        rec_price, rec_stats = naive["price"], naive["stats"]
+        rationale = "Risk-neutral optimization. Deploy profit-maximizing price."
+        tone = "success"
+
+    elif leverage_frac < leverage_threshold:
+        decision = "HOLD"
+        rec_price, rec_stats = baseline_price, baseline_stats
+        rationale = "Profit is positive but not sensitive to price. Deploy baseline (status quo)."
+        tone = "warning"
+
+    elif (downside_fraction > max_downside_frac) or (downside_risk > abs_downside_cap):
+        decision = "HOLD"
+        rec_price, rec_stats = baseline_price, baseline_stats
+        rationale = "Downside variability is too high for a price change (relative/absolute cap). Deploy baseline (status quo)."
+        tone = "warning"
+
+    else:
+        decision = "OPTIMIZE"
+        rec_price, rec_stats = rob["price"], rob["stats"]
+        rationale = "Price materially affects outcomes with acceptable downside risk. Deploy robust price."
+        tone = "success"
+
+    rec_med = float(rec_stats["median_profit"])
+    naive_med = float(naive["stats"]["median_profit"])
+    med_delta_pct = 100.0 * (rec_med - naive_med) / max(abs(naive_med), 1e-9)
+
+    # Bootstrap elasticity summary (use this in UI)
+    beta_med = float(np.median(beta))
+    beta_p10, beta_p90 = (float(x) for x in np.quantile(beta, [0.10, 0.90]))
+
+    return {
+        "b_hat": float(b_hat),  # kept for reference; UI should prefer bootstrap summary
+        "beta_boot": {"median": beta_med, "p10": beta_p10, "p90": beta_p90},
+        "grid": p_grid,
+        "bands": {"median": med_profit, "q_low": q_low, "q_high": q_high},
+        "rob": rob,
+        "naive": naive,
+        "baseline": {"price": float(baseline_price), "stats": baseline_stats},
+        "recommended": {"price": float(rec_price), "stats": rec_stats},
+        "kpis": {
+            "decision": decision,
+            "tone": tone,
+            "rationale": rationale,
+            "downside_risk": float(downside_risk),
+            "downside_fraction": float(downside_fraction),
+            "max_downside_frac": float(max_downside_frac),
+            "abs_downside_cap": float(abs_downside_cap),
+            "abs_downside_cap_frac": float(abs_downside_cap_frac),
+            "leverage_frac": float(leverage_frac),
+            "median_profit": float(rec_stats["median_profit"]),
+            "q_down_profit": float(rec_stats["q_down_profit"]),
+            "median_delta_pct": float(med_delta_pct),
+        },
+    }
+
+
+def render_plot(bundle: dict, downside_q: float) -> go.Figure:
+    p = bundle["grid"]
+    b = bundle["bands"]
+    rec = bundle["recommended"]["price"]
+    naive = bundle["naive"]["price"]
+
+    fig = go.Figure()
+    fig.add_trace(go.Scatter(x=p, y=b["median"], mode="lines", name="Median profit"))
+    fig.add_trace(go.Scatter(x=p, y=b["q_low"], mode="lines", name=f"Downside (q{int(downside_q*100)})"))
+    fig.add_trace(go.Scatter(x=p, y=b["q_high"], mode="lines", name=f"Upside (q{int((1-downside_q)*100)})"))
+
+    fig.add_vline(x=rec, line=dict(color="green", width=3))
+    fig.add_annotation(
+        x=rec,
+        y=float(np.max(b["median"])),
+        text=f"Deploy €{rec:.2f}",
+        showarrow=True,
+        arrowhead=2,
+        ax=40,
+        ay=-40,
+        font=dict(color="green"),
+    )
+
+    fig.add_vline(x=naive, line=dict(color="gray", width=2, dash="dash"))
+    fig.add_annotation(
+        x=naive,
+        y=float(np.min(b["median"])),
+        text=f"Naïve-opt €{naive:.2f}",
+        showarrow=True,
+        arrowhead=2,
+        ax=-40,
+        ay=40,
+        font=dict(color="gray"),
+    )
+
+    fig.update_layout(
+        xaxis_title="Price",
+        yaxis_title="Profit",
+        hovermode="x unified",
+        margin=dict(l=10, r=10, t=10, b=10),
+    )
+    return fig
+
+
+# ============================
+# UI
+# ============================
+st.set_page_config(page_title="Pricing Decision — Robust Price", layout="wide")
+st.title("Pricing Decision")
+st.caption("One output: deploy price. One proof: profit vs price under uncertainty.")
+
+# Attempt to load UCI panel (optional)
+panel_uci = None
+try:
+    panel_uci = load_uci_panel()
+except Exception:
+    panel_uci = None
+
+# If parquet isn't available, hide UCI mode entirely (public-repo safe)
+st.sidebar.header("1) Choose data")
+dataset_choices = ["Synthetic"] if panel_uci is None else ["Synthetic", "UCI Online Retail"]
+dataset_mode = st.sidebar.radio("Dataset", dataset_choices, index=0)
+
+st.sidebar.header("2) Economics")
+if dataset_mode == "Synthetic":
+    cost = st.sidebar.slider("Unit cost", 0.0, 20.0, 4.0, 0.1)
+else:
+    cost_frac = st.sidebar.slider("Cost as % of current median price", 0.30, 0.80, 0.50, 0.05)
+
+st.sidebar.header("3) Risk appetite")
+risk_profile = st.sidebar.radio("Risk appetite", ["Risk-neutral", "Risk-averse"], index=0)
+
+profile_map = {
+    "Risk-neutral": {"lambda": 0.0, "q": 0.20, "max_downside_frac": 0.20, "abs_cap_frac": 0.50},
+    "Risk-averse":  {"lambda": 1.5, "q": 0.05, "max_downside_frac": 0.03, "abs_cap_frac": 0.20},
+}
+risk_lambda = float(profile_map[risk_profile]["lambda"])
+downside_q = float(profile_map[risk_profile]["q"])
+max_downside_frac = float(profile_map[risk_profile]["max_downside_frac"])
+abs_cap_frac = float(profile_map[risk_profile]["abs_cap_frac"])
+
+show_advanced = st.sidebar.checkbox("Show advanced controls", value=False)
+if show_advanced:
+    n_boot = st.sidebar.slider("Bootstrap draws", 100, 800, DEFAULT_N_BOOT, 50)
+    max_move_frac = st.sidebar.slider("Allowed price move (±)", 0.05, 0.25, DEFAULT_MAX_MOVE_FRAC, 0.01)
+    leverage_th = st.sidebar.slider("Leverage threshold", 0.01, 0.15, DEFAULT_LEVERAGE_TH, 0.01)
+    abs_cap_frac = st.sidebar.slider("Abs downside cap (% of baseline median)", 0.05, 1.00, abs_cap_frac, 0.05)
+else:
+    n_boot = DEFAULT_N_BOOT
+    max_move_frac = DEFAULT_MAX_MOVE_FRAC
+    leverage_th = DEFAULT_LEVERAGE_TH
+
+# Build df_engine
+if dataset_mode == "Synthetic":
+    st.sidebar.header("4) Synthetic (optional)")
+    seed = int(st.sidebar.number_input("Seed", min_value=0, value=42, step=1))
+    noise_std = st.sidebar.slider("Demand noise", 0.01, 0.40, 0.15, 0.01)
+    true_elasticity = st.sidebar.slider("True elasticity", -3.0, -0.2, -1.5, 0.1)
+
+    df_raw = generate_synthetic_sku(elasticity=true_elasticity, noise_std=noise_std, seed=seed)
+    df_engine = standardize_engine_df(df_raw)
+    source_label = "Synthetic"
+    time_unit = "per period (synthetic)"
+else:
+    st.sidebar.header("4) Pick SKU")
+    min_weeks = st.sidebar.slider("Min weeks", 10, 52, 26, 1)
+    min_price_points = st.sidebar.slider("Min distinct prices", 3, 20, 8, 1)
+    min_total_qty = st.sidebar.slider("Min total qty", 50, 2000, 200, 50)
+
+    skus = eligible_skus(panel_uci, min_weeks=min_weeks, min_price_points=min_price_points, min_total_qty=min_total_qty)
+    if not skus:
+        st.error("No SKUs pass governance thresholds. Lower thresholds or use Synthetic.")
+        st.stop()
+
+    sku_selected = st.sidebar.selectbox("SKU (StockCode)", skus, index=0)
+    wk = _week_col(panel_uci)
+    sku_panel = panel_uci.loc[panel_uci["StockCode"] == sku_selected].sort_values(wk)
+
+    df_engine = standardize_engine_df(sku_panel[["avg_price", "qty"]].rename(columns={"avg_price": "price"}))
+    median_price = float(df_engine["price"].median())
+    cost = float(cost_frac * median_price)
+
+    seed = DEFAULT_BOOT_SEED
+    source_label = f"UCI — {sku_selected}"
+    time_unit = "per week"
+
+# ============================
+# Risk sensitivity check (compute both profiles every run)
+# ============================
+bundle_rn = compute_bundle(
+    df_engine=df_engine,
+    cost=float(cost),
+    n_boot=int(n_boot),
+    risk_lambda=float(profile_map["Risk-neutral"]["lambda"]),
+    downside_q=float(profile_map["Risk-neutral"]["q"]),
+    seed=int(seed),
+    max_move_frac=float(max_move_frac),
+    leverage_threshold=float(leverage_th),
+    max_downside_frac=float(profile_map["Risk-neutral"]["max_downside_frac"]),
+    abs_downside_cap_frac=float(profile_map["Risk-neutral"]["abs_cap_frac"]),
+)
+
+bundle_ra = compute_bundle(
+    df_engine=df_engine,
+    cost=float(cost),
+    n_boot=int(n_boot),
+    risk_lambda=float(profile_map["Risk-averse"]["lambda"]),
+    downside_q=float(profile_map["Risk-averse"]["q"]),
+    seed=int(seed),
+    max_move_frac=float(max_move_frac),
+    leverage_threshold=float(leverage_th),
+    max_downside_frac=float(profile_map["Risk-averse"]["max_downside_frac"]),
+    abs_downside_cap_frac=float(profile_map["Risk-averse"]["abs_cap_frac"]),
+)
+
+PRICE_EPS = 1e-6
+same_price = abs(bundle_rn["recommended"]["price"] - bundle_ra["recommended"]["price"]) <= PRICE_EPS
+same_decision = bundle_rn["kpis"]["decision"] == bundle_ra["kpis"]["decision"]
+risk_sensitive = (not same_price) or (not same_decision)
+
+bundle = bundle_rn if risk_profile == "Risk-neutral" else bundle_ra
+
+k = bundle["kpis"]
+rec_price = bundle["recommended"]["price"]
+naive_price = bundle["naive"]["price"]
+
+# Risk message
+if risk_sensitive:
+    st.warning(
+        f"Risk appetite changes the recommendation: "
+        f"Risk-neutral {bundle_rn['kpis']['decision']} @ €{bundle_rn['recommended']['price']:.2f} | "
+        f"Risk-averse {bundle_ra['kpis']['decision']} @ €{bundle_ra['recommended']['price']:.2f}"
+    )
+else:
+    st.info("Risk appetite does not change the recommendation for this dataset/SKU.")
+
+# Time-unit clarification (pre-empts “model units” criticism)
+st.caption(f"Source: {source_label} | Risk profile: {risk_profile} | Profit shown: **{time_unit}**")
+
+# Decision banner
+left, right = st.columns([2, 1], vertical_alignment="center")
+with left:
+    st.markdown(f"### Deploy **€{rec_price:.2f}**")
+    st.write(k["rationale"])
+with right:
+    st.metric("Decision", k["decision"])
+
+# KPI tiles
+beta_boot = bundle["beta_boot"]
+c1, c2, c3, c4 = st.columns(4)
+c1.metric("Elasticity (bootstrap median)", f"{beta_boot['median']:.3f}")
+c2.metric("Median profit", f"{k['median_profit']:.2f}")
+c3.metric(f"Downside profit (q{int(downside_q*100)})", f"{k['q_down_profit']:.2f}")
+c4.metric("Median vs Naïve-opt", f"{k['median_delta_pct']:+.1f}%")
+
+st.caption(f"Elasticity uncertainty: p10={beta_boot['p10']:.3f}, p90={beta_boot['p90']:.3f}")
+
+# HOLD explanation
+if k["decision"] == "HOLD":
+    if k["leverage_frac"] < leverage_th:
+        st.caption(
+            f"HOLD triggered: low leverage {k['leverage_frac']*100:.1f}% "
+            f"< threshold {leverage_th*100:.1f}% (price barely changes profit)."
+        )
+    else:
+        st.caption(
+            f"HOLD triggered: downside risk too high. "
+            f"Relative={k['downside_fraction']*100:.1f}% (cap {k['max_downside_frac']*100:.1f}%), "
+            f"Absolute={k['downside_risk']:.2f} (cap {k['abs_downside_cap']:.2f})."
+        )
+
+# Tone message
+if k["tone"] == "success":
+    st.success(f"DECISION: {k['decision']} — deploy €{rec_price:.2f}")
+elif k["tone"] == "warning":
+    st.warning(f"DECISION: {k['decision']} — deploy baseline €{rec_price:.2f}")
+else:
+    st.error(f"DECISION: {k['decision']} — no deployment recommended")
+
+# Proof chart
+st.subheader("Proof: profit under uncertainty across feasible prices")
+st.caption(
+    "This is an observational pricing decision demo (not a causal price elasticity estimate)."
+)
+
+fig = render_plot(bundle, downside_q=downside_q)
+st.plotly_chart(fig, use_container_width=True)
+
+# Decision card
+baseline_price = bundle["baseline"]["price"]
+baseline_med = bundle["baseline"]["stats"]["median_profit"]
+baseline_q = bundle["baseline"]["stats"]["q_down_profit"]
+
+st.subheader("Decision Card (copy/paste)")
+st.code(
+    f"""Decision: Deploy price €{rec_price:.2f}
+
+Source: {source_label}
+Risk profile: {risk_profile}
+Profit unit: {time_unit}
+
+Why:
+- Elasticity (bootstrap median): {beta_boot['median']:.3f} (p10 {beta_boot['p10']:.3f}, p90 {beta_boot['p90']:.3f})
+- Median profit (recommended): {k['median_profit']:.2f}
+- Downside profit (q{int(downside_q*100)}): {k['q_down_profit']:.2f}
+- Downside risk (absolute): {k['downside_risk']:.2f} (cap {k['abs_downside_cap']:.2f} = {k['abs_downside_cap_frac']*100:.0f}% of baseline median)
+- Downside fraction (vs baseline median): {k['downside_fraction']*100:.1f}% (cap {k['max_downside_frac']*100:.1f}%)
+- Leverage: {k['leverage_frac']*100:.1f}%
+
+Baseline (current median price): €{baseline_price:.2f}
+- Baseline median profit: {baseline_med:.2f}
+- Baseline downside profit (q{int(downside_q*100)}): {baseline_q:.2f}
+
+Naïve-opt (risk-neutral optimizer): €{naive_price:.2f}
+Rationale: {k['rationale']}
+""",
+    language="text",
+)
+
+with st.expander("Data preview"):
+    st.dataframe(df_engine.head(30), use_container_width=True)

data/processed/uci_sku_week.parquet

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:9f0258efb9e8373fae8f237f1902854aa8e4ed02e63b57a7c5c8024672868b25
+size 908235

docs/Appendix.md

@@ -0,0 +1,63 @@
+# **Appendix — Methodological Notes**
+
+---
+
+## **Why Elasticity Is Observational**
+
+Retail prices are not randomized.
+Observed price–quantity relationships reflect correlation, not causal response.
+
+This system does not attempt causal identification.
+It focuses on robust decision-making given observed behavior.
+
+---
+
+## **Why Bootstrap Is Used**
+
+Closed-form uncertainty assumptions are fragile in pricing contexts.
+
+Bootstrap resampling:
+
+* captures parameter uncertainty
+* avoids distributional assumptions
+* supports downside-aware evaluation
+
+---
+
+## **Why No Machine Learning Models Are Used**
+
+The pricing decision is low-dimensional.
+
+Additional model complexity:
+
+* increases opacity
+* complicates governance
+* does not improve decision quality at this stage
+
+ML pricing belongs to later integration phases.
+
+---
+
+## **Out-of-Scope Extensions**
+
+The following are intentionally excluded:
+
+* causal pricing experiments
+* promotion-response modeling
+* multi-SKU or portfolio pricing
+* inventory-constrained pricing
+* dynamic or reinforcement learning pricing
+
+These extensions require additional data and governance structures.
+
+---
+
+## **Closing Note**
+
+The system is designed to answer one question well:
+
+> **What price can be deployed with confidence under uncertainty?**
+
+Everything else is deliberately deferred.
+
+---

docs/Executive_Brief.md

@@ -0,0 +1,75 @@
+# **Executive Brief — Robust Pricing Decisions Under Uncertainty**
+
+---
+
+## **Why This Matters**
+
+Pricing decisions are often made using point estimates of demand or profit.
+While convenient, this approach ignores a critical reality:
+
+* profit outcomes are uncertain
+* downside risk is asymmetric
+* naïve optimization frequently selects fragile prices
+
+When a deployed price performs poorly, teams are forced into:
+
+* reactive discounting
+* margin erosion
+* post-hoc justification
+* loss of confidence in pricing governance
+
+This system was designed to answer a single operational question:
+
+> **What price should be deployed when profit is uncertain and downside risk matters?**
+
+---
+
+## **What the System Evaluates**
+
+For each feasible price, the system evaluates:
+
+* **Median profit** — expected outcome
+* **Downside profit** — exposure to adverse scenarios
+* **Governance thresholds** — leverage and risk limits
+
+These elements are combined into a **deploy / hold / reject** decision framework.
+
+---
+
+## **The Resulting Decisions**
+
+The system outputs one of three outcomes:
+
+| Decision | Interpretation |
+|-------|----------------|
+| **OPTIMIZE** | Deploy the recommended price |
+| **HOLD** | Maintain the current (baseline) price |
+| **NO-GO** | No price change is viable under uncertainty |
+
+Each decision is accompanied by a **justification card** suitable for approval or review.
+
+---
+
+## **What This Is**
+
+* A pricing **decision-support system**
+* Designed for **risk-aware governance**
+* Built to produce **defensible deployment choices**
+
+---
+
+## **What This Is Not**
+
+* Not a causal elasticity estimator
+* Not a promotion optimization engine
+* Not an experimentation framework
+
+---
+
+## **Leadership Takeaway**
+
+**Robust pricing decisions outperform naïve profit maximization when uncertainty is material.**
+
+By explicitly incorporating downside risk and governance thresholds, this system produces prices that are not only profitable, but **operationally defensible**.
+
+---

docs/Technical_Brief.md

@@ -0,0 +1,96 @@
+# **Pricing Decision — Technical Summary**
+
+---
+
+## **1. Purpose**
+
+The difficulty of pricing under uncertainty is not estimating demand,
+but deciding **which price can be safely deployed**.
+
+Prices that maximize expected profit often expose unacceptable downside risk,
+leading to reversals, overrides, and erosion of trust in pricing decisions.
+
+This system addresses the technical question:
+
+> **How should prices be selected when profit distributions—not point estimates—matter?**
+
+---
+
+## **2. Data Basis**
+
+Two operating modes are supported:
+
+### **Synthetic Mode**
+* controlled elasticity parameter
+* additive demand noise
+* known cost structure
+* fully reproducible
+
+Used to demonstrate idealized pricing behavior.
+
+### **Observational Retail Mode (UCI, local only)**
+* transactional retail data
+* time-varying prices
+* non-randomized price changes
+* aggregated per period
+
+Elasticity estimates in this mode are **observational, not causal**.
+
+---
+
+## **3. Model Structure**
+
+Demand is modeled using a log–log specification.
+Parameter uncertainty is captured via bootstrap resampling.
+
+The objective is **distributional robustness**, not causal identification.
+
+---
+
+## **4. Profit Evaluation**
+
+For each candidate price:
+
+* profit distributions are computed
+* median profit represents expected outcome
+* downside quantiles (q10 or q5) represent risk exposure
+
+All profit values are expressed **per aggregation period**.
+
+---
+
+## **5. Decision Logic**
+
+Candidate prices are evaluated within a constrained grid around the current median price.
+
+Decisions follow explicit governance rules:
+
+* **Feasibility:** at least one price must yield positive median and downside profit
+* **Leverage:** price must materially affect profit
+* **Risk:** downside exposure must remain within relative and absolute caps
+
+Violations trigger HOLD or NO-GO outcomes.
+
+---
+
+## **6. Output**
+
+The system produces:
+
+* decision state (OPTIMIZE / HOLD / NO-GO)
+* recommended deploy price
+* profit distribution diagnostics
+* traceable justification metrics
+
+This design prioritizes **auditability, explainability, and governance** over model complexity.
+
+---
+
+## **Closing Position**
+
+Pricing under uncertainty is a decision problem, not a curve-fitting exercise.
+
+This system converts uncertain demand response into **deployable pricing actions**
+that remain defensible under review, volatility, and downside exposure.
+
+---

pricing_engine/core.py

@@ -0,0 +1,348 @@
+"""
+Pricing Decision Core — Robust Optimization under Elasticity Uncertainty
+
+Purpose:
+Select a price that maximizes risk-adjusted profit by propagating
+uncertainty in demand elasticity into profit distributions and penalizing
+downside fragility.
+
+Core Assumptions:
+- Demand follows a power-law response to price: q = A * p^beta
+- Elasticity uncertainty is captured via bootstrap resampling
+- Decisions are evaluated using median profit and downside risk
+
+What this module DOES:
+- Estimates elasticity
+- Propagates uncertainty to profit
+- Selects robust prices
+
+What this module DOES NOT do:
+- Forecast demand over time
+- Handle multiple SKUs
+- Perform MLOps or deployment
+"""
+
+
+import numpy as np
+import pandas as pd
+from typing import Tuple
+
+
+def _qty_col(df: pd.DataFrame) -> str:
+    if "qty" in df.columns:
+        return "qty"
+    if "demand" in df.columns:
+        return "demand"
+    raise KeyError("Input df must contain 'qty' or 'demand' column.")
+
+
+def estimate_loglog_elasticity(df: pd.DataFrame) -> Tuple[float, float]:
+    """
+    log(q) = a + b*log(p)
+    Returns: (a, b) where a=log(A), b=elasticity
+    """
+    data = df.copy()
+    q_col = _qty_col(data)
+
+    data = data[(data["price"] > 0) & (data[q_col] > 0)].copy()
+    if len(data) < 3:
+        raise ValueError("Need at least 3 valid observations to fit elasticity.")
+
+    X = np.log(data["price"].astype(float).values)
+    y = np.log(data[q_col].astype(float).values)
+
+    X_mat = np.column_stack([np.ones(len(X)), X])
+
+    # Stable OLS: least squares instead of explicit inverse
+    beta_hat, *_ = np.linalg.lstsq(X_mat, y, rcond=None)
+
+    return float(beta_hat[0]), float(beta_hat[1])
+
+
+
+def profit_curve(
+    prices: np.ndarray,
+    intercept: float,
+    elasticity: float,
+    cost: float,
+) -> pd.DataFrame:
+    """
+    Compute demand and profit for a grid of prices.
+    """
+    # Recover A from intercept: intercept = log(A)
+    A = np.exp(intercept)
+
+    demand = A * (prices ** elasticity)
+    profit = (prices - cost) * demand
+
+    return pd.DataFrame(
+        {
+            "price": prices,
+            "demand": demand,
+            "profit": profit,
+        }
+    )
+
+
+
+def optimal_price(
+    curve: pd.DataFrame,
+) -> dict:
+    """
+    Select price that maximizes profit.
+    """
+    idx = curve["profit"].idxmax()
+    row = curve.loc[idx]
+
+    return {
+        "price": float(row["price"]),
+        "profit": float(row["profit"]),
+        "demand": float(row["demand"]),
+    }
+
+
+def bootstrap_optimal_price(
+    df: pd.DataFrame,
+    cost: float,
+    n_boot: int = 200,
+    n_grid: int = 200,
+    seed: int = 42,
+) -> pd.DataFrame:
+    """
+    Bootstrap uncertainty over elasticity by resampling rows (time periods) with replacement.
+    Returns a table of bootstrap draws with (intercept, elasticity, opt_price, opt_profit).
+    """
+    rng = np.random.default_rng(seed)
+
+    # Defensive cleaning for log transforms
+    data = df[(df["price"] > 0) & (df[_qty_col(df)] > 0)].copy()
+    q_col = _qty_col(data)
+    data = data.rename(columns={q_col: "qty"})
+
+    n = len(data)
+    if n < 10:
+        raise ValueError("Need at least 10 observations for bootstrap stability.")
+
+    # Keep optimization inside observed price range (no extrapolation)
+    p_min, p_max = float(data["price"].min()), float(data["price"].max())
+    price_grid = np.linspace(p_min, p_max, n_grid)
+
+    rows = []
+    for _ in range(n_boot):
+        # sample indices with replacement
+        idx = rng.integers(0, n, size=n)
+        sample = data.iloc[idx]
+
+        a_hat, b_hat = estimate_loglog_elasticity(sample)
+
+        curve = profit_curve(price_grid, a_hat, b_hat, cost)
+        dec = optimal_price(curve)
+
+        rows.append(
+            {
+                "intercept": a_hat,
+                "elasticity": b_hat,
+                "opt_price": dec["price"],
+                "opt_profit": dec["profit"],
+                "opt_demand": dec["demand"],
+            }
+        )
+
+    return pd.DataFrame(rows)
+
+
+def decision_stability_summary(boot: pd.DataFrame) -> dict:
+    """
+    Summarize stability of the optimal price decision.
+    """
+    q10, q50, q90 = boot["opt_price"].quantile([0.1, 0.5, 0.9]).tolist()
+    spread = q90 - q10
+
+    return {
+        "opt_price_median": float(q50),
+        "opt_price_q10": float(q10),
+        "opt_price_q90": float(q90),
+        "opt_price_spread_q90_q10": float(spread),
+        "elasticity_median": float(boot["elasticity"].median()),
+        "elasticity_q10": float(boot["elasticity"].quantile(0.1)),
+        "elasticity_q90": float(boot["elasticity"].quantile(0.9)),
+    }
+
+
+def stability_flag(summary: dict, max_spread_frac: float = 0.15) -> dict:
+    """
+    Flag whether decision is stable: opt price spread <= max_spread_frac * median price.
+    """
+    denom = max(summary["opt_price_median"], 1e-9)
+    frac = summary["opt_price_spread_q90_q10"] / denom
+    return {
+        "stable": bool(frac <= max_spread_frac),
+        "spread_fraction_of_median": float(frac),
+        "threshold": float(max_spread_frac),
+    }
+
+
+def robust_optimal_price(
+    boot_params: pd.DataFrame,
+    cost: float,
+    price_grid: np.ndarray,
+    risk_lambda: float = 0.5,
+    downside_quantile: float = 0.1,
+) -> dict:
+    """
+    Robust price selection: maximize median(profit) - lambda * downside_risk(profit)
+
+    downside_risk(profit) = median(profit) - q_downside(profit)
+    where q_downside is e.g. 10th percentile across bootstrap draws.
+
+    Inputs:
+      boot_params: DataFrame with columns ["intercept", "elasticity"] from bootstrap
+      cost: unit cost
+      price_grid: candidate prices to evaluate
+      risk_lambda: penalty weight (0 = median-only, higher = more conservative)
+      downside_quantile: e.g. 0.1 for q10
+
+    Returns dict with chosen price and diagnostics.
+    """
+    if boot_params.empty:
+        raise ValueError("boot_params is empty.")
+
+    A = np.exp(boot_params["intercept"].values)          # shape (B,)
+    beta = boot_params["elasticity"].values              # shape (B,)
+    prices = price_grid.astype(float)                    # shape (P,)
+
+    # Profit across draws for each price:
+    # demand_{b,p} = A_b * p^beta_b
+    # profit_{b,p} = (p - cost) * demand_{b,p}
+    # We build profit matrix shape (B, P)
+    demand = A[:, None] * (prices[None, :] ** beta[:, None])
+    profit = (prices[None, :] - float(cost)) * demand
+
+    med = np.median(profit, axis=0)  # shape (P,)
+    q_down = np.quantile(profit, downside_quantile, axis=0)
+    downside_risk = med - q_down
+
+    score = med - risk_lambda * downside_risk
+
+    best_idx = int(np.argmax(score))
+    p_star = float(prices[best_idx])
+
+    return {
+        "price": p_star,
+        "score": float(score[best_idx]),
+        "median_profit": float(med[best_idx]),
+        "q_down_profit": float(q_down[best_idx]),
+        "downside_risk": float(downside_risk[best_idx]),
+        "risk_lambda": float(risk_lambda),
+        "downside_quantile": float(downside_quantile),
+    }
+
+
+def profit_distribution_at_price(
+    boot_params: pd.DataFrame,
+    cost: float,
+    price: float,
+    q: float = 0.1,
+) -> dict:
+    A = np.exp(boot_params["intercept"].values)
+    beta = boot_params["elasticity"].values
+    p = float(price)
+
+    profit = (p - float(cost)) * (A * (p ** beta))
+
+    med = float(np.median(profit))
+    q_down = float(np.quantile(profit, q))
+    q_up = float(np.quantile(profit, 1 - q))
+
+    return {
+        "price": p,
+        "median_profit": med,
+        "q_down_profit": q_down,
+        "q_up_profit": q_up,
+        "downside_risk": med - q_down,
+        "upside_spread": q_up - med,
+    }
+
+
+def decision_justification_card(
+    robust_stats: dict,
+    naive_stats: dict,
+    decision_status: dict,
+) -> dict:
+    denom_profit = max(abs(float(naive_stats["median_profit"])), 1e-9)
+    denom_risk = max(abs(float(naive_stats["downside_risk"])), 1e-9)
+
+    med_delta_pct = (robust_stats["median_profit"] - naive_stats["median_profit"]) / denom_profit * 100.0
+    downside_improvement_pct = (naive_stats["downside_risk"] - robust_stats["downside_risk"]) / denom_risk * 100.0
+
+    rationale = (
+        "The selected price sacrifices negligible median profit to materially reduce downside risk "
+        "across plausible demand elasticities, producing a more stable and defensible pricing decision under uncertainty."
+        if decision_status["status"] == "ROBUST"
+        else
+        "The price decision shows excessive downside variability relative to expected payoff and should not be deployed "
+        "without further constraints or additional data."
+    )
+
+    return {
+        "recommended_price": round(float(robust_stats["price"]), 2),
+        "naive_price": round(float(naive_stats["price"]), 2),
+        "median_profit_delta_pct": round(float(med_delta_pct), 2),
+        "downside_risk_improvement_pct": round(float(downside_improvement_pct), 2),
+        "decision_status": decision_status["status"],
+        "rationale": rationale,
+    }
+
+
+
+
+def decision_status(
+    stats: dict,
+    max_downside_frac: float = 0.05,
+) -> dict:
+    frac = stats["downside_risk"] / max(stats["median_profit"], 1e-9)
+
+    return {
+        "status": "ROBUST" if frac <= max_downside_frac else "FRAGILE",
+        "downside_fraction": round(frac, 3),
+        "threshold": max_downside_frac,
+    }
+
+
+def sensitivity_table_at_price(
+    boot_params: pd.DataFrame,
+    base_cost: float,
+    price: float,
+    q: float = 0.1,
+    elasticity_scales: Tuple[float, ...] = (0.9, 1.0, 1.1),
+    cost_scales: Tuple[float, ...] = (0.9, 1.0, 1.1),
+) -> pd.DataFrame:
+    """
+    Returns median profit (and downside) at a fixed price under perturbations:
+    - scale elasticity draws by factors
+    - scale cost by factors
+    """
+    A = np.exp(boot_params["intercept"].values)
+    beta0 = boot_params["elasticity"].values
+    p = float(price)
+
+    rows = []
+    for e_scale in elasticity_scales:
+        beta = beta0 * float(e_scale)
+        demand = A * (p ** beta)
+
+        for c_scale in cost_scales:
+            c = float(base_cost) * float(c_scale)
+            profit = (p - c) * demand
+            med = float(np.median(profit))
+            q_down = float(np.quantile(profit, q))
+            rows.append(
+                {
+                    "elasticity_scale": float(e_scale),
+                    "cost_scale": float(c_scale),
+                    "median_profit": med,
+                    f"q{int(q*100)}_profit": q_down,
+                    "downside_risk": med - q_down,
+                }
+            )
+    return pd.DataFrame(rows)

pricing_engine/data.py

@@ -0,0 +1,35 @@
+import numpy as np
+import pandas as pd
+
+
+def generate_synthetic_sku(
+    n_periods: int = 52,
+    base_price: float = 10.0,
+    base_demand: float = 120.0,
+    elasticity: float = -1.5,
+    noise_std: float = 0.1,
+    seed: int = 42,
+) -> pd.DataFrame:
+    """
+    Generate a single-SKU synthetic price–demand time series.
+    Elasticity is constant and known (ground truth).
+    """
+    rng = np.random.default_rng(seed)
+
+    # price variation around base price
+    prices = base_price * (1 + rng.normal(0, 0.15, size=n_periods))
+    prices = np.clip(prices, base_price * 0.6, base_price * 1.4)
+
+    # demand model: q = A * p^elasticity * noise
+    noise = np.exp(rng.normal(0, noise_std, size=n_periods))
+    demand = base_demand * (prices ** elasticity) * noise
+
+    df = pd.DataFrame(
+        {
+            "t": np.arange(n_periods),
+            "price": prices,
+            "qty": demand,
+        }
+    )
+
+    return df

pricing_engine/data_uci.py

@@ -0,0 +1,70 @@
+import pandas as pd 
+import numpy as np 
+
+def make_sku_week_panel(df: pd.DataFrame) -> pd.DataFrame:
+    '''
+    Process raw UCI Online Retail data into SKU-week panel format.
+    
+    :param df: Input raw data
+    :type df: pd.DataFrame  
+    :return: Processed SKU-week panel data
+    :rtype: pd.DataFrame
+    '''
+    # "Normalise"
+    df_panel = df.copy()
+
+    df['InvoiceDate'] = pd.to_datetime(df['InvoiceDate'], errors='coerce')
+    df['StockCode'] = df['StockCode'].astype(str)
+
+    # Hard governance filters
+    df = df[~df['InvoiceNo'].str.startswith('C', na=False)]  # Remove cancellations
+    df = df[df['Quantity'] > 0]  # Keep only positive quantities
+    df = df[df['UnitPrice'] > 0]  # Keep only positive prices
+    df = df.dropna(subset=['InvoiceDate', 'StockCode', 'Quantity', 'UnitPrice'])  # Remove rows with missing values
+
+    # Select one country for simplicity
+    if 'Country' in df.columns:
+        df = df[df['Country'] == 'United Kingdom']
+    
+    # Create 'Week' column
+    df['Week'] = df['InvoiceDate'].dt.to_period('W').apply(lambda r: r.start_time)
+
+    # Aggregate to SKU-Week level
+    df_panel = (df.groupby(['StockCode', 'Week']).apply(
+        lambda x: pd.Series({
+            'qty': x['Quantity'].sum(),
+            'price': x['UnitPrice'].mean(),
+            'avg_price': np.average(x['UnitPrice'], weights=x['Quantity']),
+            'n_txn': len(x),
+        })
+    ).reset_index())
+
+    return df_panel
+
+
+def eligible_skus(df_panel: pd.DataFrame, min_weeks: int = 26, min_price_points: int = 10, min_total_qty: int = 200) -> list[str]:
+    '''
+    Identify SKUs eligible for analysis based on data sufficiency criteria.
+
+    :param df_panel: Processed SKU-week panel data  
+    :type df_panel: pd.DataFrame
+    :param min_weeks: Minimum number of weeks of data required
+    :type min_weeks: int
+    :param min_price_points: Minimum number of distinct price points required
+    :type min_price_points: int
+    :param min_total_qty: Minimum total quantity required
+    :type min_total_qty: int
+    :return: List of eligible SKU codes
+    :rtype: list[str]
+    '''
+    sku_stats = df_panel.groupby('StockCode').agg(
+        n_weeks=('Week', 'nunique'),
+        n_price_points=('avg_price', 'nunique'),
+        total_qty=('qty', 'sum'),
+    )
+    eligible = sku_stats[
+        (sku_stats['n_weeks'] >= min_weeks) &
+        (sku_stats['total_qty'] >= min_total_qty) &
+        (sku_stats['n_price_points'] >= min_price_points)
+    ].index.tolist()
+    return eligible

requirements.txt

@@ -0,0 +1,8 @@
+numpy>=1.23
+pandas>=1.5
+streamlit>=1.30
+plotly>=5.15
+scikit-learn>=1.2
+scipy>=1.9
+pyarrow>=12.0
+openpyxl>=3.1

tools/build_uci_panel.py

@@ -0,0 +1,27 @@
+from pathlib import Path
+import pandas as pd
+
+from pricing_engine.data_uci import make_sku_week_panel
+
+RAW_PATH = Path("data/Online Retail.xlsx")
+OUT_PATH = Path("data/processed/uci_sku_week.parquet")
+
+def main():
+    if not RAW_PATH.exists():
+        raise FileNotFoundError(f"Missing raw file: {RAW_PATH}")
+
+    print("Loading raw UCI Online Retail…")
+    df_raw = pd.read_excel(RAW_PATH)
+
+    print("Building SKU-week panel…")
+    panel = make_sku_week_panel(df_raw)
+
+    OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
+
+    print(f"Writing parquet → {OUT_PATH}")
+    panel.to_parquet(OUT_PATH, index=False)
+
+    print("Done.")
+
+if __name__ == "__main__":
+    main()