Source code for mmm_framework.eda.remediation

"""
Turn outlier flags into concrete, applyable treatments.

The mapping encodes MMM-specific semantics:

* A media **isolated_spike** is treated as a recording error — winsorize it.
  Leaving it in corrupts the channel's max-normalization and flattens its
  saturation curve.
* A **kpi_shock** is treated as a real demand event — add an event dummy
  control so the model explains it instead of blaming/crediting media.
  Winsorizing the KPI would bias media coefficients.
* A **heavy_tail_member** cluster is a noise property: dummy out the largest
  shocks and note that the model's likelihood is Normal (a Student-t
  likelihood is not currently supported), so the residual tail risk should be
  recorded as an assumption.
* A **level_shift** is a structural changepoint: switch the trend to
  ``piecewise`` (supported by the model) or add a step dummy — never
  winsorize it away.
"""

from __future__ import annotations

import numpy as np
import pandas as pd

from .config import OutlierConfig
from .loading import EDAPanel
from .results import OutlierFlag, RemediationAction

STUDENT_T_ADVISORY = (
    "the model's likelihood is Normal (Student-t is not supported), so heavy-"
    "tailed noise cannot be absorbed by the likelihood — record this as a "
    "modeling assumption (category: data)"
)


def _dummy_name(flag: OutlierFlag) -> str:
    date_part = flag.period.replace("-", "_")
    var_part = "".join(ch if ch.isalnum() else "_" for ch in flag.variable)
    return f"outlier_{var_part}_{date_part}"


def _winsorize_cap(panel: EDAPanel, flag: OutlierFlag) -> float:
    """Cap value: max(robust expected, p99 of the series excluding the point)."""
    series = panel.series(flag.variable, flag.dims or None).astype(float)
    others = series[series.index != pd.Timestamp(flag.period)]
    p99 = (
        float(np.nanpercentile(others.to_numpy(), 99)) if len(others) else flag.expected
    )
    return float(max(flag.expected, p99))


[docs] def recommend_treatments( panel: EDAPanel, flags: list[OutlierFlag], config: OutlierConfig | None = None, ) -> list[RemediationAction]: """Build remediation recommendations for a set of flags.""" cfg = config or OutlierConfig() actions: list[RemediationAction] = [] spikes = [f for f in flags if f.kind == "isolated_spike"] drops = [f for f in flags if f.kind == "isolated_drop"] shocks = [f for f in flags if f.kind == "kpi_shock"] heavy = [f for f in flags if f.kind == "heavy_tail_member"] shifts = [f for f in flags if f.kind == "level_shift"] for f in spikes: cap = _winsorize_cap(panel, f) damage = (panel.df_wide[f.variable].max() / cap) if cap > 0 else float("nan") actions.append( RemediationAction( action_id=f"winsorize:{f.flag_id}", flag_ids=[f.flag_id], strategy="winsorize", params={"cap_value": cap}, rationale=( f"{f.variable} on {f.period} is {f.value:.4g} vs an expected " f"~{f.expected:.4g} — an isolated spike consistent with a " f"data-entry error. It alone sets the channel's normalization " f"scale (~{damage:.1f}x the capped max), which would flatten " f"its saturation curve. Cap it at {cap:.4g}; alternative: an " f"event dummy if the spend was real." ), ) ) for f in drops: impute_value = float(max(f.expected, 0.0)) actions.append( RemediationAction( action_id=f"impute:{f.flag_id}", flag_ids=[f.flag_id], strategy="impute", params={"value": impute_value}, rationale=( f"{f.variable} on {f.period} is recorded as {f.value:.4g} " f"while neighboring weeks spend ~{f.expected:.4g} — the " "pattern of a missed data load on an always-on channel. " f"Impute the local baseline ({impute_value:.4g}) after " "confirming with the user the channel wasn't actually dark " "that week." ), ) ) for f in shocks: name = _dummy_name(f) actions.append( RemediationAction( action_id=f"dummy:{f.flag_id}", flag_ids=[f.flag_id], strategy="dummy", params={"dummy_name": name, "periods": [f.period]}, rationale=( f"{f.variable} on {f.period} is {f.value:.4g} vs expected " f"~{f.expected:.4g} — likely a real demand event (promo/PR). " f"Add event dummy control `{name}` so the model explains it " "instead of crediting media. Only winsorize if the user " "confirms it's a data error." ), spec_change={"add_control": name}, ) ) if heavy: top = sorted(heavy, key=lambda f: -abs(f.value - f.expected))[ : cfg.heavy_tail_top_k ] for f in top: name = _dummy_name(f) actions.append( RemediationAction( action_id=f"dummy:{f.flag_id}", flag_ids=[f.flag_id], strategy="dummy", params={"dummy_name": name, "periods": [f.period]}, rationale=( f"{f.variable} on {f.period} is one of the largest shocks " "in a heavy-tailed residual pattern — dummy it out so it " "doesn't distort coefficient estimates." ), spec_change={"add_control": name}, ) ) var = heavy[0].variable actions.append( RemediationAction( action_id=f"note:heavy_tails:{var}", flag_ids=[f.flag_id for f in heavy], strategy="note", params={}, rationale=( f"`{var}` shows heavy-tailed residuals " f"({len(heavy)} shock(s) beyond the dummied top " f"{min(len(top), cfg.heavy_tail_top_k)}); " + STUDENT_T_ADVISORY + "." ), ) ) for f in shifts: actions.append( RemediationAction( action_id=f"trend:{f.flag_id}", flag_ids=[f.flag_id], strategy="note", params={"break_period": f.period}, rationale=( f"`{f.variable}` has a sustained level shift around {f.period} " "— a structural changepoint, not a point outlier. Do NOT " "winsorize it. Set the trend to `piecewise` (or add a step-" "dummy control starting at the break) so the baseline can " "absorb it; otherwise channels whose spend moves at the break " "will absorb the misfit." ), spec_change={"setting_path": "trend.type", "value": "piecewise"}, ) ) return actions
# --------------------------------------------------------------------------- # application (pure — caller handles file I/O) # ---------------------------------------------------------------------------
[docs] def apply_treatments( df_long: pd.DataFrame, actions: list[RemediationAction], flags: list[OutlierFlag], *, date_col: str = "Period", kpi: str | None = None, ) -> pd.DataFrame: """Apply winsorize / dummy / exclude actions to an MFF long frame. Returns a NEW frame; the input is not mutated. ``flags`` provides the (variable, period, dims) targeting for each action's ``flag_ids``. """ flag_by_id = {f.flag_id: f for f in flags} out = df_long.copy() dates = pd.to_datetime(out[date_col]) def _row_mask(flag: OutlierFlag) -> pd.Series: mask = (out["VariableName"] == flag.variable) & ( dates == pd.Timestamp(flag.period) ) for dim, val in (flag.dims or {}).items(): if dim in out.columns: mask &= out[dim].astype(str) == val return mask new_rows: list[pd.DataFrame] = [] excluded_periods: set[pd.Timestamp] = set() for action in actions: if action.strategy in ("winsorize", "impute"): new_value = float( action.params["cap_value"] if action.strategy == "winsorize" else action.params["value"] ) for fid in action.flag_ids: flag = flag_by_id.get(fid) if flag is None: continue out.loc[_row_mask(flag), "VariableValue"] = new_value elif action.strategy == "dummy": name = action.params["dummy_name"] periods = {pd.Timestamp(p) for p in action.params.get("periods", [])} # Build the dummy on the grain of the KPI (or the flagged variable): # one row per existing (period, dims) row of that variable. flag = next( (flag_by_id[fid] for fid in action.flag_ids if fid in flag_by_id), None, ) grain_var = kpi or (flag.variable if flag else None) if grain_var is None: continue template = df_long[df_long["VariableName"] == grain_var].copy() template["VariableName"] = name template["VariableValue"] = ( pd.to_datetime(template[date_col]).isin(periods).astype(float) ) new_rows.append(template) elif action.strategy == "exclude_periods": excluded_periods.update( pd.Timestamp(p) for p in action.params.get("periods", []) ) # "note" actions change the spec / record assumptions — no data edit. if new_rows: out = pd.concat([out, *new_rows], ignore_index=True) if excluded_periods: out = out[~pd.to_datetime(out[date_col]).isin(excluded_periods)].reset_index( drop=True ) return out
__all__ = ["recommend_treatments", "apply_treatments", "STUDENT_T_ADVISORY"]