Target Function Reference

The target function defines what a scenario model predicts. It receives historical and future events for each entity and returns a label, value, or sketch — or None to exclude the entity.

Signature

import numpy as np
from monad.targets import Events, Attributes, Sketch

def target_fn(
    history: Events,
    future: Events,
    attributes: Attributes,
    ctx: dict,
) -> np.ndarray | Sketch | None:
    ...

Argument	Type	Description
`history`	`Events`	All events before the split point. This is what the model sees as input.
`future`	`Events`	All events after the split point. Use this to compute the prediction target.
`attributes`	`Attributes`	Static entity properties (e.g., customer demographics).
`ctx`	`dict`	Context dictionary with `SPLIT_TIMESTAMP`, `MODE`, `TRAINING_END_TIMESTAMP`.

Return Types by Task

Task	Return Type	Example
Binary Classification	`np.array([0 or 1], dtype=np.float32)`	`np.array([1.0], dtype=np.float32)`
Multiclass Classification	normalized class distribution with `shape=(num_classes,)`	`np.array([0.0, 0.0, 1.0], dtype=np.float32)`
Multilabel Classification	`np.array([0,1,0,...], dtype=np.float32)`	`np.array([1,0,1], dtype=np.float32)`
Regression	`np.array([value], dtype=np.float32)`	`np.array([1500.0], dtype=np.float32)`
Recommendation	`Sketch`	`sketch(items, weights)`
Skip entity	`None`	`return None`

Use None for the declared population, not as outcome cleanup

Return None for insufficient history, incomplete time windows, invalid entities (test accounts or bots), eligibility known before the split, or an outcome that the business question truly leaves undefined. No future events is not, by itself, a universal reason to drop an entity. When absence is a real outcome (for example, no purchase or zero spend), return that label and keep it in training and evaluation.

A question can explicitly define a label only after a future prerequisite happens — for example, "among entities that act in the window, which class dominates?" Returning None when that prerequisite does not happen implements the stated conditional evaluation population. That population is retrospective: the model can compare outcomes among entities known after the window to meet the prerequisite, but it cannot identify those entities prospectively at the split. State this limitation, report the conditional denominator, and do not present the result as prospective coverage of all entities. Never use future activity merely to remove valid negative labels or improve class balance.

`Events` API

Access event data sources by name:

txns = history["transactions"]   # Returns DataSourceEvents
txns = future["transactions"]

Aggregation Methods

Methods on DataSourceEvents:

Method	Return Type	Description
`.count()`	`int`	Number of events.
`.sum(column, ignore_nan=True)`	`float`	Sum of values. `column`: name or callable.
`.mean(column, ignore_nan=True)`	`float`	Mean of values. `column`: name or callable.
`.min(column, ignore_nan=True)`	`float`	Minimum value. `column`: name or callable.
`.max(column, ignore_nan=True)`	`float`	Maximum value. `column`: name or callable.

Lambda expressions are also supported for computed aggregations:

total = txns.sum(lambda data: data["price"] * data["quantity"])

Column Access

# Access column data (returns ModalityEvents)
product_ids = txns["product_id"]
product_ids.events     # np.ndarray of values
product_ids.timestamps # np.ndarray of timestamps

# Timestamps for the data source
txns.timestamps  # np.ndarray of unix timestamps

ModalityEvents.events exposes the fitted modality representation, not necessarily the source database type. A low-cardinality integer inferred as categorical can therefore arrive as strings such as "1" and "2"; normalize comparisons or use column_type_overrides when numeric behavior matters.

Extra Columns

Columns declared in extra_columns are accessible via .extra. To keep a raw source column out of model features, also list it in that source's disallowed_columns:

order_id = txns.extra["order_id"]  # np.ndarray

For a recommendation target that shares a raw extra column with the model, use the fitted alias as the sketch input; this provides ModalityEvents instead of a NumPy array.

Recommendation sketches need a fitted event column

sketch() requires a fitted ModalityEvents item column. It does not accept the raw NumPy array returned by .extra[...]. Do not construct an internal wrapper. When the same source ID is needed as raw target metadata and as a model input, preserve the raw name with extra_columns plus disallowed_columns, add a distinct source sql_lambdas alias, and leave that alias out of allowed_columns. Use the fitted alias as the sketch input.

import numpy as np
from monad.ui.target_function import sketch

future_txns = future["transactions"]
items = future_txns["article_id_model"]  # fitted alias -> ModalityEvents
weights = np.ones(len(items.events), dtype=np.float32)
return sketch(items, weights)

See Foundation Model YAML → Extra Columns for the matching source and data_params configuration.

Filtering

# Filter by column value with lambda
app_txns = txns.filter("channel", lambda x: x == "APP")
expensive = txns.filter("price", lambda x: x > 100)

# Filter using a callable expression
high_value = txns.filter(
    lambda data: data["price"] * data["quantity"],
    lambda x: x > 500,
)

# Filter by exact match (single value or list)
app_txns = txns.where_eq("channel", "APP")
app_web = txns.where_eq("channel", ["APP", "WEB"])

The by parameter accepts a column name (str) or a callable that computes values from the data mapping.

Time Windows

from datetime import timedelta

# Future window: 30 days from split point
future_30d = future.interval_from(ctx[SPLIT_TIMESTAMP], timedelta(days=30))

# SPLIT_TIMESTAMP is numeric Unix seconds, not a datetime.
# Use seconds for timestamp arithmetic; interval durations remain timedeltas.
next_day = ctx[SPLIT_TIMESTAMP] + 86400
future_days_1_to_30 = future.interval_from(next_day, timedelta(days=30))

# History window: [split - 90 days, split), including the lower calendar boundary
recent = history.interval_between(
    start=ctx[SPLIT_TIMESTAMP] - 90 * 86400,
    end=ctx[SPLIT_TIMESTAMP],
    include="start",
)

# Explicit time range
window = history.interval_between(
    start=ctx[SPLIT_TIMESTAMP] - 30 * 86400,
    end=ctx[SPLIT_TIMESTAMP],
    include="start",  # "start" or "end"
)

Note

ctx[SPLIT_TIMESTAMP] is a numeric Unix timestamp in seconds (the runtime can expose an int or float), not a datetime. Add calendar offsets with seconds, such as + 86400 for one day. Pass timedelta only where an interval method or completeness check expects a duration.

interval_from(start, duration) is half-open: it includes start and excludes start + duration. It and interval_between are available on both Events (all sources) and DataSourceEvents (single source). For date-only events when the split day belongs to history, a business window covering calendar days 1 through N inclusive needs timedelta(days=N + 1) so that day N falls before the exclusive end.

Last Basket

Get events from the most recent timestamp:

last_basket = history["transactions"].get_last_basket()

get_last_basket() returns events at the last timestamp, but events with the same timestamp have no guaranteed ordering. If a target needs the first future line, apply an explicit business tie-break to a stable column instead of selecting by array position.

GroupBy

grouped = txns.groupBy("category")
# or multiple columns:
grouped = txns.groupBy(["category", "channel"])

Methods on EventsGroupBy:

Method	Returns	Description
`.count()`	`(np.ndarray, list[str])`	Count per group and group names.
`.sum(target)`	`(np.ndarray, list[str])`	Sum per group and group names. `target`: column name or callable. Optional `ignore_nan=True`.
`.mean(target)`	`(np.ndarray, list[str])`	Mean per group and group names. `target`: column name or callable. Optional `ignore_nan=True`.
`.min(target)`	`(np.ndarray, list[str])`	Minimum per group and group names. `target`: column name or callable. Optional `ignore_nan=True`.
`.max(target)`	`(np.ndarray, list[str])`	Maximum per group and group names. `target`: column name or callable. Optional `ignore_nan=True`.
`.exists(groups=[...])`	`(np.ndarray, list[str])`	Existence flags per group and group names.
`.apply(func, ...)`	`(np.ndarray, list[str])`	Custom aggregation per group.

# Get counts by category
counts, names = txns.groupBy("category").count()

# With explicit group ordering
counts, names = txns.groupBy("category").count(
    groups=["Electronics", "Fashion", "Home"]
)

# Check existence
exists, names = txns.groupBy("category").exists(
    groups=["Electronics", "Fashion", "Home"]
)

# Custom aggregation
medians, names = txns.groupBy("category").apply(
    func=np.median,
    default_value=0.0,
    target="price",
    groups=["Electronics", "Fashion"],
)

`Attributes` API

Access entity attributes by data source name and column:

# Access a specific attribute value
segment = attributes["customers"]["segment"].attribute  # always a str
age = float(attributes["customers"]["age"].attribute)  # always a str; cast numeric values yourself

# Extra columns on attributes
signup_source = attributes["customers"].extra["signup_source"]

Properties of ModalityAttribute:

Property	Type	Description
`.attribute`	`str`	The attribute value, always converted to a string (`"None"` if the entity has no row in that source). Cast numeric values (e.g. `float(...)`) before comparing.
`.column_name`	`str`	Column name.
`.dataset_name`	`str`	Data source name.

Context Dictionary

from monad.batch import SPLIT_TIMESTAMP, MODE, TRAINING_END_TIMESTAMP
from monad.config import DataMode

Key	Type	Description
`SPLIT_TIMESTAMP`	`int` or `float`	Unix timestamp in seconds at the split point dividing history and future.
`MODE`	`DataMode`	Current processing mode: `DataMode.TRAIN`, `DataMode.VALIDATION`, or `DataMode.TEST`. The target function is not called during prediction, so it never sees `DataMode.PREDICT`.
`TRAINING_END_TIMESTAMP`	`float`	Unix timestamp marking the end of the training period.

split_time = ctx[SPLIT_TIMESTAMP]
if ctx[MODE] == DataMode.TRAIN:
    training_end = ctx[TRAINING_END_TIMESTAMP]  # training-specific logic

Target functions during prediction

predict() does not call the target function, because prediction has no future window to build a label from. A DataMode.PREDICT branch in the target function never runs, and returning None does not remove an entity from the prediction file. To limit which entities are scored, pass EntityIds as TestingParams.entity_ids, or filter the data source with where_condition.

Helper Functions

`has_incomplete_training_window()`

Check if the training window is too short for the required target window.

from monad.ui.target_function import has_incomplete_training_window
from datetime import timedelta

if has_incomplete_training_window(ctx, timedelta(days=30)):
    return None  # Skip — can't observe full future window

Parameter	Type	Description
`ctx`	`dict`	The context dictionary.
`required_length`	`timedelta`	Minimum required future window length.

`verify_target()`

Validate a target function against real data before training.

from monad.ui.target_function import verify_target
from monad.ui.module import BinaryClassificationTask

results = verify_target(
    target_fn=my_target_fn,
    fm_checkpoint_path="./foundation_model",
    task=BinaryClassificationTask(),
    num_percentage_entities=1,  # Test on 1% of entities
)

Parameter	Type	Default	Description
`target_fn`	`TargetFunction`	required	The target function to validate.
`fm_checkpoint_path`	`str \\| Path`	required	Foundation model checkpoint path.
`task`	`Task`	required	Task type matching the target function.
`data_params_overrides`	`DataParams \\| None`	`None`	Override data parameters for validation.
`num_percentage_entities`	`int`	`1`	Percentage of entities to evaluate on.
`percentage_nones_allowed`	`int`	`90`	Maximum percentage of `None` returns allowed.
`log_every_n_steps`	`int \\| None`	`None`	Logging frequency.
`limit`	`int \\| None`	`None`	Maximum number of entities to evaluate.

Sketch Functions (Recommendations)

For recommendation tasks, the target function returns a Sketch object representing items and their weights.

from monad.ui.target_function import sketch, sequential_decay, sketch_filtering_mask
from monad.targets.recommendation import time_decay

`sketch()`

Create a sketch from items and weights.

items = future["transactions"]["product_id"]
weights = np.ones(len(items), dtype=np.float32)
return sketch(items, weights)

`sequential_decay()`

Compute position-based decay weights. Earlier events in the future receive higher weight.

weights = sequential_decay(future["transactions"], gamma=0.5)

Parameter	Type	Default	Description
`events`	`DataSourceEvents`	required	The events to compute weights for.
`gamma`	`float`	`0.0`	Decay factor. `0.0` = full decay (only first event matters), `1.0` = uniform weights.
`init_weights`	`ModalityEvents \\| None`	`None`	Initial weights to scale by.

`time_decay()`

Compute time-based decay weights. Each event's weight is daily_decay ** days, where days is the time elapsed since the first event passed in (not since the split point). Before any init_weights are applied, the first event gets weight 1.0 and later events get less when daily_decay < 1.

weights = time_decay(future["transactions"], daily_decay=0.1)

Parameter	Type	Default	Description
`events`	`DataSourceEvents`	required	The events to compute weights for.
`daily_decay`	`float`	`0.0`	Fraction of weight kept per day. `0.1` keeps 10 % after one day and 1 % after two; `1.0` gives every event weight `1.0`. With the default `0.0`, only events at the first timestamp get weight `1.0` and all later events get `0`.
`init_weights`	`ModalityEvents \\| None`	`None`	Per-event weights multiplied into the result. Must be a decimal (float) column; any other type raises `TypeError`.

`sketch_filtering_mask()`

Create a mask that excludes items the entity has already interacted with (e.g., already purchased products) from the training loss. Returning a (sketch, mask) tuple works only with OneHotRecommendationTask; RecommendationTask raises TypeError on a tuple.

# Exclude previously purchased items from the training loss (OneHotRecommendationTask only)
future_txns = future["transactions"]
future_sketch = sketch(future_txns["product_id"], sequential_decay(future_txns, gamma=1.0))
mask = sketch_filtering_mask(history["transactions"]["product_id"])
return (future_sketch, mask)

Complete Examples

Binary Classification — Churn Prediction

import numpy as np
from datetime import timedelta
from monad.targets import Events, Attributes
from monad.batch import SPLIT_TIMESTAMP
from monad.ui.target_function import has_incomplete_training_window

TARGET_WINDOW_DAYS = 30

def churn_target_fn(history: Events, future: Events, attributes: Attributes, ctx: dict):
    if history["transactions"].count() < 2:
        return None
    if has_incomplete_training_window(ctx, timedelta(days=TARGET_WINDOW_DAYS)):
        return None

    future_window = future.interval_from(ctx[SPLIT_TIMESTAMP], timedelta(days=TARGET_WINDOW_DAYS))
    churned = 1 if future_window["transactions"].count() == 0 else 0
    return np.array([churned], dtype=np.float32)

Multiclass Classification — Next Category

import numpy as np
from monad.targets import Events, Attributes

CATEGORIES = ["Electronics", "Fashion", "Home", "Sports", "Beauty"]

def next_category_target(history: Events, future: Events, attributes: Attributes, ctx: dict):
    if history["transactions"].count() < 2:
        return None
    if future["transactions"].count() == 0:
        return None

    first_category = future["transactions"]["category"].events[0]
    if first_category not in CATEGORIES:
        return None
    target = np.zeros(len(CATEGORIES), dtype=np.float32)
    target[CATEGORIES.index(first_category)] = 1.0
    return target

Regression — Customer Lifetime Value

import numpy as np
from datetime import timedelta
from monad.targets import Events, Attributes
from monad.batch import SPLIT_TIMESTAMP
from monad.ui.target_function import has_incomplete_training_window

def ltv_target(history: Events, future: Events, attributes: Attributes, ctx: dict):
    if history["transactions"].count() < 3:
        return None
    if has_incomplete_training_window(ctx, timedelta(days=30)):
        return None

    future_30d = future.interval_from(ctx[SPLIT_TIMESTAMP], timedelta(days=30))
    total_spend = future_30d["transactions"].sum("price")
    return np.array([float(total_spend)], dtype=np.float32)

Recommendation — Product Recommendations

from monad.targets import Events, Attributes
from monad.ui.target_function import sketch, sequential_decay

def products_target(history: Events, future: Events, attributes: Attributes, ctx: dict):
    if history["transactions"].count() == 0:
        return None
    if future["transactions"].count() == 0:
        return None

    future_txns = future["transactions"]
    weights = sequential_decay(future_txns, gamma=0.5)
    return sketch(future_txns["product_id"], weights)

Recommendation — With Filtering Mask

Use with OneHotRecommendationTask; RecommendationTask does not accept a (sketch, mask) tuple.

from monad.targets import Events, Attributes
from monad.ui.target_function import sketch, sequential_decay, sketch_filtering_mask

def filtered_products_target(history: Events, future: Events, attributes: Attributes, ctx: dict):
    if history["transactions"].count() == 0:
        return None
    if future["transactions"].count() == 0:
        return None

    future_txns = future["transactions"]
    weights = sequential_decay(future_txns, gamma=0.5)

    # Exclude previously purchased items from the training loss
    mask = sketch_filtering_mask(history["transactions"]["product_id"])
    return (sketch(future_txns["product_id"], weights), mask)

Recommendation — Train vs Eval Behavior

from monad.targets import Events, Attributes
from monad.batch import MODE
from monad.config import DataMode
from monad.ui.target_function import sketch, sequential_decay

def next_basket_target(history: Events, future: Events, attributes: Attributes, ctx: dict):
    if history["transactions"].count() == 0:
        return None
    if future["transactions"].count() == 0:
        return None

    future_txns = future["transactions"]

    if ctx[MODE] == DataMode.TRAIN:
        # During training: focus on the next basket only
        weights = sequential_decay(future_txns, gamma=0)
    else:
        # During eval: weight all future purchases equally
        weights = sequential_decay(future_txns, gamma=1)

    return sketch(future_txns["product_id"], weights)

Weights do not affect the recommendation metrics

The *AtK metrics (HitRateAtK, NDCGAtK, MeanAveragePrecisionAtK, PrecisionAtK) score every item in the target sketch with binary relevance and ignore its weights, including zero weights. Changing gamma in the evaluation branch therefore changes only the validation loss, not HR, NDCG, MAP, or Precision. Vary gamma by mode only when you want a different loss target during evaluation; if the goal is simply "which items, in any order", uniform weights are equivalent and simpler.