#!/usr/bin/env python3 """ Spring Weekend Blowout — Promotional Replenishment Plan Generator Event: April 17-19, 2026 (3-day weekend) PO Deadline: April 13, 2026 EOD """ import csv import math from dataclasses import dataclass from typing import Optional # ───────────────────────────────────────────── # DATA LOADING # ───────────────────────────────────────────── def load_csv(path): with open(path) as f: return list(csv.DictReader(f)) sku_master = load_csv("sku_master.csv") pos_data = load_csv("pos_weekly_sales.csv") promo_hist = load_csv("promo_history.csv") # ───────────────────────────────────────────── # UPDATED LEAD TIMES (vendor notice Apr 8) # ───────────────────────────────────────────── UPDATED_LEAD_TIMES = { "VND-201": {"lt_days": 21, "lt_stdev_days": 4}, # NorthStar Foods (was 14/2) "VND-118": {"lt_days": 28, "lt_stdev_days": 6}, # Pacific Roasters (was 21/4) } # ───────────────────────────────────────────── # PROMO BRIEF — lift estimates and dip % # ───────────────────────────────────────────── # Lift multipliers derived from historical promo analogs + promo type/display/circular adjustments # See methodology notes at bottom PROMO_CONFIG = { "SKU-1001": {"type": "25% Off", "display": "Endcap", "circular": True, "lift": 1.70, "dip_pct": 0.10}, "SKU-1002": {"type": "20% Off", "display": "Shelf Talker", "circular": True, "lift": 1.55, "dip_pct": 0.08}, "SKU-1003": {"type": "BOGO 50%", "display": "Floor Stack", "circular": True, "lift": 2.20, "dip_pct": 0.12}, "SKU-1004": {"type": "25% Off", "display": "Endcap", "circular": False, "lift": 1.60, "dip_pct": 0.10}, "SKU-1005": {"type": "30% Off", "display": "Floor Stack", "circular": True, "lift": 1.65, "dip_pct": 0.10}, "SKU-1006": {"type": "BOGO 50%", "display": "Endcap", "circular": True, "lift": 2.10, "dip_pct": 0.12}, "SKU-1007": {"type": "25% Off", "display": "Shelf Talker", "circular": True, "lift": 1.60, "dip_pct": 0.10}, "SKU-1008": {"type": "30% Off", "display": "Endcap", "circular": False, "lift": 1.80, "dip_pct": 0.08}, "SKU-1009": {"type": "25% Off", "display": "Endcap", "circular": True, "lift": 1.70, "dip_pct": 0.10}, "SKU-1010": {"type": "15% Off", "display": "Shelf Talker", "circular": False, "lift": 1.40, "dip_pct": 0.05}, "SKU-1011": {"type": "BOGO", "display": "Floor Stack", "circular": True, "lift": 2.30, "dip_pct": 0.12}, "SKU-1012": {"type": "20% Off", "display": "Endcap", "circular": True, "lift": 1.75, "dip_pct": 0.08}, } # Service level Z-scores Z_SCORES = {0.85: 1.04, 0.90: 1.28, 0.92: 1.41, 0.95: 1.65, 0.975: 1.96, 0.99: 2.33} # ───────────────────────────────────────────── # STEP 1: COMPUTE BASELINE DEMAND & VARIABILITY # ───────────────────────────────────────────── def compute_baselines(): """8-week simple average from POS data (Feb 22 - Apr 12, 2026).""" demand = {} # (sku_id, dc) -> [weekly net units] for row in pos_data: key = (row["sku_id"], row["dc"]) demand.setdefault(key, []).append(int(row["net_units"])) baselines = {} for key, values in demand.items(): n = len(values) avg = sum(values) / n if n > 1: var = sum((v - avg) ** 2 for v in values) / (n - 1) # sample variance else: var = 0 baselines[key] = {"avg": avg, "stdev": math.sqrt(var), "weeks": n} return baselines baselines = compute_baselines() # ───────────────────────────────────────────── # STEP 2: PROMO WEEK DEMAND CALCULATION # ───────────────────────────────────────────── def promo_week_demand(baseline_avg, lift_multiplier): """ 3-day promo within a 7-day week: Promo week demand = baseline * (4/7 + 3/7 * lift) """ return baseline_avg * (4.0 / 7.0 + 3.0 / 7.0 * lift_multiplier) def dip_week_demand(baseline_avg, dip_pct): """Post-promo dip week = baseline * (1 - dip_pct).""" return baseline_avg * (1.0 - dip_pct) # ───────────────────────────────────────────── # STEP 3: SAFETY STOCK WITH UPDATED LEAD TIMES # ───────────────────────────────────────────── def safety_stock(z, lt_weeks, lt_stdev_weeks, demand_avg, demand_stdev): """ SS = Z * sqrt(LT * sigma_d^2 + d_avg^2 * sigma_LT^2) Accounts for both demand and lead time variability. """ return z * math.sqrt(lt_weeks * demand_stdev**2 + demand_avg**2 * lt_stdev_weeks**2) # ───────────────────────────────────────────── # STEP 4: BUILD REPLENISHMENT PLAN # ───────────────────────────────────────────── @dataclass class SKUPlan: sku_id: str sku_name: str dc: str vendor: str vendor_id: str category: str abc_class: str promo_type: str lift: float baseline_weekly: float demand_stdev: float promo_week_fcst: int dip_week_fcst: int lt_days: int lt_stdev_days: int coverage_weeks: float coverage_demand: int safety_stock: int on_hand: int in_transit: int inventory_position: int required_inventory: int raw_order_qty: int moq: int pack_size: int eoq: int po_qty: int unit_cost: float po_value: float weeks_of_cover_current: float stockout_risk: str stockout_week: Optional[str] notes: str def round_up_to_pack(qty, pack_size, moq): """Round order qty up to nearest pack_size, enforcing MOQ minimum.""" if qty <= 0: return 0 qty_packs = math.ceil(qty / pack_size) * pack_size return max(qty_packs, moq) def build_plan(): plans = [] for sku in sku_master: sku_id = sku["sku_id"] vendor_id = sku["vendor_id"] promo = PROMO_CONFIG[sku_id] service_level = float(sku["service_level_target"]) z = Z_SCORES[service_level] moq = int(sku["moq"]) pack_size = int(sku["pack_size"]) eoq = int(sku["eoq"]) unit_cost = float(sku["unit_cost"]) # Updated lead times if vendor_id in UPDATED_LEAD_TIMES: lt_days = UPDATED_LEAD_TIMES[vendor_id]["lt_days"] lt_stdev_days = UPDATED_LEAD_TIMES[vendor_id]["lt_stdev_days"] else: lt_days = int(sku["lead_time_days"]) lt_stdev_days = int(sku["lead_time_stdev_days"]) lt_weeks = lt_days / 7.0 lt_stdev_weeks = lt_stdev_days / 7.0 # Determine active DCs dcs = [] if sku["dc_east_active"] == "Y": dcs.append(("DC-East", int(sku["on_hand_dc_east"]), int(sku["in_transit_dc_east"]))) if sku["dc_west_active"] == "Y": dcs.append(("DC-West", int(sku["on_hand_dc_west"]), int(sku["in_transit_dc_west"]))) for dc, on_hand, in_transit in dcs: key = (sku_id, dc) if key not in baselines: continue bl = baselines[key] base_avg = bl["avg"] base_stdev = bl["stdev"] # Demand forecasts promo_wk = promo_week_demand(base_avg, promo["lift"]) dip_wk = dip_week_demand(base_avg, promo["dip_pct"]) # Safety stock ss = safety_stock(z, lt_weeks, lt_stdev_weeks, base_avg, base_stdev) ss = math.ceil(ss) # Coverage period: weeks from today until PO arrives coverage_weeks = lt_weeks # PO placed today arrives in lt_days # Demand during coverage: promo week + dip week + remaining baseline weeks if coverage_weeks <= 1: cov_demand = promo_wk elif coverage_weeks <= 2: cov_demand = promo_wk + min(coverage_weeks - 1, 1) * dip_wk else: remaining_baseline_weeks = coverage_weeks - 2 cov_demand = promo_wk + dip_wk + remaining_baseline_weeks * base_avg cov_demand = math.ceil(cov_demand) # Inventory position ip = on_hand + in_transit # Required inventory = coverage demand + safety stock required = cov_demand + ss # Raw order qty raw_qty = max(0, required - ip) # Round to pack/MOQ po_qty = round_up_to_pack(raw_qty, pack_size, moq) if raw_qty > 0 else 0 po_value = po_qty * unit_cost # Weeks of cover with current IP (no PO) if base_avg > 0: woc = ip / base_avg else: woc = float('inf') # Stockout risk assessment — simulate week-by-week depletion remaining = ip stockout_risk = "LOW" stockout_week = None # Week 1: promo week (Apr 13-19) remaining -= promo_wk if remaining < 0 and stockout_week is None: stockout_risk = "CRITICAL" stockout_week = "Apr 13-19 (promo week)" # Week 2: dip week (Apr 20-26) if remaining > 0: remaining -= dip_wk if remaining < 0 and stockout_week is None: stockout_risk = "CRITICAL" stockout_week = "Apr 20-26 (dip week)" # Weeks 3+: baseline until PO arrives if remaining > 0: extra_weeks = max(0, coverage_weeks - 2) full_extra = int(extra_weeks) frac = extra_weeks - full_extra for w in range(full_extra): remaining -= base_avg if remaining < 0 and stockout_week is None: week_start_day = 14 + w * 7 # days from Apr 13 stockout_risk = "HIGH" if remaining > -base_avg * 0.5 else "CRITICAL" from datetime import datetime, timedelta d = datetime(2026, 4, 13) + timedelta(days=14 + w * 7) stockout_week = f"~{d.strftime('%b %d')} (week {w + 3})" if remaining > 0 and frac > 0: remaining -= frac * base_avg if remaining < 0 and stockout_week is None: stockout_risk = "MODERATE" from datetime import datetime, timedelta d = datetime(2026, 4, 13) + timedelta(days=int(coverage_weeks * 7) - 3) stockout_week = f"~{d.strftime('%b %d')} (near PO arrival)" # If remaining stays positive but thin (< 0.5 weeks of baseline) if stockout_risk == "LOW" and remaining < base_avg * 0.5 and remaining >= 0: stockout_risk = "WATCH" stockout_week = "Tight buffer — any demand spike breaks SS" notes_parts = [] if vendor_id in UPDATED_LEAD_TIMES: notes_parts.append(f"LT extended to {lt_days}d (was {sku['lead_time_days']}d)") if promo["type"].startswith("BOGO"): notes_parts.append("BOGO — expect forward-buy behavior") if sku_id == "SKU-1002": notes_parts.append("Perishable — coordinate short-shelf-life units with FreshDairy") if sku_id == "SKU-1011": notes_parts.append(f"MOQ={moq} forces overage of {po_qty - raw_qty} units" if po_qty > raw_qty > 0 else "") plan = SKUPlan( sku_id=sku_id, sku_name=sku["sku_name"], dc=dc, vendor=sku["vendor"], vendor_id=vendor_id, category=sku["category"], abc_class=sku["abc_class"], promo_type=promo["type"], lift=promo["lift"], baseline_weekly=round(base_avg, 1), demand_stdev=round(base_stdev, 1), promo_week_fcst=math.ceil(promo_wk), dip_week_fcst=math.ceil(dip_wk), lt_days=lt_days, lt_stdev_days=lt_stdev_days, coverage_weeks=round(coverage_weeks, 1), coverage_demand=cov_demand, safety_stock=ss, on_hand=on_hand, in_transit=in_transit, inventory_position=ip, required_inventory=required, raw_order_qty=raw_qty, moq=moq, pack_size=pack_size, eoq=eoq, po_qty=po_qty, unit_cost=unit_cost, po_value=round(po_value, 2), weeks_of_cover_current=round(woc, 1), stockout_risk=stockout_risk, stockout_week=stockout_week, notes="; ".join(n for n in notes_parts if n), ) plans.append(plan) return plans # ───────────────────────────────────────────── # OUTPUT GENERATION # ───────────────────────────────────────────── def generate_report(plans): lines = [] L = lines.append L("=" * 90) L("SPRING WEEKEND BLOWOUT — PROMOTIONAL REPLENISHMENT PLAN") L("=" * 90) L(f"Event: April 17–19, 2026 (Fri–Sun, 3-day weekend)") L(f"Plan Date: April 13, 2026 (Monday)") L(f"PO Deadline: EOD Today") L(f"Prepared by: Demand Planning") L("") # ── EXECUTIVE SUMMARY ── L("─" * 90) L("EXECUTIVE SUMMARY") L("─" * 90) pos_needed = [p for p in plans if p.po_qty > 0] critical = [p for p in plans if p.stockout_risk == "CRITICAL"] high = [p for p in plans if p.stockout_risk == "HIGH"] total_po_value = sum(p.po_value for p in plans) total_po_units = sum(p.po_qty for p in plans) L(f" Total POs required: {len(pos_needed)} line items across {len(set(p.vendor_id for p in pos_needed))} vendors") L(f" Total PO units: {total_po_units:,}") L(f" Total PO value (cost): ${total_po_value:,.2f}") L(f" CRITICAL stockout risks: {len(critical)} SKU-DC combos") L(f" HIGH stockout risks: {len(high)} SKU-DC combos") L(f" No PO needed: {len([p for p in plans if p.po_qty == 0])} SKU-DC combos (adequate coverage)") L("") if critical: L(" *** CRITICAL ALERTS ***") for p in critical: L(f" !!! {p.sku_id} {p.sku_name} @ {p.dc}: stocks out {p.stockout_week}") L(f" IP={p.inventory_position}, promo wk demand={p.promo_week_fcst}, LT={p.lt_days}d") L(f" PO of {p.po_qty} units placed today won't arrive until ~{p.lt_days} days") L("") L(" >> ACTION: Contact Pacific Roasters (VND-118) and NorthStar Foods (VND-201)") L(" for expedited/partial shipments on critical items. Explore DC-to-DC") L(" transfers or substitute sourcing for bridge coverage.") L("") # ── DEMAND FORECAST DETAIL ── L("─" * 90) L("DEMAND FORECAST BY SKU × DC") L("─" * 90) L(f"{'SKU':<10} {'DC':<9} {'Base/wk':>8} {'σ_d':>5} {'Lift':>5} {'Promo Wk':>9} {'Dip Wk':>8} {'Dip%':>5}") L("-" * 70) for p in plans: L(f"{p.sku_id:<10} {p.dc:<9} {p.baseline_weekly:>8.0f} {p.demand_stdev:>5.1f} {p.lift:>5.2f} {p.promo_week_fcst:>9,} {p.dip_week_fcst:>8,} {p.dip_pct * 100 if hasattr(p, 'dip_pct') else PROMO_CONFIG[p.sku_id]['dip_pct'] * 100:>4.0f}%") L("") # ── SAFETY STOCK DETAIL ── L("─" * 90) L("SAFETY STOCK (recalculated with updated lead times)") L("─" * 90) L(f"{'SKU':<10} {'DC':<9} {'SL%':>5} {'Z':>5} {'LT(d)':>6} {'σ_LT':>5} {'SS':>6} {'Notes'}") L("-" * 80) for p in plans: lt_note = "** EXTENDED" if p.vendor_id in UPDATED_LEAD_TIMES else "" L(f"{p.sku_id:<10} {p.dc:<9} {float(next(s['service_level_target'] for s in sku_master if s['sku_id']==p.sku_id))*100:>4.0f}% {Z_SCORES[float(next(s['service_level_target'] for s in sku_master if s['sku_id']==p.sku_id))]:>5.2f} {p.lt_days:>5}d {p.lt_stdev_days:>4}d {p.safety_stock:>6,} {lt_note}") L("") # ── INVENTORY POSITION & COVERAGE ── L("─" * 90) L("INVENTORY POSITION & COVERAGE ANALYSIS") L("─" * 90) L(f"{'SKU':<10} {'DC':<9} {'OnHand':>7} {'InTrns':>7} {'IP':>7} {'CovDmd':>7} {'SS':>6} {'Req':>7} {'Gap':>7} {'Risk':<10}") L("-" * 85) for p in plans: gap = max(0, p.required_inventory - p.inventory_position) risk_marker = "***" if p.stockout_risk == "CRITICAL" else "**" if p.stockout_risk == "HIGH" else "*" if p.stockout_risk == "WATCH" else "" L(f"{p.sku_id:<10} {p.dc:<9} {p.on_hand:>7,} {p.in_transit:>7,} {p.inventory_position:>7,} {p.coverage_demand:>7,} {p.safety_stock:>6,} {p.required_inventory:>7,} {gap:>7,} {p.stockout_risk:<10} {risk_marker}") L("") # ── PO RECOMMENDATIONS BY VENDOR ── L("─" * 90) L("PURCHASE ORDER RECOMMENDATIONS (submit by EOD April 13)") L("─" * 90) L("") # Group by vendor vendors = {} for p in plans: vendors.setdefault(p.vendor_id, []).append(p) for vid in sorted(vendors.keys()): vplans = vendors[vid] vname = vplans[0].vendor v_total_qty = sum(p.po_qty for p in vplans) v_total_val = sum(p.po_value for p in vplans) po_lines = [p for p in vplans if p.po_qty > 0] L(f" VENDOR: {vname} ({vid})") if vid in UPDATED_LEAD_TIMES: L(f" *** LEAD TIME EXTENDED: {UPDATED_LEAD_TIMES[vid]['lt_days']}d (σ={UPDATED_LEAD_TIMES[vid]['lt_stdev_days']}d) ***") L(f" {'─' * 75}") if not po_lines: L(f" No PO required — all SKUs adequately covered.") L("") continue L(f" {'SKU':<10} {'DC':<9} {'PO Qty':>8} {'Packs':>6} {'Unit$':>7} {'PO Value':>10} {'Risk':<10}") L(f" {'-' * 65}") for p in po_lines: packs = p.po_qty // p.pack_size L(f" {p.sku_id:<10} {p.dc:<9} {p.po_qty:>8,} {packs:>5}x{p.pack_size} ${p.unit_cost:>6.2f} ${p.po_value:>9,.2f} {p.stockout_risk:<10}") L(f" {'-' * 65}") L(f" VENDOR TOTAL: {v_total_qty:>7,} units ${v_total_val:>10,.2f}") L("") # Flag critical items v_critical = [p for p in po_lines if p.stockout_risk in ("CRITICAL", "HIGH")] if v_critical: L(f" >> ESCALATION: Request expedited/partial shipment for:") for p in v_critical: L(f" - {p.sku_id} {p.sku_name} @ {p.dc}: stocks out {p.stockout_week}") L("") L("") # ── STOCKOUT RISK SUMMARY ── L("─" * 90) L("STOCKOUT RISK SUMMARY & BRIDGE ORDER RECOMMENDATIONS") L("─" * 90) L("") risk_items = [p for p in plans if p.stockout_risk in ("CRITICAL", "HIGH", "WATCH")] if risk_items: for p in sorted(risk_items, key=lambda x: {"CRITICAL": 0, "HIGH": 1, "WATCH": 2}[x.stockout_risk]): L(f" [{p.stockout_risk}] {p.sku_id} {p.sku_name} @ {p.dc}") L(f" Current IP: {p.inventory_position:,} | Promo wk demand: {p.promo_week_fcst:,} | LT: {p.lt_days}d") L(f" Estimated stockout: {p.stockout_week}") if p.stockout_risk == "CRITICAL": # Calculate bridge qty needed bridge = max(0, p.coverage_demand - p.inventory_position) L(f" Bridge qty needed: ~{bridge:,} units to cover gap until PO arrives") L(f" >> Recommend: expedited partial ship / DC transfer / alternate source") elif p.stockout_risk == "HIGH": L(f" >> Recommend: request expedited shipment, monitor daily POS") elif p.stockout_risk == "WATCH": L(f" >> Recommend: monitor daily POS, prepare contingency reorder") L("") else: L(" No stockout risks identified.") L("") # ── DAIRY COORDINATION NOTE ── L("─" * 90) L("SPECIAL COORDINATION NOTES") L("─" * 90) L("") L(" 1. DAIRY (SKU-1002 Greek Yogurt): No PO needed — ample stock at both DCs.") L(" However, coordinate with FreshDairy Co on shelf-life dating for promo units.") L(" Ensure units arriving have >14 days remaining shelf life for weekend event.") L("") L(" 2. PACIFIC ROASTERS (SKU-1003, SKU-1008): 28-day lead time means today's PO") L(" arrives ~May 11. Current stock covers only ~1.5-2 weeks. CRITICAL gap.") L(" Escalate to supply chain director per protocol (vendor LT increase >25%).") L("") L(" 3. NORTHSTAR FOODS (SKU-1001, 1004, 1006, 1012): 21-day lead time (was 14d).") L(" SKU-1006 and SKU-1012 West have tight coverage through lead time.") L(" NorthStar POs should include a request for earliest possible ship date.") L("") L(" 4. POST-PROMO WEEK (Apr 20-26): Dip expected across all SKUs. Do NOT reorder") L(" aggressively during dip week — let inventory normalize before next cycle.") L("") # ── LIFT METHODOLOGY ── L("─" * 90) L("LIFT ESTIMATION METHODOLOGY") L("─" * 90) L("") L(" Lifts derived from own-item promo history with recency weighting, adjusted for:") L(" - Promo type (BOGO > % Off at equivalent depth)") L(" - Display type (Floor Stack/Endcap > Shelf Talker)") L(" - Circular feature (adds ~10-15% incremental lift)") L(" - Seasonal timing (spring wellness trend supports protein/health items)") L("") L(" 3-day weekend scaling: lift multiplier applies to Fri-Sun only.") L(" Promo week = baseline × (4/7 + 3/7 × lift), NOT baseline × lift.") L("") L(" Post-promo dip: based on historical observed dip %, scaled by promo depth") L(" and product storability. BOGO on shelf-stable = higher dip (12%).") L(" Perishables/low-depth = lower dip (5-8%).") L("") L(" Confidence interval: ±20-25% on lift estimates. Safety stock sized accordingly.") L("=" * 90) return "\n".join(lines) def generate_po_csv(plans): """Generate a clean PO-ready CSV for vendor submission.""" lines = ["vendor_id,vendor_name,sku_id,sku_name,dc,po_qty,pack_size,num_packs,unit_cost,po_line_value,requested_ship_date,priority,notes"] from datetime import datetime, timedelta for p in sorted(plans, key=lambda x: (x.vendor_id, x.sku_id, x.dc)): if p.po_qty == 0: continue ship_date = "ASAP" if p.stockout_risk in ("CRITICAL", "HIGH") else "Standard" packs = p.po_qty // p.pack_size priority = "URGENT" if p.stockout_risk in ("CRITICAL", "HIGH") else "STANDARD" notes = p.notes.replace(",", ";") if p.notes else "" lines.append(f"{p.vendor_id},{p.vendor},{p.sku_id},{p.sku_name},{p.dc},{p.po_qty},{p.pack_size},{packs},{p.unit_cost},{p.po_value},{ship_date},{priority},{notes}") return "\n".join(lines) # ── MAIN ── if __name__ == "__main__": plans = build_plan() # Add dip_pct to plans for report formatting for p in plans: p.dip_pct = PROMO_CONFIG[p.sku_id]["dip_pct"] report = generate_report(plans) print(report) # Write report to file with open("replenishment_plan_apr13.txt", "w") as f: f.write(report) # Write PO CSV po_csv = generate_po_csv(plans) with open("po_recommendations_apr13.csv", "w") as f: f.write(po_csv) print("\n\nFiles written:") print(" - replenishment_plan_apr13.txt (full plan)") print(" - po_recommendations_apr13.csv (PO-ready CSV for vendor submission)")