phiusiil/preprocess/stats.py

"""Everything learned at fit time, in one immutable record.

The governing rule of this package: **transform() may not compute any statistic.**

Operationally, inside ``transform`` there is no call to ``.mean()``, ``.median()``,
``.mode()``, ``.std()``, ``.skew()``, ``.quantile()``, ``.value_counts()``,
``.groupby(...).agg(...)``, ``fit()`` or ``fit_transform()`` on the input frame. Anything
depending on more than the current row is a fitted statistic: computed once in ``fit()``,
stored here, serialized, and read back at serving time.

This is not a style preference. The original recomputed every statistic from whatever
batch it was handed, and standardized the training and validation splits by their own
separate means -- so the scaling that produced its headline numbers does not exist at
serving time and cannot be reconstructed, because there is no batch to take a mean over
when a user pastes one URL. Making the statistics data rather than control flow is the
precondition for the service existing at all.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Literal

FillMethod = Literal["mean", "median"]
NbDropReason = Literal["empty_contingency_cell", "zero_std"]
FallbackSource = Literal["numeric_mean", "numeric_median", "categorical_mode"]


@dataclass(frozen=True)
class NumericFill:
    """Imputation value for one numeric column.

    ``skew`` is recorded even though ``transform`` never reads it, so a reader can audit
    which branch was taken without re-running training.
    """

    column: str
    method: FillMethod  # median iff abs(skew) > 3
    value: float
    skew: float


@dataclass(frozen=True)
class CategoricalFillStep:
    """One step of the ordered categorical cascade.

    ``groupby_cols`` is a snapshot of the group key *at this step*, persisted explicitly
    rather than reconstructed as "the first k-1 filled columns plus HasObfuscation". If
    the cascade is ever edited, the artifact still describes what actually ran.
    """

    column: str
    groupby_cols: tuple[str, ...]
    modes: dict[tuple[float, ...], float]
    global_mode: float


@dataclass(frozen=True)
class ClipBound:
    column: str
    lower: float  # 1st percentile of the training split
    upper: float  # 99th percentile of the training split


@dataclass(frozen=True)
class ScalerParams:
    """An affine map, stored as plain numbers.

    Deliberately not a pickled StandardScaler. A pickled estimator carries a
    ``fit_transform`` method, and the entire class of defect this package exists to
    eliminate is someone calling it at serving time. Numbers cannot be re-fitted.
    """

    columns: tuple[str, ...]  # exactly the 30 numeric columns, ordered
    mean_: tuple[float, ...]
    scale_: tuple[float, ...]  # zeros replaced by 1.0 so transform cannot divide by zero

    def __post_init__(self) -> None:
        if not (len(self.columns) == len(self.mean_) == len(self.scale_)):
            raise ValueError(
                f"scaler arity mismatch: {len(self.columns)} columns, "
                f"{len(self.mean_)} means, {len(self.scale_)} scales"
            )
        if any(s == 0.0 for s in self.scale_):
            raise ValueError("scale_ contains a zero; constant columns must be mapped to 1.0")


@dataclass(frozen=True)
class HtmlFallback:
    """The value a page-derived feature takes when the fetch failed or the feature was
    demoted.

    Derived, not separately computed: each entry is exactly the value this column's own
    fill rule already produces. It is lifted into its own record purely so the interface
    can *display* the fallback and where it came from, rather than silently applying it.
    A mismatch between this table and the fills it was derived from is a bug, and a test
    asserts they agree.
    """

    column: str
    value: float
    source: FallbackSource


@dataclass(frozen=True)
class FittedStats:
    """The complete serialized state of a fitted preprocessor."""

    # --- URL character model -------------------------------------------------
    char_prob: dict[str, float]

    # --- TLD legitimacy ------------------------------------------------------
    tld_prob_mean: dict[str, float]
    tld_prob_global_fill: float
    tld_skew: float
    tld_fill_method: FillMethod  # median iff abs(tld_skew) > 1 -- note: 1, not 3

    # --- numeric imputation --------------------------------------------------
    numeric_fill: tuple[NumericFill, ...]

    # --- categorical cascade. ORDER IS SEMANTIC ------------------------------
    categorical_cascade: tuple[CategoricalFillStep, ...]

    # --- outlier clipping ----------------------------------------------------
    clip_bounds: tuple[ClipBound, ...]

    # --- scaling -------------------------------------------------------------
    scaler: ScalerParams

    # --- Naive-Bayes column drop, fitted here for locality -------------------
    nb_drop: tuple[str, ...]
    nb_drop_reasons: dict[str, NbDropReason]

    # --- page-feature fallbacks ---------------------------------------------
    html_fallbacks: tuple[HtmlFallback, ...]

    # --- frozen column lists, so nothing closes over a module global ---------
    feature_order: tuple[str, ...]
    numerical_columns: tuple[str, ...]
    continuous_columns: tuple[str, ...]
    discrete_columns: tuple[str, ...]
    categorical_columns_filtered: tuple[str, ...]

    # --- provenance ----------------------------------------------------------
    n_train_rows: int = 0
    demoted_features: tuple[str, ...] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        if len(self.feature_order) != 49:
            raise ValueError(f"feature_order must have 49 entries, got {len(self.feature_order)}")
        if len(self.numerical_columns) != 30:
            raise ValueError(
                f"numerical_columns must have 30 entries, got {len(self.numerical_columns)}"
            )
        if len(self.categorical_columns_filtered) != 19:
            raise ValueError(
                "categorical_columns_filtered must have 19 entries, got "
                f"{len(self.categorical_columns_filtered)}"
            )
        if len(self.categorical_cascade) != 18:
            raise ValueError(
                f"categorical_cascade must have 18 steps, got {len(self.categorical_cascade)}"
            )
        if tuple(self.scaler.columns) != tuple(self.numerical_columns):
            raise ValueError(
                "scaler.columns must be exactly numerical_columns, in the same order. "
                "A mismatch here means the matrix was scaled on a different column set "
                "than the one recorded, which silently changes every distance."
            )

    @property
    def numeric_fill_map(self) -> dict[str, float]:
        return {f.column: f.value for f in self.numeric_fill}

    @property
    def html_fallback_map(self) -> dict[str, float]:
        return {f.column: f.value for f in self.html_fallbacks}