Spaces:
Running on Zero
Running on Zero
| """ | |
| 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 | |
| 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")) | |