Api partition

Summary

Regional effects are found on a global effect object: find_regions(feature) searches for heterogeneity-reducing subregions of a feature and returns a Partition — a value object, not stored state (design contract R12).

1. create and .fit() a global effect method (PDP, RHALE, ALE, ShapDP, DerPDP)
2. partition = effect.find_regions(feature) to search for subregions
3. partition.show() to print the partition tree
4. partition.plot(idx) to plot the effect within region idx
5. partition.eval(idx, xs) / partition.eval_heter(idx, xs) to evaluate it

---

Usage

# set up the input
X = ...        # input data
predict = ...  # model to be explained
jacobian = ... # jacobian of the model (RHALE / DerPDP only)

1. Create and fit a global effect method:

   PDPRHALEShapDPALEDerPDP

   effect = effector.PDP(data=X, model=predict)
   effect.fit(features=[0, 1])
   

   effect = effector.RHALE(data=X, model=predict, model_jac=jacobian)
   effect.fit(features=[0, 1])
   

   effect = effector.ShapDP(data=X, model=predict, nof_instances=500)
   effect.fit(features=[0, 1])
   

   effect = effector.ALE(data=X, model=predict)
   effect.fit(features=[0, 1])
   

   effect = effector.DerPDP(data=X, model=predict, model_jac=jacobian)
   effect.fit(features=[0, 1])
   

2. Search for subregions of a feature (returns a Partition):

   find_regions(feature, *, finder="best", candidate_conditioning_features="all")

   Customize the search

   finder accepts a name ("best" / "best_level_wise") or a configured partitioner instance:

   finder = effector.space_partitioning.Best(
       min_heterogeneity_decrease_pcg=0.3,  # drop threshold (default: 0.1)
       max_depth=1,                         # max split levels (default: 2)
   )
   partition = effect.find_regions(0, finder=finder)
   

3. Print the partition tree:

   partition.show()

   Example Output

   Feature 3 - Full partition tree:
   🌳 Full Tree Structure:
   ───────────────────────
   hr 🔹 [id: 0 | heter: 0.43 | inst: 3476 | w: 1.00]
       workingday = 0.00 🔹 [id: 1 | heter: 0.36 | inst: 1129 | w: 0.32]
           temp < 6.50 🔹 [id: 3 | heter: 0.17 | inst: 568 | w: 0.16]
           temp ≥ 6.50 🔹 [id: 4 | heter: 0.21 | inst: 561 | w: 0.16]
       workingday = 1.00 🔹 [id: 2 | heter: 0.28 | inst: 2347 | w: 0.68]
           temp < 6.50 🔹 [id: 5 | heter: 0.19 | inst: 953 | w: 0.27]
           temp ≥ 6.50 🔹 [id: 6 | heter: 0.20 | inst: 1394 | w: 0.40]
   --------------------------------------------------
   Feature 3 - Statistics per tree level:
   🌳 Tree Summary:
   ─────────────────
   Level 0🔹heter: 0.43
       Level 1🔹heter: 0.31 | 🔻0.12 (28.15%)
           Level 2🔹heter: 0.19 | 🔻0.11 (37.10%)
   

4. Plot the effect within a region:

   partition.plot(idx) — idx is a region id from the tree (0 is the full data).

5. Evaluate the effect / heterogeneity within a region:

   y = partition.eval(idx, xs)             # mean effect within region idx
   h = partition.eval_heter(idx, xs)       # heterogeneity within region idx
   

Rules

Every Region is defined by a Rule — a normalized conjunction of per-feature conditions (effector.rules). Membership, display, and serialization all derive from the same rule, and rules are also a direct query surface:

# ad-hoc subregion queries, no Partition needed
effect.eval(feature, xs, rule="temp < 6.5 and workingday == 0")
effect.heter_score(feature, rule="temp < 6.5")

# user-authored partitions — same object find_regions returns
part = effector.Partition.from_rules(
    ["temp < 6.5", "temp >= 6.5"], effect=effect, feature=3
)

# serialization round-trip: rules travel, masks are recomputed on bind
d = part.to_dict()                              # small: rules + stats, no masks
restored = effector.Partition.from_dict(d).bind(effect)

API

effector.global_effect.GlobalEffectBase.find_regions(feature=None, *, features=None, finder='best', candidate_conditioning_features='all')

Search for subregions that resolve a feature's heterogeneity.

part = pdp.find_regions("hr")                       # one feature -> Partition
part.show()                                         # the tree + level stats
pdp.plot("hr", rule=part.leaves[0].rule)            # drill into a leaf

parts = pdp.find_regions(features="heterogeneous")  # several -> {name: Partition}
effector.plot_triage(pdp, partitions=parts)         # the before/after picture

A query, not a mutation (R12)

The result is a value — nothing is stored on the effect. Don't like a partition? Search again with different finder kwargs; nothing needs resetting.

Model-free

Every candidate split is scored by heter_score(feature, mask) on the cached local effects — zero model calls, whatever the grid size. Binning/scope are those the feature was fitted with, replayed.

Parameters:

Name	Type	Description	Default
feature	Union[int, str, None]	index or name of the one feature to partition (→ Partition).	None
features	Union[list, str, None]	several at once — a list, "all", or "heterogeneous" (heter_score at/above the median, the same convention effector.explain uses) → {feature_name: Partition}. Exactly one of feature/features must be given.	None
finder		"best" (default), "best_level_wise", or a configured finder instance (e.g. effector.space_partitioning.Best(...)).	'best'
candidate_conditioning_features		features allowed to define splits ("all" or a list of indices/names).	'all'

Returns:

Type	Description
	a Partition bound to this effect — or {feature_name: Partition}
	with features=.

Source code in effector/global_effect.py

def find_regions(
    self,
    feature: Union[int, str, None] = None,
    *,
    features: Union[list, str, None] = None,
    finder="best",
    candidate_conditioning_features="all",
):
    """Search for subregions that resolve a feature's heterogeneity.

    ```python
    part = pdp.find_regions("hr")                       # one feature -> Partition
    part.show()                                         # the tree + level stats
    pdp.plot("hr", rule=part.leaves[0].rule)            # drill into a leaf

    parts = pdp.find_regions(features="heterogeneous")  # several -> {name: Partition}
    effector.plot_triage(pdp, partitions=parts)         # the before/after picture
    ```

    !!! note "A query, not a mutation (R12)"
        The result is a value — nothing is stored on the effect. Don't
        like a partition? Search again with different finder kwargs;
        nothing needs resetting.

    !!! note "Model-free"
        Every candidate split is scored by `heter_score(feature, mask)`
        on the cached local effects — zero model calls, whatever the grid
        size. Binning/scope are those the feature was fitted with,
        replayed.

    Args:
        feature: index or name of the one feature to partition
            (→ `Partition`).
        features: several at once — a list, `"all"`, or `"heterogeneous"`
            (heter_score at/above the median, the same convention
            `effector.explain` uses) → `{feature_name: Partition}`.
            Exactly one of `feature`/`features` must be given.
        finder: `"best"` (default), `"best_level_wise"`, or a configured
            finder instance (e.g. `effector.space_partitioning.Best(...)`).
        candidate_conditioning_features: features allowed to define splits
            (`"all"` or a list of indices/names).

    Returns:
        a `Partition` bound to this effect — or `{feature_name: Partition}`
        with `features=`.
    """
    if (feature is None) == (features is None):
        raise ValueError(
            "find_regions takes exactly one of `feature` (singular -> "
            "Partition) or `features` (plural -> {name: Partition})"
        )
    if features is not None:
        return self._find_regions_plural(
            features,
            finder=finder,
            candidate_conditioning_features=candidate_conditioning_features,
        )

    from effector import space_partitioning  # lazy: one-way dep guard

    feature = self._resolve_feature(feature)
    if isinstance(candidate_conditioning_features, list):
        candidate_conditioning_features = [
            self._resolve_feature(f) for f in candidate_conditioning_features
        ]
    self._check_feature_type_supported(feature)
    self._ensure_local(feature)

    if isinstance(finder, str):
        finder = space_partitioning.return_default(finder)

    def score_fn(mask):
        return self.heter_score(feature, mask=mask)  # RAW; guard is the finder's

    partition = finder.find_regions(
        feature,
        self.data,
        score_fn,
        axis_limits=self.axis_limits,
        feature_types=self.feature_types,
        cat_limit=self.cat_limit,
        candidate_conditioning_features=candidate_conditioning_features,
        feature_names=self.feature_names,
        target_name=self.target_name,
    )
    return partition.bind(self)

effector.partition.Partition(regions, *, feature, feature_name, finder_name, feature_names=None)

An ordered set of Regions covering one feature's domain — a value, not stored state.

part = pdp.find_regions("hr")     # a Partition, bound to the effect
part.show()                       # tree + per-level stats
part.plot(1)                      # the effect inside region 1

regions[0] is always the root — full data, weight 1.0, the global effect itself — and the leaves partition it: pairwise disjoint, jointly covering (verified whenever masks are present). Each region's identity is its Rule; masks are derived caches, recomputed and re-verified by bind.

Heterogeneity = heter_score(feature, mask)

The heterogeneity a region reports is exactly the effect's heter_score(feature, mask=region_mask) — the same scalar the region finders minimize.

Methods:

Name	Description
mask	The boolean mask of region idx — a copy, safe to mutate.
label	Human-readable label for region idx.
show	Print the partition tree and per-level heterogeneity statistics.
show_axes	Print the leaves as a partition of the conditioning axes.
eval	The mean effect within region idx at positions xs.
eval_heter	The heterogeneity curve within region idx at positions xs.
plot	Plot the effect within region idx, titled with the region's rule.
bind	Attach an effect: recompute every region's mask from its rule and verify it.
from_rules	Build a partition from your own rules — no search.
to_dict	Serialize to a plain JSON-able dict.

Attributes:

Name	Type	Description
leaves		The regions that are no other region's parent — the finest partition.

Source code in effector/partition.py

def __init__(
    self, regions, *, feature, feature_name, finder_name, feature_names=None
):
    if not regions:
        raise ValueError("Partition needs at least one region (the root).")
    for i, r in enumerate(regions):
        if r.idx != i:
            raise ValueError(
                f"Region.idx must equal its position; region {i} has idx {r.idx}."
            )
    root = regions[0]
    if root.parent_idx is not None:
        raise ValueError("Root region (idx 0) must have parent_idx=None.")
    if root.weight != 1.0:
        raise ValueError("Root region (idx 0) must have weight == 1.0.")
    if not root.rule.is_root:
        raise ValueError("Root region (idx 0) must have the root rule (Rule({})).")

    self.regions = list(regions)
    self.feature = feature
    self.feature_name = feature_name
    self.finder_name = finder_name
    self.feature_names = list(feature_names) if feature_names is not None else None
    self._effect = None
    self._default_scale_x_list = None
    self._category_names = None
    self._check_leaves_partition()

leaves property

The regions that are no other region's parent — the finest partition.

A one-region partition's only leaf is the root.

mask(idx)

The boolean mask of region idx — a copy, safe to mutate.

Parameters:

Name	Type	Description	Default
idx		region index (0 = root).	required

Returns:

Type	Description
	boolean array (N,) over the bound effect's data.

Raises:

Type	Description
RuntimeError	the region has no mask (unbound partition — rebuilt from to_dict()); call bind(effect) first.

Source code in effector/partition.py

def mask(self, idx):
    """The boolean mask of region `idx` — a copy, safe to mutate.

    Args:
        idx: region index (0 = root).

    Returns:
        boolean array `(N,)` over the bound effect's data.

    Raises:
        RuntimeError: the region has no mask (unbound partition — rebuilt
            from `to_dict()`); call `bind(effect)` first.
    """
    region = self[idx]
    if region.mask is None:
        raise RuntimeError(
            f"Region {idx} has no mask (unbound partition — rebuilt from "
            f"to_dict()); call bind(effect) first."
        )
    return region.mask.copy()

label(idx, scale_x_list=None)

Human-readable label for region idx.

Root -> the feature name; otherwise "<feature> where <formatted rule>".

Parameters:

Name	Type	Description	Default
idx		region index.	required
scale_x_list		optional per-feature {"mean": ..., "std": ...} list to display values in original units; defaults to the bound effect's.	None

Returns:

Type	Description
	the label string.

Source code in effector/partition.py

def label(self, idx, scale_x_list=None):
    """Human-readable label for region `idx`.

    Root -> the feature name; otherwise ``"<feature> where <formatted rule>"``.

    Args:
        idx: region index.
        scale_x_list: optional per-feature ``{"mean": ..., "std": ...}``
            list to display values in original units; defaults to the
            bound effect's.

    Returns:
        the label string.
    """
    scale_x_list = helpers.resolve_scale(scale_x_list, self._default_scale_x_list)
    region = self[idx]
    if region.rule.is_root:
        return self.feature_name
    return (
        f"{self.feature_name} where {self._format_rule(region.rule, scale_x_list)}"
    )

show(scale_x_list=None)

Print the partition tree and per-level heterogeneity statistics.

Each node shows its splitting condition plus a [id | heter | inst | w] chip; the summary shows the heterogeneity drop per level. Works on unbound partitions too.

Parameters:

Name	Type	Description	Default
scale_x_list		optional per-feature {"mean": ..., "std": ...} list for display in original units; defaults to the bound effect's.	None

Source code in effector/partition.py

def show(self, scale_x_list=None):
    """Print the partition tree and per-level heterogeneity statistics.

    Each node shows its splitting condition plus a
    ``[id | heter | inst | w]`` chip; the summary shows the heterogeneity
    drop per level. Works on unbound partitions too.

    Args:
        scale_x_list: optional per-feature ``{"mean": ..., "std": ...}``
            list for display in original units; defaults to the bound
            effect's.
    """
    scale_x_list = helpers.resolve_scale(scale_x_list, self._default_scale_x_list)
    feature = self.feature

    # A flat producer yields non-root regions with parent_idx=None;
    # tree rendering does not apply there.
    is_flat = any(r.level > 0 and r.parent_idx is None for r in self.regions)

    print("\n")
    print("Feature {} - Full partition tree:".format(feature))
    if is_flat:
        for r in self.regions:
            print(self.label(r.idx, scale_x_list))
        print("-" * 50)
        print("Feature {} - Statistics per tree level:".format(feature))
        print("\n")
        return

    if len(self) == 1:
        print("No splits found for feature {}".format(feature))
    else:
        self._print_full_tree(scale_x_list)

    print("-" * 50)
    print("Feature {} - Statistics per tree level:".format(feature))
    if len(self) == 1:
        print("No splits found for feature {}".format(feature))
    else:
        self._print_level_stats()
    print("\n")

show_axes(scale_x_list=None)

Print the leaves as a partition of the conditioning axes.

When the leaves differ on one conditioning feature, print them as a partition of that axis; on two, as a grid. Anything else falls back to the tree print (show).

Parameters:

Name	Type	Description	Default
scale_x_list		optional per-feature {"mean": ..., "std": ...} list for display in original units; defaults to the bound effect's.	None

Source code in effector/partition.py

def show_axes(self, scale_x_list=None):
    """Print the leaves as a partition of the conditioning axes.

    When the leaves differ on one conditioning feature, print them as a
    partition of that axis; on two, as a grid. Anything else falls back
    to the tree print (`show`).

    Args:
        scale_x_list: optional per-feature ``{"mean": ..., "std": ...}``
            list for display in original units; defaults to the bound
            effect's.
    """
    scale_x_list = helpers.resolve_scale(scale_x_list, self._default_scale_x_list)
    leaves = self.leaves
    feats = sorted({f for leaf in leaves for f in leaf.rule.features})

    def fmt(f, subset):
        return self._format_rule(Rule({f: subset}), scale_x_list)

    if not feats:
        return self.show(scale_x_list)

    if len(feats) == 1:
        f = feats[0]
        if any(leaf.rule.get(f) is None for leaf in leaves):
            return self.show(scale_x_list)
        print("\n")
        print(f"Feature {self.feature} - Partition along 1 axis:")
        for leaf in sorted(leaves, key=lambda r: _subset_sort_key(r.rule[f])):
            print(f"{fmt(f, leaf.rule[f])} 🔹 {self._stat_chip(leaf)}")
        print("\n")
        return

    if len(feats) == 2:
        f_row, f_col = feats
        cells = {}
        for leaf in leaves:
            sr, sc = leaf.rule.get(f_row), leaf.rule.get(f_col)
            if sr is None or sc is None or (sr, sc) in cells:
                return self.show(scale_x_list)
            cells[(sr, sc)] = leaf
        rows = sorted({k[0] for k in cells}, key=_subset_sort_key)
        cols = sorted({k[1] for k in cells}, key=_subset_sort_key)
        if len(rows) * len(cols) != len(cells):
            return self.show(scale_x_list)

        row_labels = [fmt(f_row, s) for s in rows]
        col_labels = [fmt(f_col, s) for s in cols]
        cell_strs = [
            [self._stat_chip(cells[(sr, sc)], with_weight=False) for sc in cols]
            for sr in rows
        ]
        w0 = max(len(lab) for lab in row_labels)
        widths = [
            max(len(col_labels[j]), *(len(row[j]) for row in cell_strs))
            for j in range(len(cols))
        ]
        print("\n")
        print(f"Feature {self.feature} - Partition along 2 axes:")
        print(
            " " * w0
            + "   "
            + "   ".join(lab.ljust(widths[j]) for j, lab in enumerate(col_labels))
        )
        for i, row_lab in enumerate(row_labels):
            print(
                row_lab.ljust(w0)
                + "   "
                + "   ".join(
                    cell_strs[i][j].ljust(widths[j]) for j in range(len(cols))
                )
            )
        print("\n")
        return

    return self.show(scale_x_list)

eval(idx, xs, **kwargs)

The mean effect within region idx at positions xs.

Sugar for effect.eval(feature, xs, mask=part.mask(idx)) — re-summarized from cached local effects, zero model calls.

Parameters:

Name	Type	Description	Default
idx		region index (0 = root = the global effect).	required
xs		where to evaluate, (T,).	required
**kwargs		forwarded to effect.eval (e.g. centering).	{}

Returns:

Type	Description
	the mean effect at xs, shape (T,).

Raises:

Type	Description
RuntimeError	the partition is unbound; call bind(effect) first.

Source code in effector/partition.py

def eval(self, idx, xs, **kwargs):
    """The mean effect within region `idx` at positions `xs`.

    Sugar for ``effect.eval(feature, xs, mask=part.mask(idx))`` —
    re-summarized from cached local effects, zero model calls.

    Args:
        idx: region index (0 = root = the global effect).
        xs: where to evaluate, `(T,)`.
        **kwargs: forwarded to `effect.eval` (e.g. `centering`).

    Returns:
        the mean effect at `xs`, shape `(T,)`.

    Raises:
        RuntimeError: the partition is unbound; call `bind(effect)` first.
    """
    return self._require_effect().eval(
        self.feature, xs, mask=self.mask(idx), **kwargs
    )

eval_heter(idx, xs)

The heterogeneity curve within region idx at positions xs.

Sugar for effect.eval_heter(feature, xs, mask=part.mask(idx)) — model-free.

Parameters:

Name	Type	Description	Default
idx		region index (0 = root = the global effect).	required
xs		where to evaluate, (T,).	required

Returns:

Type	Description
	the heterogeneity curve at xs, shape (T,), non-negative.

Raises:

Type	Description
RuntimeError	the partition is unbound; call bind(effect) first.

Source code in effector/partition.py

def eval_heter(self, idx, xs):
    """The heterogeneity curve within region `idx` at positions `xs`.

    Sugar for ``effect.eval_heter(feature, xs, mask=part.mask(idx))`` —
    model-free.

    Args:
        idx: region index (0 = root = the global effect).
        xs: where to evaluate, `(T,)`.

    Returns:
        the heterogeneity curve at `xs`, shape `(T,)`, non-negative.

    Raises:
        RuntimeError: the partition is unbound; call `bind(effect)` first.
    """
    return self._require_effect().eval_heter(self.feature, xs, mask=self.mask(idx))

plot(idx, scale_x_list=None, **plot_kwargs)

Plot the effect within region idx, titled with the region's rule.

Sugar for effect.plot(feature, mask=part.mask(idx), ...) with the region's label as the feature label.

Parameters:

Name	Type	Description	Default
idx		region index (0 = root = the global effect).	required
scale_x_list		optional per-feature {"mean": ..., "std": ...} list — scales both the axis and the rule in the label.	None
**plot_kwargs		forwarded to effect.plot (e.g. heterogeneity, centering, show_plot).	{}

Raises:

Type	Description
RuntimeError	the partition is unbound; call bind(effect) first.

Source code in effector/partition.py

def plot(self, idx, scale_x_list=None, **plot_kwargs):
    """Plot the effect within region `idx`, titled with the region's rule.

    Sugar for ``effect.plot(feature, mask=part.mask(idx), ...)`` with the
    region's `label` as the feature label.

    Args:
        idx: region index (0 = root = the global effect).
        scale_x_list: optional per-feature ``{"mean": ..., "std": ...}``
            list — scales both the axis and the rule in the label.
        **plot_kwargs: forwarded to `effect.plot` (e.g. `heterogeneity`,
            `centering`, `show_plot`).

    Raises:
        RuntimeError: the partition is unbound; call `bind(effect)` first.
    """
    effect = self._require_effect()
    scale_x = scale_x_list[self.feature] if isinstance(scale_x_list, list) else None
    return effect.plot(
        self.feature,
        mask=self.mask(idx),
        feature_label=self.label(idx, scale_x_list),
        scale_x=scale_x,
        **plot_kwargs,
    )

bind(effect)

Attach an effect: recompute every region's mask from its rule and verify it.

part = effector.Partition.from_dict(d).bind(pdp)   # live again

Masks are recomputed from the rules against effect.data, then checked against the stored evidence — the finder's mask when present (the find_regions path), else the serialized nof_instances (the from_dict path). This is what makes a deserialized partition safely re-attachable.

Parameters:

Name	Type	Description	Default
effect		a fitted global effect whose data the rules apply to.	required

Returns:

Type	Description
	self, live — eval/eval_heter/plot/mask work afterwards.

Raises:

Type	Description
ValueError	a rule selects different instances than the stored evidence — the effect's data differs from the data the partition was built on (check nof_instances subsampling and random_state).

Source code in effector/partition.py

def bind(self, effect):
    """Attach an effect: recompute every region's mask from its rule and verify it.

    ```python
    part = effector.Partition.from_dict(d).bind(pdp)   # live again
    ```

    Masks are recomputed from the rules against `effect.data`, then checked
    against the stored evidence — the finder's mask when present (the
    `find_regions` path), else the serialized `nof_instances` (the
    `from_dict` path). This is what makes a deserialized partition safely
    re-attachable.

    Args:
        effect: a fitted global effect whose `data` the rules apply to.

    Returns:
        `self`, live — `eval`/`eval_heter`/`plot`/`mask` work afterwards.

    Raises:
        ValueError: a rule selects different instances than the stored
            evidence — the effect's data differs from the data the
            partition was built on (check `nof_instances` subsampling
            and `random_state`).
    """
    rebuilt = []
    for r in self.regions:
        m = r.rule.contains(effect.data)
        if r.mask is not None:
            if not (m.shape == r.mask.shape and np.array_equal(m, r.mask)):
                raise ValueError(
                    f"Region {r.idx} ({r.name!r}): the rule selects different "
                    f"instances than the stored mask — the partition was built "
                    f"on different data."
                )
            rebuilt.append(r)
        else:
            n = int(m.sum())
            if n != r.nof_instances:
                raise ValueError(
                    f"Region {r.idx} ({r.name!r}): the rule selects {n} "
                    f"instances on this effect's data but the partition was "
                    f"built with {r.nof_instances}. The effect's data differs "
                    f"— check `nof_instances` subsampling and `random_state`."
                )
            rebuilt.append(replace(r, mask=m))
    self.regions = rebuilt
    self._effect = effect
    self._default_scale_x_list = effect.scale_x_list
    self._category_names = effect.feature_metadata.category_names
    if self.feature_names is None:
        self.feature_names = list(effect.feature_names)
    self._check_leaves_partition()
    return self

from_rules(rules, *, effect, feature, finder_name='user') classmethod

Build a partition from your own rules — no search.

part = effector.Partition.from_rules(
    ["workingday == 0", "workingday == 1"], effect=pdp, feature=3
)
part.plot(1)

Strings are parsed with the effect's metadata (Rule.parse); stats (heterogeneity, counts, weights) are stamped from the effect's cached local effects — zero model calls — and the result comes back bound.

Parameters:

Name	Type	Description	Default
rules		Rule objects or rule strings like "temp < 3".	required
effect		a fitted global effect providing data and metadata.	required
feature		index of the feature of interest.	required
finder_name		label recorded on the partition (default "user").	'user'

Returns:

Type	Description
	a bound Partition: the root plus one level-1 region per rule.

Raises:

Type	Description
ValueError	the rules do not partition the data — some instance is covered more than once or not at all.

Source code in effector/partition.py

@classmethod
def from_rules(cls, rules, *, effect, feature, finder_name="user"):
    """Build a partition from your own rules — no search.

    ```python
    part = effector.Partition.from_rules(
        ["workingday == 0", "workingday == 1"], effect=pdp, feature=3
    )
    part.plot(1)
    ```

    Strings are parsed with the effect's metadata (`Rule.parse`); stats
    (heterogeneity, counts, weights) are stamped from the effect's cached
    local effects — zero model calls — and the result comes back bound.

    Args:
        rules: `Rule` objects or rule strings like `"temp < 3"`.
        effect: a fitted global effect providing data and metadata.
        feature: index of the feature of interest.
        finder_name: label recorded on the partition (default `"user"`).

    Returns:
        a bound `Partition`: the root plus one level-1 region per rule.

    Raises:
        ValueError: the rules do not partition the data — some instance
            is covered more than once or not at all.
    """
    parsed = []
    for r in rules:
        if isinstance(r, str):
            levels = {
                j: np.unique(effect.data[:, j])
                for j in range(effect.dim)
                if ingestion.is_categorical(effect.feature_types[j])
            }
            r = Rule.parse(
                r,
                feature_names=effect.feature_names,
                feature_types=effect.feature_types,
                levels=levels,
                category_names=effect.feature_metadata.category_names,
            )
        parsed.append(r)

    n_total = effect.data.shape[0]
    feature_name = effect.feature_names[feature]
    regions = [
        Region(
            idx=0,
            name=feature_name,
            rule=Rule({}),
            heterogeneity=float(effect.heter_score(feature)),
            nof_instances=n_total,
            weight=1.0,
            level=0,
            parent_idx=None,
            mask=np.ones(n_total, dtype=bool),
        )
    ]
    for i, rule in enumerate(parsed):
        mask = rule.contains(effect.data)
        n = int(mask.sum())
        regions.append(
            Region(
                idx=i + 1,
                name=f"{feature_name} where {rule.format(effect.feature_names)}",
                rule=rule,
                heterogeneity=float(effect.heter_score(feature, mask=mask)),
                nof_instances=n,
                weight=n / n_total,
                level=1,
                parent_idx=0,
                mask=mask,
            )
        )
    partition = cls(
        regions,
        feature=feature,
        feature_name=feature_name,
        finder_name=finder_name,
        feature_names=effect.feature_names,
    )
    return partition.bind(effect)

to_dict()

Serialize to a plain JSON-able dict.

Rules + stats only

Each region's rule and stamped statistics are serialized — never masks, never the data, never the effect or the model. from_dict(...) + bind(effect) restore the rest.

Returns:

Type	Description
	a dict with schema_version 2, feature metadata, and one entry
	per region (rule, heterogeneity, counts, weight, tree links).

Source code in effector/partition.py

def to_dict(self):
    """Serialize to a plain JSON-able dict.

    !!! note "Rules + stats only"
        Each region's rule and stamped statistics are serialized — never
        masks, never the data, never the effect or the model.
        `from_dict(...)` + `bind(effect)` restore the rest.

    Returns:
        a dict with `schema_version` 2, feature metadata, and one entry
        per region (rule, heterogeneity, counts, weight, tree links).
    """
    return {
        "schema_version": 2,
        "feature": self.feature,
        "feature_name": self.feature_name,
        "feature_names": self.feature_names,
        "finder": self.finder_name,
        "regions": [
            {
                "idx": r.idx,
                "name": r.name,
                "rule": r.rule.to_dict(),
                "heterogeneity": float(r.heterogeneity),
                "nof_instances": int(r.nof_instances),
                "weight": float(r.weight),
                "level": int(r.level),
                "parent_idx": r.parent_idx,
            }
            for r in self.regions
        ],
    }

effector.partition.Region(idx, name, rule, heterogeneity, nof_instances, weight, level=0, parent_idx=None, mask=None) dataclass

One subregion of a feature's domain — a frozen value whose identity is its Rule.

idx == 0 is the root: full data, weight 1.0, the global effect itself. heterogeneity is effect.heter_score(feature, mask) within the region, stamped when the region was built. mask is a derived cache — the rule applied to one dataset — and is None on a deserialized region until Partition.bind recomputes it.

Attributes:

Name	Type	Description
idx	int	position in the partition; 0 is the root.
name	str	display name — the feature name, or "<feature> | <rule>".
rule	Rule	the Rule that defines membership — the region's identity.
heterogeneity	float	heter_score(feature, mask) within the region.
nof_instances	int	how many instances the rule selects.
weight	float	nof_instances / N.
level	int	depth in the partition tree (root = 0).
parent_idx	Optional[int]	index of the parent region; None for the root.
mask	Optional[ndarray]	boolean (N,) cache of rule.contains(data); None until bound.

effector.rules.Rule(conditions=())

A normalized conjunction of per-feature conditions — a region's identity.

rule = effector.Rule.parse("temp < 3 and season == 0", ...)
mask = rule.contains(X)                  # (N,) bool
pdp.plot("hr", rule=rule)                # effects accept rules directly

At most one subset per feature (a hyperbox with categorical level sets); membership, display, and serialization all derive from this one object. Rules are immutable and hashable; equality is order-insensitive. Rule({}) is the root rule — it contains everything and formats to "". A full (unbounded) Interval constrains nothing and is dropped at construction, so Rule({0: Interval()}) == Rule({}). Emptiness (is_empty) is representable and never raises; only parse rejects contradictions, as a courtesy to humans.

Methods:

Name	Description
contains	Which rows of X satisfy the rule — the only place rules become masks.
intersect	The conjunction of two rules — subsets on shared features intersect.
refine	A new rule with one more condition — sugar for intersect(Rule([condition])).
format	Human-readable conjunction. A single condition stays bare
parse	Parse "temp < 3 and season != 2 and hr in {7, 8}" into a Rule.
to_dict	Serialize to a plain JSON-able dict — a list of per-feature conditions.
from_dict	Rebuild a Rule from to_dict() output.

Source code in effector/rules.py

def __init__(self, conditions: Union[dict, Iterable[Condition]] = ()):
    merged: dict = {}
    if isinstance(conditions, dict):
        items = [Condition(f, s) for f, s in conditions.items()]
    else:
        items = list(conditions)
    for cond in items:
        if not isinstance(cond, Condition):
            raise ValueError(
                f"Rule accepts Conditions or a {{feature: subset}} dict; "
                f"got {cond!r}"
            )
        f, s = cond.feature, cond.subset
        merged[f] = merged[f].intersect(s) if f in merged else s
    # a full interval constrains nothing — drop it (canonical form)
    self._conditions = {
        f: s
        for f, s in merged.items()
        if not (isinstance(s, Interval) and s == Interval())
    }

contains(X)

Which rows of X satisfy the rule — the only place rules become masks.

Parameters:

Name	Type	Description	Default
X	ndarray	data matrix (N, D).	required

Returns:

Type	Description
ndarray	boolean mask (N,); all-True for the root rule.

Raises:

Type	Description
ValueError	X is not 2D, or the rule references a feature index beyond X.shape[1].

Source code in effector/rules.py

def contains(self, X: np.ndarray) -> np.ndarray:
    """Which rows of `X` satisfy the rule — the only place rules become masks.

    Args:
        X: data matrix `(N, D)`.

    Returns:
        boolean mask `(N,)`; all-`True` for the root rule.

    Raises:
        ValueError: `X` is not 2D, or the rule references a feature index
            beyond `X.shape[1]`.
    """
    X = np.asarray(X)
    if X.ndim != 2:
        raise ValueError(f"X must be a 2D (N, D) array; got ndim {X.ndim}")
    if self._conditions and max(self._conditions) >= X.shape[1]:
        raise ValueError(
            f"rule references feature {max(self._conditions)} but X has "
            f"only {X.shape[1]} columns"
        )
    mask = np.ones(X.shape[0], dtype=bool)
    for f, subset in self._conditions.items():
        mask &= subset.contains(X[:, f])
    return mask

intersect(other)

The conjunction of two rules — subsets on shared features intersect.

Returns:

Type	Description
'Rule'	a new Rule; may be empty (is_empty), never raises.

Source code in effector/rules.py

def intersect(self, other: "Rule") -> "Rule":
    """The conjunction of two rules — subsets on shared features intersect.

    Returns:
        a new `Rule`; may be empty (`is_empty`), never raises.
    """
    conditions = dict(self._conditions)
    for f, s in other._conditions.items():
        conditions[f] = conditions[f].intersect(s) if f in conditions else s
    return Rule(conditions)

refine(condition)

A new rule with one more condition — sugar for intersect(Rule([condition])).

Source code in effector/rules.py

def refine(self, condition: Condition) -> "Rule":
    """A new rule with one more condition — sugar for `intersect(Rule([condition]))`."""
    return self.intersect(Rule([condition]))

format(feature_names=None, scale_x_list=None, category_names=None)

Human-readable conjunction. A single condition stays bare ("temp < 3.00"); two or more are parenthesized: "(temp < 3.00) and (season = winter)".

Parameters:

Name	Type	Description	Default
feature_names		display names, position = index; defaults to x_<i>.	None
scale_x_list		per-feature {"mean": ..., "std": ...} list to show numeric values in original units.	None
category_names		{feature: {level: name}} map to show level names instead of numbers.	None

Returns:

Type	Description
str	the formatted string; the root rule formats to "".

Source code in effector/rules.py

def format(self, feature_names=None, scale_x_list=None, category_names=None) -> str:
    """Human-readable conjunction. A single condition stays bare
    (``"temp < 3.00"``); two or more are parenthesized:
    ``"(temp < 3.00) and (season = winter)"``.

    Args:
        feature_names: display names, position = index; defaults to
            ``x_<i>``.
        scale_x_list: per-feature ``{"mean": ..., "std": ...}`` list to
            show numeric values in original units.
        category_names: ``{feature: {level: name}}`` map to show level
            names instead of numbers.

    Returns:
        the formatted string; the root rule formats to ``""``.
    """
    parts = [
        Condition(f, s).format(feature_names, scale_x_list, category_names)
        for f, s in self._conditions.items()
    ]
    if len(parts) <= 1:
        return "".join(parts)
    return " and ".join(f"({p})" for p in parts)

parse(text, *, feature_names, feature_types=None, levels=None, category_names=None) classmethod

Parse "temp < 3 and season != 2 and hr in {7, 8}" into a Rule.

Grammar: clauses joined by and; each clause <feature_name> <op> <value> with op in < <= > >= == != in (= is accepted as ==). Continuous features take the four inequalities; discrete features take ==, in {…}, and !=. Level names are accepted where category_names covers the feature.

Parameters:

Name	Type	Description	Default
text	str	the rule string.	required
feature_names		feature display names, position = index.	required
feature_types		per-feature types; required for the discrete-only ops ==/!=/in.	None
levels		{feature_index: observed levels} map; required by != to materialize the complement.	None
category_names		{feature: {level: name}} map so values can be given by name.	None

Returns:

Type	Description
'Rule'	the parsed Rule.

Raises:

Type	Description
ValueError	unknown feature, unparsable clause, an op unsupported for the feature's type, or contradictory clauses.

Source code in effector/rules.py

@classmethod
def parse(
    cls,
    text: str,
    *,
    feature_names,
    feature_types=None,
    levels=None,
    category_names=None,
) -> "Rule":
    """Parse ``"temp < 3 and season != 2 and hr in {7, 8}"`` into a `Rule`.

    Grammar: clauses joined by ``and``; each clause
    ``<feature_name> <op> <value>`` with op in ``< <= > >= == != in``
    (``=`` is accepted as ``==``). Continuous features take the four
    inequalities; discrete features take ``==``, ``in {…}``, and ``!=``.
    Level *names* are accepted where `category_names` covers the feature.

    Args:
        text: the rule string.
        feature_names: feature display names, position = index.
        feature_types: per-feature types; required for the discrete-only
            ops ``==``/``!=``/``in``.
        levels: ``{feature_index: observed levels}`` map; required by
            ``!=`` to materialize the complement.
        category_names: ``{feature: {level: name}}`` map so values can be
            given by name.

    Returns:
        the parsed `Rule`.

    Raises:
        ValueError: unknown feature, unparsable clause, an op unsupported
            for the feature's type, or contradictory clauses.
    """
    name_to_idx = {str(n): i for i, n in enumerate(feature_names)}
    conditions: dict = {}

    for clause in re.split(r"(?i)\s+and\s+", text.strip()):
        feature, subset = cls._parse_clause(
            clause, name_to_idx, feature_types, levels, category_names
        )
        merged = (
            conditions[feature].intersect(subset)
            if feature in conditions
            else subset
        )
        if merged.is_empty:
            raise ValueError(
                f"contradictory conditions on feature "
                f"{feature_names[feature]!r} in {text!r}"
            )
        conditions[feature] = merged
    return cls(conditions)

to_dict()

Serialize to a plain JSON-able dict — a list of per-feature conditions.

Source code in effector/rules.py

def to_dict(self) -> dict:
    """Serialize to a plain JSON-able dict — a list of per-feature conditions."""
    return {
        "conditions": [
            {"feature": f, **s.to_dict()} for f, s in self._conditions.items()
        ]
    }

from_dict(d) classmethod

Rebuild a Rule from to_dict() output.

Source code in effector/rules.py

@classmethod
def from_dict(cls, d: dict) -> "Rule":
    """Rebuild a `Rule` from `to_dict()` output."""
    return cls({c["feature"]: subset_from_dict(c) for c in d["conditions"]})