""" Retail Analytics Toolkit - interactive demo. Builds the toolkit's fully synthetic retail dataset on startup, runs the same SQL queries as the GitHub project, and shows the results as interactive charts. Source: https://github.com/Anjub004/retail-analytics-toolkit """ import re import sqlite3 import subprocess import sys from pathlib import Path import gradio as gr import pandas as pd import plotly.express as px # Hugging Face ZeroGPU hardware expects at least one @spaces.GPU function at startup. # This app runs entirely on CPU, so the placeholder below is never called. try: import spaces @spaces.GPU def _zero_gpu_placeholder(): return None except ImportError: pass ROOT = Path(__file__).resolve().parent DB_PATH = ROOT / "data" / "retail.db" SQL_DIR = ROOT / "sql" TEAL, RED, AMBER, BLUE = "#14b8a6", "#e5534b", "#f0a030", "#5b8def" # ---------- data ---------- def ensure_database() -> None: if not DB_PATH.exists(): subprocess.run([sys.executable, str(ROOT / "data" / "generate_data.py")], check=True) def load_queries(path: Path) -> dict[str, str]: text = path.read_text() parts = re.split(r"^--\s*name:\s*(\w+)\s*$", text, flags=re.MULTILINE) return {parts[i]: parts[i + 1].strip() for i in range(1, len(parts), 2)} def run_queries() -> dict[str, pd.DataFrame]: results = {} with sqlite3.connect(DB_PATH) as conn: for sql_file in sorted(SQL_DIR.glob("*.sql")): for name, query in load_queries(sql_file).items(): results[name] = pd.read_sql_query(query, conn) for table in ["branches", "products", "stock_adjustments", "sales"]: results[f"t_{table}"] = pd.read_sql_query(f"SELECT * FROM {table}", conn) return results ensure_database() R = run_queries() BRANCH_NAMES = ["All branches"] + R["t_branches"]["branch_name"].tolist() def style(fig): fig.update_layout( template="plotly_white", margin=dict(l=10, r=10, t=50, b=10), font=dict(family="Inter, system-ui, sans-serif", size=13), title_font=dict(size=16), legend_title_text="", ) return fig # ---------- overview ---------- def overview_md() -> str: sb = R["shrink_by_branch"] net = sb["shrink_value"].sum() / sb["revenue"].sum() * 100 worst = sb.iloc[0] reason = R["shrink_by_category_reason"].groupby("reason")["shrink_value"].sum() cat = R["shrink_by_category_reason"].groupby("category")["shrink_value"].sum().idxmax() wk = R["weekday_pattern"] busiest = wk.loc[wk["avg_visitors"].idxmax(), "weekday"][2:] quietest = wk.loc[wk["avg_visitors"].idxmin(), "weekday"][2:] conv = R["branch_conversion"] best = conv.loc[conv["conversion_pct"].idxmax()] risk = R["expiry_risk"] return f""" ### Key findings (synthetic data) | Metric | Result | |---|---| | **Network shrink** | {net:.2f}% of sales | | **Highest-shrink branch** | {worst.branch_name}, {worst.shrink_pct:.2f}% ({worst.shrink_pct / net:.1f}x the network average) | | **Biggest loss reason** | {reason.idxmax()} ({reason.max() / reason.sum() * 100:.0f}% of shrink value) | | **Category losing the most** | {cat} | | **Busiest / quietest day** | {busiest} / {quietest} | | **Best conversion** | {best.branch_name} at {best.conversion_pct}% | | **Batches at expiry risk (30 days)** | {len(risk)}, worth {risk['value_at_risk'].sum():,.2f} at cost | """ # ---------- shrinkage ---------- def shrink_branch_fig(): df = R["shrink_by_branch"].sort_values("shrink_pct") avg = df["shrink_value"].sum() / df["revenue"].sum() * 100 df["vs_network"] = ["Above average" if v > avg else "At or below average" for v in df["shrink_pct"]] fig = px.bar(df, x="shrink_pct", y="branch_name", orientation="h", color="vs_network", color_discrete_map={"Above average": RED, "At or below average": BLUE}, labels={"shrink_pct": "Shrink % of revenue", "branch_name": ""}, title="Shrinkage % of sales by branch", hover_data={"revenue": ":,.0f", "shrink_value": ":,.0f", "vs_network": False}) fig.add_vline(x=avg, line_dash="dash", line_color="grey", annotation_text=f"Network {avg:.2f}%", annotation_position="top") return style(fig) def shrink_mix(branch: str): adj = R["t_stock_adjustments"].merge(R["t_products"][["product_id", "category", "product_name"]], on="product_id") if branch != "All branches": bid = R["t_branches"].set_index("branch_name").loc[branch, "branch_id"] adj = adj[adj["branch_id"] == bid] mix = adj.groupby(["category", "reason"], as_index=False)["cost_value"].sum() order = mix.groupby("category")["cost_value"].sum().sort_values().index.tolist() fig = px.bar(mix, x="cost_value", y="category", color="reason", orientation="h", category_orders={"category": order}, color_discrete_sequence=[AMBER, RED, "#8b5cf6", BLUE], labels={"cost_value": "Shrink value (cost)", "category": ""}, title=f"Where is the loss coming from? ({branch})") top = (adj.groupby(["product_name", "category"], as_index=False) .agg(units_lost=("qty", "sum"), shrink_value=("cost_value", "sum")) .sort_values("shrink_value", ascending=False).head(10)) top["shrink_value"] = top["shrink_value"].round(2) return style(fig), top def shrink_trend_fig(): df = R["shrink_monthly_trend"] fig = px.line(df, x="month", y="shrink_pct", markers=True, title="Monthly shrink % of sales", labels={"shrink_pct": "Shrink %", "month": ""}) fig.update_traces(line_color=TEAL) return style(fig) # ---------- footfall ---------- def weekday_fig(): df = R["weekday_pattern"].copy() df["day"] = df["weekday"].str[2:] fig = px.bar(df, x="day", y="avg_visitors", text="avg_visitors", title="Average daily visitors by weekday", labels={"avg_visitors": "Visitors per branch", "day": ""}, hover_data={"conversion_pct": True}) fig.update_traces(marker_color=BLUE, texttemplate="%{text:,.0f}", textposition="outside") return style(fig) def weekly_fig(): df = R["weekly_trend"].melt(id_vars="week", value_vars=["visitors", "transactions"], var_name="metric", value_name="count") fig = px.line(df, x="week", y="count", color="metric", markers=True, title="Weekly footfall vs transactions (all branches)", color_discrete_map={"visitors": BLUE, "transactions": TEAL}, labels={"count": "", "week": ""}) return style(fig) # ---------- expiry ---------- def expiry_view(branch: str, days: int): risk = R["expiry_risk"] risk = risk[risk["days_left"] <= days] if branch != "All branches": risk = risk[risk["branch_name"] == branch] total = risk["value_at_risk"].sum() summary = (f"**{len(risk)}** batches projected to have unsold stock at expiry within **{days} days**, " f"worth **{total:,.2f}** at cost ({branch}).") by_cat = risk.groupby("category", as_index=False)["value_at_risk"].sum().sort_values("value_at_risk") fig = px.bar(by_cat, x="value_at_risk", y="category", orientation="h", title="Value at risk by category", labels={"value_at_risk": "Value at risk (cost)", "category": ""}) fig.update_traces(marker_color=AMBER) cols = ["branch_name", "product_name", "category", "days_left", "qty_on_hand", "avg_daily_units", "projected_unsold", "value_at_risk"] return summary, style(fig), risk[cols].head(25) def buckets_fig(): df = R["expiry_buckets"] fig = px.bar(df, x="bucket", y="stock_value", text="batches", color="bucket", color_discrete_sequence=[RED, AMBER, "#f7d774", "#22c55e"], title="All stock value by days to expiry", labels={"stock_value": "Stock value (cost)", "bucket": ""}) fig.update_traces(texttemplate="%{text} batches", textposition="outside") fig.update_layout(showlegend=False) return style(fig) # ---------- UI ---------- INTRO = """ # 🛒 Retail Analytics Toolkit Three questions every retail business asks, answered with SQL and Python on a **fully synthetic dataset** (8 branches, 60 products, 6 months of daily sales): 1. **Where are we losing stock?** 2. **Are visitors turning into buyers?** 3. **What will expire before it sells?** Code: [github.com/Anjub004/retail-analytics-toolkit](https://github.com/Anjub004/retail-analytics-toolkit) · Author: [Anju Barai](https://huggingface.co/Anju4). All names and numbers are fake. """ HOW = """ ### How the metrics work - **Shrink %** = value of stock adjustments (at cost) ÷ sales revenue × 100 - **Conversion %** = transactions ÷ visitors × 100 - **Projected unsold at expiry** = stock on hand − (average daily sales over the last 28 days × days until expiry). If it's positive, that stock will probably expire on the shelf, and *value at risk* is what it costs. The SQL is plain SQLite and ports easily to BigQuery, PostgreSQL or MySQL. Outputs plug straight into Looker Studio, Power BI or Google Sheets. """ with gr.Blocks(title="Retail Analytics Toolkit") as demo: gr.Markdown(INTRO) with gr.Tab("Overview"): gr.Markdown(overview_md()) gr.Plot(shrink_branch_fig()) with gr.Tab("Shrinkage"): branch_s = gr.Dropdown(BRANCH_NAMES, value="All branches", label="Branch") mix_plot = gr.Plot() gr.Markdown("**Top 10 loss products**") top_table = gr.Dataframe(interactive=False) gr.Plot(shrink_trend_fig()) branch_s.change(shrink_mix, branch_s, [mix_plot, top_table]) demo.load(shrink_mix, branch_s, [mix_plot, top_table]) with gr.Tab("Footfall & conversion"): gr.Plot(weekday_fig()) gr.Plot(weekly_fig()) gr.Markdown("**Conversion by branch**") gr.Dataframe(R["branch_conversion"].drop(columns=["branch_id"]), interactive=False) with gr.Tab("Short-expiry risk"): with gr.Row(): branch_e = gr.Dropdown(BRANCH_NAMES, value="All branches", label="Branch") days_e = gr.Slider(1, 30, value=30, step=1, label="Expiring within (days)") risk_md = gr.Markdown() risk_plot = gr.Plot() gr.Markdown("**Batches to act on first (mark down, transfer or promote)**") risk_table = gr.Dataframe(interactive=False) gr.Plot(buckets_fig()) for ctrl in (branch_e, days_e): ctrl.change(expiry_view, [branch_e, days_e], [risk_md, risk_plot, risk_table]) demo.load(expiry_view, [branch_e, days_e], [risk_md, risk_plot, risk_table]) with gr.Tab("How it works"): gr.Markdown(HOW) if __name__ == "__main__": demo.launch(theme=gr.themes.Soft(primary_hue="teal"))