Skip to content

Product Acquisition Recommendation

Task type: RecommendationTask Industry: Retail / E-commerce

This recipe recommends products that a customer has not previously purchased, focusing on product discovery and repertoire expansion. Unlike the Next Basket recipe which may re-recommend familiar items, this one explicitly filters out repeat purchases.

When to use this: Use acquisition recommendations when your goal is to increase category penetration, introduce customers to new brands, or expand basket diversity. For "buy again" scenarios, use the Next Basket recipe instead.

Prerequisites

Before writing a target function you need:

  • A trained foundation model built on event data that includes a transactions data source with a product ID column (e.g., PROD_ID).
  • The monad library installed in your environment (for Python App).

Target Function

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:

  • A Sketch object — a weighted set of new target items.
  • None — exclude this customer (e.g., no new products in future).

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 = "transactions"
PRODUCT_ID_COLUMN = "PROD_ID"
GAMMA = 0  # 0 = next basket only


def acquisition_target_fn(
    history: Events,
    future: Events,
    attributes: Attributes,
    ctx: Dict,
) -> Sketch:
    """Recommend products the customer has never purchased before."""

    # 1. Identify products already purchased in history
    products_in_history = set(history[TRANSACTION_DATA_SOURCE].modality_events[PRODUCT_ID_COLUMN])

    # 2. Filter future transactions to new products only
    new_product_transactions = future[TRANSACTION_DATA_SOURCE].filter(
        by=PRODUCT_ID_COLUMN,
        condition=lambda x: x not in products_in_history,
    )

    # 3. Exclude customers with no new-product purchases
    if new_product_transactions.count() == 0:
        return None

    # 4. Extract item IDs and compute weights
    new_product_ids = new_product_transactions[PRODUCT_ID_COLUMN]
    training_weights = sequential_decay(new_product_transactions, gamma=GAMMA)

    # 5. Return a sketch of new products
    return sketch(new_product_ids, training_weights)
Python
def acquisition_target_fn(
    history: target_function.Events,
    future: target_function.Events,
    attributes: target_function.Attributes,
    ctx: Dict,
) -> target_function.Sketch:
    """Recommend products the customer has never purchased before."""

    # === Configuration ===
    TRANSACTION_DATA_SOURCE = "transactions"
    PRODUCT_ID_COLUMN = "PROD_ID"
    GAMMA = 0  # 0 = next basket only

    # 1. Identify products already purchased in history
    products_in_history = set(history[TRANSACTION_DATA_SOURCE].modality_events[PRODUCT_ID_COLUMN])

    # 2. Filter future transactions to new products only
    new_product_transactions = future[TRANSACTION_DATA_SOURCE].filter(
        by=PRODUCT_ID_COLUMN,
        condition=lambda x: x not in products_in_history,
    )

    # 3. Exclude customers with no new-product purchases
    if new_product_transactions.count() == 0:
        return None

    # 4. Extract item IDs and compute weights
    new_product_ids = new_product_transactions[PRODUCT_ID_COLUMN]
    training_weights = target_function.sequential_decay(new_product_transactions, gamma=GAMMA)

    # 5. Return a sketch of new products
    return target_function.sketch(new_product_ids, training_weights)

Step-by-Step Breakdown

① Build a set of historically purchased products

Python
products_in_history = set(history[TRANSACTION_DATA_SOURCE].modality_events[PRODUCT_ID_COLUMN])

modality_events returns a mapping of column names to NumPy arrays. We extract the product ID column and convert it to a Python set for fast lookup. This set contains every product the customer has ever bought.

② Filter future transactions to new products

Python
new_product_transactions = future[TRANSACTION_DATA_SOURCE].filter(
    by=PRODUCT_ID_COLUMN,
    condition=lambda x: x not in products_in_history,
)

The filter method keeps only future transactions where the product ID is not in the history set. This ensures the recommendation target contains only genuinely new products.

③ Exclude customers with no new purchases

Python
if new_product_transactions.count() == 0:
    return None

If the customer only repurchases items they already own, there is no acquisition signal. Returning None excludes them.

④ Compute sequential decay weights

Python
training_weights = sequential_decay(new_product_transactions, gamma=GAMMA)
Python
training_weights = target_function.sequential_decay(new_product_transactions, gamma=GAMMA)

With gamma=0, only the first basket of new products gets weight. This targets the very next new-product purchase.

⑤ Return the sketch

Python
return sketch(new_product_ids, training_weights)
Python
return target_function.sketch(new_product_ids, training_weights)

Creates a Sketch of the new product IDs with their computed weights.

Training

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=acquisition_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)

Variations

Category-restricted acquisition

Only recommend new products from specific categories:

Python
TARGET_CATEGORIES = ["Electronics", "Home & Garden"]

new_product_transactions = future[TRANSACTION_DATA_SOURCE].filter(
    by=PRODUCT_ID_COLUMN,
    condition=lambda x: x not in products_in_history,
)
# Further filter to target categories
new_product_transactions = new_product_transactions.filter(
    by="category",
    condition=lambda x: x in TARGET_CATEGORIES,
)
Python
def category_acquisition_target_fn(
    history: target_function.Events,
    future: target_function.Events,
    attributes: target_function.Attributes,
    ctx: Dict,
) -> target_function.Sketch:
    # === Configuration ===
    TRANSACTION_DATA_SOURCE = "transactions"
    PRODUCT_ID_COLUMN = "PROD_ID"
    GAMMA = 0
    TARGET_CATEGORIES = ["Electronics", "Home & Garden"]

    products_in_history = set(history[TRANSACTION_DATA_SOURCE].modality_events[PRODUCT_ID_COLUMN])

    new_product_transactions = future[TRANSACTION_DATA_SOURCE].filter(
        by=PRODUCT_ID_COLUMN,
        condition=lambda x: x not in products_in_history,
    )
    # Further filter to target categories
    new_product_transactions = new_product_transactions.filter(
        by="category",
        condition=lambda x: x in TARGET_CATEGORIES,
    )

    if new_product_transactions.count() == 0:
        return None

    new_product_ids = new_product_transactions[PRODUCT_ID_COLUMN]
    training_weights = target_function.sequential_decay(new_product_transactions, gamma=GAMMA)
    return target_function.sketch(new_product_ids, training_weights)

Broader discovery horizon

Increase gamma to recommend products the customer will discover over a longer period:

Python
GAMMA = 0.3  # Include items from multiple future baskets with decay
Python
def acquisition_broad_target_fn(
    history: target_function.Events,
    future: target_function.Events,
    attributes: target_function.Attributes,
    ctx: Dict,
) -> target_function.Sketch:
    # === Configuration ===
    TRANSACTION_DATA_SOURCE = "transactions"
    PRODUCT_ID_COLUMN = "PROD_ID"
    GAMMA = 0.3  # Include items from multiple future baskets with decay

    products_in_history = set(history[TRANSACTION_DATA_SOURCE].modality_events[PRODUCT_ID_COLUMN])

    new_product_transactions = future[TRANSACTION_DATA_SOURCE].filter(
        by=PRODUCT_ID_COLUMN,
        condition=lambda x: x not in products_in_history,
    )

    if new_product_transactions.count() == 0:
        return None

    new_product_ids = new_product_transactions[PRODUCT_ID_COLUMN]
    training_weights = target_function.sequential_decay(new_product_transactions, gamma=GAMMA)
    return target_function.sketch(new_product_ids, training_weights)
Metric Why it matters
NDCG Ranking quality — are the most relevant new products ranked highest?
Hit Rate Fraction of customers who bought at least one recommended new product.
Recall@K Fraction of actually purchased new products appearing in top-K recommendations.

Production Tips

  1. Position as "Discover something new". Customers respond well to acquisition recommendations when framed as discovery rather than generic suggestions.

  2. Combine with popularity signals. Post-process results to boost trending or new-arrival items for better engagement.

  3. Monitor novelty vs. relevance. Purely novel recommendations may feel random. Balance novelty with affinity (e.g., recommend new products from brands the customer already likes).

  4. Retrain frequently. New products enter the catalog regularly — retrain weekly to include them in recommendations.