Extending

Summary

The estimators resolve their teacher and their regional model by name from two registries. Registering a new name makes it available as teacher= or curve_fitter=. Below the estimators sit two functions that turn regions you already have into a model.

---

Usage

A new teacher: factory(task, **params) returns an object with fit(X, y) and forward(X) (values for regression, positive-class probability for classification). params is the user's teacher_params; random_state is added if the factory accepts it.

from calm_additive import CALMRegressor, register_teacher

register_teacher("my_teacher", lambda task, **params: MyTeacher(**params))
calm = CALMRegressor(teacher="my_teacher", teacher_params={"depth": 6}).fit(X, y)

A new regional model: implement MaskedFitter. X is the masked design, one column per (feature, region); mask marks which entries are active.

from calm_additive import MaskedFitter, register_curve_fitter

class MyFitter(MaskedFitter):
    def fit(self, X, y, mask): ...
    def predict(self, X, mask): ...
    def curves(self): ...          # optional: what the model learned

register_curve_fitter("mine", lambda ctx, **params: MyFitter(**params))
calm = CALMRegressor(curve_fitter="mine").fit(X, y)

curves() returns the fitted model as an AdditiveTerms: the intercept, one Curve per masked column, and what each column adds when it is inactive. With it, every view (print_summary, plot_curves, …) works for your fitter; without it the model still predicts.

from calm_additive import AdditiveTerms, Curve

def curves(self):
    columns = [Curve(knots, values, "linear") for knots, values in self.splines]
    return AdditiveTerms(self.intercept, columns, self.inactive_scores)

👉 A teacher needs no registration if you pass it directly: CALMRegressor(teacher=my_model).

API

calm_additive.engine.register_teacher(name, factory)

Register a teacher factory: factory(task, **params) → object with fit(X, y) and forward(X) (and optionally jac(X) for derivative methods), where params is the user's teacher_params (plus random_state if the factory accepts it).

Source code in calm_additive/engine.py

def register_teacher(name: str, factory: Callable[..., object]):
    """Register a teacher factory: `factory(task, **params)` → object with
    `fit(X, y)` and `forward(X)` (and optionally `jac(X)` for derivative
    methods), where `params` is the user's `teacher_params` (plus
    `random_state` if the factory accepts it)."""
    _TEACHERS[name] = factory

calm_additive.engine.register_curve_fitter(name, factory)

Register a fitter factory under a name (calm_additive does this for "ebm"): factory(ctx, **params) → a MaskedFitter, where params is the user's curve_fitter_params (plus random_state if the factory accepts it). "surrogate" is built in and reserved.

Source code in calm_additive/engine.py

def register_curve_fitter(name: str, factory: Callable[..., MaskedFitter]):
    """Register a fitter factory under a name (`calm_additive` does this
    for `"ebm"`): `factory(ctx, **params)` → a `MaskedFitter`, where `params`
    is the user's `curve_fitter_params` (plus `random_state` if the factory
    accepts it). `"surrogate"` is built in and reserved."""
    if name == "surrogate":
        raise ValueError("'surrogate' is the built-in free mode; pick another name")
    _CURVE_FITTERS[name] = factory

calm_additive.engine.MaskedFitter

The refit contract: an additive model over masked columns.

X is the masked design (one column per (feature, region), inactive entries zeroed), mask the parallel boolean activity matrix. Implement fit(X, y, mask) and predict(X, mask); predict must return (N,) scores (probabilities of the positive class for classification).

Implement curves() as well — the fitted model as an AdditiveTerms — and the model can be printed and plotted (print_summary, plot_curves, …); without it the model still predicts.

calm_additive.engine.Curve(x, y, kind='step', other=0.0, fn=None) dataclass

A one-dimensional function, the way a fitter stores it.

kind	x	y	value at v
"step"	(B-1,) ascending cut points	(B,)	y[i] on [x[i-1], x[i])
"linear"	(T,) ascending knots	(T,)	linear between knots, flat outside
"levels"	(L,) ascending levels	(L,)	y of the matching level, else other

fn, when given, is the exact evaluator and x/y only its drawable sampling (a curve read from an effector effect).

Methods:

Name	Description
shifted	The same curve moved up by c.

shifted(c)

The same curve moved up by c.

Source code in calm_additive/engine.py

def shifted(self, c: float) -> "Curve":
    """The same curve moved up by `c`."""
    fn = self.fn
    return replace(
        self,
        y=self.y + c,
        other=self.other + c,
        fn=None if fn is None else (lambda v: np.asarray(fn(v), dtype=float) + c),
    )

calm_additive.engine.AdditiveTerms(intercept, columns, inactive, link='identity') dataclass

What a MaskedFitter learned, column by column.

The fitter's score is intercept plus, per masked column, the column's curve at the row's value when the column is active and inactive[c] when it is not. link names the scale that sum lives on: "identity" (the prediction itself) or "logit" (log-odds of the positive class).

calm_additive.engine.fit_chain(chain, X=None, y=None, *, curve_fitter='surrogate', curve_fitter_params=None, snapshot='final', task='regression', target='teacher', random_state=None)

Materialize a chain you inspected: snapshot (default the final CALM) picks which decision-sequence prefix becomes the model.

Source code in calm_additive/engine.py

def fit_chain(
    chain: CalmSequence,
    X: Optional[np.ndarray] = None,
    y: Optional[np.ndarray] = None,
    *,
    curve_fitter="surrogate",
    curve_fitter_params: Optional[dict] = None,
    snapshot: Union[str, int] = "final",
    task: str = "regression",
    target: str = "teacher",
    random_state: Optional[int] = None,
) -> ExecutableCALM:
    """Materialize a chain you inspected: `snapshot` (default the final CALM)
    picks which decision-sequence prefix becomes the model."""
    calm = chain.final if snapshot == "final" else chain[int(snapshot)]
    effect = calm._require_effect()
    return fit_partitions(
        effect,
        dict(calm.partitions),
        task=task,
        curve_fitter=curve_fitter,
        curve_fitter_params=curve_fitter_params,
        X=X,
        y=y,
        target=target,
        random_state=random_state,
        chain=chain,
        calm=calm,
    )

calm_additive.engine.fit_partitions(effect, partitions, *, task='regression', curve_fitter='surrogate', curve_fitter_params=None, X=None, y=None, target='teacher', random_state=None, chain=None, calm=None)

Materialize {feature: Partition} into an executable model.

The lowest-level entry: both fit_chain and the sklearn twins land here. partitions must be bound to effect. For the surrogate mode the offset target is the teacher's predictions (target="teacher") or y (target="y", requires y). Refit modes require y (and use X — default effect.data — as the design source); curve_fitter_params and random_state reach a fitter given by name.

Source code in calm_additive/engine.py

def fit_partitions(
    effect,
    partitions: dict,
    *,
    task: str = "regression",
    curve_fitter="surrogate",
    curve_fitter_params: Optional[dict] = None,
    X: Optional[np.ndarray] = None,
    y: Optional[np.ndarray] = None,
    target: str = "teacher",
    random_state: Optional[int] = None,
    chain: Optional[CalmSequence] = None,
    calm: Optional[CALM] = None,
) -> ExecutableCALM:
    """Materialize `{feature: Partition}` into an executable model.

    The lowest-level entry: both `fit_chain` and the sklearn twins land here.
    `partitions` must be bound to `effect`. For the surrogate mode the offset
    target is the teacher's predictions (``target="teacher"``) or `y`
    (``target="y"``, requires `y`). Refit modes require `y` (and use
    `X` — default `effect.data` — as the design source);
    `curve_fitter_params` and `random_state` reach a fitter given by name.
    """
    if task not in ("regression", "classification"):
        raise ValueError(f"task must be 'regression' or 'classification'; got {task!r}")
    if target not in ("teacher", "y"):
        raise ValueError(f"target must be 'teacher' or 'y'; got {target!r}")
    partitions = {j: p for j, p in partitions.items() if len(p.leaves) > 1}

    if curve_fitter == "surrogate":
        if curve_fitter_params:
            raise ValueError("curve_fitter='surrogate' takes no curve_fitter_params")
        if target == "y":
            if y is None:
                raise ValueError("target='y' needs y")
            tvec = np.asarray(y, dtype=float).reshape(-1)
        else:
            if effect._y_pred is None:
                effect._y_pred = np.asarray(effect.model(effect.data))
            tvec = np.asarray(effect._y_pred, dtype=float).reshape(-1)
        features = [j for j in range(effect.dim)]
        surrogate = SurrogateModel().fit(effect, partitions, features, tvec)
        return ExecutableCALM(
            task, partitions=partitions, chain=chain, calm=calm, surrogate=surrogate
        )

    if y is None:
        raise ValueError(
            f"curve_fitter={curve_fitter!r} refits on labels — y is required"
        )
    X = effect.data if X is None else np.asarray(X, dtype=float)
    transform = MaskedTransform(partitions, effect.dim, effect.feature_names)
    Xt, mask = transform.transform(X)
    src_types = [
        effect.feature_types[j] for j, _, _ in transform.columns
    ]
    ctx = FitterContext(
        task=task,
        dim=Xt.shape[1],
        feature_names=tuple(transform.new_names),
        feature_types=tuple(src_types),
    )
    fitter_obj = _resolve_curve_fitter(
        curve_fitter, ctx, curve_fitter_params, random_state
    )
    fitter_obj.fit(Xt, np.asarray(y).reshape(-1), mask)
    return ExecutableCALM(
        task,
        partitions=partitions,
        chain=chain,
        calm=calm,
        transform=transform,
        curve_fitter=fitter_obj,
    )

calm_additive.engine.ExecutableCALM(task, *, partitions, chain=None, calm=None, surrogate=None, transform=None, curve_fitter=None)

A fitted, predict-capable CALM. Built by fit_chain/fit_partitions.

Carries its provenance: .chain (when built from one), .calm (the snapshot), .partitions, .transform + .curve_fitter (refit mode) or .surrogate (free mode). predict returns labels for classification (predict_score gives the raw score / positive-class probability).

Methods:

Name	Description
predict
predict_score
terms	The model as functions — one curve per (feature, region), each a

Attributes:

Name	Type	Description
interactions_info	dict
n_conditional_interactions	int	The paper's metric: Σ_i |S_i| over the split features — one per

Source code in calm_additive/engine.py

def __init__(
    self,
    task: str,
    *,
    partitions: dict,
    chain: Optional[CalmSequence] = None,
    calm: Optional[CALM] = None,
    surrogate: Optional[SurrogateModel] = None,
    transform: Optional[MaskedTransform] = None,
    curve_fitter: Optional[MaskedFitter] = None,
):
    self.task = task
    self.partitions = partitions
    self.chain = chain
    self.calm = calm
    self.surrogate = surrogate
    self.transform = transform
    self.curve_fitter = curve_fitter

interactions_info property

n_conditional_interactions property

The paper's metric: Σ_i |S_i| over the split features — one per (feature, conditioning feature) pair, so x1 | x2 and x2 | x1 count as two.

predict(X)

Source code in calm_additive/engine.py

def predict(self, X) -> np.ndarray:
    score = self.predict_score(X)
    if self.task == "classification":
        return (score > 0.5).astype(int)
    return score

predict_score(X)

Source code in calm_additive/engine.py

def predict_score(self, X) -> np.ndarray:
    X = np.asarray(X, dtype=float)
    if self.surrogate is not None:
        return self.surrogate.predict(X)
    Xt, mask = self.transform.transform(X)
    return np.asarray(self.curve_fitter.predict(Xt, mask)).reshape(-1)

terms()

The model as functions — one curve per (feature, region), each a feature's whole contribution inside its region. What print_summary and plot_curves read; the same for every curve fitter.

Source code in calm_additive/engine.py

def terms(self) -> ModelTerms:
    """The model as functions — one curve per (feature, region), each a
    feature's whole contribution inside its region. What `print_summary` and
    `plot_curves` read; the same for every curve fitter."""
    if self.surrogate is not None:
        return self.surrogate.terms(self.partitions)
    fitted = self.curve_fitter.curves()
    transform = self.transform
    features = []
    for j in range(transform.dim):
        cols = transform.foi_2_cols[j]
        total = float(sum(fitted.inactive[c] for c in cols))
        # inside a region the feature contributes its active column's
        # curve plus what its other, inactive columns add
        curves = [
            fitted.columns[c].shifted(total - float(fitted.inactive[c]))
            for c in cols
        ]
        leaves = [transform.columns[c][1] for c in cols]
        if leaves[0] is None:
            features.append(FeatureTerms(j, curves, [], curves[0]))
        else:
            features.append(FeatureTerms(j, curves, leaves, Curve.constant(total)))
    return ModelTerms(float(fitted.intercept), fitted.link, features)