Target Function Reference

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

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

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

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

# 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

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

# 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"])

from datetime import timedelta

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

# History window: last 90 days before split
recent = history.interval_from(ctx[SPLIT_TIMESTAMP], timedelta(days=-90))

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

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

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

# 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"],
)

# Access a specific attribute value
segment = attributes["customers"]["segment"].attribute  # str or numeric value
age = attributes["customers"]["age"].attribute

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

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

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

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

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
)

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

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

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

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

# Exclude previously purchased items from recommendations
mask = sketch_filtering_mask(history["transactions"]["product_id"])
future_sketch = sketch(future_items, weights)
return (future_sketch, mask)  # Return tuple for filtered recommendations

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)

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
    return np.array([CATEGORIES.index(first_category)], dtype=np.float32)

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)

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)

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
    mask = sketch_filtering_mask(history["transactions"]["product_id"])
    return (sketch(future_txns["product_id"], weights), mask)

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: consider all future purchases equally
        weights = sequential_decay(future_txns, gamma=0)
    else:
        # During eval: focus on the next purchase only
        weights = sequential_decay(future_txns, gamma=1)

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