"""Side-by-side verification: app.py (Streamlit / CSV) vs FastAPI (Postgres). Runs all PART 1 / PART 2 checks from the verification spec, replicating the EXACT load_sales_data() logic from app.py:92 against backend.repositories.demand_repo + raw Postgres SQL. Output is a plain-text report — copy-pasteable into the conversation. """ from __future__ import annotations import sys from pathlib import Path import pandas as pd from sqlalchemy import text # Force UTF-8 stdout so the unicode box-drawing chars and arrows print on # Windows consoles (default codepage is cp1252). try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass ROOT = Path(__file__).resolve().parent sys.path.insert(0, str(ROOT)) from db.connection import get_engine # noqa: E402 DATA_DIR = ROOT / "data" # -------------------------------------------------------------------------- # OLD path — replicate app.py:92 load_sales_data() verbatim # -------------------------------------------------------------------------- def load_sales_data() -> pd.DataFrame: sc = pd.read_csv(DATA_DIR / "sales_clean.csv") sc["yw"] = sc["year"] * 100 + sc["week"] cat_df = pd.read_csv(DATA_DIR / "sku_category_map.csv") cat_map = dict(zip(cat_df["sku"], cat_df["cat"])) name_map = dict(zip(cat_df["sku"], cat_df["name"])) if "name" in cat_df.columns else {} sc["cat"] = sc["sku"].map(cat_map).fillna("OTHER") sc["name"] = sc["sku"].map(name_map).fillna("") plan = pd.read_csv(DATA_DIR / "sku_plan_list.csv") ozn_map = dict(zip(plan["sku"], plan["oznaka"])) sc["oznaka"] = sc["sku"].map(ozn_map).fillna("UNCLASSIFIED") if "total_xyz" in plan.columns: xyz_map = dict(zip(plan["sku"], plan["total_xyz"])) sc["xyz"] = sc["sku"].map(xyz_map).fillna("N/A") else: sc["xyz"] = "N/A" return sc def row(label: str, old, new, note: str = "") -> str: """Format a side-by-side row. Use ✓ / ✗ icons for match.""" matches = old == new icon = "✓" if matches else "✗" o = f"{old:,}" if isinstance(old, int) else str(old) n = f"{new:,}" if isinstance(new, int) else str(new) pad_label = label.ljust(40) pad_old = str(o).ljust(20) pad_new = str(n).ljust(20) suffix = f" {note}" if note else "" return f" {icon} {pad_label} {pad_old} {pad_new}{suffix}" def main() -> int: print("=" * 100) print("STREAMLIT (sales_clean.csv) vs POSTGRES (erp_transactions → v_sales_weekly)") print("=" * 100) print() # Load both sides print("Loading sales_clean.csv (OLD path)...") sc = load_sales_data() plan = pd.read_csv(DATA_DIR / "sku_plan_list.csv") bt_old = pd.read_csv(DATA_DIR / "backtest_fa.csv") print("Connecting to Postgres (NEW path)...") engine = get_engine() # ---------------------------------------------------------------------- # PART 1 — counts # ---------------------------------------------------------------------- print() print("─" * 100) print("PART 1 — DATA COUNTS") print("─" * 100) print(f" {'CHECK':<40} {'OLD (csv)':<20} {'NEW (postgres)':<20}") print() # 1. unique SKUs with sales old_skus = sc["sku"].nunique() with engine.connect() as conn: new_skus_view = conn.execute(text( "SELECT COUNT(DISTINCT product_id) FROM v_sales_weekly")).scalar() new_skus_tx = conn.execute(text( "SELECT COUNT(DISTINCT product_id) FROM erp_transactions")).scalar() print(row("1. unique SKUs in sales", old_skus, new_skus_view, f"(erp_transactions has {new_skus_tx:,} including transfers)")) # 2. weekly rows old_rows = len(sc) with engine.connect() as conn: new_rows = conn.execute(text("SELECT COUNT(*) FROM v_sales_weekly")).scalar() print(row("2. weekly sales rows", old_rows, new_rows)) # 3. SKUs in planning old_plan = len(plan) with engine.connect() as conn: new_plan = conn.execute(text("SELECT COUNT(*) FROM sku_planning")).scalar() print(row("3. SKUs in sku_planning", old_plan, new_plan)) # 4. date range (year*100+week) old_yw_min = int((sc["year"] * 100 + sc["week"]).min()) old_yw_max = int((sc["year"] * 100 + sc["week"]).max()) with engine.connect() as conn: rng = conn.execute(text( "SELECT MIN(year*100+week), MAX(year*100+week) FROM v_sales_weekly")).fetchone() new_yw_min, new_yw_max = int(rng[0]), int(rng[1]) print(row("4a. earliest yw", old_yw_min, new_yw_min)) print(row("4b. latest yw", old_yw_max, new_yw_max)) # 5. total qty per channel old_r = int(sc["qty_retail"].sum()) old_w = int(sc["qty_webshop"].sum()) old_s = int(sc["qty_wholesale"].sum()) old_t = int(sc["qty_total"].sum()) with engine.connect() as conn: sums = conn.execute(text( "SELECT SUM(qty_retail), SUM(qty_webshop), SUM(qty_wholesale), SUM(qty_total) " "FROM v_sales_weekly")).fetchone() new_r, new_w, new_s, new_t = [int(x or 0) for x in sums] print(row("5a. SUM qty_retail", old_r, new_r)) print(row("5b. SUM qty_webshop", old_w, new_w)) print(row("5c. SUM qty_wholesale", old_s, new_s)) print(row("5d. SUM qty_total", old_t, new_t, f"(diff: {new_t - old_t:+,} = {(new_t - old_t) / old_t * 100:+.1f}%)" if old_t else "")) # 6. backtest rows old_bt = len(bt_old) with engine.connect() as conn: new_bt = conn.execute(text("SELECT COUNT(*) FROM backtest_results")).scalar() print(row("6. backtest rows", old_bt, new_bt)) # 7. categories old_cats = sorted(sc["cat"].dropna().unique().tolist()) with engine.connect() as conn: new_cats = [r[0] for r in conn.execute(text( "SELECT name FROM dim_categories ORDER BY name")).all() if r[0]] new_cats_sorted = sorted(new_cats) print(row("7. category count", len(old_cats), len(new_cats_sorted))) only_old = set(old_cats) - set(new_cats_sorted) only_new = set(new_cats_sorted) - set(old_cats) if only_old: print(f" ! ONLY in OLD: {sorted(only_old)}") if only_new: print(f" ! ONLY in NEW: {sorted(only_new)}") # ---------------------------------------------------------------------- # PART 2 — five Gold SKUs, recent 13 weeks # ---------------------------------------------------------------------- print() print("─" * 100) print("PART 2 — FIVE GOLD SKUS · last 13 weeks, channel split") print("─" * 100) # Pick 5 Gold SKUs with sales in last 8 weeks sc_gold = sc[sc["oznaka"].astype(str).str.contains("GOLD", na=False)].copy() sc_gold["yw"] = sc_gold["year"] * 100 + sc_gold["week"] max_yw = int(sc_gold["yw"].max()) cutoff_yw = max_yw - 8 # rough: last ~8 weeks recent = sc_gold[sc_gold["yw"] >= cutoff_yw] top_skus = (recent.groupby("sku")["qty_total"].sum() .sort_values(ascending=False).head(5).index.tolist()) # 13-week cutoff for the comparison cutoff_13 = max_yw - 13 for sku in top_skus: print() print(f" SKU: {sku}") # OLD old_sub = sc[(sc["sku"] == sku) & (sc["yw"] >= cutoff_13)] o_qt = int(old_sub["qty_total"].sum()) o_qr = int(old_sub["qty_retail"].sum()) o_qw = int(old_sub["qty_webshop"].sum()) o_qs = int(old_sub["qty_wholesale"].sum()) # NEW (same yw window from v_sales_weekly) with engine.connect() as conn: r = conn.execute(text(""" SELECT COALESCE(SUM(v.qty_total), 0), COALESCE(SUM(v.qty_retail), 0), COALESCE(SUM(v.qty_webshop), 0), COALESCE(SUM(v.qty_wholesale), 0) FROM v_sales_weekly v JOIN dim_products p ON v.product_id = p.id WHERE p.sku = :sku AND v.year * 100 + v.week >= :cutoff """), {"sku": sku, "cutoff": cutoff_13}).fetchone() n_qt = int(r[0]); n_qr = int(r[1]); n_qw = int(r[2]); n_qs = int(r[3]) # OLD: also from erp_transactions DIRECTLY (bypass the channel CASE) to # see whether the v_sales_weekly aggregation is what's causing drift. with engine.connect() as conn: r2 = conn.execute(text(""" SELECT COALESCE(SUM(t.quantity), 0) FROM erp_transactions t JOIN dim_products p ON t.product_id = p.id WHERE p.sku = :sku AND EXTRACT(ISOYEAR FROM t.transaction_date) * 100 + EXTRACT(WEEK FROM t.transaction_date) >= :cutoff """), {"sku": sku, "cutoff": cutoff_13}).scalar() or 0 tx_qt = int(r2) print(row(" qty_total", o_qt, n_qt, f"erp_tx.SUM(quantity) = {tx_qt:,}")) print(row(" qty_retail", o_qr, n_qr)) print(row(" qty_webshop", o_qw, n_qw)) print(row(" qty_wholesale", o_qs, n_qs)) # ---------------------------------------------------------------------- # PART 3 — formulas (verbatim from app.py:4178+, _per_week_avg_metrics:4430) # ---------------------------------------------------------------------- print() print("─" * 100) print("PART 3 — FORMULAS") print("─" * 100) print(""" STREAMLIT (app.py:4172-4192) — per-row enrichment (after `actual > 0` filter): error = |F - A| fa = max(0, 1 - |F-A| / A) # 0..1, scaled ×100 at display fa_signed = F / A # 1.0 = perfect, >1 over, <1 under bias = (F - A) / A # computed in agg from sum_F, sum_A hit = (|F-A| / max(A, 1) <= 0.30) # bool / 0-1 int STREAMLIT — HEADLINE weekly view (`_per_week_avg_metrics`, app.py:4430-4442): Group rows by week → sum F, sum A per week. Drop weeks where sum_A = 0. Per week: fa_w = max(0, 1 - |sum_F - sum_A| / sum_A) × 100 fa_signed_w = sum_F / sum_A × 100 bias_w = (sum_F - sum_A) / sum_A × 100 HEADLINE = MEAN of per-week values across the selected weeks. Hit rate = MEAN of per-row hits × 100. STREAMLIT — MONTHLY view (app.py:4624-4636) uses SUM-THEN-DIVIDE on totals. BACKEND (backend/services/demand_service.py: get_fa_summary, _agg): Per-row: fa = max(0, 1 - |F-A|/A) × 100 ✓ matches (just scaled here, app does it at display) Per-row: fa_signed = F/A × 100 ✓ matches Per-row: hit = (|F-A|/A <= 0.30) ✓ matches (we don't use clip(1) but actual > 0 already filtered) HEADLINE: overall_fa = SUM(fa) / N ← simple mean of per-ROW FAs overall_fa_signed = sum_F / sum_A × 100 ← SUM-then-divide (totals) overall_bias = (sum_F - sum_A) / sum_A × 100 ← SUM-then-divide (totals) hit_rate = SUM(hit) / N × 100 ← simple mean of per-row hits ✓ ┌──────────────────────────────────────────────────────────────────────────┐ │ MISMATCH — headline FA / FA-signed / BIAS in the BACKEND do NOT match │ │ either Streamlit "weekly" or "monthly" aggregation rule. │ │ │ │ Backend vs Streamlit weekly vs Streamlit monthly │ │ ---------- --- ------------------ --- ------------------- │ │ FA mean-of-row weekly-avg totals │ │ FA signed totals weekly-avg totals │ │ BIAS totals weekly-avg totals │ │ Hit rate mean-of-row mean-of-row mean-of-row ✓ │ │ │ │ To match the Streamlit "weekly" view (the default Forecast Accuracy │ │ tab), the backend should aggregate by week first and average across │ │ weeks for FA, FA signed, and BIAS. This is a fix for prompt 1C, not 1A. │ └──────────────────────────────────────────────────────────────────────────┘ """) return 0 if __name__ == "__main__": sys.exit(main())