"""Backfill erp_costs.ruc with the realized average margin per unit. Previously erp_costs.ruc was a static ERP-provided number (one figure per SKU, didn't match real channel mix). Now we derive it directly from erp_transactions: ruc_per_unit = SUM(ruc_eur) / SUM(quantity) across every sales document for that SKU. Two windows are computed: • RECENT (last 13 weeks) — what UI labels as "realized 13w" • ALL-TIME — broader coverage, stored as the canonical value Strategy: store the recent value where available, fall back to all-time for SKUs with no recent sales. Write the same numbers back to data/sku_costs.csv so the next migrate_remaining run stays consistent. Run: python scripts/recompute_ruc.py """ from __future__ import annotations from pathlib import Path import pandas as pd from sqlalchemy import text from backend.models.database import SessionLocal ROOT = Path(__file__).resolve().parents[1] SKU_COSTS = ROOT / "data" / "sku_costs.csv" AUDIT = ROOT / "data" / "_ruc_recompute_audit.csv" def aggregate_ruc(db) -> pd.DataFrame: """Per product_id: realized RUC per unit, recent + all-time.""" print(" reading erp_transactions ...") # INNER JOIN lookup_channel_map drops transactions whose doc_type is # not in our 10-row sales-channel mapping (e.g. returns, refunds, # internal transfers). Those rows have RUC near zero and would drag # the per-unit average toward zero — matches the filter the live # demand_repo.get_sku_pricing query uses. rows = db.execute(text(""" SELECT et.product_id, p.sku, SUM(CASE WHEN et.transaction_date >= (CURRENT_DATE - INTERVAL '13 weeks') THEN et.quantity ELSE 0 END)::float AS qty_13w, SUM(CASE WHEN et.transaction_date >= (CURRENT_DATE - INTERVAL '13 weeks') THEN et.ruc_eur ELSE 0 END)::float AS ruc_13w, SUM(et.quantity)::float AS qty_all, SUM(et.ruc_eur)::float AS ruc_all, COUNT(*)::int AS n_docs 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 GROUP BY et.product_id, p.sku """)).mappings().all() df = pd.DataFrame([dict(r) for r in rows]) print(f" {len(df):,} SKUs with at least one transaction") def _avg(num, den): if den is None or den == 0: return None return float(num) / float(den) df["ruc_per_unit_13w"] = df.apply(lambda r: _avg(r["ruc_13w"], r["qty_13w"]), axis=1) df["ruc_per_unit_all"] = df.apply(lambda r: _avg(r["ruc_all"], r["qty_all"]), axis=1) # Canonical: recent if available (>= 5 units in 13w), else all-time def _canonical(r): if r["qty_13w"] and r["qty_13w"] >= 5 and r["ruc_per_unit_13w"] is not None: return r["ruc_per_unit_13w"], "13w" if r["ruc_per_unit_all"] is not None: return r["ruc_per_unit_all"], "all_time" return None, "none" can = df.apply(_canonical, axis=1) df["ruc_canonical"] = [c[0] for c in can] df["ruc_window"] = [c[1] for c in can] return df def update_db(db, df: pd.DataFrame) -> tuple[int, int]: """UPDATE erp_costs.ruc per product. INSERT a stub row for SKUs that have transactions but no erp_costs row (so RUC is preserved even when cost_price is unknown).""" n_updated = 0 n_inserted = 0 rows_to_update = df[df["ruc_canonical"].notna()][ ["product_id", "ruc_canonical"] ].to_dict("records") # Existing product_ids in erp_costs existing = {int(r["product_id"]) for r in db.execute(text( "SELECT product_id FROM erp_costs" )).mappings()} for r in rows_to_update: pid = int(r["product_id"]) ruc = round(float(r["ruc_canonical"]), 4) if pid in existing: db.execute(text(""" UPDATE erp_costs SET ruc = :ruc WHERE product_id = :pid """), {"ruc": ruc, "pid": pid}) n_updated += 1 else: db.execute(text(""" INSERT INTO erp_costs (product_id, cost_price, ruc, valid_from) VALUES (:pid, NULL, :ruc, CURRENT_DATE) """), {"pid": pid, "ruc": ruc}) n_inserted += 1 db.commit() return n_updated, n_inserted def update_csv(db, df: pd.DataFrame) -> int: """Rewrite data/sku_costs.csv with the new RUC values so the next migrate_remaining run stays consistent. We keep existing cost_price untouched — only the ruc column is replaced.""" if not SKU_COSTS.exists(): print(f" [skip] {SKU_COSTS} missing") return 0 old = pd.read_csv(SKU_COSTS) new_ruc = df.set_index("sku")["ruc_canonical"] old["ruc"] = old["sku"].map(new_ruc).fillna(old["ruc"]).round(4) old.to_csv(SKU_COSTS, index=False) return int(old["ruc"].notna().sum()) def main() -> None: db = SessionLocal() try: print("Step 1 — aggregate RUC from erp_transactions") df = aggregate_ruc(db) # Print headline stats + a few samples for spot-check valid = df[df["ruc_canonical"].notna()] print(f"\n SKUs with RUC: {len(valid):,} " f"({(df['ruc_window']=='13w').sum():,} via 13w, " f"{(df['ruc_window']=='all_time').sum():,} via all-time)") print(f" median ruc/unit: €{valid['ruc_canonical'].median():.2f}") print(f" mean ruc/unit: €{valid['ruc_canonical'].mean():.2f}") sample_skus = ["POL12885", "POL12887", "POL12862", "POL09753"] print("\n Spot check:") for sku in sample_skus: row = df[df["sku"] == sku] if row.empty: print(f" {sku}: no transactions") continue r = row.iloc[0] print(f" {sku:<10} ruc_13w={r['ruc_per_unit_13w']!s:<8} " f"ruc_all={r['ruc_per_unit_all']!s:<8} " f"canonical={r['ruc_canonical']:.2f} ({r['ruc_window']})") # Save audit CSV df[["sku", "n_docs", "qty_13w", "ruc_13w", "ruc_per_unit_13w", "qty_all", "ruc_all", "ruc_per_unit_all", "ruc_canonical", "ruc_window"]].round(4).to_csv(AUDIT, index=False) print(f"\n audit written: {AUDIT}") print("\nStep 2 — UPDATE erp_costs.ruc") n_up, n_ins = update_db(db, df) print(f" updated {n_up:,} rows, inserted {n_ins:,} new rows") print("\nStep 3 — refresh sku_costs.csv") n_csv = update_csv(db, df) print(f" {n_csv:,} ruc values written to {SKU_COSTS}") # Final DB stats print("\n=== DB after recompute ===") n_rows = db.execute(text("SELECT COUNT(*) FROM erp_costs")).scalar() n_ruc = db.execute(text("SELECT COUNT(*) FROM erp_costs WHERE ruc IS NOT NULL AND ruc > 0")).scalar() med_ruc = db.execute(text("SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY ruc) " "FROM erp_costs WHERE ruc > 0")).scalar() print(f" erp_costs rows: {n_rows:,}") print(f" with ruc > 0: {n_ruc:,}") print(f" median ruc: €{float(med_ruc):.2f}") finally: db.close() if __name__ == "__main__": main()