API Reference
Start typing to search…

MonadModule

class monad.ui.module.MonadModule

Trainable module.

from typing import Dict
from monad.ui.config import TrainingParams
from monad.ui.module import load_from_foundation_model, BinaryClassificationTask
from monad.ui.target_function import Attributes, Events, has_incomplete_training_window, next_n_days, SPLIT_TIMESTAMP

import numpy as np

# specifying the target
def target_fn(history: Events, future: Events, _entity: Attributes, _ctx: Dict) -> np.ndarray:
    target_window_days = 28
    if has_incomplete_training_window(_ctx, target_window_days):
        return None
    future = next_n_days(future, _ctx[SPLIT_TIMESTAMP], target_window_days)
    churn = 0 if future["product.buy"].count() > 0 else 1
    return np.array([churn], dtype=np.float32) 

# selecting the foundation model
fm_path = "/path/to/your/models/pretrain/fm" # parameter for: load_from_foundation_model
checkpoint_dir = "path/to/your/models/downstream/model_name" # parameter for: training_params

# adapting training parameters
training_params = TrainingParams(
  checkpoint_dir=checkpoint_dir, # location to save your scenario model
  epochs=1,
  learning_rate=0.0001,
  devices=[0],
)

# instantiating the trainer (MonadModule object)
trainer = load_from_foundation_model(
  checkpoint_path=fm_path, # location of foundation model
  downstream_task=BinaryClassificationTask(), # task aligned with business scenario
  target_fn=target_fn, 
)
# training the model
trainer.fit(training_params=training_params)

Methods

MonadModule.fit(self, training_params, resume=False, overwrite=False, seed=None)

Trains scenario model.

...
# instantiating the trainer (MonadModule object)
trainer = load_from_foundation_model(
  checkpoint_path=...,
  downstream_task=...,
  target_fn=...,
)
# training the model
trainer.fit(training_params=...)
Parameters

training_params: TrainingParams
Parameters defining the training.

resume : bool
Default: False
If True, training will be resumed from the last checkpoint if such exists, an error will be thrown otherwise.

overwrite: bool
Default: False
If True, any previous training results will be overwritten. Otherwise, if resume is not set and checkpoints from previous training are present, error will be raised.

seed: Optional[int]
Default: None
Seed for the training, when provided, ensures reproducibility of the results.

Returns

Saves scenario model under the path defined in TrainingParams.

MonadModule.evaluate(self, testing_params, seed=None)

Evaluates scenario model.

...
# instantiating the trainer (MonadModule object)
trainer = load_from_checkpoint(
  checkpoint_path=...,
  downstream_task=...,
  target_fn=...,
)
# training the model
trainer.evaluate(testing_params=...)
Parameters

testing_params: TestingParams
Parameters defining the evaluation.

seed: Optional[int]
Default: None
Seed for the training, when provided, ensures reproducibility of the results.

Returns

Metrics computed for evaluation

MonadModule.test(self, testing_params, seed=None)

Generates model predictions, ground truth and evaluates scenario model.

from monad.ui.config import OutputType, TestingParams
from monad.ui.module import load_from_checkpoint

# instantiating the trainer (MonadModule object)
trainer = load_from_checkpoint(
  checkpoint_path=...,
  downstream_task=...,
  target_fn=...,
)
testing_params = TestingParams(
  local_save_location = ...,
  output_type = OutputType.DECODED,
)
# training the model
trainer.test(testing_params=testing_params)
Parameters

testing_params: TestingParams
Parameters defining the evaluation.

seed: Optional[int]
Default: None
Seed for the training, when provided, ensures reproducibility of the results.

Returns

Return test metrics and saves a tsv file with predictions and ground truths under the path provided via TestingParams.

MonadModule.predict(self, testing_params, seed=None)

Generates model predictions for scenario model.

from datetime import datetime
from monad.ui.config import OutputType, TestingParams
from monad.ui.module import load_from_checkpoint

# instantiating the trainer (MonadModule object)
trainer = load_from_checkpoint(
  checkpoint_path=...,
  downstream_task=...,
  target_fn=...,
)
testing_params = TestingParams(
  local_save_location = ...,
  output_type = OutputType.DECODED,
	prediction_date = datetime(2025, 10, 1)
)
# training the model
trainer.test(testing_params=testing_params)
Parameters

testing_params: TestingParams
Parameters defining the evaluation.

seed: Optional[int]
Default: None
Seed for the training, when provided, ensures reproducibility of the results.

Returns

Return saves a tsv file with predictions under the path and prediction date provided via TestingParams.