Skip to content

Recommend Products Never Bought at Full Price

Task type: RecommendationTask Industry: Retail

Discount-sensitive customers represent a distinct behavioral segment. By identifying products a customer has never bought at full price, marketing teams can target them with strategic discounts on those specific products — turning occasional discount buyers into regular customers while protecting margins on products they already buy at full price.

What makes this advanced? groupBy min aggregation — uses groupBy("product_id").min("discount_applied") to find products where the minimum discount > 0, meaning the customer never paid full price.

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): purchases_events with product_id and discount_applied columns

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.

For recommendation tasks, the function must return one of:

  • A Sketch object — a weighted set of target items.
  • None — exclude this entity (e.g., no discount-only products found).

Full Example

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

from monad.ui.target_function import Events, Attributes, Sketch, sketch, sequential_decay
from monad.ui.target_function import SPLIT_TIMESTAMP
from monad.ui.target_function import has_incomplete_training_window


# === Configuration ===
TRANSACTION_DATA_SOURCE = "purchases_events"

def discount_products_recommendation_target_fn(
    history: Events,
    future: Events,
    attributes: Attributes,
    ctx: Dict,
) -> Sketch | None:
    """Recommend products only bought with discounts."""

    # 1. Find minimum discount per product
    min_discount, product_ids = (
        history[TRANSACTION_DATA_SOURCE]
        .groupBy("product_id")
        .min("discount_applied")
    )

    # 2. Filter to products never bought at full price
    target_ids = set(
        pid for pid, md in zip(product_ids, min_discount) if md > 0
    )

    if not target_ids:
        return None

    # 3. Build recommendation from matching transactions
    targets = history[TRANSACTION_DATA_SOURCE].filter(
        "product_id", lambda pid: pid in target_ids
    )

    return sketch(targets["product_id"], sequential_decay(targets, gamma=0))

Step-by-Step Breakdown

① Find minimum discount per product

Python
min_discount, product_ids = (
    history[TRANSACTION_DATA_SOURCE]
    .groupBy("product_id")
    .min("discount_applied")
)

groupBy("product_id").min("discount_applied") groups all historical purchases by product and finds the minimum discount applied to each product. If the minimum discount for a product is 0, the customer bought it at full price at least once. If the minimum discount is > 0, every purchase of that product included a discount.

② Filter to discount-only products

Python
target_ids = set(
    pid for pid, md in zip(product_ids, min_discount) if md > 0
)

A set comprehension filters product IDs to those where the minimum discount exceeds zero. These are products the customer has never paid full price for — prime candidates for targeted discount campaigns.

③ Build the recommendation sketch

Python
targets = history[TRANSACTION_DATA_SOURCE].filter(
    "product_id", lambda pid: pid in target_ids
)

return sketch(targets["product_id"], sequential_decay(targets, gamma=0))

Historical transactions matching the discount-only products are filtered, and a sketch is built with uniform weights. Products that were purchased more frequently at a discount will appear multiple times, implicitly receiving higher weight in the sketch.

④ Handle the empty case

Python
if not target_ids:
    return None

Customers who have bought every product at full price at least once are excluded — there are no discount-only products to recommend.

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, RecommendationTask

module = load_from_foundation_model(
    checkpoint_path=Path("./foundation_model"),
    downstream_task=RecommendationTask(),
    target_fn=discount_products_recommendation_target_fn,
)

training_params = TrainingParams(
    checkpoint_dir=Path("./<this_model>"),
    learning_rate=1e-4,
    epochs=20,
    devices=[0],
    metrics=[
        MetricParams(alias="ndcg", metric_name="NDCGAtK", kwargs={"k": 10}),
        MetricParams(alias="hitrate", metric_name="HitRateAtK", kwargs={"k": 10}),
        MetricParams(alias="recall", metric_name="MultipleTargetsRecall", kwargs={"k": 10}),
    ],
    metric_to_monitor="val_ndcg_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],
    top_k=10,
    metrics=[
        MetricParams(alias="ndcg", metric_name="NDCG"),
        MetricParams(alias="hitrate", metric_name="HitRate"),
        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)
Metric Why it matters
Hit Rate Proportion of users with at least one relevant recommendation.
NDCG Measures ranking quality — rewards relevant items ranked higher.
MAP Mean average precision across all users.
Coverage Fraction of the item catalogue that appears in recommendations.

Production Tips

  1. Distinguish discount types. Not all discounts are equal — loyalty rewards, clearance sales, and promotional coupons have different implications. Consider filtering by discount type if your data supports it.
  2. Set a minimum purchase count. A single discounted purchase of a product is weak evidence. Require 2+ purchases to increase confidence that the customer genuinely prefers the product at a discount.
  3. Combine with margin data. Some products have healthy margins even with discounts. Prioritise recommendations for products where a discount still yields acceptable profit.
  4. A/B test discount levels. Use the model to identify discount-sensitive products per customer, then test different discount levels (5%, 10%, 15%) to find the minimum discount that drives conversion.