class `monad.targets.DataSourceEvents`

Represents main entity events from a single data source (e.g. "transactions") defined in the pretrain phase.
You obtain it by indexing history or future:

transactions = future["transactions"]

Properties

TIMESTAMPS

`DataSourceEvents.timestamps`

Timestamps of the events.

Returns

numpy.ndarray

Methods

GETITEM

`DataSourceEvents.getitem(self, column_name: str | list[str])`

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

    # below we get price feature from the datasource transactions
    # note, it is ModalityEvents object, not the value of the feature
    price = attributes["transactions"]["price"]

    ...

Parameters

column_name: str | list[str]
Name of the feature from the data source.

Returns

ModalityEvents

LEN

`DataSourceEvents.len(self) -> int`

Returns the length of the DataSourceEvents object, ie. the count of events.

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

    no_events = len(future["transactions"])

    ...

Returns

int

COUNT

`Events.count(self) -> int`

Returns the total number of events across all data sources.

churn = 0 if future["transactions"].count() > 0 else 1

Returns

int — number of events

SUM

`Events.sum(self, column: str | Callable[[Any], Any], ignore_nan: Optional[bool] = True) -> float`

Returns the sum of values for a column or computed expression.

    ...
    
    # sum of a column    
    total_spend = future["transactions"].sum(
        column="purchase_value"
    )
    
    # sum of a computed expression (price * quantity)
    total_spend = future["transactions"].sum(
        column=lambda data: data["price"] * data["quantity"],
    )

    ...

Parameters

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

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

Returns

float

MEAN

`Events.mean(self, column: str | Callable[[Any], Any], ignore_nan: Optional[bool] = True) -> float`

Returns the mean value for a column or computed expression.

    ...

    # mean price
    avg_price = future["transactions"].mean(column="price")
    
    # mean basket line value
    avg_line_value = future["transactions"].mean(
        column=lambda data: data["price"] * data["quantity"],
    )

    ...

Parameters

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

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

Returns

float

MIN

`Events.min(self, column: str | Callable[[Any], Any], ignore_nan: Optional[bool] = True) -> float`

Returns the minimum value for a column or computed expression.

    ...

    min_value = future["transactions"].min(
        column=lambda data: data["price"] * data["quantity"],
    )

    ...

Parameters

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

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

Returns

float

MAX

`Events.max(self, column: str | Callable[[Any], Any], ignore_nan: Optional[bool] = True) -> float`

Returns the maximum value for a column or computed expression.

    ...

    max_value = future["transactions"].max(
        column=lambda data: data["price"] * data["quantity"],
    )

    ...

Parameters

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

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

Returns

float

GROUPBY

`Events.groupBy(self, by: str | list[str]) -> EventsGroupBy`

Groups events by one or more columns and returns an EventsGroupBy proxy.

    ...
    
    grouped = future["transactions"].groupBy("category")

    ...

Parameters

by: str
Column name or list of column names to group by.

Returns

EventsGroupBy

⚠️
Note
groupBy only produces output when followed by an aggregation (sum, mean, exists, etc.).

APPLY

`Events.apply(self, func: Callable[[Any], Any], target: str) -> DataSourceEvents`

Applies a transformation function to a column.

    ...

    normalized = future["product_buy"].apply(
        func=lambda x: x.lower(),
        target="brand",
    )

    ...

Parameters

func: Callable[[Any], Any]
Function applied element-wise.

target: str
Column to transform.

Returns

DataSourceEvents

FILTER

`Events.filter(self, by: str | Callable[[Any], Any], condition: Callable[[Any], bool]) -> DataSourceEvents`

Filters events using a column or computed expression.

    ...

    # filter on column value
    filtered = future["transactions"].filter(
        by="category",
        condition=lambda x: x > 100,
    )

    ...

Parameters

by: str | Callable[[Mapping[str, np.ndarray]], np.ndarray]
Column name or callable expression that receives a mapping (column_name -> np.ndarray).

condition: Callable[[Any], bool]
Predicate applied to each value.

Returns

DataSourceEvents