Build a Music Discovery Engine

Introduction

30 Second Summary

Music apps often serve the same trending songs to listeners with completely different tastes. A generic playlist quickly makes discovery feel repetitive.

In this project, you will build a deterministic local recommendation system that returns a personalized 30-song playlist through FastAPI. A leakage-safe offline evaluation will reveal whether hybrid ranking improves on a popularity baseline.

What You'll Build

You will demo a local API that reshapes a personalized 30-song discovery playlist after each play, skip, or save event.

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

• A reproducible benchmark that regenerates the same 120-track catalog while keeping every future listening event out of training.
• A two-stage discovery API whose 30-song slate reflects listener history plus play, skip, or save feedback.
• An evidence report that compares popularity with hybrid ranking across Precision@10, Recall@10, catalog_coverage, and diversity.
• Secret Mission: A recency-aware hybrid model that reproduces stale taste failure before correcting it on the unchanged future holdout.

Are there any prerequisites?

You should be comfortable with advanced Python, FastAPI, recommender-system fundamentals, and backend testing. The setup checks Python 3.12 or later before installing every pinned dependency.

Before We Start

Before the hands-on work begins, define the engineering claim that your finished recommendation benchmark must support. You are building a deterministic local music-discovery service whose leakage-safe temporal benchmark exposes relevance, coverage, diversity, cold-start behavior, popularity bias, and taste drift while testing hybrid ranking plus recency-aware ranking against a popularity-only baseline on one immutable future holdout.

Set Up the Local Recommendation Service

A recommendation benchmark is only credible when every result comes from a known runtime. An unexpected Python version or dependency upgrade can change the scores before you change the model.

This step creates an isolated virtual environment with pinned versions of FastAPI, Uvicorn, NumPy, pytest, plus HTTPX. You will also prepare the project structure in VS Code.

In this step, get ready to:

• Verify that Python 3.12 or later is available.
• Create the local project scaffold.
• Verify all pinned dependencies.

Verify Python 3.12 or later

NumPy 2.5.4 requires Python 3.12 or later. Confirming the runtime first prevents an incompatible dependency installation.

• Press Cmd+Space to open Spotlight.
• Type Terminal into Spotlight.
• Press Enter to open Terminal.

Before you run the check, do you expect your installed version to meet the 3.12 minimum?

• Check the installed Python version by running this command:

python3 --version

What does this check prove?

The command reports the Python interpreter available through python3. The version determines whether the pinned NumPy build can be installed.

✔️ I see Python 3.12 or higher

Your runtime meets the project requirement. You can build the isolated environment without changing your system installation.

• Continue to the workspace setup below.

ⓧ I see an older version

The existing interpreter cannot install the pinned NumPy release. A current universal2 macOS build provides the required runtime on an Intel-based Mac.

• Open the Python.org macOS downloads page.
• Download a current universal2 macOS installer.
• Run the downloaded installer.
• Close the existing Terminal window.
• Open a new Terminal window through Spotlight.
• Check the new Python version by running this command:

python3 --version

The reported version now needs to be 3.12 or later.

ⓧ Command not found

Your shell cannot find a Python 3 installation. Install a current universal2 macOS build before creating the project environment.

• Open the Python.org macOS downloads page.
• Download a current universal2 macOS installer.
• Run the downloaded installer.
• Close the existing Terminal window.
• Open a new Terminal window through Spotlight.
• Check the installed version by running this command:

python3 --version

You should now see Python 3.12 or later.

Still seeing the previous version?

Close every Terminal window after the installer finishes. A new shell session reloads the command search path.

Use the interpreter reported by the new python3 command when you create the virtual environment.

Help me resolve my Python version check.

Create the project workspace

The project needs one named folder that keeps its source code, tests, dependency manifest, plus isolated interpreter together. A predictable structure also makes every later command reproducible.

• Return to the Terminal window from the version check.
• Move to your Desktop by running this command:

cd ~/Desktop

Why start on the Desktop?

The ~ shortcut identifies your home folder. Starting from ~/Desktop gives the project a location you can find quickly.

• Build the complete folder scaffold by running these commands:

mkdir music-discovery-engine
cd music-discovery-engine
mkdir app tests
ls

What does this scaffold create?

• The music-discovery-engine folder contains every project artifact.
• The app folder holds the recommendation service.
• The tests folder holds the regression suite.
• The final ls command lists the new directories.

You should see app plus tests in the output.

Folder setup did not complete?

Check that the Terminal prompt points to your Desktop before creating the project folder. If a folder with the same name already exists, inspect it before reusing or removing it.

Help me diagnose my project scaffold.

A virtual environment keeps these project dependencies separate from packages used by other Python applications. Activating .venv also makes its local interpreter the default for the current terminal session.

• Prepare the isolated environment by running these commands:

python3 -m venv .venv
source .venv/bin/activate

What do these commands do?

• The first command creates the .venv environment inside music-discovery-engine.
• The second command activates that environment for the current shell.

Your Terminal prompt should now include .venv. That prefix confirms the environment is active.

Virtual environment not active?

Confirm that your prompt points to the music-discovery-engine folder. Run the activation command again from that folder if the prompt lacks the .venv prefix.

Help me activate my virtual environment.

• Open the current project folder in VS Code by running this command:

code .

What does this command open?

The code . command opens the current folder as a VS Code workspace. The . represents the folder your Terminal currently uses.

You should see music-discovery-engine at the top of the Explorer view. Its contents should include app plus tests.

VS Code did not open?

The VS Code shell command may be missing from your command search path. You can install it from the Command Palette.

• Press Cmd+Shift+P in VS Code.
• Run Shell Command: Install 'code' and optionally 'copilot' commands in PATH.
• Choose the option that installs only code if the optional Copilot CLI prompt appears.
• Restart Terminal after the command finishes.
• Return to the music-discovery-engine folder.
• Run the opening command again.

Help me open the project in VS Code.

Install and verify the dependencies

A pinned requirements file gives every benchmark run the same library builds. The empty app/__init__.py file makes app an importable Python package.

• Open the integrated terminal from the View menu in VS Code.
• Select Terminal from the menu.
• Create the two starter files by running these commands:

touch requirements.txt app/__init__.py
ls app

Why create an empty package file?

The touch command creates the two empty files. Listing app confirms that __init__.py exists before any service modules depend on it.

You should see __init__.py in the terminal output. The file remains empty in the finished artifact.

• Select requirements.txt in the Explorer view.
• Paste the exact dependency pins below into the file:

fastapi==0.162.0
uvicorn==0.54.0
numpy==2.5.4
pytest==9.2.0
httpx==0.28.1

What do these dependencies provide?

• FastAPI defines the local recommendation API.
• Uvicorn serves that API through an ASGI runtime.
• NumPy supplies vector operations for similarity scoring.
• pytest runs the regression suite.
• HTTPX supports FastAPI endpoint tests through its test client dependency.

• Save requirements.txt.
• Install the pinned dependencies by running:

pip install -r requirements.txt

What does this installation use?

The command reads every exact version from requirements.txt. Because .venv is active, the packages stay inside the project environment.

The installation output should finish without a dependency error. Your prompt should still include .venv.

Dependency installation failed?

Confirm that the terminal prompt includes .venv. Check that requirements.txt contains the five pins exactly as shown.

A NumPy compatibility failure usually means the active environment was created with a Python version earlier than 3.12. Recreate .venv with the verified interpreter.

Help me debug my pinned dependency installation.

• Compare your starter artifacts with the complete reference below.

✔️ Awesome, I've got everything!

Your dependency manifest is saved. Your app package marker is empty.

ⓧ I'd like to double check the full code

Your requirements.txt file should match this reference exactly.

fastapi==0.162.0
uvicorn==0.54.0
numpy==2.5.4
pytest==9.2.0
httpx==0.28.1

These five lines preserve the exact dependency versions used by the project.

Your app/__init__.py file should contain no text.

Before you run the final check, do you expect both imports to complete without an error?

• Verify the runtime plus installed packages by running these commands:

python3 --version
python -c "import fastapi, numpy, pytest, httpx; print(fastapi.__version__, numpy.__version__)"

What should the verification prove?

The first line confirms that the active runtime is Python 3.12 or later. The second line imports the required test packages before printing the FastAPI plus NumPy versions.

You should see a Python version of 3.12 or later. The package version line should begin with 0.162.0 2.5.4.

Imports or versions do not match?

Check that the terminal prompt still includes .venv. Reactivate the environment if VS Code opened a new shell without it.

Run the installation command again after confirming that requirements.txt matches the reference. A successful reinstall restores the pinned package set.

Help me resolve my environment verification result.

Your reproducible runtime is ready. The service now has a verified interpreter, isolated dependencies, plus an importable package structure.

Next, you will generate the deterministic music data and expose the popularity-only baseline under one global temporal split.

Generate Data and Expose Popularity Bias

Your isolated Python environment is ready. You can now build a repeatable recommendation benchmark without external data or API calls.

A recommendation benchmark can appear successful while learning from future behavior. One global temporal cutoff prevents that data leakage before the popularity failure is measured.

In this step, get ready to:

• Generate a deterministic catalog with timestamped listener events.
• Separate training events from one immutable future holdout.
• Run a popularity baseline that exposes trend-dominated recommendations.

Generate the catalog and event stream

The synthetic stream uses implicit feedback from plays, skips, and saves. A fixed seed makes every generated track and event reproducible.

• In the VS Code file sidebar, use the new-file control beside app to create core.py.
• Build the event model in app/core.py by pasting this code:

from __future__ import annotations

from dataclasses import dataclass
from random import Random

import numpy as np

EVENT_WEIGHT = {"play": 1.0, "skip": -2.0, "save": 3.0}
CUTOFF_DAY = 19


@dataclass(frozen=True)
class Track:
    id: str
    title: str
    artist: str
    genre: str
    mood: str
    energy: float


@dataclass(frozen=True)
class Event:
    user_id: str
    track_id: str
    kind: str
    day: int

What does this foundation define?

• Frozen dataclasses give each track and event a stable value-based identity.
• EVENT_WEIGHT makes saves the strongest positive signal. It makes skips a negative signal.
• CUTOFF_DAY establishes day 19 as the simulated deployment boundary.
• NumPy is imported for deterministic score ordering in the baseline.

• Add the dataset scaffold below the Event class by pasting this code:

def build_dataset(seed: int = 7) -> tuple[list[Track], list[Event], tuple[str, ...]]:
    rng = Random(seed)
    genres = ("ambient", "jazz", "rock", "electronic", "folk", "soul")
    moods = ("calm", "bright", "dark", "driving")
    tracks: list[Track] = []
    by_genre = {genre: [track for track in tracks if track.genre == genre] for genre in genres}
    events: list[Event] = []
    evaluation_users = tuple(f"u{number:02d}" for number in range(18))
    return tracks, events, evaluation_users

What does the scaffold establish?

• Random(seed) creates the independent seeded generator behind the synthetic catalog.
• genres and moods define the categorical space used by every track.
• evaluation_users creates listener IDs from u00 through u17.

• Save app/core.py.
• Check that the model definitions parse by running this command in the activated terminal from earlier:

python -m app.core

What does this check prove?

The command imports and executes app/core.py as a module. Returning to the terminal prompt without a traceback proves that the current definitions are valid Python.

Seeing a Python traceback?

• Check that core.py sits inside the existing app directory.
• Compare the indentation of the Track and Event fields with the code above.

Ask for help with the traceback: Diagnose my app.core Python traceback and identify the exact line I should correct.

• Insert the catalog-generation loop directly below tracks: list[Track] = [] by pasting this code:

    for genre_index, genre in enumerate(genres):
        for number in range(20):
            tracks.append(
                Track(
                    id=f"t{genre_index:02d}{number:02d}",
                    title=f"{genre.title()} Signal {number + 1}",
                    artist=f"{genre.title()} Artist {number // 2 + 1}",
                    genre=genre,
                    mood=moods[(number + genre_index) % len(moods)],
                    energy=round(0.2 + 0.6 * rng.random(), 2),
                )
            )

How is the catalog generated?

• The nested loops create 20 tracks for each of the six genres.
• Track IDs encode the genre index plus the track number. This keeps ordering deterministic.
• The seeded generator assigns reproducible energy values between the project bounds.

• Insert the listener and trend event loops directly above the return tracks, events, evaluation_users line by pasting this code:

    for user_number, user_id in enumerate(evaluation_users):
        old_genre = genres[user_number % len(genres)]
        current_genre = genres[(user_number + 1) % len(genres)]
        noise_genre = genres[(user_number + 3) % len(genres)]
        for offset, track in enumerate(by_genre[old_genre][:10], start=1):
            kind = "save" if offset <= 6 else "play"
            events.append(Event(user_id, track.id, kind, offset))
        events.append(Event(user_id, by_genre[noise_genre][0].id, "skip", 11))
        for offset, track in enumerate(by_genre[current_genre][:4], start=13):
            events.append(Event(user_id, track.id, "save", offset))
        for offset, track in enumerate(by_genre[current_genre][4:], start=20):
            events.append(Event(user_id, track.id, "save", offset))

    for trend_number in range(12):
        trend_user = f"trend{trend_number:02d}"
        for track in by_genre["rock"]:
            events.append(Event(trend_user, track.id, "save", 12))

How does the stream create a realistic failure?

• Each evaluation listener has a larger old-genre history plus a smaller recent-current-genre history.
• Events from day 20 onward represent future preferences that the recommender must not use for training.
• The trend cohort saves every rock track on day 12. This creates a dominant global trend for the popularity baseline.

• Save app/core.py.
• Check that the completed generator parses by running this command:

python -m app.core

What does this check prove?

The module now loads the complete build_dataset() function. A return to the terminal prompt without a traceback confirms that both event loops are nested correctly.

Seeing an indentation error?

• Keep both listener loops inside build_dataset() with four leading spaces.
• Keep the trend loop aligned with the evaluation-user loop.
• Keep the final return aligned with both outer loops.

Ask for help with the loop structure: Help me correct the indentation in build_dataset without changing its generated data.

Enforce the global cutoff

A global temporal split gives every model the same simulated deployment boundary. Training receives events at or before day 19.

• Add the split function below build_dataset() by pasting this code:

def temporal_split(
    events: list[Event],
    evaluation_users: tuple[str, ...],
    cutoff_day: int = CUTOFF_DAY,
) -> tuple[list[Event], list[Event], dict[str, set[str]]]:
    users = set(evaluation_users)
    train = [event for event in events if event.day <= cutoff_day]
    test = [event for event in events if event.day > cutoff_day and event.user_id in users]
    holdouts = {
        user_id: {event.track_id for event in test if event.user_id == user_id and event.kind != "skip"}
        for user_id in evaluation_users
    }
    return train, test, holdouts

How does the split prevent leakage?

• train accepts only events whose day is at or before the cutoff.
• test accepts only later events from the evaluation listeners.
• holdouts keeps future non-skip track IDs as the relevance target for each listener.

• Save app/core.py.
• Check that the split function parses by running this command:

python -m app.core

What does this check prove?

The module now loads the cutoff logic alongside the generator. Returning to the prompt without a traceback confirms that the split comprehensions and holdout dictionary are valid.

Does the split fail to load?

• Check that temporal_split() starts at the left edge of the file.
• Confirm that the holdout comprehension excludes only events whose kind equals skip.

Ask for help with the split: Review my temporal_split function for syntax or indentation mistakes.

• Add the shared evaluation fixture below temporal_split() by pasting this code:

def evaluation_fixture() -> tuple[list[Track], list[Event], list[Event], dict[str, set[str]]]:
    tracks, events, users = build_dataset()
    train, test, holdouts = temporal_split(events, users)
    return tracks, train, test, holdouts

Why use one evaluation fixture?

evaluation_fixture() creates the catalog once per evaluation run. It returns one training partition plus one future test partition for every model.

• Save app/core.py.
• Check that the shared fixture parses by running this command:

python -m app.core

What does this check prove?

The module can now connect deterministic generation to the global split. Returning to the prompt without a traceback confirms that the fixture resolves both functions.

Run the popularity baseline

A popularity baseline ranks tracks using positive interaction weight from the training partition. It ignores each listener's genre profile beyond filtering tracks they already encountered.

• Insert the initial DiscoveryEngine class directly above evaluation_fixture() by pasting this code:

class DiscoveryEngine:
    def __init__(self, tracks: list[Track], events: list[Event]):
        self.tracks = tracks
        self.events = list(events)
        self.track_index = {track.id: index for index, track in enumerate(tracks)}

    def _weight(self, event: Event) -> float:
        return EVENT_WEIGHT[event.kind]

    def _history(self, user_id: str) -> list[Event]:
        return [event for event in self.events if event.user_id == user_id]

    def _seen(self, user_id: str) -> set[str]:
        return {event.track_id for event in self._history(user_id)}

    def _popularity(self) -> list[float]:
        scores = [0.0] * len(self.tracks)
        for event in self.events:
            scores[self.track_index[event.track_id]] += max(0.0, self._weight(event))
        return scores

    @staticmethod
    def _top_indices(scores: list[float], count: int) -> list[int]:
        values = np.asarray(scores, dtype=float)
        return [int(index) for index in np.argsort(values, stable=True, descending=True)[:count]]

    def popularity_recommend(self, user_id: str, limit: int = 30) -> list[Track]:
        seen = self._seen(user_id)
        ranked = self._top_indices(self._popularity(), len(self.tracks))
        return [self.tracks[index] for index in ranked if self.tracks[index].id not in seen][:limit]

How does the baseline rank tracks?

• _weight() converts each event kind into its configured numeric signal.
• _popularity() adds only non-negative signal to each track's global score.
• _seen() prevents the service from returning tracks already present in the listener's history.
• popularity_recommend() returns the highest-scoring unseen tracks with deterministic tie ordering.

• Add the demonstration function below evaluation_fixture() by pasting this code:

def demo() -> None:
    tracks, train, test, holdouts = evaluation_fixture()
    engine = DiscoveryEngine(tracks, train)
    print(
        "catalog", len(tracks),
        "train_max_day", max(event.day for event in train),
        "test_min_day", min(event.day for event in test),
    )
    print("holdout_users", len(holdouts))
    for user_id in ("u00", "u07"):
        popularity = [track.genre for track in engine.popularity_recommend(user_id)[:8]]
        print(user_id, "popularity", popularity)


if __name__ == "__main__":
    demo()

What does the demonstration reveal?

• The first line exposes the catalog size plus both sides of the temporal boundary.
• The listener lines show the first eight popularity genres for u00 and u07.
• The main guard runs demo() only when the module is executed directly.

• Save app/core.py.

Before you run the module, do you expect two listeners with different histories to escape the rock trend?

• Run the completed popularity demonstration with this command:

python -m app.core

What does this command test?

The module regenerates the same catalog plus the same events from seed 7. It ranks only the training partition before printing both listener examples.

You should see catalog 120 train_max_day 16 test_min_day 20 on the first line. This proves every held-out event occurs after every training event.

The u00 and u07 popularity sequences should both be similarly rock-heavy. That shared output is the intended popularity failure.

Seeing different counts or no listener output?

• Confirm that the catalog loop creates 20 tracks for each entry in genres.
• Check that the trend cohort saves every rock track on day 12.
• Confirm that the main guard calls demo().

Ask for help with the output: Compare my app.core output with the expected temporal boundary and popularity behavior.

• Run the same command a second time to confirm the output is identical:

python -m app.core

Why repeat the run?

The second run starts from a fresh process. Identical output proves the seed controls the generated catalog plus the complete timestamped event stream.

That is the failure captured: a globally popular trend overwhelms individual taste even though the evaluation boundary is leakage-safe.

Use these tabs to confirm that your cumulative app/core.py matches the step.

✔️ Awesome, I've got everything!

Your deterministic dataset, global split, and popularity baseline are ready. Make sure app/core.py is saved.

ⓧ I'd like to double check the full code

from __future__ import annotations

from dataclasses import dataclass
from random import Random

import numpy as np

EVENT_WEIGHT = {"play": 1.0, "skip": -2.0, "save": 3.0}
CUTOFF_DAY = 19


@dataclass(frozen=True)
class Track:
    id: str
    title: str
    artist: str
    genre: str
    mood: str
    energy: float


@dataclass(frozen=True)
class Event:
    user_id: str
    track_id: str
    kind: str
    day: int


def build_dataset(seed: int = 7) -> tuple[list[Track], list[Event], tuple[str, ...]]:
    rng = Random(seed)
    genres = ("ambient", "jazz", "rock", "electronic", "folk", "soul")
    moods = ("calm", "bright", "dark", "driving")
    tracks: list[Track] = []
    for genre_index, genre in enumerate(genres):
        for number in range(20):
            tracks.append(
                Track(
                    id=f"t{genre_index:02d}{number:02d}",
                    title=f"{genre.title()} Signal {number + 1}",
                    artist=f"{genre.title()} Artist {number // 2 + 1}",
                    genre=genre,
                    mood=moods[(number + genre_index) % len(moods)],
                    energy=round(0.2 + 0.6 * rng.random(), 2),
                )
            )

    by_genre = {genre: [track for track in tracks if track.genre == genre] for genre in genres}
    events: list[Event] = []
    evaluation_users = tuple(f"u{number:02d}" for number in range(18))
    for user_number, user_id in enumerate(evaluation_users):
        old_genre = genres[user_number % len(genres)]
        current_genre = genres[(user_number + 1) % len(genres)]
        noise_genre = genres[(user_number + 3) % len(genres)]
        for offset, track in enumerate(by_genre[old_genre][:10], start=1):
            kind = "save" if offset <= 6 else "play"
            events.append(Event(user_id, track.id, kind, offset))
        events.append(Event(user_id, by_genre[noise_genre][0].id, "skip", 11))
        for offset, track in enumerate(by_genre[current_genre][:4], start=13):
            events.append(Event(user_id, track.id, "save", offset))
        for offset, track in enumerate(by_genre[current_genre][4:], start=20):
            events.append(Event(user_id, track.id, "save", offset))

    for trend_number in range(12):
        trend_user = f"trend{trend_number:02d}"
        for track in by_genre["rock"]:
            events.append(Event(trend_user, track.id, "save", 12))
    return tracks, events, evaluation_users


def temporal_split(
    events: list[Event],
    evaluation_users: tuple[str, ...],
    cutoff_day: int = CUTOFF_DAY,
) -> tuple[list[Event], list[Event], dict[str, set[str]]]:
    users = set(evaluation_users)
    train = [event for event in events if event.day <= cutoff_day]
    test = [event for event in events if event.day > cutoff_day and event.user_id in users]
    holdouts = {
        user_id: {event.track_id for event in test if event.user_id == user_id and event.kind != "skip"}
        for user_id in evaluation_users
    }
    return train, test, holdouts


class DiscoveryEngine:
    def __init__(self, tracks: list[Track], events: list[Event]):
        self.tracks = tracks
        self.events = list(events)
        self.track_index = {track.id: index for index, track in enumerate(tracks)}

    def _weight(self, event: Event) -> float:
        return EVENT_WEIGHT[event.kind]

    def _history(self, user_id: str) -> list[Event]:
        return [event for event in self.events if event.user_id == user_id]

    def _seen(self, user_id: str) -> set[str]:
        return {event.track_id for event in self._history(user_id)}

    def _popularity(self) -> list[float]:
        scores = [0.0] * len(self.tracks)
        for event in self.events:
            scores[self.track_index[event.track_id]] += max(0.0, self._weight(event))
        return scores

    @staticmethod
    def _top_indices(scores: list[float], count: int) -> list[int]:
        values = np.asarray(scores, dtype=float)
        return [int(index) for index in np.argsort(values, stable=True, descending=True)[:count]]

    def popularity_recommend(self, user_id: str, limit: int = 30) -> list[Track]:
        seen = self._seen(user_id)
        ranked = self._top_indices(self._popularity(), len(self.tracks))
        return [self.tracks[index] for index in ranked if self.tracks[index].id not in seen][:limit]


def evaluation_fixture() -> tuple[list[Track], list[Event], list[Event], dict[str, set[str]]]:
    tracks, events, users = build_dataset()
    train, test, holdouts = temporal_split(events, users)
    return tracks, train, test, holdouts


def demo() -> None:
    tracks, train, test, holdouts = evaluation_fixture()
    engine = DiscoveryEngine(tracks, train)
    print(
        "catalog", len(tracks),
        "train_max_day", max(event.day for event in train),
        "test_min_day", min(event.day for event in test),
    )
    print("holdout_users", len(holdouts))
    for user_id in ("u00", "u07"):
        popularity = [track.genre for track in engine.popularity_recommend(user_id)[:8]]
        print(user_id, "popularity", popularity)


if __name__ == "__main__":
    demo()

You now have a deterministic popularity failure measured against a protected future holdout. Next, you will separate candidate retrieval from ranking so individual listening history can overcome the global trend.

Build the Hybrid Discovery Engine

Your global temporal split now gives every recommender the same leakage-safe training history. That hard boundary makes the popularity failure credible.

The next goal is a two-stage recommendation pipeline that separates candidate retrieval from hybrid ranking. This design makes personalization, cold-start behavior, and slate diversity explicit.

In this step, get ready to:

• Generate collaborative candidates from listeners with similar weighted histories.
• Generate content candidates from genre, mood, and energy similarity.
• Rank a deterministic 30-song slate with cold-start behavior and diversity caps.

Why use two stages?

Candidate retrieval searches broadly for plausible tracks. Ranking then combines the strongest signals into one ordered slate.

This split keeps each signal inspectable. It also gives product policies such as artist caps a clear place in the pipeline.

Build feature-based candidate signals

A hybrid recommender needs two independent views of listener taste. Collaborative retrieval learns from similar listeners. Content retrieval compares tracks through an auditable feature vector.

• In `app/core.py`, replace the current import section with this expanded import section:

from collections import Counter
from dataclasses import dataclass
from itertools import combinations
from random import Random

import numpy as np

What do these imports support?

• `Counter` tracks how many chosen songs belong to each artist or genre.
• `combinations` creates every track pair needed for a later slate-diversity measurement.
• `numpy` supplies vectors, dot products, norms, and deterministic sorting indices.

• Inside `DiscoveryEngine`, replace the existing constructor with this constructor and content-vector method:

    def __init__(self, tracks: list[Track], events: list[Event]):
        self.tracks = tracks
        self.events = list(events)
        self.track_by_id = {track.id: track for track in tracks}
        self.track_index = {track.id: index for index, track in enumerate(tracks)}
        self.genres = sorted({track.genre for track in tracks})
        self.moods = sorted({track.mood for track in tracks})
        self.content = [np.asarray(self._content_vector(track), dtype=float) for track in tracks]

    def _content_vector(self, track: Track) -> list[float]:
        genre = [1.0 if track.genre == value else 0.0 for value in self.genres]
        mood = [1.0 if track.mood == value else 0.0 for value in self.moods]
        return genre + mood + [track.energy, 1.0 - track.energy]

How does the content vector work?

Each track receives one-hot genre values and one-hot mood values. Its energy contributes a final two-value gradient.

Every track now occupies the same feature space. Cosine similarity can compare that track with a listener profile.

• Save `app/core.py`.
• Confirm the expanded feature setup still preserves the deterministic data pipeline by running:

python -m app.core

What does this check prove?

The command executes the module against the seeded catalog. A successful run proves that every track received a valid numeric vector without breaking the existing popularity baseline.

You should still see `catalog 120 train_max_day 16 test_min_day 20` near the top of the output.

Seeing an import or attribute error?

• Confirm the virtual environment from the previous step remains active.
• Check that `_content_vector` is indented inside `DiscoveryEngine`.
• Compare `self.genres`, `self.moods`, and `self.content` with the snippet above.

Ask for targeted help with this edit: Help me debug the content-vector setup in my DiscoveryEngine class.

Collaborative retrieval represents each listener as weighted track interactions. Positive cosine similarity identifies listeners whose play, skip, and save patterns point in a similar direction.

• Inside `DiscoveryEngine`, add `_collaborative` immediately after `_popularity`:

    def _collaborative(self, user_id: str) -> list[float]:
        users = sorted({event.user_id for event in self.events})
        if user_id not in users:
            return [0.0] * len(self.tracks)
        matrix = [[0.0] * len(self.tracks) for _ in users]
        user_index = {value: index for index, value in enumerate(users)}
        for event in self.events:
            matrix[user_index[event.user_id]][self.track_index[event.track_id]] += self._weight(event)
        target = np.asarray(matrix[user_index[user_id]], dtype=float)
        target_norm = float(np.linalg.norm(target))
        scores = [0.0] * len(self.tracks)
        if target_norm == 0.0:
            return scores
        for other_user, row_values in zip(users, matrix):
            if other_user == user_id:
                continue
            row = np.asarray(row_values, dtype=float)
            denominator = target_norm * float(np.linalg.norm(row))
            similarity = float(np.dot(target, row) / denominator) if denominator else 0.0
            if similarity <= 0.0:
                continue
            for index, value in enumerate(row_values):
                scores[index] += similarity * max(0.0, value)
        return scores

How does collaborative retrieval work?

• The interaction matrix gives every listener one weighted value per track.
• Cosine similarity compares the target listener with every other listener.
• Only positive similarities contribute candidate scores.
• `max(0.0, value)` prevents another listener's negative track interaction from becoming a recommendation boost.

Content retrieval builds one listener profile from the same weighted events. Signal normalization then places collaborative, content, and popularity scores on comparable zero-to-one ranges.

• Add `_content_scores` and `_scale` immediately after `_collaborative`:

    def _content_scores(self, user_id: str) -> list[float]:
        profile = [0.0] * len(self.content[0])
        for event in self._history(user_id):
            vector = self.content[self.track_index[event.track_id]]
            weight = self._weight(event)
            for index, value in enumerate(vector):
                profile[index] += weight * float(value)
        profile_array = np.asarray(profile, dtype=float)
        profile_norm = float(np.linalg.norm(profile_array))
        if profile_norm == 0.0:
            return [0.0] * len(self.tracks)
        return [float(np.dot(profile_array, vector) / (profile_norm * float(np.linalg.norm(vector)))) for vector in self.content]

    @staticmethod
    def _scale(scores: list[float]) -> list[float]:
        low, high = min(scores), max(scores)
        if high == low:
            return [1.0 if value > 0.0 else 0.0 for value in scores]
        return [(value - low) / (high - low) for value in scores]

What do these methods calculate?

`_content_scores` adds track vectors into a weighted listener profile. Saves pull the profile toward similar tracks. Skips push it away.

`_scale` preserves each signal's ordering while mapping its values onto a shared range. That makes the final ranking weights meaningful.

• Save `app/core.py`.
• Check that both candidate generators load without disrupting the seeded run by running:

python -m app.core

What does this run confirm?

The run constructs every content vector and loads both scoring methods as part of `DiscoveryEngine`. The deterministic catalog summary confirms that the original split remains intact.

You should see the same catalog size and temporal boundary as the previous run.

Seeing a vector-shape error?

• Confirm every vector comes from `self.content`.
• Check that the profile length uses `len(self.content[0])`.
• Verify that `_scale` remains a static method inside `DiscoveryEngine`.

Ask for help with the failing calculation: Help me debug my collaborative or content score calculation.

Rank and diversify the discovery slate

Candidate generation creates a broad pool. The ranker now unions three candidate sources and applies an explicit score blend.

A listener without history follows a cold-start path based on popularity. Every path passes through the same deterministic slate policy.

• Add `recommend` immediately after `popularity_recommend`:

    def recommend(self, user_id: str, limit: int = 30) -> list[Track]:
        seen = self._seen(user_id)
        popularity = self._popularity()
        if not self._history(user_id):
            return self._diversify(self._top_indices(popularity, len(self.tracks)), seen, limit)
        collaborative = self._collaborative(user_id)
        content = self._content_scores(user_id)
        candidates = set(self._top_indices(collaborative, 60))
        candidates.update(self._top_indices(content, 60))
        candidates.update(self._top_indices(popularity, 40))
        collab_scaled = self._scale(collaborative)
        content_scaled = self._scale(content)
        popularity_scaled = self._scale(popularity)
        ranked = sorted(candidates, key=lambda index: (0.30 * collab_scaled[index] + 0.55 * content_scaled[index] + 0.15 * popularity_scaled[index], self.tracks[index].id), reverse=True)
        return self._diversify(ranked, seen, limit)

How is the hybrid score composed?

• Collaborative retrieval contributes `0.30` of the final score.
• Content similarity contributes `0.55` of the final score.
• Popularity contributes `0.15` as a safety-net signal.
• Track IDs provide a deterministic tie-breaker.

Ranking alone can fill a playlist with near-duplicates. Greedy slate construction enforces product constraints while walking through the ranked candidates.

• Add `_diversify` immediately after `recommend`:

    def _diversify(self, ranked: list[int], seen: set[str], limit: int) -> list[Track]:
        chosen: list[Track] = []
        artists: Counter[str] = Counter()
        genres: Counter[str] = Counter()
        for index in ranked:
            track = self.tracks[index]
            if track.id in seen or artists[track.artist] >= 3 or genres[track.genre] >= 6:
                continue
            chosen.append(track)
            artists[track.artist] += 1
            genres[track.genre] += 1
            if len(chosen) == limit:
                return chosen
        for index in ranked:
            track = self.tracks[index]
            if track.id not in seen and track not in chosen:
                chosen.append(track)
            if len(chosen) == limit:
                break
        return chosen

How does greedy diversification work?

The first pass skips seen tracks. It also caps each artist at three tracks and each genre at six tracks.

The second pass fills any remaining positions from unseen ranked tracks. Membership checks prevent duplicates from entering the 30-song result.

• Save `app/core.py`.
• Confirm the new methods preserve a runnable module by running:

python -m app.core

Why run the module here?

This check catches indentation and symbol errors before the demo begins calling `recommend`. The catalog summary proves the engine still initializes against the training partition.

You should see the deterministic catalog summary without a traceback.

Seeing a ranking error?

• Check that `_diversify` is defined inside `DiscoveryEngine`.
• Confirm the three scaled lists use the same track indices.
• Verify that `Counter` is present in the import section.

Ask for help with the ranking path: Help me debug the hybrid ranker or slate constraints.

Complete and verify the engine contract

The engine also needs stable extension points for feedback and offline evidence. These methods reuse the same event weights and content representation.

• Add `add_feedback` and `diversity` immediately after `_diversify`:

    def add_feedback(self, user_id: str, track_id: str, kind: str) -> Event:
        if track_id not in self.track_by_id:
            raise KeyError(track_id)
        if kind not in EVENT_WEIGHT:
            raise ValueError(kind)
        next_day = max((event.day for event in self.events), default=CUTOFF_DAY) + 1
        event = Event(user_id, track_id, kind, next_day)
        self.events.append(event)
        return event

    def diversity(self, playlist: list[Track]) -> float:
        if len(playlist) < 2:
            return 0.0
        distances = []
        for left, right in combinations(playlist, 2):
            left_vector = self.content[self.track_index[left.id]]
            right_vector = self.content[self.track_index[right.id]]
            similarity = float(np.dot(left_vector, right_vector) / (float(np.linalg.norm(left_vector)) * float(np.linalg.norm(right_vector))))
            distances.append(1.0 - similarity)
        return sum(distances) / len(distances)

What do these engine hooks provide?

`add_feedback` appends one validated event after the latest known day. The next recommendation can use that event immediately.

`diversity` measures mean pairwise content-vector distance across a playlist. The measurement stays separate from the hard slate caps.

Shared evaluation helpers ensure every model uses the same holdouts and metric calculations. This keeps later comparisons tied to one immutable future partition.

• Add `evaluation_fixture` and `metric_report` below `DiscoveryEngine`:

def evaluation_fixture() -> tuple[list[Track], list[Event], list[Event], dict[str, set[str]]]:
    tracks, events, users = build_dataset()
    train, test, holdouts = temporal_split(events, users)
    return tracks, train, test, holdouts


def metric_report(engine: DiscoveryEngine, recommendations: dict[str, list[Track]], holdouts: dict[str, set[str]], k: int = 10) -> dict[str, float]:
    hits = 0
    relevant = 0
    recommended_ids: set[str] = set()
    diversities = []
    for user_id, playlist in recommendations.items():
        top_k = {track.id for track in playlist[:k]}
        hits += len(top_k & holdouts[user_id])
        relevant += len(holdouts[user_id])
        recommended_ids.update(track.id for track in playlist)
        diversities.append(engine.diversity(playlist))
    return {"Precision@10": round(hits / (len(recommendations) * k), 4), "Recall@10": round(hits / relevant, 4), "catalog_coverage": round(len(recommended_ids) / len(engine.tracks), 4), "diversity": round(sum(diversities) / len(diversities), 4)}

Why share one metric function?

`evaluation_fixture` reconstructs the deterministic training and future partitions. `metric_report` scores any recommendation dictionary against the same holdouts.

One implementation prevents metric drift between the popularity baseline and the hybrid model.

• Save `app/core.py`.
• Check that the support methods leave the module runnable by running:

python -m app.core

What does this checkpoint protect?

The command reconstructs the seeded fixture and initializes the expanded engine. A clean run confirms that the new helper functions did not alter the temporal split.

You should still see training stop before the held-out events begin.

• Add `report_for` and `evaluate` immediately before `demo`:

def report_for(engine: DiscoveryEngine, holdouts: dict[str, set[str]], popularity: bool = False):
    recommendations = {user_id: (engine.popularity_recommend(user_id) if popularity else engine.recommend(user_id)) for user_id in sorted(holdouts)}
    return metric_report(engine, recommendations, holdouts)


def evaluate() -> dict[str, dict[str, float]]:
    tracks, train, _, holdouts = evaluation_fixture()
    engine = DiscoveryEngine(tracks, train)
    return {"popularity": report_for(engine, holdouts, popularity=True), "hybrid": report_for(engine, holdouts)}

What does the report adapter do?

`report_for` builds recommendations for the same sorted listener set. `evaluate` compares popularity and hybrid behavior without rebuilding metric logic.

The later evidence report can call this stable interface directly.

The final demo exposes the difference that the popularity baseline concealed. It prints both strategies for `u00` and `u07` against the same training data.

• Replace the existing `demo` function with this completed version:

def demo() -> None:
    tracks, train, test, holdouts = evaluation_fixture()
    engine = DiscoveryEngine(tracks, train)
    print("catalog", len(tracks), "train_max_day", max(event.day for event in train), "test_min_day", min(event.day for event in test))
    print("holdout_users", len(holdouts))
    for user_id in ("u00", "u07"):
        print(user_id, "popularity", [track.genre for track in engine.popularity_recommend(user_id)[:8]])
        print(user_id, "hybrid", [track.genre for track in engine.recommend(user_id)[:8]])


if __name__ == "__main__":
    demo()

What does the demo compare?

Both listeners first receive the popularity baseline. They then receive hybrid results from their own weighted histories.

The shared catalog and split isolate personalization as the reason the hybrid genre sequences differ.

• Save `app/core.py`.

Before you run the final check, do you expect the two hybrid genre sequences to remain as similar as their popularity sequences?

• Run the completed discovery engine:

python -m app.core

What should this final run prove?

The first line confirms the unchanged catalog and global time boundary. The listener lines expose the difference between trend-driven popularity and personalized hybrid ranking.

You should see `u00 hybrid` and `u07 hybrid` followed by different genre sequences. Each call to `recommend` now produces a deterministic 30-track slate.

Do the hybrid lines fail or match?

• Confirm `demo` calls `engine.recommend(user_id)` for each hybrid line.
• Check that the hybrid score uses `0.30`, `0.55`, and `0.15`.
• Verify that collaborative and content candidates are both added to `candidates`.

Ask for help comparing the observed output: Help me debug why my hybrid recommendations are missing or identical.

That is the recommendation core working. Two listeners now receive personalized deterministic slates while the popularity baseline still exposes the original trend bias.

Compare your completed `app/core.py` with the reference before moving on.

✔️ Awesome, I've got everything!

Great work. Save `app/core.py` so the complete hybrid pipeline is ready for the API layer.

ⓧ I'd like to double check the full code

from __future__ import annotations

from collections import Counter
from dataclasses import dataclass
from itertools import combinations
from random import Random

import numpy as np

EVENT_WEIGHT = {"play": 1.0, "skip": -2.0, "save": 3.0}
CUTOFF_DAY = 19


@dataclass(frozen=True)
class Track:
    id: str
    title: str
    artist: str
    genre: str
    mood: str
    energy: float


@dataclass(frozen=True)
class Event:
    user_id: str
    track_id: str
    kind: str
    day: int


def build_dataset(seed: int = 7) -> tuple[list[Track], list[Event], tuple[str, ...]]:
    rng = Random(seed)
    genres = ("ambient", "jazz", "rock", "electronic", "folk", "soul")
    moods = ("calm", "bright", "dark", "driving")
    tracks: list[Track] = []
    for genre_index, genre in enumerate(genres):
        for number in range(20):
            tracks.append(Track(id=f"t{genre_index:02d}{number:02d}", title=f"{genre.title()} Signal {number + 1}", artist=f"{genre.title()} Artist {number // 2 + 1}", genre=genre, mood=moods[(number + genre_index) % len(moods)], energy=round(0.2 + 0.6 * rng.random(), 2)))
    by_genre = {genre: [track for track in tracks if track.genre == genre] for genre in genres}
    events: list[Event] = []
    evaluation_users = tuple(f"u{number:02d}" for number in range(18))
    for user_number, user_id in enumerate(evaluation_users):
        old_genre = genres[user_number % len(genres)]
        current_genre = genres[(user_number + 1) % len(genres)]
        noise_genre = genres[(user_number + 3) % len(genres)]
        for offset, track in enumerate(by_genre[old_genre][:10], start=1):
            events.append(Event(user_id, track.id, "save" if offset <= 6 else "play", offset))
        events.append(Event(user_id, by_genre[noise_genre][0].id, "skip", 11))
        for offset, track in enumerate(by_genre[current_genre][:4], start=13):
            events.append(Event(user_id, track.id, "save", offset))
        for offset, track in enumerate(by_genre[current_genre][4:], start=20):
            events.append(Event(user_id, track.id, "save", offset))
    for trend_number in range(12):
        trend_user = f"trend{trend_number:02d}"
        for track in by_genre["rock"]:
            events.append(Event(trend_user, track.id, "save", 12))
    return tracks, events, evaluation_users


def temporal_split(events: list[Event], evaluation_users: tuple[str, ...], cutoff_day: int = CUTOFF_DAY) -> tuple[list[Event], list[Event], dict[str, set[str]]]:
    users = set(evaluation_users)
    train = [event for event in events if event.day <= cutoff_day]
    test = [event for event in events if event.day > cutoff_day and event.user_id in users]
    holdouts = {user_id: {event.track_id for event in test if event.user_id == user_id and event.kind != "skip"} for user_id in evaluation_users}
    return train, test, holdouts


class DiscoveryEngine:
    def __init__(self, tracks: list[Track], events: list[Event]):
        self.tracks = tracks
        self.events = list(events)
        self.track_by_id = {track.id: track for track in tracks}
        self.track_index = {track.id: index for index, track in enumerate(tracks)}
        self.genres = sorted({track.genre for track in tracks})
        self.moods = sorted({track.mood for track in tracks})
        self.content = [np.asarray(self._content_vector(track), dtype=float) for track in tracks]

    def _content_vector(self, track: Track) -> list[float]:
        genre = [1.0 if track.genre == value else 0.0 for value in self.genres]
        mood = [1.0 if track.mood == value else 0.0 for value in self.moods]
        return genre + mood + [track.energy, 1.0 - track.energy]

    def _weight(self, event: Event) -> float:
        return EVENT_WEIGHT[event.kind]

    def _history(self, user_id: str) -> list[Event]:
        return [event for event in self.events if event.user_id == user_id]

    def _seen(self, user_id: str) -> set[str]:
        return {event.track_id for event in self._history(user_id)}

    def _popularity(self) -> list[float]:
        scores = [0.0] * len(self.tracks)
        for event in self.events:
            scores[self.track_index[event.track_id]] += max(0.0, self._weight(event))
        return scores

    def _collaborative(self, user_id: str) -> list[float]:
        users = sorted({event.user_id for event in self.events})
        if user_id not in users:
            return [0.0] * len(self.tracks)
        matrix = [[0.0] * len(self.tracks) for _ in users]
        user_index = {value: index for index, value in enumerate(users)}
        for event in self.events:
            matrix[user_index[event.user_id]][self.track_index[event.track_id]] += self._weight(event)
        target = np.asarray(matrix[user_index[user_id]], dtype=float)
        target_norm = float(np.linalg.norm(target))
        scores = [0.0] * len(self.tracks)
        if target_norm == 0.0:
            return scores
        for other_user, row_values in zip(users, matrix):
            if other_user == user_id:
                continue
            row = np.asarray(row_values, dtype=float)
            denominator = target_norm * float(np.linalg.norm(row))
            similarity = float(np.dot(target, row) / denominator) if denominator else 0.0
            if similarity <= 0.0:
                continue
            for index, value in enumerate(row_values):
                scores[index] += similarity * max(0.0, value)
        return scores

    def _content_scores(self, user_id: str) -> list[float]:
        profile = [0.0] * len(self.content[0])
        for event in self._history(user_id):
            vector = self.content[self.track_index[event.track_id]]
            weight = self._weight(event)
            for index, value in enumerate(vector):
                profile[index] += weight * float(value)
        profile_array = np.asarray(profile, dtype=float)
        profile_norm = float(np.linalg.norm(profile_array))
        if profile_norm == 0.0:
            return [0.0] * len(self.tracks)
        return [float(np.dot(profile_array, vector) / (profile_norm * float(np.linalg.norm(vector)))) for vector in self.content]

    @staticmethod
    def _scale(scores: list[float]) -> list[float]:
        low, high = min(scores), max(scores)
        if high == low:
            return [1.0 if value > 0.0 else 0.0 for value in scores]
        return [(value - low) / (high - low) for value in scores]

    @staticmethod
    def _top_indices(scores: list[float], count: int) -> list[int]:
        values = np.asarray(scores, dtype=float)
        return [int(index) for index in np.argsort(values, stable=True, descending=True)[:count]]

    def popularity_recommend(self, user_id: str, limit: int = 30) -> list[Track]:
        seen = self._seen(user_id)
        ranked = self._top_indices(self._popularity(), len(self.tracks))
        return [self.tracks[index] for index in ranked if self.tracks[index].id not in seen][:limit]

    def recommend(self, user_id: str, limit: int = 30) -> list[Track]:
        seen = self._seen(user_id)
        popularity = self._popularity()
        if not self._history(user_id):
            return self._diversify(self._top_indices(popularity, len(self.tracks)), seen, limit)
        collaborative = self._collaborative(user_id)
        content = self._content_scores(user_id)
        candidates = set(self._top_indices(collaborative, 60))
        candidates.update(self._top_indices(content, 60))
        candidates.update(self._top_indices(popularity, 40))
        collab_scaled = self._scale(collaborative)
        content_scaled = self._scale(content)
        popularity_scaled = self._scale(popularity)
        ranked = sorted(candidates, key=lambda index: (0.30 * collab_scaled[index] + 0.55 * content_scaled[index] + 0.15 * popularity_scaled[index], self.tracks[index].id), reverse=True)
        return self._diversify(ranked, seen, limit)

    def _diversify(self, ranked: list[int], seen: set[str], limit: int) -> list[Track]:
        chosen: list[Track] = []
        artists: Counter[str] = Counter()
        genres: Counter[str] = Counter()
        for index in ranked:
            track = self.tracks[index]
            if track.id in seen or artists[track.artist] >= 3 or genres[track.genre] >= 6:
                continue
            chosen.append(track)
            artists[track.artist] += 1
            genres[track.genre] += 1
            if len(chosen) == limit:
                return chosen
        for index in ranked:
            track = self.tracks[index]
            if track.id not in seen and track not in chosen:
                chosen.append(track)
            if len(chosen) == limit:
                break
        return chosen

    def add_feedback(self, user_id: str, track_id: str, kind: str) -> Event:
        if track_id not in self.track_by_id:
            raise KeyError(track_id)
        if kind not in EVENT_WEIGHT:
            raise ValueError(kind)
        next_day = max((event.day for event in self.events), default=CUTOFF_DAY) + 1
        event = Event(user_id, track_id, kind, next_day)
        self.events.append(event)
        return event

    def diversity(self, playlist: list[Track]) -> float:
        if len(playlist) < 2:
            return 0.0
        distances = []
        for left, right in combinations(playlist, 2):
            left_vector = self.content[self.track_index[left.id]]
            right_vector = self.content[self.track_index[right.id]]
            similarity = float(np.dot(left_vector, right_vector) / (float(np.linalg.norm(left_vector)) * float(np.linalg.norm(right_vector))))
            distances.append(1.0 - similarity)
        return sum(distances) / len(distances)


def evaluation_fixture() -> tuple[list[Track], list[Event], list[Event], dict[str, set[str]]]:
    tracks, events, users = build_dataset()
    train, test, holdouts = temporal_split(events, users)
    return tracks, train, test, holdouts


def metric_report(engine: DiscoveryEngine, recommendations: dict[str, list[Track]], holdouts: dict[str, set[str]], k: int = 10) -> dict[str, float]:
    hits = 0
    relevant = 0
    recommended_ids: set[str] = set()
    diversities = []
    for user_id, playlist in recommendations.items():
        top_k = {track.id for track in playlist[:k]}
        hits += len(top_k & holdouts[user_id])
        relevant += len(holdouts[user_id])
        recommended_ids.update(track.id for track in playlist)
        diversities.append(engine.diversity(playlist))
    return {"Precision@10": round(hits / (len(recommendations) * k), 4), "Recall@10": round(hits / relevant, 4), "catalog_coverage": round(len(recommended_ids) / len(engine.tracks), 4), "diversity": round(sum(diversities) / len(diversities), 4)}


def report_for(engine: DiscoveryEngine, holdouts: dict[str, set[str]], popularity: bool = False):
    recommendations = {user_id: (engine.popularity_recommend(user_id) if popularity else engine.recommend(user_id)) for user_id in sorted(holdouts)}
    return metric_report(engine, recommendations, holdouts)


def evaluate() -> dict[str, dict[str, float]]:
    tracks, train, _, holdouts = evaluation_fixture()
    engine = DiscoveryEngine(tracks, train)
    return {"popularity": report_for(engine, holdouts, popularity=True), "hybrid": report_for(engine, holdouts)}


def demo() -> None:
    tracks, train, test, holdouts = evaluation_fixture()
    engine = DiscoveryEngine(tracks, train)
    print("catalog", len(tracks), "train_max_day", max(event.day for event in train), "test_min_day", min(event.day for event in test))
    print("holdout_users", len(holdouts))
    for user_id in ("u00", "u07"):
        print(user_id, "popularity", [track.genre for track in engine.popularity_recommend(user_id)[:8]])
        print(user_id, "hybrid", [track.genre for track in engine.recommend(user_id)[:8]])


if __name__ == "__main__":
    demo()

How to use this reference

Compare method order, indentation, identifiers, and numeric weights with your saved file. The reference represents the complete engine state for this step.

Your hybrid engine now retrieves broadly and ranks explicitly. Next up, you will expose it through FastAPI and protect its claims with regression evidence.

Expose Feedback and Evaluate Models

Your ranking engine can generate personalized slates, but its quality claims still live inside Python functions. A stable API contract makes those recommendations available to other applications.

This step exposes the engine through FastAPI. It also creates shared evaluation evidence that protects model quality and the train-test boundary from future code changes.

In this step, get ready to:

• Serve discovery playlists through health, recommendation, and feedback endpoints.
• Compare popularity and hybrid recommendations against one immutable future holdout.
• Protect the API contract and evaluation claims with automated regression tests.

Expose discovery and feedback endpoints

The application keeps one in-memory engine built from the training partition. Each request uses that same engine, so accepted feedback can influence the next playlist without contaminating the immutable evaluation fixture.

• Create `main.py` inside the existing `app` folder using the file sidebar in VS Code.
• Initialize the application by adding this code to `app/main.py`:

from typing import Literal

from fastapi import FastAPI, HTTPException

from app.core import DiscoveryEngine, build_dataset, temporal_split

TRACKS, ALL_EVENTS, EVALUATION_USERS = build_dataset()
TRAIN_EVENTS, _, _ = temporal_split(ALL_EVENTS, EVALUATION_USERS)
ENGINE = DiscoveryEngine(TRACKS, TRAIN_EVENTS)
app = FastAPI(title="Personalized Music Discovery Engine")

What does this setup do?

• `temporal_split()` gives the service only the training events.
• `ENGINE` stores a mutable copy of those events for live feedback.
• `app` defines the FastAPI application that Uvicorn serves.

• Save `app/main.py`.
• Start the application by running this command in the activated terminal:

uvicorn app.main:app

What should you see?

Uvicorn starts the application without an import or startup failure. This proves the dataset, temporal split, engine, and FastAPI application can initialize together.

• Stop the server by pressing Control and C in the terminal.

Does the application fail to start?

Confirm the virtual environment remains active. Check that `app/main.py` imports each identifier exactly as it appears in `app/core.py`.

If the traceback mentions an import or syntax problem, use the first failing file and line as your starting point. Help me diagnose the FastAPI startup traceback.

The first endpoint gives monitoring tools a small readiness signal. The discovery endpoint returns the strategy plus the complete 30-track slate for a listener.

• Add the health endpoint below the application setup in `app/main.py`:

@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}

What does the health endpoint prove?

`GET /health` returns a small response when the application is initialized and accepting requests. External monitoring can use this route without generating a playlist.

• Add the discovery endpoint below `health()` in `app/main.py`:

@app.get("/users/{user_id}/discover")
def discover(user_id: str) -> dict[str, object]:
    strategy = "hybrid" if ENGINE._history(user_id) else "cold_start"
    playlist = ENGINE.recommend(user_id, limit=30)
    return {"user_id": user_id, "strategy": strategy, "tracks": [{"id": track.id, "title": track.title, "artist": track.artist, "genre": track.genre, "mood": track.mood, "energy": track.energy} for track in playlist]}

How does discovery work?

• `strategy` exposes whether the listener has enough history for hybrid ranking.
• `ENGINE.recommend()` applies the existing retrieval, ranking, and diversity pipeline.
• The response serializes each selected track into fields a client can consume.

• Save `app/main.py`.
• Restart the application by running:

uvicorn app.main:app

The read endpoints are live

The running server now exposes the health and discovery routes. The interactive documentation provides a browser-based client for checking both responses.

• Open `http://127.0.0.1:8000/docs` in your browser.
• Expand `GET /health`.
• Click **Try it out**.
• Click **Execute**.
• Confirm the response contains `{"status": "ok"}`.
• Expand `GET /users/{user_id}/discover`.
• Enter `api-listener` as `user_id`.
• Execute the request.

You should see `cold_start` as the strategy. The response should contain 30 tracks for this unseen listener.

• Stop the server by pressing Control and C in the terminal.

The feedback endpoint translates a play, skip, or save request into a new in-memory event. A listener with one accepted event moves from the cold start path to hybrid ranking on the next request.

• Add the feedback endpoint below `discover()` in `app/main.py`:

@app.post("/users/{user_id}/feedback")
def feedback(user_id: str, track_id: str, event: Literal["play", "skip", "save"]) -> dict[str, object]:
    try:
        accepted = ENGINE.add_feedback(user_id, track_id, event)
    except KeyError as error:
        raise HTTPException(status_code=404, detail="Track not found") from error
    return {"accepted": True, "day": accepted.day, "event": accepted.kind}

How does feedback change the engine?

• `Literal` limits accepted feedback kinds to play, skip, or save.
• `ENGINE.add_feedback()` validates the track before appending the event.
• An unknown track becomes an HTTP response with a missing-resource status.

• Save `app/main.py`.
• Restart the completed application by running:

uvicorn app.main:app

The feedback loop is ready

The server now exposes all three routes from `app/main.py`. The next requests prove that one accepted save changes the listener's strategy.

• Return to `http://127.0.0.1:8000/docs`.
• Expand `POST /users/{user_id}/feedback`.
• Enter `api-listener` as `user_id`.
• Enter `t0000` as `track_id`.
• Select `save` as the event.
• Execute the request.
• Execute `GET /users/{user_id}/discover` again for `api-listener`.

You should now see `hybrid` as the strategy. The response still contains a complete 30-track playlist.

Does the strategy remain on cold start?

Confirm the feedback request and the discovery request use the same `api-listener` value. Check that the feedback response confirms acceptance before repeating discovery.

Restarting the server clears in-memory feedback, so submit the save again after every restart. Help me trace why the listener remains on cold start.

• Stop the server by pressing Control and C in the terminal.

The API feedback loop now works. Double-check the completed service file before moving into model evidence.

✔️ Awesome, I've got everything!

Your `app/main.py` file now exposes health, discovery, and feedback routes. Make sure the file is saved.

ⓧ I'd like to double check the full code

from typing import Literal

from fastapi import FastAPI, HTTPException

from app.core import DiscoveryEngine, build_dataset, temporal_split

TRACKS, ALL_EVENTS, EVALUATION_USERS = build_dataset()
TRAIN_EVENTS, _, _ = temporal_split(ALL_EVENTS, EVALUATION_USERS)
ENGINE = DiscoveryEngine(TRACKS, TRAIN_EVENTS)
app = FastAPI(title="Personalized Music Discovery Engine")


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/users/{user_id}/discover")
def discover(user_id: str) -> dict[str, object]:
    strategy = "hybrid" if ENGINE._history(user_id) else "cold_start"
    playlist = ENGINE.recommend(user_id, limit=30)
    return {"user_id": user_id, "strategy": strategy, "tracks": [{"id": track.id, "title": track.title, "artist": track.artist, "genre": track.genre, "mood": track.mood, "energy": track.energy} for track in playlist]}


@app.post("/users/{user_id}/feedback")
def feedback(user_id: str, track_id: str, event: Literal["play", "skip", "save"]) -> dict[str, object]:
    try:
        accepted = ENGINE.add_feedback(user_id, track_id, event)
    except KeyError as error:
        raise HTTPException(status_code=404, detail="Track not found") from error
    return {"accepted": True, "day": accepted.day, "event": accepted.kind}

What should match?

Compare the imports, shared engine initialization, three route decorators, response fields, and feedback error handling. Your file should match this reference exactly.

Build the evidence report and regression suite

A model comparison is credible when every recommender receives the same training partition and the same future holdout. The existing `evaluate()` function already enforces that protocol, so the report script only serializes its result as JSON.

• Create `evaluate.py` inside the existing `app` folder using the file sidebar in VS Code.
• Add this report entry point to `app/evaluate.py`:

import json

from app.core import evaluate


if __name__ == "__main__":
    print(json.dumps(evaluate(), indent=2, sort_keys=True))

What does the report script do?

`evaluate()` computes both model reports through the same metric function. Sorted, indented JSON gives the evidence a stable structure that humans and automated checks can inspect.

• Save `app/evaluate.py`.

Before you run the report, consider whether the personalized model should improve relevance and catalog coverage over the popularity baseline.

• Generate the shared evaluation report by running:

python -m app.evaluate

What should the report contain?

You should see `popularity` and `hybrid` objects. Each object contains `Precision@10`, `Recall@10`, `catalog_coverage`, and `diversity`.

The hybrid values should improve relevance and catalog coverage over popularity. Both models are scored against the same future events.

Does the report fail or omit metrics?

Confirm `app/evaluate.py` imports `evaluate` from `app.core`. Check that the activated terminal is inside the project folder that contains `app`.

If the metric relationships fail, keep the report unchanged and inspect the ranking logic in `app/core.py`. Help me diagnose my evaluation report.

✔️ Awesome, I've got everything!

Your `app/evaluate.py` file now prints the two-model evidence report. Make sure the file is saved.

ⓧ I'd like to double check the full code

import json

from app.core import evaluate


if __name__ == "__main__":
    print(json.dumps(evaluate(), indent=2, sort_keys=True))

What should match?

The file should contain one standard-library import, one project import, and one guarded print call. This keeps report generation separate from the metric implementation.

The regression suite turns architectural assumptions into executable contracts. pytest checks deterministic data, temporal isolation, slate constraints, cold-start behavior, model improvement, report shape, and API feedback.

• Create `test_system.py` inside the existing `tests` folder using the file sidebar in VS Code.
• Add the imports and shared test client to `tests/test_system.py`:

from collections import Counter

from fastapi.testclient import TestClient

from app.core import DiscoveryEngine, build_dataset, evaluate, temporal_split
from app.main import app

client = TestClient(app)

Why use one test client?

`TestClient` sends requests directly to the FastAPI application during testing. This exercises request handling without starting a separate network server.

• Add the deterministic split test below `client`:

def test_dataset_and_split_are_deterministic() -> None:
    assert build_dataset() == build_dataset()
    tracks, events, users = build_dataset()
    train, test, holdouts = temporal_split(events, users)
    assert len(tracks) == 120
    assert len(holdouts) == 18
    assert max(event.day for event in train) < min(event.day for event in test)
    assert set(train).isdisjoint(test)

What boundary does this protect?

The test proves repeated dataset construction is identical. It also proves every training timestamp precedes the held-out period and that no event belongs to both partitions.

• Save `tests/test_system.py`.
• Run the current regression suite with:

pytest

What should pass now?

Every test currently present in `tests/test_system.py` should pass. This first result locks down deterministic data and global temporal isolation before the suite expands.

• Add the playlist contract test below the deterministic split test:

def test_playlist_contract_and_diversity_caps() -> None:
    tracks, events, users = build_dataset()
    train, _, _ = temporal_split(events, users)
    playlist = DiscoveryEngine(tracks, train).recommend("u00")
    assert len(playlist) == 30
    assert len({track.id for track in playlist}) == 30
    assert max(Counter(track.artist for track in playlist).values()) <= 3
    assert max(Counter(track.genre for track in playlist).values()) <= 6

What does the playlist contract cover?

The test protects the 30-track response size and track uniqueness. It also enforces the explicit artist and genre caps applied during slate construction.

• Add the unseen-listener test below the playlist contract test:

def test_cold_start_returns_a_full_playlist() -> None:
    tracks, events, users = build_dataset()
    train, _, _ = temporal_split(events, users)
    assert len(DiscoveryEngine(tracks, train).recommend("new-listener")) == 30

Why test an unseen listener?

An unseen listener has no collaborative or content history. This test proves the popularity fallback still produces the full service contract.

• Save `tests/test_system.py`.
• Run the expanded suite with:

pytest

What should the expanded run prove?

Every test currently in the file should pass. The suite now protects deterministic splitting, personalized slate constraints, and unseen-listener behavior.

• Add the baseline-improvement test below the cold-start test:

def test_hybrid_beats_the_popularity_failure() -> None:
    report = evaluate()
    assert report["hybrid"]["Precision@10"] > report["popularity"]["Precision@10"]
    assert report["hybrid"]["Recall@10"] > report["popularity"]["Recall@10"]
    assert report["hybrid"]["catalog_coverage"] > report["popularity"]["catalog_coverage"]
    assert set(report["hybrid"]) == {"Precision@10", "Recall@10", "catalog_coverage", "diversity"}

What model claim becomes executable?

The hybrid recommender must improve precision, recall, and catalog coverage over the designed popularity failure. The final assertion prevents accidental changes to the evidence schema.

• Add the API feedback test below the model comparison test:

def test_api_feedback_changes_a_cold_start_user() -> None:
    before = client.get("/users/api-listener/discover")
    assert before.status_code == 200
    assert before.json()["strategy"] == "cold_start"
    feedback = client.post("/users/api-listener/feedback?track_id=t0000&event=save")
    assert feedback.status_code == 200
    after = client.get("/users/api-listener/discover")
    assert after.status_code == 200
    assert after.json()["strategy"] == "hybrid"
    assert len(after.json()["tracks"]) == 30

What does the API regression prove?

The test begins with an unseen listener and submits one save event. The next request must use hybrid ranking while preserving the 30-track response contract.

• Save `tests/test_system.py`.

Before you run the completed suite, consider which failure would reveal leakage and which failure would reveal a broken serving contract.

• Run every regression test by running:

pytest

The evidence contracts should pass

Every test should pass. The result protects the temporal boundary, recommendation contract, expected metric relationships, and mutable API feedback behavior.

Is a regression test failing?

Read the first failing assertion before changing any code. A temporal test points to dataset partitioning, while a playlist test points to ranking or diversity behavior.

An API failure can come from shared in-memory state or a mismatched route parameter. Help me investigate the first failing regression test.

✔️ Awesome, I've got everything!

Your `tests/test_system.py` file now protects the data, ranking, evaluation, and API contracts. Make sure the completed suite is saved.

ⓧ I'd like to double check the full code

from collections import Counter

from fastapi.testclient import TestClient

from app.core import DiscoveryEngine, build_dataset, evaluate, temporal_split
from app.main import app

client = TestClient(app)


def test_dataset_and_split_are_deterministic() -> None:
    assert build_dataset() == build_dataset()
    tracks, events, users = build_dataset()
    train, test, holdouts = temporal_split(events, users)
    assert len(tracks) == 120
    assert len(holdouts) == 18
    assert max(event.day for event in train) < min(event.day for event in test)
    assert set(train).isdisjoint(test)


def test_playlist_contract_and_diversity_caps() -> None:
    tracks, events, users = build_dataset()
    train, _, _ = temporal_split(events, users)
    playlist = DiscoveryEngine(tracks, train).recommend("u00")
    assert len(playlist) == 30
    assert len({track.id for track in playlist}) == 30
    assert max(Counter(track.artist for track in playlist).values()) <= 3
    assert max(Counter(track.genre for track in playlist).values()) <= 6


def test_cold_start_returns_a_full_playlist() -> None:
    tracks, events, users = build_dataset()
    train, _, _ = temporal_split(events, users)
    assert len(DiscoveryEngine(tracks, train).recommend("new-listener")) == 30


def test_hybrid_beats_the_popularity_failure() -> None:
    report = evaluate()
    assert report["hybrid"]["Precision@10"] > report["popularity"]["Precision@10"]
    assert report["hybrid"]["Recall@10"] > report["popularity"]["Recall@10"]
    assert report["hybrid"]["catalog_coverage"] > report["popularity"]["catalog_coverage"]
    assert set(report["hybrid"]) == {"Precision@10", "Recall@10", "catalog_coverage", "diversity"}


def test_api_feedback_changes_a_cold_start_user() -> None:
    before = client.get("/users/api-listener/discover")
    assert before.status_code == 200
    assert before.json()["strategy"] == "cold_start"
    feedback = client.post("/users/api-listener/feedback?track_id=t0000&event=save")
    assert feedback.status_code == 200
    after = client.get("/users/api-listener/discover")
    assert after.status_code == 200
    assert after.json()["strategy"] == "hybrid"
    assert len(after.json()["tracks"]) == 30

What should match?

Compare the import order, shared client, five test names, route values, and metric assertions. The complete file should match this reference exactly.

Document and verify the finished service

The README connects the implementation to its engineering evidence. It records the architecture, reproducible commands, metric definitions, leakage controls, and production limitations without presenting the simulation as a production system.

• Create `README.md` inside the top-level project folder using the file sidebar in VS Code.
• Add this project documentation to `README.md`:

# Personalized Music Discovery Engine

A deterministic, local-first recommendation service that returns a personalized 30-song playlist without external APIs, credentials, or heavyweight training.

## Architecture

The service uses seeded timestamped events, a global day-19 split, collaborative candidates, content candidates, a popularity safety net, hybrid ranking, artist and genre constraints, and FastAPI feedback.

## Run locally

```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
pytest
python -m app.evaluate
uvicorn app.main:app
```

## Leakage-safe evaluation

Every model receives the same catalog and events at or before day 19. Events from day 20 onward are held out. Precision@10 is held-out hits divided by ten recommendations per listener; Recall@10 is held-out hits divided by all held-out relevant tracks; catalog coverage is unique recommended tracks divided by 120; diversity is mean pairwise content-vector dissimilarity.

## Production limitations

Events are in memory, retrieval uses exact local cosine scans, ranking weights are hand tuned, synthetic offline metrics do not replace online experiments, and feedback has no authentication, idempotency, or abuse controls.

What does the README capture?

• The architecture summary names the complete recommendation path.
• The run commands make the local evidence reproducible.
• The evaluation section defines all four metrics and the global temporal boundary.
• The limitations separate this educational system from a production deployment.

• Save `README.md`.

✔️ Awesome, I've got everything!

Your `README.md` now documents how the system works, how to reproduce its evidence, and where its production limits begin.

ⓧ I'd like to double check the full code

# Personalized Music Discovery Engine

A deterministic, local-first recommendation service that returns a personalized 30-song playlist without external APIs, credentials, or heavyweight training.

## Architecture

The service uses seeded timestamped events, a global day-19 split, collaborative candidates, content candidates, a popularity safety net, hybrid ranking, artist and genre constraints, and FastAPI feedback.

## Run locally

```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
pytest
python -m app.evaluate
uvicorn app.main:app
```

## Leakage-safe evaluation

Every model receives the same catalog and events at or before day 19. Events from day 20 onward are held out. Precision@10 is held-out hits divided by ten recommendations per listener; Recall@10 is held-out hits divided by all held-out relevant tracks; catalog coverage is unique recommended tracks divided by 120; diversity is mean pairwise content-vector dissimilarity.

## Production limitations

Events are in memory, retrieval uses exact local cosine scans, ranking weights are hand tuned, synthetic offline metrics do not replace online experiments, and feedback has no authentication, idempotency, or abuse controls.

What should match?

Compare each heading, command, metric definition, and limitation. The documentation should match this reference exactly.

Before you run the final checks, consider whether the tests and report now support every quality claim documented in the README.

• Run the completed regression suite by running:

pytest

The regression evidence should hold

All tests should pass. This confirms deterministic data, temporal isolation, slate constraints, cold start, model improvement, report schema, and API feedback.

• Generate the final two-model report by running:

python -m app.evaluate

The report should use one schema

The output should contain popularity and hybrid results under the same four metric names. Hybrid should improve relevance and catalog coverage over the popularity baseline.

• Start the verified service by running:

uvicorn app.main:app

The complete service is running

Uvicorn should start without an application error. The same code that passed the API regression test is now available through the local server.

• Open `http://127.0.0.1:8000/docs` in your browser.
• Execute `GET /health`.
• Confirm the response contains `{"status": "ok"}`.

That is the full evidence loop working: the tests pass, the report compares both models, and the API responds through its documented contract.

You have closed the loop from ranking logic to a tested service. Feedback changes listener state, while one immutable holdout keeps the offline comparison honest.

Secret mission

Secret Mission: Detect and Correct Taste Drift

A listener's recent choices can disappear beneath a larger history of older behavior. Extend the benchmark with recency-aware event weighting while preserving the shared future holdout.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep your local resources ready for more experiments, pause any running processes, or delete the project artifacts. Everything runs on your computer, so keeping the project creates no ongoing service costs.

Resources you used:

• Local project artifacts, including app, tests, .venv, README.md, and requirements.txt.
• An isolated Python virtual environment containing the pinned project dependencies.

Keep everything running

No action is needed. Choose this if you want to continue testing recommendation strategies or extending the evidence report.

• Keep app, tests, .venv, README.md, and requirements.txt together in their current directory.
• Use the existing isolated environment when you return to the project.

Pause - I'll come back to this later

Pausing shuts down local processes to free memory while preserving every project file and installed dependency.

• Stop the local API if it is running by pressing Ctrl+C in its terminal.
• Close any terminal window that still has the isolated environment active.
• Leave the project artifacts in their current directory for your next session.

After stopping a running API, you'll see its request logs stop. Your code and evidence report remain ready for another session.

Delete - I don't want to use this again

Remove the listed local artifacts if you are finished with the project. Deleting these paths is permanent. The cleanup command is limited to the artifacts you confirm below.

• Return to the terminal where you ran the project tests.
• Confirm that the terminal is positioned in the project directory by running this command:

ls

Check your location first

You should see app, tests, README.md, and requirements.txt in the output. This confirms that the terminal is in the project directory.

• Stop here if any of those paths are missing.
• Return to the terminal where you ran the test suite before continuing.

• Delete the project artifacts by running this command:

rm -rf app tests .venv README.md requirements.txt

What does this command remove?

This removes the application modules, test suite, isolated environment, documentation, and pinned dependency file. The command leaves unrelated paths outside this list untouched.

• Verify that the visible project paths are gone by running this command:

ls

What remains?

Your output should no longer include app, tests, README.md, or requirements.txt. The cleanup command also removed the isolated .venv directory.

Still see a project path?

• Check that you returned to the same directory where you ran the project tests.
• Run the cleanup command again after correcting the terminal location.

Help me remove a remaining local project path.

Nice Work!

Nice Work!

You made it! Your local recommendation service now returns personalized 30-song playlists from deterministic listening histories.

Your evidence report compares three ranking strategies against one unchanged future holdout.

What you learned:

• Generated a deterministic dataset of 120 tracks with timestamped implicit feedback. Enforced a global day-19 temporal split that keeps future listening events outside training. Reproduced the popularity trap before introducing personalization.
• Built a two-stage recommendation pipeline. Combined collaborative retrieval with content retrieval for candidate generation. Applied hybrid ranking under explicit slate constraints. Returned 30 unique songs for established listeners plus cold-start listeners.
• Published personalized playlists through FastAPI endpoints. Accepted play, skip, or save feedback in memory. Reported Precision@10 plus Recall@10 for relevance. Measured catalog coverage plus diversity for breadth. Protected the contracts with pytest regression tests.
• Secret Mission: Corrected taste drift by weighting recent training events more heavily. Added recency_hybrid to the shared evidence report without changing the global split. Proved higher Precision@10 plus Recall@10 against the ordinary hybrid.

Ready to quiz yourself?