Expose Temporal Leakage in a Recommender

Build a recommender evaluation harness that detects temporal leakage.

Introduction

30 Second Summary

A recommendation score can look convincing when a test quietly borrows information from the future. A product launched on that score can disappoint as soon as real customers behave differently.

In this project, you will build a reproducible evaluation harness for a popularity recommender. You will expose temporal leakage by comparing a mixed-time score with genuinely future behavior.

What You'll Build

Your finished notebook turns an inflated recommendation score into a visible warning that guides a defensible launch decision.

By the end of this project, you'll have:

A reproducible interaction log where the same 2,400-event simulation shows popularity shifting after day 30.
A side-by-side Hit Rate@3 comparison that makes the mixed-time inflation visible against a future holdout.
An automated leakage guard that rejects invalid chronology before a launch decision uses the score.
Secret Mission: Compare rolling cutoff dates to identify when the recommender becomes stale.

Are there any prerequisites?

This project requires a Google account plus basic familiarity with Python lists, dictionaries, loops, and functions. Everything runs in the free-of-charge Google Colab service through your browser, so local Python and GitHub are not required.

Before We Start

Before any notebook work, commit to exposing how mixed-time validation leaks future popularity into training. An inflated Hit Rate@3 can support a bad recommendation launch decision.

Set Up the PulseRank Notebook

PulseRank needs a reproducible place to run each evaluation. A browser notebook keeps each result beside the code that produced it.

Google Colab is a hosted Jupyter Notebook service that requires no local setup. This keeps your first session focused on evaluation reasoning.

A short Python check will prove the runtime is ready.

In this step, get ready to:

Create a Google Colab notebook named PulseRank_Temporal_Evaluation.
Run a Python code cell that prints PulseRank ready.

Open and name the notebook

Colab can store your notebook in Google Drive. A clear name makes the project easy to find later.

Why Google Colab instead of local Python?

A local workflow adds editor setup before the evaluation begins. Colab removes that setup from this project.

Your existing Google account handles access to the workspace.

Open Google Colab in your browser.
Sign in with your Google account if Colab shows the sign-in screen.
Create a new notebook from the Colab landing page.
Click the notebook name in the top-left corner.
Replace the generated name with PulseRank_Temporal_Evaluation.
Press Enter to apply the name.

You should see PulseRank_Temporal_Evaluation at the top of the open notebook.

Good progress. Your notebook now has a stable name you can find later.

Can't create the notebook?

Confirm your browser is signed in to your Google account.
Reload Colab if the page remains on its loading screen.
Help me open a new Google Colab notebook.

Run the environment check

A code cell sends one piece of Python to the runtime. Colab displays the result beneath that same cell.

Select the empty code cell in the notebook.
Enter the environment check by pasting this code:

print("PulseRank ready")

What does this check prove?

The cell asks Python to send a fixed message to the notebook output. Matching output proves the runtime executed the cell successfully.

Before you run the cell, what exact line do you expect to appear underneath it?

Run the cell by clicking the run button on its left edge.

You should see PulseRank ready directly beneath the cell.

That first checkpoint is complete. Your PulseRank notebook can now execute the evaluation code you add next.

Don't see the readiness output?

Confirm the cell contains the readiness line exactly as shown above.
Run the cell again after Colab finishes connecting its runtime.
Refresh the browser tab if the run button stays inactive.
Help me run the PulseRank readiness cell.

Your browser notebook is ready. Next, you will generate a behavior log with a visible popularity shift.

Generate Changing Behavior

Your Google Colab notebook is ready to run PulseRank code. The evaluation now needs behavior that changes as user demand moves.

A recommendation evaluation becomes meaningful when the data reproduces a product challenge. You'll generate a repeatable event log whose popular items shift after day 30.

In this step, get ready to:

Generate 2,400 timestamped interactions across 40 days.
Count the interactions for each item.
Compare the top three items before the demand shift with the top three items after it.

Create the interaction generator

The generator creates a controlled distribution shift inside a repeatable event history. Days 1 through 30 favor one item group. The final 10 days favor another group.

In the open notebook, use the toolbar to add a code cell below the environment check.
Paste the event generator into the new cell:

def build_events():
    events = []
    all_items = list(range(1, 13))
    event_id = 0

    for day in range(1, 41):
        hot_items = [1, 2, 3] if day <= 30 else [10, 11, 12]

        for user_id in range(1, 61):
            event_id += 1

            if (user_id + day) % 5 != 0:
                item_id = hot_items[(user_id + day) % len(hot_items)]
            else:
                item_id = all_items[(user_id * 7 + day) % len(all_items)]

            events.append(
                {
                    "event_id": event_id,
                    "day": day,
                    "user_id": user_id,
                    "item_id": item_id,
                }
            )

    return events

What does this code do?

The events list holds every simulated interaction.
The all_items list creates item IDs from 1 through 12.
The event_id value gives each interaction a unique position in the event sequence.
The hot_items list changes after day 30. This creates the popularity shift.
The inner loop creates 60 user interactions for each day.
The condition sends most interactions to the current hot items.
The remaining pattern spreads interactions across all 12 items.
Each event dictionary records its event ID, day, user ID, and item ID.

Run the generator definition cell.

Before you run the first check, predict the total produced by 40 days of 60 user interactions.

Add a code cell beneath the generator definition.
Build the event list by running this check:

events = build_events()
print("Total events:", len(events))

What does this check prove?

The call to build_events() creates the complete interaction log. The length check confirms that every planned user interaction reached the list.

You'll see Total events: 2400. That first data milestone is in place. Your generator now produces all 2,400 interactions on demand.

Count different from 2,400?

Confirm the day loop uses range(1, 41).
Confirm the user loop uses range(1, 61).
Check that events.append remains inside both loops.
Help me debug the unexpected PulseRank event count.

Inspect the distribution change

The event total proves that the generator created the expected volume. Item frequencies reveal whether the intended demand pattern exists inside that volume.

Add a code cell beneath the event count.
Paste the item-counting helper into the new cell:

def count_items(events):
    counts = {}

    for event in events:
        item_id = event["item_id"]
        counts[item_id] = counts.get(item_id, 0) + 1

    return counts

How does the counter work?

The counts dictionary maps each item ID to its interaction total.
The loop reads the item_id from each event.
The counts.get(item_id, 0) expression starts an unseen item at zero before adding its first interaction.

Run the item-counting definition cell.
Add another code cell beneath it.
Inspect the item totals by running this expression:

count_items(events)

What does this output show?

The displayed dictionary contains item IDs as keys. Each value shows how many interactions selected that item across all 40 days.

You'll see higher totals for the deliberately favored items. The next helper turns those raw counts into a ranked recommendation list.

Missing item counts?

Run the cell that creates events before calling the helper.
Check that return counts aligns with the for loop.
Confirm the dictionary key is spelled item_id.
Help me fix my PulseRank item counter.

A popularity recommender ranks items by interaction count. PulseRank also uses the item ID as a stable tie-breaker so repeated runs return the same order.

Add a code cell beneath the item-counting check.
Paste the ranking helper into the new cell:

def top_k_popular(events, k=3):
    counts = count_items(events)
    ranked_items = sorted(
        counts,
        key=lambda item_id: (-counts[item_id], item_id),
    )
    return ranked_items[:k]

How does the ranking work?

The counts value reuses the item totals from count_items().
The negative count places items with more interactions first.
The item ID places tied items in ascending numerical order.
The [:k] slice returns the requested number of popular items.

Run the popularity-ranking definition cell.

Before you run the final comparison, predict which three item IDs lead each period.

Add a code cell beneath the ranking helper.
Compare the early period with the late period by running this code:

early_events = [event for event in events if event["day"] <= 30]
late_events = [event for event in events if event["day"] > 30]

print("Early-period top items:", top_k_popular(early_events))
print("Late-period top items:", top_k_popular(late_events))

What does this comparison prove?

The early_events list contains interactions from days 1 through 30.
The late_events list contains interactions after day 30.
The two rankings expose the change in demand that future evaluation must respect.

You'll see Early-period top items: [1, 2, 3] on the first line. You'll see Late-period top items: [10, 11, 12] on the second line.

You now have the product behavior PulseRank needs. Historical favorites lose their lead when the late demand regime begins.

✔️ Awesome, I've got everything!

Your notebook now defines all three helpers. The output confirms 2,400 events plus different early and late top-three items.

ⓧ I'd like to double check the full code

The notebook now contains the same logic as this cumulative Python reference.

def build_events():
    events = []
    all_items = list(range(1, 13))
    event_id = 0

    for day in range(1, 41):
        hot_items = [1, 2, 3] if day <= 30 else [10, 11, 12]

        for user_id in range(1, 61):
            event_id += 1

            if (user_id + day) % 5 != 0:
                item_id = hot_items[(user_id + day) % len(hot_items)]
            else:
                item_id = all_items[(user_id * 7 + day) % len(all_items)]

            events.append(
                {
                    "event_id": event_id,
                    "day": day,
                    "user_id": user_id,
                    "item_id": item_id,
                }
            )

    return events


def count_items(events):
    counts = {}

    for event in events:
        item_id = event["item_id"]
        counts[item_id] = counts.get(item_id, 0) + 1

    return counts


def top_k_popular(events, k=3):
    counts = count_items(events)
    ranked_items = sorted(
        counts,
        key=lambda item_id: (-counts[item_id], item_id),
    )
    return ranked_items[:k]

Your event log now captures a controlled demand shift. Next, you'll build the mixed-time baseline that turns this shift into a tempting score.

Build the Mixed-Time Baseline

Your simulated event log now makes the demand shift visible. The next question is whether an evaluation can hide that shift.

A mixed-time split places interactions from all 40 days in the training data. It also places interactions from all 40 days in the test data.

That design gives a popularity baseline access to future behavior before scoring. The resulting data leakage can make a risky launch decision look safe.

In this step, get ready to:

Create a Hit Rate@3 metric for the popularity baseline.
Split all 40 days across mixed-time training data and test data.
Evaluate the top-three recommendations to capture the tempting score.

Measure recommendation hits

Hit Rate@K measures the share of held-out interactions whose item appears in the recommendation list. This baseline uses three recommended items.

Define the metric beneath top_k_popular by pasting this code into a new cell:

def hit_rate_at_k(test_events, recommendations):
    hits = sum(
        1
        for event in test_events
        if event["item_id"] in recommendations
    )
    return hits / len(test_events)

What does this metric do?

The generator counts one hit whenever a test interaction contains an item from recommendations.
Dividing hits by the number of test interactions converts that count into a comparable rate.
A higher rate means the same short recommendation list covers more held-out interactions.

Run the cell.

You should see the cell finish without an exception. The notebook now knows how to calculate Hit Rate@3.

Metric cell does not finish?

Check that the lines inside sum keep their indentation.
Confirm the final return line remains inside hit_rate_at_k.
Ask for help with the function structure: Help me debug my hit_rate_at_k function in Google Colab.

Mix every time period

The mixed-time split uses event order instead of day. Every fifth event enters the test data across the entire 40-day log.

Define the split beneath hit_rate_at_k by pasting this code into a new cell:

def make_mixed_time_split(events):
    train_events = [
        event for event in events if event["event_id"] % 5 != 0
    ]
    test_events = [
        event for event in events if event["event_id"] % 5 == 0
    ]
    return train_events, test_events

What does this split do?

The train_events list keeps events whose identifiers are not divisible by five.
The test_events list receives every fifth event.
Both lists span days 1 through 40. The training data therefore includes interactions from the future demand regime.

Run the cell.

You should see the cell finish without an exception. The mixed-time splitting function is now ready to use.

Split function does not run?

Check that both list comprehensions use event["event_id"].
Confirm the training condition uses != 0.
Confirm the test condition uses == 0.
Ask for help comparing the two conditions: Help me debug my make_mixed_time_split function.

✔️ Awesome, I've got everything!

Great. Confirm that both new function cells have run successfully.

ⓧ I'd like to double check the full code

def build_events():
    events = []
    all_items = list(range(1, 13))
    event_id = 0

    for day in range(1, 41):
        hot_items = [1, 2, 3] if day <= 30 else [10, 11, 12]

        for user_id in range(1, 61):
            event_id += 1

            if (user_id + day) % 5 != 0:
                item_id = hot_items[(user_id + day) % len(hot_items)]
            else:
                item_id = all_items[(user_id * 7 + day) % len(all_items)]

            events.append(
                {
                    "event_id": event_id,
                    "day": day,
                    "user_id": user_id,
                    "item_id": item_id,
                }
            )

    return events


def count_items(events):
    counts = {}

    for event in events:
        item_id = event["item_id"]
        counts[item_id] = counts.get(item_id, 0) + 1

    return counts


def top_k_popular(events, k=3):
    counts = count_items(events)
    ranked_items = sorted(
        counts,
        key=lambda item_id: (-counts[item_id], item_id),
    )
    return ranked_items[:k]


def hit_rate_at_k(test_events, recommendations):
    hits = sum(
        1
        for event in test_events
        if event["item_id"] in recommendations
    )
    return hits / len(test_events)


def make_mixed_time_split(events):
    train_events = [
        event for event in events if event["event_id"] % 5 != 0
    ]
    test_events = [
        event for event in events if event["event_id"] % 5 == 0
    ]
    return train_events, test_events

Record the tempting result

The split and metric can now judge the popularity recommender. Before you run the evaluation, do you expect a score shaped by all 40 days to look persuasive?

Evaluate the mixed-time baseline by running this code in a new cell:

mixed_train, mixed_test = make_mixed_time_split(events)
mixed_recommendations = top_k_popular(mixed_train)
mixed_score = hit_rate_at_k(mixed_test, mixed_recommendations)

print("\nMixed-time recommendations:", mixed_recommendations)
print("Mixed-time Hit Rate@3:", round(mixed_score, 3))

Why does this score look strong?

The training set spans the full event log. Its recommendation list can reflect behavior from periods that would still be in the future at launch time.

The test set also contains many early-period interactions where items 1 through 3 are popular. Those overlapping periods make the score look reassuring.

You should see Mixed-time recommendations: [1, 2, 3] followed by Mixed-time Hit Rate@3: 0.65.

That is a valuable catch: you now have a reproducible benchmark whose persuasive score is not valid for a launch decision.

Mixed-time output missing?

Run the earlier cell that creates events if the notebook runtime no longer contains the event list.
Run the cells that define hit_rate_at_k and make_mixed_time_split before running the evaluation cell.
Ask for help tracing the evaluation variables: Help me debug the mixed-time PulseRank evaluation.

Your mixed-time baseline now makes the leakage trap visible. Next, you will test the same recommender against genuinely future behavior.

Evaluate a Future Holdout

Your mixed-time baseline produced a tempting Hit Rate@3 because future behavior leaked into training. That data leakage makes the score unsuitable for a launch decision.

Temporal validation places every training event before every test event. This recreates the information boundary at launch time.

In this step, get ready to:

Split the events at day 30.
Evaluate the existing popularity recommender on days 31 through 40.
Connect the lower future score to the shift toward items 10 through 12.

Split events on the launch date

A launch-date cutoff creates a clean boundary between known history and unseen behavior. Day 30 represents the final day available when the recommender launches.

In the Google Colab notebook from earlier, add the chronological split below make_mixed_time_split by pasting this function:

def make_temporal_split(events, cutoff_day=30):
    train_events = [
        event for event in events if event["day"] <= cutoff_day
    ]
    test_events = [
        event for event in events if event["day"] > cutoff_day
    ]
    return train_events, test_events

What does this code do?

The cutoff_day=30 parameter sets the simulated launch date.
The train_events list contains every interaction from day 30 or earlier.
The test_events list contains every interaction after day 30.
The return statement packages both event lists for the evaluation.

Run the new code cell by selecting its play icon.

The cell should complete without an exception. This confirms that make_temporal_split is ready to separate launch history from future behavior.

Does the split cell report a problem?

Check that make_temporal_split starts at the left edge of the cell.
Match the indentation inside both list comprehensions.
Run the earlier function cells if the notebook reports an undefined name.
Help me debug my temporal split function.

✔️ Awesome, I've got everything!

Your temporal split function is saved. The notebook can now create a genuine future holdout.

ⓧ I'd like to double check the full code

Compare your cumulative code with this version of pulserank_temporal_evaluation.py.

def build_events():
    events = []
    all_items = list(range(1, 13))
    event_id = 0

    for day in range(1, 41):
        hot_items = [1, 2, 3] if day <= 30 else [10, 11, 12]

        for user_id in range(1, 61):
            event_id += 1

            if (user_id + day) % 5 != 0:
                item_id = hot_items[(user_id + day) % len(hot_items)]
            else:
                item_id = all_items[(user_id * 7 + day) % len(all_items)]

            events.append(
                {
                    "event_id": event_id,
                    "day": day,
                    "user_id": user_id,
                    "item_id": item_id,
                }
            )

    return events


def count_items(events):
    counts = {}

    for event in events:
        item_id = event["item_id"]
        counts[item_id] = counts.get(item_id, 0) + 1

    return counts


def top_k_popular(events, k=3):
    counts = count_items(events)
    ranked_items = sorted(
        counts,
        key=lambda item_id: (-counts[item_id], item_id),
    )
    return ranked_items[:k]


def hit_rate_at_k(test_events, recommendations):
    hits = sum(
        1
        for event in test_events
        if event["item_id"] in recommendations
    )
    return hits / len(test_events)


def make_mixed_time_split(events):
    train_events = [
        event for event in events if event["event_id"] % 5 != 0
    ]
    test_events = [
        event for event in events if event["event_id"] % 5 == 0
    ]
    return train_events, test_events


def make_temporal_split(events, cutoff_day=30):
    train_events = [
        event for event in events if event["day"] <= cutoff_day
    ]
    test_events = [
        event for event in events if event["day"] > cutoff_day
    ]
    return train_events, test_events

Reuse the same model and metric

A fair comparison changes only the split design. The recommender still ranks items by popularity using the same top-three limit.

Add a fresh code cell immediately below the temporal split cell.

Before you run this cell, do you expect the mixed-time score or the future score to be higher?

Evaluate the future holdout by running this code:

temporal_train, temporal_test = make_temporal_split(events)
temporal_recommendations = top_k_popular(temporal_train)
temporal_score = hit_rate_at_k(
    temporal_test,
    temporal_recommendations,
)

print("\nMixed-time recommendations:", mixed_recommendations)
print("Mixed-time Hit Rate@3:", round(mixed_score, 3))
print("Temporal recommendations:", temporal_recommendations)
print("Temporal Hit Rate@3:", round(temporal_score, 3))

What does this evaluation do?

The first line creates temporal_train from days through 30.
The same line creates temporal_test from days after 30.
The top_k_popular call trains the same recommendation baseline on historical events.
The hit_rate_at_k call measures how often those recommendations match future interactions.

Read the four lines printed beneath the cell.
Compare the two score lines.

You should see mixed-time recommendations of [1, 2, 3] with a score of 0.65. You should see temporal recommendations of [1, 2, 3] with a score of 0.05.

That is the core finding: the same recommender falls from an impressive mixed-time result to a weak future result.

Do the comparison lines fail to print?

Run the earlier cells from top to bottom if Python reports an undefined name.
Confirm that events still contains all 2400 interactions.
Check that make_temporal_split uses the day field.
Help me debug my mixed-time and temporal score comparison.

Explain the performance collapse

The historical recommender selects items 1 through 3 because they dominate the first 30 days. The distribution shift moves future demand toward items 10 through 12.

Compare temporal_recommendations with the late-period top items from earlier.
Use the checkpoint below to connect the score collapse to the popularity shift.

Why does the future score collapse?

Training stops at day 30. The popularity baseline therefore ranks the old demand regime highest.

Days 31 through 40 favor a different item group. The chronological result captures the prediction challenge the product faces after launch.

You have exposed how mixed-time leakage can reverse a launch decision. Next up, you'll add a guard that blocks invalid chronological evaluations.

Add a Leakage Guard

Your Google Colab notebook now shows how the mixed-time score hides the future popularity shift. The lower future score gives you a more realistic view of launch performance.

Manual checks make data leakage easy to miss when an evaluation changes. One invalid split could restore the optimistic result without warning.

You will enforce temporal validation in code. You will also convert the difference in Hit Rate@3 into a launch decision.

In this step, get ready to:

Enforce chronological order between training events and test events.
Calculate how much the mixed-time evaluation overstates future performance.
Run the complete evaluation with accepted and blocked split checks.

Enforce chronological order

A temporal-order guard compares the final training day with the first test day. A valid evaluation keeps the training boundary strictly earlier.

Return to the code cell that defines make_temporal_split().
Place your cursor below the make_temporal_split() function.
Add the chronological guard by pasting this code:

def validate_temporal_order(train_events, test_events):
    latest_train_day = max(event["day"] for event in train_events)
    earliest_test_day = min(event["day"] for event in test_events)

    if latest_train_day >= earliest_test_day:
        raise ValueError(
            "Temporal leakage detected: training must end before testing begins."
        )

    return "PASS: all training events occur before all test events."

What does this guard check?

The latest_train_day variable finds the final day represented in training.
The earliest_test_day variable finds the first day represented in testing.
The comparison requires the training period to finish before the test period begins.
The ValueError stops an evaluation that violates this rule.

Run the cell that defines validate_temporal_order().
Check the chronological split by running this code in a new cell:

print(validate_temporal_order(temporal_train, temporal_test))

What should you see?

You should see PASS: all training events occur before all test events. The day 30 cutoff keeps every training event earlier than every test event.

Before you test the mixed-time split, consider whether its training events finish before its test events begin.

Test the mixed-time split by running this code in a new cell:

print(validate_temporal_order(mixed_train, mixed_test))

The failure is intentional

You should see a ValueError stating that temporal leakage was detected. The mixed-time split places overlapping days in training and testing.

This is the guard doing its job. The evaluation now stops before an invalid metric can influence a launch decision.

Does the guard behave differently?

Confirm that latest_train_day uses max() over the training events.
Confirm that earliest_test_day uses min() over the test events.
Check that the comparison uses >=.
Help me debug the temporal-order guard.

Turn metrics into a decision

The score difference measures how much optimism came from the invalid evaluation design. A decision statement makes that gap useful to people choosing whether the recommender is ready to launch.

Return to the cell that contains validate_temporal_order().
Place your cursor below the temporal-order guard.
Add the business interpretation by pasting this code:

def business_interpretation(mixed_score, temporal_score):
    leakage_gap = mixed_score - temporal_score
    return (
        "Mixed-time evaluation overstates future Hit Rate@3 by "
        f"{leakage_gap:.3f}. Use temporal validation for launch decisions."
    )

How does the interpretation work?

The leakage_gap variable subtracts realistic future performance from the optimistic mixed-time result.
The formatted value keeps the difference to three decimal places.
The returned statement identifies temporal validation as the evidence for launch decisions.

Run the cell that defines business_interpretation().
Generate the decision statement by running this code in a new cell:

print(business_interpretation(mixed_score, temporal_score))

What should the decision say?

You should see a sentence that reports the numeric leakage gap. Its final instruction says to use temporal validation for launch decisions.

Seeing an unexpected gap?

Check that mixed_score appears first in the subtraction.
Check that temporal_score appears second in the subtraction.
Rerun the earlier cells that calculate both scores.
Help me fix the PulseRank business interpretation.

Run the complete evaluation

A single main() function makes the experiment reproducible. It rebuilds the events before evaluating both designs with the same recommender and metric.

Place your cursor below business_interpretation().
Add the event setup and both evaluations by pasting this code:

def main():
    events = build_events()
    early_events = [event for event in events if event["day"] <= 30]
    late_events = [event for event in events if event["day"] > 30]

    print("Total events:", len(events))
    print("Early-period top items:", top_k_popular(early_events))
    print("Late-period top items:", top_k_popular(late_events))

    mixed_train, mixed_test = make_mixed_time_split(events)
    mixed_recommendations = top_k_popular(mixed_train)
    mixed_score = hit_rate_at_k(mixed_test, mixed_recommendations)

    temporal_train, temporal_test = make_temporal_split(events)
    temporal_recommendations = top_k_popular(temporal_train)
    temporal_score = hit_rate_at_k(
        temporal_test,
        temporal_recommendations,
    )

    print("\nMixed-time recommendations:", mixed_recommendations)
    print("Mixed-time Hit Rate@3:", round(mixed_score, 3))
    print("Temporal recommendations:", temporal_recommendations)
    print("Temporal Hit Rate@3:", round(temporal_score, 3))

What does the first part of main do?

The function rebuilds the deterministic event log each time it runs.
The early and late lists expose the popularity shift around day 30.
The mixed-time path calculates the optimistic score.
The temporal path calculates future performance with the same model and metric.

Run the cell that defines main().
Call the current version of main() by running this code in a new cell:

main()

What should the evaluation print?

You should see 2400 total events. You should also see different early and late popularity lists.

The output then shows both recommendation lists and both Hit Rate@3 scores. The temporal score remains much lower than the mixed-time score.

Does main stop before printing scores?

Rerun every earlier function cell before calling main().
Check that the body of main() stays indented by four spaces.
Confirm that temporal_score uses temporal_test.
Help me debug the first part of main.

The final part of main() applies the guard to both evaluation designs. A try block catches the intended mixed-time failure so the program can still print its business conclusion.

Return to the cell that defines main().
Place your cursor below print("Temporal Hit Rate@3:", round(temporal_score, 3)).
Add the split checks and business decision by pasting this code:

    print("\nTemporal split check:")
    print(validate_temporal_order(temporal_train, temporal_test))

    print("Mixed-time split check:")
    try:
        print(validate_temporal_order(mixed_train, mixed_test))
    except ValueError as error:
        print("BLOCKED:", error)

    print("\nBusiness decision:")
    print(business_interpretation(mixed_score, temporal_score))

How does main handle the invalid split?

The temporal split check prints the guard result directly because its periods are correctly ordered.
The mixed-time split check runs inside try because its invalid chronology raises an error.
The except block labels that invalid evaluation as BLOCKED.
The final print turns the metric gap into the recommendation for launch evaluation.

Run the updated cell that defines main().
Return to the cell containing the main() call.
Prepare to rerun that cell.

Before you call main(), predict which split the guard accepts.

Run the complete evaluation with this code:

main()

What should the final program prove?

The chronological split prints PASS: all training events occur before all test events..
The mixed-time split prints BLOCKED with the temporal leakage message.
The business decision says to use temporal validation for launch decisions.

Missing PASS or BLOCKED?

Confirm that the temporal check passes temporal_train before temporal_test.
Confirm that the mixed-time check passes mixed_train before mixed_test.
Check that the mixed-time call remains inside the try block.
Help me fix the final PulseRank split checks.

You have turned chronology from a manual check into an enforced rule. PulseRank now bases its launch decision on the future holdout.

✔️ Awesome, I've got everything!

Great. Your notebook now blocks temporal leakage and prints a defensible launch decision.

ⓧ I'd like to double check the full code

def build_events():
    events = []
    all_items = list(range(1, 13))
    event_id = 0

    for day in range(1, 41):
        hot_items = [1, 2, 3] if day <= 30 else [10, 11, 12]

        for user_id in range(1, 61):
            event_id += 1

            if (user_id + day) % 5 != 0:
                item_id = hot_items[(user_id + day) % len(hot_items)]
            else:
                item_id = all_items[(user_id * 7 + day) % len(all_items)]

            events.append(
                {
                    "event_id": event_id,
                    "day": day,
                    "user_id": user_id,
                    "item_id": item_id,
                }
            )

    return events


def count_items(events):
    counts = {}

    for event in events:
        item_id = event["item_id"]
        counts[item_id] = counts.get(item_id, 0) + 1

    return counts


def top_k_popular(events, k=3):
    counts = count_items(events)
    ranked_items = sorted(
        counts,
        key=lambda item_id: (-counts[item_id], item_id),
    )
    return ranked_items[:k]


def hit_rate_at_k(test_events, recommendations):
    hits = sum(
        1
        for event in test_events
        if event["item_id"] in recommendations
    )
    return hits / len(test_events)


def make_mixed_time_split(events):
    train_events = [
        event for event in events if event["event_id"] % 5 != 0
    ]
    test_events = [
        event for event in events if event["event_id"] % 5 == 0
    ]
    return train_events, test_events


def make_temporal_split(events, cutoff_day=30):
    train_events = [
        event for event in events if event["day"] <= cutoff_day
    ]
    test_events = [
        event for event in events if event["day"] > cutoff_day
    ]
    return train_events, test_events


def validate_temporal_order(train_events, test_events):
    latest_train_day = max(event["day"] for event in train_events)
    earliest_test_day = min(event["day"] for event in test_events)

    if latest_train_day >= earliest_test_day:
        raise ValueError(
            "Temporal leakage detected: training must end before testing begins."
        )

    return "PASS: all training events occur before all test events."


def business_interpretation(mixed_score, temporal_score):
    leakage_gap = mixed_score - temporal_score
    return (
        "Mixed-time evaluation overstates future Hit Rate@3 by "
        f"{leakage_gap:.3f}. Use temporal validation for launch decisions."
    )


def main():
    events = build_events()
    early_events = [event for event in events if event["day"] <= 30]
    late_events = [event for event in events if event["day"] > 30]

    print("Total events:", len(events))
    print("Early-period top items:", top_k_popular(early_events))
    print("Late-period top items:", top_k_popular(late_events))

    mixed_train, mixed_test = make_mixed_time_split(events)
    mixed_recommendations = top_k_popular(mixed_train)
    mixed_score = hit_rate_at_k(mixed_test, mixed_recommendations)

    temporal_train, temporal_test = make_temporal_split(events)
    temporal_recommendations = top_k_popular(temporal_train)
    temporal_score = hit_rate_at_k(
        temporal_test,
        temporal_recommendations,
    )

    print("\nMixed-time recommendations:", mixed_recommendations)
    print("Mixed-time Hit Rate@3:", round(mixed_score, 3))
    print("Temporal recommendations:", temporal_recommendations)
    print("Temporal Hit Rate@3:", round(temporal_score, 3))

    print("\nTemporal split check:")
    print(validate_temporal_order(temporal_train, temporal_test))

    print("Mixed-time split check:")
    try:
        print(validate_temporal_order(mixed_train, mixed_test))
    except ValueError as error:
        print("BLOCKED:", error)

    print("\nBusiness decision:")
    print(business_interpretation(mixed_score, temporal_score))


main()

Secret mission

Test Model Freshness Across Time

One future holdout catches leakage at a single launch date. Add rolling five-day evaluations at days 20, 25, and 30 to reveal when shifting demand makes the recommender stale.

Clean Up Your Resources

Your PulseRank notebook uses Google Colab's free-of-charge service. Decide whether to keep the notebook available, close it for now, or delete it from Google Drive.

Resources you used:

The PulseRank_Temporal_Evaluation notebook in Google Drive. It contains the completed pulserank_temporal_evaluation.py program.

Keep everything running

No action needed. Choose this if you are still testing temporal cutoffs or refining your evaluation explanation.

Keep the browser tab containing PulseRank_Temporal_Evaluation open while you continue working.
Keep the notebook in Google Drive so its code plus outputs remain available.
Run main() whenever you want to reproduce the mixed-time comparison or rolling cutoff results.

Pause - I'll come back to this later

Pausing ends your active browser session while keeping the notebook available in Google Drive. Colab deletes its temporary execution virtual machine after inactivity under its managed lifecycle.

Close the browser tab that contains PulseRank_Temporal_Evaluation.
Return to PulseRank_Temporal_Evaluation from Google Drive when you want to continue.
Run main() to regenerate the evaluation output in a new runtime.

Delete - I don't want to use this again

Deleting the notebook removes your saved code plus outputs from Google Drive. Save anything you want to reuse before removing it.

Search Google Drive for PulseRank_Temporal_Evaluation.
Select the notebook in the search results.
Open the file actions menu for the selected notebook.
Choose the option that moves the notebook to the trash.
Open the trash area from the Google Drive sidebar.
Select PulseRank_Temporal_Evaluation in the trash area.
Choose the permanent deletion option from its file actions menu.
Search Google Drive again for PulseRank_Temporal_Evaluation.

Cleanup complete. You should see no result for the PulseRank notebook in Google Drive.

Nice Work!

You pulled it off! You built PulseRank, a runnable recommendation evaluation harness that exposes temporal leakage before it can distort a launch decision.

You've learned how to:

Generate a deterministic interaction log that makes a late popularity shift visible across 2,400 events.
Evaluate a top-three recommendation baseline with Hit Rate@3 across mixed-time versus future-holdout designs.
Block invalid splits with an automated chronology guard. Turn the leakage gap into a defensible launch decision.
Complete the optional Secret Mission by measuring model freshness across rolling temporal cutoffs. Identify the demand-shift window that signals retraining.

Ready to quiz yourself?

Expose Temporal Leakage in a Recommender

Introduction

30 Second Summary

What You'll Build

Are there any prerequisites?

Before We Start

Set Up the PulseRank Notebook

In this step, get ready to:

Open and name the notebook

Why Google Colab instead of local Python?

Can't create the notebook?

Run the environment check

What does this check prove?

Don't see the readiness output?

Generate Changing Behavior

In this step, get ready to:

Create the interaction generator

What does this code do?

What does this check prove?

Count different from 2,400?

Inspect the distribution change

How does the counter work?

What does this output show?

Missing item counts?

How does the ranking work?

What does this comparison prove?

Top items look the same?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Build the Mixed-Time Baseline

In this step, get ready to:

Measure recommendation hits

What does this metric do?

Metric cell does not finish?

Mix every time period

What does this split do?

Split function does not run?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Record the tempting result

Why does this score look strong?

Mixed-time output missing?

Evaluate a Future Holdout

In this step, get ready to:

Split events on the launch date

What does this code do?

Does the split cell report a problem?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Reuse the same model and metric

What does this evaluation do?

Do the comparison lines fail to print?

Explain the performance collapse

Why does the future score collapse?

Add a Leakage Guard

In this step, get ready to:

Enforce chronological order

What does this guard check?

What should you see?

The failure is intentional

Does the guard behave differently?

Turn metrics into a decision

How does the interpretation work?

What should the decision say?

Seeing an unexpected gap?

Run the complete evaluation

What does the first part of main do?

What should the evaluation print?

Does main stop before printing scores?

How does main handle the invalid split?

What should the final program prove?

Missing PASS or BLOCKED?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Secret mission

Test Model Freshness Across Time

Clean Up Your Resources

Clean Up Your Resources

Keep everything running

Pause - I'll come back to this later