⚠️
Check This First!
This article refers to BaseModel accessed via Docker container. Please refer to Snowflake Native App section if you are using BaseModel as SF GUI application.

In this article we show example configuration for all scenario model tasks.

Binary Classification

Below you can see all the elements of the scenario training script in an example of binary classification.

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)

    if history["product.buy"].count() == 0:
        return None

    # churn definition:
    # 1 - churned
    # 0 - not churned
    
    churn = 0 if future["product.buy"].count() > 0 else 1

    # output should be a float32 numpy array
    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 & training the model
if __name__ == "__main__":
    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, 
    )
    trainer.fit(training_params=training_params, overwrite=True)

Once ready, you can save the training script as *.py file and run it from your Python console.

Multiclass Classification

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

import numpy as np


# specify targets
TARGET_ENTITY = "department_name"
TARGET_NAMES = ["Denim Trousers", "Swimwear", "Trousers"]
# specifying the target
def target_fn(history: Events, future: Events, attribites: Attributes, ctx: Dict) -> np.ndarray:
    # trim the future to the desired target window
    target_window_days = 21
    if has_incomplete_training_window(ctx, target_window_days):
        return None
    future = next_n_days(future, ctx[SPLIT_TIMESTAMP], target_window_days)

    purchase_target, _ = (
      future["transactions"]
      .groupBy(TARGET_ENTITY)
      .count(normalize=True, groups=TARGET_NAMES)
    )
    
    return purchase_target

# 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 & training the model
trainer = load_from_foundation_model(
    checkpoint_path=fm_path, # location of foundation model
    downstream_task=MulticlassClassificationTask(class_names=TARGET_NAMES), # task aligned with business scenario
    target_fn=target_fn, 
)
trainer.fit(training_params=training_params, overwrite=True)

Multi-Label Classification

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

import numpy as np


    
# specify targets
TARGET_ENTITY = "department_name"
TARGET_NAMES = [
    "Denim Trousers",
    "Swimwear",
    "Trousers",
    "Jersey Basic",
    "Ladies Sport Bottoms",
]
# specifying the target
def target_fn(history: Events, future: Events, attribuites: Attributes, ctx: Dict) -> np.ndarray:

    # trim the future to the desired target window
    target_window_days = 21
    if has_incomplete_training_window(ctx, target_window_days):
        return None
    future = next_n_days(future, ctx[SPLIT_TIMESTAMP], target_window_days)

    # apply logic
    purchase_target, _ = (
      future["transactions"]
      .groupBy(TARGET_ENTITY)
      .exists(groups=TARGET_NAMES)
    )
    
    if purchase_target.sum() == 0:
        return None
    
    return purchase_target

# 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 & training the model
trainer = load_from_foundation_model(
    checkpoint_path=fm_path, # location of foundation model
    downstream_task=MultilabelClassificationTask(class_names=TARGET_NAMES), # task aligned with business scenario
    target_fn=target_fn, 
)
trainer.fit(training_params=training_params, overwrite=True)

Regression

from typing import Dict
from monad.ui.config import TrainingParams
from monad.ui.module import load_from_foundation_model, RegressionTask
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, attribuites: Attributes, ctx: Dict) -> np.ndarray:

    # trim the future to the desired target window
    target_window_days = 21
    if has_incomplete_training_window(ctx, target_window_days):
        return None
    future = next_n_days(future, ctx[SPLIT_TIMESTAMP], target_window_days)
		
    # apply the transformation reflecting target value logic - total amount spent
    spent = future["transactions"]["price"].events.sum()
		
    # return the result as float32 array
    return np.array([spent], 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 & training the model
trainer = load_from_foundation_model(
    checkpoint_path=fm_path, # location of foundation model
    downstream_task=RegressionTask(num_targets=1, max_value=1000), # task aligned with business scenario
    target_fn=target_fn, 
)
trainer.fit(training_params=training_params, overwrite=True)

Recommendation

from typing import Dict
from monad.ui.config import TrainingParams
from monad.ui.module import load_from_foundation_model, RecommendationTask
from monad.ui.target_function import Attributes, Events, sequential_decay, sketch, Sketch

import numpy as np


# specifying the target
def target_fn(history: Events, future: Events, attribuites: Attributes, ctx: Dict) -> Sketch:        
    # transformations
    future_transactions = future["transactions"]
    article_ids = future["transactions"]["article_id"]
    training_weights = sequential_decay(future_transactions, gamma=0)
    
    # output - a SKETCH
    return sketch(article_ids, training_weights)

# 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 & training the model
trainer = load_from_foundation_model(
    checkpoint_path=fm_path, # location of foundation model
    downstream_task=RecommendationTask(), # task aligned with business scenario
    target_fn=target_fn, 
)
trainer.fit(training_params=training_params, overwrite=True)

One-Hot Recommendation

from typing import Dict
from monad.ui.config import TrainingParams
from monad.ui.module import load_from_foundation_model, OneHotRecommendationTask
from monad.ui.target_function import Attributes, Events, sequential_decay, sketch, Sketch

import numpy as np


# specifying the target
def target_fn(history: Events, future: Events, attribuites: Attributes, ctx: Dict) -> Sketch:        
    # transformations
    future_transactions = future["transactions"]
    article_ids = future["transactions"]["brand"]
    training_weights = sequential_decay(future_transactions, gamma=0)
    
    # output - a SKETCH
    return sketch(article_ids, training_weights)

# 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 & training the model
trainer = load_from_foundation_model(
    checkpoint_path=fm_path, # location of foundation model
    downstream_task=OneHotRecommendationTask(), # task aligned with business scenario
    target_fn=target_fn, 
)
trainer.fit(training_params=training_params, overwrite=True)