New Arrival Recommendations

Rank recently introduced products for each customer when you want a new arrivals shelf. The target is a product the customer may purchase in the next 30 days. This example selects products introduced in the previous 14 days with at least one recorded purchase from a general purchase ranking. It assumes those products were represented in the graph when the model was fitted. It does not establish recommendations for products added after fitting.

Data and relationships

The example assumes a prepared Kumo Engine graph, a matching snapshot, and these tables:

Table	Required columns	Purpose
`users`	`user_id` as the primary key	Defines the customers to score.
`items`	`item_id` as the primary key, `introduced_at`	Defines catalog products and when each was introduced.
`transactions`	`user_id`, `item_id`, `timestamp`	Records purchases and their times.

Connect transactions.user_id to users.user_id and transactions.item_id to items.item_id. Mark transactions.timestamp as the time column. The items and transactions pandas DataFrames used below must come from the same extracts as the snapshot. Interpret their times consistently as UTC timestamps without timezone offsets.

Rank purchases and select new arrivals

This Python query ranks items each user may purchase over the next 30 days. LIST_DISTINCT treats repeat purchases of an item as one target item within a window. The query does not constrain product introduction dates; the catalog filter does that after ranking.

from datetime import datetime, timedelta
from kumo.engine import PredictiveQuery
query_text = (
    "PREDICT LIST_DISTINCT(transactions.item_id, 0, 30, days) "
    "RANK TOP 50 FOR EACH users.user_id"
)
query = PredictiveQuery(graph, query_text)
query.validate()
print(query.get_task_type())  # temporal_link_prediction
prediction_time = datetime(2024, 5, 1)  # UTC in the source extracts.
prediction_table = query.generate_prediction_table(
    snapshot, prediction_time=prediction_time
)
ranked = trainer.predict(
    prediction_table, snapshot=snapshot, num_classes_to_return=50
).frame

Here trainer is a fitted model for this query. Replace the example prediction time with a UTC time after the latest observed purchase. The prediction table selects users and the time; it does not fit a model. In ranked, ENTITY identifies a user, CLASS an item, and SCORE orders that user’s returned candidates. A score does not guarantee a purchase.

Use the catalog introduction time at the same prediction anchor to keep products introduced during the previous 14 days. The interval includes both the cutoff and prediction time. Keep only products with a recorded purchase by that time. This does not require a particular customer to have purchased the product before.

cutoff = prediction_time - timedelta(days=14)
recent_items = items.loc[
    items["introduced_at"].ge(cutoff)
    & items["introduced_at"].le(prediction_time),
    ["item_id", "introduced_at"],
]
observed_items = transactions.loc[
    transactions["timestamp"].le(prediction_time), ["item_id"]
].drop_duplicates()
recent_items = recent_items.merge(observed_items, on="item_id", how="inner")
new_arrivals = (
    ranked.merge(recent_items, left_on="CLASS", right_on="item_id", how="inner")
    .sort_values(["ENTITY", "SCORE"], ascending=[True, False])
    [["ENTITY", "CLASS", "SCORE"]]
)

Each remaining row is a returned product candidate introduced within that interval, with recorded purchase history, ordered for a customer by score. You can use these rows to order a new arrivals shelf after applying your catalog’s separate saleability rules. The shown filter checks introduction time and purchase history only. The top 50 general purchase candidates are chosen before the catalog filter, so some customers may have no new arrival in the result. This example does not establish scoring for an item with no purchase history or one added after fitting.