from __future__ import annotations
import numpy as np
import polars as pl
from .metrics import compute_metrics
from .rule_evaluation import apply_rules
[docs]
def validate_rules_cv(
X: pl.DataFrame,
y: pl.Series,
rules: list[str],
n_folds: int = 5,
cv_metrics: list[str] | None = None,
weight_column: str | None = None,
shuffle: bool = True,
random_state: int | None = None,
) -> pl.DataFrame:
"""Evaluate rule stability across K folds.
Validates already-generated rules on K folds without re-generating them.
For each fold the rules are evaluated on the validation split, and the
mean, standard deviation, and minimum of each requested metric across folds
are returned. Rules with a high ``{metric}_cv_std`` or a low
``{metric}_cv_min`` are likely over-fitted to the training data.
.. warning::
**The folds are not truly held out, and the reported statistics are
optimistically biased.** This function does not generate rules; it takes
rules that were already produced from the *entire* dataset and then
re-scores them on subsets of that same dataset. Every validation fold
was therefore part of the data used to choose the rules' features and
thresholds. ``{metric}_cv_mean``, ``{metric}_cv_std`` and
``{metric}_cv_min`` must **not** be reported as out-of-sample or
generalisation estimates. See Notes for an unbiased protocol.
Parameters
----------
X : pl.DataFrame
Feature DataFrame. Must contain all columns referenced in ``rules``.
y : pl.Series
Target series (boolean or binary).
rules : list[str]
Rule expressions to validate. Typically obtained from
:func:`~iguanas.rule_evaluation.apply_filter_and_deduplicate_rules`
or similar.
n_folds : int, default=5
Number of CV folds. The data is split into ``n_folds`` contiguous
blocks (after optional shuffling).
cv_metrics : list[str] | None, default=None
Metric names to compute CV statistics for. When ``None``, defaults to
``["precision", "recall", "f1"]``.
weight_column : str | None, default=None
Name of a column in ``X`` to use as sample weights when computing
metrics. If ``None``, all samples are weighted equally.
shuffle : bool, default=True
Whether to shuffle the row indices before splitting into folds.
random_state : int | None, default=None
Random seed for reproducibility when ``shuffle=True``.
Returns
-------
pl.DataFrame
One row per rule with columns:
- ``rule``
- ``{metric}_cv_mean`` — mean of the metric across folds
- ``{metric}_cv_std`` — standard deviation across folds
- ``{metric}_cv_min`` — worst-fold value (lowest)
Sorted by ``rule`` name.
Examples
--------
>>> import polars as pl
>>> X = pl.DataFrame({"age": [25, 30, 35, 40, 45, 50, 55, 60, 65, 70]})
>>> y = pl.Series([0, 0, 0, 0, 0, 1, 1, 1, 1, 1])
>>> rules = ['(X["age"] >= 50)']
>>> validate_rules_cv(X, y, rules, n_folds=2, random_state=0)
Notes
-----
**Optimism bias.** The intended usage is: generate rules on ``X``/``y``,
then call this function on the same ``X``/``y``. Rule selection has
therefore already seen every fold, which leaks information into each
"validation" split. The consequences are:
- ``{metric}_cv_mean`` is inflated relative to true held-out performance.
- ``{metric}_cv_std`` is deflated and ``{metric}_cv_min`` is inflated, so
the numbers understate how badly a rule can degrade on genuinely unseen
data. They are a *lower bound* on overfitting, not a measure of it.
- A rule that looks stable here can still fail out of sample; a rule that
looks unstable here is almost certainly unstable.
Use these statistics as a *relative* screen to rank and discard fragile
rules within a candidate set — never as an estimate of deployment
performance, and never as a headline result.
For an unbiased estimate, use a nested protocol in which rule generation
happens strictly inside the training split of each outer fold:
1. Split the data into outer folds (or a single held-out test set).
2. For each outer fold, run the full pipeline — weight transformations,
grid search, filtering, deduplication — on the training portion only.
3. Evaluate the resulting rules on the outer fold, which no step of the
pipeline has seen.
4. Aggregate across outer folds.
Only step 3 yields a defensible generalisation estimate.
See Also
--------
apply_filter_and_deduplicate_rules : Complete evaluation pipeline that
produces the ``rules`` input for this function.
"""
if cv_metrics is None:
cv_metrics = ["precision", "recall", "f1"]
n = len(y)
indices = list(range(n))
if shuffle:
rng = np.random.default_rng(random_state)
rng.shuffle(indices)
fold_size = n // n_folds
fold_frames: list[pl.DataFrame] = []
for fold in range(n_folds):
val_start = fold * fold_size
val_end = val_start + fold_size if fold < n_folds - 1 else n
val_idx = indices[val_start:val_end]
X_val = X[val_idx]
y_val = y[val_idx]
weights_val = X_val[weight_column] if weight_column is not None else None
R_val = apply_rules(X_val, rules)
m = compute_metrics(R_val, y_val, weights=weights_val)
available = [c for c in cv_metrics if c in m.columns]
fold_frames.append(m.select(["rule"] + available))
stacked = pl.concat(fold_frames)
available_metrics = [c for c in cv_metrics if c in stacked.columns]
agg_exprs = []
for metric in available_metrics:
agg_exprs.extend(
[
pl.col(metric).mean().alias(f"{metric}_cv_mean"),
pl.col(metric).std().alias(f"{metric}_cv_std"),
pl.col(metric).min().alias(f"{metric}_cv_min"),
]
)
return stacked.group_by("rule").agg(agg_exprs).sort("rule")
[docs]
def identify_unstable_rules(
cv_result: pl.DataFrame,
metric: str = "f1",
max_std: float = 0.05,
min_mean: float | None = None,
) -> pl.DataFrame:
"""Return rules whose cross-fold metric is unstable or consistently poor.
Filters the output of :func:`validate_rules_cv` to surface rules that are
likely over-fitted (high variance across folds) or simply weak (low mean
metric).
.. warning::
Inherits the optimism bias of :func:`validate_rules_cv`: the folds were
already seen during rule generation, so ``{metric}_cv_std`` is deflated
and ``{metric}_cv_mean`` inflated. This function is therefore a
one-sided screen — rules it flags are genuinely unstable, but rules it
does **not** flag are not thereby shown to generalise. Absence from the
returned set is not evidence of stability.
Parameters
----------
cv_result : pl.DataFrame
Output of :func:`validate_rules_cv`. Must contain columns
``{metric}_cv_std`` and (if ``min_mean`` is set) ``{metric}_cv_mean``.
metric : str, default="f1"
Metric prefix to inspect (must match one used in :func:`validate_rules_cv`).
max_std : float, default=0.05
Rules whose ``{metric}_cv_std`` **exceeds** this threshold are flagged
as unstable.
min_mean : float | None, default=None
If provided, also flag rules whose ``{metric}_cv_mean`` is **below**
this threshold (consistently weak rules).
Returns
-------
pl.DataFrame
Subset of ``cv_result`` containing only flagged rules, sorted by
``{metric}_cv_std`` descending (most unstable first).
Raises
------
ValueError
If required columns are absent from ``cv_result``.
Examples
--------
>>> cv = validate_rules_cv(X, y, rules, n_folds=5)
>>> identify_unstable_rules(cv, metric="f1", max_std=0.05, min_mean=0.3)
"""
std_col = f"{metric}_cv_std"
mean_col = f"{metric}_cv_mean"
if std_col not in cv_result.columns:
raise ValueError(
f"Column '{std_col}' not found. "
f"Run validate_rules_cv with cv_metrics=['{metric}', ...]."
)
if min_mean is not None and mean_col not in cv_result.columns:
raise ValueError(
f"Column '{mean_col}' not found. "
f"Run validate_rules_cv with cv_metrics=['{metric}', ...]."
)
mask = pl.col(std_col) > max_std
if min_mean is not None:
mask = mask | (pl.col(mean_col) < min_mean)
return cv_result.filter(mask).sort(std_col, descending=True)
