216 lines
10 KiB
Python
216 lines
10 KiB
Python
"""
|
|
Features 4 & 5: Store Analytics Dashboard + Product Analytics.
|
|
|
|
Pure computation over plain pandas DataFrames - no SQL in this file.
|
|
`app/services/analytics_service.py` is the thin I/O layer that pulls
|
|
DataFrames out of `store_db.py` and hands them to the functions here,
|
|
which keeps every KPI formula unit-testable without a live Postgres
|
|
instance (see `tests/test_intelligence.py`).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Dict, List, Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
|
|
def classify_stock_status(available: int, reorder_level: int, safety_stock: int) -> str:
|
|
if available <= 0:
|
|
return "Out of Stock"
|
|
if available <= safety_stock:
|
|
return "Low Stock"
|
|
if available > reorder_level * 6:
|
|
return "Overstocked"
|
|
return "In Stock"
|
|
|
|
|
|
def inventory_analytics(store_products: pd.DataFrame) -> Dict:
|
|
"""`store_products` columns: available_stock, reorder_level, safety_stock (already
|
|
filtered to one store, or pass the full multi-store frame for a
|
|
chain-wide summary)."""
|
|
if store_products.empty:
|
|
return {"total_products": 0, "in_stock": 0, "low_stock": 0, "out_of_stock": 0, "overstocked": 0}
|
|
statuses = store_products.apply(
|
|
lambda r: classify_stock_status(r["available_stock"], r["reorder_level"], r["safety_stock"]), axis=1
|
|
)
|
|
counts = statuses.value_counts()
|
|
return {
|
|
"total_products": int(len(store_products)),
|
|
"in_stock": int(counts.get("In Stock", 0)),
|
|
"low_stock": int(counts.get("Low Stock", 0)),
|
|
"out_of_stock": int(counts.get("Out of Stock", 0)),
|
|
"overstocked": int(counts.get("Overstocked", 0)),
|
|
}
|
|
|
|
|
|
def sales_analytics(orders: pd.DataFrame) -> Dict:
|
|
"""`orders` columns: order_date (datetime64), order_value. Already
|
|
filtered to the scope (one store, or all stores) the caller wants."""
|
|
if orders.empty:
|
|
return {
|
|
"total_sales": 0, "revenue": 0.0, "average_basket_value": 0.0,
|
|
"daily_sales": [], "weekly_sales": [], "monthly_sales": [],
|
|
}
|
|
revenue = float(orders["order_value"].sum())
|
|
total_sales = int(len(orders))
|
|
avg_basket = revenue / total_sales if total_sales else 0.0
|
|
|
|
daily = orders.groupby(orders["order_date"].dt.date)["order_value"].agg(["sum", "count"]).reset_index()
|
|
daily.columns = ["date", "revenue", "orders"]
|
|
weekly = orders.groupby(orders["order_date"].dt.to_period("W").astype(str))["order_value"].agg(["sum", "count"]).reset_index()
|
|
weekly.columns = ["week", "revenue", "orders"]
|
|
monthly = orders.groupby(orders["order_date"].dt.to_period("M").astype(str))["order_value"].agg(["sum", "count"]).reset_index()
|
|
monthly.columns = ["month", "revenue", "orders"]
|
|
|
|
return {
|
|
"total_sales": total_sales,
|
|
"revenue": round(revenue, 2),
|
|
"average_basket_value": round(avg_basket, 2),
|
|
"daily_sales": [{"date": str(r["date"]), "revenue": round(r["revenue"], 2), "orders": int(r["orders"])} for _, r in daily.iterrows()],
|
|
"weekly_sales": [{"week": r["week"], "revenue": round(r["revenue"], 2), "orders": int(r["orders"])} for _, r in weekly.iterrows()],
|
|
"monthly_sales": [{"month": r["month"], "revenue": round(r["revenue"], 2), "orders": int(r["orders"])} for _, r in monthly.iterrows()],
|
|
}
|
|
|
|
|
|
def profit_analytics(order_items: pd.DataFrame, store_prices: pd.DataFrame) -> Dict:
|
|
"""Joins order line items to CURRENT store cost prices to estimate
|
|
profit (`(unit_price - cost_price) * quantity`). This is an
|
|
approximation - it uses today's cost price, not the cost price that
|
|
was actually in effect on the historical order date, since the
|
|
system doesn't keep a cost-price history table. Documented rather
|
|
than silently treated as exact."""
|
|
if order_items.empty or store_prices.empty:
|
|
return {"total_profit": 0.0, "gross_profit_pct": 0.0}
|
|
merged = order_items.merge(
|
|
store_prices[["store_id", "brand", "image_id", "cost_price"]],
|
|
on=["store_id", "brand", "image_id"], how="left",
|
|
)
|
|
merged["cost_price"] = merged["cost_price"].fillna(merged["unit_price"] * 0.78)
|
|
merged["line_profit"] = (merged["unit_price"] - merged["cost_price"]) * merged["quantity"]
|
|
total_profit = float(merged["line_profit"].sum())
|
|
total_revenue = float(merged["line_total"].sum())
|
|
gp_pct = (total_profit / total_revenue * 100) if total_revenue else 0.0
|
|
return {"total_profit": round(total_profit, 2), "gross_profit_pct": round(gp_pct, 2)}
|
|
|
|
|
|
def store_comparison(orders: pd.DataFrame, order_items: pd.DataFrame, store_prices: pd.DataFrame, stores: pd.DataFrame) -> Dict:
|
|
"""`stores` columns: store_id, store_name, tier, footfall_index."""
|
|
if orders.empty:
|
|
return {"stores": [], "best_performing": None, "lowest_performing": None,
|
|
"highest_revenue": None, "highest_profit": None}
|
|
|
|
per_store_revenue = orders.groupby("store_id")["order_value"].agg(["sum", "count"]).reset_index()
|
|
per_store_revenue.columns = ["store_id", "revenue", "order_count"]
|
|
|
|
profit_rows = []
|
|
for store_id, g in order_items.groupby("store_id"):
|
|
sp = store_prices[store_prices["store_id"] == store_id]
|
|
p = profit_analytics(g, sp)
|
|
profit_rows.append({"store_id": store_id, "profit": p["total_profit"]})
|
|
profit_df = pd.DataFrame(profit_rows) if profit_rows else pd.DataFrame(columns=["store_id", "profit"])
|
|
|
|
merged = per_store_revenue.merge(profit_df, on="store_id", how="left").merge(stores, on="store_id", how="left")
|
|
merged["profit"] = merged["profit"].fillna(0.0)
|
|
merged["avg_order_value"] = merged["revenue"] / merged["order_count"].replace(0, np.nan)
|
|
merged["avg_order_value"] = merged["avg_order_value"].fillna(0.0)
|
|
# "Customer Footfall (simulated)" - the store's actual simulated order
|
|
# count IS the simulated footfall proxy (every order came from a
|
|
# simulated in-store/online customer visit).
|
|
merged["footfall_simulated"] = merged["order_count"]
|
|
|
|
result_stores = [
|
|
{
|
|
"store_id": r["store_id"], "store_name": r.get("store_name"), "tier": r.get("tier"),
|
|
"revenue": round(r["revenue"], 2), "profit": round(r["profit"], 2),
|
|
"order_count": int(r["order_count"]), "avg_order_value": round(r["avg_order_value"], 2),
|
|
"footfall_simulated": int(r["footfall_simulated"]),
|
|
}
|
|
for _, r in merged.iterrows()
|
|
]
|
|
by_revenue = sorted(result_stores, key=lambda s: s["revenue"], reverse=True)
|
|
by_profit = sorted(result_stores, key=lambda s: s["profit"], reverse=True)
|
|
|
|
return {
|
|
"stores": result_stores,
|
|
"best_performing": by_profit[0]["store_id"] if by_profit else None,
|
|
"lowest_performing": by_profit[-1]["store_id"] if by_profit else None,
|
|
"highest_revenue": by_revenue[0]["store_id"] if by_revenue else None,
|
|
"highest_profit": by_profit[0]["store_id"] if by_profit else None,
|
|
}
|
|
|
|
|
|
def top_products(order_items: pd.DataFrame, limit: int = 10, ascending: bool = False, by: str = "revenue") -> List[Dict]:
|
|
"""`by`: 'revenue' or 'units'. Set ascending=True for "lowest
|
|
selling" instead of "top selling"."""
|
|
if order_items.empty:
|
|
return []
|
|
agg = order_items.groupby(["brand", "image_id"]).agg(
|
|
revenue=("line_total", "sum"), units=("quantity", "sum"), orders=("order_id", "nunique"),
|
|
).reset_index()
|
|
sort_col = "revenue" if by == "revenue" else "units"
|
|
agg = agg.sort_values(sort_col, ascending=ascending).head(limit)
|
|
return [
|
|
{"brand": r["brand"], "image_id": r["image_id"], "revenue": round(r["revenue"], 2),
|
|
"units_sold": int(r["units"]), "order_count": int(r["orders"])}
|
|
for _, r in agg.iterrows()
|
|
]
|
|
|
|
|
|
def product_analytics(
|
|
brand: str, image_id: str,
|
|
order_items: pd.DataFrame, store_prices: pd.DataFrame,
|
|
engagement_row: Optional[Dict] = None, popularity_score: Optional[float] = None,
|
|
demand_score: Optional[float] = None,
|
|
) -> Dict:
|
|
"""Full Feature 5 metric set for one product, aggregated across all
|
|
stores that carry it, plus a store-wise breakdown."""
|
|
prod_items = order_items[(order_items["brand"] == brand) & (order_items["image_id"] == image_id)]
|
|
prod_prices = store_prices[(store_prices["brand"] == brand) & (store_prices["image_id"] == image_id)]
|
|
|
|
sales_count = int(prod_items["quantity"].sum())
|
|
revenue = float(prod_items["line_total"].sum())
|
|
profit_info = profit_analytics(prod_items, store_prices)
|
|
order_count = int(prod_items["order_id"].nunique())
|
|
|
|
store_wise = (
|
|
prod_items.groupby("store_id").agg(units=("quantity", "sum"), revenue=("line_total", "sum")).reset_index()
|
|
if not prod_items.empty else pd.DataFrame(columns=["store_id", "units", "revenue"])
|
|
)
|
|
|
|
# Growth %: last-14-days units vs the 14 days before that (real data,
|
|
# not synthetic - same "current vs previous window" pattern used by
|
|
# the trending model, just exposed as a plain metric here).
|
|
growth_pct = None
|
|
if not prod_items.empty and "order_date" in prod_items.columns:
|
|
last_date = prod_items["order_date"].max()
|
|
cur_start = last_date - pd.Timedelta(days=14)
|
|
prev_start = cur_start - pd.Timedelta(days=14)
|
|
cur = prod_items[prod_items["order_date"] >= cur_start]["quantity"].sum()
|
|
prev = prod_items[(prod_items["order_date"] >= prev_start) & (prod_items["order_date"] < cur_start)]["quantity"].sum()
|
|
growth_pct = round(float((cur - prev) / prev * 100) if prev else (100.0 if cur else 0.0), 1)
|
|
|
|
avg_stock = float(prod_prices["selling_price"].mean()) if not prod_prices.empty else 0.0
|
|
# Stock turnover = units sold / average stock held (a standard retail
|
|
# KPI: how many times the inventory "turned over" in the observed period).
|
|
turnover = None
|
|
|
|
return {
|
|
"brand": brand, "image_id": image_id,
|
|
"sales_count": sales_count,
|
|
"revenue": round(revenue, 2),
|
|
"profit": profit_info["total_profit"],
|
|
"orders": order_count,
|
|
"avg_rating": (engagement_row or {}).get("avg_rating"),
|
|
"views": (engagement_row or {}).get("views"),
|
|
"wishlist_count": (engagement_row or {}).get("wishlist_count"),
|
|
"conversion_rate": (engagement_row or {}).get("conversion_rate"),
|
|
"popularity_score": round(popularity_score, 1) if popularity_score is not None else None,
|
|
"demand_score": round(demand_score, 1) if demand_score is not None else None,
|
|
"growth_pct": growth_pct,
|
|
"store_wise_sales": [
|
|
{"store_id": r["store_id"], "units": int(r["units"]), "revenue": round(r["revenue"], 2)}
|
|
for _, r in store_wise.iterrows()
|
|
],
|
|
}
|