"""Monthly plan snapshot service. Builds and stores per-month plan snapshots that other modules (Margin Bridge, Revenue Forecast monthly view, NPL Report) consume as the canonical "plan" for that calendar month. Created when the user clicks "Lock plan for month X" on the Consensus Plan page. Two methodologies stored in every snapshot: - **day_fraction**: each ISO week is split pro-rata across calendar months by day count. Mass-preserving — required for revenue/RUC/qty totals. A week 4/7 in April + 3/7 in May contributes those exact fractions. - **thursday_rule**: each ISO week snaps to the month containing its Thursday. Required for Forecast Accuracy and similar bucketing where a week must belong to a single month. The snapshot stores per-SKU detail (JSONB) plus aggregate totals. Source cascade per (SKU, week): 1. `backtest_results.forecast_*` — for past weeks the engine has back- tested (CW12-CW19/2026 currently) 2. `forecasts.forecast_*` — for future weeks (CW21+ currently) 3. **Gap fallback**: when neither source has the week (e.g. CW20 in May 2026 — the current-week gap), use the SKU's nearest available baseline (most recent backtest week ≤ target week, or earliest forecast week ≥ target week, whichever is closer). On-top quantities from `on_top_inputs` are added when present for the (SKU, year_week) regardless of which baseline source was used. """ from __future__ import annotations from dataclasses import dataclass from datetime import datetime from typing import Optional import pandas as pd from sqlalchemy import text from sqlalchemy.orm import Session from backend.services.time_utils import ( days_in_calendar_month, iso_week_first_monday, split_iso_week_across_months, ) # ────────────────────────────────────────────────────────────────────── # Per-month calibration overrides # ────────────────────────────────────────────────────────────────────── # When the model output diverges from the realistic plan number — e.g., # the engine over-predicts retail-channel baseline qty by ~15-20% in # May/June 2026 vs trailing actuals — apply a multiplier here. Only # affects BASELINE qty (engine forecast for that channel); on-top # commits from KAM/CM (on_top_retail / on_top_wholesale) are NEVER # scaled — those are the planner-set promo quantities and should pass # through untouched. # # Format: { month_key (YYYYMM): {'retail_baseline_factor': 0.85, ...} } # Remove entries here once the engine re-train resolves the bias. CALIBRATION_BY_MONTH: dict[int, dict[str, float]] = { # `retail_pct_above_mgmt`: pin the plan's retail RUC to mgmt-plan # retail × (1 + pct). 0.10 = "10% above mgmt commitment". This is a # direct target, applied AFTER the RF + MCI headline alignment so it # actually moves the locked plan total (not just rebalances channels). # The delta vs the engine-implied retail is absorbed by reducing the # aligned headline (revenue + RUC) — wholesale & webshop stay at # their model-implied values. 202605: {"retail_pct_above_mgmt": 0.10}, # May 2026 202606: {"retail_pct_above_mgmt": 0.10}, # Jun 2026 } # ────────────────────────────────────────────────────────────────────── # Helpers # ────────────────────────────────────────────────────────────────────── def _iso_weeks_in_month(year: int, month: int) -> list[tuple[int, int, float]]: """Return list of (iso_year, iso_week, day_fraction) for every ISO week with at least one day in the given calendar month.""" from datetime import date, timedelta # All days in this calendar month d = date(year, month, 1) last_day = days_in_calendar_month(year, month) weeks: dict[tuple[int, int], int] = {} for offset in range(last_day): cur = d + timedelta(days=offset) iso = cur.isocalendar() key = (int(iso[0]), int(iso[1])) weeks[key] = weeks.get(key, 0) + 1 return [(y, w, n / 7.0) for (y, w), n in sorted(weeks.items())] def _thursday_month(iso_year: int, iso_week: int) -> tuple[int, int]: """Thursday-of-week → (calendar_year, calendar_month).""" mon = iso_week_first_monday(iso_year, iso_week) from datetime import timedelta thu = mon + timedelta(days=3) return thu.year, thu.month # ────────────────────────────────────────────────────────────────────── # Source loader — per-SKU per-week qty (with channel split) # ────────────────────────────────────────────────────────────────────── def _load_plan_qty_per_sku_week(db: Session) -> pd.DataFrame: """Combined long-format DataFrame: (sku, product_id, year, week, qty_total, qty_ws, qty_b2c, source). One row per (sku, week) — covering the union of backtest_results and forecasts. Critical: this MUST mirror what Revenue Forecast page shows, so totals agree between the two views. Both use: qty_baseline = forecasts.baseline × planner_factor (NOT forecasts.total, because total has on-tops baked in and on-tops are added separately downstream — using total double-counts) ws_share = sku_planning.ws_share_26w (canonical 26-week channel split; NOT recomputed from v_sales_weekly_full) Backtest rows still come with their own per-channel splits (engine populates those for past weeks). """ bt = pd.read_sql(text(""" SELECT p.sku, b.product_id, b.year, b.week, b.forecast::float AS qty_total, b.forecast_wholesale::float AS qty_ws, b.forecast_retail::float AS qty_b2c, COALESCE(sp.ws_share_26w, 0.5)::float AS _ws_share_26w, 'backtest' AS source FROM backtest_results b JOIN dim_products p ON p.id = b.product_id LEFT JOIN sku_planning sp ON sp.product_id = b.product_id WHERE b.forecast IS NOT NULL """), db.bind) # When per-channel forecast columns are NULL (some backfilled rows # don't have them), fall back to ws_share_26w split — otherwise the # qty_total just gets dropped from both channel buckets downstream. if not bt.empty: bt_mask = bt["qty_ws"].isna() | bt["qty_b2c"].isna() bt.loc[bt_mask, "qty_ws"] = bt.loc[bt_mask, "qty_total"] * bt.loc[bt_mask, "_ws_share_26w"] bt.loc[bt_mask, "qty_b2c"] = bt.loc[bt_mask, "qty_total"] * (1.0 - bt.loc[bt_mask, "_ws_share_26w"]) bt = bt.drop(columns=["_ws_share_26w"]) # Forecasts: use baseline × planner_factor for qty_total. For the # WS/B2C split prefer the engine's per-channel output # (forecast_wholesale + forecast_retail) when populated. Falls back # to sku_planning.ws_share_26w only when the engine didn't write them # (legacy rows pre-May-2026 engine fix). DISTINCT ON picks the latest # run per (sku, year, week) so CW22 falls back to run #5 if run #6 # didn't predict it. fc = pd.read_sql(text(""" WITH latest_fc AS ( SELECT DISTINCT ON (product_id, year, week) product_id, year, week, baseline, planner_factor, forecast_wholesale, forecast_retail FROM forecasts ORDER BY product_id, year, week, run_id DESC ) SELECT p.sku, f.product_id, f.year, f.week, (COALESCE(f.baseline, 0) * COALESCE(f.planner_factor, 1))::float AS qty_total, f.forecast_wholesale::float AS _fc_ws, f.forecast_retail::float AS _fc_mp, COALESCE(sp.ws_share_26w, 0)::float AS _ws_share, 'forecasts' AS source FROM latest_fc f JOIN dim_products p ON p.id = f.product_id LEFT JOIN sku_planning sp ON sp.product_id = f.product_id """), db.bind) if not fc.empty: # Use engine's per-channel output where populated (real model # output for that week, not a trailing proxy). Fall back to # ws_share_26w only when both per-channel cols are NULL. per_ch_mask = fc["_fc_ws"].notna() & fc["_fc_mp"].notna() fc.loc[per_ch_mask, "qty_ws"] = fc.loc[per_ch_mask, "_fc_ws"] fc.loc[per_ch_mask, "qty_b2c"] = fc.loc[per_ch_mask, "_fc_mp"] # Where total ≠ ws + mp (webshop slice, rounding), assign the # residual to whichever channel was empty so qty_ws+qty_b2c keeps # equal to qty_total. Difference is usually <5%. resid = fc.loc[per_ch_mask, "qty_total"] - (fc.loc[per_ch_mask, "_fc_ws"] + fc.loc[per_ch_mask, "_fc_mp"]) fc.loc[per_ch_mask, "qty_b2c"] += resid # webshop typically lives in B2C bucket # Fallback rows: use ws_share_26w fb_mask = ~per_ch_mask fc.loc[fb_mask, "qty_ws"] = fc.loc[fb_mask, "qty_total"] * fc.loc[fb_mask, "_ws_share"] fc.loc[fb_mask, "qty_b2c"] = fc.loc[fb_mask, "qty_total"] * (1.0 - fc.loc[fb_mask, "_ws_share"]) fc = fc.drop(columns=["_fc_ws", "_fc_mp", "_ws_share"]) combined = pd.concat([bt, fc], ignore_index=True) # Backtest takes precedence over forecasts where both exist combined = combined.sort_values(["sku", "year", "week", "source"]).drop_duplicates( subset=["sku", "year", "week"], keep="first" ) return combined def _apply_month_calibration(qpw: pd.DataFrame, year: int, month: int) -> pd.DataFrame: """Apply per-month calibration (e.g., retail baseline factor) to the per-(sku, week) qty dataframe. Pure function — no side effects on the underlying forecasts/backtest tables. Idempotent. Retail factor scales qty_b2c (which at this point is baseline retail + webshop residual, no on-tops yet — on_top_inputs is summed in later). The split between retail and webshop within qty_b2c is estimated from trailing-13w mix (~86% retail / 14% webshop) so we only scale the retail portion. """ cfg = CALIBRATION_BY_MONTH.get(year * 100 + month, {}) if not cfg: return qpw retail_factor = float(cfg.get("retail_baseline_factor", 1.0)) if retail_factor == 1.0: return qpw # Approximation: blended factor on qty_b2c. 86% retail × 0.85 + 14% # webshop × 1.0 = 0.871 effective B2C multiplier. RETAIL_SHARE_WITHIN_B2C = 0.86 blended_b2c_factor = ( RETAIL_SHARE_WITHIN_B2C * retail_factor + (1.0 - RETAIL_SHARE_WITHIN_B2C) * 1.0 ) # Only the weeks that fall inside this calendar month — we don't want # to affect other months that share an ISO week (rare but possible at # month boundaries). iso_yws_in_month = {(y, w) for (y, w, _) in _iso_weeks_in_month(year, month)} mask = qpw.apply( lambda r: (int(r["year"]), int(r["week"])) in iso_yws_in_month, axis=1, ) if mask.any(): qpw = qpw.copy() # Reduce qty_b2c (and proportionally adjust qty_total). qty_ws # untouched. qpw.loc[mask, "qty_b2c"] = qpw.loc[mask, "qty_b2c"] * blended_b2c_factor qpw.loc[mask, "qty_total"] = qpw.loc[mask, "qty_ws"] + qpw.loc[mask, "qty_b2c"] return qpw def _load_on_top_inputs(db: Session) -> pd.DataFrame: """Long-format: (product_id, year_week, channel, quantity). Aggregated across all buyers within a (product, week, channel).""" return pd.read_sql(text(""" SELECT product_id, year_week, channel, SUM(quantity)::float AS quantity FROM on_top_inputs GROUP BY product_id, year_week, channel """), db.bind) def _compute_grossup_ratios(db: Session) -> dict: """Per-channel gross-up ratio to estimate non-planned SKU contribution. Mirrors `get_revenue_grossup_ratios()` in demand_repo so snapshot totals are consistent with Revenue Forecast's "include non-planned" toggle. Ratio = total_channel_revenue / planned_channel_revenue over the last 13 ISO weeks of v_sales_weekly_full. Always ≥ 1.0 (degenerates to 1.0 when there's no non-planned activity). """ r = db.execute(text(""" WITH last_weeks AS ( SELECT year, week FROM ( SELECT DISTINCT year, week FROM v_sales_weekly_full ORDER BY year DESC, week DESC LIMIT 13 ) sub ), base AS ( SELECT v.product_id, SUM(v.qty_wholesale * COALESCE(ep.avg_sell_price, 0)) AS ws_rev, SUM((v.qty_retail + v.qty_webshop) * COALESCE(ep.avg_sell_price, 0)) AS rt_rev FROM v_sales_weekly_full v JOIN last_weeks lw ON lw.year = v.year AND lw.week = v.week LEFT JOIN erp_prices ep ON ep.product_id = v.product_id GROUP BY v.product_id ) SELECT SUM(ws_rev)::float AS ws_all, SUM(rt_rev)::float AS rt_all, SUM(CASE WHEN EXISTS ( SELECT 1 FROM sku_planning sp WHERE sp.product_id = base.product_id) THEN ws_rev ELSE 0 END)::float AS ws_plan, SUM(CASE WHEN EXISTS ( SELECT 1 FROM sku_planning sp WHERE sp.product_id = base.product_id) THEN rt_rev ELSE 0 END)::float AS rt_plan FROM base """)).mappings().first() or {} ws_all = float(r.get("ws_all") or 0) rt_all = float(r.get("rt_all") or 0) ws_plan = float(r.get("ws_plan") or 0) rt_plan = float(r.get("rt_plan") or 0) def _ratio(all_: float, plan: float) -> float: return float(all_ / plan) if (plan > 0 and all_ > plan) else 1.0 ws_ratio = _ratio(ws_all, ws_plan) b2c_ratio = _ratio(rt_all, rt_plan) all_total = ws_all + rt_all all_plan = ws_plan + rt_plan nonplan_share_pct = ( (1 - all_plan / all_total) * 100.0 if all_total > all_plan else 0.0 ) return { "ws_ratio": round(ws_ratio, 4), "b2c_ratio": round(b2c_ratio, 4), "nonplanned_share_pct": round(nonplan_share_pct, 2), } def _load_ruc_rates(db: Session, plan_year: int, plan_month: int, window_weeks: int = 8) -> pd.DataFrame: """Per-SKU per-channel realised rates over an **anchored trailing window** ending BEFORE the planned month starts. Prices use NET of VAT. **Anchor**: 7 days before plan_first_day so anchor's ISO week ends before the planned month. For April 2026 plan → anchor = March 25 (CW13) → baseline weeks = CW06–CW13 (Feb–Mar). This avoids the self-reference trap where snapshot locked in May uses May data as "April plan baseline". **Window**: 8 ISO weeks by default. Spans roughly two monthly promo cycles for Polleo, smoothing regime swings (promo month vs non-promo month). A non-promo carve-out via `erp_promo_weeks` is intentionally NOT applied: that table is dominated by permanent OUTLET flags (~80% of rows), so filtering would collapse the baseline to a fraction of SKUs. The 8w window mitigates regime mismatch instead. `tax_base` is the authoritative NET column when populated. ~30% of rows have tax_base=0, so we fall back to `total_value × 0.80`. RUC is net by definition (RUC = revenue − COGS, both excl. VAT).""" from datetime import date, timedelta plan_first_day = date(plan_year, plan_month, 1) # Anchor 7 days before plan_first_day so the anchor's ISO week is # guaranteed to end BEFORE the plan month starts (avoids the boundary # week that straddles month-end/month-start being counted in baseline). # For April 1 plan → anchor = March 25 (CW13) → baseline = CW06–CW13. anchor_date = plan_first_day - timedelta(days=7) sql = f""" WITH anchor AS ( SELECT EXTRACT(ISOYEAR FROM DATE '{anchor_date.isoformat()}')::int AS y, EXTRACT(WEEK FROM DATE '{anchor_date.isoformat()}')::int AS w ), last_weeks AS ( SELECT DISTINCT year, week FROM v_sales_weekly_full v WHERE (v.year * 100 + v.week) <= (SELECT y*100 + w FROM anchor) ORDER BY year DESC, week DESC LIMIT {int(window_weeks)} ), recent_yws AS ( SELECT year * 100 + week AS yw FROM last_weeks ) SELECT p.sku, p.id AS product_id, NULLIF(SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.quantity ELSE 0 END), 0)::float AS qty_r, SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.ruc_eur ELSE 0 END)::float AS ruc_r, SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80) ELSE 0 END)::float AS rev_r, NULLIF(SUM(CASE WHEN cm.channel = 'wholesale' THEN et.quantity ELSE 0 END), 0)::float AS qty_w, SUM(CASE WHEN cm.channel = 'wholesale' THEN et.ruc_eur ELSE 0 END)::float AS ruc_w, SUM(CASE WHEN cm.channel = 'wholesale' THEN COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80) ELSE 0 END)::float AS rev_w FROM erp_transactions et JOIN dim_products p ON p.id = et.product_id JOIN lookup_channel_map cm ON cm.id = et.channel_map_id WHERE (EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100 + EXTRACT(WEEK FROM et.transaction_date)::int) IN (SELECT yw FROM recent_yws) GROUP BY p.sku, p.id """ return pd.read_sql(text(sql), db.bind) def _load_planned_promo_discounts(db: Session, plan_year: int, plan_month: int) -> dict[int, float]: """For each product on planned promo overlapping the plan month, return the **planned discount percentage** from `erp_promo_items`. Multiple overlapping campaigns are averaged. Returns dict[product_id] -> discount_pct (e.g., 25.0 = 25% off catalog). This is the **forward-anchored** plan source: the rabatna politika is set 1.5 months ahead (loaded via VrstaRabatnePolitike Excel files into promo_policies + automatically via erp_promo_items from the ERP promo sync). Margin Bridge uses these to know what discount was supposed to apply, then variance vs actual measures KAM compliance. SKUs without a row in result have no curated promo plan → fall back to trailing baseline downstream. """ from datetime import date, timedelta month_start = date(plan_year, plan_month, 1) if plan_month == 12: month_end = date(plan_year + 1, 1, 1) - timedelta(days=1) else: month_end = date(plan_year, plan_month + 1, 1) - timedelta(days=1) # Threshold 5%: erp_promo_items often has OUTLET-tag entries with tiny # (< 5%) "discounts" that are bookkeeping artifacts, not real campaigns. # Real Polleo promo discounts start at 10%+ (Akcije X/26 typically # 15-40%). 5% cutoff drops the noise while keeping all genuine campaigns. PROMO_DISCOUNT_THRESHOLD = 5.0 rows = db.execute(text(""" SELECT product_id, AVG(discount_pct)::float AS pct FROM erp_promo_items WHERE valid_from <= :end AND valid_to >= :start AND discount_pct IS NOT NULL AND discount_pct > :thresh GROUP BY product_id HAVING AVG(discount_pct) > :thresh """), {"start": month_start, "end": month_end, "thresh": PROMO_DISCOUNT_THRESHOLD}).all() return {int(pid): float(pct) for pid, pct in rows if pct and pct > PROMO_DISCOUNT_THRESHOLD} def _load_catalog_prices(db: Session) -> dict[int, dict]: """Per-SKU catalog list prices (retail_ppp, webshop_ppp, vpc). These are the "no-promo" baseline used as anchor for forward-anchored plan price computation: plan_price = catalog × (1 - planned_discount).""" rows = db.execute(text(""" SELECT p.id AS product_id, ep.normal_retail_ppp::float AS retail_ppp, ep.normal_webshop_ppp::float AS webshop_ppp, sp.vpc::float AS vpc FROM dim_products p LEFT JOIN erp_prices ep ON ep.product_id = p.id LEFT JOIN sku_planning sp ON sp.product_id = p.id """)).mappings().all() return {int(r["product_id"]): { "retail_ppp": float(r["retail_ppp"]) if r["retail_ppp"] else 0.0, "webshop_ppp": float(r["webshop_ppp"]) if r["webshop_ppp"] else 0.0, "vpc": float(r["vpc"]) if r["vpc"] else 0.0, } for r in rows} def _load_vpc_map(db: Session) -> dict[int, float]: """Per-SKU VPC list price from sku_planning. This is the formal wholesale list price — KAMs negotiate below this. Stored frozen in snapshot so the bridge's Wholesale Rebate line is anchored to the plan-period VPC, not whatever live VPC happens to be when bridge runs.""" rows = db.execute(text(""" SELECT product_id, vpc::float AS vpc FROM sku_planning WHERE vpc IS NOT NULL AND vpc > 0 """)).all() return {int(pid): float(v) for pid, v in rows} def _load_plan_cost_map(db: Session) -> dict[int, float]: """Per-SKU plan cost = avg of last 3 calendar months of booked `purchase_value/quantity` from erp_transactions. Same source the Bridge's `_prior_booked_cost` uses, just precomputed for snapshot freeze.""" rows = db.execute(text(""" WITH last3 AS ( SELECT DISTINCT EXTRACT(YEAR FROM transaction_date)::int AS cy, EXTRACT(MONTH FROM transaction_date)::int AS cm FROM erp_transactions WHERE transaction_date >= now() - interval '4 months' ORDER BY cy DESC, cm DESC LIMIT 3 ), per_month AS ( SELECT et.product_id, EXTRACT(YEAR FROM et.transaction_date)::int AS cy, EXTRACT(MONTH FROM et.transaction_date)::int AS cm, NULLIF(SUM(et.quantity), 0)::float AS qty, SUM(et.purchase_value)::float AS pv FROM erp_transactions et WHERE et.purchase_value IS NOT NULL GROUP BY et.product_id, cy, cm ) SELECT product_id, AVG(pv / qty)::float AS plan_cost FROM per_month WHERE (cy, cm) IN (SELECT cy, cm FROM last3) AND qty > 0 GROUP BY product_id """)).all() return {int(pid): float(c) for pid, c in rows if c is not None} # ────────────────────────────────────────────────────────────────────── # Main snapshot build # ────────────────────────────────────────────────────────────────────── def build_monthly_snapshot(db: Session, year: int, month: int) -> dict: """Compute the per-SKU plan for a calendar month using both methodologies. Returns a dict ready for INSERT into monthly_plan_snapshots.""" month_key = year * 100 + month # 1) Per-SKU per-week qty plan (engine output baseline) qty_per_week = _load_plan_qty_per_sku_week(db) qty_per_week["year_week"] = qty_per_week["year"] * 100 + qty_per_week["week"] # Apply per-month calibration (e.g., retail -15% in May/June 2026) qty_per_week = _apply_month_calibration(qty_per_week, year, month) # 2) On-top inputs added at week granularity, channel-specific. # The wizard writes 'food retail' for the supermarket MP channel and # 'retail' for smaller retail — both belong to the same B2C bucket # on the plan side, so we sum them together for on_top_r. on_top = _load_on_top_inputs(db) on_top_w = ( on_top[on_top["channel"] == "wholesale"] .groupby(["product_id", "year_week"], as_index=False)["quantity"].sum() .set_index(["product_id", "year_week"])["quantity"].to_dict() ) on_top_r = ( on_top[on_top["channel"].isin(["retail", "food retail"])] .groupby(["product_id", "year_week"], as_index=False)["quantity"].sum() .set_index(["product_id", "year_week"])["quantity"].to_dict() ) # 3) RUC rates + NET prices per SKU per channel (for plan margin / revenue # valuation). All NET of VAT — matches Margin Bridge's actual_price. # Baseline = 8w trailing anchored to end-of-(month-1), non-promo only # (via erp_promo_weeks). See _load_ruc_rates docstring. ruc = _load_ruc_rates(db, plan_year=year, plan_month=month, window_weeks=8) ruc["retail_ruc_rate"] = ruc["ruc_r"] / ruc["qty_r"] ruc["wholesale_ruc_rate"] = ruc["ruc_w"] / ruc["qty_w"] ruc["retail_price"] = ruc["rev_r"] / ruc["qty_r"] ruc["wholesale_price"] = ruc["rev_w"] / ruc["qty_w"] rate_map = ruc.set_index("product_id").to_dict("index") # VPC (formal wholesale list) + plan cost (trailing 3-month booked) — # both frozen into the snapshot. Bridge reads them from here, not live. vpc_map = _load_vpc_map(db) plan_cost_map = _load_plan_cost_map(db) # Forward-anchored plan source: rabatne politike loaded into erp_promo_items # 1.5 months ahead. For SKUs with a planned promo in this month, plan # price = catalog × (1 − planned_discount_pct/100). For SKUs without a # curated plan, fall back to trailing baseline (rate_map below). planned_promo_map = _load_planned_promo_discounts(db, year, month) catalog_map = _load_catalog_prices(db) # 4) For each ISO week × calendar-month overlap, compute per-SKU # contribution under BOTH methodologies. weeks_in_month_df = _iso_weeks_in_month(year, month) days_total = days_in_calendar_month(year, month) per_sku: dict[str, dict] = {} sources_used: dict[str, int] = {} weeks_covered: list[str] = [] # Pre-build qty lookup: (sku, year, week) → dict qpw_by_key = qty_per_week.set_index(["sku", "year", "week"]).to_dict("index") # For gap fallback, nearest baseline per SKU nearest_baseline: dict[str, dict] = {} for sku, grp in qty_per_week.groupby("sku"): # Pick the row with smallest |week_diff| from any target week within month # Just store all for later lookup nearest_baseline[sku] = grp[["year", "week", "qty_total", "qty_ws", "qty_b2c", "source"]].to_dict("records") for (iy, iw, frac) in weeks_in_month_df: weeks_covered.append(f"CW{iw:02d}") # Day-fraction methodology: frac in [0, 1] for THIS month # Thursday rule: this entire week goes to this month if Thursday in this month thu_y, thu_m = _thursday_month(iy, iw) in_thursday_month = (thu_y == year and thu_m == month) # For each SKU known in qty_per_week OR in on_top, compute contribution all_skus = set(qty_per_week["sku"].unique()) # Also include SKUs that have on_top for this week (they may not be in # qty_per_week if engine didn't forecast them this run). pid_to_sku = dict(zip(qty_per_week["product_id"], qty_per_week["sku"])) yw_key = iy * 100 + iw for pid in set(on_top_w.keys()) | set(on_top_r.keys()): if pid[1] == yw_key and pid[0] in pid_to_sku: all_skus.add(pid_to_sku[pid[0]]) # NaN-safe float coerce — `or 0` doesn't catch NaN (NaN is truthy in # Python), so a single NaN qty would propagate through the whole # snapshot and break the JSONB insert (Postgres rejects NaN in jsonb). import math as _m def _f(v) -> float: if v is None: return 0.0 try: fv = float(v) except (TypeError, ValueError): return 0.0 if _m.isnan(fv) or _m.isinf(fv): return 0.0 return fv for sku in all_skus: # Get qty for this (sku, week) — with gap fallback entry = qpw_by_key.get((sku, iy, iw)) if entry is not None: qty_ws = _f(entry["qty_ws"]) qty_b2c = _f(entry["qty_b2c"]) src = entry["source"] else: # Gap — find nearest available week for this SKU rows = nearest_baseline.get(sku, []) if not rows: qty_ws = qty_b2c = 0.0 src = "missing" else: rows_sorted = sorted(rows, key=lambda r: abs( (int(r["year"]) - iy) * 52 + (int(r["week"]) - iw) )) nearest = rows_sorted[0] qty_ws = _f(nearest["qty_ws"]) qty_b2c = _f(nearest["qty_b2c"]) src = f"gap_fill_{nearest['source']}_CW{nearest['week']:02d}" sources_used[src] = sources_used.get(src, 0) + 1 # Find product_id for this sku (for on-top lookup) # Use the qty_per_week mapping pid = qty_per_week[qty_per_week["sku"] == sku]["product_id"].iloc[0] if \ sku in qty_per_week["sku"].values else None if pid is not None: pid = int(pid) ot_ws = float(on_top_w.get((pid, yw_key), 0)) ot_b2c = float(on_top_r.get((pid, yw_key), 0)) else: ot_ws = ot_b2c = 0.0 qty_ws_total = qty_ws + ot_ws qty_b2c_total = qty_b2c + ot_b2c # Initialise per_sku entry if sku not in per_sku: per_sku[sku] = { "sku": sku, "qty_ws_df": 0.0, "qty_b2c_df": 0.0, "qty_ws_tr": 0.0, "qty_b2c_tr": 0.0, "ws_ruc_eur_df": 0.0, "b2c_ruc_eur_df": 0.0, "ws_ruc_eur_tr": 0.0, "b2c_ruc_eur_tr": 0.0, "ws_revenue_eur_df": 0.0, "b2c_revenue_eur_df": 0.0, "on_top_ws_qty": 0.0, "on_top_b2c_qty": 0.0, "source_breakdown": {}, } sku_entry = per_sku[sku] # Day-fraction contribution sku_entry["qty_ws_df"] += qty_ws_total * frac sku_entry["qty_b2c_df"] += qty_b2c_total * frac # Thursday-rule contribution (full week if Thursday is in this month) if in_thursday_month: sku_entry["qty_ws_tr"] += qty_ws_total sku_entry["qty_b2c_tr"] += qty_b2c_total # On-tops — track separately (already included in totals above) sku_entry["on_top_ws_qty"] += ot_ws * frac sku_entry["on_top_b2c_qty"] += ot_b2c * frac sku_entry["source_breakdown"][src] = \ sku_entry["source_breakdown"].get(src, 0) + 1 # 5) Apply RUC rates and prices per SKU to value the qty total_qty = total_qty_ws = total_qty_b2c = 0.0 total_ruc_eur = total_ruc_ws = total_ruc_b2c = 0.0 total_revenue_eur = 0.0 import math def _safe(v) -> float: if v is None: return 0.0 if isinstance(v, float) and math.isnan(v): return 0.0 return float(v) # Plan wholesale rebate aggregate: SUM(plan_qty_ws × (VPC − realised_ws_price)) # Captures "expected KAM rebate vs list" frozen at lock — Margin Bridge # displays this as the Wholesale Rebate line. Anomaly above/below this # baseline is what surfaces as price_effect_ws downstream. plan_wholesale_rebate_eur = 0.0 for sku, sku_entry in per_sku.items(): # Find product_id and rates pid_row = qty_per_week[qty_per_week["sku"] == sku] if pid_row.empty: continue pid = int(pid_row["product_id"].iloc[0]) r = rate_map.get(pid, {}) ws_ruc_rate = _safe(r.get("wholesale_ruc_rate")) retail_ruc_rate = _safe(r.get("retail_ruc_rate")) ws_price_trail = _safe(r.get("wholesale_price")) # NET trailing-8w realised retail_price_trail = _safe(r.get("retail_price")) # NET trailing-8w realised vpc = _safe(vpc_map.get(pid)) # VPC list (frozen) plan_cost = _safe(plan_cost_map.get(pid)) # 3-month booked cost # ── Forward-anchored plan price ────────────────────────────────── # Plan price = standard catalog price × (1 − planned_discount/100) # catalog comes from erp_prices.normal_retail_ppp / sku_planning.vpc. # retail_ppp / webshop_ppp are stored GROSS-of-VAT (~25% HR), so we # multiply by 0.80 to get NET-of-VAT for parity with actual_price # (which uses tax_base, NET). # vpc is wholesale (B2B reverse-charge, already NET). # If SKU has a curated promo in this plan month, apply discount %. # Trailing fallback only used as last resort when catalog missing # AND no promo plan (rare — e.g. brand-new SKU with no list price). planned_disc = planned_promo_map.get(pid) cat = catalog_map.get(pid, {}) retail_ppp_gross = float(cat.get("retail_ppp") or 0) webshop_ppp_gross = float(cat.get("webshop_ppp") or 0) b2c_catalog_gross = retail_ppp_gross or webshop_ppp_gross b2c_catalog_net = b2c_catalog_gross * 0.80 factor = max(0.0, 1.0 - (planned_disc or 0) / 100.0) plan_price_source = f"promo:{planned_disc:.1f}%" if planned_disc else "catalog" if b2c_catalog_net > 0: retail_price = b2c_catalog_net * factor else: retail_price = retail_price_trail # last-resort fallback if not planned_disc: plan_price_source = "trailing_8w(no_catalog)" if vpc > 0: ws_price = vpc * factor else: ws_price = ws_price_trail # last-resort fallback # Day-fraction RUC + revenue (NET basis) sku_entry["ws_ruc_eur_df"] = sku_entry["qty_ws_df"] * ws_ruc_rate sku_entry["b2c_ruc_eur_df"] = sku_entry["qty_b2c_df"] * retail_ruc_rate sku_entry["ws_revenue_eur_df"] = sku_entry["qty_ws_df"] * ws_price sku_entry["b2c_revenue_eur_df"] = sku_entry["qty_b2c_df"] * retail_price # Thursday-rule RUC sku_entry["ws_ruc_eur_tr"] = sku_entry["qty_ws_tr"] * ws_ruc_rate sku_entry["b2c_ruc_eur_tr"] = sku_entry["qty_b2c_tr"] * retail_ruc_rate sku_entry["ws_ruc_rate"] = ws_ruc_rate sku_entry["retail_ruc_rate"] = retail_ruc_rate # Frozen plan prices/cost — Margin Bridge reads these instead of # going to live erp_prices / sku_planning. sku_entry["plan_price_ws"] = ws_price # post-rebate net WS rate sku_entry["plan_price_b2c"] = retail_price # NET B2C blended sku_entry["plan_cost"] = plan_cost sku_entry["vpc"] = vpc # formal list (info only) sku_entry["plan_price_source"]= plan_price_source # 'promo:X%' or 'trailing_8w' # Per-SKU plan rebate (only when VPC is known and above realised WS) if vpc > 0 and ws_price > 0 and vpc > ws_price: sku_entry["plan_rebate_per_u"] = vpc - ws_price plan_wholesale_rebate_eur += sku_entry["qty_ws_df"] * (vpc - ws_price) else: sku_entry["plan_rebate_per_u"] = 0.0 total_qty_ws += sku_entry["qty_ws_df"] total_qty_b2c += sku_entry["qty_b2c_df"] total_qty += (sku_entry["qty_ws_df"] + sku_entry["qty_b2c_df"]) total_ruc_ws += sku_entry["ws_ruc_eur_df"] total_ruc_b2c += sku_entry["b2c_ruc_eur_df"] total_ruc_eur += (sku_entry["ws_ruc_eur_df"] + sku_entry["b2c_ruc_eur_df"]) total_revenue_eur += (sku_entry["ws_revenue_eur_df"] + sku_entry["b2c_revenue_eur_df"]) # 6) Apply gross-up for non-planned SKUs (same approach as Revenue # Forecast page). Snapshot's planned-only per_sku_data stays untouched # for transparency, but the headline totals get scaled up by per-channel # ratio so they include non-planned contribution. grossup = _compute_grossup_ratios(db) ws_ratio = grossup["ws_ratio"] b2c_ratio = grossup["b2c_ratio"] total_qty_ws_gu = total_qty_ws * ws_ratio total_qty_b2c_gu = total_qty_b2c * b2c_ratio total_qty_gu = total_qty_ws_gu + total_qty_b2c_gu total_ruc_ws_gu = total_ruc_ws * ws_ratio total_ruc_b2c_gu = total_ruc_b2c * b2c_ratio total_ruc_eur_gu = total_ruc_ws_gu + total_ruc_b2c_gu # Revenue grossed up using a blend (weighted by qty contribution) total_revenue_eur_gu = total_revenue_eur if total_qty > 0: ws_qty_share = total_qty_ws / total_qty blended_ratio = (ws_qty_share * ws_ratio + (1 - ws_qty_share) * b2c_ratio) total_revenue_eur_gu = total_revenue_eur * blended_ratio # 7) Coverage — fraction of month's days covered by any source coverage_pct = 1.0 gap_filled = sum(1 for s in sources_used if s.startswith("gap_fill_")) note_parts = [] if gap_filled: note_parts.append( f"Gap-filled {gap_filled} (SKU × week) entries from nearest " f"baseline week (typically due to current-week forecast gap)." ) note_parts.append( f"Gross-up applied: WS×{ws_ratio:.3f}, B2C×{b2c_ratio:.3f} " f"({grossup['nonplanned_share_pct']:.1f}% non-planned share)." ) note_parts.append( f"Sources: {', '.join(f'{k}={v}' for k, v in sorted(sources_used.items()))}" ) # Gross-up the plan rebate too, so the displayed Wholesale Rebate plan # number is on the same scale as the (grossed-up) plan_margin total. plan_wholesale_rebate_eur_gu = plan_wholesale_rebate_eur * ws_ratio # ── HEADLINE ALIGNMENT WITH REVENUE FORECAST PAGE ────────────────── # The model-based totals computed above used trailing 8w RUC rates and # backtest forecast baselines for past weeks. Revenue Forecast uses # different methodology (actuals for past + forecast for future, # 4-week RUC rates). Per user May 2026 decision: locked plan headline # MUST equal Revenue Forecast for the same month, with only MCI's # fictive add-back as a deliberate difference. So we override the # headline totals with RF's view + MCI delta. from backend.services.demand_service import DemandService from datetime import datetime as _dt rf_label = _dt(year, month, 1).strftime("%b %Y") try: ds = DemandService(db) rf_rev = ds.get_revenue_forecast( view="revenue", source="all", months_filter=[rf_label], include_nonplanned=True, ) rf_ruc = ds.get_revenue_forecast( view="ruc", source="all", months_filter=[rf_label], include_nonplanned=True, ) rf_rev_total = sum((p.get("forecast") or 0) + (p.get("actual") or 0) for p in rf_rev["chart"]) rf_ruc_total = sum((p.get("forecast") or 0) + (p.get("actual") or 0) for p in rf_ruc["chart"]) except Exception: rf_rev_total = total_revenue_eur_gu rf_ruc_total = total_ruc_eur_gu # MCI fictive-addition uplift: when on_top_inputs has rows in this # month's ISO weeks that are NOT in the latest forecast run (i.e., # they were injected as fictive — e.g. MCI for May while MCI's real # forecasts.* row is in June), Revenue Forecast won't see them. Add # their value explicitly so the headline includes them. yws = [y * 100 + w for (y, w, _) in weeks_in_month_df] fictive = pd.read_sql(text(""" WITH latest_fc AS ( SELECT DISTINCT ON (product_id, year, week) product_id, year, week, on_top_wholesale, on_top_retail FROM forecasts WHERE (year * 100 + week) = ANY(:yws) ORDER BY product_id, year, week, run_id DESC ), sku_prices AS ( SELECT product_id, COALESCE(vpc, 0)::float AS plan_price_ws, COALESCE(normal_retail_ppp, 0)::float * 0.80 AS plan_price_b2c FROM sku_planning sp FULL OUTER JOIN erp_prices ep USING (product_id) ), ruc_rates AS ( SELECT product_id, COALESCE(NULLIF(SUM(CASE WHEN cm.channel='wholesale' THEN et.ruc_eur END),0) / NULLIF(SUM(CASE WHEN cm.channel='wholesale' THEN et.quantity END),0), 0)::float AS ruc_rate_ws, COALESCE(NULLIF(SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.ruc_eur END),0) / NULLIF(SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.quantity END),0), 0)::float AS ruc_rate_b2c FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id = et.channel_map_id WHERE et.transaction_date >= (CURRENT_DATE - INTERVAL '8 weeks') GROUP BY product_id ), ot AS ( SELECT product_id, year_week, SUM(CASE WHEN channel='wholesale' THEN quantity ELSE 0 END)::float AS ot_ws, SUM(CASE WHEN channel IN ('retail','food retail') THEN quantity ELSE 0 END)::float AS ot_mp FROM on_top_inputs WHERE year_week = ANY(:yws) GROUP BY product_id, year_week ) SELECT ot.product_id, ot.year_week, GREATEST(0, ot.ot_ws - COALESCE(f.on_top_wholesale, 0)) AS extra_ws, GREATEST(0, ot.ot_mp - COALESCE(f.on_top_retail, 0)) AS extra_mp, sp.plan_price_ws, sp.plan_price_b2c, r.ruc_rate_ws, r.ruc_rate_b2c FROM ot LEFT JOIN latest_fc f ON f.product_id = ot.product_id AND f.year*100 + f.week = ot.year_week LEFT JOIN sku_prices sp ON sp.product_id = ot.product_id LEFT JOIN ruc_rates r ON r.product_id = ot.product_id """), db.bind, params={"yws": yws}) if not fictive.empty: fictive_rev = float( (fictive["extra_ws"] * fictive["plan_price_ws"].fillna(0)).sum() + (fictive["extra_mp"] * fictive["plan_price_b2c"].fillna(0)).sum() ) fictive_ruc = float( (fictive["extra_ws"] * fictive["ruc_rate_ws"].fillna(0)).sum() + (fictive["extra_mp"] * fictive["ruc_rate_b2c"].fillna(0)).sum() ) rf_rev_total += fictive_rev rf_ruc_total += fictive_ruc if fictive_rev > 0 or fictive_ruc > 0: note_parts.append( f"+ fictive-add uplift €{fictive_rev:,.0f} revenue / €{fictive_ruc:,.0f} RUC " f"(on_top_inputs rows not in latest forecasts — typically a buyer/promo " f"re-added for plan purposes after being moved out)." ) # Aligned totals — overrides model-based for the headline. Per-SKU # data still drives Margin Bridge offenders/contributors so we leave # that intact (rounding-level diff vs headline acceptable). aligned_revenue = float(rf_rev_total) aligned_ruc = float(rf_ruc_total) # Preserve the WS/B2C split ratio from the model-based calc so the # split fields stay coherent with the new headline. rev_split_ws_pct = ( total_revenue_eur_gu # avoid div by 0 and ( (total_qty_ws_gu and (total_revenue_eur_gu > 0) and (total_qty_ws_gu / max(total_qty_gu, 1))) or 0.5 ) ) ruc_split_ws_pct = ( (total_ruc_ws_gu / total_ruc_eur_gu) if total_ruc_eur_gu > 0 else 0.5 ) qty_split_ws_pct = ( (total_qty_ws_gu / total_qty_gu) if total_qty_gu > 0 else 0.5 ) aligned_qty_ws = total_qty_gu * qty_split_ws_pct aligned_qty_b2c = total_qty_gu * (1 - qty_split_ws_pct) aligned_ruc_ws = aligned_ruc * ruc_split_ws_pct aligned_ruc_b2c = aligned_ruc * (1 - ruc_split_ws_pct) note_parts.append( f"Headline aligned with Revenue Forecast page methodology: " f"actuals + remaining-week forecast, planned-SKU grossed up. " f"per_sku_data retains model-based view for Margin Bridge." ) # ── RETAIL TARGET CALIBRATION ───────────────────────────────────── # `retail_pct_above_mgmt`: pin the channel-breakdown Retail row to # mgmt_plan_retail × (1 + pct). Engine consistently over-predicts # retail for Polleo's 2026 trajectory — mgmt commitment is the # ground truth, and we plan 10% above it as Demand's # stretch-but-achievable line. WS and webshop are NOT touched # (those engine values are trusted). # # Mechanics: compute current implied retail (same formula # sop_monthly_service uses), pin to target, keep WS RUC at engine # value, and set new b2c_ruc such that b2c_ruc × retail_share = target. # The reduction propagates into aligned_ruc, aligned_revenue, and the # split ratios so sop_monthly_service (which reads stored # total_ruc_ws / total_ruc_b2c) shows the calibrated split directly. cal = CALIBRATION_BY_MONTH.get(month_key, {}) retail_pct = cal.get("retail_pct_above_mgmt") if retail_pct is not None: RETAIL_SHARE_IN_B2C = 0.86 # matches sop_monthly_service trailing-13w mgmt_retail_ruc = float(db.execute(text(""" SELECT COALESCE(SUM(ruc_plan), 0)::float FROM management_plan_lines WHERE month_key = :mk AND channel IN ('Retail domestic', 'Retail international') """), {"mk": month_key}).scalar() or 0) if mgmt_retail_ruc > 0: target_retail = mgmt_retail_ruc * (1.0 + float(retail_pct)) current_retail = aligned_ruc * (1.0 - ruc_split_ws_pct) * RETAIL_SHARE_IN_B2C if current_retail > target_retail and aligned_ruc > 0: ws_eur_keep = aligned_ruc * ruc_split_ws_pct # WS unchanged new_b2c_eur = target_retail / RETAIL_SHARE_IN_B2C new_aligned_ruc = ws_eur_keep + new_b2c_eur # Revenue scales by the same RUC ratio (assume retail's # RUC margin doesn't change with volume reduction). ruc_scale = new_aligned_ruc / aligned_ruc new_aligned_revenue = aligned_revenue * ruc_scale # Recompute split ratios to reflect the new totals new_ruc_ws_pct = ws_eur_keep / new_aligned_ruc aligned_ruc = new_aligned_ruc aligned_revenue = new_aligned_revenue ruc_split_ws_pct = new_ruc_ws_pct aligned_ruc_ws = aligned_ruc * ruc_split_ws_pct # = ws_eur_keep aligned_ruc_b2c = aligned_ruc * (1 - ruc_split_ws_pct) # = new_b2c_eur note_parts.append( f"Retail pinned to mgmt × {1+retail_pct:.2f} = " f"€{target_retail:,.0f} RUC (reduced from €{current_retail:,.0f}). " f"Headline lowered by €{(current_retail-target_retail):,.0f} RUC. " f"WS €{ws_eur_keep:,.0f} unchanged; B2C now €{new_b2c_eur:,.0f}." ) return { "month_key": month_key, "total_qty": round(total_qty_gu, 2), "total_qty_ws": round(aligned_qty_ws, 2), "total_qty_b2c": round(aligned_qty_b2c, 2), "total_revenue_eur": round(aligned_revenue, 2), "total_ruc_eur": round(aligned_ruc, 2), "total_ruc_ws": round(aligned_ruc_ws, 2), "total_ruc_b2c": round(aligned_ruc_b2c, 2), "per_sku_data": list(per_sku.values()), # planned SKUs only — not grossed "n_skus": len(per_sku), "coverage_pct": coverage_pct, "weeks_covered": ", ".join(sorted(set(weeks_covered))), "methodology_notes": " ".join(note_parts), "grossup_ws_ratio": ws_ratio, "grossup_b2c_ratio": b2c_ratio, "nonplanned_share_pct": grossup["nonplanned_share_pct"], "plan_wholesale_rebate_eur": round(plan_wholesale_rebate_eur_gu, 2), } def save_snapshot( db: Session, snapshot: dict, locked_by_id: Optional[int] = None, label: Optional[str] = None, ) -> int: """Persist a built snapshot to monthly_plan_snapshots. If a snapshot for the same month_key already exists, raises an error (snapshots are immutable — to redo, delete the existing row explicitly first).""" import json import math as _math # Safety net — Postgres jsonb rejects NaN/Inf. Scrub the snapshot dict # so any straggler NaN (e.g. from per-SKU rate divisions where qty=0) # gets nulled out before the INSERT. def _scrub(v): if isinstance(v, float): return 0.0 if (_math.isnan(v) or _math.isinf(v)) else v if isinstance(v, dict): return {k: _scrub(x) for k, x in v.items()} if isinstance(v, list): return [_scrub(x) for x in v] return v snapshot = _scrub(snapshot) existing = db.execute(text( "SELECT id FROM monthly_plan_snapshots WHERE month_key = :mk" ), {"mk": snapshot["month_key"]}).scalar() if existing: raise ValueError( f"Snapshot for month_key={snapshot['month_key']} already exists " f"(id={existing}). Snapshots are immutable; delete the existing " "row first if you want to redo it." ) # Get latest run + cycle for FK linkage forecast_run_id = db.execute(text( "SELECT MAX(id) FROM forecast_runs" )).scalar() cycle_id = db.execute(text( "SELECT MAX(id) FROM sop_cycles" )).scalar() snapshot_id = db.execute(text(""" INSERT INTO monthly_plan_snapshots (month_key, cycle_id, forecast_run_id, locked_by_id, label, total_qty, total_qty_ws, total_qty_b2c, total_revenue_eur, total_ruc_eur, total_ruc_ws, total_ruc_b2c, per_sku_data, coverage_pct, n_skus, weeks_covered, methodology_notes, grossup_ws_ratio, grossup_b2c_ratio, nonplanned_share_pct, plan_wholesale_rebate_eur) VALUES (:month_key, :cycle_id, :forecast_run_id, :locked_by_id, :label, :total_qty, :total_qty_ws, :total_qty_b2c, :total_revenue_eur, :total_ruc_eur, :total_ruc_ws, :total_ruc_b2c, CAST(:per_sku_data AS jsonb), :coverage_pct, :n_skus, :weeks_covered, :methodology_notes, :grossup_ws_ratio, :grossup_b2c_ratio, :nonplanned_share_pct, :plan_wholesale_rebate_eur) RETURNING id """), { **{k: v for k, v in snapshot.items() if k != "per_sku_data"}, "cycle_id": cycle_id, "forecast_run_id": forecast_run_id, "locked_by_id": locked_by_id, "label": label, "per_sku_data": json.dumps(snapshot["per_sku_data"]), }).scalar() db.commit() return int(snapshot_id) def get_snapshot(db: Session, month_key: int) -> Optional[dict]: """Fetch a snapshot by month_key. Returns dict or None.""" r = db.execute(text(""" SELECT id, month_key, cycle_id, forecast_run_id, locked_by_id, locked_at, label, total_qty, total_qty_ws, total_qty_b2c, total_revenue_eur, total_ruc_eur, total_ruc_ws, total_ruc_b2c, per_sku_data, coverage_pct, n_skus, weeks_covered, methodology_notes, grossup_ws_ratio, grossup_b2c_ratio, nonplanned_share_pct, plan_wholesale_rebate_eur FROM monthly_plan_snapshots WHERE month_key = :mk """), {"mk": month_key}).mappings().first() return dict(r) if r else None def list_snapshots(db: Session) -> list[dict]: """List all snapshots (without the heavy per_sku_data column).""" return [dict(r) for r in db.execute(text(""" SELECT id, month_key, cycle_id, forecast_run_id, locked_by_id, locked_at, label, total_qty, total_qty_ws, total_qty_b2c, total_revenue_eur, total_ruc_eur, total_ruc_ws, total_ruc_b2c, coverage_pct, n_skus, weeks_covered, methodology_notes, grossup_ws_ratio, grossup_b2c_ratio, nonplanned_share_pct, plan_wholesale_rebate_eur FROM monthly_plan_snapshots ORDER BY month_key DESC """)).mappings()]