""" Polleo Stock & Cash Conversion Cycle (CCC) Analysis ==================================================== Goal (management deck): split warehouse stock into stock that **earns** (fast turn + good margin yield) versus stock that **traps capital** (slow / dead). Quantify the capital we could release by clearing the trapped portion and how much margin we could redeploy into more of the high-yield items. Reads live from PostgreSQL (read-only), produces a single Excel workbook with seven sheets plus charts. Console summary mirrors the workbook. Health classification (per SKU, based on weeks-of-stock over the observed sales window): * STAR — WOS < 8w AND positive sales → fast turn, candidate to expand * HEALTHY — WOS 8-16w AND positive sales → normal, maintain * SLOW — WOS 16-52w → capital getting trapped * DEAD — WOS > 52w OR zero sales → liquidate / write-off Margin yield (€ annual margin per € stock) is computed from RUC over the same window and annualized — gives an ROI-on-inventory metric. Assumptions documented at top of "Executive Summary" sheet. No DB writes. """ from __future__ import annotations import sys from datetime import datetime, date from pathlib import Path import pandas as pd import numpy as np from sqlalchemy import text if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") from db.connection import get_engine, is_db_available OUT_DIR = Path(__file__).resolve().parent TODAY = date.today() OUT_XLSX = OUT_DIR / f"polleo_stock_analysis_{TODAY.strftime('%Y%m%d')}.xlsx" # --- assumptions ---------------------------------------------------------- COGS_WINDOW_DAYS = 91 # ~13 weeks; clipped to actual data span DSO_ASSUMED = 30 DPO_ASSUMED = 45 # Health bucket cutoffs (weeks of stock) WOS_STAR_MAX = 8 # < this = STAR WOS_HEALTHY_MAX = 16 # < this = HEALTHY WOS_SLOW_MAX = 52 # < this = SLOW; >= this = DEAD # Dead also includes zero-sales SKUs regardless of WOS. HEALTH_ORDER = ["STAR", "HEALTHY", "SLOW", "DEAD"] HEALTH_FILL = { "STAR": "C6EFCE", # green "HEALTHY": "DDEBF7", # light blue "SLOW": "FFE699", # amber "DEAD": "F8CBAD", # red } # --------------------------------------------------------------------------- # Data loading # --------------------------------------------------------------------------- def load_data(engine): """Pull warehouse stock + costs + sales/RUC over window + tier/category.""" stock_q = text(""" SELECT p.sku, p.name AS sku_name, COALESCE(dc.name, '(no category)') AS category, sp.tier AS tier, sc.stock_qty AS qty_on_hand, ec.cost_price AS cost_price FROM erp_stock_current sc JOIN dim_stores ds ON sc.store_id = ds.id JOIN dim_products p ON sc.product_id = p.id LEFT JOIN dim_categories dc ON p.category_id = dc.id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN erp_costs ec ON ec.product_id = p.id WHERE ds.is_warehouse = TRUE """) stock = pd.read_sql(stock_q, engine) # Actual transaction date span span = pd.read_sql( "SELECT MIN(transaction_date) AS min_d, MAX(transaction_date) AS max_d " "FROM erp_transactions", engine, ).iloc[0] max_d = pd.Timestamp(span["max_d"]).date() min_d = pd.Timestamp(span["min_d"]).date() window_start = max(max_d - pd.Timedelta(days=COGS_WINDOW_DAYS - 1), min_d) actual_days = (max_d - window_start).days + 1 # Sales + COGS + RUC margin over window sales_q = text(""" SELECT p.sku AS sku, SUM(GREATEST(t.quantity, 0)) AS qty_sold, SUM(CASE WHEN ec.cost_price IS NOT NULL THEN GREATEST(t.quantity,0) * ec.cost_price ELSE 0 END) AS cogs_eur, SUM(GREATEST(t.ruc_eur, 0)) AS ruc_eur, SUM(GREATEST(t.total_value, 0)) AS revenue_eur FROM erp_transactions t JOIN dim_products p ON t.product_id = p.id LEFT JOIN erp_costs ec ON ec.product_id = p.id LEFT JOIN lookup_channel_map cm ON t.channel_map_id = cm.id WHERE t.transaction_date BETWEEN :wstart AND :wend AND cm.channel IN ('retail', 'webshop', 'wholesale') GROUP BY p.sku """) sales = pd.read_sql( sales_q, engine, params={"wstart": window_start, "wend": max_d} ) # Per-week per-SKU for run-rate (excluding promo weeks) weekly_q = text(""" SELECT p.sku AS sku, EXTRACT(ISOYEAR FROM t.transaction_date)::INT AS year, EXTRACT(WEEK FROM t.transaction_date)::INT AS week, SUM(GREATEST(t.quantity, 0)) AS qty FROM erp_transactions t JOIN dim_products p ON t.product_id = p.id LEFT JOIN lookup_channel_map cm ON t.channel_map_id = cm.id WHERE t.transaction_date BETWEEN :wstart AND :wend AND cm.channel IN ('retail', 'webshop', 'wholesale') GROUP BY p.sku, year, week """) weekly = pd.read_sql( weekly_q, engine, params={"wstart": window_start, "wend": max_d} ) promo_q = text(""" SELECT p.sku AS sku, pw.year, pw.week FROM erp_promo_weeks pw JOIN dim_products p ON pw.product_id = p.id WHERE pw.is_erp_promo = TRUE """) try: promo_weeks = pd.read_sql(promo_q, engine) except Exception: promo_weeks = pd.DataFrame(columns=["sku", "year", "week"]) return { "stock": stock, "sales": sales, "weekly": weekly, "promo_weeks": promo_weeks, "window_start": window_start, "window_end": max_d, "actual_days": actual_days, "data_min": min_d, "data_max": max_d, } # --------------------------------------------------------------------------- # Metric computation # --------------------------------------------------------------------------- def classify_health(wos: float, qty_sold: float) -> str: if qty_sold <= 0: return "DEAD" if not np.isfinite(wos): return "DEAD" if wos < WOS_STAR_MAX: return "STAR" if wos < WOS_HEALTHY_MAX: return "HEALTHY" if wos < WOS_SLOW_MAX: return "SLOW" return "DEAD" def compute_metrics(data: dict) -> pd.DataFrame: stock = data["stock"].copy() sales = data["sales"] weekly = data["weekly"] promo_weeks = data["promo_weeks"] actual_days = data["actual_days"] # tier normalization stock["tier"] = stock["tier"].fillna("Unplanned") stock["tier_display"] = stock["tier"].replace({ "01 GOLD": "Gold", "02 SILVER": "Silver", "03 BRONZE": "Bronze", }) stock["tier_display"] = stock["tier_display"].where( stock["tier_display"].isin(["Gold", "Silver", "Bronze"]), "Unplanned", ) stock["cost_missing_flag"] = stock["cost_price"].isna() stock["cost_price"] = stock["cost_price"].fillna(0).astype(float) stock["qty_on_hand"] = stock["qty_on_hand"].astype(float) stock["stock_value"] = stock["qty_on_hand"] * stock["cost_price"] # sales / margin merge sales = sales.copy() for c in ["qty_sold", "cogs_eur", "ruc_eur", "revenue_eur"]: sales[c] = sales[c].astype(float) stock = stock.merge( sales[["sku", "qty_sold", "cogs_eur", "ruc_eur", "revenue_eur"]], on="sku", how="left", ) stock[["qty_sold", "cogs_eur", "ruc_eur", "revenue_eur"]] = ( stock[["qty_sold", "cogs_eur", "ruc_eur", "revenue_eur"]].fillna(0) ) # weekly run-rate (exclude promo weeks) promo_set = set(zip(promo_weeks["sku"], promo_weeks["year"], promo_weeks["week"])) if not weekly.empty: weekly = weekly.copy() weekly["is_promo"] = [ (s, int(y), int(w)) in promo_set for s, y, w in zip(weekly["sku"], weekly["year"], weekly["week"]) ] non_promo = weekly[~weekly["is_promo"]] n_weeks_actual = max( non_promo.drop_duplicates(["year", "week"]).shape[0], 1 ) run_rate = ( non_promo.groupby("sku")["qty"].sum() / n_weeks_actual ).rename("weekly_run_rate") else: run_rate = pd.Series(dtype=float, name="weekly_run_rate") stock = stock.merge(run_rate, on="sku", how="left") stock["weekly_run_rate"] = stock["weekly_run_rate"].fillna(0) stock["weeks_of_stock"] = np.where( stock["weekly_run_rate"] > 0, stock["qty_on_hand"] / stock["weekly_run_rate"], np.inf, ) stock["annual_turns"] = np.where( np.isfinite(stock["weeks_of_stock"]) & (stock["weeks_of_stock"] > 0), 52.0 / stock["weeks_of_stock"], 0.0, ) # DIO per SKU daily_cogs = stock["cogs_eur"] / actual_days stock["dio_days"] = np.where( daily_cogs > 0, stock["stock_value"] / daily_cogs, np.inf, ) # annualized margin & margin yield annualization = 365.0 / actual_days if actual_days else 0 stock["annual_margin_eur"] = stock["ruc_eur"] * annualization stock["annual_revenue_eur"] = stock["revenue_eur"] * annualization stock["margin_yield"] = np.where( stock["stock_value"] > 0, stock["annual_margin_eur"] / stock["stock_value"], np.nan, ) stock["margin_pct"] = np.where( stock["revenue_eur"] > 0, stock["ruc_eur"] / stock["revenue_eur"], np.nan, ) # health classification stock["health"] = [ classify_health(w, q) for w, q in zip(stock["weeks_of_stock"], stock["qty_sold"]) ] # Suggested action per row def action(row): h = row["health"] if h == "STAR": return "Expand — increase reorder point if supply allows" if h == "HEALTHY": return "Maintain — current coverage is right-sized" if h == "SLOW": if row["tier_display"] in ("Gold", "Silver"): return "Stop reorder; review forecast — top-tier shouldn't be slow" return "Stop reorder until WOS<16; consider -10/-20% nudge" # DEAD if row["qty_sold"] == 0: return "Liquidate — outlet/clearance or write-off" return "Liquidate — deep discount or transfer to outlet store" stock["suggested_action"] = stock.apply(action, axis=1) return stock # --------------------------------------------------------------------------- # Aggregates # --------------------------------------------------------------------------- def _safe_div(a, b): return a / b if b else float("inf") def compute_totals(df: pd.DataFrame, data: dict) -> dict: actual_days = data["actual_days"] annualization = 365.0 / actual_days if actual_days else 0 total_stock_value = float(df["stock_value"].sum()) total_cogs = float(df["cogs_eur"].sum()) total_ruc = float(df["ruc_eur"].sum()) total_revenue = float(df["revenue_eur"].sum()) annual_margin = total_ruc * annualization daily_cogs = total_cogs / actual_days if actual_days else 0 overall_dio = _safe_div(total_stock_value, daily_cogs) ccc_proxy = overall_dio + DSO_ASSUMED - DPO_ASSUMED # Per health bucket bucket_rows = [] for h in HEALTH_ORDER: sub = df[df["health"] == h] sv = float(sub["stock_value"].sum()) cg = float(sub["cogs_eur"].sum()) rc = float(sub["ruc_eur"].sum()) rv = float(sub["revenue_eur"].sum()) dc = cg / actual_days if actual_days else 0 dio = _safe_div(sv, dc) avg_wos = sub.loc[np.isfinite(sub["weeks_of_stock"]), "weeks_of_stock"].mean() ann_margin = rc * annualization yld = (ann_margin / sv) if sv > 0 else float("nan") bucket_rows.append({ "health": h, "n_skus": len(sub), "stock_value": sv, "share_pct": sv / total_stock_value if total_stock_value else 0, "cogs_window": cg, "ruc_window": rc, "annual_margin": ann_margin, "margin_yield": yld, "dio_days": dio if np.isfinite(dio) else 9999, "avg_wos": avg_wos if pd.notna(avg_wos) else 0, }) health_table = pd.DataFrame(bucket_rows) # Per-tier health breakdown (value + count matrix) tiers = ["Gold", "Silver", "Bronze", "Unplanned"] health_x_tier_value = pd.DataFrame( 0.0, index=tiers, columns=HEALTH_ORDER ) health_x_tier_count = pd.DataFrame( 0, index=tiers, columns=HEALTH_ORDER ) for t in tiers: for h in HEALTH_ORDER: sub = df[(df["tier_display"] == t) & (df["health"] == h)] health_x_tier_value.loc[t, h] = float(sub["stock_value"].sum()) health_x_tier_count.loc[t, h] = int(len(sub)) # DIO by tier (kept for CCC sheet) tier_rows = [] for t in tiers: sub = df[df["tier_display"] == t] sv = float(sub["stock_value"].sum()) cg = float(sub["cogs_eur"].sum()) dc = cg / actual_days if actual_days else 0 dio = _safe_div(sv, dc) rc = float(sub["ruc_eur"].sum()) tier_rows.append({ "tier_display": t, "stock_value": sv, "cogs_eur": cg, "annual_margin": rc * annualization, "dio_days": dio if np.isfinite(dio) else 9999, }) tier_table = pd.DataFrame(tier_rows) # DIO by category cat_rows = [] for cat, sub in df.groupby("category"): sv = float(sub["stock_value"].sum()) cg = float(sub["cogs_eur"].sum()) dc = cg / actual_days if actual_days else 0 dio = _safe_div(sv, dc) cat_rows.append({ "category": cat, "stock_value": sv, "cogs_eur": cg, "dio_days": dio if np.isfinite(dio) else 9999, }) category_table = pd.DataFrame(cat_rows).sort_values( "stock_value", ascending=False ) # Working vs trapped working_mask = df["health"].isin(["STAR", "HEALTHY"]) trapped_mask = df["health"].isin(["SLOW", "DEAD"]) working_value = float(df.loc[working_mask, "stock_value"].sum()) trapped_value = float(df.loc[trapped_mask, "stock_value"].sum()) working_annual_margin = float( df.loc[working_mask, "ruc_eur"].sum() * annualization ) trapped_annual_margin = float( df.loc[trapped_mask, "ruc_eur"].sum() * annualization ) # Yields (value-weighted averages, not per-SKU medians — avoids tiny-base # SKUs from inflating the figure) working_yield = (working_annual_margin / working_value) if working_value else 0.0 trapped_yield = (trapped_annual_margin / trapped_value) if trapped_value else 0.0 overall_yield = (annual_margin / total_stock_value) if total_stock_value else 0.0 # Redeployment scenarios — what extra annual margin if trapped capital # were redeployed at *X* yield. We show two anchors: # - Realistic floor: redeploy at the OVERALL blended yield (treats the # freed capital as if it joins the average mix) # - Optimistic ceiling: redeploy at the WORKING yield (assumes we only # buy more of the fast-turn stuff) redeployment_floor = trapped_value * overall_yield - trapped_annual_margin redeployment_ceiling = trapped_value * working_yield - trapped_annual_margin # Top liquidation candidates by stock value (DEAD + SLOW) top_liquidate = ( df[trapped_mask] .sort_values("stock_value", ascending=False) .head(30) ) # Top star/healthy by margin yield (to recommend expansion) top_expand = ( df[working_mask & df["margin_yield"].notna() & np.isfinite(df["margin_yield"])] .sort_values(["margin_yield", "stock_value"], ascending=[False, False]) .head(30) ) # action items (narrative bullets) bullets = [] bullets.append( f"€{working_value:,.0f} ({working_value/total_stock_value:.0%}) of stock " f"is WORKING (STAR + HEALTHY) — earns ~€{working_annual_margin:,.0f}/yr in margin " f"(yield {working_yield:.2f} €/€/yr)." ) bullets.append( f"€{trapped_value:,.0f} ({trapped_value/total_stock_value:.0%}) is TRAPPED " f"(SLOW + DEAD) — earns only ~€{trapped_annual_margin:,.0f}/yr " f"(yield {trapped_yield:.2f} €/€/yr — 3x worse return on each €)." ) bullets.append( f"Same € redeployed at the overall blended yield ({overall_yield:.2f}) " f"would add ~€{redeployment_floor:,.0f}/yr extra margin (realistic floor); " f"redeployed at the WORKING-stock yield ({working_yield:.2f}) " f"would add ~€{redeployment_ceiling:,.0f}/yr (optimistic ceiling)." ) dead_val = float(df.loc[df["health"] == "DEAD", "stock_value"].sum()) dead_n = int((df["health"] == "DEAD").sum()) bullets.append( f"Quickest win: €{dead_val:,.0f} in {dead_n:,} DEAD SKUs — " "outlet/clearance or write-off. No future sales expected." ) bullets.append( f"Top 30 liquidation candidates alone hold €" f"{top_liquidate['stock_value'].sum():,.0f} — concentrated action releases " "most of the capital." ) bullets.append( f"Overall DIO {overall_dio:.0f}d (CCC proxy {ccc_proxy:.0f}d). " "Each €1 in stock is locked ~" f"{overall_dio:.0f} days before turning back into cash." ) return { "total_stock_value": total_stock_value, "total_cogs": total_cogs, "total_ruc": total_ruc, "total_revenue": total_revenue, "annual_margin": annual_margin, "annualization": annualization, "overall_dio": overall_dio, "ccc_proxy": ccc_proxy, "health_table": health_table, "tier_table": tier_table, "category_table": category_table, "health_x_tier_value": health_x_tier_value, "health_x_tier_count": health_x_tier_count, "working_value": working_value, "trapped_value": trapped_value, "working_annual_margin": working_annual_margin, "trapped_annual_margin": trapped_annual_margin, "working_yield": working_yield, "trapped_yield": trapped_yield, "overall_yield": overall_yield, "redeployment_floor": redeployment_floor, "redeployment_ceiling": redeployment_ceiling, "top_liquidate": top_liquidate, "top_expand": top_expand, "bullets": bullets, } # --------------------------------------------------------------------------- # Console summary # --------------------------------------------------------------------------- def fmt_eur(x): try: return f"€{x:,.0f}" except Exception: return str(x) def print_summary(data, df, totals): print("\n=== POLLEO STOCK & CCC ANALYSIS ===\n") print("Database connected: polleo_demand @ localhost:5432") print() print("--- Data check ---") print(f" Transactions span: {data['data_min']} -> {data['data_max']}") print(f" COGS window: {data['window_start']} -> {data['window_end']} " f"({data['actual_days']} days)") print(f" Warehouse SKUs in stock: {len(df):,}") print(f" SKUs missing cost price: {int(df['cost_missing_flag'].sum()):,}") print() print("--- Stock totals ---") print(f" Total stock value: {fmt_eur(totals['total_stock_value'])}") print(f" Annual margin (RUC): {fmt_eur(totals['annual_margin'])} (extrapolated from window)") print(f" Overall DIO: {totals['overall_dio']:.1f} days") print(f" CCC proxy: {totals['ccc_proxy']:.1f} days " f"(DIO + {DSO_ASSUMED} DSO - {DPO_ASSUMED} DPO)") print() print("--- Health classification ---") print(f" {'Bucket':<8} {'SKUs':>7} {'Stock value':>14} " f"{'% total':>8} {'Avg WOS':>9} {'Yield €/€/yr':>13}") for _, r in totals["health_table"].iterrows(): yld = r["margin_yield"] yld_s = f"{yld:.2f}" if pd.notna(yld) and np.isfinite(yld) else "n/a" print(f" {r['health']:<8} {int(r['n_skus']):>7,} " f"{fmt_eur(r['stock_value']):>14} " f"{r['share_pct']:>7.1%} " f"{r['avg_wos']:>7.1f}w {yld_s:>10}") print() print("--- Working vs Trapped ---") print(f" WORKING (STAR+HEALTHY): {fmt_eur(totals['working_value'])} stock " f"-> {fmt_eur(totals['working_annual_margin'])}/yr margin " f"(yield {totals['working_yield']:.2f} €/€/yr)") print(f" TRAPPED (SLOW+DEAD): {fmt_eur(totals['trapped_value'])} stock " f"-> {fmt_eur(totals['trapped_annual_margin'])}/yr margin " f"(yield {totals['trapped_yield']:.2f} €/€/yr)") print(f" Overall blended yield: {totals['overall_yield']:.2f} €/€/yr") print(f" Redeployment upside FLOOR (at blended yield): " f"{fmt_eur(totals['redeployment_floor'])}/yr extra margin") print(f" Redeployment upside CEILING (at working yield): " f"{fmt_eur(totals['redeployment_ceiling'])}/yr extra margin") print() print("--- Action Items for Management ---") for b in totals["bullets"]: print(f" • {b}") print() # --------------------------------------------------------------------------- # Excel output # --------------------------------------------------------------------------- def write_excel(df: pd.DataFrame, totals: dict, data: dict, out_path: Path): from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter from openpyxl.formatting.rule import CellIsRule, FormulaRule from openpyxl.chart import BarChart, PieChart, Reference wb = Workbook() H1 = Font(bold=True, size=14, color="FFFFFF") H2 = Font(bold=True, size=11, color="FFFFFF") H3 = Font(bold=True, size=11) FILL_DARK = PatternFill("solid", fgColor="1F4E79") FILL_MID = PatternFill("solid", fgColor="2E75B6") FILL_GRN = PatternFill("solid", fgColor="C6EFCE") FILL_AMB = PatternFill("solid", fgColor="FFE699") FILL_RED = PatternFill("solid", fgColor="F8CBAD") THIN = Side(border_style="thin", color="B4B4B4") BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN) def title(ws, text, row=1, span=6): ws.cell(row=row, column=1, value=text).font = H1 ws.cell(row=row, column=1).fill = FILL_DARK ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=span) ws.cell(row=row, column=1).alignment = Alignment( horizontal="left", vertical="center" ) ws.row_dimensions[row].height = 22 def autosize(ws, max_w=46): for col in ws.columns: try: letter = get_column_letter(col[0].column) except AttributeError: continue length = 10 for c in col: v = c.value if v is None: continue length = max(length, min(max_w, len(str(v)) + 2)) ws.column_dimensions[letter].width = length # ------------------------------------------------------------------- # Sheet 1: Executive Summary # ------------------------------------------------------------------- ws = wb.active ws.title = "Executive Summary" title(ws, "POLLEO SPORT — STOCK HEALTH & CAPITAL EFFICIENCY", span=4) ws["A2"] = ( f"Generated: {datetime.now():%Y-%m-%d %H:%M} " f"Window: {data['window_start']} → {data['window_end']} " f"({data['actual_days']}d) Warehouse: dim_stores.is_warehouse=TRUE" ) ws["A2"].font = Font(italic=True, color="595959") ws.merge_cells("A2:F2") row = 4 ws.cell(row=row, column=1, value="THE STORY").font = H3 row += 1 headline = [ ("Total warehouse stock value", fmt_eur(totals["total_stock_value"])), (" → WORKING (STAR + HEALTHY)", f"{fmt_eur(totals['working_value'])} " f"({totals['working_value']/totals['total_stock_value']:.0%})"), (" → TRAPPED (SLOW + DEAD)", f"{fmt_eur(totals['trapped_value'])} " f"({totals['trapped_value']/totals['total_stock_value']:.0%})"), ("", ""), ("Annual margin from WORKING stock (est.)", f"{fmt_eur(totals['working_annual_margin'])} " f"(yield {totals['working_yield']:.2f} €/€/yr)"), ("Annual margin from TRAPPED stock (est.)", f"{fmt_eur(totals['trapped_annual_margin'])} " f"(yield {totals['trapped_yield']:.2f} €/€/yr)"), ("Overall blended yield", f"{totals['overall_yield']:.2f} €margin / €stock / yr"), ("", ""), ("Redeployment upside — FLOOR (at blended yield)", f"{fmt_eur(totals['redeployment_floor'])} extra annual margin"), ("Redeployment upside — CEILING (at working yield)", f"{fmt_eur(totals['redeployment_ceiling'])} extra annual margin"), ("", ""), ("Overall DIO", f"{totals['overall_dio']:.0f} days"), ("CCC proxy", f"{totals['ccc_proxy']:.0f} days " f"(DIO + {DSO_ASSUMED}d DSO − {DPO_ASSUMED}d DPO assumed)"), ] for label, val in headline: ws.cell(row=row, column=1, value=label).font = Font(bold=True) ws.cell(row=row, column=3, value=val) row += 1 row += 1 ws.cell(row=row, column=1, value="Health bucket overview").font = H3 row += 1 cols = ["Bucket", "SKUs", "Stock value (€)", "% of total", "Avg WOS (weeks)", "Annual margin (€)", "Yield €/€/yr"] for i, c in enumerate(cols, 1): cell = ws.cell(row=row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID row += 1 for _, r in totals["health_table"].iterrows(): ws.cell(row=row, column=1, value=r["health"]) ws.cell(row=row, column=2, value=int(r["n_skus"])) ws.cell(row=row, column=3, value=float(r["stock_value"])) ws.cell(row=row, column=4, value=float(r["share_pct"])) ws.cell(row=row, column=5, value=float(r["avg_wos"])) ws.cell(row=row, column=6, value=float(r["annual_margin"])) ws.cell(row=row, column=7, value=(float(r["margin_yield"]) if pd.notna(r["margin_yield"]) and np.isfinite(r["margin_yield"]) else None)) ws.cell(row=row, column=3).number_format = "#,##0" ws.cell(row=row, column=4).number_format = "0.0%" ws.cell(row=row, column=5).number_format = "0.0" ws.cell(row=row, column=6).number_format = "#,##0" ws.cell(row=row, column=7).number_format = "0.00" ws.cell(row=row, column=1).fill = PatternFill( "solid", fgColor=HEALTH_FILL[r["health"]] ) row += 1 row += 1 ws.cell(row=row, column=1, value="Action Items").font = H3 row += 1 for b in totals["bullets"]: ws.cell(row=row, column=1, value="• " + b) ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=7) row += 1 row += 2 ws.cell(row=row, column=1, value="Assumptions & Notes").font = H3 row += 1 notes = [ "Warehouse scope = dim_stores.is_warehouse=TRUE (central HR warehouse only).", f"COGS / margin window = {data['actual_days']} days " f"({data['window_start']} → {data['window_end']}). " "Annual figures = window × (365 / window days); assumes no seasonality.", "Margin = SUM(GREATEST(ruc_eur, 0)) over the window from ERP transactions.", f"Health buckets: STAR <{WOS_STAR_MAX}w, HEALTHY <{WOS_HEALTHY_MAX}w, " f"SLOW <{WOS_SLOW_MAX}w, DEAD ≥{WOS_SLOW_MAX}w or zero sales.", "Weekly run-rate excludes ERP-flagged promo weeks (erp_promo_weeks.is_erp_promo).", f"DSO {DSO_ASSUMED}d / DPO {DPO_ASSUMED}d are placeholders — " "replace with finance-team actuals before final read-out.", f"{int(df['cost_missing_flag'].sum())} SKUs have no cost_price → " "stock_value treated as €0 (flagged in Raw Data sheet).", "Redeployment upside applies the working-stock median yield to the " "trapped capital — illustrative ceiling, not a forecast.", ] for n in notes: ws.cell(row=row, column=1, value="• " + n) ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=7) row += 1 autosize(ws) # ------------------------------------------------------------------- # Sheet 2: CCC Analysis # ------------------------------------------------------------------- ws = wb.create_sheet("CCC Analysis") title(ws, "Days Inventory Outstanding (DIO) — by Tier & Category", span=4) row = 3 ws.cell(row=row, column=1, value="DIO by Tier").font = H3 row += 1 for i, c in enumerate( ["Tier", "Stock value (€)", "COGS in window (€)", "Annual margin (€)", "DIO (days)"], 1, ): cell = ws.cell(row=row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID row += 1 start = row for _, r in totals["tier_table"].iterrows(): ws.cell(row=row, column=1, value=r["tier_display"]) ws.cell(row=row, column=2, value=float(r["stock_value"])).number_format = "#,##0" ws.cell(row=row, column=3, value=float(r["cogs_eur"])).number_format = "#,##0" ws.cell(row=row, column=4, value=float(r["annual_margin"])).number_format = "#,##0" ws.cell(row=row, column=5, value=float(r["dio_days"])).number_format = "0.0" row += 1 end = row - 1 chart = BarChart() chart.type = "bar" chart.title = "DIO by Tier (days)" data_ref = Reference(ws, min_col=5, min_row=start - 1, max_row=end, max_col=5) cats_ref = Reference(ws, min_col=1, min_row=start, max_row=end) chart.add_data(data_ref, titles_from_data=True) chart.set_categories(cats_ref) chart.height = 7; chart.width = 14 ws.add_chart(chart, f"G{start - 1}") row += 2 ws.cell(row=row, column=1, value="DIO by Category").font = H3 row += 1 for i, c in enumerate( ["Category", "Stock value (€)", "COGS in window (€)", "DIO (days)"], 1 ): cell = ws.cell(row=row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID row += 1 start = row for _, r in totals["category_table"].iterrows(): ws.cell(row=row, column=1, value=r["category"]) ws.cell(row=row, column=2, value=float(r["stock_value"])).number_format = "#,##0" ws.cell(row=row, column=3, value=float(r["cogs_eur"])).number_format = "#,##0" ws.cell(row=row, column=4, value=float(r["dio_days"])).number_format = "0.0" row += 1 end = row - 1 chart2 = BarChart() chart2.type = "bar" chart2.title = "DIO by Category (days)" data_ref = Reference(ws, min_col=4, min_row=start - 1, max_row=end, max_col=4) cats_ref = Reference(ws, min_col=1, min_row=start, max_row=end) chart2.add_data(data_ref, titles_from_data=True) chart2.set_categories(cats_ref) chart2.height = 9; chart2.width = 14 ws.add_chart(chart2, f"G{start - 1}") autosize(ws) # ------------------------------------------------------------------- # Sheet 3: Health Classification (with pie + bar charts) # ------------------------------------------------------------------- ws = wb.create_sheet("Stock Health") title(ws, "Stock Health Classification — Working vs Trapped", span=7) row = 3 ws.cell(row=row, column=1, value="Health bucket summary").font = H3 row += 1 for i, c in enumerate( ["Bucket", "SKUs", "Stock value (€)", "% of total", "Avg WOS", "Annual margin (€)", "Yield €/€/yr"], 1 ): cell = ws.cell(row=row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID row += 1 start = row for _, r in totals["health_table"].iterrows(): ws.cell(row=row, column=1, value=r["health"]).fill = PatternFill( "solid", fgColor=HEALTH_FILL[r["health"]]) ws.cell(row=row, column=2, value=int(r["n_skus"])) ws.cell(row=row, column=3, value=float(r["stock_value"])).number_format = "#,##0" ws.cell(row=row, column=4, value=float(r["share_pct"])).number_format = "0.0%" ws.cell(row=row, column=5, value=float(r["avg_wos"])).number_format = "0.0" ws.cell(row=row, column=6, value=float(r["annual_margin"])).number_format = "#,##0" ws.cell(row=row, column=7, value=(float(r["margin_yield"]) if pd.notna(r["margin_yield"]) and np.isfinite(r["margin_yield"]) else None)) ws.cell(row=row, column=7).number_format = "0.00" row += 1 end = row - 1 # Pie chart of stock value by bucket pie = PieChart() pie.title = "Stock value by health bucket" data_ref = Reference(ws, min_col=3, min_row=start - 1, max_row=end, max_col=3) cats_ref = Reference(ws, min_col=1, min_row=start, max_row=end) pie.add_data(data_ref, titles_from_data=True) pie.set_categories(cats_ref) pie.height = 9; pie.width = 12 ws.add_chart(pie, f"I{start - 1}") # Bar chart of annual margin by bucket bar = BarChart() bar.type = "col" bar.title = "Annual margin by health bucket (€)" data_ref = Reference(ws, min_col=6, min_row=start - 1, max_row=end, max_col=6) bar.add_data(data_ref, titles_from_data=True) bar.set_categories(cats_ref) bar.height = 9; bar.width = 12 ws.add_chart(bar, f"I{start + 12}") # Tier × Health value matrix row = end + 3 ws.cell(row=row, column=1, value="Tier × Health — Stock value (€)").font = H3 row += 1 ws.cell(row=row, column=1, value="Tier \\ Health").font = H2 ws.cell(row=row, column=1).fill = FILL_MID for j, h in enumerate(HEALTH_ORDER, 2): c = ws.cell(row=row, column=j, value=h) c.font = H2; c.fill = PatternFill("solid", fgColor=HEALTH_FILL[h]) row += 1 for tier in ["Gold", "Silver", "Bronze", "Unplanned"]: ws.cell(row=row, column=1, value=tier).font = Font(bold=True) for j, h in enumerate(HEALTH_ORDER, 2): v = float(totals["health_x_tier_value"].loc[tier, h]) cell = ws.cell(row=row, column=j, value=v) cell.number_format = "#,##0" if h in ("SLOW", "DEAD"): cell.fill = FILL_RED row += 1 row += 2 ws.cell(row=row, column=1, value="Tier × Health — SKU count").font = H3 row += 1 ws.cell(row=row, column=1, value="Tier \\ Health").font = H2 ws.cell(row=row, column=1).fill = FILL_MID for j, h in enumerate(HEALTH_ORDER, 2): c = ws.cell(row=row, column=j, value=h) c.font = H2; c.fill = PatternFill("solid", fgColor=HEALTH_FILL[h]) row += 1 for tier in ["Gold", "Silver", "Bronze", "Unplanned"]: ws.cell(row=row, column=1, value=tier).font = Font(bold=True) for j, h in enumerate(HEALTH_ORDER, 2): n = int(totals["health_x_tier_count"].loc[tier, h]) cell = ws.cell(row=row, column=j, value=n) cell.number_format = "#,##0" if h in ("SLOW", "DEAD"): cell.fill = FILL_RED row += 1 autosize(ws) # ------------------------------------------------------------------- # Sheet 4: KEEP & EXPAND (STAR + HEALTHY, ranked by yield) # ------------------------------------------------------------------- ws = wb.create_sheet("Keep & Expand") title(ws, "Keep & Expand — STAR + HEALTHY SKUs (sorted by margin yield)", span=11) keep = df[df["health"].isin(["STAR", "HEALTHY"])].copy() keep = keep.sort_values( ["margin_yield", "annual_margin_eur"], ascending=[False, False], na_position="last", ) cols = [ "sku", "sku_name", "category", "tier_display", "health", "qty_on_hand", "stock_value", "weeks_of_stock", "annual_turns", "annual_margin_eur", "margin_yield", "suggested_action", ] header_row = 3 for i, c in enumerate(cols, 1): cell = ws.cell(row=header_row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID for i, r in enumerate(keep.itertuples(index=False), start=header_row + 1): ws.cell(row=i, column=1, value=r.sku) ws.cell(row=i, column=2, value=r.sku_name) ws.cell(row=i, column=3, value=r.category) ws.cell(row=i, column=4, value=r.tier_display) ws.cell(row=i, column=5, value=r.health) ws.cell(row=i, column=6, value=float(r.qty_on_hand)) ws.cell(row=i, column=7, value=float(r.stock_value)) ws.cell(row=i, column=8, value=(None if not np.isfinite(r.weeks_of_stock) else float(r.weeks_of_stock))) ws.cell(row=i, column=9, value=float(r.annual_turns)) ws.cell(row=i, column=10, value=float(r.annual_margin_eur)) ws.cell(row=i, column=11, value=(None if pd.isna(r.margin_yield) or not np.isfinite(r.margin_yield) else float(r.margin_yield))) ws.cell(row=i, column=12, value=r.suggested_action) # bucket color in column 5 ws.cell(row=i, column=5).fill = PatternFill( "solid", fgColor=HEALTH_FILL[r.health]) last_row = header_row + len(keep) for col_idx, fmt in [(6, "#,##0.0"), (7, "#,##0"), (8, "0.0"), (9, "0.00"), (10, "#,##0"), (11, "0.00")]: for r in range(header_row + 1, last_row + 1): ws.cell(row=r, column=col_idx).number_format = fmt ws.freeze_panes = ws.cell(row=header_row + 1, column=1) autosize(ws) # ------------------------------------------------------------------- # Sheet 5: LIQUIDATE (SLOW + DEAD, ranked by value) # ------------------------------------------------------------------- ws = wb.create_sheet("Liquidate") title(ws, "Liquidate — SLOW + DEAD SKUs (sorted by stock value)", span=11) liq = df[df["health"].isin(["SLOW", "DEAD"])].copy() liq = liq.sort_values("stock_value", ascending=False) cols = [ "sku", "sku_name", "category", "tier_display", "health", "qty_on_hand", "stock_value", "weeks_of_stock", "qty_sold_window", "annual_margin_eur", "margin_yield", "suggested_action", ] header_row = 3 for i, c in enumerate(cols, 1): cell = ws.cell(row=header_row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID for i, r in enumerate(liq.itertuples(index=False), start=header_row + 1): ws.cell(row=i, column=1, value=r.sku) ws.cell(row=i, column=2, value=r.sku_name) ws.cell(row=i, column=3, value=r.category) ws.cell(row=i, column=4, value=r.tier_display) ws.cell(row=i, column=5, value=r.health) ws.cell(row=i, column=6, value=float(r.qty_on_hand)) ws.cell(row=i, column=7, value=float(r.stock_value)) ws.cell(row=i, column=8, value=(None if not np.isfinite(r.weeks_of_stock) else float(r.weeks_of_stock))) ws.cell(row=i, column=9, value=float(r.qty_sold)) ws.cell(row=i, column=10, value=float(r.annual_margin_eur)) ws.cell(row=i, column=11, value=(None if pd.isna(r.margin_yield) or not np.isfinite(r.margin_yield) else float(r.margin_yield))) ws.cell(row=i, column=12, value=r.suggested_action) ws.cell(row=i, column=5).fill = PatternFill( "solid", fgColor=HEALTH_FILL[r.health]) last_row = header_row + len(liq) for col_idx, fmt in [(6, "#,##0.0"), (7, "#,##0"), (8, "0.0"), (9, "#,##0.0"), (10, "#,##0"), (11, "0.00")]: for r in range(header_row + 1, last_row + 1): ws.cell(row=r, column=col_idx).number_format = fmt # Summary row at bottom summary_row = last_row + 2 ws.cell(row=summary_row, column=1, value="TOTAL capital release if liquidated").font = H3 ws.cell(row=summary_row, column=7, value=float(liq["stock_value"].sum())).number_format = "#,##0" summary_row += 1 ws.cell(row=summary_row, column=1, value="Annual margin currently earned on this stock").font = Font(bold=True) ws.cell(row=summary_row, column=10, value=float(liq["annual_margin_eur"].sum())).number_format = "#,##0" ws.freeze_panes = ws.cell(row=header_row + 1, column=1) autosize(ws) # ------------------------------------------------------------------- # Sheet 6: Top 30 picks (Liquidate + Expand) # ------------------------------------------------------------------- ws = wb.create_sheet("Top 30 Actions") title(ws, "Top 30 SKUs to LIQUIDATE & Top 30 to EXPAND", span=10) row = 3 ws.cell(row=row, column=1, value="TOP 30 TO LIQUIDATE — biggest trapped capital").font = H3 row += 1 for i, c in enumerate( ["sku", "sku_name", "category", "tier_display", "health", "stock_value", "weeks_of_stock", "annual_margin_eur", "suggested_action"], 1 ): cell = ws.cell(row=row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID row += 1 for r in totals["top_liquidate"].itertuples(index=False): ws.cell(row=row, column=1, value=r.sku) ws.cell(row=row, column=2, value=r.sku_name) ws.cell(row=row, column=3, value=r.category) ws.cell(row=row, column=4, value=r.tier_display) ws.cell(row=row, column=5, value=r.health).fill = PatternFill( "solid", fgColor=HEALTH_FILL[r.health]) ws.cell(row=row, column=6, value=float(r.stock_value)).number_format = "#,##0" ws.cell(row=row, column=7, value=(None if not np.isfinite(r.weeks_of_stock) else float(r.weeks_of_stock))).number_format = "0.0" ws.cell(row=row, column=8, value=float(r.annual_margin_eur)).number_format = "#,##0" ws.cell(row=row, column=9, value=r.suggested_action) row += 1 row += 2 ws.cell(row=row, column=1, value="TOP 30 TO EXPAND — highest yield STAR/HEALTHY").font = H3 row += 1 for i, c in enumerate( ["sku", "sku_name", "category", "tier_display", "health", "stock_value", "weeks_of_stock", "annual_margin_eur", "margin_yield"], 1 ): cell = ws.cell(row=row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID row += 1 for r in totals["top_expand"].itertuples(index=False): ws.cell(row=row, column=1, value=r.sku) ws.cell(row=row, column=2, value=r.sku_name) ws.cell(row=row, column=3, value=r.category) ws.cell(row=row, column=4, value=r.tier_display) ws.cell(row=row, column=5, value=r.health).fill = PatternFill( "solid", fgColor=HEALTH_FILL[r.health]) ws.cell(row=row, column=6, value=float(r.stock_value)).number_format = "#,##0" ws.cell(row=row, column=7, value=(None if not np.isfinite(r.weeks_of_stock) else float(r.weeks_of_stock))).number_format = "0.0" ws.cell(row=row, column=8, value=float(r.annual_margin_eur)).number_format = "#,##0" ws.cell(row=row, column=9, value=(None if pd.isna(r.margin_yield) or not np.isfinite(r.margin_yield) else float(r.margin_yield))).number_format = "0.00" row += 1 autosize(ws) # ------------------------------------------------------------------- # Sheet 7: Raw Data # ------------------------------------------------------------------- ws = wb.create_sheet("Raw Data") title(ws, "Raw per-SKU data (warehouse stock universe)", span=18) cols = [ "sku", "sku_name", "category", "tier", "tier_display", "health", "qty_on_hand", "cost_price", "cost_missing_flag", "stock_value", "qty_sold", "cogs_eur", "ruc_eur", "weekly_run_rate", "weeks_of_stock", "annual_turns", "dio_days", "annual_margin_eur", "margin_yield", "suggested_action", ] header_row = 3 for i, c in enumerate(cols, 1): cell = ws.cell(row=header_row, column=i, value=c) cell.font = H2; cell.fill = FILL_MID for i, r in enumerate(df.itertuples(index=False), start=header_row + 1): ws.cell(row=i, column=1, value=r.sku) ws.cell(row=i, column=2, value=r.sku_name) ws.cell(row=i, column=3, value=r.category) ws.cell(row=i, column=4, value=r.tier) ws.cell(row=i, column=5, value=r.tier_display) ws.cell(row=i, column=6, value=r.health) ws.cell(row=i, column=6).fill = PatternFill( "solid", fgColor=HEALTH_FILL[r.health]) ws.cell(row=i, column=7, value=float(r.qty_on_hand)) ws.cell(row=i, column=8, value=float(r.cost_price)) ws.cell(row=i, column=9, value=bool(r.cost_missing_flag)) ws.cell(row=i, column=10, value=float(r.stock_value)) ws.cell(row=i, column=11, value=float(r.qty_sold)) ws.cell(row=i, column=12, value=float(r.cogs_eur)) ws.cell(row=i, column=13, value=float(r.ruc_eur)) ws.cell(row=i, column=14, value=float(r.weekly_run_rate)) ws.cell(row=i, column=15, value=(None if not np.isfinite(r.weeks_of_stock) else float(r.weeks_of_stock))) ws.cell(row=i, column=16, value=float(r.annual_turns)) ws.cell(row=i, column=17, value=(None if not np.isfinite(r.dio_days) else float(r.dio_days))) ws.cell(row=i, column=18, value=float(r.annual_margin_eur)) ws.cell(row=i, column=19, value=(None if pd.isna(r.margin_yield) or not np.isfinite(r.margin_yield) else float(r.margin_yield))) ws.cell(row=i, column=20, value=r.suggested_action) ws.freeze_panes = ws.cell(row=header_row + 1, column=1) autosize(ws) wb.save(out_path) # --------------------------------------------------------------------------- # main # --------------------------------------------------------------------------- def main(): if not is_db_available(): print("ERROR: PostgreSQL not reachable on localhost:5432 — " "is docker compose up?", file=sys.stderr) sys.exit(1) engine = get_engine() print("Loading data…") data = load_data(engine) if data["stock"].empty: print("WARNING: erp_stock_current returned no warehouse rows. " "Aborting.", file=sys.stderr) sys.exit(2) print("Computing metrics…") df = compute_metrics(data) print("Aggregating totals…") totals = compute_totals(df, data) print_summary(data, df, totals) print(f"Writing Excel: {OUT_XLSX} …") write_excel(df, totals, data, OUT_XLSX) print(f"Excel saved: {OUT_XLSX}") if __name__ == "__main__": main()