class `monad.targets.EventsGroupBy`

Contains grouped events from a single data source (a DataSourceEvents stream).
You obtain it by calling groupBy(...) on a data source.

def target_fn(history: Events, future: Events, attributes: Attributes, ctx: Dict) -> np.ndarray:
    ...
    
    # first we group events (future purchases) by selected column (brand)
    # then we do other operations; below we filter to 2 brands and check for existence

    prop_target = (
        future["transactions"]
        .groupby("brand")
        .count(normalize=True, groups=["Garmin", "Suunto"])
    )

    ...

Methods

COUNT

`EventsGroupBy.count(self, normalize: Optional[bool] = False, groups: Optional[list[Any]] = None) -> tuple[numpy.ndarray, list[str]]`

Counts elements in each group.

def target_fn(history: Events, future: Events, attributes: Attributes, ctx: Dict) -> np.ndarray:
    ...

    # count events per brand, normalized so the sum is 1
    counts, brands = (
        future["transactions"]
        .groupBy("brand")
        .count(normalize=True, groups=["Garmin", "Suunto"])
    )

    ...

Parameters

normalize: bool
Default: False
Scales counts so that they sum to 1 (only if total count is > 0).

groups: list[Any] | None
Default: None
Limits output to the provided group keys.

Returns

tuple[numpy.ndarray, list[str]], a tuple with counts per group and group names.

SUM

`EventsGroupBy.sum(self, target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray], groups: Optional[list[Any]] = None, ignore_nan: Optional[bool] = True) -> tuple[numpy.ndarray, list[str]]`

Sums the values of target in each group. target can be a column name or a callable expression.

def target_fn(history: Events, future: Events, attributes: Attributes, ctx: Dict) -> np.ndarray:
    ...

    # sum a column per category
    spend_by_cat, categories = (
        future["transactions"]
        .groupBy("category")
        .sum(target="purchase_value")
    )

    # sum a computed expression (price * quantity) per category
    spend_by_cat, categories = (
        future["transactions"]
        .groupBy("category")
        .sum(target=lambda data: data["price"] * data["quantity"])
    )

    ...

Parameters

target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray]
Column name, or callable that receives a mapping (column_name -> np.ndarray) and returns an array aligned with events.

groups: list[Any] | None
Default: None
Limits output to the provided group keys.

ignore_nan: bool
Default: True
Whether to ignore NaN values.

Returns

tuple[numpy.ndarray, list[str]], a tuple with sums per group and group names.

MEAN

`EventsGroupBy.mean(self, target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray], groups: Optional[list[Any]] = None, ignore_nan: Optional[bool] = True) -> tuple[numpy.ndarray, list[str]]`

Computes the mean of target in each group.

def target_fn(history: Events, future: Events, attributes: Attributes, ctx: Dict) -> np.ndarray:
    ...

    avg_price_by_brand, brands = (
        future["transactions"]
        .groupBy("brand")
        .mean(target="price")
    )

    avg_line_value_by_brand, brands = (
        future["transactions"]
        .groupBy("brand")
        .mean(target=lambda data: data["price"] * data["quantity"])
    )

    ...

Parameters

target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray]
Column name, or callable that receives a mapping (column_name -> np.ndarray) and returns an array aligned with events.

groups: list[Any] | None
Default: None
Limits output to the provided group keys.

ignore_nan: bool
Default: True
Whether to ignore NaN values.

Returns

tuple[numpy.ndarray, list[str]], a tuple with means per group and group names.

MIN

`EventsGroupBy.min(self, target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray], groups: Optional[list[Any]] = None, ignore_nan: Optional[bool] = True) -> tuple[numpy.ndarray, list[str]]`

Computes the minimum of target in each group.

def target_fn(history: Events, future: Events, attributes: Attributes, ctx: Dict) -> np.ndarray:
    ...

    min_line_value_by_cat, categories = (
        future["transactions"]
        .groupBy("category")
        .min(target=lambda data: data["price"] * data["quantity"])
    )

    ...

Parameters

target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray]
Column name, or callable that receives a mapping (column_name -> np.ndarray) and returns an array aligned with events.

groups: list[Any] | None
Default: None
Limits output to the provided group keys.

ignore_nan: bool
Default: True
Whether to ignore NaN values.

Returns

tuple[numpy.ndarray, list[str]], a tuple with minimums per group and group names.

MAX

`EventsGroupBy.max(self, target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray], groups: Optional[list[Any]] = None, ignore_nan: Optional[bool] = True) -> tuple[numpy.ndarray, list[str]]`

Computes the maximum of target in each group.

def target_fn(history: Events, future: Events, attributes: Attributes, ctx: Dict) -> np.ndarray:
    ...

    max_line_value_by_store, stores = (
        future["transactions"]
        .groupBy("store")
        .max(target=lambda data: data["price"] * data["quantity"])
    )

    ...

Parameters

target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray]
Column name, or callable that receives a mapping (column_name -> np.ndarray) and returns an array aligned with events.

groups: list[Any] | None
Default: None
Limits output to the provided group keys.

ignore_nan: bool
Default: True
Whether to ignore NaN values.

Returns

tuple[numpy.ndarray, list[str]], a tuple with maximums per group and group names.

EXISTS

`EventsGroupBy.exists(self, groups: list[Any]) -> tuple[numpy.ndarray, list[str]]`

Checks whether each provided group has any events.

def target_fn(history: Events, future: Events, attributes: Attributes, ctx: Dict) -> np.ndarray:
    ...

    target_brands = ["Nike", "Adidas"]

    present, brands = (
        future["product_buy"]
        .groupBy("brand")
        .exists(groups=target_brands)
    )

    ...

Parameters

groups: list[Any]
Group keys to check for existence.

Returns

tuple[numpy.ndarray, list[str]], a tuple with 0/1 indicators and group names.

APPLY

`EventsGroupBy.apply(self, func: Callable[[numpy.ndarray], Any], default_value: Any, target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray], groups: Optional[list[Any]] = None) -> tuple[Any, list[str]]`

Applies func to each group’s values.

target can be a column name or a callable expression.
default_value is used for groups with no events (only when groups is provided).

import numpy as np

def target_fn(history: Events, future: Events, attributes: Attributes, ctx: Dict) -> np.ndarray:
    ...

    max_line_value, brands = (
        future["transactions"]
        .groupBy("brand")
        .apply(
            func=lambda x: float(np.nanmax(x)),
            default_value=0.0,
            target=lambda data: data["price"] * data["quantity"],
            groups=["Garmin", "Suunto"],
        )
    )

    ...

Parameters

func: Callable[[numpy.ndarray], Any]
Function applied to values in each group.

default_value: Any
Default output value for empty groups (used only when groups is set).
The result returned by func will be cast to the type of default_value.
Therefore, ensure that the type of default_value matches the expected output.
For example, avoid setting an integer if your column contains floating-point values.

target: str | Callable[[Mapping[str, numpy.ndarray]], numpy.ndarray]
Column name or callable expression producing values to pass to func.

groups: list[Any] | None
Default: None
Limits output to the provided group keys.

Returns

tuple[Any, list[str]], a tuple with values returned by func per group and group names.