"""Grouped cross-validation for constrained models."""
from __future__ import annotations
from collections.abc import Mapping, Sequence
import logging
from typing import Any, Union
import numpy as np
import pandas as pd
from p2p.models.cross_validation import ModelCV
from p2p.models.folds import (
DEFAULT_FOLD_METHOD,
DEFAULT_FOLD_SEED,
FOLD_METHODS,
FoldAssignment,
resolve_fold_assignment,
)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
if not logger.hasHandlers():
handler = logging.StreamHandler()
formatter = logging.Formatter(
"[%(levelname)s] %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
[docs]
def build_group_folds(
groups,
*,
n_splits: int,
assignments: FoldAssignment | Mapping[str, int] | None = None,
method: str = DEFAULT_FOLD_METHOD,
seed: int | None = DEFAULT_FOLD_SEED,
) -> tuple[tuple[np.ndarray, np.ndarray], ...]:
"""Build grouped train/validation indices, one pair per fold.
``groups`` labels each observation with its question. Without explicit
``assignments`` the folds are generated by ``method`` (see
``p2p.models.folds``); an explicit ``FoldAssignment`` or plain mapping of
question -> fold index is used as given, after validation.
"""
return resolve_fold_assignment(
groups,
n_splits=n_splits,
assignments=assignments,
method=method,
seed=seed,
).indices(groups)
[docs]
class ConstrainedModelCV(ModelCV):
"""Cross-validation where constraints follow observation groups."""
def __init__(
self,
*args,
fold_groups: Mapping[str, Sequence[str]] | None = None,
fold_method: str = DEFAULT_FOLD_METHOD,
**kwargs,
):
"""Cross-validate a constrained model.
Args:
fold_groups: The question partition used to build folds. A
question's option rows must stay together, so grouping is
required, not optional. Stating it separately from
``constraint_groups`` lets the mass constraints be switched
off or restated without changing how folds are cut. When
omitted it falls back to ``model_kwargs["constraint_groups"]``;
if neither is given, fitting raises rather than silently
cutting folds through the middle of a question.
fold_method: How folds are generated when none are passed to
``fit``: ``"hashed"`` (default) or ``"sklearn"``. The seed is
``random_state``. See ``p2p.models.folds``.
*args: Forwarded to ``ModelCV``.
**kwargs: Forwarded to ``ModelCV``.
"""
super().__init__(*args, **kwargs)
if fold_groups is not None and not isinstance(fold_groups, Mapping):
raise TypeError("fold_groups must be a mapping or None")
if fold_method not in FOLD_METHODS or fold_method == "replay":
raise ValueError(
"fold_method must be 'hashed' or 'sklearn'; replay is done by "
"passing fold_assignments to fit"
)
self.fold_groups = fold_groups
self.fold_method = fold_method
self.fold_assignment_: FoldAssignment | None = None
@staticmethod
def _same_partition(
first: Mapping[str, Sequence[str]] | None,
second: Mapping[str, Sequence[str]] | None,
) -> bool:
"""Report whether two group mappings describe the same partition."""
if first is None or second is None:
return False
return {
str(name): tuple(members)
for name, members in first.items()
} == {
str(name): tuple(members)
for name, members in second.items()
}
def _resolve_observation_groups(
self,
A: pd.DataFrame,
constraint_groups: Mapping[str, list[str]],
) -> np.ndarray:
"""Return group labels aligned with the observation index."""
observation_to_group = {}
for group, members in constraint_groups.items():
for observation_name in members:
observation_to_group[observation_name] = group
try:
return np.asarray(
[
observation_to_group[str(index)]
for index in A.index
],
dtype=object,
)
except KeyError as error:
raise ValueError(
f"observation {error.args[0]!r} in A.index "
"was not found in any constraint group"
) from None
[docs]
def fit(
self,
A: pd.DataFrame,
y: Union[pd.Series, np.ndarray],
*,
fold_assignments: FoldAssignment | Mapping[str, int] | None = None,
):
"""Cross-validate over question-grouped folds.
Folds come from ``fold_assignments`` when given (replay), otherwise
from ``fold_method`` with ``random_state`` as the seed.
"""
self.A_ = A
self.y_ = y
constraint_groups = self.model_kwargs.get(
"constraint_groups"
)
fold_groups = (
self.fold_groups
if self.fold_groups is not None
else constraint_groups
)
if fold_groups is None:
raise ValueError(
"cross-validation folds must be grouped by question: a "
"question's option rows sum to one, so splitting them across "
"folds leaks the held-out answer. Pass fold_groups, or "
"constraint_groups for the folds to follow."
)
if constraint_groups is not None and not self._same_partition(
fold_groups,
constraint_groups,
):
raise ValueError(
"fold_groups and constraint_groups must describe the same "
"partition: both are the survey's questions. Constraints may "
"be switched off (constraint_groups=None) without affecting "
"how folds are cut, but the two must not disagree."
)
group_labels = self._resolve_observation_groups(
A,
fold_groups,
)
if len(group_labels) != len(A):
raise ValueError(
"resolved group labels do not match "
"the number of observations"
)
y_array = (
y.to_numpy()
if isinstance(y, pd.Series)
else np.asarray(y)
)
fold_assignment = resolve_fold_assignment(
group_labels,
n_splits=self.k,
assignments=fold_assignments,
method=self.fold_method,
seed=self.random_state,
)
folds = fold_assignment.indices(group_labels)
self.fold_assignment_ = fold_assignment
# Every fold fit's solver outcome is counted, so a run that fell back to
# another solver says so in its diagnostics.
fold_fits = 0
fold_fallbacks = 0
total_configs = len(self.alphas) * (
len(self.l1_ratios)
if self.l1_ratios is not None
else 1
)
config_count = 0
for alpha in self.alphas:
ratios = (
self.l1_ratios
if self.l1_ratios is not None
else [None]
)
for l1_ratio in ratios:
config_count += 1
if self.verbose:
logger.info(
"Evaluating config %s/%s: "
"alpha=%s, l1_ratio=%s",
config_count,
total_configs,
alpha,
l1_ratio,
)
errors = []
for train_indices, validation_indices in folds:
A_train = A.iloc[train_indices]
A_validation = A.iloc[
validation_indices
]
y_train = y_array[train_indices]
y_validation = y_array[
validation_indices
]
kwargs: dict[str, Any] = (
self.model_kwargs.copy()
)
kwargs["alpha"] = alpha
if constraint_groups is not None:
training_observations = set(
A.index[train_indices]
)
filtered_groups = {
group: observation_names
for group, observation_names
in constraint_groups.items()
if set(
observation_names
).issubset(
training_observations
)
}
grouped_observation_count = sum(
len(observation_names)
for observation_names
in filtered_groups.values()
)
if (
grouped_observation_count
!= len(train_indices)
):
raise ValueError(
"training fold constraint groups "
"do not cover every observation: "
f"got {grouped_observation_count}"
", expected "
f"{len(train_indices)}"
)
kwargs["constraint_groups"] = (
filtered_groups
)
if l1_ratio is not None:
kwargs["l1_ratio"] = l1_ratio
model = self.model_cls(**kwargs)
model.fit(A_train, y_train)
fold_fits += 1
fold_fallbacks += int(
bool(getattr(model, "solver_fallback_", False))
)
errors.append(
model.score(
A_validation,
y_validation,
)
)
average_error = float(np.mean(errors))
standard_deviation = float(
np.std(errors)
)
parameters = (alpha, l1_ratio)
self.cv_mean_mse_[parameters] = (
average_error
)
self.cv_std_mse_[parameters] = (
standard_deviation
)
if self.verbose:
logger.info(
"Mean MSE: %.4f ± %.4f",
average_error,
standard_deviation,
)
if average_error < self.best_score_:
self.best_score_ = average_error
self.best_params_ = parameters
if self.l1_ratios is None:
mean_mse = np.asarray(
[
self.cv_mean_mse_[(alpha, None)]
for alpha in self.alphas
]
)
std_mse = np.asarray(
[
self.cv_std_mse_[(alpha, None)]
for alpha in self.alphas
]
)
else:
mean_mse = np.asarray(
[
[
self.cv_mean_mse_[
(alpha, l1_ratio)
]
for alpha in self.alphas
]
for l1_ratio in self.l1_ratios
]
)
std_mse = np.asarray(
[
[
self.cv_std_mse_[
(alpha, l1_ratio)
]
for alpha in self.alphas
]
for l1_ratio in self.l1_ratios
]
)
resolved_assignments = {}
for fold, (_, validation_indices) in enumerate(
folds
):
for group in set(
group_labels[validation_indices]
):
resolved_assignments[str(group)] = fold
self.diagnostics_ = {
"alphas": self.alphas,
"l1_ratios": self.l1_ratios,
"mean_mse": mean_mse,
"std_mse": std_mse,
"best_alpha": self.best_params_[0],
"best_l1_ratio": self.best_params_[1],
"fold_assignments": resolved_assignments,
"fold_method": fold_assignment.method,
"fold_seed": fold_assignment.seed,
"fold_partition_digest": fold_assignment.partition_digest,
"cv_fits": fold_fits,
"cv_solver_fallbacks": fold_fallbacks,
}
return self
__all__ = [
"ConstrainedModelCV",
"build_group_folds",
]