"""Production model monitoring: watch for drift, retrain, and swap -- with provenance and a DOE hook.
A :class:`Monitor` wraps a fitted model + its estimator + a reference (training) sample. Feed it each
production batch via :meth:`check` (drift report only) or :meth:`update` (check, and on drift retrain a
fresh model -- recording a new provenance header -- and swap it in). Every action is appended to a history
for audit. :meth:`suggest_samples` ties in ``mixle.doe`` so a drift signal (or any model objective) can
drive *where to collect new data* -- space-filling by default, or active-learning against a model objective.
"""
from __future__ import annotations
import time
from typing import Any
from mixle.inference.production.drift import DriftReport, detect_drift, validate_drift_thresholds
from mixle.inference.production.provenance import Header, fit_with_provenance
[docs]
class Monitor:
"""Stateful monitor for one deployed model: drift detection + retrain-and-swap + DOE-driven sampling."""
def __init__(
self,
model: Any,
estimator: Any,
reference: Any,
*,
psi_threshold: float = 0.25,
ks_threshold: float = 0.2,
loglik_shift_threshold: float = -0.5,
) -> None:
# The same validation ``detect_drift`` runs, at construction rather than one call later.
# A local copy of it here was weaker -- it accepted a positive ``loglik_shift_threshold``
# (which flags drift on identical data) and a negative ``psi_threshold`` (likewise, PSI
# being non-negative), so a Monitor built with either constructed cleanly and then raised
# from every ``check()`` and ``update()`` it was ever given, naming a constructor argument
# from a method that does not take one (R06-F05).
validate_drift_thresholds(
psi_threshold=psi_threshold,
ks_threshold=ks_threshold,
loglik_shift_threshold=loglik_shift_threshold,
)
self.model = model
self.estimator = estimator
self.reference = self._materialize(reference, "reference", model)
self.thresholds = {
"psi_threshold": float(psi_threshold),
"ks_threshold": float(ks_threshold),
"loglik_shift_threshold": float(loglik_shift_threshold),
}
self.history: list[dict] = []
self.header: Header | None = getattr(model, "header", None)
@staticmethod
def _materialize(data: Any, name: str, model: Any = None) -> list:
"""Through the fit verbs' front door: a drift batch is read the same way a fit reads it.
This carried its own copy of "records() if present, else iterate", so a DataFrame handed to
check() or update() became its column labels and a bare string became its characters -- and
update() then retrained on that (Q05-F01).
"""
from mixle.inference.estimation import _tabular_records
batch = _tabular_records(data, f"Monitor({name})", target=model)
if not batch:
raise ValueError(f"{name} data must contain at least one observation")
return batch
[docs]
def check(self, current: Any) -> DriftReport:
"""Drift report of ``current`` (production) data against the reference under the current model."""
batch = self._materialize(current, "current", self.model)
return detect_drift(self.model, self.reference, batch, **self.thresholds)
[docs]
def update(self, current: Any, *, retrain: bool = True, combine_reference: bool = True, **fit_kw: Any) -> dict:
"""Check drift on ``current`` and, if drift is flagged and ``retrain``, fit a fresh model (with a
new provenance header) and swap it in. Returns ``{report, action, model, header}`` and appends to
:attr:`history`. ``combine_reference`` retrains on reference + current (else current only)."""
if not isinstance(retrain, bool) or not isinstance(combine_reference, bool):
raise TypeError("retrain and combine_reference must be booleans")
batch = self._materialize(current, "current", self.model)
report = detect_drift(self.model, self.reference, batch, **self.thresholds)
action = "none"
header = self.header
if report.drift and retrain:
train = (self.reference + batch) if combine_reference else batch
new_model, header = fit_with_provenance(train, self.estimator, **fit_kw)
self.model = new_model
self.reference = train
self.header = header
action = "retrained"
entry = {
"time": time.time(),
"n_current": len(batch),
"drift": report.drift,
"score": report.score,
"action": action,
}
self.history.append(entry)
return {
"report": report,
"action": action,
"model": self.model,
"header": header,
"processed_count": len(batch),
}
[docs]
def suggest_samples(
self, bounds: Any, n: int = 10, *, method: str = "lhs", objective: Any = None, seed: int | None = None
) -> Any:
"""Use ``mixle.doe`` to propose where to collect new data (e.g. after drift, or to meet an
objective). ``method='lhs'``/``'sobol'`` gives a space-filling batch over ``bounds`` (list of
``(lo, hi)``); pass an ``objective(x)->float`` to switch to active learning (ALC/ALM) that targets
where the model is most informative."""
from mixle import doe as _doe
if objective is not None:
return _doe.active_learning_design(objective, bounds, max_evals=n, seed=seed)
if method == "sobol":
return _doe.sobol_design(bounds, n, seed)
return _doe.latin_hypercube(bounds, n, seed)