"""Analytics service — read-only business-performance views (decision support).

Built on the same physical tables the operational modules use (no datamart):
sales = erp_transactions x lookup_channel_map, category = dim_products ->
dim_categories, country = dim_stores (retail/web) or dim_partners (wholesale),
margin = erp_transactions.ruc_eur.

First view: category performance over time, sliceable by channel and country,
with YoY growth (comparing the months present in the latest year to the same
months a year earlier) and contribution share.
"""
from __future__ import annotations

from typing import Optional

import pandas as pd
from sqlalchemy import text
from sqlalchemy.orm import Session

# Net revenue: prefer tax_base; fall back to 80% of gross when tax_base is 0/null.
_REV = "COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80)"
_COUNTRY = ("CASE WHEN cm.channel IN ('retail','webshop') "
            "THEN COALESCE(ds.country,'UNK') ELSE COALESCE(dp.country,'UNK') END")

# Non-product categories excluded from the sales/growth view: services /
# mediaplans (HR/SI/AT language variants), packaging materials, and the misc
# gift/promo bucket. These are real financial rows but not product demand.
EXCLUDE_CATEGORIES = ("USLUGE", "STORITVE", "DIENSTLEISTUNGEN",
                      "PARIS", "OTHER", "GADGETS")


def category_performance(
    db: Session, *, granularity: str = "month",
    channel: Optional[str] = None, country: Optional[str] = None,
    category: Optional[str] = None,
) -> dict:
    """Units / net revenue / margin by category over time, optionally filtered to
    one channel and/or country. Returns per-category monthly (or weekly) series
    plus YoY growth and share."""
    period_sql = ("to_char(et.transaction_date,'YYYY-MM')" if granularity == "month"
                  else "(EXTRACT(isoyear FROM et.transaction_date)::int*100"
                       "+EXTRACT(week FROM et.transaction_date)::int)::text")

    rows = db.execute(text(f"""
        SELECT COALESCE(dc.name,'(uncategorised)') AS category,
               {period_sql} AS period,
               SUM(et.quantity) AS units,
               SUM({_REV})      AS revenue,
               SUM(et.ruc_eur)  AS margin
        FROM erp_transactions et
        JOIN lookup_channel_map cm ON cm.id = et.channel_map_id
        JOIN dim_products p ON p.id = et.product_id
        LEFT JOIN dim_categories dc ON dc.id = p.category_id
        LEFT JOIN dim_stores ds ON ds.id = et.store_id
        LEFT JOIN dim_partners dp ON dp.id = et.partner_id
        WHERE et.quantity > 0
          AND (:channel IS NULL OR cm.channel = :channel)
          AND (:country IS NULL OR {_COUNTRY} = :country)
          AND (:category IS NULL OR dc.name = :category)
          AND (dc.name IS NULL OR NOT (dc.name = ANY(:excl)))
        GROUP BY 1, 2
        ORDER BY 1, 2
    """), {"channel": channel, "country": country, "category": category,
           "excl": list(EXCLUDE_CATEGORIES)}).mappings().all()

    df = pd.DataFrame(rows, columns=["category", "period", "units", "revenue", "margin"])
    if df.empty:
        return {"granularity": granularity, "channel": channel, "country": country,
                "periods": [], "categories": [], "totals": {}}
    for c in ("units", "revenue", "margin"):
        df[c] = df[c].astype(float)

    periods = sorted(df["period"].unique())

    # YoY: months present in the latest calendar year vs the same months prior year.
    yoy = {}
    if granularity == "month":
        df["y"] = df["period"].str[:4].astype(int)
        df["m"] = df["period"].str[5:7]
        years = sorted(df["y"].unique())
        if len(years) >= 2:
            cur, prev = years[-1], years[-2]
            common = sorted(set(df.loc[df.y == cur, "m"]) & set(df.loc[df.y == prev, "m"]))
            cur_df = df[(df.y == cur) & (df.m.isin(common))]
            prev_df = df[(df.y == prev) & (df.m.isin(common))]
            cur_rev = cur_df.groupby("category")["revenue"].sum()
            prev_rev = prev_df.groupby("category")["revenue"].sum()
            for cat in df["category"].unique():
                p = float(prev_rev.get(cat, 0.0)); c0 = float(cur_rev.get(cat, 0.0))
                yoy[cat] = round((c0 / p - 1.0) * 100, 1) if p > 0 else None
            tot_yoy = (round((float(cur_rev.sum()) / float(prev_rev.sum()) - 1.0) * 100, 1)
                       if prev_rev.sum() > 0 else None)
            yoy_window = {"current_year": int(cur), "prev_year": int(prev), "months": common}
        else:
            tot_yoy, yoy_window = None, None
    else:
        tot_yoy, yoy_window = None, None

    total_rev_all = float(df["revenue"].sum()) or 1.0
    cats = []
    for cat, g in df.groupby("category"):
        g = g.sort_values("period")
        series = [{"period": r["period"], "units": round(r["units"]),
                   "revenue": round(r["revenue"], 2), "margin": round(r["margin"], 2)}
                  for _, r in g.iterrows()]
        rev = float(g["revenue"].sum())
        cats.append({
            "category": cat, "series": series,
            "total_units": round(float(g["units"].sum())),
            "total_revenue": round(rev, 2),
            "total_margin": round(float(g["margin"].sum()), 2),
            "margin_pct": round(float(g["margin"].sum()) / rev * 100, 1) if rev > 0 else None,
            "share_pct": round(rev / total_rev_all * 100, 1),
            "yoy_pct": yoy.get(cat),
        })
    cats.sort(key=lambda c: c["total_revenue"], reverse=True)

    return {
        "granularity": granularity, "channel": channel, "country": country,
        "periods": periods, "categories": cats, "yoy_window": yoy_window,
        "totals": {
            "units": round(float(df["units"].sum())),
            "revenue": round(float(df["revenue"].sum()), 2),
            "margin": round(float(df["margin"].sum()), 2),
            "yoy_pct": tot_yoy,
        },
    }


_WS = "cm.channel = 'wholesale' AND et.partner_id IS NOT NULL AND et.quantity > 0"


def buyer_performance(db: Session, *, country: Optional[str] = None,
                      category: Optional[str] = None) -> dict:
    """Per-wholesale-buyer performance: revenue / margin / growth, category mix,
    order recency, and a churn-risk status. Decision support for account mgmt."""
    last = db.execute(text(
        f"SELECT MAX(et.transaction_date) FROM erp_transactions et "
        f"JOIN lookup_channel_map cm ON cm.id=et.channel_map_id WHERE {_WS}"
    )).scalar()
    if last is None:
        return {"buyers": [], "periods": [], "totals": {}, "status_counts": {}}

    flt = ("AND (:country IS NULL OR dp.country=:country) "
           "AND (:category IS NULL OR dc.name=:category)")
    params = {"country": country, "category": category, "last": last}

    rows = db.execute(text(f"""
        SELECT dp.id AS pid, COALESCE(dp.name,'(unknown)') AS buyer,
               COALESCE(dp.country,'?') AS country,
               to_char(et.transaction_date,'YYYY-MM') AS period,
               SUM(et.quantity) AS units, SUM({_REV}) AS revenue, SUM(et.ruc_eur) AS margin
        FROM erp_transactions et
        JOIN lookup_channel_map cm ON cm.id=et.channel_map_id
        JOIN dim_partners dp ON dp.id=et.partner_id
        JOIN dim_products p ON p.id=et.product_id
        LEFT JOIN dim_categories dc ON dc.id=p.category_id
        WHERE {_WS} {flt}
        GROUP BY 1,2,3,4
    """), params).mappings().all()

    meta = db.execute(text(f"""
        SELECT dp.id AS pid,
               MAX(et.transaction_date) AS last_order,
               COUNT(DISTINCT dc.name) AS n_categories,
               COUNT(DISTINCT et.product_id) AS n_skus,
               SUM({_REV}) FILTER (WHERE et.transaction_date > :last - INTERVAL '90 days') AS rev_recent,
               SUM({_REV}) FILTER (WHERE et.transaction_date <= :last - INTERVAL '90 days'
                                    AND et.transaction_date > :last - INTERVAL '180 days') AS rev_prior
        FROM erp_transactions et
        JOIN lookup_channel_map cm ON cm.id=et.channel_map_id
        JOIN dim_partners dp ON dp.id=et.partner_id
        JOIN dim_products p ON p.id=et.product_id
        LEFT JOIN dim_categories dc ON dc.id=p.category_id
        WHERE {_WS} {flt}
        GROUP BY dp.id
    """), params).mappings().all()
    meta_by = {m["pid"]: m for m in meta}

    df = pd.DataFrame(rows, columns=["pid", "buyer", "country", "period", "units", "revenue", "margin"])
    if df.empty:
        return {"buyers": [], "periods": [], "totals": {}, "status_counts": {}}
    for c in ("units", "revenue", "margin"):
        df[c] = df[c].astype(float)
    periods = sorted(df["period"].unique())

    # YoY per buyer: latest-year months vs same prior-year months
    df["y"] = df["period"].str[:4].astype(int)
    df["m"] = df["period"].str[5:7]
    yoy = {}
    years = sorted(df["y"].unique())
    if len(years) >= 2:
        cur, prev = years[-1], years[-2]
        common = sorted(set(df.loc[df.y == cur, "m"]) & set(df.loc[df.y == prev, "m"]))
        cr = df[(df.y == cur) & (df.m.isin(common))].groupby("pid")["revenue"].sum()
        pr = df[(df.y == prev) & (df.m.isin(common))].groupby("pid")["revenue"].sum()
        for pid in df["pid"].unique():
            p = float(pr.get(pid, 0.0)); c0 = float(cr.get(pid, 0.0))
            yoy[pid] = round((c0 / p - 1.0) * 100, 1) if p > 0 else None

    total_rev = float(df["revenue"].sum()) or 1.0

    def status(pid):
        m = meta_by.get(pid, {})
        lo = m.get("last_order")
        wks = (last - lo).days // 7 if lo else 999
        rec = float(m.get("rev_recent") or 0); pri = float(m.get("rev_prior") or 0)
        if wks >= 12:
            return "lost", wks
        if wks >= 6:
            return "at_risk", wks
        if pri == 0 and rec > 0:
            return "new", wks
        if pri > 0 and rec > pri * 1.25:
            return "growing", wks
        if pri > 0 and rec < pri * 0.6:
            return "declining", wks
        return "active", wks

    buyers = []
    for pid, g in df.groupby("pid"):
        g = g.sort_values("period")
        rev = float(g["revenue"].sum())
        st, wks = status(pid)
        m = meta_by.get(pid, {})
        buyers.append({
            "partner_id": int(pid), "buyer": g["buyer"].iloc[0], "country": g["country"].iloc[0],
            "total_revenue": round(rev, 2), "total_margin": round(float(g["margin"].sum()), 2),
            "total_units": round(float(g["units"].sum())),
            "margin_pct": round(float(g["margin"].sum()) / rev * 100, 1) if rev > 0 else None,
            "share_pct": round(rev / total_rev * 100, 1),
            "yoy_pct": yoy.get(pid),
            "n_categories": int(m.get("n_categories") or 0),
            "n_skus": int(m.get("n_skus") or 0),
            "weeks_since_order": int(wks), "status": st,
            "series": [{"period": r["period"], "revenue": round(r["revenue"], 2)}
                       for _, r in g.iterrows()],
        })
    buyers.sort(key=lambda b: b["total_revenue"], reverse=True)

    status_counts: dict = {}
    for b in buyers:
        status_counts[b["status"]] = status_counts.get(b["status"], 0) + 1

    return {
        "country": country, "category": category, "periods": periods,
        "buyers": buyers, "status_counts": status_counts,
        "totals": {"revenue": round(float(df["revenue"].sum()), 2),
                   "margin": round(float(df["margin"].sum()), 2),
                   "n_buyers": len(buyers)},
    }


_NONSTORE = ("'%warehouse%','%transfer%','%proizvodnja%','%testni%',"
             "'%FBA%','%oštećena%','%aggregate%'")


def _status(rec: float, pri: float, wks: int):
    if wks >= 12:
        return "lost"
    if wks >= 6:
        return "at_risk"
    if pri == 0 and rec > 0:
        return "new"
    if pri > 0 and rec > pri * 1.25:
        return "growing"
    if pri > 0 and rec < pri * 0.6:
        return "declining"
    return "active"


def store_performance(db: Session, *, country: Optional[str] = None,
                      category: Optional[str] = None) -> dict:
    """Per-store retail performance (real stores only): revenue/margin/growth,
    share, recency, status. Retail-only — web/wholesale are central."""
    last = db.execute(text(
        "SELECT MAX(et.transaction_date) FROM erp_transactions et "
        "JOIN lookup_channel_map cm ON cm.id=et.channel_map_id WHERE cm.channel='retail'"
    )).scalar()
    if last is None:
        return {"stores": [], "periods": [], "totals": {}, "status_counts": {}}
    base = (f"FROM erp_transactions et "
            f"JOIN lookup_channel_map cm ON cm.id=et.channel_map_id "
            f"JOIN dim_stores ds ON ds.id=et.store_id "
            f"JOIN dim_products p ON p.id=et.product_id "
            f"LEFT JOIN dim_categories dc ON dc.id=p.category_id "
            f"WHERE cm.channel='retail' AND et.quantity>0 AND ds.is_warehouse IS NOT TRUE "
            f"AND NOT (ds.name ILIKE ANY(ARRAY[{_NONSTORE}])) "
            f"AND (:country IS NULL OR ds.country=:country) "
            f"AND (:category IS NULL OR dc.name=:category)")
    params = {"country": country, "category": category, "last": last}
    rows = db.execute(text(f"""
        SELECT ds.id AS sid, ds.name AS store, COALESCE(ds.country,'?') AS country,
               to_char(et.transaction_date,'YYYY-MM') AS period,
               SUM(et.quantity) AS units, SUM({_REV}) AS revenue, SUM(et.ruc_eur) AS margin
        {base} GROUP BY 1,2,3,4
    """), params).mappings().all()
    meta = db.execute(text(f"""
        SELECT ds.id AS sid, MAX(et.transaction_date) AS last_order,
               SUM({_REV}) FILTER (WHERE et.transaction_date > :last - INTERVAL '90 days') AS rev_recent,
               SUM({_REV}) FILTER (WHERE et.transaction_date <= :last - INTERVAL '90 days'
                                    AND et.transaction_date > :last - INTERVAL '180 days') AS rev_prior
        {base} GROUP BY ds.id
    """), params).mappings().all()
    meta_by = {m["sid"]: m for m in meta}

    df = pd.DataFrame(rows, columns=["sid", "store", "country", "period", "units", "revenue", "margin"])
    if df.empty:
        return {"stores": [], "periods": [], "totals": {}, "status_counts": {}}
    for c in ("units", "revenue", "margin"):
        df[c] = df[c].astype(float)
    periods = sorted(df["period"].unique())
    df["y"] = df["period"].str[:4].astype(int); df["m"] = df["period"].str[5:7]
    yoy = {}
    yrs = sorted(df["y"].unique())
    if len(yrs) >= 2:
        cur, prev = yrs[-1], yrs[-2]
        common = sorted(set(df.loc[df.y == cur, "m"]) & set(df.loc[df.y == prev, "m"]))
        cr = df[(df.y == cur) & df.m.isin(common)].groupby("sid")["revenue"].sum()
        pr = df[(df.y == prev) & df.m.isin(common)].groupby("sid")["revenue"].sum()
        for sid in df["sid"].unique():
            p = float(pr.get(sid, 0)); c0 = float(cr.get(sid, 0))
            yoy[sid] = round((c0 / p - 1) * 100, 1) if p > 0 else None
    total = float(df["revenue"].sum()) or 1.0
    stores = []
    for sid, g in df.groupby("sid"):
        g = g.sort_values("period"); rev = float(g["revenue"].sum())
        m = meta_by.get(sid, {}); lo = m.get("last_order")
        wks = (last - lo).days // 7 if lo else 999
        st = _status(float(m.get("rev_recent") or 0), float(m.get("rev_prior") or 0), wks)
        stores.append({"store_id": int(sid), "store": g["store"].iloc[0], "country": g["country"].iloc[0],
                       "total_revenue": round(rev, 2), "total_margin": round(float(g["margin"].sum()), 2),
                       "total_units": round(float(g["units"].sum())),
                       "margin_pct": round(float(g["margin"].sum()) / rev * 100, 1) if rev > 0 else None,
                       "share_pct": round(rev / total * 100, 1), "yoy_pct": yoy.get(sid),
                       "weeks_since_order": int(wks), "status": st,
                       "series": [{"period": r["period"], "revenue": round(r["revenue"], 2)} for _, r in g.iterrows()]})
    stores.sort(key=lambda s: s["total_revenue"], reverse=True)
    sc: dict = {}
    for s in stores:
        sc[s["status"]] = sc.get(s["status"], 0) + 1
    return {"country": country, "category": category, "periods": periods, "stores": stores,
            "status_counts": sc, "totals": {"revenue": round(float(df["revenue"].sum()), 2),
            "margin": round(float(df["margin"].sum()), 2), "n_stores": len(stores)}}


def profitability(db: Session, *, channel: Optional[str] = None,
                  country: Optional[str] = None) -> dict:
    """Margin €/% by category with a value-quadrant (cash-cow / traffic-driver /
    niche / drag), plus a channel margin breakdown."""
    flt = ("AND (:channel IS NULL OR cm.channel=:channel) "
           f"AND (:country IS NULL OR {_COUNTRY}=:country) "
           "AND (dc.name IS NULL OR NOT (dc.name = ANY(:excl)))")
    params = {"channel": channel, "country": country, "excl": list(EXCLUDE_CATEGORIES)}
    join = ("FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id=et.channel_map_id "
            "JOIN dim_products p ON p.id=et.product_id "
            "LEFT JOIN dim_categories dc ON dc.id=p.category_id "
            "LEFT JOIN dim_stores ds ON ds.id=et.store_id "
            "LEFT JOIN dim_partners dp ON dp.id=et.partner_id "
            "WHERE et.quantity>0 " + flt)
    cat_rows = db.execute(text(f"""
        SELECT COALESCE(dc.name,'(uncategorised)') AS category,
               SUM({_REV}) AS revenue, SUM(et.ruc_eur) AS margin {join} GROUP BY 1
    """), params).mappings().all()
    ch_rows = db.execute(text(f"""
        SELECT cm.channel AS channel, SUM({_REV}) AS revenue, SUM(et.ruc_eur) AS margin {join} GROUP BY 1
    """), params).mappings().all()

    cats = [{"category": r["category"], "revenue": round(float(r["revenue"]), 2),
             "margin": round(float(r["margin"]), 2),
             "margin_pct": round(float(r["margin"]) / float(r["revenue"]) * 100, 1)
             if r["revenue"] else None} for r in cat_rows if float(r["revenue"] or 0) > 0]
    if cats:
        revs = sorted(c["revenue"] for c in cats)
        med = revs[len(revs) // 2]
        for c in cats:
            hi_rev = c["revenue"] >= med
            hi_mgn = (c["margin_pct"] or 0) >= 30
            c["quadrant"] = ("cash_cow" if hi_rev and hi_mgn else
                             "traffic_driver" if hi_rev and not hi_mgn else
                             "niche" if not hi_rev and hi_mgn else "drag")
    cats.sort(key=lambda c: c["revenue"], reverse=True)
    chans = [{"channel": r["channel"], "revenue": round(float(r["revenue"]), 2),
              "margin": round(float(r["margin"]), 2),
              "margin_pct": round(float(r["margin"]) / float(r["revenue"]) * 100, 1)
              if r["revenue"] else None} for r in ch_rows if float(r["revenue"] or 0) > 0]
    chans.sort(key=lambda c: c["revenue"], reverse=True)
    tot_rev = sum(c["revenue"] for c in cats) or 1.0
    tot_mgn = sum(c["margin"] for c in cats)
    return {"channel": channel, "country": country, "categories": cats, "channels": chans,
            "rev_median": round(med, 2) if cats else 0, "margin_threshold": 30,
            "totals": {"revenue": round(tot_rev, 2), "margin": round(tot_mgn, 2),
                       "margin_pct": round(tot_mgn / tot_rev * 100, 1)}}


# ── Wholesale FA per buyer (S&OE) ─────────────────────────────────────────
# KAM input buyer names are short/free-text; map to the real partner (country-
# aware — Spar HR != Spar SI). Pattern + optional country; first match wins.
BUYER_ALIASES = {
    "mercator":       ("Mercator%", "SI"),
    "konzum":         ("Konzum%", "HR"),
    "spar hrvatska":  ("SPAR HRVATSKA%", None),
    "spar ljubljana": ("SPAR LJUBLJANA%", "SI"),
    "bipa":           ("Bipa%", "HR"),
    "kaufland":       ("Kaufland%", "HR"),
    "sbi":            ("SBI%", "ME"),
    "dm":             ("dm-drogerie markt%", "HR"),
    "plodine":        ("Plodine%", None),
    "tisak":          ("Tisak%", None),
    "mci":            ("MCI%", None),
    "ostalo":         (None, None),     # "Other" — residual bucket, no single partner
}
_MAD = 1.4826
_ONTOP_WINDOW = 2          # ±weeks to match a forecast on-top to an actual one-off
_HIST_WEEKS = 26          # trailing window for the recurring baseline


def _resolve_buyers(db: Session) -> dict:
    """{lower(input buyer): partner_id} via the curated alias map."""
    out = {}
    for alias, (pat, ctry) in BUYER_ALIASES.items():
        if pat is None:                 # residual bucket (e.g. "ostalo") — no partner
            out[alias] = None
            continue
        row = db.execute(text(
            "SELECT id FROM dim_partners WHERE name ILIKE :p "
            "AND (:c IS NULL OR country=:c) ORDER BY id LIMIT 1"
        ), {"p": pat, "c": ctry}).first()
        out[alias] = int(row[0]) if row else None
    return out


def _yw_index(weeks):
    return {w: i for i, w in enumerate(sorted(weeks))}


def _oneoff_bar(qty):
    """(median, bar) defining one-offs for a weekly array, or None if too sparse
    (<4 non-zero weeks). bar = max(median + 3*MAD, 300u)."""
    import numpy as np
    qty = np.asarray(qty, float)
    nz = qty[qty > 0]
    if len(nz) < 4:
        return None
    med = float(np.median(nz)); mad = float(np.median(np.abs(nz - med))) * _MAD
    return med, max(med + 3 * mad, 300.0)


def _strip_oneoffs(qty, ref=None):
    """Split a weekly array into (recurring, oneoff): a week is a one-off if its
    qty exceeds median + 3*MAD of the non-zero orders AND >= 300u (mirrors the
    forecast_v4 wholesale cleaning).

    If `ref` (median, bar) is supplied — e.g. derived from a longer history via
    _oneoff_bar — it is used instead of computing the threshold from `qty` itself.
    This lets a short window (e.g. 4 scored weeks) be decomposed with the SAME
    threshold as the baseline, so a campaign week isn't mislabelled 'recurring'
    just because the window is too short to find its own median."""
    import numpy as np
    qty = np.asarray(qty, float)
    rec, one = qty.copy(), np.zeros_like(qty)
    stats = ref if ref is not None else _oneoff_bar(qty)
    if stats is not None:
        med, bar = stats
        mask = qty > bar
        one[mask] = qty[mask] - med
        rec[mask] = med
    return rec, one


def _rate(qty):
    """Recency-weighted mean weekly rate of the recurring (one-off-stripped) history."""
    import numpy as np
    qty = np.asarray(qty, float)
    if len(qty) == 0:
        return 0.0
    rec, _ = _strip_oneoffs(qty)
    w = np.array([0.95 ** (len(rec) - 1 - i) for i in range(len(rec))])
    return float(np.average(rec, weights=w))


def forecast_outlook(db: Session) -> dict:
    """Next-13-week v4 forecast by category vs the same weeks last year (actual),
    with projected growth. Uses the latest forecasts run."""
    run = db.execute(text("SELECT MAX(run_id) FROM forecasts")).scalar()
    if run is None:
        return {"run_id": None, "categories": [], "totals": {}}
    fc = db.execute(text("""
        SELECT COALESCE(dc.name,'(uncategorised)') AS category,
               (f.year*100+f.week) AS yw, SUM(f.total) AS fc
        FROM forecasts f JOIN dim_products p ON p.id=f.product_id
        LEFT JOIN dim_categories dc ON dc.id=p.category_id
        WHERE f.run_id=:r AND NOT (dc.name = ANY(:excl)) GROUP BY 1,2
    """), {"r": run, "excl": list(EXCLUDE_CATEGORIES)}).mappings().all()
    if not fc:
        return {"run_id": run, "categories": [], "totals": {}}
    fdf = pd.DataFrame(fc, columns=["category", "yw", "fc"]); fdf["fc"] = fdf["fc"].astype(float)
    weeks = [int(w) for w in sorted(fdf["yw"].unique())]
    ly_weeks = [(w // 100 - 1) * 100 + (w % 100) for w in weeks]   # same ISO weeks, prior year
    ly = db.execute(text(f"""
        SELECT COALESCE(dc.name,'(uncategorised)') AS category,
               SUM({_REV}) AS actual
        FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id=et.channel_map_id
        JOIN dim_products p ON p.id=et.product_id
        LEFT JOIN dim_categories dc ON dc.id=p.category_id
        WHERE et.quantity>0
          AND (EXTRACT(isoyear FROM transaction_date)::int*100+EXTRACT(week FROM transaction_date)::int) = ANY(:lyw)
        GROUP BY 1
    """), {"lyw": ly_weeks}).mappings().all()
    # NB forecast is UNITS (forecasts.total), LY actual here is revenue — so we
    # report forecast units vs LY units instead for apples-to-apples:
    ly_units = db.execute(text("""
        SELECT COALESCE(dc.name,'(uncategorised)') AS category, SUM(et.quantity) AS units
        FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id=et.channel_map_id
        JOIN dim_products p ON p.id=et.product_id
        LEFT JOIN dim_categories dc ON dc.id=p.category_id
        WHERE et.quantity>0
          AND (EXTRACT(isoyear FROM transaction_date)::int*100+EXTRACT(week FROM transaction_date)::int) = ANY(:lyw)
        GROUP BY 1
    """), {"lyw": ly_weeks}).mappings().all()
    ly_by = {r["category"]: float(r["units"]) for r in ly_units}

    cats = []
    for cat, g in fdf.groupby("category"):
        fc_units = float(g["fc"].sum())
        ly_u = ly_by.get(cat, 0.0)
        cats.append({"category": cat, "forecast_units": round(fc_units),
                     "ly_units": round(ly_u),
                     "growth_pct": round((fc_units / ly_u - 1) * 100, 1) if ly_u > 0 else None,
                     "series": [{"yw": int(r["yw"]), "forecast": round(float(r["fc"]))}
                                for _, r in g.sort_values("yw").iterrows()]})
    cats.sort(key=lambda c: c["forecast_units"], reverse=True)
    tot_fc = sum(c["forecast_units"] for c in cats)
    tot_ly = sum(ly_by.values())
    return {"run_id": int(run), "horizon": [int(w) for w in weeks],
            "ly_horizon": [int(w) for w in ly_weeks], "categories": cats,
            "totals": {"forecast_units": round(tot_fc), "ly_units": round(tot_ly),
                       "growth_pct": round((tot_fc / tot_ly - 1) * 100, 1) if tot_ly > 0 else None}}


def _fa(fc: float, act: float) -> float:
    denom = abs(act) if act else (abs(fc) or 1.0)
    return round(max(0.0, (1 - abs(fc - act) / denom)) * 100, 1)


def wholesale_fa_review(db: Session, from_yw: Optional[int] = None,
                        to_yw: Optional[int] = None, n_weeks: int = 8) -> dict:
    """Wholesale forecast review over a week window (default last n_weeks):
    cumulative 3 lines (baseline / baseline+ontop+regular / actual) per week, and
    top-10 offender SKUs by |full-forecast - actual| with per-buyer detail.

    Baseline = each series' one-off-stripped recency-weighted run-rate as of the
    window start, held flat. KAM regular/on-top from on_top_inputs where present
    (older weeks have none -> full = baseline)."""
    import numpy as np

    all_w = [int(w) for w in db.execute(text(
        "SELECT DISTINCT EXTRACT(isoyear FROM transaction_date)::int*100"
        "+EXTRACT(week FROM transaction_date)::int AS yw "
        "FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id=et.channel_map_id "
        "WHERE cm.channel='wholesale' ORDER BY 1"
    )).scalars().all()]
    if not all_w:
        return {"weeks": [], "lines": [], "offenders": [], "totals": {}}
    to_yw = to_yw or all_w[-1]
    avail = [w for w in all_w if w <= to_yw]
    from_yw = from_yw or (avail[-n_weeks] if len(avail) >= n_weeks else avail[0])
    weeks = [w for w in avail if w >= from_yw]
    hist_w = [w for w in all_w if w < from_yw][-_HIST_WEEKS:]
    nW = len(weeks)
    if nW == 0:
        return {"weeks": [], "lines": [], "offenders": [], "totals": {}}

    # actuals per (sku, partner, yw) over history + window
    span = hist_w + weeks
    arows = db.execute(text("""
        SELECT et.product_id AS sku, et.partner_id AS pid,
               EXTRACT(isoyear FROM transaction_date)::int*100
                 + EXTRACT(week FROM transaction_date)::int AS yw,
               SUM(et.quantity) AS q
        FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id=et.channel_map_id
        WHERE cm.channel='wholesale'
          AND (EXTRACT(isoyear FROM transaction_date)::int*100
                 + EXTRACT(week FROM transaction_date)::int) = ANY(:span)
        GROUP BY 1,2,3
    """), {"span": [int(w) for w in span]}).mappings().all()
    df = pd.DataFrame(arows, columns=["sku", "pid", "yw", "q"])
    if df.empty:
        return {"weeks": weeks, "lines": [], "offenders": [], "totals": {}}
    df["q"] = df["q"].astype(float)
    df["sku"] = df["sku"].astype(int)
    df["yw"] = df["yw"].astype(int)

    # KAM inputs for the window (regular + ontop), mapped to partner
    pid_of = _resolve_buyers(db)
    irows = db.execute(text("""
        SELECT lower(oi.buyer) AS alias, oi.buyer AS buyer_name,
               COALESCE(u.display_name,u.username) AS kam, oi.product_id AS sku,
               oi.year_week AS yw,
               COALESCE(oi.regular_increase_qty,0) AS regular,
               GREATEST(COALESCE(oi.quantity,0)-COALESCE(oi.regular_increase_qty,0),0) AS ontop
        FROM on_top_inputs oi LEFT JOIN users u ON u.id=oi.submitted_by_id
        WHERE oi.channel='wholesale' AND oi.buyer IS NOT NULL AND oi.year_week = ANY(:weeks)
    """), {"weeks": [int(w) for w in weeks]}).mappings().all()
    inp = pd.DataFrame(irows, columns=["alias", "buyer_name", "kam", "sku", "yw", "regular", "ontop"])
    if not inp.empty:
        inp["sku"] = inp["sku"].astype(int)
        for c in ("regular", "ontop"):
            inp[c] = inp[c].astype(float)
        inp["pid"] = inp["alias"].map(pid_of)

    sku_name = dict(db.execute(text("SELECT id, name FROM dim_products")).all())
    part_name = dict(db.execute(text("SELECT id, name FROM dim_partners")).all())

    def rate_of(sub_hist):
        """Run-rate baseline for a week-summed hist dict (keyed by yw). Returns 0
        when the history is too sparse for an established rhythm — i.e. <4 order
        weeks, so `_oneoff_bar` can't even form a one-off threshold. A sporadic /
        distributor pattern (e.g. SBI orders only via large on-tops) has no
        recurring baseline; only the KAM on-top counts."""
        v = np.array([float(sub_hist.get(w, 0.0)) for w in hist_w])
        if not len(v) or _oneoff_bar(v) is None:
            return 0.0
        return _rate(v)

    # ---- per-SKU baseline = Σ per-buyer recurring (each sparse-guarded) ----
    # Summing guarded PER-BUYER run-rates (not the single aggregate run-rate) keeps
    # sporadic buyers out of the SKU baseline entirely — otherwise their volume
    # would survive in the SKU total and leak into the residual "Other" bucket
    # even after their own row is zeroed.
    hist_df = df[df.yw.isin(hist_w)]
    sku_base = {}
    for sku in df["sku"].unique():
        sub = hist_df[hist_df.sku == sku]
        tot = 0.0
        for pid in sub["pid"].dropna().unique():
            ph = sub[sub.pid == pid].groupby("yw")["q"].sum().to_dict()
            tot += rate_of(ph)
        sku_base[int(sku)] = tot

    act_sw = df[df.yw.isin(weeks)].groupby(["sku", "yw"])["q"].sum()
    # inputs per (sku, yw)
    inp_sw = (inp.groupby(["sku", "yw"])[["regular", "ontop"]].sum()
              if not inp.empty else pd.DataFrame())

    def inp_at(sku, w):
        try:
            r = inp_sw.loc[(sku, w)]; return float(r["regular"]) + float(r["ontop"])
        except Exception:
            return 0.0

    base_total = sum(sku_base.values())
    lines = []
    for w in weeks:
        actual_w = float(act_sw[act_sw.index.get_level_values(1) == w].sum()) if len(act_sw) else 0.0
        adds = float(inp_sw.xs(w, level=1)[["regular", "ontop"]].sum().sum()) if (not inp_sw.empty and w in inp_sw.index.get_level_values(1)) else 0.0
        lines.append({"yw": int(w), "baseline": round(base_total),
                      "full": round(base_total + adds), "actual": round(actual_w)})

    # ---- offenders: per SKU full vs actual over window ----
    rows = []
    skus = set(int(s) for s in df["sku"].unique()) | (set(int(s) for s in inp["sku"].unique()) if not inp.empty else set())
    for sku in skus:
        base = sku_base.get(sku, 0.0) * nW
        adds = float(inp[inp.sku == sku][["regular", "ontop"]].sum().sum()) if not inp.empty else 0.0
        act = float(act_sw[sku].sum()) if (sku in act_sw.index.get_level_values(0)) else 0.0
        full = base + adds
        rows.append((sku, base, full, act, abs(full - act)))
    rows.sort(key=lambda r: r[4], reverse=True)
    top = rows[:10]

    # ---- per-buyer detail for the top offenders ----
    named_pids = {p for p in pid_of.values() if p}      # KAM-named buyers only
    offenders = []
    for sku, base, full, act, err in top:
        sb = df[(df.sku == sku)]
        buyers = []
        bset = (set(int(p) for p in sb["pid"].dropna().unique())
                | (set(int(p) for p in inp[(inp.sku == sku)]["pid"].dropna().unique()) if not inp.empty else set())) & named_pids
        named_base = named_act = 0.0
        for pid in bset:
            bh = sb[(sb.pid == pid) & (sb.yw.isin(hist_w))].groupby("yw")["q"].sum().to_dict()
            b_base = rate_of(bh) * nW
            b_adds = float(inp[(inp.sku == sku) & (inp.pid == pid)][["regular", "ontop"]].sum().sum()) if not inp.empty else 0.0
            b_act = float(sb[(sb.pid == pid) & (sb.yw.isin(weeks))]["q"].sum())
            kam = (inp[(inp.sku == sku) & (inp.pid == pid)]["kam"].iloc[0]
                   if (not inp.empty and len(inp[(inp.sku == sku) & (inp.pid == pid)])) else None)
            named_base += b_base; named_act += b_act
            buyers.append({"partner_id": pid, "buyer": part_name.get(pid, str(pid)), "kam": kam,
                           "baseline": round(b_base), "full": round(b_base + b_adds),
                           "actual": round(b_act), "gap": round(b_base + b_adds - b_act)})
        # "Other (Ostalo)" residual = KAM's other-buyers on-top vs the actual of
        # everyone not named above.
        ost_adds = float(inp[(inp.sku == sku) & (inp.alias == "ostalo")][["regular", "ontop"]].sum().sum()) if not inp.empty else 0.0
        resid_base = max(base - named_base, 0.0)
        resid_act = max(act - named_act, 0.0)
        if ost_adds > 0 or resid_act > 1:
            buyers.append({"partner_id": -1, "buyer": "Other (Ostalo)", "kam": "Selma",
                           "baseline": round(resid_base), "full": round(resid_base + ost_adds),
                           "actual": round(resid_act), "gap": round(resid_base + ost_adds - resid_act)})
        buyers.sort(key=lambda b: abs(b["gap"]), reverse=True)
        offenders.append({"product_id": sku, "sku": sku_name.get(sku, str(sku)),
                          "baseline": round(base), "full": round(full), "actual": round(act),
                          "gap": round(full - act), "fa_pct": _fa(full, act), "buyers": buyers})

    tb = sum(l["baseline"] for l in lines); tf = sum(l["full"] for l in lines); ta = sum(l["actual"] for l in lines)
    return {"from_yw": from_yw, "to_yw": to_yw, "weeks": weeks, "lines": lines,
            "available_weeks": all_w, "offenders": offenders,
            "totals": {"baseline": round(tb), "full": round(tf), "actual": round(ta),
                       "fa_pct": _fa(tf, ta), "bias_pct": round((tf - ta) / (ta or 1) * 100, 1)}}


def wholesale_fa_excel(db: Session, from_yw: Optional[int] = None,
                       to_yw: Optional[int] = None) -> bytes:
    """Styled S&OE wholesale review workbook: Summary (window, totals, weekly
    3-line, top-10 offenders) + Detail (per offender, per-buyer base/full/actual/gap)."""
    from io import BytesIO
    from openpyxl import Workbook
    from openpyxl.styles import Font, PatternFill, Alignment, Border, Side

    d = wholesale_fa_review(db, from_yw=from_yw, to_yw=to_yw)
    NAVY, BLUE, LGREY, GREEN, RED = "1B2A4A", "2F5496", "EEF2F9", "C6EFCE", "FFC7CE"
    hdr = Font(name="Aptos", bold=True, color="FFFFFF", size=10)
    title = Font(name="Aptos", bold=True, color=NAVY, size=15)
    bold = Font(name="Aptos", bold=True, size=10)
    fill_hdr = PatternFill("solid", fgColor=BLUE)
    fill_navy = PatternFill("solid", fgColor=NAVY)
    fill_band = PatternFill("solid", fgColor=LGREY)
    thin = Border(*([Side(style="thin", color="D9D9D9")] * 4))
    R = Alignment(horizontal="right"); C = Alignment(horizontal="center")

    def cwl(yw): return f"W{yw % 100}/{yw // 100}"

    wb = Workbook(); ws = wb.active; ws.title = "Summary"
    ws["A1"] = "Wholesale Forecast Review (S&OE)"; ws["A1"].font = title
    wkr = f"{cwl(d['from_yw'])} – {cwl(d['to_yw'])}" if d.get("from_yw") else "—"
    ws["A2"] = f"Window: {wkr}"; ws["A2"].font = Font(name="Aptos", size=9, color="808080")
    t = d.get("totals", {})
    ws["A4"] = "Totals (units)"; ws["A4"].font = bold
    for i, (lbl, key) in enumerate([("Baseline", "baseline"), ("Full forecast", "full"),
                                    ("Actual", "actual"), ("FA %", "fa_pct"), ("Bias %", "bias_pct")]):
        c = ws.cell(5, 1 + i, lbl); c.font = hdr; c.fill = fill_hdr; c.alignment = C
        v = ws.cell(6, 1 + i, t.get(key, 0))
        v.number_format = '#,##0' if key in ("baseline", "full", "actual") else '0.0"%"'
        v.alignment = C; v.font = bold

    # weekly 3-line
    r0 = 8
    ws.cell(r0, 1, "Week").font = hdr
    for j, lbl in enumerate(["Week", "Baseline", "Full (base+ontop+reg)", "Actual", "Gap (full-act)"]):
        c = ws.cell(r0, 1 + j, lbl); c.font = hdr; c.fill = fill_hdr; c.alignment = C
    for i, l in enumerate(d.get("lines", [])):
        rr = r0 + 1 + i
        ws.cell(rr, 1, cwl(l["yw"])).alignment = C
        for j, key in enumerate(["baseline", "full", "actual"]):
            cc = ws.cell(rr, 2 + j, l[key]); cc.number_format = '#,##0'; cc.alignment = R
        gap = ws.cell(rr, 5, l["full"] - l["actual"]); gap.number_format = '#,##0'; gap.alignment = R
        gap.fill = PatternFill("solid", fgColor=RED if abs(l["full"] - l["actual"]) > 0.3 * (l["actual"] or 1) else GREEN)

    # top-10 offenders
    r1 = r0 + len(d.get("lines", [])) + 3
    ws.cell(r1, 1, "Top-10 offender SKUs (by |full − actual|)").font = bold
    for j, lbl in enumerate(["SKU", "Baseline", "Full", "Actual", "Gap", "FA %"]):
        c = ws.cell(r1 + 1, 1 + j, lbl); c.font = hdr; c.fill = fill_hdr; c.alignment = C
    for i, o in enumerate(d.get("offenders", [])):
        rr = r1 + 2 + i
        ws.cell(rr, 1, o["sku"])
        for j, key in enumerate(["baseline", "full", "actual", "gap"]):
            cc = ws.cell(rr, 2 + j, o[key]); cc.number_format = '#,##0'; cc.alignment = R
        fa = ws.cell(rr, 6, o["fa_pct"]); fa.number_format = '0.0"%"'; fa.alignment = R
        fa.fill = PatternFill("solid", fgColor=GREEN if o["fa_pct"] >= 70 else RED if o["fa_pct"] < 50 else "FFEB9C")
    for col, w in (("A", 42), ("B", 12), ("C", 18), ("D", 12), ("E", 12), ("F", 10)):
        ws.column_dimensions[col].width = w

    # Detail sheet — per offender, per buyer
    ws2 = wb.create_sheet("Offender detail")
    ws2["A1"] = "Per-buyer detail — top-10 offenders"; ws2["A1"].font = title
    row = 3
    for o in d.get("offenders", []):
        ws2.cell(row, 1, o["sku"]).font = Font(name="Aptos", bold=True, color="FFFFFF", size=11)
        for j in range(1, 7): ws2.cell(row, j).fill = fill_navy
        ws2.cell(row, 6, f"FA {o['fa_pct']}%").font = Font(name="Aptos", bold=True, color="FFFFFF"); ws2.cell(row, 6).alignment = R
        row += 1
        for j, lbl in enumerate(["Buyer", "KAM", "Baseline", "Full", "Actual", "Gap"]):
            c = ws2.cell(row, 1 + j, lbl); c.font = hdr; c.fill = fill_hdr; c.alignment = C
        row += 1
        for k, b in enumerate(o.get("buyers", [])):
            ws2.cell(row, 1, b["buyer"]); ws2.cell(row, 2, b.get("kam") or "")
            for j, key in enumerate(["baseline", "full", "actual", "gap"]):
                cc = ws2.cell(row, 3 + j, b[key]); cc.number_format = '#,##0'; cc.alignment = R
            ws2.cell(row, 6).fill = PatternFill("solid", fgColor=RED if abs(b["gap"]) > 0.3 * (b["actual"] or 1) else GREEN)
            row += 1
        row += 1
    for col, w in (("A", 34), ("B", 12), ("C", 12), ("D", 12), ("E", 12), ("F", 12)):
        ws2.column_dimensions[col].width = w

    bio = BytesIO(); wb.save(bio); return bio.getvalue()


def wholesale_fa(db: Session, from_yw: Optional[int] = None,
                 to_yw: Optional[int] = None) -> dict:
    """S&OE wholesale forecast accuracy per buyer (KAM on-top + regular-increase),
    scored on CLOSED weeks that have KAM inputs (optionally restricted to a
    [from_yw, to_yw] window so it matches the page's timeframe picker). Forecast =
    recurring baseline + KAM regular + KAM on-top; actual decomposed into
    recurring + one-off; on-tops matched ±2 weeks. Rolls up per KAM, a global
    cumulative number, top-10 offender SKUs, and an error driver."""
    import numpy as np

    latest = db.execute(text(
        "SELECT MAX(EXTRACT(isoyear FROM transaction_date)::int*100"
        "+EXTRACT(week FROM transaction_date)::int) FROM erp_transactions et "
        "JOIN lookup_channel_map cm ON cm.id=et.channel_map_id WHERE cm.channel='wholesale'"
    )).scalar()
    if latest is None:
        return {"scored_weeks": [], "buyers": [], "kams": [], "offenders": [], "global": {}}
    latest = int(latest)
    if to_yw:
        latest = min(latest, int(to_yw))     # never score beyond actuals

    inputs = db.execute(text("""
        SELECT lower(oi.buyer) AS alias, oi.buyer AS buyer_name,
               COALESCE(u.display_name, u.username) AS kam,
               oi.product_id, oi.year_week,
               COALESCE(oi.regular_increase_qty,0) AS regular,
               GREATEST(COALESCE(oi.quantity,0) - COALESCE(oi.regular_increase_qty,0), 0) AS ontop
        FROM on_top_inputs oi LEFT JOIN users u ON u.id=oi.submitted_by_id
        WHERE oi.channel='wholesale' AND oi.buyer IS NOT NULL
          AND oi.year_week <= :latest
          AND (:from_yw IS NULL OR oi.year_week >= :from_yw)
    """), {"latest": latest, "from_yw": int(from_yw) if from_yw else None}).mappings().all()
    if not inputs:
        return {"scored_weeks": [], "buyers": [], "kams": [], "offenders": [], "global": {},
                "note": "No closed weeks with KAM wholesale inputs yet (inputs are for future weeks)."}

    scored = sorted({int(r["year_week"]) for r in inputs})
    pid_of = _resolve_buyers(db)
    sku_name = dict(db.execute(text("SELECT id, name FROM dim_products")).all())

    # actuals for the involved (partner, SKU) over history + scored + on-top window
    pairs = {(pid_of.get(r["alias"]), r["product_id"]) for r in inputs}
    partner_ids = sorted({p for (p, _) in pairs if p})
    product_ids = sorted({s for (_, s) in pairs})
    acts = {}
    if partner_ids and product_ids:
        rows = db.execute(text(f"""
            SELECT et.partner_id AS pid, et.product_id AS sku,
                   EXTRACT(isoyear FROM transaction_date)::int*100
                     + EXTRACT(week FROM transaction_date)::int AS yw,
                   SUM(et.quantity) AS q
            FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id=et.channel_map_id
            WHERE cm.channel='wholesale' AND et.partner_id = ANY(:pids)
              AND et.product_id = ANY(:skus)
            GROUP BY 1,2,3
        """), {"pids": partner_ids, "skus": product_ids}).mappings().all()
        for r in rows:
            acts[(int(r["pid"]), int(r["sku"]), int(r["yw"]))] = float(r["q"])

    def act_series(pid, sku, weeks):
        return np.array([acts.get((pid, sku, w), 0.0) for w in weeks])

    hist_weeks = [w for w in sorted({(y * 100 + wk) for y in (scored[0] // 100 - 1, scored[0] // 100)
                                     for wk in range(1, 53)}) if w < scored[0]][-_HIST_WEEKS:]
    win_weeks = sorted(set(scored) | {w for w in range(scored[0] - _ONTOP_WINDOW, scored[-1] + _ONTOP_WINDOW + 1)})

    # per (buyer, SKU) FA
    by_pair = {}
    for r in inputs:
        k = (r["alias"], r["buyer_name"], r["kam"], int(r["product_id"]))
        d = by_pair.setdefault(k, {})
        d[int(r["year_week"])] = {"regular": float(r["regular"]), "ontop": float(r["ontop"])}

    buyers_agg, kam_agg, sku_agg = {}, {}, {}
    g = {"fc": 0.0, "act": 0.0, "rec_fc": 0.0, "rec_act": 0.0,
         "ontop_fc": 0.0, "ontop_act": 0.0, "ontop_matched": 0.0, "ontop_missed": 0.0, "ontop_surprise": 0.0}

    for (alias, bname, kam, sku), wk in by_pair.items():
        pid = pid_of.get(alias)
        hist_arr = act_series(pid, sku, hist_weeks) if pid else np.zeros(len(hist_weeks))
        ref = _oneoff_bar(hist_arr)            # history-derived one-off threshold (per SKU)
        # A recurring baseline only makes sense when the buyer has an established
        # ordering rhythm. When the history is too sparse for _oneoff_bar to even
        # form a threshold (<4 order weeks), the buyer-SKU has no recurring rate —
        # it's a sporadic / distributor pattern (e.g. SBI is a distributor that
        # always orders via large on-tops, never a steady weekly baseline). Force
        # baseline 0 so a couple of initial bulk orders aren't spread into a
        # phantom weekly baseline; only the KAM on-top/regular counts.
        baseline = _rate(hist_arr) if (pid and ref is not None) else 0.0
        rec_act_arr, one_act_arr = _strip_oneoffs(act_series(pid, sku, scored), ref=ref) if pid else (np.zeros(len(scored)), np.zeros(len(scored)))
        rec_act = float(rec_act_arr.sum())
        rec_fc = sum(baseline + wk.get(w, {}).get("regular", 0.0) for w in scored)
        baseline_only = baseline * len(scored)

        # on-top event matching (±window) on the full window
        win = sorted(win_weeks)
        one_win = {w: v for w, v in zip(win, _strip_oneoffs(act_series(pid, sku, win), ref=ref)[1] if pid else np.zeros(len(win)))}
        fc_ontops = [(w, wk[w]["ontop"]) for w in scored if wk.get(w, {}).get("ontop", 0) > 0]
        act_ontops = {w: q for w, q in one_win.items() if q > 0}
        matched = missed = 0.0
        used = set()
        for w, q in fc_ontops:
            cand = sorted([(abs(w - aw), aw) for aw in act_ontops if aw not in used and abs(w - aw) <= _ONTOP_WINDOW])
            if cand:
                aw = cand[0][1]; used.add(aw); matched += min(q, act_ontops[aw])
            else:
                missed += q
        surprise = sum(q for aw, q in act_ontops.items() if aw not in used)
        ontop_fc = sum(q for _, q in fc_ontops)
        ontop_act = sum(act_ontops.values())      # ±window one-offs — feeds matched/missed/surprise only
        oneoff_scored = float(one_act_arr.sum())  # one-offs *within* scored weeks (reconciles with weekly series)

        fc = rec_fc + ontop_fc
        act = rec_act + oneoff_scored
        # error driver
        err_ontop = missed + surprise + abs(ontop_fc - matched - missed)
        err_rec = abs(rec_fc - rec_act)
        if err_ontop >= err_rec and err_ontop > 0:
            driver = "ontop"
        elif abs(rec_fc - rec_act) > abs(baseline_only - rec_act) + 1:
            driver = "regular"
        elif err_rec > 0:
            driver = "baseline"
        else:
            driver = "ok"

        for agg, key in ((buyers_agg, (alias, bname, kam)), (kam_agg, kam)):
            a = agg.setdefault(key, {"fc": 0.0, "act": 0.0, "rec_fc": 0.0, "rec_act": 0.0,
                                     "ontop_fc": 0.0, "ontop_act": 0.0, "matched": 0.0,
                                     "missed": 0.0, "surprise": 0.0, "country": None,
                                     "wk": {w: {"fcb": 0.0, "fcr": 0.0, "fco": 0.0, "ar": 0.0, "ao": 0.0} for w in scored}})
            a["fc"] += fc; a["act"] += act; a["rec_fc"] += rec_fc; a["rec_act"] += rec_act
            a["ontop_fc"] += ontop_fc; a["ontop_act"] += oneoff_scored
            a["matched"] += matched; a["missed"] += missed; a["surprise"] += surprise
            for i, w in enumerate(scored):
                a["wk"][w]["fcb"] += baseline
                a["wk"][w]["fcr"] += wk.get(w, {}).get("regular", 0.0)
                a["wk"][w]["fco"] += wk.get(w, {}).get("ontop", 0.0)
                a["wk"][w]["ar"] += float(rec_act_arr[i])
                a["wk"][w]["ao"] += float(one_act_arr[i])
        sa = sku_agg.setdefault(sku, {"fc": 0.0, "act": 0.0, "abs_err": 0.0, "buyer": bname,
                                      "drivers": {}})
        sa["fc"] += fc; sa["act"] += act; sa["abs_err"] += abs(fc - act)
        sa["drivers"][driver] = sa["drivers"].get(driver, 0.0) + abs(fc - act)
        for fld, val in (("fc", fc), ("act", act), ("rec_fc", rec_fc), ("rec_act", rec_act),
                         ("ontop_fc", ontop_fc), ("ontop_act", oneoff_scored),
                         ("ontop_matched", matched), ("ontop_missed", missed), ("ontop_surprise", surprise)):
            g[fld] += val

    def pack_buyer(key, a):
        alias, bname, kam = key
        wkser = a["wk"]
        series = [{"yw": w,
                   "fc": round(d["fcb"] + d["fcr"] + d["fco"]), "act": round(d["ar"] + d["ao"]),
                   "fc_base": round(d["fcb"]), "fc_reg": round(d["fcr"]), "fc_ontop": round(d["fco"]),
                   "act_rec": round(d["ar"]), "act_one": round(d["ao"])}
                  for w, d in sorted(wkser.items())]
        return {"buyer": bname, "kam": kam, "country": (None),
                "forecast": round(a["fc"]), "actual": round(a["act"]),
                "fa_pct": _fa(a["fc"], a["act"]),
                "bias_pct": round((a["fc"] - a["act"]) / (a["act"] or 1) * 100, 1),
                "recurring_fa": _fa(a["rec_fc"], a["rec_act"]),
                "ontop_fa": _fa(a["ontop_fc"], a["ontop_act"]),
                "ontop_matched": round(a["matched"]), "ontop_missed": round(a["missed"]),
                "ontop_surprise": round(a["surprise"]),
                "forecast_components": {"baseline": round(sum(d["fcb"] for d in wkser.values())),
                                        "regular": round(sum(d["fcr"] for d in wkser.values())),
                                        "ontop": round(sum(d["fco"] for d in wkser.values()))},
                "actual_components": {"recurring": round(sum(d["ar"] for d in wkser.values())),
                                      "oneoff": round(sum(d["ao"] for d in wkser.values()))},
                "series": series}

    buyers = sorted([pack_buyer(k, a) for k, a in buyers_agg.items()],
                    key=lambda b: b["actual"], reverse=True)
    kams = sorted([{"kam": k, "forecast": round(a["fc"]), "actual": round(a["act"]),
                    "fa_pct": _fa(a["fc"], a["act"]),
                    "bias_pct": round((a["fc"] - a["act"]) / (a["act"] or 1) * 100, 1)}
                   for k, a in kam_agg.items()], key=lambda x: x["actual"], reverse=True)
    offenders = sorted(
        [{"product_id": s, "sku": sku_name.get(s, str(s)), "buyer": v["buyer"],
          "forecast": round(v["fc"]), "actual": round(v["act"]), "abs_error": round(v["abs_err"]),
          "driver": max(v["drivers"], key=v["drivers"].get) if v["drivers"] else "ok"}
         for s, v in sku_agg.items()],
        key=lambda x: x["abs_error"], reverse=True)[:10]

    # global per-week forecast vs actual (sum across buyers)
    weekly_map = {w: {"yw": w, "fc": 0, "act": 0,
                      "fc_base": 0, "fc_reg": 0, "fc_ontop": 0, "act_rec": 0, "act_one": 0} for w in scored}
    for b in buyers:
        for s in b["series"]:
            m = weekly_map[s["yw"]]
            for k in ("fc", "act", "fc_base", "fc_reg", "fc_ontop", "act_rec", "act_one"):
                m[k] += s[k]
    weekly = [weekly_map[w] for w in scored]

    return {
        "scored_weeks": scored, "latest_actual_yw": latest, "n_buyers": len(buyers),
        "weekly": weekly,
        "global": {"forecast": round(g["fc"]), "actual": round(g["act"]),
                   "fa_pct": _fa(g["fc"], g["act"]),
                   "bias_pct": round((g["fc"] - g["act"]) / (g["act"] or 1) * 100, 1),
                   "recurring_fa": _fa(g["rec_fc"], g["rec_act"]),
                   "ontop_fa": _fa(g["ontop_fc"], g["ontop_act"]),
                   "ontop_matched": round(g["ontop_matched"]), "ontop_missed": round(g["ontop_missed"]),
                   "ontop_surprise": round(g["ontop_surprise"]),
                   "forecast_components": {
                       "baseline": round(sum(b["forecast_components"]["baseline"] for b in buyers)),
                       "regular": round(sum(b["forecast_components"]["regular"] for b in buyers)),
                       "ontop": round(sum(b["forecast_components"]["ontop"] for b in buyers))},
                   "actual_components": {
                       "recurring": round(sum(b["actual_components"]["recurring"] for b in buyers)),
                       "oneoff": round(sum(b["actual_components"]["oneoff"] for b in buyers))}},
        "kams": kams, "buyers": buyers, "offenders": offenders,
    }