Scenario Model API

A scenario model fine-tunes the foundation model for a specific prediction task. Unlike the foundation model (configured via YAML), scenario models are configured entirely via the Python API.

`load_from_foundation_model()`

Creates a scenario model from a trained foundation model checkpoint.

from pathlib import Path
from monad.ui.module import load_from_foundation_model, BinaryClassificationTask

module = load_from_foundation_model(
    checkpoint_path=Path("./foundation_model"),
    downstream_task=BinaryClassificationTask(),
    target_fn=my_target_fn,
)

Parameters

Parameter	Type	Default	Description
`checkpoint_path`	`str \\| Path`	required	Path to the foundation model checkpoint directory.
`downstream_task`	`Task`	required	The prediction task type. See Task Types below.
`target_fn`	`TargetFunction`	required	Target function defining what to predict. See Target Function Reference.
`with_head`	`bool`	`False`	Whether to reuse the foundation model head (advanced).
`pl_logger`	`Logger \\| None`	`None`	PyTorch Lightning logger instance.
`worker_init_fn`	`Callable[[int], None] \\| None`	`None`	Custom worker initialization function for data loading.
`predictions_to_include_fn`	`PredictionsFilteringFnType \\| None`	`None`	Function returning items/classes to include in predictions. Mutually exclusive with `predictions_to_exclude_fn`.
`predictions_to_exclude_fn`	`PredictionsFilteringFnType \\| None`	`None`	Function returning items/classes to exclude from predictions. Mutually exclusive with `predictions_to_include_fn`.
`split`	`TimeSplitOverride \\| EntitySplitOverride \\| None`	`None`	Override the data split from the foundation model checkpoint.
`ignored_modalities`	`dict[str, set[str]] \\| None`	`None`	Per-data-source set of column names whose modalities are always dropped (replaced with the default sketch) for this downstream model. See Ignored Modalities below.
`downstream_head`	`"linear" \\| "prefix"`	`"linear"`	`"linear"` adds a new dense head on top of the foundation model. `"prefix"` (recommendation tasks only) reuses the block of the foundation model head that predicts the next basket of the target column, with no new parameters, and implies `with_head=True`. See Loading Overrides.
`**kwargs`			`DataParams` fields to override (e.g., `data_start_date`, `extra_columns`, `target_sampling_strategy`, `maximum_splitpoints_per_entity`). Any other keyword raises `InvalidConfigurationError` before the checkpoint is read. See Loading Overrides.

Note

predictions_to_include_fn and predictions_to_exclude_fn are only supported for RecommendationTask, MulticlassClassificationTask, and MultilabelClassificationTask.

Ignored Modalities

Marking a modality as ignored suppresses a foundation model feature for this fine-tuned downstream model: its sketch is always replaced with the default (empty) sketch, so the modality contributes nothing to the scenario. Use it to drop a column that is noisy, unavailable at scoring time, or irrelevant to a specific task — without retraining the foundation model.

Pass a mapping from data source name to the set of column names to ignore:

from pathlib import Path
from monad.ui.module import load_from_foundation_model, BinaryClassificationTask

module = load_from_foundation_model(
    checkpoint_path=Path("./foundation_model"),
    downstream_task=BinaryClassificationTask(),
    target_fn=my_target_fn,
    ignored_modalities={
        "transactions": {"discount_code"},
    },
)

Note

This is the scenario-stage counterpart of ignorable_modalities in pretrain(). ignored_modalities drops a modality always (deterministically at the training stage); ignorable_modalities drops it probabilistically during foundation model pretraining as a regularizer.

`load_from_checkpoint()`

Loads a previously trained scenario model for resumed training, testing, or inference.

from pathlib import Path
from monad.ui.module import load_from_checkpoint

module = load_from_checkpoint(
    checkpoint_path=Path("./my_scenario_model"),
)

Parameters

Parameter	Type	Default	Description
`checkpoint_path`	`str \\| Path`	required	Path to the scenario model checkpoint directory.
`pl_logger`	`Logger \\| None`	`None`	PyTorch Lightning logger instance.
`scoring`	`bool`	`False`	Whether loading for inference (`True`) or resumed training (`False`).
`worker_init_fn`	`Callable[[int], None] \\| None`	`None`	Custom worker initialization function.
`split`	`TimeSplitOverride \\| EntitySplitOverride \\| None`	`None`	Override the data split from the checkpoint.
`predictions_to_include_fn`	`PredictionsFilteringFnType \\| None`	`None`	Items/classes to include in predictions.
`predictions_to_exclude_fn`	`PredictionsFilteringFnType \\| None`	`None`	Items/classes to exclude from predictions.
`**kwargs`			`DataParams` fields to override. Any other keyword raises `InvalidConfigurationError` before the checkpoint is read. See Loading Overrides.

Which loading function to use

load_from_foundation_model() — Use when creating a new scenario model from a foundation model checkpoint. The explicit, version-portable checkpoint_path is the inner fm/ subdirectory (for example, output_path/fm/). Monad 1.12 and later also auto-detect this subdirectory when given the outer output directory.
load_from_checkpoint() — Use when loading an already-trained scenario model for test, predict, or interpret. Using load_from_foundation_model() for inference re-initializes the scenario head and discards trained weights.

Task Types

Each scenario model requires a task type that defines the prediction objective.

from monad.ui.module import (
    BinaryClassificationTask,
    MulticlassClassificationTask,
    MultilabelClassificationTask,
    RegressionTask,
    RecommendationTask,
    OneHotRecommendationTask,
)

Task	Constructor	Description
`BinaryClassificationTask()`	No parameters	Predict yes/no outcomes (e.g., churn, conversion).
`MulticlassClassificationTask(class_names=[...])`	`class_names: list[str]`	Predict one of N mutually exclusive classes.
`MultilabelClassificationTask(class_names=[...])`	`class_names: list[str]`	Predict multiple labels simultaneously.
`RegressionTask(num_targets=..., max_value=2000000.0)`	`num_targets: int` (required), `max_value: float`	Predict continuous values (e.g., spend, LTV).
`RecommendationTask()`	No parameters	Rank items for each entity. Default for high-cardinality catalogs.
`OneHotRecommendationTask()`	No parameters	Rank items for each entity. For low-cardinality item sets.

Workflow Example

Complete training and inference pipeline:

from pathlib import Path
import numpy as np
from datetime import datetime, timedelta, timezone

from monad.ui.module import load_from_foundation_model, BinaryClassificationTask
from monad.config import TrainingParams, TestingParams, OutputType, MetricParams
from monad.targets import Events, Attributes
from monad.batch import SPLIT_TIMESTAMP
from monad.ui.target_function import has_incomplete_training_window


# 1. Define target function
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=30)):
        return None
    future_30d = future.interval_from(ctx[SPLIT_TIMESTAMP], timedelta(days=30))
    churned = 1 if future_30d["transactions"].count() == 0 else 0
    return np.array([churned], dtype=np.float32)


# 2. Load foundation model and define task
module = load_from_foundation_model(
    checkpoint_path=Path("./foundation_model"),
    downstream_task=BinaryClassificationTask(),
    target_fn=churn_target_fn,
)

# 3. Train
training_params = TrainingParams(
    epochs=5,
    learning_rate=1e-4,
    devices=[0],
    checkpoint_dir=Path("./churn_model"),
    metrics=[MetricParams(alias="auroc", metric_name="AUROC", kwargs={"task": "binary"})],
)
module.fit(training_params)

# 4. Test on test split (requires test split in data config)
testing_params = TestingParams(
    output_type=OutputType.DECODED,
    devices=[0],
    local_save_location=Path("./predictions.tsv"),
    metrics=[MetricParams(alias="auroc", metric_name="AUROC", kwargs={"task": "binary"})],
)
module.test(testing_params)

# 5. Predict (prediction_date is required)
predict_params = TestingParams(
    output_type=OutputType.DECODED,
    devices=[0],
    prediction_date=datetime(2024, 6, 1, tzinfo=timezone.utc),
    local_save_location=Path("./predictions.tsv"),
)
module.predict(predict_params)

Scenario Model API

load_from_foundation_model()

Parameters

Ignored Modalities

load_from_checkpoint()

Parameters

Task Types

Workflow Example

See Also

`load_from_foundation_model()`

`load_from_checkpoint()`