"""Equipment/durable promo analysis — companion to promo_calc_w22_w26_actual.py. Computes equipment-specific promo metrics for low-frequency durable SKUs (Theragun, LARQ, LUMI, SafeJawz, Aquaphor, Atleticore, FIT, etc.) where the FMCG baseline x uplift bridge breaks down. Five metric sheets + portfolio scorecard: 1. Margin per transaction (unit margin on normal vs. observed promo price) 2. Basket attach rate -> N/A unless order_id is present 3. Pull-forward (pre/during/post 8w around each ERP promo period) 4. Stock coverage (weeks of cover = on_hand / avg_weekly_sales_12w) 5. Customer acquisition -> N/A unless customer_id is present 6. Equipment scorecard -> per-group rollup with recommendation CLI: python equipment_promo_analysis.py --skus batch2_skus.txt [--out data/equipment_promo_analysis.xlsx] """ from __future__ import annotations import argparse import sys from pathlib import Path ROOT = Path(__file__).resolve().parent sys.stdout.reconfigure(encoding="utf-8") # type: ignore[attr-defined] import numpy as np import pandas as pd DATA_DIR = ROOT / "data" PULL_WINDOW = 8 # weeks pre/post around each promo period RECENT_WEEKS = 12 # avg-weekly-sales lookback for stock cover TX_LOOKBACK_WEEKS = 52 # margin per transaction window EQUIPMENT_GROUPS: dict[str, callable] = { "LARQ boca PureVis 500ml": lambda s, n: s.startswith("LRQ") and "purevis" in n.lower() and "500" in n, "LARQ boca PureVis 740ml": lambda s, n: s.startswith("LRQ") and "purevis" in n.lower() and "740" in n, "LARQ boca s filterom": lambda s, n: s.startswith("LRQ") and "filter" in n.lower() and "boc" in n.lower(), "LARQ Twist Top": lambda s, n: s.startswith("LRQ") and "twist" in n.lower(), "LARQ Flip Top": lambda s, n: s.startswith("LRQ") and "flip" in n.lower(), "LARQ Swig Top": lambda s, n: s.startswith("LRQ") and "swig" in n.lower(), "LARQ vrč PureVis": lambda s, n: s.startswith("LRQ") and "vrč" in n.lower(), "LARQ dodaci": lambda s, n: s.startswith("LRQ"), "Kineziološka traka": lambda s, n: s.startswith("GMS"), "Theragun PRO PLUS": lambda s, n: s == "THG03001", "Theragun Prime": lambda s, n: s.startswith("THG") and "prime" in n.lower(), "Theragun Elite": lambda s, n: s.startswith("THG") and "elite" in n.lower(), "Theragun Relief": lambda s, n: s.startswith("THG") and "relief" in n.lower(), "Theragun Sense": lambda s, n: s.startswith("THG") and "sense" in n.lower(), "Theragun Mini": lambda s, n: s.startswith("THG") and "mini" in n.lower(), "Theragun ostalo": lambda s, n: s.startswith("THG"), "LUMI Ice Bath": lambda s, n: s.startswith("LMT"), "Masažni pištolj mini": lambda s, n: s == "TNT25825", "Aquaphor City boca": lambda s, n: s.startswith("AQP") and "zamjenski" not in n.lower(), "Aquaphor zamjenski uložak": lambda s, n: s.startswith("AQP") and "zamjenski" in n.lower(), "SafeJawz Extro": lambda s, n: s.startswith("SJW") and "extro" in n.lower(), "SafeJawz Nitro": lambda s, n: s.startswith("SJW") and "nitro" in n.lower(), "SafeJawz Intro": lambda s, n: s.startswith("SJW") and "intro" in n.lower(), "SafeJawz Ortho": lambda s, n: s.startswith("SJW") and "ortho" in n.lower(), "SafeJawz Marvel": lambda s, n: s.startswith("SJW") and "marvel" in n.lower(), "SafeJawz Lip Protector": lambda s, n: s.startswith("SJW") and "lip" in n.lower(), "SafeJawz dodaci": lambda s, n: s.startswith("SJW"), "Bučica gumirana": lambda s, n: s.startswith("ATC") and "bučica" in n.lower(), "Girja gumirana": lambda s, n: s.startswith("ATC") and "girja" in n.lower(), "Trening prsluk": lambda s, n: s.startswith("ATC") and "prsluk" in n.lower(), "Traka za trčanje": lambda s, n: s.startswith("TNT") and s != "TNT25825", "Tatami strunjača": lambda s, n: s.startswith("FIT"), } def assign_group(sku: str, name: str) -> str: for label, pred in EQUIPMENT_GROUPS.items(): try: if pred(sku, name or ""): return label except Exception: continue return "Other" # -------------------------------------------------------------------------- # Loaders # -------------------------------------------------------------------------- def _csv(name: str) -> pd.DataFrame: p = DATA_DIR / name return pd.read_csv(p, low_memory=False) if p.exists() else pd.DataFrame() def _read_sku_list(path: Path) -> list[str]: raw = path.read_text(encoding="utf-8-sig").splitlines() out, seen = [], set() for line in raw: s = line.strip().lstrip("").strip() if not s or s.startswith("#") or s.lower() == "sku": continue sku = s.split(",", 1)[0].strip().strip('"').strip("'") if sku and sku not in seen: seen.add(sku) out.append(sku) return out def _resolve_names(skus: list[str], plan_df: pd.DataFrame) -> dict[str, str]: out: dict[str, str] = {} if not plan_df.empty and "sku" in plan_df.columns and "name" in plan_df.columns: out.update(dict(zip(plan_df["sku"], plan_df["name"].fillna("")))) sd_path = DATA_DIR / "sales_detailed.csv" if sd_path.exists(): sd = pd.read_csv(sd_path, usecols=["sku", "naziv"], low_memory=False) sd = sd.dropna(subset=["sku"]).drop_duplicates("sku") for s, n in zip(sd["sku"], sd["naziv"]): out.setdefault(s, str(n)) sif_path = DATA_DIR / "SifrarnikArtikala.xlsx" if sif_path.exists(): try: sif = pd.read_excel(sif_path, usecols=["Šifra", "Naziv artikla/usluge"]) sif.columns = ["sku", "naziv"] sif = sif.dropna(subset=["sku"]).drop_duplicates("sku") for s, n in zip(sif["sku"], sif["naziv"]): out.setdefault(s, str(n)) except Exception: pass return {s: out.get(s, "") for s in skus} # -------------------------------------------------------------------------- # yw helpers # -------------------------------------------------------------------------- def yw_offset(yw: int, delta: int) -> int: year, week = divmod(yw, 100) week += delta while week < 1: year -= 1 week += 52 while week > 52: year += 1 week -= 52 return year * 100 + week def yws_range(start_yw: int, end_yw: int, cap: int = 200) -> list[int]: out, cur, n = [], start_yw, 0 while cur <= end_yw and n < cap: out.append(cur) cur = yw_offset(cur, 1) n += 1 return out # -------------------------------------------------------------------------- # Metric 1: Margin per transaction # -------------------------------------------------------------------------- def margin_per_transaction( skus: list[str], sales: pd.DataFrame, prices: pd.DataFrame, costs: pd.DataFrame, name_map: dict[str, str], group_map: dict[str, str], ) -> pd.DataFrame: price_map = ( dict(zip(prices["sku"], prices["normal_retail_ppp"])) if {"sku", "normal_retail_ppp"}.issubset(prices.columns) else {} ) cost_map = ( dict(zip(costs["sku"], costs["cost_price"])) if {"sku", "cost_price"}.issubset(costs.columns) else {} ) max_year = int(sales["year"].max()) max_week = int(sales.loc[sales["year"] == max_year, "week"].max()) end_yw = max_year * 100 + max_week start_yw = end_yw for _ in range(TX_LOOKBACK_WEEKS - 1): start_yw = yw_offset(start_yw, -1) sub = sales[sales["sku"].isin(skus)].copy() sub["yw"] = sub["year"].astype(int) * 100 + sub["week"].astype(int) in_window = sub["yw"].apply(lambda v: start_yw <= v <= end_yw) sub = sub[in_window].copy() rows = [] for sku in skus: normal_price = float(price_map.get(sku, 0) or 0) cost = float(cost_map.get(sku, 0) or 0) margin_unit = normal_price - cost margin_pct = (margin_unit / normal_price * 100) if normal_price > 0 else None ssub = sub[sub["sku"] == sku] promo_rows = ssub[(ssub["is_retail_promo"] == 1) & (ssub["avg_ppp_retail"].fillna(0) > 0)] promo_qty_w = promo_rows["qty_retail"].fillna(0) promo_price_w = promo_rows["avg_ppp_retail"].fillna(0) if promo_qty_w.sum() > 0: promo_price_obs = float((promo_price_w * promo_qty_w).sum() / promo_qty_w.sum()) elif not promo_rows.empty: promo_price_obs = float(promo_price_w.mean()) else: promo_price_obs = None promo_margin_unit = (promo_price_obs - cost) if (promo_price_obs is not None and cost > 0) else None margin_erosion_unit = (margin_unit - promo_margin_unit) if (promo_margin_unit is not None and margin_unit) else None margin_erosion_pct = (margin_erosion_unit / margin_unit * 100) if (margin_erosion_unit is not None and margin_unit) else None rows.append({ "SKU": sku, "Naziv": name_map.get(sku, ""), "Grupa": group_map.get(sku, "Other"), "Normal price (EUR)": round(normal_price, 2) if normal_price else None, "Cost (EUR)": round(cost, 4) if cost else None, "Normal margin/unit (EUR)": round(margin_unit, 2) if normal_price else None, "Normal margin %": round(margin_pct, 1) if margin_pct is not None else None, "Promo price observed (EUR)": round(promo_price_obs, 2) if promo_price_obs is not None else None, "Promo margin/unit (EUR)": round(promo_margin_unit, 2) if promo_margin_unit is not None else None, "Margin erosion/unit (EUR)": round(margin_erosion_unit, 2) if margin_erosion_unit is not None else None, "Margin erosion %": round(margin_erosion_pct, 1) if margin_erosion_pct is not None else None, "Promo weeks observed": int(len(promo_rows)), }) df = pd.DataFrame(rows) group = df.groupby("Grupa", as_index=False).agg( n_skus=("SKU", "count"), median_normal_price=("Normal price (EUR)", "median"), median_normal_margin_unit=("Normal margin/unit (EUR)", "median"), median_promo_price=("Promo price observed (EUR)", "median"), median_promo_margin_unit=("Promo margin/unit (EUR)", "median"), median_margin_erosion_pct=("Margin erosion %", "median"), ) return df, group # -------------------------------------------------------------------------- # Metric 3: Pull-forward # -------------------------------------------------------------------------- def pull_forward( skus: list[str], sales: pd.DataFrame, erp: pd.DataFrame, name_map: dict[str, str], group_map: dict[str, str], ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: sc = sales[sales["sku"].isin(skus)].copy() sc["yw"] = sc["year"].astype(int) * 100 + sc["week"].astype(int) sc["units"] = sc["qty_total"].fillna(0) sc["ruc"] = sc["ruc_total"].fillna(0) sc_idx = sc.set_index(["sku", "yw"])[["units", "ruc"]].sort_index() pc = erp[erp["sku"].isin(skus)].copy() pc["yw"] = pc["year"].astype(int) * 100 + pc["week"].astype(int) pc = pc.sort_values(["sku", "promo_types", "yw"]) periods = [] for (sku, ptype), g in pc.groupby(["sku", "promo_types"]): yws = g["yw"].tolist() if not yws: continue start = prev = yws[0] for cur in yws[1:]: if cur == yw_offset(prev, 1): prev = cur continue periods.append((sku, ptype, start, prev)) start = prev = cur periods.append((sku, ptype, start, prev)) def collect(sku, s_yw, e_yw, label): weeks = yws_range(s_yw, e_yw) units = ruc = 0.0 n_with = 0 for yw in weeks: try: r = sc_idx.loc[(sku, yw)] units += float(r["units"]) ruc += float(r["ruc"]) n_with += 1 except KeyError: pass return { "period": label, "weeks_span": len(weeks), "weeks_with_data": n_with, "total_units": units, "avg_units_per_week": units / len(weeks) if weeks else 0.0, "total_ruc": ruc, "avg_ruc_per_week": ruc / len(weeks) if weeks else 0.0, } long_rows = [] for sku, ptype, s_yw, e_yw in periods: pre = collect(sku, yw_offset(s_yw, -PULL_WINDOW), yw_offset(s_yw, -1), "pre_promo") during = collect(sku, s_yw, e_yw, "during_promo") post = collect(sku, yw_offset(e_yw, 1), yw_offset(e_yw, PULL_WINDOW), "post_promo") for d in (pre, during, post): d.update({ "sku": sku, "promo_types": ptype, "start_yw": s_yw, "end_yw": e_yw, "ruc_reliable": (s_yw // 100) >= 2026, "naziv": name_map.get(sku, ""), "grupa": group_map.get(sku, "Other"), }) long_rows.append(d) long_df = pd.DataFrame(long_rows) if long_df.empty: return long_df, pd.DataFrame(), pd.DataFrame() long_df = long_df[[ "sku", "naziv", "grupa", "promo_types", "start_yw", "end_yw", "period", "weeks_span", "weeks_with_data", "total_units", "avg_units_per_week", "total_ruc", "avg_ruc_per_week", "ruc_reliable", ]].sort_values(["sku", "start_yw", "period"]) pivot = long_df.pivot_table( index=["sku", "naziv", "grupa", "promo_types", "start_yw", "end_yw", "ruc_reliable"], columns="period", values=["avg_units_per_week", "avg_ruc_per_week", "weeks_span"], aggfunc="first", ).reset_index() pivot.columns = ["_".join([c for c in col if c]).strip("_") if isinstance(col, tuple) else col for col in pivot.columns] def safe_ratio(num, den): return np.where((den > 0), num / den.replace(0, np.nan), np.nan) pre_u = pivot["avg_units_per_week_pre_promo"] during_u = pivot["avg_units_per_week_during_promo"] post_u = pivot["avg_units_per_week_post_promo"] promo_weeks = pivot["weeks_span_during_promo"].fillna(0) pivot["pull_forward_ratio"] = (post_u / pre_u.replace(0, np.nan)).round(3) pivot["uplift_during_vs_pre"] = (during_u / pre_u.replace(0, np.nan)).round(3) pivot["net_incremental_units"] = ( (during_u - pre_u) * promo_weeks - (pre_u - post_u) * PULL_WINDOW ).round(1) # Per-SKU summary (median across promo events for that SKU) per_sku = ( pivot.groupby(["sku", "naziv", "grupa"], as_index=False) .agg( n_promo_periods=("promo_types", "count"), median_pre_uw=("avg_units_per_week_pre_promo", "median"), median_during_uw=("avg_units_per_week_during_promo", "median"), median_post_uw=("avg_units_per_week_post_promo", "median"), median_pull_fwd_ratio=("pull_forward_ratio", "median"), median_uplift=("uplift_during_vs_pre", "median"), total_net_incremental=("net_incremental_units", "sum"), ) ) # Per-group summary per_group = ( per_sku.groupby("grupa", as_index=False) .agg( n_skus=("sku", "count"), median_pull_fwd_ratio=("median_pull_fwd_ratio", "median"), median_uplift=("median_uplift", "median"), total_net_incremental=("total_net_incremental", "sum"), ) ) return pivot, per_sku, per_group # -------------------------------------------------------------------------- # Metric 4: Stock coverage # -------------------------------------------------------------------------- def stock_coverage( skus: list[str], sales: pd.DataFrame, name_map: dict[str, str], group_map: dict[str, str], ) -> tuple[pd.DataFrame, pd.DataFrame, str | None]: stock_path = DATA_DIR / "stock.csv" if not stock_path.exists(): return pd.DataFrame(), pd.DataFrame(), "N/A — treba inventory data, dodaj data/stock.csv (sku, on_hand)" stock = pd.read_csv(stock_path) sub = sales[sales["sku"].isin(skus)].copy() sub["yw"] = sub["year"].astype(int) * 100 + sub["week"].astype(int) max_year = int(sub["year"].max()) max_week = int(sub.loc[sub["year"] == max_year, "week"].max()) end_yw = max_year * 100 + max_week last_yws = {end_yw} cur = end_yw for _ in range(RECENT_WEEKS - 1): cur = yw_offset(cur, -1) last_yws.add(cur) last = sub[sub["yw"].isin(last_yws)] weekly = last.groupby("sku", as_index=False).agg( weeks_with_sales=("qty_total", lambda s: int((s > 0).sum())), total_units_recent=("qty_total", "sum"), ) weekly["avg_weekly_sales"] = (weekly["total_units_recent"] / RECENT_WEEKS).round(2) cov = pd.DataFrame({"sku": skus}) cov = cov.merge(stock, on="sku", how="left") cov = cov.merge(weekly, on="sku", how="left") cov["weeks_of_cover"] = np.where( cov["avg_weekly_sales"].fillna(0) > 0, (cov["on_hand"].fillna(0) / cov["avg_weekly_sales"]).round(1), np.nan, ) cov["Naziv"] = cov["sku"].map(name_map) cov["Grupa"] = cov["sku"].map(group_map) cov = cov.rename(columns={"sku": "SKU"})[[ "SKU", "Naziv", "Grupa", "on_hand", "weeks_with_sales", "total_units_recent", "avg_weekly_sales", "weeks_of_cover", ]].sort_values("weeks_of_cover", ascending=False, na_position="last") per_group = ( cov.groupby("Grupa", as_index=False) .agg( n_skus=("SKU", "count"), total_on_hand=("on_hand", "sum"), sum_avg_weekly_sales=("avg_weekly_sales", "sum"), median_weeks_of_cover=("weeks_of_cover", "median"), ) ) per_group["group_weeks_of_cover"] = np.where( per_group["sum_avg_weekly_sales"] > 0, (per_group["total_on_hand"] / per_group["sum_avg_weekly_sales"]).round(1), np.nan, ) return cov, per_group, None # -------------------------------------------------------------------------- # Scorecard # -------------------------------------------------------------------------- def recommend(margin_unit, pull_fwd, stock_weeks) -> str: if pd.notna(stock_weeks) and stock_weeks > 30: return "ČIŠĆENJE ZALIHA" if pd.notna(margin_unit) and margin_unit > 5 and pd.notna(pull_fwd) and pull_fwd > 0.7: if pd.notna(stock_weeks) and stock_weeks > 20: return "ZADRŽATI U AKCIJI" return "ZADRŽATI U AKCIJI (provjeri attach)" if pd.notna(margin_unit) and margin_unit > 3 and pd.notna(pull_fwd) and pull_fwd < 0.7: return "SMANJITI RABAT" if (pd.notna(margin_unit) and margin_unit < 3 and pd.notna(pull_fwd) and pull_fwd < 0.7 and pd.notna(stock_weeks) and stock_weeks < 12): return "MAKNUTI IZ AKCIJE" return "PROMOTRITI" def scorecard(margin_group: pd.DataFrame, pull_group: pd.DataFrame, cov_group: pd.DataFrame) -> pd.DataFrame: if margin_group.empty: return pd.DataFrame() sc = margin_group.copy() if not pull_group.empty: sc = sc.merge(pull_group, left_on="Grupa", right_on="grupa", how="left").drop(columns=["grupa"]) if not cov_group.empty: sc = sc.merge(cov_group, on="Grupa", how="left", suffixes=("", "_cov")) sc["Preporuka"] = sc.apply(lambda r: recommend( r.get("median_promo_margin_unit"), r.get("median_pull_fwd_ratio"), r.get("group_weeks_of_cover") if "group_weeks_of_cover" in sc.columns else r.get("median_weeks_of_cover"), ), axis=1) cols = [ "Grupa", "n_skus", "median_promo_margin_unit", "median_margin_erosion_pct", "median_pull_fwd_ratio", "median_uplift", "total_net_incremental", "total_on_hand", "sum_avg_weekly_sales", "group_weeks_of_cover", "median_weeks_of_cover", "Preporuka", ] return sc.reindex(columns=[c for c in cols if c in sc.columns]) # -------------------------------------------------------------------------- # Excel writer # -------------------------------------------------------------------------- def _na_sheet(reason: str) -> pd.DataFrame: return pd.DataFrame({"status": ["N/A"], "reason": [reason]}) def _autofit(ws, df: pd.DataFrame) -> None: for i, col in enumerate(df.columns, 1): ml = max([len(str(col))] + [len(str(v)) for v in df[col].astype(str).head(200)]) ws.column_dimensions[ws.cell(row=1, column=i).column_letter].width = min(ml + 2, 50) # -------------------------------------------------------------------------- # CLI # -------------------------------------------------------------------------- def main() -> None: ap = argparse.ArgumentParser(description="Equipment promo analysis.") ap.add_argument("--skus", type=Path, required=True, help="Text/CSV file with one SKU per line.") ap.add_argument("--out", type=Path, default=DATA_DIR / "equipment_promo_analysis.xlsx", help="Output xlsx path.") args = ap.parse_args() skus = _read_sku_list(args.skus) print(f"Loading data for {len(skus)} equipment SKUs...") sales = _csv("sales_clean.csv") prices = _csv("sku_prices.csv") costs = _csv("sku_costs.csv") plan_df = _csv("sku_plan_list.csv") erp = _csv("erp_promo_calendar.csv") has_order_id = "order_id" in sales.columns or "transaction_id" in sales.columns has_customer_id = "customer_id" in sales.columns name_map = _resolve_names(skus, plan_df) group_map = {s: assign_group(s, name_map.get(s, "")) for s in skus} print(f" names resolved: {sum(1 for v in name_map.values() if v)}/{len(skus)}") print(f" has order_id: {has_order_id}") print(f" has customer_id: {has_customer_id}") print("Metric 1: margin per transaction...") m1_sku, m1_group = margin_per_transaction(skus, sales, prices, costs, name_map, group_map) print(f" {len(m1_sku)} SKUs, {len(m1_group)} groups") print("Metric 3: pull-forward...") pf_pivot, pf_per_sku, pf_per_group = pull_forward(skus, sales, erp, name_map, group_map) print(f" {len(pf_pivot)} promo periods, {len(pf_per_sku)} SKUs with history, {len(pf_per_group)} groups") print("Metric 4: stock coverage...") cov_sku, cov_group, cov_warn = stock_coverage(skus, sales, name_map, group_map) if cov_warn: print(f" WARN: {cov_warn}") else: print(f" {len(cov_sku)} SKUs, {len(cov_group)} groups") print("Scorecard...") score = scorecard(m1_group, pf_per_group, cov_group) print(f" {len(score)} groups scored") args.out.parent.mkdir(parents=True, exist_ok=True) print(f"\nWriting {args.out}...") with pd.ExcelWriter(args.out, engine="openpyxl") as wr: # README first readme = pd.DataFrame({ "Field": [ "Generated", "SKUs in scope", "sales_clean window", "order_id available", "customer_id available", "Pull-forward window", "Stock coverage lookback", "Margin window", "Notes", ], "Value": [ pd.Timestamp.now().strftime("%Y-%m-%d %H:%M"), str(len(skus)), f"{int(sales['year'].min())} W{int(sales[sales['year']==sales['year'].min()]['week'].min())} -> " f"{int(sales['year'].max())} W{int(sales[sales['year']==sales['year'].max()]['week'].max())}", str(has_order_id), str(has_customer_id), f"+/- {PULL_WINDOW} weeks around each contiguous promo period", f"{RECENT_WEEKS} weeks", f"{TX_LOOKBACK_WEEKS} weeks", "ruc_total in sales_clean.csv is 0 for all 2025 — pull-forward RUC reliable only for 2026+ (see ruc_reliable column).", ], }) readme.to_excel(wr, sheet_name="README", index=False) _autofit(wr.sheets["README"], readme) # 1. Margin per transaction m1_sku.to_excel(wr, sheet_name="1_Margin_per_SKU", index=False) _autofit(wr.sheets["1_Margin_per_SKU"], m1_sku) m1_group.to_excel(wr, sheet_name="1_Margin_per_group", index=False) _autofit(wr.sheets["1_Margin_per_group"], m1_group) # 2. Basket attach if has_order_id: pd.DataFrame({"note": ["order_id present but basket calc not implemented in this build."]}) \ .to_excel(wr, sheet_name="2_Basket_attach", index=False) else: na2 = _na_sheet("N/A — treba order-level data (order_id ili transaction_id) u sales_clean.csv") na2.to_excel(wr, sheet_name="2_Basket_attach_NA", index=False) _autofit(wr.sheets["2_Basket_attach_NA"], na2) # 3. Pull-forward if not pf_pivot.empty: pf_pivot.to_excel(wr, sheet_name="3_PullForward_pivot", index=False) _autofit(wr.sheets["3_PullForward_pivot"], pf_pivot) pf_per_sku.to_excel(wr, sheet_name="3_PullForward_per_SKU", index=False) _autofit(wr.sheets["3_PullForward_per_SKU"], pf_per_sku) pf_per_group.to_excel(wr, sheet_name="3_PullForward_per_group", index=False) _autofit(wr.sheets["3_PullForward_per_group"], pf_per_group) else: pd.DataFrame({"note": ["No ERP promo history for any SKU in scope."]}) \ .to_excel(wr, sheet_name="3_PullForward", index=False) # 4. Stock coverage if cov_warn: na4 = _na_sheet(cov_warn) na4.to_excel(wr, sheet_name="4_StockCoverage_NA", index=False) _autofit(wr.sheets["4_StockCoverage_NA"], na4) else: cov_sku.to_excel(wr, sheet_name="4_StockCoverage_per_SKU", index=False) _autofit(wr.sheets["4_StockCoverage_per_SKU"], cov_sku) cov_group.to_excel(wr, sheet_name="4_StockCoverage_per_group", index=False) _autofit(wr.sheets["4_StockCoverage_per_group"], cov_group) # 5. Customer acquisition if has_customer_id: pd.DataFrame({"note": ["customer_id present but cohort calc not implemented in this build."]}) \ .to_excel(wr, sheet_name="5_Customer_acq", index=False) else: na5 = _na_sheet("N/A — treba customer_id u sales_clean.csv (ne postoji)") na5.to_excel(wr, sheet_name="5_Customer_acq_NA", index=False) _autofit(wr.sheets["5_Customer_acq_NA"], na5) # 6. Scorecard if not score.empty: score.to_excel(wr, sheet_name="6_Scorecard", index=False) _autofit(wr.sheets["6_Scorecard"], score) print(f"\n✓ wrote {args.out}") if not score.empty: print("\n=== Equipment scorecard (group) ===") with pd.option_context("display.max_columns", None, "display.width", 200): print(score.to_string(index=False)) if __name__ == "__main__": main()