import os
import numpy as np
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import streamlit as st
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
st.set_page_config(
page_title="FreshWise Studio",
page_icon="🥬",
layout="wide",
initial_sidebar_state="expanded",
)
DATA_CANDIDATES = [
os.environ.get("DATA_PATH", ""),
"perishable_goods_management.csv",
"/app/perishable_goods_management.csv",
"/data/perishable_goods_management.csv",
"/mnt/data/perishable_goods_management.csv",
]
BRAND = {
"name": "FreshWise Studio",
"tagline": "Turn perishables into profit, loyalty, and lower waste.",
"manager_copy": "For operators who need sharper replenishment, clearer risk alerts, and smarter campaign design.",
"consumer_copy": "For shoppers who want better deals, curated bundles, and easy discovery of time-sensitive offers.",
}
REGION_COORDS = {
"North": (53.48, -2.24),
"South": (51.50, -0.12),
"East": (52.20, 0.12),
"West": (51.48, -3.18),
"Central": (52.48, -1.89),
}
def inject_css():
st.markdown(
"""
""",
unsafe_allow_html=True,
)
def find_data_path() -> str:
for path in DATA_CANDIDATES:
if path and os.path.exists(path):
return path
raise FileNotFoundError("perishable_goods_management.csv not found. Put it next to app.py or set DATA_PATH.")
@st.cache_data(show_spinner=False)
def load_data() -> pd.DataFrame:
df = pd.read_csv(find_data_path())
df["transaction_date"] = pd.to_datetime(df["transaction_date"], errors="coerce")
df["expiration_date"] = pd.to_datetime(df["expiration_date"], errors="coerce")
df["sell_through_pct"] = np.where(df["initial_quantity"] > 0, df["units_sold"] / df["initial_quantity"], 0)
df["stock_demand_ratio"] = np.where(df["daily_demand"] > 0, df["initial_quantity"] / df["daily_demand"], np.nan)
df["gross_margin"] = df["selling_price"] - df["cost_price"]
df["leftover_units"] = (df["initial_quantity"] - df["units_sold"]).clip(lower=0)
df["savings"] = df["base_price"] - df["selling_price"]
df["value_score"] = (
(1 - df["waste_pct"].clip(0, 1)) * 0.25
+ df["discount_pct"].clip(0, 0.6) * 0.35
+ df["sell_through_pct"].clip(0, 1) * 0.2
+ (1 - np.minimum(df["days_until_expiry"], 14) / 14) * 0.2
)
df["expiry_bucket"] = pd.cut(
df["days_until_expiry"],
bins=[-1, 1, 3, 7, 30, 10000],
labels=["<=1d", "2-3d", "4-7d", "8-30d", ">30d"],
)
df["high_waste_flag"] = (df["waste_pct"] >= df["waste_pct"].quantile(0.75)).astype(int)
df["promo_readiness"] = (
df["discount_pct"] * 0.3
+ (1 - np.minimum(df["days_until_expiry"], 7) / 7) * 0.3
+ df["spoilage_risk"] * 0.2
+ (1 - df["sell_through_pct"].clip(0, 1)) * 0.2
)
return attach_store_locations(df)
def _region_anchor(region: str):
return REGION_COORDS.get(region, (52.0, 0.0))
def attach_store_locations(df: pd.DataFrame) -> pd.DataFrame:
stores = sorted(df["store_id"].dropna().unique())
rows = []
for store in stores:
sub = df[df["store_id"] == store]
region = str(sub["region"].mode().iloc[0]) if not sub.empty else "Central"
base_lat, base_lon = _region_anchor(region)
seed = abs(hash(store)) % 10000
rng = np.random.default_rng(seed)
rows.append(
{
"store_id": store,
"store_lat": base_lat + rng.uniform(-0.35, 0.35),
"store_lon": base_lon + rng.uniform(-0.45, 0.45),
}
)
loc = pd.DataFrame(rows)
return df.merge(loc, on="store_id", how="left")
@st.cache_resource(show_spinner=False)
def risk_model(df: pd.DataFrame):
features = [
"daily_demand", "initial_quantity", "shelf_life_days", "days_until_expiry",
"temp_deviation", "temp_abuse_events", "handling_score", "packaging_score",
"spoilage_risk", "discount_pct", "markdown_applied", "is_weekend", "supplier_score",
]
X = df[features]
y = df["high_waste_flag"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)
model = RandomForestClassifier(n_estimators=150, random_state=42, max_depth=10, n_jobs=-1)
model.fit(X_train, y_train)
importances = pd.Series(model.feature_importances_, index=features).sort_values(ascending=False)
return model, importances
def ensure_state():
if "auth_role" not in st.session_state:
st.session_state.auth_role = None
if "auth_name" not in st.session_state:
st.session_state.auth_name = ""
if "logged_in" not in st.session_state:
st.session_state.logged_in = False
def hero(title: str, subtitle: str):
st.markdown(
f"""
Fresh retail intelligence
{title}
{subtitle}
""",
unsafe_allow_html=True,
)
def landing_page():
hero(BRAND["name"], BRAND["tagline"])
c1, c2 = st.columns(2)
with c1:
st.markdown(
f"""
Manager portal
Operate smarter stores
{BRAND['manager_copy']}
Waste alerts
Store map
Promotion simulation
Replenishment
""",
unsafe_allow_html=True,
)
with st.form("manager_login"):
name = st.text_input("Manager name", placeholder="Alex Chen")
team = st.text_input("Team / region", placeholder="Operations - North")
submitted = st.form_submit_button("Enter Manager Portal", use_container_width=True)
if submitted:
st.session_state.logged_in = True
st.session_state.auth_role = "Manager"
st.session_state.auth_name = name or "Manager"
st.session_state.auth_team = team or "Operations"
st.rerun()
with c2:
st.markdown(
f"""
Consumer app
Find timely deals that fit life
{BRAND['consumer_copy']}
Deal finder
Smart bundles
Store map
Personalized picks
""",
unsafe_allow_html=True,
)
with st.form("consumer_login"):
name = st.text_input("Your name", placeholder="Jamie")
home_store = st.text_input("Preferred store / area", placeholder="South")
submitted = st.form_submit_button("Enter Consumer App", use_container_width=True)
if submitted:
st.session_state.logged_in = True
st.session_state.auth_role = "Consumer"
st.session_state.auth_name = name or "Guest"
st.session_state.auth_team = home_store or "Nearby stores"
st.rerun()
a, b, c = st.columns(3)
a.markdown("Reduce waste
Turn near-expiry inventory into higher sell-through with timing-aware recommendations.
Improve margin
Simulate promotions before launch and balance discount depth, channel, and duration.
Boost shopper value
Guide customers toward bundles, bargains, and nearby offers that still feel fresh.
Launch a {discount}% {promo_category} campaign for {expiry_target} inventory via {channel} for {duration} days.
Prioritize the objective {objective}, {'bundle with complementary items' if bundle else 'keep as a single-item offer'},
and {'target high-risk stores first' if geo_boost else 'deploy evenly across stores'}.
This scenario is expected to improve sell-through while lowering expiry-driven waste pressure.
""",
unsafe_allow_html=True,
)
def manager_risk(df: pd.DataFrame):
st.markdown("## Risk monitor")
_, importances = risk_model(df)
c1, c2 = st.columns([1.05, 1])
with c1:
fig = px.bar(importances.head(10).sort_values(), orientation="h", title="Top drivers of high waste risk")
st.plotly_chart(fig, use_container_width=True)
with c2:
st.plotly_chart(store_map(df, "temp_deviation", "temp_abuse_events", "Store map: temperature risk and handling exposure"), use_container_width=True)
alerts = (
df.groupby("store_id")[["temp_deviation", "temp_abuse_events", "waste_pct", "profit", "spoilage_risk"]]
.mean()
.assign(alert_score=lambda x: 0.28 * x["temp_deviation"] + 0.18 * x["temp_abuse_events"] + 0.34 * x["waste_pct"] * 10 + 0.2 * x["spoilage_risk"])
.sort_values("alert_score", ascending=False)
.head(15)
.reset_index()
)
st.dataframe(alerts, use_container_width=True, hide_index=True)
def consumer_deals(df: pd.DataFrame):
hero("Consumer deal finder", "Discover nearby offers, value-packed products, and time-sensitive bargains.")
c1, c2, c3 = st.columns(3)
budget = c1.slider("Budget (€)", 5, 60, 20)
preferred_category = c2.selectbox("Category", ["All"] + sorted(df["category"].unique()))
max_expiry = c3.slider("Max days until expiry", 1, 14, 5)
deals = df[df["days_until_expiry"] <= max_expiry].copy()
if preferred_category != "All":
deals = deals[deals["category"] == preferred_category]
deals["deal_score"] = deals["discount_pct"] * 0.4 + deals["value_score"] * 0.4 + deals["savings"].clip(lower=0) / deals["base_price"].replace(0, np.nan).fillna(1) * 0.2
deals = deals.sort_values(["deal_score", "savings"], ascending=False)
st.plotly_chart(store_map(deals, "discount_pct", "units_sold", "Nearby stores with strong deal intensity"), use_container_width=True)
st.dataframe(deals[["product_name", "category", "store_id", "days_until_expiry", "base_price", "selling_price", "discount_pct", "savings"]].head(30), use_container_width=True, hide_index=True)
best = deals[deals["selling_price"] <= budget].head(9)
cols = st.columns(3)
for i, (_, row) in enumerate(best.iterrows()):
with cols[i % 3]:
st.markdown(
f"""
{row['category']}
{row['product_name']}
€{row['selling_price']:.2f} now · save €{row['savings']:.2f}
Store: {row['store_id']}
Expires in {int(row['days_until_expiry'])} day(s)
""",
unsafe_allow_html=True,
)
def build_bundle(df: pd.DataFrame, budget: float, people: int, theme: str):
work = df[df["days_until_expiry"] <= 7].copy()
work["score"] = work["value_score"] + work["discount_pct"] + np.where(work["selling_price"] <= budget / max(people, 1), 0.15, 0)
theme_map = {
"Quick dinner": ["Ready_to_Eat", "Produce", "Bakery", "Dairy"],
"Healthy protein": ["Meat", "Seafood", "Dairy", "Produce"],
"Family breakfast": ["Bakery", "Dairy", "Beverages", "Produce"],
"Budget saver": list(work["category"].unique()),
}
work = work[work["category"].isin(theme_map.get(theme, list(work["category"].unique())))]
work = work.sort_values(["score", "selling_price"], ascending=[False, True])
chosen, remaining, target_items, used = [], budget, min(max(people + 1, 3), 6), set()
for _, row in work.iterrows():
if row["selling_price"] <= remaining:
if theme != "Budget saver" and row["category"] in used:
continue
chosen.append(row)
remaining -= row["selling_price"]
used.add(row["category"])
if len(chosen) >= target_items:
break
if not chosen:
return pd.DataFrame(), 0.0, 0.0
bundle = pd.DataFrame(chosen)
return bundle, float(bundle["selling_price"].sum()), float(bundle["savings"].sum())
def consumer_bundles(df: pd.DataFrame):
st.markdown("## Bundle builder")
c1, c2, c3 = st.columns(3)
budget = c1.slider("Bundle budget (€)", 8, 90, 28)
people = c2.slider("People", 1, 6, 2)
theme = c3.selectbox("Bundle theme", ["Quick dinner", "Healthy protein", "Family breakfast", "Budget saver"])
bundle, total, saved = build_bundle(df, budget, people, theme)
if bundle.empty:
st.warning("No bundle found for these settings.")
return
m1, m2, m3 = st.columns(3)
m1.metric("Bundle total", f"€{total:.2f}")
m2.metric("You save", f"€{saved:.2f}")
m3.metric("Items", f"{len(bundle)}")
st.dataframe(bundle[["product_name", "category", "store_id", "selling_price", "base_price", "discount_pct", "days_until_expiry"]], use_container_width=True, hide_index=True)
st.info("Managers can reuse these bundles as prebuilt campaign templates for near-expiry conversion.")
def consumer_personal(df: pd.DataFrame):
st.markdown("## Personalized promotions")
c1, c2, c3 = st.columns(3)
favorite = c1.selectbox("Favorite category", sorted(df["category"].unique()))
price_cap = c2.slider("Max item price (€)", 1, 30, 10)
safe_window = c3.checkbox("Hide items expiring within 1 day", value=False)
recs = df[df["category"] == favorite].copy()
recs = recs[recs["selling_price"] <= price_cap]
if safe_window:
recs = recs[recs["days_until_expiry"] > 1]
recs["score"] = recs["discount_pct"] * 0.5 + recs["value_score"] * 0.5
recs = recs.sort_values("score", ascending=False).head(12)
cols = st.columns(3)
for i, (_, row) in enumerate(recs.iterrows()):
with cols[i % 3]:
st.markdown(
f"""
Recommended for you
{row['product_name']}
{row['category']} · {row['store_id']}
€{row['selling_price']:.2f} now · save €{row['savings']:.2f}
Expires in {int(row['days_until_expiry'])} day(s)
""",
unsafe_allow_html=True,
)
st.button("Save offer", key=f"save_{i}", use_container_width=True)
def manager_shell(df: pd.DataFrame):
tabs = st.tabs(["Overview", "Inventory", "Promotion Studio", "Risk", "Store Map"])
with tabs[0]:
manager_dashboard(df)
with tabs[1]:
manager_inventory(df)
with tabs[2]:
manager_promotions(df)
with tabs[3]:
manager_risk(df)
with tabs[4]:
st.markdown("## Store network view")
st.plotly_chart(store_map(df, "profit", "units_sold", "Store map: profitability and sales concentration"), use_container_width=True)
by_store = df.groupby(["store_id", "region"])[["profit", "waste_pct", "units_sold", "temp_deviation"]].mean().reset_index()
st.dataframe(by_store.sort_values("profit", ascending=False), use_container_width=True, hide_index=True)
def consumer_shell(df: pd.DataFrame):
tabs = st.tabs(["Deals", "Bundles", "Personalized", "Store Map"])
with tabs[0]:
consumer_deals(df)
with tabs[1]:
consumer_bundles(df)
with tabs[2]:
consumer_personal(df)
with tabs[3]:
st.markdown("## Nearby store map")
st.plotly_chart(store_map(df, "discount_pct", "units_sold", "Store map: where deals are strongest right now"), use_container_width=True)
shortlist = df.sort_values(["discount_pct", "value_score"], ascending=False)[["store_id", "region", "product_name", "category", "selling_price", "discount_pct", "days_until_expiry"]].head(25)
st.dataframe(shortlist, use_container_width=True, hide_index=True)
def main():
inject_css()
ensure_state()
try:
df = load_data()
except Exception as e:
st.error(str(e))
st.stop()
if not st.session_state.logged_in:
landing_page()
st.stop()
filtered = sidebar_filters(df)
if filtered.empty:
st.warning("No data left after filtering.")
st.stop()
st.caption(f"Logged in as {st.session_state.auth_name} · {st.session_state.auth_role}")
if st.session_state.auth_role == "Manager":
manager_shell(filtered)
else:
consumer_shell(filtered)
st.markdown(
"""
""",
unsafe_allow_html=True,
)
if __name__ == "__main__":
main()