"""Piecewise-linear fuzzy-set calibration."""
from __future__ import annotations
import warnings
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any
import numpy as np
import pandas as pd
from qca._constants import (
CALIBRATION_FULL_IN,
CALIBRATION_FULL_OUT,
CROSSOVER_POINT,
ZERO_EPSILON,
)
from qca.calibration._validators import (
validate_column_exists,
validate_columns_exist,
validate_membership_bounds,
validate_min_nonnull,
validate_numeric_column,
validate_quantile_triplet,
)
[docs]
@dataclass(frozen=True)
class PiecewiseAnchor:
"""Anchor values estimated for one calibration group."""
group_key: tuple[Any, ...]
p_low: float
p_mid: float
p_high: float
n_total: int
n_nonnull: int
is_valid: bool
def to_dict(self) -> dict[str, Any]:
return {
"group_key": self.group_key,
"p_low": self.p_low,
"p_mid": self.p_mid,
"p_high": self.p_high,
"n_total": self.n_total,
"n_nonnull": self.n_nonnull,
"is_valid": self.is_valid,
}
[docs]
@dataclass(frozen=True)
class PiecewiseCalibrationResult:
"""Result returned by :func:`calibrate_piecewise`."""
df: pd.DataFrame
anchors: list[PiecewiseAnchor]
out_col: str
n_calibrated: int
n_skipped: int
[docs]
def anchors_to_df(self) -> pd.DataFrame:
"""Return anchor metadata as a DataFrame."""
if not self.anchors:
return pd.DataFrame()
return pd.DataFrame([anchor.to_dict() for anchor in self.anchors])
def __repr__(self) -> str:
return (
"PiecewiseCalibrationResult("
f"n_calibrated={self.n_calibrated}, "
f"n_skipped={self.n_skipped}, "
f"out_col={self.out_col!r}, "
f"n_groups={len(self.anchors)}"
")"
)
[docs]
def piecewise_fuzzy_scalar(
x: float,
p_low: float,
p_mid: float,
p_high: float,
lower: float = CALIBRATION_FULL_OUT,
upper: float = CALIBRATION_FULL_IN,
) -> float:
"""Convert one numeric value into fuzzy membership with linear segments."""
validate_membership_bounds(lower, upper)
try:
x_value = float(x)
except (TypeError, ValueError):
return np.nan
if np.isnan(x_value):
return np.nan
w_low = max(p_mid - p_low, ZERO_EPSILON)
w_high = max(p_high - p_mid, ZERO_EPSILON)
if x_value <= p_low:
return lower
if x_value <= p_mid:
return lower + (x_value - p_low) / w_low * (CROSSOVER_POINT - lower)
if x_value <= p_high:
return CROSSOVER_POINT + (x_value - p_mid) / w_high * (upper - CROSSOVER_POINT)
return upper
[docs]
def piecewise_fuzzy_series(
series: pd.Series,
p_low: float,
p_mid: float,
p_high: float,
lower: float = CALIBRATION_FULL_OUT,
upper: float = CALIBRATION_FULL_IN,
) -> pd.Series:
"""Convert a Series into fuzzy membership with linear segments."""
validate_membership_bounds(lower, upper)
numeric = pd.to_numeric(series, errors="coerce").astype(float)
arr = numeric.to_numpy(dtype=float)
w_low = max(p_mid - p_low, ZERO_EPSILON)
w_high = max(p_high - p_mid, ZERO_EPSILON)
lower_segment = lower + (arr - p_low) / w_low * (CROSSOVER_POINT - lower)
upper_segment = CROSSOVER_POINT + (arr - p_mid) / w_high * (upper - CROSSOVER_POINT)
result = np.select(
[np.isnan(arr), arr <= p_low, arr <= p_mid, arr <= p_high],
[np.nan, lower, lower_segment, upper_segment],
default=upper,
)
return pd.Series(result, index=series.index, dtype=float)
def _compute_quantile(series: pd.Series, q: float) -> float:
clean = series.dropna()
if clean.empty:
return np.nan
return float(clean.quantile(q))
def _estimate_anchors(
series: pd.Series,
quantiles: tuple[float, float, float],
min_nonnull: int,
) -> tuple[float, float, float, bool]:
"""Estimate low/mid/high anchors for a numeric series."""
n_nonnull = int(series.notna().sum())
if n_nonnull < min_nonnull:
return np.nan, np.nan, np.nan, False
q_low, q_mid, q_high = quantiles
p_low = _compute_quantile(series, q_low)
p_mid = _compute_quantile(series, q_mid)
p_high = _compute_quantile(series, q_high)
if not all(np.isfinite(value) for value in (p_low, p_mid, p_high)):
return np.nan, np.nan, np.nan, False
if not p_low < p_mid < p_high:
warnings.warn(
"Estimated anchors do not satisfy p_low < p_mid < p_high; "
"this group will be skipped.",
UserWarning,
stacklevel=3,
)
return np.nan, np.nan, np.nan, False
return p_low, p_mid, p_high, True
[docs]
def calibrate_piecewise(
df: pd.DataFrame,
value_col: str,
out_col: str | None = None,
group_cols: Sequence[str] | None = None,
quantiles: tuple[float, float, float] = (0.10, 0.50, 0.90),
lower: float = CALIBRATION_FULL_OUT,
upper: float = CALIBRATION_FULL_IN,
min_nonnull: int = 5,
) -> PiecewiseCalibrationResult:
"""Calibrate a numeric column to fuzzy membership by quantile anchors."""
validate_column_exists(df, value_col)
if group_cols:
validate_columns_exist(df, list(group_cols))
validate_quantile_triplet(quantiles)
validate_membership_bounds(lower, upper)
validate_min_nonnull(min_nonnull)
out_col = out_col or f"{value_col}_fuzzy"
work = df.copy()
work[value_col] = validate_numeric_column(work, value_col)
anchors: list[PiecewiseAnchor] = []
fuzzy_values = pd.Series(np.nan, index=work.index, dtype=float)
if not group_cols:
p_low, p_mid, p_high, is_valid = _estimate_anchors(
work[value_col], quantiles, min_nonnull
)
anchors.append(
PiecewiseAnchor(
group_key=(),
p_low=p_low,
p_mid=p_mid,
p_high=p_high,
n_total=len(work),
n_nonnull=int(work[value_col].notna().sum()),
is_valid=is_valid,
)
)
if is_valid:
fuzzy_values = piecewise_fuzzy_series(
work[value_col], p_low, p_mid, p_high, lower, upper
)
else:
group_list = list(group_cols)
for keys, index in work.groupby(
group_list, dropna=False, sort=True
).groups.items():
group_key = keys if isinstance(keys, tuple) else (keys,)
group_series = work.loc[index, value_col]
p_low, p_mid, p_high, is_valid = _estimate_anchors(
group_series, quantiles, min_nonnull
)
anchors.append(
PiecewiseAnchor(
group_key=group_key,
p_low=p_low,
p_mid=p_mid,
p_high=p_high,
n_total=len(index),
n_nonnull=int(group_series.notna().sum()),
is_valid=is_valid,
)
)
if is_valid:
fuzzy_values.loc[index] = piecewise_fuzzy_series(
group_series, p_low, p_mid, p_high, lower, upper
)
work[out_col] = fuzzy_values
n_calibrated = int(fuzzy_values.notna().sum())
n_skipped = int(fuzzy_values.isna().sum())
if n_skipped > 0:
warnings.warn(
f"{n_skipped} cases were not calibrated and are NaN. ",
UserWarning,
stacklevel=2,
)
return PiecewiseCalibrationResult(
df=work,
anchors=anchors,
out_col=out_col,
n_calibrated=n_calibrated,
n_skipped=n_skipped,
)