Building the Simulator Front End

74.1 Who uses the simulator, and for what (40 min)

The users are a category manager preparing a pricing review and a CFO checking the plan. Neither knows what an elasticity is; both know what a price is. So:

• Inputs are decisions (price change by segment, coupon share and cost), never parameters.
• Assumptions are inputs too, but labelled as assumptions: pull-forward and error correlation sit at the bottom of the sidebar with plain-language labels.
• The first screen is the counterfactual. Before anyone moves a slider, the app says what happens if nothing changes.

Why Streamlit? The course website is Next.js, which is the right choice for a product used by thousands. An internal decision tool for twenty people needs to be built by the analyst, in Python, in a day. Streamlit reruns the script top to bottom on every interaction; no front-end code at all. Its limits (layout control, many concurrent users) don't matter here.

74.2 A Streamlit app in one file (50 min)

The app uses three charts from app/charts.py. Tomorrow is about their design; for now, create the file as it is:

"""Charts for counterfactuals (Day 75). Each function takes a simulation result and returns an Altair chart."""
import altair as alt
import numpy as np
import pandas as pd

BLUE, GREY, GREEN, RED = "#2563eb", "#94a3b8", "#16a34a", "#dc2626"


def projected_vs_counterfactual(r: dict, measure: str = "gross_profit") -> alt.Chart:
    """Two bars: what happens if nothing changes, and the scenario (median, with its 90% range)."""
    label = "Gross profit" if measure == "gross_profit" else "Revenue"
    p = r["projected"][measure]
    df = pd.DataFrame({"case": ["No change", "Scenario"],
                       "value": [r["counterfactual"][measure] / 1e6, p["median"] / 1e6],
                       "low": [r["counterfactual"][measure] / 1e6, p["p5"] / 1e6],
                       "high": [r["counterfactual"][measure] / 1e6, p["p95"] / 1e6]})
    base = alt.Chart(df).encode(y=alt.Y("case:N", title=None, sort=None, axis=alt.Axis(labelLimit=200)))
    bars = base.mark_bar(size=28).encode(
        x=alt.X("value:Q", title=f"{label}, $M a year"),
        color=alt.Color("case:N", scale=alt.Scale(range=[GREY, BLUE]), legend=None),
        tooltip=[alt.Tooltip("value:Q", format=",.2f", title="$M (median)"),
                 alt.Tooltip("low:Q", format=",.2f", title="5th pct"), alt.Tooltip("high:Q", format=",.2f", title="95th pct")])
    ranges = base.mark_rule(strokeWidth=2, color="#1e293b").encode(x="low:Q", x2="high:Q")
    return (bars + ranges).properties(height=120)


def change_by_segment(r: dict) -> alt.Chart:
    """Where the profit change comes from: each segment's pricing effect, plus the coupon programme."""
    cells = pd.DataFrame(r["cells"])
    seg = cells.groupby("segment", as_index=False).profit_change_median.sum()
    rows = list(seg.itertuples(index=False, name=None))
    if r["coupons"]["share"] > 0:
        rows.append(("coupon programme", r["coupons"]["profit_change"]["median"]))
    df = pd.DataFrame(rows, columns=["part", "change"])
    df["change"] = df["change"] / 1e3
    df["sign"] = df["change"].ge(0).map({True: "gain", False: "loss"})
    return alt.Chart(df).mark_bar().encode(
        y=alt.Y("part:N", title=None, sort="-x", axis=alt.Axis(labelLimit=200, labelOverlap=False)),
        x=alt.X("change:Q", title="change in annual gross profit, $k (median)"),
        color=alt.Color("sign:N", scale=alt.Scale(domain=["gain", "loss"], range=[GREEN, RED]), legend=None),
        tooltip=["part", alt.Tooltip("change:Q", format=",.0f", title="$k")]).properties(height=42 * len(df))


def outcome_distribution(r: dict) -> alt.Chart:
    """The spread of plausible outcomes, with the line that matters: zero."""
    x = np.array(r["draws"]["gross_profit_change"]) / 1e3
    counts, edges = np.histogram(x, bins=30)
    df = pd.DataFrame({"start": edges[:-1], "end": edges[1:], "count": counts})
    hist = alt.Chart(df).mark_bar(color=BLUE, opacity=0.7).encode(
        x=alt.X("start:Q", title="change in annual gross profit, $k"), x2="end:Q",
        y=alt.Y("count:Q", title="simulated outcomes"), y2=alt.datum(0))
    zero = alt.Chart(pd.DataFrame({"x": [0]})).mark_rule(color=RED, strokeWidth=2).encode(x="x:Q")
    return (hist + zero).properties(height=200)

Create app/streamlit_app.py:

"""
NorthMart scenario simulator (Module 8, Days 74-75).

    streamlit run app/streamlit_app.py

By default it runs the engine in-process. Set API_URL (e.g. http://127.0.0.1:8001) to call the scenario
API instead, which is how it runs once the API is deployed.
"""
import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import pandas as pd  # noqa: E402
import requests  # noqa: E402
import streamlit as st  # noqa: E402

from app.charts import change_by_segment, outcome_distribution, projected_vs_counterfactual  # noqa: E402
from engine import Scenario, load_parameters, simulate  # noqa: E402
from engine.narrate import money, summary  # noqa: E402


def md(text: str) -> str:
    """Streamlit reads $...$ as LaTeX; escape dollar signs in anything that shows money."""
    return text.replace("$", "\\$")

st.set_page_config(page_title="NorthMart scenario simulator", layout="wide")
API_URL = os.environ.get("API_URL")


@st.cache_data
def get_parameters():
    if API_URL:
        return requests.get(f"{API_URL}/parameters", timeout=10).json()
    return load_parameters(os.environ.get("PARAMS_PATH", "models/causal_parameters.json"))


@st.cache_data
def run(scenario_json: str):
    sc = Scenario.model_validate_json(scenario_json)
    if API_URL:
        resp = requests.post(f"{API_URL}/scenario", data=scenario_json, params={"include_draws": True}, timeout=30)
        resp.raise_for_status()
        return resp.json()
    return simulate(get_parameters(), sc)


params = get_parameters()
segments = sorted({c["segment"] for c in params["pricing"]})

st.title("NorthMart scenario simulator")
st.caption(f"Causal parameters version {params['version']} · projections are a year ahead, against the "
           "counterfactual of changing nothing")

# Every scenario is shareable: its settings live in the URL, e.g. ?price=5&coupons=50
q = st.query_params


def from_url(key, default, cast=int):
    try:
        return cast(q.get(key, default))
    except ValueError:
        return default


with st.sidebar:
    st.header("Pricing")
    everywhere = st.slider("Price change everywhere (%)", -10, 15, from_url("price", 0), 1)
    overrides = {}
    with st.expander("Override by segment"):
        for s in segments:
            if st.checkbox(s, key=f"use_{s}"):
                overrides[s] = st.slider(f"{s} (%)", -10, 15, everywhere, 1, key=f"pct_{s}")
    st.header("Coupons")
    share = st.slider("Share of customers couponed (top by predicted response, %)", 0, 100, from_url("coupons", 0), 10)
    cost = st.slider("Cost per coupon ($)", 2.0, 10.0, from_url("cost", 5.0, float), 0.5)
    pull = st.slider("Pull-forward: extra spend that is just earlier purchases (%)", 0, 40,
                     int(params["assumptions"]["pull_forward"] * 100), 5)
    st.header("Uncertainty")
    rho = st.slider("How much do estimation errors move together? (0 = independent)", 0.0, 1.0, 0.0, 0.25)

changes = {"all": everywhere / 100} | {s: v / 100 for s, v in overrides.items()}
scenario = Scenario(price_change={k: v for k, v in changes.items() if v != 0 or k != "all"},
                    coupon_share=share / 100, coupon_cost=cost, pull_forward=pull / 100,
                    elasticity_correlation=rho)
r = run(scenario.model_dump_json())
st.query_params.update({"price": everywhere, "coupons": share, "cost": cost})

if not scenario.price_change and scenario.coupon_share == 0:
    st.info(md(f"Counterfactual: gross profit {money(r['counterfactual']['gross_profit'])} and revenue "
               f"{money(r['counterfactual']['revenue'])} a year if nothing changes. Move a slider to build a scenario."))
    st.stop()

g, rev = r["change"]["gross_profit"], r["change"]["revenue"]
c1, c2, c3 = st.columns(3)
c1.metric("Gross profit, change per year", md(money(g["median"])), help="Median of the simulated outcomes")
c2.metric("90% range", md(f"{money(g['p5'])} to {money(g['p95'])}"))
p_up = r["change"]["prob_profit_up"]
c3.metric("Chance profit goes up", "-" if p_up is None else f"{p_up:.0%}")
st.write(md(summary("This scenario", r)))
if r["warnings"]:
    st.warning(f"{len(r['warnings'])} price changes go beyond what the data have seen: those projections extrapolate.")
    with st.expander("Which ones?"):
        for w in r["warnings"]:
            st.write(md(w))

left, right = st.columns(2)
with left:
    st.subheader("Projected vs counterfactual")
    st.altair_chart(projected_vs_counterfactual(r, "gross_profit"), width="stretch")
    st.altair_chart(projected_vs_counterfactual(r, "revenue"), width="stretch")
with right:
    st.subheader("Where the change comes from")
    st.altair_chart(change_by_segment(r), width="stretch")
    st.subheader("Plausible outcomes")
    st.altair_chart(outcome_distribution(r), width="stretch")

with st.expander("Every segment x category cell"):
    cells = pd.DataFrame(r["cells"])
    st.dataframe(cells.assign(price_change=cells.price_change.map("{:+.0%}".format),
                              profit_change_median=(cells.profit_change_median / 1e3).round(1))
                 [["segment", "category", "price_change", "profit_change_median", "prob_profit_up"]]
                 .rename(columns={"profit_change_median": "profit change ($k)", "prob_profit_up": "P(up)"}),
                 hide_index=True)
st.caption("Uncertainty shown is from the causal estimates (and their correlation, if set). It does not include "
           "uncertainty in assumptions such as pull-forward or coupon fatigue: change those with the sliders.")

streamlit run app/streamlit_app.py

Codespaces offers to open port 8501. Move the "Price change everywhere" slider to +5%: the metrics, sentence and charts update in about a tenth of a second. Then tick "student" and set it to +10%. The warning appears, and the expander lists the two student cells whose supported range is below 10%.

Two details worth copying:

• @st.cache_data on the parameters load and on run: the same scenario is never simulated twice.
• md() escapes dollar signs. Streamlit reads $...$ as LaTeX, so "$459k to $502k" would otherwise render as a garbled formula. Every money string goes through it.

74.3 Defaults, empty states and warnings (45 min)

Good tools are designed for the moment nobody has done anything yet, and for the moment someone asks something unreasonable.

Each of these was a bug in the first version of this app. Test them by hand now: set everything to zero; set +15% everywhere; set coupons to 0% and price to 0% but correlation to 1.

74.4 Shareable scenarios and the API switch (40 min)

Shareable scenarios. st.query_params writes the main sliders into the URL. Copy the address bar and send it: the recipient opens the same scenario. This turns a tool into a conversation: "here's what I mean, click this".

The API switch. By default the app imports the engine. Set API_URL and it calls the Day 71 service instead:

uvicorn service.app:app --port 8001 &
API_URL=http://127.0.0.1:8001 streamlit run app/streamlit_app.py

Same answers, but now the app and the pricing system use one deployed copy of the parameters. When the parameters are rebuilt, both change together. Stop the background server afterwards with kill %1.

74.5 Day 74 mini-project: testing the app without a browser (75 min)

Clicking through an app after every change doesn't scale. Streamlit's AppTest runs the script headless, lets you set widgets and read the outputs.

Try it. With AppTest, set coupons to the top 50% and print the three metrics; then raise the coupon cost to $9 and print the profit change again. Create day74/try_apptest.py. (AppTest resolves paths relative to the calling file.)

# day74/try_apptest.py   (python -m day74.try_apptest)
from streamlit.testing.v1 import AppTest

at = AppTest.from_file("../app/streamlit_app.py", default_timeout=60).run()
at.slider[1].set_value(50).run()          # coupons to the top 50%
for m in at.metric:
    print(f"{m.label}: {m.value}")
at.slider[2].set_value(9.0).run()         # dearer coupons
print("at $9 a coupon:", at.metric[0].value)

Gross profit, change per year: \$5.6M
90% range: \$4.8M to \$6.4M
Chance profit goes up: 100%
at $9 a coupon: -\$4.0M

74.6 Save your work with Git (20 min)

git add app day74
git commit -m "Day 74: Streamlit scenario simulator"
git push

Day 74 checklist

• My app's inputs are decisions and labelled assumptions, not model parameters
• The first screen shows the counterfactual
• Extrapolation warnings are visible but not overwhelming
• A scenario can be shared as a link
• I can drive the app from a script with AppTest