"""harness/transforms.py — governed ANALYTICS TRANSFORMS over query results (OM-4).

The model never touches SQL or raw math: it requests named transforms on a result_id
(tools.transform_result) and the platform computes them here, deterministically, over the exact
rows the governed query returned. The chain is RECORDED on the derived result ("transforms"), so
a saved view replays query -> transforms at render time (harness/views.run_view) and stays live.

This is the AIOS Analyst's TABLE-CALCULATION + ANALYTICS library — the open-source Tableau-class
capability set (Tableau table calcs + Analytics pane, Power BI DAX quick measures, pandas/polars
window ops, scipy/statsmodels stats) mapped onto ONE governed dispatcher. Grounding brief:
.claude/wiki/research/analytics-tools.md. Design laws (2026-07-18):

  1. ONE tool, many ops. Every capability is a pure function rows->rows (or rows->summary) in the
     OPS registry, dispatched through tools.transform_result — so the small model's prompt carries
     ONE tool + a skill file, never 50 signatures. Ops CHAIN.
  2. ADDITIVITY IS LAW (no-unverifiable-aggregates). Additivity comes from the metric's `agg` in
     the semantic model, NOT the column name: sum/count are additive; count_distinct and ratio are
     NOT. Accumulating ops (running_total, cum_share, share_of_total, moving_sum) REFUSE a known
     non-additive measure with a readable error; collapse ops (top_n's Other, add_total, pivot)
     land it as None, never a fabricated sum. A cumulative/rolling DISTINCT count must re-run the
     governed query per widening window (the ytd/rolling ops) — never sum displayed values.
  3. RE-AGGREGATION belongs in the QUERY, not here. Transforms are row-wise / window / reshape.
     To regroup or re-aggregate, run a new run_semantic_query with a different group_by — that keeps
     the parity proof. pivot is a pure RESHAPE (collision -> error, never a hidden sum).
  4. NO black-box models. Trend/forecast are transparent least-squares / seasonal-naive, fully
     hand-computable and unit-tested; we do not pull in Prophet/Merlion/k-means (our own AVOID
     briefs). Every op has a hand-computed unit test (_demo_analytics.py) — that suite is the ship
     gate, not the provider-flaky eval gate.
  5. Reference/trend/forecast LINES are delivered as ADDED COLUMNS and drawn with the existing
     combo/line chart kinds — the renderer stays untouched.
"""
import ast
import datetime as dt
import math

import harness.semantic as SEM


# ================================================================== additivity (law #2)

# Column-name suffixes that are inherently non-additive regardless of the base metric (ratios,
# indices, standardized/derived series that must never be re-summed).
_NON_ADDITIVE_SUFFIXES = (
    "_pct", "_ratio", "_share", "_share_pct", "_yoy_pct", "_cum_pct", "_pct_change",
    "_pct_of_max", "_rank_pct", "_zscore", "_norm", "_idx", "_index100", "_wavg",
    "_running_avg", "_ref", "_band_lo", "_band_hi", "_trend", "_ytd",
)
_NON_ADDITIVE_NAMES = {"aov", "margin_pct"}
_INHERIT_SUFFIXES = ("_ly", "_delta")     # <m>_ly / <m>_delta inherit <m>'s additivity


def _metric_agg(key):
    try:
        m = SEM.metrics().get(key)
    except Exception:
        m = None
    return m.get("agg") if m else None


def _base_metric(col):
    base = col
    for suf in _INHERIT_SUFFIXES:
        if base.endswith(suf):
            return base[: -len(suf)]
    return base


def known_non_additive(col):
    """True when we KNOW a column must not be summed across rows/groups: a non-additive suffix,
    a registered ratio/count_distinct metric, or a ratio/count_distinct-derived _ly/_delta. Unknown
    numeric columns default to additive (we only block what we can prove wrong)."""
    if col.endswith(_NON_ADDITIVE_SUFFIXES) or col.endswith("_z") or col in _NON_ADDITIVE_NAMES:
        return True
    return _metric_agg(_base_metric(col)) in ("count_distinct", "ratio")


def _is_additive(col):
    return not known_non_additive(col)


# ================================================================== small helpers

def _numeric(v):
    return isinstance(v, (int, float)) and not isinstance(v, bool)


def _num_cols(rows):
    return [k for k in (rows[0] if rows else {}) if any(_numeric(r.get(k)) for r in rows)]


def _text_cols(rows):
    nums = set(_num_cols(rows))
    return [k for k in (rows[0] if rows else {}) if k not in nums]


def _need_col(rows, col, what):
    if not rows:
        raise SEM.ModelError("cannot transform an empty result")
    if col not in rows[0]:
        raise SEM.ModelError(f"{what}={col!r} not in result columns {sorted(rows[0])}")


def _need_additive(rows, of, op):
    _need_col(rows, of, "of")
    if known_non_additive(of):
        raise SEM.ModelError(
            f"{op} sums {of!r} across rows, but {of!r} is a distinct-count or ratio measure and is "
            f"NOT additive — summing it would over-count. For a cumulative/rolling distinct count "
            f"use the 'ytd' or 'rolling' op (they re-run the query per window); otherwise pick an "
            f"additive measure (revenue, units, margin, orders, cogs).")


def _vals(rows, col):
    return [r.get(col) for r in rows if _numeric(r.get(col))]


def _mean(v):
    return (sum(v) / len(v)) if v else None


def _median(v):
    s = sorted(v)
    n = len(s)
    if not n:
        return None
    return s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2.0


def _std(v):
    """Sample standard deviation (ddof=1); None for n<2."""
    n = len(v)
    if n < 2:
        return None
    m = sum(v) / n
    return math.sqrt(sum((x - m) ** 2 for x in v) / (n - 1))


def _percentile(v, p):
    """Linear-interpolation percentile, p in [0,100] (the numpy 'linear' method)."""
    s = sorted(v)
    n = len(s)
    if not n:
        return None
    if n == 1:
        return float(s[0])
    k = (n - 1) * (p / 100.0)
    lo = math.floor(k)
    hi = math.ceil(k)
    if lo == hi:
        return float(s[int(k)])
    return s[lo] + (s[hi] - s[lo]) * (k - lo)


# ================================================================== ordering / rank

def sort_rows(rows, by, direction="desc"):
    """Sort rows by a column, Nones last either way; works for numeric and text columns."""
    _need_col(rows, by, "by")
    present = [r for r in rows if r.get(by) is not None]
    absent = [r for r in rows if r.get(by) is None]
    return sorted(present, key=lambda r: r[by], reverse=(direction != "asc")) + absent


def head(rows, n=20):
    """Keep the FIRST n rows in the current order — pairs with sort for 'most negative first'
    asks, where top_n (largest-by-value) is the wrong shape."""
    return rows[: max(1, min(int(n or 20), 500))]


def bottom_n(rows, by, n=10):
    """The N SMALLEST rows by `by` (sort ascending + take N) — Tableau's 'Bottom N'."""
    _need_col(rows, by, "by")
    return sort_rows(rows, by, "asc")[: max(1, min(int(n or 10), 500))]


def rank_rows(rows, by, direction="desc", out="rank"):
    """Sort by `by` and add a 1-based rank column (ties keep row order — 'first' method)."""
    ordered = sort_rows(rows, by, direction)
    return [{**r, out: i + 1} for i, r in enumerate(ordered)]


def rank_pct(rows, by, direction="desc", out="rank_pct"):
    """Percentile rank (0-100): position among the rows after sorting by `by`. Top row = 100
    (desc) — 'this row beats X% of the rest'."""
    ordered = sort_rows(rows, by, direction)
    n = len([r for r in ordered if r.get(by) is not None])
    res = []
    for i, r in enumerate(ordered):
        pr = (100.0 * (n - 1 - i) / (n - 1)) if (n > 1 and r.get(by) is not None) else (
            100.0 if r.get(by) is not None else None)
        res.append({**r, out: pr})
    return res


def ntile(rows, by, tiles=4, direction="asc", out="ntile"):
    """Assign each row to one of `tiles` equal-count buckets by `by` (quartile=4, decile=10).
    direction='asc' -> tile 1 is the smallest values (Tableau/pandas qcut convention)."""
    _need_col(rows, by, "by")
    tiles = max(2, min(int(tiles or 4), 100))
    ordered = sort_rows(rows, by, direction)
    present = [r for r in ordered if r.get(by) is not None]
    n = len(present)
    res = []
    for i, r in enumerate(ordered):
        if r.get(by) is None:
            res.append({**r, out: None})
        else:
            res.append({**r, out: min(tiles, int(i * tiles / n) + 1)})
    return res


# ================================================================== top-N / composition

def top_n(rows, by, n=10, other=True, other_label="Other"):
    """Keep the N largest rows by `by`; the rest collapse into ONE labelled bucket (additive
    columns summed, non-additive columns None — never a fake average). other=False just truncates,
    and the caller must surface the cut (the tool notes it)."""
    _need_col(rows, by, "by")
    n = max(1, min(int(n or 10), 500))
    ordered = sort_rows(rows, by, "desc")
    head_rows, tail = ordered[:n], ordered[n:]
    if not tail or not other:
        return head_rows
    bucket = _collapse(rows, tail, f"{other_label} ({len(tail)})")
    return head_rows + [bucket]


def add_total(rows, label="Total"):
    """Append a grand-total row: additive columns summed, non-additive columns None (honest),
    the first text column = `label`. Same discipline as make_table's totals row, but in-data so a
    chart can show it."""
    if not rows:
        raise SEM.ModelError("cannot total an empty result")
    return rows + [_collapse(rows, rows, label)]


def _collapse(all_rows, subset, label):
    bucket = {}
    for k in all_rows[0]:
        vals = [t.get(k) for t in subset if _numeric(t.get(k))]
        if vals and _is_additive(k):
            bucket[k] = sum(vals)
        elif vals:
            bucket[k] = None
        else:
            bucket[k] = ""
    texts = _text_cols(all_rows)
    if texts:
        bucket[texts[0]] = label
    return bucket


# ================================================================== part-to-whole / cumulative

def share_of_total(rows, of, out=None):
    """Add `<of>_share_pct` (0-100): each row's share of the column total across THESE rows."""
    _need_additive(rows, of, "share_of_total")
    out = out or f"{of}_share_pct"
    total = sum(r.get(of) or 0 for r in rows)
    return [{**r, out: (100.0 * (r.get(of) or 0) / total) if total else None} for r in rows]


def cum_share(rows, of, out=None):
    """Pareto prep: sort desc by `of`, add cumulative share % (0-100). The 'top X carry Y%' read."""
    _need_additive(rows, of, "cum_share")
    out = out or f"{of}_cum_pct"
    ordered = sort_rows(rows, of, "desc")
    total = sum(r.get(of) or 0 for r in ordered)
    run, res = 0.0, []
    for r in ordered:
        run += r.get(of) or 0
        res.append({**r, out: (100.0 * run / total) if total else None})
    return res


# ================================================================== running / moving (window)

def running_total(rows, of, out=None):
    """Add `<of>_running`: cumulative sum in the rows' current order (sort first if needed)."""
    _need_additive(rows, of, "running_total")
    out = out or f"{of}_running"
    run, res = 0.0, []
    for r in rows:
        run += r.get(of) or 0
        res.append({**r, out: run})
    return res


def running_avg(rows, of, out=None):
    """Add `<of>_running_avg`: expanding (cumulative) mean in row order — a smoothing, valid on
    any numeric column (it never claims a total)."""
    _need_col(rows, of, "of")
    out = out or f"{of}_running_avg"
    tot, cnt, res = 0.0, 0, []
    for r in rows:
        v = r.get(of)
        if _numeric(v):
            tot += v
            cnt += 1
        res.append({**r, out: (tot / cnt) if cnt else None})
    return res


def running_extreme(rows, of, kind="max", out=None):
    """Add `<of>_running_max` / `_running_min`: the cumulative max/min so far, in row order."""
    _need_col(rows, of, "of")
    out = out or f"{of}_running_{kind}"
    best, res = None, []
    for r in rows:
        v = r.get(of)
        if _numeric(v):
            best = v if best is None else (max(best, v) if kind == "max" else min(best, v))
        res.append({**r, out: best})
    return res


def moving_average(rows, of, window=3, out=None):
    """Add `<of>_ma<window>`: trailing moving average in row order; the first window-1 rows get
    None (a partial-window average reads as a level change and lies). A smoothing — any numeric."""
    _need_col(rows, of, "of")
    window = max(2, min(int(window or 3), 24))
    out = out or f"{of}_ma{window}"
    vals = [r.get(of) or 0 for r in rows]
    res = []
    for i, r in enumerate(rows):
        ma = sum(vals[i - window + 1:i + 1]) / window if i >= window - 1 else None
        res.append({**r, out: ma})
    return res


def moving_sum(rows, of, window=3, out=None):
    """Add `<of>_msum<window>`: trailing moving SUM (additive measures only) — the first window-1
    rows get None."""
    _need_additive(rows, of, "moving_sum")
    window = max(2, min(int(window or 3), 24))
    out = out or f"{of}_msum{window}"
    vals = [r.get(of) or 0 for r in rows]
    res = []
    for i, r in enumerate(rows):
        ms = sum(vals[i - window + 1:i + 1]) if i >= window - 1 else None
        res.append({**r, out: ms})
    return res


def moving_median(rows, of, window=3, out=None):
    """Add `<of>_mmed<window>`: trailing moving MEDIAN (robust smoothing; outlier-resistant)."""
    _need_col(rows, of, "of")
    window = max(2, min(int(window or 3), 24))
    out = out or f"{of}_mmed{window}"
    res = []
    for i, r in enumerate(rows):
        if i >= window - 1:
            win = [rows[j].get(of) for j in range(i - window + 1, i + 1) if _numeric(rows[j].get(of))]
            res.append({**r, out: _median(win) if win else None})
        else:
            res.append({**r, out: None})
    return res


def rolling_std(rows, of, window=3, out=None):
    """Add `<of>_rstd<window>`: trailing-window SAMPLE standard deviation — demand/sales
    VOLATILITY over time (feeds control-limit / safety-stock work). First window-1 rows None."""
    _need_col(rows, of, "of")
    window = max(2, min(int(window or 3), 24))
    out = out or f"{of}_rstd{window}"
    res = []
    for i, r in enumerate(rows):
        if i >= window - 1:
            win = [rows[j].get(of) for j in range(i - window + 1, i + 1) if _numeric(rows[j].get(of))]
            res.append({**r, out: _std(win)})
        else:
            res.append({**r, out: None})
    return res


def running_count(rows, out="running_count"):
    """Add `running_count`: cumulative 1-based ROW count in the current order (cumulative # of
    orders/SKUs/whatever the rows are). Counts rows — always safe (never a distinct-measure sum)."""
    return [{**r, out: i + 1} for i, r in enumerate(rows)]


# ================================================================== period-over-period (row-wise)

def diff(rows, of, out=None):
    """Add `<of>_diff`: value minus the previous row's value (first row None). Row-wise — safe on
    any numeric column (period-over-period change, incl. of a ratio)."""
    _need_col(rows, of, "of")
    out = out or f"{of}_diff"
    res, prev = [], None
    for r in rows:
        v = r.get(of)
        res.append({**r, out: (v - prev) if _numeric(v) and _numeric(prev) else None})
        prev = v
    return res


def pct_change(rows, of, out=None):
    """Add `<of>_pct_change` (0-100 signed): percent change from the previous row. prev=0 -> None."""
    _need_col(rows, of, "of")
    out = out or f"{of}_pct_change"
    res, prev = [], None
    for r in rows:
        v = r.get(of)
        pc = (100.0 * (v - prev) / abs(prev)) if _numeric(v) and _numeric(prev) and prev else None
        res.append({**r, out: pc})
        prev = v
    return res


def lag(rows, of, k=1, out=None):
    """Add `<of>_lag<k>`: the value from k rows earlier (Tableau LOOKUP / pandas shift)."""
    _need_col(rows, of, "of")
    k = max(1, min(int(k or 1), 100))
    out = out or f"{of}_lag{k}"
    vals = [r.get(of) for r in rows]
    return [{**r, out: (vals[i - k] if i - k >= 0 else None)} for i, r in enumerate(rows)]


def lead(rows, of, k=1, out=None):
    """Add `<of>_lead<k>`: the value from k rows later."""
    _need_col(rows, of, "of")
    k = max(1, min(int(k or 1), 100))
    out = out or f"{of}_lead{k}"
    vals = [r.get(of) for r in rows]
    n = len(rows)
    return [{**r, out: (vals[i + k] if i + k < n else None)} for i, r in enumerate(rows)]


def diff_from_first(rows, of, out=None):
    """Add `<of>_vs_first`: value minus the FIRST row's value (Tableau 'difference from first')."""
    _need_col(rows, of, "of")
    out = out or f"{of}_vs_first"
    base = next((r.get(of) for r in rows if _numeric(r.get(of))), None)
    return [{**r, out: (r.get(of) - base) if _numeric(r.get(of)) and _numeric(base) else None}
            for r in rows]


def index_to_100(rows, of, out=None):
    """Add `<of>_idx`: rebase the series to 100 at the first value (index-to-100 — compare shapes
    of series at different levels). first=0 -> None."""
    _need_col(rows, of, "of")
    out = out or f"{of}_idx"
    base = next((r.get(of) for r in rows if _numeric(r.get(of))), None)
    return [{**r, out: (100.0 * r.get(of) / base) if _numeric(r.get(of)) and base else None}
            for r in rows]


def percent_of_max(rows, of, out=None):
    """Add `<of>_pct_of_max` (0-100): each row as a % of the largest value ('how far below best')."""
    _need_col(rows, of, "of")
    out = out or f"{of}_pct_of_max"
    vals = _vals(rows, of)
    mx = max(vals) if vals else None
    return [{**r, out: (100.0 * r.get(of) / mx) if _numeric(r.get(of)) and mx else None}
            for r in rows]


def compare(rows, a, b, how="diff", out=None):
    """Column-wise comparison of two EXISTING columns per row (e.g. revenue vs revenue_ly already
    in the result): how='diff' (a-b), 'pct' (100*(a-b)/|b|), or 'ratio' (a/b). Distinct from `diff`
    (which is cross-row on ONE column). Safe on any columns; b=0 -> None for pct/ratio."""
    _need_col(rows, a, "a")
    _need_col(rows, b, "b")
    if how not in ("diff", "pct", "ratio"):
        raise SEM.ModelError("how must be diff|pct|ratio")
    out = out or f"{a}_vs_{b}_{how}"

    def _cmp(x, y):
        if not (_numeric(x) and _numeric(y)):
            return None
        if how == "diff":
            return x - y
        if not y:
            return None
        return (100.0 * (x - y) / abs(y)) if how == "pct" else (x / y)
    return [{**r, out: _cmp(r.get(a), r.get(b))} for r in rows]


# ================================================================== distribution / stats

def bin_values(rows, of, bins=10):
    """Equal-width histogram buckets over `of` across these rows -> one row per bucket:
    {bucket, bucket_lo, bucket_hi, count, <of>_sum}. Empty buckets kept (an honest gap)."""
    _need_col(rows, of, "of")
    bins = max(2, min(int(bins or 10), 50))
    vals = _vals(rows, of)
    if not vals:
        raise SEM.ModelError(f"no numeric values in {of!r} to bin")
    lo, hi = min(vals), max(vals)
    if lo == hi:
        return [{"bucket": f"{lo:,.4g}", "bucket_lo": lo, "bucket_hi": hi,
                 "count": len(vals), f"{of}_sum": sum(vals)}]
    width = (hi - lo) / bins
    out = []
    for i in range(bins):
        b_lo, b_hi = lo + i * width, lo + (i + 1) * width
        hit = [v for v in vals if (b_lo <= v < b_hi) or (i == bins - 1 and v == hi)]
        out.append({"bucket": f"{b_lo:,.4g} to {b_hi:,.4g}", "bucket_lo": b_lo, "bucket_hi": b_hi,
                    "count": len(hit), f"{of}_sum": sum(hit)})
    return out


def describe(rows, of):
    """Summary statistics of `of` -> ONE row: count, mean, median, std, min, p25, p75, max
    (+ sum when the measure is additive). The pandas .describe() of a column."""
    _need_col(rows, of, "of")
    v = _vals(rows, of)
    if not v:
        raise SEM.ModelError(f"no numeric values in {of!r} to describe")
    row = {"stat_of": of, "count": len(v), "mean": _mean(v), "median": _median(v),
           "std": _std(v), "min": min(v), "p25": _percentile(v, 25),
           "p75": _percentile(v, 75), "max": max(v)}
    if _is_additive(of):
        row["sum"] = sum(v)
    return [row]


def zscore(rows, of, out=None):
    """Add `<of>_zscore`: standardized value (v-mean)/std (sample std). Constant column -> None."""
    _need_col(rows, of, "of")
    out = out or f"{of}_zscore"
    v = _vals(rows, of)
    m, s = _mean(v), _std(v)
    return [{**r, out: ((r.get(of) - m) / s if _numeric(r.get(of)) and s else None)} for r in rows]


def outliers(rows, of, method="zscore", k=None, out=None):
    """Add `<of>_outlier` (bool): flag statistical outliers. method='zscore' (|z|>k, default 3) or
    'iqr' (outside [Q1-k*IQR, Q3+k*IQR], default k=1.5 — Tukey's fences)."""
    _need_col(rows, of, "of")
    out = out or f"{of}_outlier"
    v = _vals(rows, of)
    if method == "iqr":
        k = 1.5 if k is None else float(k)
        q1, q3 = _percentile(v, 25), _percentile(v, 75)
        iqr = (q3 - q1) if (q1 is not None and q3 is not None) else None
        lo = (q1 - k * iqr) if iqr is not None else None
        hi = (q3 + k * iqr) if iqr is not None else None
        return [{**r, out: (bool(r.get(of) < lo or r.get(of) > hi)
                            if _numeric(r.get(of)) and lo is not None else None)} for r in rows]
    k = 3.0 if k is None else float(k)
    m, s = _mean(v), _std(v)
    return [{**r, out: (abs((r.get(of) - m) / s) > k if _numeric(r.get(of)) and s else None)}
            for r in rows]


def winsorize(rows, of, p=5, out=None):
    """Add `<of>_winsor`: `of` clipped to its [p, 100-p] percentiles (tame outliers before a mean
    or chart without dropping rows)."""
    _need_col(rows, of, "of")
    p = max(0.0, min(float(p or 5), 49.0))
    out = out or f"{of}_winsor"
    v = _vals(rows, of)
    lo, hi = _percentile(v, p), _percentile(v, 100 - p)
    return [{**r, out: (min(max(r.get(of), lo), hi) if _numeric(r.get(of)) else None)} for r in rows]


def clip(rows, of, lo=None, hi=None, out=None):
    """Add `<of>_clip`: `of` clamped to [lo, hi] (either bound optional)."""
    _need_col(rows, of, "of")
    if lo is None and hi is None:
        raise SEM.ModelError("clip needs lo and/or hi")
    out = out or f"{of}_clip"

    def _c(x):
        if not _numeric(x):
            return None
        if lo is not None:
            x = max(x, lo)
        if hi is not None:
            x = min(x, hi)
        return x
    return [{**r, out: _c(r.get(of))} for r in rows]


def normalize(rows, of, out=None):
    """Add `<of>_norm` (0-1): min-max scale of `of`. Constant column -> 0.0 for all."""
    _need_col(rows, of, "of")
    out = out or f"{of}_norm"
    v = _vals(rows, of)
    lo, hi = (min(v), max(v)) if v else (None, None)
    span = (hi - lo) if (lo is not None) else None
    return [{**r, out: ((r.get(of) - lo) / span if span else 0.0) if _numeric(r.get(of)) else None}
            for r in rows]


def correlate(rows, x, y):
    """Pearson correlation between two columns -> ONE row {x, y, pearson_r, n}. r in [-1,1];
    'do these two measures move together?'. Zero-variance column -> r None."""
    _need_col(rows, x, "x")
    _need_col(rows, y, "y")
    pairs = [(r[x], r[y]) for r in rows if _numeric(r.get(x)) and _numeric(r.get(y))]
    n = len(pairs)
    if n < 2:
        return [{"x": x, "y": y, "pearson_r": None, "n": n}]
    xs, ys = [p[0] for p in pairs], [p[1] for p in pairs]
    mx, my = _mean(xs), _mean(ys)
    cov = sum((a - mx) * (b - my) for a, b in pairs)
    sx = math.sqrt(sum((a - mx) ** 2 for a in xs))
    sy = math.sqrt(sum((b - my) ** 2 for b in ys))
    r = (cov / (sx * sy)) if (sx and sy) else None
    return [{"x": x, "y": y, "pearson_r": r, "n": n}]


def weighted_average(rows, of, weight):
    """Weighted mean of `of` by `weight` -> ONE row {<of>_wavg, weight_col, n}. The correct way to
    average a per-unit figure (e.g. price weighted by units) — never a mean of means."""
    _need_col(rows, of, "of")
    _need_col(rows, weight, "weight")
    num = sum((r[of] * r[weight]) for r in rows if _numeric(r.get(of)) and _numeric(r.get(weight)))
    den = sum(r[weight] for r in rows if _numeric(r.get(of)) and _numeric(r.get(weight)))
    return [{f"{of}_wavg": (num / den) if den else None, "weight_col": weight,
             "n": sum(1 for r in rows if _numeric(r.get(of)) and _numeric(r.get(weight)))}]


def safe_ratio(rows, numerator, denominator, out="ratio"):
    """Add a row-wise ratio `numerator/denominator` (denominator 0 -> None). Build an ad-hoc rate
    the semantic layer doesn't predefine, correctly guarded."""
    _need_col(rows, numerator, "numerator")
    _need_col(rows, denominator, "denominator")
    return [{**r, out: ((r[numerator] / r[denominator])
                        if _numeric(r.get(numerator)) and r.get(denominator) else None)}
            for r in rows]


def product(rows, a, b, out="product"):
    """Add a row-wise product `a*b` (e.g. price * quantity)."""
    _need_col(rows, a, "a")
    _need_col(rows, b, "b")
    return [{**r, out: (r[a] * r[b] if _numeric(r.get(a)) and _numeric(r.get(b)) else None)}
            for r in rows]


# ================================================================== business templates

def abc_classify(rows, of, a=80, b=95):
    """Pareto ABC classification: sort desc by `of`, add `<of>_cum_pct` and `abc_class` (A = the
    vital few up to a% of the total, B up to b%, C the long tail). The classic 80/20 inventory /
    customer / SKU segmentation."""
    _need_additive(rows, of, "abc_classify")
    a, b = float(a), float(b)
    ordered = sort_rows(rows, of, "desc")
    total = sum(r.get(of) or 0 for r in ordered)
    run, res = 0.0, []
    for r in ordered:
        run += r.get(of) or 0
        cum = (100.0 * run / total) if total else None
        cls = None if cum is None else ("A" if cum <= a else ("B" if cum <= b else "C"))
        res.append({**r, f"{of}_cum_pct": cum, "abc_class": cls})
    return res


def concentration(rows, of):
    """Concentration statistics of `of` across these rows -> ONE row: HHI (Herfindahl-Hirschman
    Index, 0-10000), Gini (0-1), and top-1/5/10 share %. 'How concentrated is the book?'."""
    _need_additive(rows, of, "concentration")
    vals = [r.get(of) or 0 for r in rows if _numeric(r.get(of))]
    vals = [v for v in vals if v > 0]
    n = len(vals)
    total = sum(vals)
    if not total:
        return [{"of": of, "n": n, "hhi": None, "gini": None,
                 "top1_share_pct": None, "top5_share_pct": None, "top10_share_pct": None}]
    shares = [v / total for v in vals]
    hhi = sum(s * s for s in shares) * 10000.0
    asc = sorted(vals)
    gini = (2.0 * sum((i + 1) * x for i, x in enumerate(asc))) / (n * total) - (n + 1.0) / n
    desc = sorted(vals, reverse=True)

    def topk(k):
        return 100.0 * sum(desc[:k]) / total
    return [{"of": of, "n": n, "hhi": hhi, "gini": gini, "top1_share_pct": topk(1),
             "top5_share_pct": topk(5), "top10_share_pct": topk(10)}]


def contribution_to_change(rows, of):
    """Given a `<of>_delta` column (run yoy first), add `<of>_delta_share_pct`: each row's share of
    the TOTAL change (who drove the movement — the bridge / contribution decomposition)."""
    delta = f"{of}_delta"
    _need_col(rows, delta, "of (expected <of>_delta from yoy)")
    total = sum(r.get(delta) or 0 for r in rows)
    return [{**r, f"{of}_delta_share_pct": (100.0 * (r.get(delta) or 0) / total) if total else None}
            for r in rows]


def rfm(rows, recency, frequency, monetary, tiles=5):
    """RFM scoring: quintile-score each customer on Recency (LOWER days = better), Frequency and
    Monetary (higher = better), 1..tiles. Adds r_score/f_score/m_score, rfm_cell ('545'),
    rfm_score (sum) and rfm_segment (Champions / Loyal / Potential / At Risk / Lost / Others)."""
    for c, nm in ((recency, "recency"), (frequency, "frequency"), (monetary, "monetary")):
        _need_col(rows, c, nm)
    tiles = max(2, min(int(tiles or 5), 10))

    def _score(col, reverse):
        # reverse=True -> smaller value scores higher (recency). Bucket by sorted position.
        ordered = sort_rows(rows, col, "asc")
        present = [r for r in ordered if r.get(col) is not None]
        n = len(present)
        sc = {}
        for i, r in enumerate(ordered):
            if r.get(col) is None:
                sc[id(r)] = None
            else:
                t = min(tiles, int(i * tiles / n) + 1)
                sc[id(r)] = (tiles + 1 - t) if reverse else t
        return sc
    rs, fs, ms = _score(recency, True), _score(frequency, False), _score(monetary, False)
    out = []
    for r in rows:
        rr, ff, mm = rs[id(r)], fs[id(r)], ms[id(r)]
        seg = _rfm_segment(rr, ff, mm, tiles)
        out.append({**r, "r_score": rr, "f_score": ff, "m_score": mm,
                    "rfm_cell": (f"{rr}{ff}{mm}" if None not in (rr, ff, mm) else None),
                    "rfm_score": (rr + ff + mm if None not in (rr, ff, mm) else None),
                    "rfm_segment": seg})
    return out


def _rfm_segment(r, f, m, tiles):
    if None in (r, f, m):
        return None
    hi = tiles - 1
    lo = 2
    if r >= hi and f >= hi:
        return "Champions"
    if f >= hi:
        return "Loyal"
    if r >= hi:
        return "Recent / Promising"
    if r <= lo and f >= 3:
        return "At Risk"
    if r <= lo and f <= lo:
        return "Lost"
    return "Others"


def funnel_rates(rows, of):
    """Stage conversion: over ordered stage rows carrying a count `of`, add `<of>_step_pct` (vs the
    previous stage) and `<of>_overall_pct` (vs the first stage). The funnel drop-off read."""
    _need_col(rows, of, "of")
    first = next((r.get(of) for r in rows if _numeric(r.get(of))), None)
    res, prev = [], None
    for r in rows:
        v = r.get(of)
        step = (100.0 * v / prev) if _numeric(v) and _numeric(prev) and prev else None
        overall = (100.0 * v / first) if _numeric(v) and first else None
        res.append({**r, f"{of}_step_pct": step, f"{of}_overall_pct": overall})
        prev = v
    return res


# ================================================================== modeling (transparent)

def _ols(xs, ys):
    """Ordinary least squares y = slope*x + intercept over paired numerics -> (slope, intercept,
    r2). None,None,None if degenerate."""
    pts = [(x, y) for x, y in zip(xs, ys) if _numeric(x) and _numeric(y)]
    n = len(pts)
    if n < 2:
        return None, None, None
    mx = sum(p[0] for p in pts) / n
    my = sum(p[1] for p in pts) / n
    sxx = sum((p[0] - mx) ** 2 for p in pts)
    sxy = sum((p[0] - mx) * (p[1] - my) for p in pts)
    if not sxx:
        return None, None, None
    slope = sxy / sxx
    intercept = my - slope * mx
    syy = sum((p[1] - my) ** 2 for p in pts)
    ss_res = sum((y - (slope * x + intercept)) ** 2 for x, y in pts)
    r2 = (1 - ss_res / syy) if syy else None
    return slope, intercept, r2


def trend_line(rows, of, out=None):
    """Add `<of>_trend`: the linear least-squares fitted value (a straight trend over the rows'
    order) — draw it over `of` with a combo/line chart. The Analytics-pane trend line."""
    _need_col(rows, of, "of")
    out = out or f"{of}_trend"
    xs = list(range(len(rows)))
    ys = [r.get(of) for r in rows]
    slope, intercept, _ = _ols(xs, ys)
    if slope is None:
        return [{**r, out: None} for r in rows]
    return [{**r, out: slope * i + intercept} for i, r in enumerate(rows)]


def regression(rows, x, y):
    """Linear regression of `y` on `x` -> ONE row {slope, intercept, r2, n}. 'Is there a
    relationship, how strong?' (r2 near 1 = tight fit)."""
    _need_col(rows, x, "x")
    _need_col(rows, y, "y")
    slope, intercept, r2 = _ols([r.get(x) for r in rows], [r.get(y) for r in rows])
    n = sum(1 for r in rows if _numeric(r.get(x)) and _numeric(r.get(y)))
    return [{"x": x, "y": y, "slope": slope, "intercept": intercept, "r2": r2, "n": n}]


def cagr(rows, of):
    """Compound annual (per-period) growth rate first->last -> ONE row {cagr_pct, periods}.
    ((last/first)^(1/periods) - 1) * 100. Needs first > 0."""
    _need_col(rows, of, "of")
    vals = [r.get(of) for r in rows if _numeric(r.get(of))]
    if len(vals) < 2:
        return [{"of": of, "cagr_pct": None, "periods": max(0, len(vals) - 1)}]
    first, last = vals[0], vals[-1]
    p = len(vals) - 1
    c = ((last / first) ** (1.0 / p) - 1) * 100.0 if first > 0 and last > 0 else None
    return [{"of": of, "cagr_pct": c, "periods": p}]


def growth_rate(rows, of):
    """Total growth first->last -> ONE row {growth_pct, first, last}. (last-first)/|first| * 100."""
    _need_col(rows, of, "of")
    vals = [r.get(of) for r in rows if _numeric(r.get(of))]
    if len(vals) < 2:
        return [{"of": of, "growth_pct": None, "first": (vals[0] if vals else None),
                 "last": (vals[-1] if vals else None)}]
    first, last = vals[0], vals[-1]
    g = (100.0 * (last - first) / abs(first)) if first else None
    return [{"of": of, "growth_pct": g, "first": first, "last": last}]


# ================================================================== reference annotations (columns)

def reference_line(rows, of, stat="mean", out=None):
    """Add a constant column = a summary stat of `of` (mean|median|min|max), so a combo/line chart
    can draw the reference line. The Analytics-pane average/median/constant line."""
    _need_col(rows, of, "of")
    out = out or f"{of}_ref"
    v = _vals(rows, of)
    val = {"mean": _mean(v), "median": _median(v), "min": (min(v) if v else None),
           "max": (max(v) if v else None)}.get(stat)
    if stat not in ("mean", "median", "min", "max"):
        raise SEM.ModelError("stat must be mean|median|min|max")
    return [{**r, out: val} for r in rows]


def reference_band(rows, of, method="stddev", k=1):
    """Add `<of>_band_lo`/`<of>_band_hi` constant columns for a shaded reference band. method
    'stddev' -> mean ± k·std; method 'percentile' -> the k-th and (100-k)-th percentiles."""
    _need_col(rows, of, "of")
    v = _vals(rows, of)
    if method == "percentile":
        lo, hi = _percentile(v, float(k)), _percentile(v, 100 - float(k))
    elif method == "stddev":
        m, s = _mean(v), _std(v)
        lo = (m - float(k) * s) if (m is not None and s is not None) else None
        hi = (m + float(k) * s) if (m is not None and s is not None) else None
    else:
        raise SEM.ModelError("method must be stddev|percentile")
    return [{**r, f"{of}_band_lo": lo, f"{of}_band_hi": hi} for r in rows]


def target_line(rows, value, out="target"):
    """Add a constant `target` column = value — for a bullet chart (actual vs target) or a goal
    line on a combo chart."""
    if not _numeric(value):
        raise SEM.ModelError("target_line needs a numeric value")
    return [{**r, out: value} for r in rows]


def xmr_limits(rows, of):
    """Wheeler XmR (process-behaviour) control limits on `of` in row order, as constant columns:
    `<of>_center` (mean), `<of>_ucl`/`<of>_lcl` (center ± 2.66·mR-bar, mR-bar = mean moving range),
    and `<of>_signal` (bool: this point is outside the limits). The HONEST, business-native
    alternative to mean±k·std for spotting real signals in a noisy monthly series — the same XmR
    the warehouse/expenses modules use. Draw center/ucl/lcl as reference lines over `of`."""
    _need_col(rows, of, "of")
    vals = [r.get(of) for r in rows]
    nums = [v for v in vals if _numeric(v)]
    center = _mean(nums)
    mr = [abs(vals[i] - vals[i - 1]) for i in range(1, len(vals))
          if _numeric(vals[i]) and _numeric(vals[i - 1])]
    mrbar = _mean(mr)
    ucl = (center + 2.66 * mrbar) if (center is not None and mrbar is not None) else None
    lcl = (center - 2.66 * mrbar) if (center is not None and mrbar is not None) else None
    out = []
    for r in rows:
        v = r.get(of)
        sig = (bool(v > ucl or v < lcl) if _numeric(v) and ucl is not None else None)
        out.append({**r, f"{of}_center": center, f"{of}_ucl": ucl, f"{of}_lcl": lcl,
                    f"{of}_signal": sig})
    return out


# ================================================================== reshape (pure, no hidden agg)

def pivot(rows, index, column, value):
    """Long -> wide: one row per `index`, one column per distinct `column` value, cell = `value`.
    A PURE reshape — if two source rows share an (index, column) pair it RAISES (never a hidden
    sum; regroup in the query instead). Missing cells are None."""
    for c, nm in ((index, "index"), (column, "column"), (value, "value")):
        _need_col(rows, c, nm)
    cols, order, out_map = set(), [], {}
    for r in rows:
        idx, col = r.get(index), r.get(column)
        cols.add(col)
        if idx not in out_map:
            out_map[idx] = {index: idx}
            order.append(idx)
        cell = str(col)
        if cell in out_map[idx]:
            raise SEM.ModelError(
                f"pivot collision: {index}={idx!r} has two rows for {column}={col!r} — pivot cannot "
                f"aggregate (that would hide a sum); regroup the query so (index, column) is unique")
        out_map[idx][cell] = r.get(value)
    col_names = [str(c) for c in sorted(cols, key=lambda z: (z is None, z))]
    return [{index: out_map[idx][index], **{cn: out_map[idx].get(cn) for cn in col_names}}
            for idx in order]


def unpivot(rows, keep, columns, var_name="metric", value_name="value"):
    """Wide -> long (melt): for each row, emit one output row per `columns` entry carrying the
    `keep` columns plus (`var_name`, `value_name`). The inverse of pivot."""
    keep = keep if isinstance(keep, (list, tuple)) else [keep]
    columns = columns if isinstance(columns, (list, tuple)) else [columns]
    for c in list(keep) + list(columns):
        _need_col(rows, c, "column")
    out = []
    for r in rows:
        base = {k: r.get(k) for k in keep}
        for c in columns:
            out.append({**base, var_name: c, value_name: r.get(c)})
    return out


def filter_rows(rows, col, cmp, value):
    """Keep rows where `col` `cmp` `value`. cmp in >, >=, <, <=, ==, !=, contains. The 'having'
    clause over a result (e.g. keep customers with revenue_yoy_pct < 0). (Named `cmp`, not `op`,
    to avoid colliding with the transform dispatch key.)"""
    _need_col(rows, col, "col")
    ops = {">": lambda a, b: a > b, ">=": lambda a, b: a >= b, "<": lambda a, b: a < b,
           "<=": lambda a, b: a <= b, "==": lambda a, b: a == b, "!=": lambda a, b: a != b,
           "contains": lambda a, b: str(b).lower() in str(a).lower()}
    if cmp not in ops:
        raise SEM.ModelError(f"cmp must be one of {sorted(ops)}")
    fn = ops[cmp]
    out = []
    for r in rows:
        v = r.get(col)
        try:
            if cmp in (">", ">=", "<", "<=") and not _numeric(v):
                continue
            if fn(v, value):
                out.append(r)
        except TypeError:
            continue
    return out


def dedupe(rows, by=None):
    """Keep the FIRST row per distinct key. by = a column or list of columns (default: the whole
    row). Distinct rows, order-preserving."""
    if by is None:
        keys = list(rows[0]) if rows else []
    else:
        keys = by if isinstance(by, (list, tuple)) else [by]
        for k in keys:
            _need_col(rows, k, "by")
    seen, out = set(), []
    for r in rows:
        key = tuple(r.get(k) for k in keys)
        if key not in seen:
            seen.add(key)
            out.append(r)
    return out


def resample(rows, grain):
    """Reindex a time series to a COMPLETE period spine (grain=month|week|day) from the first to
    the last period, inserting a row for every MISSING period with None measures (never
    interpolated). Run this BEFORE running_total / moving_* / diff on a sparse or filtered date
    spine — otherwise those window ops silently skip the gaps and lie. The single most-flagged
    correctness guard in the tool surveys (asfreq/reindex)."""
    if grain not in ("month", "week", "day"):
        raise SEM.ModelError("resample grain must be month|week|day")
    _need_col(rows, "period", "period")
    present = {str(r.get("period"))[:10]: r for r in rows if r.get("period") is not None}
    if not present:
        return rows
    other_cols = [c for c in rows[0] if c != "period"]
    keys = sorted(present)
    p = keys[0]
    last = keys[-1]
    out, guard = [], 0
    while True:
        norm = _period_start(p, grain) if grain == "month" else str(p)[:10]
        if norm in present:
            out.append(present[norm])
        else:
            out.append({"period": norm, **{c: None for c in other_cols}})
        if norm >= last:
            break
        p = _step_period(p, grain, 1)
        guard += 1
        if guard > 5000:                         # bounded (loop-library discipline)
            break
    return out


# ================================================================== re-query family (widening window)

def _period_start(period, grain):
    if grain == "month":
        return f"{str(period)[:7]}-01"
    return str(period)[:10]


def _period_end(period, grain):
    p = str(period)
    if grain == "month":
        y, m = int(p[:4]), int(p[5:7])
        ny, nm = (y + 1, 1) if m == 12 else (y, m + 1)
        return (dt.date(ny, nm, 1) - dt.timedelta(days=1)).isoformat()
    if grain == "week":
        return (dt.date.fromisoformat(p[:10]) + dt.timedelta(days=6)).isoformat()
    return p[:10]


def _step_period(period, grain, n):
    p = str(period)
    if grain == "month":
        idx = int(p[:4]) * 12 + (int(p[5:7]) - 1) + n
        return f"{idx // 12:04d}-{idx % 12 + 1:02d}-01"
    if grain == "week":
        return (dt.date.fromisoformat(p[:10]) + dt.timedelta(weeks=n)).isoformat()
    return (dt.date.fromisoformat(p[:10]) + dt.timedelta(days=n)).isoformat()


def _require_series(res, op):
    grain = res.get("grain")
    if not grain:
        raise SEM.ModelError(f"{op} needs a time grain (run the query with grain=month|week|day)")
    if res.get("group_by"):
        raise SEM.ModelError(f"{op} works on a SINGLE time series — drop group_by (per-group "
                             "widening windows need one query per group)")
    return grain


def _scalar(result, col):
    rows = result.get("rows") or []
    return rows[0].get(col) if rows else None


def ytd(res, run_query, of):
    """CORRECT cumulative year-to-date for ANY measure (incl. distinct counts / ratios): re-runs
    the governed query over [Jan 1 .. each period end] and reads `of`. Never sums displayed values
    (which would over-count a distinct customer count). Adds `<of>_ytd`."""
    rows = res.get("rows") or []
    grain = _require_series(res, "ytd")
    _need_col(rows, "period", "period")
    _need_col(rows, of, "of")
    q = dict(res.get("query") or {})
    ordered = sorted(rows, key=lambda r: str(r.get("period") or ""))
    out = []
    for r in ordered:
        p = r["period"]
        wq = {**q, "grain": None, "group_by": None,
              "date_from": f"{str(p)[:4]}-01-01", "date_to": _period_end(p, grain)}
        out.append({**r, f"{of}_ytd": _scalar(run_query(wq), of)})
    return out


def rolling(res, run_query, of, window=3):
    """CORRECT trailing-window value for ANY measure (incl. distinct counts): re-runs the governed
    query over the trailing `window` periods and reads `of` (trailing-N distinct customers, T12M
    revenue, ...). Adds `<of>_roll<window>`."""
    rows = res.get("rows") or []
    grain = _require_series(res, "rolling")
    window = max(2, min(int(window or 3), 36))
    _need_col(rows, "period", "period")
    _need_col(rows, of, "of")
    q = dict(res.get("query") or {})
    ordered = sorted(rows, key=lambda r: str(r.get("period") or ""))
    out = []
    for r in ordered:
        p = r["period"]
        start = _period_start(_step_period(p, grain, -(window - 1)), grain)
        wq = {**q, "grain": None, "group_by": None,
              "date_from": start, "date_to": _period_end(p, grain)}
        out.append({**r, f"{of}_roll{window}": _scalar(run_query(wq), of)})
    return out


def forecast(res, run_query, of, periods=3, method="linear", season=12):
    """Append `periods` future rows with a `<of>_forecast` value. method='linear' extends the
    least-squares trend; 'seasonal_naive' repeats the value from `season` periods ago. Transparent
    and deterministic (no ML lib). Needs a time grain (to label future periods)."""
    rows = res.get("rows") or []
    grain = _require_series(res, "forecast")
    periods = max(1, min(int(periods or 3), 24))
    _need_col(rows, "period", "period")
    _need_col(rows, of, "of")
    ordered = sorted(rows, key=lambda r: str(r.get("period") or ""))
    hist = [r.get(of) for r in ordered]
    out = [{**r, f"{of}_forecast": None} for r in ordered]
    if out:                                        # bridge: last actual seeds the line
        out[-1][f"{of}_forecast"] = ordered[-1].get(of)
    last_p = ordered[-1]["period"] if ordered else None
    if method == "seasonal_naive":
        season = max(1, int(season or 12))
        for i in range(1, periods + 1):
            src = len(hist) - season + (i - 1)
            val = hist[src] if 0 <= src < len(hist) else None
            out.append({"period": _step_period(last_p, grain, i), f"{of}_forecast": val})
    else:
        slope, intercept, _ = _ols(list(range(len(hist))), hist)
        for i in range(1, periods + 1):
            val = (slope * (len(hist) - 1 + i) + intercept) if slope is not None else None
            out.append({"period": _step_period(last_p, grain, i), f"{of}_forecast": val})
    return out


# ================================================================== yoy (re-query)

def _shift_year(iso, delta):
    y, rest = iso[:4], iso[4:]
    shifted = f"{int(y) + delta}{rest}"
    if shifted.endswith("-02-29"):                 # leap-day clamp
        shifted = shifted[:-2] + "28"
    return shifted


def yoy_compare(res, run_query):
    """Same-period-last-year compare: re-run the source GOVERNED query shifted -1 year and join
    on (period shifted +1y, *group dims). Adds `<m>_ly`, `<m>_delta` ($ change) and `<m>_yoy_pct`
    (0-100) per measure — sort by `<m>_delta` asc for 'who dropped the most'. Unmatched periods
    keep None — a partial-vs-full compare is never faked."""
    q = dict(res.get("query") or {})
    if not (q.get("date_from") and q.get("date_to")):
        raise SEM.ModelError("yoy needs an explicit date_from/date_to on the source query")
    ly_q = {**q, "date_from": _shift_year(q["date_from"], -1),
            "date_to": _shift_year(q["date_to"], -1)}
    ly_rows = run_query(ly_q)["rows"]
    gb = list(res.get("group_by") or [])
    period_col = "period" if res.get("grain") else None
    keys = ([period_col] if period_col else []) + gb
    measures = [m for m in (res.get("measures") or []) if res["rows"] and m in res["rows"][0]]

    def _key(row, shift_period):
        parts = []
        for k in keys:
            v = str(row.get(k) or "")
            if k == period_col and shift_period and len(v) >= 4:
                v = _shift_year(v, +1)
            parts.append(v)
        return tuple(parts)

    ly_map = {}
    for r in ly_rows:
        ly_map[_key(r, True)] = r
    out = []
    for r in res["rows"]:
        prev = ly_map.get(_key(r, False), {})
        row = dict(r)
        for m in measures:
            pv = prev.get(m)
            row[f"{m}_ly"] = pv
            cur = r.get(m)
            row[f"{m}_delta"] = (cur - pv) if _numeric(pv) and _numeric(cur) else None
            row[f"{m}_yoy_pct"] = (100.0 * (cur - pv) / abs(pv)
                                   if _numeric(pv) and pv and _numeric(cur) else None)
        out.append(row)
    return out, ly_q


def _yoy(res, run_query):
    out, _ = yoy_compare(res, run_query)
    return out


# ================================================================== registry + replay

def _op(fn, params=(), required=(), needs_query=False):
    return {"fn": fn, "params": set(params), "required": set(required), "needs_query": needs_query}


OPS = {
    # ordering / rank / top-N
    "sort": _op(sort_rows, {"by", "direction"}, {"by"}),
    "head": _op(head, {"n"}),
    "bottom_n": _op(bottom_n, {"by", "n"}, {"by"}),
    "rank": _op(rank_rows, {"by", "direction", "out"}, {"by"}),
    "rank_pct": _op(rank_pct, {"by", "direction", "out"}, {"by"}),
    "ntile": _op(ntile, {"by", "tiles", "direction", "out"}, {"by"}),
    "top_n": _op(top_n, {"by", "n", "other", "other_label"}, {"by"}),
    "add_total": _op(add_total, {"label"}),
    # part-to-whole / cumulative
    "share_of_total": _op(share_of_total, {"of", "out"}, {"of"}),
    "cum_share": _op(cum_share, {"of", "out"}, {"of"}),
    # running / moving windows
    "running_total": _op(running_total, {"of", "out"}, {"of"}),
    "running_avg": _op(running_avg, {"of", "out"}, {"of"}),
    "running_max": _op(lambda rows, of, out=None: running_extreme(rows, of, "max", out),
                       {"of", "out"}, {"of"}),
    "running_min": _op(lambda rows, of, out=None: running_extreme(rows, of, "min", out),
                       {"of", "out"}, {"of"}),
    "moving_average": _op(moving_average, {"of", "window", "out"}, {"of"}),
    "moving_sum": _op(moving_sum, {"of", "window", "out"}, {"of"}),
    "moving_median": _op(moving_median, {"of", "window", "out"}, {"of"}),
    "rolling_std": _op(rolling_std, {"of", "window", "out"}, {"of"}),
    "running_count": _op(running_count, {"out"}),
    # period-over-period (row-wise)
    "diff": _op(diff, {"of", "out"}, {"of"}),
    "pct_change": _op(pct_change, {"of", "out"}, {"of"}),
    "lag": _op(lag, {"of", "k", "out"}, {"of"}),
    "lead": _op(lead, {"of", "k", "out"}, {"of"}),
    "diff_from_first": _op(diff_from_first, {"of", "out"}, {"of"}),
    "index_to_100": _op(index_to_100, {"of", "out"}, {"of"}),
    "percent_of_max": _op(percent_of_max, {"of", "out"}, {"of"}),
    "compare": _op(compare, {"a", "b", "how", "out"}, {"a", "b"}),
    # distribution / stats
    "bin": _op(bin_values, {"of", "bins"}, {"of"}),
    "describe": _op(describe, {"of"}, {"of"}),
    "zscore": _op(zscore, {"of", "out"}, {"of"}),
    "outliers": _op(outliers, {"of", "method", "k", "out"}, {"of"}),
    "winsorize": _op(winsorize, {"of", "p", "out"}, {"of"}),
    "clip": _op(clip, {"of", "lo", "hi", "out"}, {"of"}),
    "normalize": _op(normalize, {"of", "out"}, {"of"}),
    "correlate": _op(correlate, {"x", "y"}, {"x", "y"}),
    "weighted_average": _op(weighted_average, {"of", "weight"}, {"of", "weight"}),
    "safe_ratio": _op(safe_ratio, {"numerator", "denominator", "out"}, {"numerator", "denominator"}),
    "product": _op(product, {"a", "b", "out"}, {"a", "b"}),
    # business templates
    "abc_classify": _op(abc_classify, {"of", "a", "b"}, {"of"}),
    "concentration": _op(concentration, {"of"}, {"of"}),
    "contribution_to_change": _op(contribution_to_change, {"of"}, {"of"}),
    "rfm": _op(rfm, {"recency", "frequency", "monetary", "tiles"},
               {"recency", "frequency", "monetary"}),
    "funnel_rates": _op(funnel_rates, {"of"}, {"of"}),
    # modeling (transparent)
    "trend_line": _op(trend_line, {"of", "out"}, {"of"}),
    "regression": _op(regression, {"x", "y"}, {"x", "y"}),
    "cagr": _op(cagr, {"of"}, {"of"}),
    "growth_rate": _op(growth_rate, {"of"}, {"of"}),
    # reference annotations (added columns)
    "reference_line": _op(reference_line, {"of", "stat", "out"}, {"of"}),
    "reference_band": _op(reference_band, {"of", "method", "k"}, {"of"}),
    "target_line": _op(target_line, {"value", "out"}, {"value"}),
    "xmr_limits": _op(xmr_limits, {"of"}, {"of"}),
    # reshape (pure)
    "pivot": _op(pivot, {"index", "column", "value"}, {"index", "column", "value"}),
    "unpivot": _op(unpivot, {"keep", "columns", "var_name", "value_name"}, {"keep", "columns"}),
    "filter_rows": _op(filter_rows, {"col", "cmp", "value"}, {"col", "cmp", "value"}),
    "dedupe": _op(dedupe, {"by"}),
    "resample": _op(resample, {"grain"}, {"grain"}),
    # re-query family (widening / trailing windows, correct for non-additive)
    "yoy": _op(_yoy, needs_query=True),
    "ytd": _op(ytd, {"of"}, {"of"}, needs_query=True),
    "rolling": _op(rolling, {"of", "window"}, {"of"}, needs_query=True),
    "forecast": _op(forecast, {"of", "periods", "method", "season"}, {"of"}, needs_query=True),
}


def apply(res, ops, run_query=None):
    """Apply an op CHAIN to a result dict; returns (rows, applied_ops). Ops are validated against
    the registry (whitelisted names + params only). `run_query(query_dict) -> result` powers the
    re-query family (yoy/ytd/rolling/forecast); both the live tool and the saved-view replay inject
    their own."""
    if not isinstance(ops, list) or not ops:
        raise SEM.ModelError("transforms must be a non-empty list of {op, ...} objects")
    rows = list(res.get("rows") or [])
    applied = []
    for spec in ops:
        if not isinstance(spec, dict) or "op" not in spec:
            raise SEM.ModelError(f"each transform needs an 'op' key: {spec!r}")
        name = spec["op"]
        entry = OPS.get(name)
        if not entry:
            raise SEM.ModelError(f"unknown transform {name!r} (transforms: {sorted(OPS)})")
        params = {k: v for k, v in spec.items() if k != "op"}
        bad = set(params) - entry["params"]
        if bad:
            raise SEM.ModelError(f"{name}: unknown params {sorted(bad)} "
                                 f"(allowed: {sorted(entry['params'])})")
        missing = entry["required"] - set(params)
        if missing:
            raise SEM.ModelError(f"{name}: missing required params {sorted(missing)}")
        if entry["needs_query"]:
            if run_query is None:
                raise SEM.ModelError(f"{name} is unavailable here (no query runner)")
            # every re-query op takes (res_with_current_rows, run_query, **params)
            rows = entry["fn"]({**res, "rows": rows}, run_query, **params)
        else:
            rows = entry["fn"](rows, **params)
        applied.append({"op": name, **params})
    return rows, applied