Extended Warranty Purchase

Task type: BinaryClassificationTask Industry: Electronics / Retail

Extended warranties are high-margin products that customers are most likely to purchase shortly after buying an expensive item. The challenge is identifying which laptop buyers are receptive to warranty upsells — poorly targeted offers annoy customers, while missed opportunities leave revenue on the table. By predicting warranty purchase likelihood within 7 days of a laptop transaction, sales teams can prioritize follow-up outreach and personalize warranty offers.

What makes this advanced? Cross-event product matching — the target function filters transactions by product type, then matches with service purchase events using both order_id AND sku, requiring a multi-column cross-event join to link the right warranty to the right product.

Prerequisites

Before writing a target function you need:

A trained foundation model built on event data that includes the relevant data sources.
The monad library installed in your environment.
Data source(s): transactions_events, services_events

Target Function

The target function tells monad how to label each entity for training. It receives four arguments:

Argument	Type	Description
`history`	`Events`	All events before the temporal split.
`future`	`Events`	All events after the temporal split.
`attributes`	`Attributes`	Static entity attributes.
`ctx`	`Dict`	Context dictionary containing `SPLIT_TIMESTAMP`, data mode, etc.

The function must return one of:

np.array([1], dtype=np.float32) — positive case
np.array([0], dtype=np.float32) — negative case
None — exclude this entity from training

Full Example

Python

import numpy as np
from datetime import timedelta
from typing import Dict

from monad.ui.target_function import Events, Attributes
from monad.ui.target_function import SPLIT_TIMESTAMP
from monad.ui.target_function import has_incomplete_training_window

from monad.constants import SECONDS_PER_DAY

# === Configuration ===
TARGET_WINDOW_DAYS = 7
TRANSACTIONS_DATA_SOURCE = "transactions_events"
SERVICES_DATA_SOURCE = "services_events"
TARGET_PRODUCT_TYPE = "laptop"
TARGET_SERVICE_TYPE = "warranty.extension"

def extended_warranty_target_fn(
    history: Events,
    future: Events,
    attributes: Attributes,
    ctx: Dict,
) -> np.ndarray | None:
    """Predict if customer buys extended warranty within 7 days of laptop purchase."""

    # 1. Find laptop transactions in the future
    laptop_transactions = future[TRANSACTIONS_DATA_SOURCE].filter(
        "product_type", lambda pt: pt == TARGET_PRODUCT_TYPE
    )
    if not len(laptop_transactions):
        return None

    # 2. Find warranty service purchases
    warranty_services = future[SERVICES_DATA_SOURCE].filter(
        "type", lambda st: st == TARGET_SERVICE_TYPE
    )
    if not len(warranty_services):
        return np.array([0], dtype=np.float32)

    # 3. Match by order_id and sku, check time window
    for order_ts, order_id, sku in zip(
        laptop_transactions.timestamps,
        laptop_transactions["order_id"],
        laptop_transactions["sku"],
    ):
        matching_warranty = warranty_services.filter(
            "order_id", lambda oid: oid == order_id,
        ).filter(
            "sku", lambda s: s == sku,
        )
        if not len(matching_warranty):
            continue

        warranty_ts = matching_warranty.timestamps[0]
        if 0 <= (warranty_ts - order_ts) < TARGET_WINDOW_DAYS * SECONDS_PER_DAY:
            return np.array([1], dtype=np.float32)

    return np.array([0], dtype=np.float32)

Step-by-Step Breakdown

① Filter laptop transactions

Python

laptop_transactions = future[TRANSACTIONS_DATA_SOURCE].filter(
    "product_type", lambda pt: pt == TARGET_PRODUCT_TYPE
)
if not len(laptop_transactions):
    return None

The function first filters future transactions to only laptop purchases. If the customer has no laptop transactions in the future, they are excluded from training (None) — the warranty prediction only applies to laptop buyers.

② Find warranty purchases

Python

warranty_services = future[SERVICES_DATA_SOURCE].filter(
    "type", lambda st: st == TARGET_SERVICE_TYPE
)
if not len(warranty_services):
    return np.array([0], dtype=np.float32)

Next, the services data source is filtered for warranty extension events. If no warranty purchases exist at all, the customer is labeled negative immediately — they bought a laptop but did not buy any warranty.

③ Match by order_id + sku

Python

for order_ts, order_id, sku in zip(
    laptop_transactions.timestamps,
    laptop_transactions["order_id"],
    laptop_transactions["sku"],
):
    matching_warranty = warranty_services.filter(
        "order_id", lambda oid: oid == order_id,
    ).filter(
        "sku", lambda s: s == sku,
    )

This is the core cross-event join. For each laptop transaction, the function searches for a warranty service event with the same order_id AND sku. The double filter ensures the warranty is matched to the correct product — a customer may buy multiple items and only get a warranty for one.

④ Check 7-day window

Python

warranty_ts = matching_warranty.timestamps[0]
if 0 <= (warranty_ts - order_ts) < TARGET_WINDOW_DAYS * SECONDS_PER_DAY:
    return np.array([1], dtype=np.float32)

When a matching warranty is found, the time difference between the laptop purchase and warranty purchase is checked. If the warranty was bought within 7 days, the entity is labeled positive. This captures both same-day warranty additions and follow-up purchases within the critical upsell window.

Training

Once the target function is defined, fine-tune a downstream model:

Python

from pathlib import Path
from monad.ui.config import TrainingParams, MetricParams, MetricMonitoringMode
from monad.config.early_stopping import EarlyStopping

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=extended_warranty_target_fn,
)

training_params = TrainingParams(
    checkpoint_dir=Path("./<this_model>"),
    learning_rate=1e-4,
    epochs=20,
    devices=[0],
    metrics=[
        MetricParams(alias="auroc", metric_name="AUROC", kwargs={"task": "binary"}),
        MetricParams(alias="auprc", metric_name="AveragePrecision", kwargs={"task": "binary"}),
        MetricParams(alias="recall", metric_name="Recall", kwargs={"task": "binary"}),
        MetricParams(alias="precision", metric_name="Precision", kwargs={"task": "binary"}),
    ],
    metric_to_monitor="val_auroc_0",
    metric_monitoring_mode=MetricMonitoringMode.MAX,
    early_stopping=EarlyStopping(min_delta=1e-4, patience=5),
)

module.fit(training_params, seed=42)

Evaluation

Python

from pathlib import Path
from datetime import datetime, timezone
from monad.ui.module import load_from_checkpoint
from monad.ui.config import TestingParams, MetricParams, OutputType

module = load_from_checkpoint(Path("./<this_model>"))

testing_params = TestingParams(
    prediction_date=datetime(2024, 5, 1, tzinfo=timezone.utc),
    output_type=OutputType.DECODED,
    devices=[0],
    metrics=[
        MetricParams(alias="auroc", metric_name="AUROC"),
        MetricParams(alias="auprc", metric_name="AveragePrecision"),
        MetricParams(alias="recall", metric_name="Recall"),
    ],
)

results = module.test(testing_params)

Prediction

Python

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

module = load_from_checkpoint(Path("./<this_model>"))

testing_params = TestingParams(
    local_save_location=Path("./predictions.tsv"),
    output_type=OutputType.DECODED,
    prediction_date=datetime(2024, 6, 1, tzinfo=timezone.utc),
    devices=[0],
)

predictions = module.predict(testing_params)

Recommended Metrics

Metric	Why it matters
AUROC	Measures overall ranking quality.
AUPRC	More informative when the positive class is rare.
Recall	Proportion of actual positives caught.
Precision	Proportion of predicted positives that are correct.
F1 Score	Harmonic mean of precision and recall.

Production Tips

Expand to other high-value product types. The same pattern works for tablets, TVs, or appliances. Parameterise TARGET_PRODUCT_TYPE and train separate models or a multi-product model.
Shorten the window for real-time upsells. If you want to offer the warranty at checkout or within the first email follow-up, consider a 1-3 day window instead of 7 days to focus the model on immediate conversion signals.
Account for bundled warranties. Some purchases include warranties as part of a bundle. Ensure your data pipeline distinguishes between separately purchased warranties and bundled ones to avoid label leakage.
Monitor the warranty attachment rate. Track the baseline warranty attachment rate over time. If it changes significantly (e.g., due to pricing changes), retrain the model to reflect the new purchasing dynamics.
Use the prediction score for offer personalisation. Instead of a binary decision, use the model's probability score to vary the offer — high-probability customers may not need a discount, while low-probability ones might need an incentive.