Api calm

Summary

select_regions decides, across features, which of the splits find_regions found actually explain the model. It returns a CalmSequence — the chain [GAM, calm1, ...] produced by greedy forward selection: each step applies the split with the largest explained-variance gain, and stops when no remaining split adds at least min_r2_gain of Var(f̂).

Each snapshot is a CALM (Conditional Additive Local Model): the global read plus the partitions accepted so far, with its surrogate R² stamped on it. Like Partition, both are values, not stored state (design contract R12): they serialize with to_dict() and re-attach with bind(effect).

---

Usage

pdp = effector.PDP(data=X, model=predict)

parts = pdp.find_regions(features="heterogeneous")   # candidates, per feature
chain = pdp.select_regions(partitions=parts)          # which ones earn their keep

chain.show()          # the explained-variance ledger, as tables
chain.gam_r2          # R² of the pure-GAM read
chain.regional_r2     # R² after the accepted splits
chain.skipped         # rejected splits, with reasons

calm = chain.final    # the selected snapshot — the report's §2
calm.importances()    # per feature, weighted mean over subregions
calm.plot_triage()    # the importance × heterogeneity plane, at this snapshot

partitions= is optional: select_regions() runs the search itself, with the same features / finder / candidate_conditioning_features arguments as find_regions. The method itself is documented per engine in Global effect; the walkthrough lives in the select_regions guide.

Derivative-scale methods cannot play

select_regions raises ValueError on DerPDP: its curves live on the derivative scale, where summing does not approximate f̂.

API

effector.global_effect.GlobalEffectBase.select_regions(partitions=None, *, features='heterogeneous', finder='best', candidate_conditioning_features='all', min_r2_gain=0.01)

Greedily select which partitions earn their complexity — the CALM chain.

chain = pdp.select_regions()      # search + select in one call
chain.show()                      # GAM R2, each accepted split, the rejected
chain.final                       # the last CALM — the regional analysis
chain[0]                          # the GAM snapshot

find_regions proposes one candidate Partition per feature; this verb decides across features which of them actually explain the model: starting from the GAM (all features global), each round applies the split with the largest explained-variance gain — the surrogate R² against f̂, measured on top of the splits already applied — and stops when no remaining split adds at least min_r2_gain. Every accepted round is a snapshot (CALM) of increased complexity; the whole chain is the report's ledger.

A query, not a mutation (R12)

The result is a value — nothing is stored on the effect.

One prediction pass

Beyond fit, the only model touch is one f̂(X) pass for the variance denominator (cached on the effect); the search, the scoring, and every snapshot's summaries are model-free.

Parameters:

Name	Type	Description	Default
partitions	Optional[dict]	pre-computed candidates — {feature_index_or_name: Partition}, exactly what find_regions(features=...) returns. None (default) runs the search here first.	None
features	Union[list, str]	which features to search when partitions is None — a list, "all", or "heterogeneous" (default; heter_score at/above the median).	'heterogeneous'
finder		region finder, as in find_regions.	'best'
candidate_conditioning_features		features allowed to define splits ("all" or a list of indices/names).	'all'
min_r2_gain	float	smallest explained-variance marginal (fraction of Var(f̂), default 0.01 = 1 pt) a split must add — on top of the splits already applied — to earn a snapshot.	0.01

Returns:

Type	Description
	a CalmSequence — [GAM, calm1, ...], R² non-decreasing along
	it, with the rejected splits in .skipped
	("redundant"/"below_threshold").

Raises:

Type	Description
ValueError	the method is derivative-scale (no output-scale surrogate) or Var(f̂) == 0.

Source code in effector/global_effect.py

def select_regions(
    self,
    partitions: Optional[dict] = None,
    *,
    features: Union[list, str] = "heterogeneous",
    finder="best",
    candidate_conditioning_features="all",
    min_r2_gain: float = 0.01,
):
    """Greedily select which partitions earn their complexity — the CALM chain.

    ```python
    chain = pdp.select_regions()      # search + select in one call
    chain.show()                      # GAM R2, each accepted split, the rejected
    chain.final                       # the last CALM — the regional analysis
    chain[0]                          # the GAM snapshot
    ```

    `find_regions` proposes one candidate `Partition` per feature;
    this verb decides *across* features which of them actually explain
    the model: starting from the GAM (all features global), each round
    applies the split with the largest explained-variance gain — the
    surrogate R² against `f̂`, measured on top of the splits already
    applied — and stops when no remaining split adds at least
    `min_r2_gain`. Every accepted round is a snapshot (`CALM`) of
    increased complexity; the whole chain is the report's ledger.

    !!! note "A query, not a mutation (R12)"
        The result is a value — nothing is stored on the effect.

    !!! note "One prediction pass"
        Beyond `fit`, the only model touch is one `f̂(X)` pass for the
        variance denominator (cached on the effect); the search, the
        scoring, and every snapshot's summaries are model-free.

    Args:
        partitions: pre-computed candidates — `{feature_index_or_name:
            Partition}`, exactly what `find_regions(features=...)`
            returns. `None` (default) runs the search here first.
        features: which features to search when `partitions` is `None` —
            a list, `"all"`, or `"heterogeneous"` (default; heter_score
            at/above the median).
        finder: region finder, as in `find_regions`.
        candidate_conditioning_features: features allowed to define
            splits (`"all"` or a list of indices/names).
        min_r2_gain: smallest explained-variance marginal (fraction of
            `Var(f̂)`, default 0.01 = 1 pt) a split must add — on top of
            the splits already applied — to earn a snapshot.

    Returns:
        a `CalmSequence` — `[GAM, calm1, ...]`, R² non-decreasing along
        it, with the rejected splits in `.skipped`
        (`"redundant"`/`"below_threshold"`).

    Raises:
        ValueError: the method is derivative-scale (no output-scale
            surrogate) or `Var(f̂) == 0`.
    """
    from effector import explained_variance as _ev  # lazy: one-way dep guard

    if partitions is None:
        partitions = self.find_regions(
            features=features,
            finder=finder,
            candidate_conditioning_features=candidate_conditioning_features,
        )
    parts = {}
    for key, p in partitions.items():
        f = self._resolve_feature(key)
        parts[f] = p if p._effect is not None else p.bind(self)

    supported = []
    for f in range(self.dim):
        try:
            self._check_feature_type_supported(f)
            supported.append(f)
        except ValueError:
            continue
    return _ev.select(self, parts, supported, min_gain=min_r2_gain)

effector.calm.CalmSequence(calms, *, skipped=None, min_gain=0.01)

The chain select_regions returns: [GAM, calm1, ...] plus the splits it rejected.

List-like (chain[0], chain[-1], len, iteration) with the chain invariants: R² is non-decreasing along it and each stage's delta_r2 is the sequential marginal, so the gains sum exactly to regional_r2 - gam_r2.

Methods:

Name	Description
show	Print the decision sequence as the report's ledger tables:
bind	Bind every snapshot (mask verification included). Returns self.
to_dict	Serialize the chain — a superset of the report's explained-variance
from_dict	Rebuild an unbound chain from to_dict() output.

Attributes:

Name	Type	Description
gam		The chain's first snapshot — the pure GAM.
final		The chain's last snapshot — what the report renders as the regional analysis.
gam_r2
regional_r2
stages		The accepted decisions, in order — one per non-GAM snapshot.

Source code in effector/calm.py

def __init__(self, calms, *, skipped=None, min_gain=0.01):
    if not calms:
        raise ValueError("CalmSequence needs at least the GAM snapshot.")
    self.calms = list(calms)
    self.skipped = list(skipped) if skipped else []
    self.min_gain = float(min_gain)

gam property

The chain's first snapshot — the pure GAM.

final property

The chain's last snapshot — what the report renders as the regional analysis.

gam_r2 property

regional_r2 property

stages property

The accepted decisions, in order — one per non-GAM snapshot.

show(ascii=False)

Print the decision sequence as the report's ledger tables: EXPLAINED VARIANCE (the GAM, each accepted split, the FINAL R²), then REJECTED SPLITS with the reason each one was refused.

Parameters:

Name	Type	Description	Default
ascii		draw with plain ASCII instead of box-drawing characters, for terminals and logs that mangle unicode.	False

Source code in effector/calm.py

def show(self, ascii=False):
    """Print the decision sequence as the report's ledger tables:
    EXPLAINED VARIANCE (the GAM, each accepted split, the FINAL R²),
    then REJECTED SPLITS with the reason each one was refused.

    Args:
        ascii: draw with plain ASCII instead of box-drawing characters,
            for terminals and logs that mangle unicode.
    """
    from effector.report import _print_explained_variance

    sy = self.final.scale_y["std"] if self.final.scale_y else 1.0
    _print_explained_variance(
        {
            "gam_r2": self.gam_r2,
            "regional_r2": self.regional_r2,
            "min_gain": self.min_gain,
            "stages": [st for st in self.stages if st is not None],
            "skipped": self.skipped,
        },
        ascii=ascii,
        sy=sy,
    )

bind(effect)

Bind every snapshot (mask verification included). Returns self.

Source code in effector/calm.py

def bind(self, effect):
    """Bind every snapshot (mask verification included). Returns `self`."""
    for c in self.calms:
        c.bind(effect)
    return self

to_dict()

Serialize the chain — a superset of the report's explained-variance payload: the flat decision-sequence keys (gam_r2, regional_r2, min_gain, stages, skipped) plus the full calms list.

Source code in effector/calm.py

def to_dict(self):
    """Serialize the chain — a superset of the report's explained-variance
    payload: the flat decision-sequence keys (`gam_r2`, `regional_r2`,
    `min_gain`, `stages`, `skipped`) plus the full `calms` list."""
    return {
        "schema_version": _SCHEMA_VERSION,
        "gam_r2": self.gam_r2,
        "regional_r2": self.regional_r2,
        "min_gain": self.min_gain,
        "stages": self.stages,
        "skipped": self.skipped,
        "calms": [c.to_dict() for c in self.calms],
    }

from_dict(d) classmethod

Rebuild an unbound chain from to_dict() output.

Source code in effector/calm.py

@classmethod
def from_dict(cls, d):
    """Rebuild an **unbound** chain from `to_dict()` output."""
    if d.get("schema_version") != _SCHEMA_VERSION:
        raise ValueError(
            f"Unsupported CalmSequence dict: expected schema_version "
            f"{_SCHEMA_VERSION}, got {d.get('schema_version')!r}."
        )
    return cls(
        [CALM.from_dict(c) for c in d["calms"]],
        skipped=d["skipped"],
        min_gain=d["min_gain"],
    )

effector.calm.CALM(*, index, r2, partitions, stage=None, feature_names, target_name, importances, heter_scores, baseline, scale_y=None)

One snapshot: global effects everywhere except the accepted split features.

index == 0 is the pure GAM (no partitions). Every later CALM adds one accepted split on top of the previous one; stage records that decision (feature, sequential delta_r2, running cum_r2, heterogeneity before/after the split).

Stamped scalars

importance(f)/heter_score(f) are computed at construction — weighted mean over subregions for split features (weights = instances per leaf), the global value otherwise — and stamped, so an unbound CALM still ranks and plots. When bound, the same calls can recompute live via the effect's masked verbs and must agree.

Methods:

Name	Description
from_effect	Stamp a snapshot off a fitted effect — used by select_regions.
importance	This snapshot's importance of feature (output units).
importances	(D,) vector of this snapshot's importances; NaN = unsupported type.
heter_score	This snapshot's heterogeneity of feature (output units).
heter_scores	(D,) vector of this snapshot's heterogeneities; NaN = unsupported type.
plot_triage	This snapshot's triage plane, from the stamped scalars (works unbound).
show	Print the snapshot: R², the decision that created it, its partitions.
bind	Re-attach a fitted effect: binds every held partition (recomputing
to_dict	Serialize to a plain JSON-able dict — rules and stamped stats only,
from_dict	Rebuild an unbound CALM from to_dict() output — tables and

Attributes:

Name	Type	Description
is_gam
features		The split features of this snapshot, ascending.

Source code in effector/calm.py

def __init__(
    self,
    *,
    index,
    r2,
    partitions,
    stage=None,
    feature_names,
    target_name,
    importances,
    heter_scores,
    baseline,
    scale_y=None,
):
    self.index = int(index)
    self.r2 = float(r2)
    self.partitions = dict(partitions)  # {feature_int: Partition}
    self.stage = stage
    self.feature_names = list(feature_names)
    self.target_name = target_name
    # the schema's y scaling ({"mean", "std"} or None): the stamped
    # scalars speak model-output units; display surfaces bridge by its std
    self.scale_y = scale_y
    self._importances = {int(k): float(v) for k, v in importances.items()}
    self._heter_scores = {int(k): float(v) for k, v in heter_scores.items()}
    # global (imp, het) start points of the split features' triage arrows
    self._baseline = {int(k): tuple(v) for k, v in baseline.items()}
    self._effect = None

is_gam property

features property

The split features of this snapshot, ascending.

from_effect(effect, partitions, *, r2, index=0, stage=None) classmethod

Stamp a snapshot off a fitted effect — used by select_regions.

Parameters:

Name	Type	Description	Default
effect		a fitted global effect (bound partitions apply to its data).	required
partitions		{feature_int: Partition} — the accepted splits.	required
r2		surrogate R² of this snapshot against f̂ (see explained_variance.surrogate_r2).	required
index		position in the chain; 0 is the GAM.	0
stage		the decision that created this snapshot (None for the GAM).	None

Source code in effector/calm.py

@classmethod
def from_effect(cls, effect, partitions, *, r2, index=0, stage=None):
    """Stamp a snapshot off a fitted effect — used by `select_regions`.

    Args:
        effect: a fitted global effect (bound partitions apply to its data).
        partitions: `{feature_int: Partition}` — the accepted splits.
        r2: surrogate R² of this snapshot against `f̂` (see
            `explained_variance.surrogate_r2`).
        index: position in the chain; 0 is the GAM.
        stage: the decision that created this snapshot (`None` for the GAM).
    """
    partitions = {int(f): p for f, p in partitions.items()}
    imps, hets, base = {}, {}, {}
    for f in range(effect.dim):
        try:
            effect._check_feature_type_supported(f)
        except ValueError:
            imps[f] = np.nan
            hets[f] = np.nan
            continue
        if f in partitions:
            part = partitions[f]
            w = np.array([leaf.nof_instances for leaf in part.leaves], dtype=float)
            imps[f] = float(
                np.average(
                    [
                        effect.importance(f, mask=part.mask(leaf.idx))
                        for leaf in part.leaves
                    ],
                    weights=w,
                )
            )
            hets[f] = float(
                np.average(
                    [float(leaf.heterogeneity) for leaf in part.leaves],
                    weights=w,
                )
            )
            base[f] = (
                float(effect.importance(f)),
                float(effect.heter_score(f)),
            )
        else:
            imps[f] = float(effect.importance(f))
            hets[f] = float(effect.heter_score(f))
    from effector.report import _scale_payload

    calm = cls(
        index=index,
        r2=r2,
        partitions=partitions,
        stage=stage,
        feature_names=list(effect.feature_names),
        target_name=effect.target_name,
        importances=imps,
        heter_scores=hets,
        baseline=base,
        scale_y=_scale_payload(effect.scale_y),
    )
    calm._effect = effect
    return calm

importance(feature, *, per_region=False)

This snapshot's importance of feature (output units).

Split feature -> instance-weighted mean over its subregions; else the global value. With per_region=True (bound CALM, split feature only) returns the per-leaf list instead.

Source code in effector/calm.py

def importance(self, feature, *, per_region=False):
    """This snapshot's importance of `feature` (output units).

    Split feature -> instance-weighted mean over its subregions; else the
    global value. With `per_region=True` (bound CALM, split feature only)
    returns the per-leaf list instead.
    """
    f = self._resolve(feature)
    if per_region:
        part = self._part_for(f)
        effect = self._require_effect()
        return [
            float(effect.importance(f, mask=part.mask(leaf.idx)))
            for leaf in part.leaves
        ]
    return self._importances[f]

importances()

(D,) vector of this snapshot's importances; NaN = unsupported type.

Source code in effector/calm.py

def importances(self):
    """`(D,)` vector of this snapshot's importances; NaN = unsupported type."""
    return np.array([self._importances[f] for f in range(len(self.feature_names))])

heter_score(feature, *, per_region=False)

This snapshot's heterogeneity of feature (output units).

Split feature -> instance-weighted mean over its subregions; else the global value. With per_region=True (split feature only) returns the stamped per-leaf list — available unbound.

Source code in effector/calm.py

def heter_score(self, feature, *, per_region=False):
    """This snapshot's heterogeneity of `feature` (output units).

    Split feature -> instance-weighted mean over its subregions; else the
    global value. With `per_region=True` (split feature only) returns the
    stamped per-leaf list — available unbound.
    """
    f = self._resolve(feature)
    if per_region:
        part = self._part_for(f)
        return [float(leaf.heterogeneity) for leaf in part.leaves]
    return self._heter_scores[f]

heter_scores()

(D,) vector of this snapshot's heterogeneities; NaN = unsupported type.

Source code in effector/calm.py

def heter_scores(self):
    """`(D,)` vector of this snapshot's heterogeneities; NaN = unsupported type."""
    return np.array([self._heter_scores[f] for f in range(len(self.feature_names))])

plot_triage(threshold=False, title=None, show_plot=True)

This snapshot's triage plane, from the stamped scalars (works unbound).

One point per feature at the CALM's own (importance, heterogeneity); each split feature gets one arrow from its global point to its weighted-mean point — the movement the accepted split bought.

Parameters:

Name	Type	Description	Default
threshold		heterogeneity line — a float draws it (given in heter_score units; rescaled with the points when the schema declared a scale_y), False (default) nothing.	False
title		figure title.	None
show_plot		if True, show and return None; else (fig, ax).	True

Source code in effector/calm.py

def plot_triage(self, threshold=False, title=None, show_plot=True):
    """This snapshot's triage plane, from the stamped scalars (works unbound).

    One point per feature at the CALM's own (importance, heterogeneity);
    each split feature gets one arrow from its global point to its
    weighted-mean point — the movement the accepted split bought.

    Args:
        threshold: heterogeneity line — a float draws it (given in
            `heter_score` units; rescaled with the points when the schema
            declared a `scale_y`), `False` (default) nothing.
        title: figure title.
        show_plot: if `True`, show and return `None`; else `(fig, ax)`.
    """
    from effector.visualization import triage_scatter

    # stamped scalars speak model-output units; the axes claim the
    # target's units, so bridge by the schema's y-std (spreads: std only)
    sy = self.scale_y["std"] if self.scale_y else 1.0
    pts = [
        (
            self.feature_names[f],
            self._importances[f] * sy,
            self._heter_scores[f] * sy,
        )
        for f in range(len(self.feature_names))
        if np.isfinite(self._importances[f])
    ]
    arrows = {
        self.feature_names[f]: (
            tuple(v * sy for v in self._baseline[f]),
            (self._importances[f] * sy, self._heter_scores[f] * sy),
        )
        for f in self.features
        if f in self._baseline
    }
    if threshold is not None and threshold is not False:
        threshold = float(threshold) * sy
    default = (
        "Feature triage" if self.is_gam else f"Feature triage — CALM {self.index}"
    )
    return triage_scatter(
        pts,
        arrows=arrows,
        threshold=threshold,
        unit=f" ({self.target_name} units)",
        title=default if title is None else title,
        show_plot=show_plot,
    )

show()

Print the snapshot: R², the decision that created it, its partitions.

Source code in effector/calm.py

def show(self):
    """Print the snapshot: R², the decision that created it, its partitions."""
    head = "GAM" if self.is_gam else f"CALM {self.index}"
    print(f"{head}: R2 = {self.r2:.1%}")
    if self.stage is not None:
        print(
            f"  + split {self.stage['name']} on {self.stage['on']} "
            f"({self.stage['n_regions']} regions, "
            f"{self.stage['delta_r2']:+.1%})"
        )
    for f in self.features:
        self.partitions[f].show()

bind(effect)

Re-attach a fitted effect: binds every held partition (recomputing and verifying masks) and re-enables the live verbs.

Returns:

Type	Description
	self, live.

Source code in effector/calm.py

def bind(self, effect):
    """Re-attach a fitted effect: binds every held partition (recomputing
    and verifying masks) and re-enables the live verbs.

    Returns:
        `self`, live.
    """
    for part in self.partitions.values():
        part.bind(effect)
    self._effect = effect
    return self

to_dict()

Serialize to a plain JSON-able dict — rules and stamped stats only, never masks/data/effect. from_dict(...) + bind(effect) restore the rest.

Source code in effector/calm.py

def to_dict(self):
    """Serialize to a plain JSON-able dict — rules and stamped stats only,
    never masks/data/effect. `from_dict(...)` + `bind(effect)` restore
    the rest."""
    return {
        "schema_version": _SCHEMA_VERSION,
        "index": self.index,
        "r2": self.r2,
        "stage": self.stage,
        "feature_names": self.feature_names,
        "target_name": self.target_name,
        "partitions": {str(f): p.to_dict() for f, p in self.partitions.items()},
        "importances": {str(f): v for f, v in self._importances.items()},
        "heter_scores": {str(f): v for f, v in self._heter_scores.items()},
        "baseline": {str(f): list(v) for f, v in self._baseline.items()},
        "scale_y": self.scale_y,
    }

from_dict(d) classmethod

Rebuild an unbound CALM from to_dict() output — tables and triage work; partition eval/plot and per-region importance need bind(effect).

Source code in effector/calm.py

@classmethod
def from_dict(cls, d):
    """Rebuild an **unbound** CALM from `to_dict()` output — tables and
    triage work; partition `eval`/`plot` and per-region importance need
    `bind(effect)`."""
    if d.get("schema_version") != _SCHEMA_VERSION:
        raise ValueError(
            f"Unsupported CALM dict: expected schema_version "
            f"{_SCHEMA_VERSION}, got {d.get('schema_version')!r}."
        )
    return cls(
        index=d["index"],
        r2=d["r2"],
        partitions={
            int(f): Partition.from_dict(p) for f, p in d["partitions"].items()
        },
        stage=d["stage"],
        feature_names=d["feature_names"],
        target_name=d["target_name"],
        importances=d["importances"],
        heter_scores=d["heter_scores"],
        baseline=d["baseline"],
        scale_y=d.get("scale_y"),
    )