Build a Hybrid Movie Recommender
Build a local movie recommender with hybrid ranking, fallbacks, and metrics.
Introduction
30 Second Summary
Streaming apps have thousands of titles competing for a few spots on your screen. When every viewer sees the same list, discovering something worth watching feels generic.
In this project, you will build StreamFlix, a local JSON API that turns MovieLens Latest Small ratings into personalized movie rows. Every response exposes the system's recommendation strategy, fallback behavior, and serving metrics.
What You'll Build
You will demo StreamFlix by showing two users receive different movie rows while a simulated personalization failure still returns ranked recommendations.
By the end of this project, you'll have:
- An offline preparation job that turns MovieLens data into a serving-ready index whose metadata and keys you can inspect.
- A Top-N overlap experiment that proves popularity gives two users identical lists. The same comparison shows lower overlap after collaborative candidate generation and hybrid ranking.
- A resilient recommendation API that returns ranked fallbacks for cold-start users and simulated dependency failures. Its serving metrics expose degraded behavior. Its scale-up map connects the local design to production serving decisions.
- Secret Mission: Add a diversity-aware post-ranker that caps genre concentration and measures the change.
Do I need any experience before starting?
Comfort with Python plus basic system design knowledge is enough to begin. No cloud account or paid service is required because every project resource stays on your Mac.
Before We Start
This is your moment to commit to StreamFlix's goal: ranked MovieLens movie recommendations through a local JSON API. This step also clarifies why offline preparation keeps expensive data work outside the request path.
Prepare an Offline Recommendation Index
Each StreamFlix request needs fast access to movie data that has already been organized for recommendations. Scanning raw MovieLens Latest Small CSV files during every request would make serving depend on repeated parsing.
This step creates an offline boundary around that expensive work. A preparation job downloads the dataset. It converts the ratings into a serving-ready JSON contract that the online service can load once.
In this step, get ready to:
- Verify that Python 3.7 or newer is available.
- Create the StreamFlix folder with four scaffold files.
- Generate a serving-ready recommendation index from MovieLens Latest Small.
Create the project and verify Python
A named workspace keeps the preparation job separate from the generated data. Visual Studio Code also gives you an integrated terminal that starts inside the folder you open.
- Press Cmd+Space to open Spotlight on your Mac.
You will see the Spotlight search field at the center of your screen.
- Type Visual Studio Code into Spotlight.
- Press Return to open Visual Studio Code.
Visual Studio Code opens with its welcome screen or your most recent workspace.
- Click File in the menu bar.
- Click Open Folder....
The macOS folder chooser opens so you can create the workspace in a location you can find again.
- Select Desktop in the folder chooser.
- Click New Folder.
- Enter streamflix as the folder name.
- Press Return to create the folder.
The folder chooser now shows the new streamflix folder on your Desktop.
- Select the streamflix folder.
- Click Open.
You will see streamflix at the top of the Explorer sidebar.
- Click View in the menu bar.
- Click Terminal.
The integrated terminal opens inside your streamflix folder. This location keeps every command scoped to the project.
- Check your installed Python version by running this command:
python3 --versionWhat does this command check?
The command asks the active Python 3 interpreter to print its version. StreamFlix needs version 3.7 or newer because its local HTTP server later uses functionality introduced in that release.
✔️ I see version 3.7 or higher
Your Python installation is ready for StreamFlix. You have confirmed that the preparation job can use the required standard-library features.
- Keep the integrated terminal open inside streamflix.
ⓧ I see an older version
Your current interpreter is older than the project requires. Install the current macOS release before adding the preparation code.
- Open the official Python downloads for macOS page.
- Download Python 3.14.7 for macOS.
- Complete the macOS installer.
- Quit Visual Studio Code after the installation finishes.
- Reopen the streamflix folder in Visual Studio Code.
- Return to the integrated terminal.
- Verify the updated interpreter by running this command:
python3 --versionWhat should I see?
The command should now report Python 3.14.7 or another version newer than 3.7.
Still seeing the older version?
Restart Visual Studio Code so its integrated terminal picks up the updated interpreter path.
If the older version remains active, help me make the new Python installation available in the VS Code terminal.
ⓧ Command not found
The terminal cannot currently find a Python 3 interpreter. Install the current macOS release from Python.org.
- Open the official Python downloads for macOS page.
- Download Python 3.14.7 for macOS.
- Complete the macOS installer.
- Quit Visual Studio Code after the installation finishes.
- Reopen the streamflix folder in Visual Studio Code.
- Return to the integrated terminal.
- Confirm that Python is now available by running this command:
python3 --versionWhat should I see?
The terminal should print Python 3.14.7 or another version newer than 3.7.
Python still unavailable?
Restart Visual Studio Code after the installer completes. This gives the integrated terminal a fresh environment.
If the command is still unavailable, help me troubleshoot Python 3 installation on macOS.
With Python verified, the workspace needs one file for each later responsibility. This step fills only prepare.py because online serving begins in the next step.
- Select the Explorer view in the Activity Bar.
- Click the New File... button.
- Enter prepare.py and press Return.
- Click the New File... button.
- Enter recommender.py and press Return.
- Click the New File... button.
- Enter app.py and press Return.
- Click the New File... button.
Three Python files now appear in the Explorer sidebar. The final scaffold holds your later architecture narrative.
- Enter ARCHITECTURE.md and press Return.
Good progress. Your streamflix workspace now has a clear place for offline preparation, online recommendation logic, the HTTP API, and architecture notes.
Build the offline preparation job
The preparation program has three responsibilities. It acquires the trusted dataset, computes reusable indexes, and writes one artifact for the future request path.
- Select prepare.py in the Explorer sidebar.
- Add the imports, paths, threshold, and first part of the download function by pasting this code:
import csv
import json
import math
from collections import Counter, defaultdict
from pathlib import Path
from urllib.request import urlopen
from zipfile import ZipFile
DATA_URL = "https://files.grouplens.org/datasets/movielens/ml-latest-small.zip"
DATA_DIR = Path("data")
ARCHIVE_PATH = DATA_DIR / "ml-latest-small.zip"
EXTRACTED_DIR = DATA_DIR / "ml-latest-small"
ARTIFACT_DIR = Path("artifacts")
ARTIFACT_PATH = ARTIFACT_DIR / "recommendation_state.json"
LIKE_THRESHOLD = 4.0
def download_and_extract():
DATA_DIR.mkdir(parents=True, exist_ok=True)
if not ARCHIVE_PATH.exists():
print("Downloading MovieLens Latest Small...")
with urlopen(DATA_URL, timeout=30) as response:
ARCHIVE_PATH.write_bytes(response.read())What does this section do?
- The imports provide CSV parsing, JSON serialization, counting, paths, downloading, and ZIP extraction using the Python standard library.
- The path constants keep the downloaded archive, extracted files, and generated artifact in predictable locations.
- The LIKE_THRESHOLD value treats ratings of 4.0 or higher as positive signals for collaborative recommendations.
- The first branch downloads the archive only when data/ml-latest-small.zip is missing.
- Save prepare.py.
- Confirm that the saved module exposes its artifact path by running this command:
python3 -c "print(__import__('prepare').ARTIFACT_PATH)"What does this check prove?
The command imports your saved module and prints the configured artifact path. A successful import confirms that Python can read the first code section.
You should see artifacts/recommendation_state.json in the terminal.
Cannot import prepare?
Confirm that the integrated terminal is inside the streamflix folder. Check that the file is named prepare.py with lowercase letters.
If Python points to a syntax problem, compare the imports and quotation marks with the code above.
For focused help, help me fix the first section of prepare.py.
The next section finishes acquisition and begins parsing the movie catalog. Keeping these operations offline means the future API never waits for a download.
- Place your cursor after ARCHIVE_PATH.write_bytes(response.read()) in prepare.py.
- Finish download_and_extract() and start prepare_state() by pasting this code:
movies_path = EXTRACTED_DIR / "movies.csv"
ratings_path = EXTRACTED_DIR / "ratings.csv"
if not movies_path.exists() or not ratings_path.exists():
print("Extracting MovieLens archive...")
with ZipFile(ARCHIVE_PATH, "r") as archive:
archive.extractall(DATA_DIR)
return movies_path, ratings_path
def prepare_state(movies_path, ratings_path):
movies = {}
with movies_path.open("r", encoding="utf-8", newline="") as movie_file:
for row in csv.DictReader(movie_file):
genres = []
if row["genres"] != "(no genres listed)":
genres = row["genres"].split("|")
movies[row["movieId"]] = {
"title": row["title"],
"genres": genres,
}How does acquisition work?
- The extraction check looks for movies.csv and ratings.csv before unpacking the trusted GroupLens archive.
- The returned paths give the preparation stage direct references to the two source files it needs.
- The movie loop converts each catalog row into a dictionary containing its title and genre list.
- Save prepare.py.
- Download and extract the official dataset by running this command:
python3 -c "print(__import__('prepare').download_and_extract())"What does this command do?
The command imports prepare.py and calls download_and_extract() directly. It prints the two extracted paths returned by the function.
You should see paths ending in data/ml-latest-small/movies.csv and data/ml-latest-small/ratings.csv. The Explorer sidebar should also show the downloaded archive and extracted folder.
Download or extraction failed?
Check that your Mac has an active internet connection. Confirm that DATA_URL matches the code above.
If extraction fails after an interrupted download, remove the incomplete archive from the data folder before rerunning the command.
For help with the failure, help me troubleshoot the MovieLens download or ZIP extraction.
Ratings provide the user behavior needed for personalization. This next section records every watched movie while separately collecting ratings that meet the project-owned like threshold.
- Place your cursor after the closing brace of the movies[row["movieId"]] assignment.
- Add the rating aggregation section by pasting this code:
rating_count = Counter()
rating_sum = defaultdict(float)
seen_by_user = defaultdict(set)
liked_by_user = defaultdict(set)
with ratings_path.open("r", encoding="utf-8", newline="") as rating_file:
for row in csv.DictReader(rating_file):
user_id = row["userId"]
movie_id = row["movieId"]
rating = float(row["rating"])
seen_by_user[user_id].add(movie_id)
liked_by_user[user_id]
rating_count[movie_id] += 1
rating_sum[movie_id] += rating
if rating >= LIKE_THRESHOLD:
liked_by_user[user_id].add(movie_id)What do these indexes capture?
- The rating counters preserve the volume and sum needed to calculate each movie's popularity score.
- The seen_by_user index records every rated title so serving can filter watched movies.
- The liked_by_user index records titles rated at least 4.0 for later neighbor matching.
- Save prepare.py.
- Confirm that Visual Studio Code shows no red syntax markers from rating_count through liked_by_user.
Seeing syntax markers?
Check that every line in this section stays indented inside prepare_state(). Confirm that the final if block uses one additional indentation level.
If the markers remain, help me fix indentation in the ratings section.
Raw rating counts still need to become serving signals. Popularity rewards well-rated movies with more evidence, while inverse liker and genre indexes support future candidate generation and ranking.
- Place your cursor after liked_by_user[user_id].add(movie_id).
- Add popularity and preference aggregation by pasting this code:
popularity = []
for movie_id, count in rating_count.items():
average = rating_sum[movie_id] / count
score = average * math.log1p(count)
popularity.append(
{
"movie_id": movie_id,
"score": round(score, 6),
"rating_count": count,
"average_rating": round(average, 3),
}
)
popularity.sort(
key=lambda item: (item["score"], item["rating_count"]),
reverse=True,
)
likers_by_movie = defaultdict(list)
genre_preferences = {}
for user_id, liked_movie_ids in liked_by_user.items():
genre_counts = Counter()
for movie_id in liked_movie_ids:
likers_by_movie[movie_id].append(user_id)
for genre in movies[movie_id]["genres"]:
genre_counts[genre] += 1
genre_preferences[user_id] = dict(genre_counts)How are the serving signals prepared?
- The popularity score combines average rating with the amount of rating evidence using math.log1p().
- The likers_by_movie index reverses the user-like mapping so the online engine can find neighbors from shared movies.
- The genre_preferences index counts the genres among each user's liked movies for the hybrid ranker.
- Save prepare.py.
- Confirm that Python can compile the expanded function by running this command:
python3 -c "print(len(__import__('prepare').prepare_state.__code__.co_code))"What does this check prove?
The command imports the saved function and prints the size of its compiled instructions. A positive number confirms that Python accepted the function structure added so far.
You should see a positive integer in the terminal. That result confirms the saved function has valid Python syntax.
Function does not compile?
Check the closing parentheses around popularity.append() and popularity.sort(). Confirm that every nested loop uses consistent four-space indentation.
For help tracing the syntax issue, help me debug the popularity and genre aggregation section.
The final preparation structure turns the in-memory dictionaries into one serializable contract. Sorted identifier lists make the artifact stable enough to inspect and straightforward for the online engine to load.
- Place your cursor after genre_preferences[user_id] = dict(genre_counts).
- Assemble and return the prepared state by pasting this code:
state = {
"metadata": {
"source": "MovieLens Latest Small",
"source_url": DATA_URL,
"like_threshold": LIKE_THRESHOLD,
"movie_count": len(movies),
"user_count": len(seen_by_user),
},
"movies": movies,
"popularity": popularity,
"seen_by_user": {
user_id: sorted(movie_ids, key=int)
for user_id, movie_ids in seen_by_user.items()
},
"liked_by_user": {
user_id: sorted(movie_ids, key=int)
for user_id, movie_ids in liked_by_user.items()
},
"likers_by_movie": {
movie_id: sorted(user_ids, key=int)
for movie_id, user_ids in likers_by_movie.items()
},
"genre_preferences": genre_preferences,
}
return stateWhat becomes part of the contract?
- The metadata object records the source, threshold, movie count, and user count.
- The movies and popularity entries provide catalog details and a stable fallback order.
- The user-like, inverse-liker, seen-item, and genre-preference entries provide direct keyed lookups for online recommendation work.
- Save prepare.py.
- Inspect the returned top-level keys by running this command:
python3 -c "print(__import__('prepare').prepare_state(*__import__('prepare').download_and_extract()).keys())"Why inspect the keys now?
This command runs the completed state builder in memory and prints its top-level contract. It confirms that each required index exists before the program writes the artifact.
You should see metadata, movies, popularity, seen_by_user, liked_by_user, likers_by_movie, and genre_preferences in the terminal output.
Missing a state key?
Compare the state dictionary with the code above. Check that every dictionary entry has a comma after it.
If the command points to a missing name, confirm that the earlier section defines that identifier with the same spelling.
For help comparing the contract, help me find the missing recommendation state key.
Run the job and inspect the contract
The state builder now produces every required index in memory. The entry point connects acquisition, preparation, and serialization so one command can regenerate the offline artifact.
- Place your cursor after return state in prepare.py.
- Add the program entry point by pasting this code:
def main():
movies_path, ratings_path = download_and_extract()
state = prepare_state(movies_path, ratings_path)
ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)
ARTIFACT_PATH.write_text(
json.dumps(state, ensure_ascii=False),
encoding="utf-8",
)
print(
"Prepared "
f"{state['metadata']['movie_count']} movies for "
f"{state['metadata']['user_count']} users"
)
print(f"Wrote {ARTIFACT_PATH}")
if __name__ == "__main__":
main()How does the entry point finish the job?
- The main() function obtains the source paths and passes them into prepare_state().
- The artifact directory is created before the state is serialized to artifacts/recommendation_state.json.
- The final messages expose the processed movie count, user count, and artifact path as concrete success signals.
- Save prepare.py.
Before you run this, picture the two terminal lines that would prove the offline job finished.
- Generate the recommendation index by running this command:
python3 prepare.pyWhat does this command do?
Python executes the saved prepare.py entry point. The program reuses the downloaded files when they already exist, rebuilds the indexes, and writes the serving artifact.
You should see a line reporting non-zero movie and user counts. The next line should say Wrote artifacts/recommendation_state.json.
That is the offline boundary working. StreamFlix now has prepared state that future requests can read without scanning raw ratings.
Artifact was not written?
Confirm that the terminal is inside streamflix and that prepare.py is saved. Check the first traceback line that points into your file.
If the message mentions a missing CSV file, rerun the download check from the previous substep before running the full job again.
For help with the traceback, help me debug the failed StreamFlix preparation job.
- Expand the artifacts folder in the Explorer sidebar.
- Select recommendation_state.json.
- Press Cmd+F to open the editor search field.
- Search for metadata.
You will find the source name, source URL, 4.0 like threshold, and non-zero counts inside metadata. The terminal key check confirms that the same artifact also contains every serving index.
Why is this artifact a contract?
Offline preparation owns downloading, CSV parsing, aggregation, and index construction. Online serving receives a predictable JSON shape with direct lookups for movies, popularity, user history, collaborative neighbors, and genre preferences.
That boundary lets the next service load one prepared snapshot at startup. Request handling stays focused on selecting and ranking recommendations.
✔️ Awesome, I've got everything!
Your dataset, preparation program, and recommendation artifact are ready. Double check that all four project files are saved before moving on.
ⓧ I'd like to double check the full code
Compare your complete prepare.py with this reference.
import csv
import json
import math
from collections import Counter, defaultdict
from pathlib import Path
from urllib.request import urlopen
from zipfile import ZipFile
DATA_URL = "https://files.grouplens.org/datasets/movielens/ml-latest-small.zip"
DATA_DIR = Path("data")
ARCHIVE_PATH = DATA_DIR / "ml-latest-small.zip"
EXTRACTED_DIR = DATA_DIR / "ml-latest-small"
ARTIFACT_DIR = Path("artifacts")
ARTIFACT_PATH = ARTIFACT_DIR / "recommendation_state.json"
LIKE_THRESHOLD = 4.0
def download_and_extract():
DATA_DIR.mkdir(parents=True, exist_ok=True)
if not ARCHIVE_PATH.exists():
print("Downloading MovieLens Latest Small...")
with urlopen(DATA_URL, timeout=30) as response:
ARCHIVE_PATH.write_bytes(response.read())
movies_path = EXTRACTED_DIR / "movies.csv"
ratings_path = EXTRACTED_DIR / "ratings.csv"
if not movies_path.exists() or not ratings_path.exists():
print("Extracting MovieLens archive...")
with ZipFile(ARCHIVE_PATH, "r") as archive:
archive.extractall(DATA_DIR)
return movies_path, ratings_path
def prepare_state(movies_path, ratings_path):
movies = {}
with movies_path.open("r", encoding="utf-8", newline="") as movie_file:
for row in csv.DictReader(movie_file):
genres = []
if row["genres"] != "(no genres listed)":
genres = row["genres"].split("|")
movies[row["movieId"]] = {
"title": row["title"],
"genres": genres,
}
rating_count = Counter()
rating_sum = defaultdict(float)
seen_by_user = defaultdict(set)
liked_by_user = defaultdict(set)
with ratings_path.open("r", encoding="utf-8", newline="") as rating_file:
for row in csv.DictReader(rating_file):
user_id = row["userId"]
movie_id = row["movieId"]
rating = float(row["rating"])
seen_by_user[user_id].add(movie_id)
liked_by_user[user_id]
rating_count[movie_id] += 1
rating_sum[movie_id] += rating
if rating >= LIKE_THRESHOLD:
liked_by_user[user_id].add(movie_id)
popularity = []
for movie_id, count in rating_count.items():
average = rating_sum[movie_id] / count
score = average * math.log1p(count)
popularity.append(
{
"movie_id": movie_id,
"score": round(score, 6),
"rating_count": count,
"average_rating": round(average, 3),
}
)
popularity.sort(
key=lambda item: (item["score"], item["rating_count"]),
reverse=True,
)
likers_by_movie = defaultdict(list)
genre_preferences = {}
for user_id, liked_movie_ids in liked_by_user.items():
genre_counts = Counter()
for movie_id in liked_movie_ids:
likers_by_movie[movie_id].append(user_id)
for genre in movies[movie_id]["genres"]:
genre_counts[genre] += 1
genre_preferences[user_id] = dict(genre_counts)
state = {
"metadata": {
"source": "MovieLens Latest Small",
"source_url": DATA_URL,
"like_threshold": LIKE_THRESHOLD,
"movie_count": len(movies),
"user_count": len(seen_by_user),
},
"movies": movies,
"popularity": popularity,
"seen_by_user": {
user_id: sorted(movie_ids, key=int)
for user_id, movie_ids in seen_by_user.items()
},
"liked_by_user": {
user_id: sorted(movie_ids, key=int)
for user_id, movie_ids in liked_by_user.items()
},
"likers_by_movie": {
movie_id: sorted(user_ids, key=int)
for movie_id, user_ids in likers_by_movie.items()
},
"genre_preferences": genre_preferences,
}
return state
def main():
movies_path, ratings_path = download_and_extract()
state = prepare_state(movies_path, ratings_path)
ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)
ARTIFACT_PATH.write_text(
json.dumps(state, ensure_ascii=False),
encoding="utf-8",
)
print(
"Prepared "
f"{state['metadata']['movie_count']} movies for "
f"{state['metadata']['user_count']} users"
)
print(f"Wrote {ARTIFACT_PATH}")
if __name__ == "__main__":
main()The recommender.py, app.py, and ARCHITECTURE.md scaffold files should still be empty.
Your offline pipeline now converts raw MovieLens ratings into a serving-ready recommendation index. Next, you will load that artifact once and expose StreamFlix's first popularity-ranked API response.
Serve the Popularity Baseline
Your offline job has turned raw MovieLens ratings into a prepared JSON contract. The request path can now read that state without rescanning CSV files.
This step turns that contract into a local recommendation API with a popularity baseline. You will test whether changing the user changes the ranking.
In this step, get ready to:
- Load the prepared recommendation index once at startup.
- Expose local routes for health checks plus popularity responses.
- Compare the baseline output for two user IDs.
Load the prepared state once
The prepared artifact forms the boundary between offline computation and online serving. Loading it during startup keeps raw ratings scans out of each request.
- Select recommender.py from the VS Code file list.
- Replace the scaffold text with the state-loading code below:
import json
import math
from collections import defaultdict
from pathlib import Path
from time import perf_counter
class RecommendationEngine:
def __init__(self, artifact_path="artifacts/recommendation_state.json"):
with Path(artifact_path).open("r", encoding="utf-8") as state_file:
self.state = json.load(state_file)
self.movies = self.state["movies"]
self.popularity = self.state["popularity"]
self.seen_by_user = self.state["seen_by_user"]
self.liked_by_user = self.state["liked_by_user"]
self.likers_by_movie = self.state["likers_by_movie"]
self.genre_preferences = self.state["genre_preferences"]
self.popularity_score = {
item["movie_id"]: item["score"] for item in self.popularity
}
self.max_popularity_score = max(self.popularity_score.values())
self.metrics = {
"request_count": 0,
"fallback_count": 0,
"strategy_counts": defaultdict(int),
"total_latency_ms": 0.0,
"last_candidate_count": 0,
"last_fallback_reason": None,
}What does this code do?
- The constructor reads artifacts/recommendation_state.json when RecommendationEngine starts.
- The index keys become attributes that serving methods can access without reopening the artifact.
- The popularity score lookup prepares fast access to each movie's precomputed score.
- The metrics dictionary creates a home for serving counters used by the running engine.
- Save recommender.py.
- Confirm RecommendationEngine now contains __init__() as its first method.
Does the class indentation look wrong?
Check that every assignment from self.movies through self.metrics remains inside __init__().
Ask for help with checking the RecommendationEngine indentation.
The loaded indexes still need a consistent response shape. A formatter gives every ranked movie the same fields.
- Add the following methods inside RecommendationEngine immediately after __init__():
def _format_movie(self, movie_id, score, reason):
movie = self.movies[movie_id]
return {
"movie_id": movie_id,
"title": movie["title"],
"genres": movie["genres"],
"score": round(score, 4),
"reason": reason,
}
def baseline(self, user_id, n=10):
recommendations = []
for item in self.popularity[:n]:
recommendations.append(
self._format_movie(
item["movie_id"],
item["score"],
"global_popularity",
)
)
return {
"user_id": str(user_id),
"strategy": "popularity_baseline",
"recommendations": recommendations,
}How does the baseline work?
- The _format_movie() helper combines catalog details with the ranking score plus its reason.
- The baseline() method selects the first n entries from the prepared popularity list.
- The response labels the strategy as popularity_baseline.
- Each movie records global_popularity as its ranking reason.
- Save recommender.py.
- Confirm the file now contains _format_movie() followed by baseline().
Are the methods outside the class?
Check that _format_movie() plus baseline() align with __init__() inside RecommendationEngine.
Ask for help with placing the baseline methods inside the class.
Serve the baseline over HTTP
A local HTTP server turns the engine into something a browser can call. The server creates one engine instance before it starts accepting requests.
- Select app.py from the VS Code file list.
- Replace the scaffold text with the server setup below:
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
from recommender import RecommendationEngine
HOST = "127.0.0.1"
PORT = 8000
ENGINE = RecommendationEngine()
class RecommendationHandler(BaseHTTPRequestHandler):
def send_json(self, payload, status=200):
body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
@staticmethod
def parse_n(parameters):
try:
value = int(parameters.get("n", ["10"])[0])
except ValueError:
value = 10
return max(1, min(value, 20))What does this server setup do?
- The global ENGINE creates one RecommendationEngine during startup.
- The send_json() method serializes response data into an indented JSON body.
- The parse_n() method converts the requested result count into a number between 1 plus 20.
- The server binds to local host 127.0.0.1 on port 8000.
- Save app.py.
- Confirm ENGINE = RecommendationEngine() appears before RecommendationHandler.
Is the artifact missing at startup?
Confirm the VS Code terminal still points to the streamflix folder. Confirm artifacts/recommendation_state.json remains inside that folder.
Ask for help with finding the prepared recommendation artifact.
The request handler now needs routes that connect browser requests to the shared engine. A health route confirms the process is responsive.
- Add the following routing plus startup code below parse_n():
def do_GET(self):
request = urlsplit(self.path)
parameters = parse_qs(request.query)
n = self.parse_n(parameters)
if request.path == "/health":
self.send_json({"status": "ok"})
return
if request.path == "/baseline":
user_id = parameters.get("user_id", ["1"])[0]
self.send_json(ENGINE.baseline(user_id, n))
return
if __name__ == "__main__":
server = ThreadingHTTPServer((HOST, PORT), RecommendationHandler)
print(f"StreamFlix listening at http://{HOST}:{PORT}")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopping StreamFlix")
finally:
server.server_close()How do the routes connect?
- The do_GET() method separates the request path from its query parameters.
- The /health route returns a small status response.
- The /baseline route passes user_id plus n to the shared engine.
- The final block starts ThreadingHTTPServer until you stop the process.
- Save app.py.
- Confirm do_GET() remains inside RecommendationHandler.
Does the startup block look nested?
Check that if __name__ == "__main__": begins at the far left edge. This keeps server startup outside RecommendationHandler.
Ask for help with checking the app.py class boundary.
- Start StreamFlix from the existing VS Code integrated terminal by running this command:
python3 app.pyWhat should the terminal show?
You should see StreamFlix listening at http://127.0.0.1:8000. The process keeps this terminal occupied while it handles browser requests.
Use Control-C when you need to stop the local server.
- Leave the server running in the current terminal.
- Open http://127.0.0.1:8000/health in a browser.
You should see a JSON response with status set to ok. This proves the local process is accepting requests.
Can the browser reach the server?
Confirm the terminal still shows the running StreamFlix process. If the process stopped then run python3 app.py again.
Ask for help with connecting to the local StreamFlix server.
Before you load the baseline, what fields do you expect each ranked movie to include?
- Open http://127.0.0.1:8000/baseline?user_id=1&n=10 in the browser.
You should see ten ranked movie objects. The response sets strategy to popularity_baseline.
Each movie includes its title plus genres. Its reason is global_popularity.
Compare two user IDs
The endpoint identifies which user made the request. This check reveals whether that input influences the ranking.
Before you send the second request, do you expect the movie order to change?
- Change user_id=1 to user_id=2 in the browser address bar.
- Press Enter to send the second request.
You should see user_id change to 2. The ten movie titles remain in the same order.
That result is valuable. Your working API now exposes the baseline's personalization gap.
Why does the order stay unchanged?
The baseline() method echoes user_id so the response identifies the request. Its recommendations come directly from self.popularity[:n].
Global popularity gives each user the same ordering. This makes the baseline useful as a stable fallback.
✔️ Awesome, I've got everything!
Your local API now loads its prepared state once. It serves a ten-movie popularity row for both test users.
ⓧ I'd like to double check the full code
Compare your complete recommender.py with this version.
import json
import math
from collections import defaultdict
from pathlib import Path
from time import perf_counter
class RecommendationEngine:
def __init__(self, artifact_path="artifacts/recommendation_state.json"):
with Path(artifact_path).open("r", encoding="utf-8") as state_file:
self.state = json.load(state_file)
self.movies = self.state["movies"]
self.popularity = self.state["popularity"]
self.seen_by_user = self.state["seen_by_user"]
self.liked_by_user = self.state["liked_by_user"]
self.likers_by_movie = self.state["likers_by_movie"]
self.genre_preferences = self.state["genre_preferences"]
self.popularity_score = {
item["movie_id"]: item["score"] for item in self.popularity
}
self.max_popularity_score = max(self.popularity_score.values())
self.metrics = {
"request_count": 0,
"fallback_count": 0,
"strategy_counts": defaultdict(int),
"total_latency_ms": 0.0,
"last_candidate_count": 0,
"last_fallback_reason": None,
}
def _format_movie(self, movie_id, score, reason):
movie = self.movies[movie_id]
return {
"movie_id": movie_id,
"title": movie["title"],
"genres": movie["genres"],
"score": round(score, 4),
"reason": reason,
}
def baseline(self, user_id, n=10):
recommendations = []
for item in self.popularity[:n]:
recommendations.append(
self._format_movie(
item["movie_id"],
item["score"],
"global_popularity",
)
)
return {
"user_id": str(user_id),
"strategy": "popularity_baseline",
"recommendations": recommendations,
}Compare your complete app.py with this version.
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
from recommender import RecommendationEngine
HOST = "127.0.0.1"
PORT = 8000
ENGINE = RecommendationEngine()
class RecommendationHandler(BaseHTTPRequestHandler):
def send_json(self, payload, status=200):
body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
@staticmethod
def parse_n(parameters):
try:
value = int(parameters.get("n", ["10"])[0])
except ValueError:
value = 10
return max(1, min(value, 20))
def do_GET(self):
request = urlsplit(self.path)
parameters = parse_qs(request.query)
n = self.parse_n(parameters)
if request.path == "/health":
self.send_json({"status": "ok"})
return
if request.path == "/baseline":
user_id = parameters.get("user_id", ["1"])[0]
self.send_json(ENGINE.baseline(user_id, n))
return
if __name__ == "__main__":
server = ThreadingHTTPServer((HOST, PORT), RecommendationHandler)
print(f"StreamFlix listening at http://{HOST}:{PORT}")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopping StreamFlix")
finally:
server.server_close()Your baseline API is live. Next, you will turn its identical rankings into a measurable Top-N overlap result.
Measure the Personalization Failure
Your popularity baseline now returns a ranked movie row from the prepared index. The API gives every user the same order.
A visual comparison can feel subjective. Top-N overlap turns that sameness into a reproducible measurement.
In this step, get ready to:
- Calculate the overlap between two ranked movie lists.
- Expose the measurement through the comparison endpoint.
- Explain why identical recommendations fail to represent different tastes.
Define the Top-N overlap measurement
Top-N overlap divides the number of shared movie IDs by the length of the shorter list. The result ranges from 0.0 for no shared items to 1.0 for complete overlap.
- Stop the running API by pressing Control-C in the VS Code integrated terminal.
- In the VS Code Explorer sidebar, select recommender.py.
- Place this method directly below baseline():
@staticmethod
def top_n_overlap(first_ids, second_ids):
denominator = min(len(first_ids), len(second_ids))
if denominator == 0:
return 0.0
return len(set(first_ids) & set(second_ids)) / denominatorWhat Does This Method Measure?
- The denominator uses the shorter list so the calculation remains valid when the requested row contains fewer items.
- The two sets reveal which movie IDs appear in both lists.
- An empty input returns 0.0 instead of causing a division error.
- Save recommender.py.
The measurement needs two active users plus their baseline movie IDs. The comparison method gathers those inputs before returning the score with both title lists.
- Place this method directly below top_n_overlap():
def compare_users(self, n=10):
eligible_users = sorted(
[
user_id
for user_id, likes in self.liked_by_user.items()
if len(likes) >= 10
],
key=int,
)[:40]
user_a, user_b = eligible_users[:2]
baseline_a = self.baseline(user_a, n)["recommendations"]
baseline_b = self.baseline(user_b, n)["recommendations"]
baseline_a_ids = [item["movie_id"] for item in baseline_a]
baseline_b_ids = [item["movie_id"] for item in baseline_b]
return {
"user_a": user_a,
"user_b": user_b,
"n": n,
"baseline_overlap": round(
self.top_n_overlap(baseline_a_ids, baseline_b_ids),
3,
),
"baseline_user_a_titles": [item["title"] for item in baseline_a],
"baseline_user_b_titles": [item["title"] for item in baseline_b],
}How Does the Comparison Work?
- The active-user filter selects MovieLens users with at least ten liked movies.
- Each call to baseline() collects a popularity-ranked row for one selected user.
- The method measures shared movie IDs before returning readable titles for inspection.
- Save recommender.py.
- Check that the updated recommendation engine imports successfully by restarting the API with this command:
python3 app.pyWhat Does This Check Prove?
Starting the API imports the updated RecommendationEngine class. It also loads artifacts/recommendation_state.json through the existing startup path.
You should see StreamFlix listening at http://127.0.0.1:8000. Your new methods now load without a syntax error.
Seeing a Python Error?
Check that top_n_overlap() has the same class-level indentation as baseline(). Make sure every line inside compare_users() remains indented beneath its definition.
Ask for help with the exact terminal output if the import still fails: help me debug the new comparison methods in recommender.py
- Stop the API again by pressing Control-C.
Expose the comparison endpoint
The engine can now calculate the measurement. The HTTP handler needs a route that connects a browser request to compare_users().
- In the VS Code Explorer sidebar, select app.py.
- Inside do_GET(), find this baseline route:
if request.path == "/baseline":
user_id = parameters.get("user_id", ["1"])[0]
self.send_json(ENGINE.baseline(user_id, n))
returnWhere Does the New Route Belong?
The baseline branch already uses the parsed n value. The comparison branch belongs directly below it because both routes return recommendation measurements.
- Add the comparison branch directly below the baseline branch so this section now looks like:
if request.path == "/baseline":
user_id = parameters.get("user_id", ["1"])[0]
self.send_json(ENGINE.baseline(user_id, n))
return
if request.path == "/compare":
self.send_json(ENGINE.compare_users(n))
returnWhat Does the Route Add?
- The path check matches requests sent to /compare.
- The handler passes the validated row size to compare_users().
- The existing send_json() method serializes the comparison result for the browser.
- Save app.py.
- Start the updated API by running this command in the integrated terminal:
python3 app.pyWhat Does This Command Start?
The command starts the local HTTP process with the updated route table. The terminal remains occupied while StreamFlix listens for browser requests.
You should see the StreamFlix listening message again. The comparison route is now ready for a request.
Is the Server Failing to Start?
A previous API process may still be using port 8000. Return to that process to stop it before running the command again.
If the terminal points to app.py, compare the indentation of the new branch with the baseline branch. help me debug the StreamFlix comparison route
✔️ Awesome, I've got everything!
Great. Your recommendation engine can calculate Top-N overlap. Your API can now return that measurement through /compare.
ⓧ I'd like to double check the full code
Compare your cumulative recommender.py with this version:
import json
import math
from collections import defaultdict
from pathlib import Path
from time import perf_counter
class RecommendationEngine:
def __init__(self, artifact_path="artifacts/recommendation_state.json"):
with Path(artifact_path).open("r", encoding="utf-8") as state_file:
self.state = json.load(state_file)
self.movies = self.state["movies"]
self.popularity = self.state["popularity"]
self.seen_by_user = self.state["seen_by_user"]
self.liked_by_user = self.state["liked_by_user"]
self.likers_by_movie = self.state["likers_by_movie"]
self.genre_preferences = self.state["genre_preferences"]
self.popularity_score = {
item["movie_id"]: item["score"] for item in self.popularity
}
self.max_popularity_score = max(self.popularity_score.values())
self.metrics = {
"request_count": 0,
"fallback_count": 0,
"strategy_counts": defaultdict(int),
"total_latency_ms": 0.0,
"last_candidate_count": 0,
"last_fallback_reason": None,
}
def _format_movie(self, movie_id, score, reason):
movie = self.movies[movie_id]
return {
"movie_id": movie_id,
"title": movie["title"],
"genres": movie["genres"],
"score": round(score, 4),
"reason": reason,
}
def baseline(self, user_id, n=10):
recommendations = []
for item in self.popularity[:n]:
recommendations.append(
self._format_movie(
item["movie_id"],
item["score"],
"global_popularity",
)
)
return {
"user_id": str(user_id),
"strategy": "popularity_baseline",
"recommendations": recommendations,
}
@staticmethod
def top_n_overlap(first_ids, second_ids):
denominator = min(len(first_ids), len(second_ids))
if denominator == 0:
return 0.0
return len(set(first_ids) & set(second_ids)) / denominator
def compare_users(self, n=10):
eligible_users = sorted(
[
user_id
for user_id, likes in self.liked_by_user.items()
if len(likes) >= 10
],
key=int,
)[:40]
user_a, user_b = eligible_users[:2]
baseline_a = self.baseline(user_a, n)["recommendations"]
baseline_b = self.baseline(user_b, n)["recommendations"]
baseline_a_ids = [item["movie_id"] for item in baseline_a]
baseline_b_ids = [item["movie_id"] for item in baseline_b]
return {
"user_a": user_a,
"user_b": user_b,
"n": n,
"baseline_overlap": round(
self.top_n_overlap(baseline_a_ids, baseline_b_ids),
3,
),
"baseline_user_a_titles": [item["title"] for item in baseline_a],
"baseline_user_b_titles": [item["title"] for item in baseline_b],
}What Should This File Contain?
This cumulative file keeps the existing state loader plus the popularity baseline. It adds only the overlap calculation and baseline comparison required in this step.
Compare your cumulative app.py with this version:
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
from recommender import RecommendationEngine
HOST = "127.0.0.1"
PORT = 8000
ENGINE = RecommendationEngine()
class RecommendationHandler(BaseHTTPRequestHandler):
def send_json(self, payload, status=200):
body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
@staticmethod
def parse_n(parameters):
try:
value = int(parameters.get("n", ["10"])[0])
except ValueError:
value = 10
return max(1, min(value, 20))
def do_GET(self):
request = urlsplit(self.path)
parameters = parse_qs(request.query)
n = self.parse_n(parameters)
if request.path == "/health":
self.send_json({"status": "ok"})
return
if request.path == "/baseline":
user_id = parameters.get("user_id", ["1"])[0]
self.send_json(ENGINE.baseline(user_id, n))
return
if request.path == "/compare":
self.send_json(ENGINE.compare_users(n))
return
if __name__ == "__main__":
server = ThreadingHTTPServer((HOST, PORT), RecommendationHandler)
print(f"StreamFlix listening at http://{HOST}:{PORT}")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopping StreamFlix")
finally:
server.server_close()What Should This File Contain?
This cumulative handler keeps the existing health and baseline routes. It adds the comparison branch before starting the same local server.
Measure the identical Top-N lists
The comparison endpoint selects two active MovieLens users. It then measures the baseline rows produced for those users.
Before you check, consider what overlap value two identically ordered Top-N lists should produce.
- Switch back to the browser from earlier.
- Enter http://127.0.0.1:8000/compare?n=10 in the address bar.
- Press Enter to request the comparison.
You should see different values for user_a and user_b. The baseline_user_a_titles and baseline_user_b_titles arrays contain the same ten titles in the same order.
The baseline_overlap value is 1.0. Popularity remains useful as a safety net, but identical Top-N output cannot express different user tastes.
You have turned the baseline's sameness into measurable evidence. Next, you will add collaborative candidates and rank different movies for different users.
Add Collaborative Candidates and Hybrid Ranking
Your comparison endpoint proved that the popularity baseline gives two different users the same movie order. A recommendation row needs a user-specific signal before it can reflect individual taste.
In this step, you will add collaborative filtering to find movies liked by similar users. You will use hybrid ranking to balance that personal signal with popularity plus genre affinity.
In this step, get ready to:
- Generate unseen movie candidates from users with similar likes.
- Rank candidates with collaborative, popularity, and genre signals.
- Measure the drop in Top-N overlap between two users.
Generate collaborative candidates
A collaborative candidate generator starts with movies the target user likes. It finds other users who liked those movies before collecting unseen titles from the strongest neighbors.
Why separate candidates from ranking?
Candidate generation narrows a large catalog to plausible choices. Ranking then decides the order using richer signals.
This boundary lets each stage evolve independently as the service grows.
- Switch back to the VS Code terminal running StreamFlix.
- Stop the local server by pressing Control+C.
- Select recommender.py in the VS Code Explorer sidebar.
- Replace the import section at the top of recommender.py with this code:
import json
import math
from collections import defaultdict
from pathlib import Path
from time import perf_counterWhat do these imports support?
- The math module supplies the square root used by binary cosine similarity.
- The defaultdict type accumulates candidate scores without manual initialization.
- The perf_counter timer measures recommendation latency for response metadata.
- Find the line that assigns self.genre_preferences inside __init__().
- Add these cached popularity values plus serving counters below that line:
self.popularity_score = {
item["movie_id"]: item["score"] for item in self.popularity
}
self.max_popularity_score = max(self.popularity_score.values())
self.metrics = {
"request_count": 0,
"fallback_count": 0,
"strategy_counts": defaultdict(int),
"total_latency_ms": 0.0,
"last_candidate_count": 0,
"last_fallback_reason": None,
}What does this state hold?
- The popularity lookup makes each candidate score available by movie ID.
- The maximum popularity score provides the scale used during normalization.
- The metrics dictionary prepares the engine to record strategy usage plus fallback behavior.
- Add _genre_affinity() below baseline() by pasting this method:
def _genre_affinity(self, user_id, movie_id):
preferences = self.genre_preferences.get(user_id, {})
if not preferences:
return 0.0
maximum = max(preferences.values())
genres = self.movies[movie_id]["genres"]
if not genres or maximum == 0:
return 0.0
return max(preferences.get(genre, 0) for genre in genres) / maximumHow is genre affinity calculated?
The method finds the strongest genre preference for the user. It returns a normalized match between 0.0 and 1.0 for the candidate movie.
- Add _collaborative_candidates() directly below _genre_affinity() by pasting this method:
def _collaborative_candidates(self, user_id):
target_likes = set(self.liked_by_user[user_id])
seen = set(self.seen_by_user[user_id])
neighbor_ids = set()
for movie_id in target_likes:
neighbor_ids.update(self.likers_by_movie.get(movie_id, []))
neighbor_ids.discard(user_id)
neighbor_scores = []
for neighbor_id in neighbor_ids:
neighbor_likes = set(self.liked_by_user[neighbor_id])
common = len(target_likes & neighbor_likes)
denominator = math.sqrt(len(target_likes) * len(neighbor_likes))
similarity = common / denominator if denominator else 0.0
if similarity > 0:
neighbor_scores.append((neighbor_id, similarity))
neighbor_scores.sort(key=lambda pair: pair[1], reverse=True)
candidate_scores = defaultdict(float)
for neighbor_id, similarity in neighbor_scores[:50]:
for movie_id in self.liked_by_user[neighbor_id]:
if movie_id not in seen:
candidate_scores[movie_id] += similarity
return dict(candidate_scores)What does candidate generation do?
- Binary cosine similarity compares the overlap between two users' liked-movie sets.
- The strongest 50 neighbors contribute movies to the candidate pool.
- The seen-item check removes titles the target user has already rated.
- Repeated recommendations from similar neighbors accumulate stronger collaborative scores.
- Save recommender.py.
Blend the ranking signals
The candidate score captures neighbor agreement. The final ranker combines it with stable popularity plus the user's prepared genre preferences.
How does the hybrid score work?
The local heuristic gives normalized collaborative evidence a weight of 0.70. Normalized popularity receives 0.20.
Genre affinity receives the remaining 0.10. These project-owned weights keep personalization dominant while preserving a stable quality signal.
- Add the first part of _rank_personalized() below _collaborative_candidates().
- Keep the method open for the next code block:
def _rank_personalized(self, user_id, candidate_scores, n):
max_collaborative = max(candidate_scores.values())
strongest_candidates = sorted(
candidate_scores.items(),
key=lambda pair: pair[1],
reverse=True,
)[:250]
ranked = []
for movie_id, collaborative_score in strongest_candidates:
normalized_collaborative = collaborative_score / max_collaborative
normalized_popularity = (
self.popularity_score.get(movie_id, 0.0)
/ self.max_popularity_score
)
genre_affinity = self._genre_affinity(user_id, movie_id)
final_score = (
0.70 * normalized_collaborative
+ 0.20 * normalized_popularity
+ 0.10 * genre_affinity
)How are candidates normalized?
The ranker keeps the strongest 250 candidates before scoring. Each signal is normalized so one raw scale cannot dominate the blend.
- Continue _rank_personalized() immediately below the final_score calculation with this code:
ranked.append(
self._format_movie(
movie_id,
final_score,
{
"collaborative": round(normalized_collaborative, 4),
"popularity": round(normalized_popularity, 4),
"genre_affinity": round(genre_affinity, 4),
},
)
)
ranked.sort(key=lambda item: item["score"], reverse=True)
return ranked[:n]Why return recommendation reasons?
Each movie includes the three normalized signals behind its final score. This makes the ranking decision inspectable from the API response.
- Add _fallback() below _rank_personalized() by pasting this method:
def _fallback(self, user_id, n):
seen = set(self.seen_by_user.get(user_id, []))
ranked = []
for item in self.popularity:
movie_id = item["movie_id"]
if movie_id in seen:
continue
normalized_popularity = item["score"] / self.max_popularity_score
genre_affinity = self._genre_affinity(user_id, movie_id)
final_score = 0.85 * normalized_popularity + 0.15 * genre_affinity
ranked.append(
self._format_movie(
movie_id,
final_score,
{
"popularity": round(normalized_popularity, 4),
"genre_affinity": round(genre_affinity, 4),
},
)
)
ranked.sort(key=lambda item: item["score"], reverse=True)
return ranked[:n]What is the fallback for?
The fallback produces an unseen popularity row with genre support when collaborative candidates are unavailable. The next step exercises its failure paths directly.
- Add _record_metrics() below _fallback() by pasting this method:
def _record_metrics(
self,
strategy,
fallback_reason,
elapsed_ms,
candidate_count,
):
self.metrics["request_count"] += 1
self.metrics["strategy_counts"][strategy] += 1
self.metrics["total_latency_ms"] += elapsed_ms
self.metrics["last_candidate_count"] = candidate_count
self.metrics["last_fallback_reason"] = fallback_reason
if fallback_reason is not None:
self.metrics["fallback_count"] += 1What does metric recording capture?
The engine records each chosen strategy plus the latest candidate count. It also tracks latency plus fallback reasons for later inspection.
- Add the opening of recommend() directly below _record_metrics().
- Keep the method open for the remaining code blocks:
def recommend(
self,
user_id,
n=10,
collaborative_available=True,
record_metrics=True,
):
started = perf_counter()
user_id = str(user_id)
fallback_reason = None
candidate_scores = {}What starts a recommendation request?
The method starts a latency timer before normalizing the user ID. It also creates an empty candidate map plus a clear fallback state.
- Continue recommend() below the candidate_scores assignment with this decision logic:
if user_id not in self.seen_by_user:
fallback_reason = "unknown_user"
elif not self.liked_by_user.get(user_id):
fallback_reason = "no_liked_movies"
elif not collaborative_available:
fallback_reason = "collaborative_unavailable"
else:
candidate_scores = self._collaborative_candidates(user_id)
if not candidate_scores:
fallback_reason = "no_usable_candidates"
if fallback_reason is None:
strategy = "hybrid_collaborative"
recommendations = self._rank_personalized(
user_id,
candidate_scores,
n,
)
else:
strategy = "popularity_genre_fallback"
recommendations = self._fallback(user_id, n)How is the strategy selected?
Known users with usable candidates follow the collaborative ranker. Any named fallback condition selects the popularity plus genre path.
- Finish recommend() immediately below the strategy selection with this code:
elapsed_ms = (perf_counter() - started) * 1000
if record_metrics:
self._record_metrics(
strategy,
fallback_reason,
elapsed_ms,
len(candidate_scores),
)
return {
"user_id": user_id,
"strategy": strategy,
"fallback_reason": fallback_reason,
"candidate_count": len(candidate_scores),
"latency_ms": round(elapsed_ms, 3),
"recommendations": recommendations,
}What does the response reveal?
The response identifies the selected strategy plus any fallback reason. It also exposes candidate volume plus request latency beside the ranked movies.
- Save recommender.py.
- Select app.py in the VS Code Explorer sidebar.
- Find the existing /compare branch inside do_GET().
- Insert this recommendations branch directly above it:
if request.path == "/recommendations":
user_id = parameters.get("user_id", [""])[0]
if not user_id:
self.send_json({"error": "user_id is required"}, status=400)
return
simulate_failure = (
parameters.get("simulate_failure", ["0"])[0] == "1"
)
self.send_json(
ENGINE.recommend(
user_id,
n=n,
collaborative_available=not simulate_failure,
)
)
returnWhat does the route pass through?
The route requires a user ID before calling the recommendation engine. It passes the requested row size plus the collaborative availability switch into recommend().
- Add the recommendations URL to the route list near the bottom of do_GET() so that section contains this entry:
"/recommendations?user_id=1&n=10",
"/recommendations?user_id=1&n=10&simulate_failure=1",Why list both recommendation URLs?
The first URL exercises the personalized path. The second URL is ready for the deliberate dependency-failure check in the next step.
- Save app.py.
Before you start the server, do you expect user 1 to receive the old popularity strategy or the new hybrid strategy?
- Start the updated StreamFlix server from the VS Code terminal by running this command:
python3 app.pyWhat should the terminal show?
You should see StreamFlix listening at http://127.0.0.1:8000. The process now serves the personalized recommendations route.
- Open http://127.0.0.1:8000/recommendations?user_id=1&n=10 in your browser.
You should see hybrid_collaborative as the strategy. You should also see a non-zero candidate_count plus ten movie objects with three ranking signals inside each reason.
Personalized route not working?
Check that the terminal is running from the existing streamflix folder. Confirm that artifacts/recommendation_state.json still exists.
A startup syntax error usually points to a misplaced method indentation in recommender.py.
Help me debug the personalized StreamFlix route.
Measure the personalization lift
One personalized response proves the new path works. The comparison endpoint still needs to run that path for two active users before the overlap metric can show the improvement.
- Switch back to the terminal running StreamFlix.
- Stop the server by pressing Control+C.
- Return to recommender.py in the VS Code Explorer sidebar.
- Replace the existing compare_users() method with the next four code blocks in order.
- Start the replacement with this user-selection section:
def compare_users(self, n=10):
eligible_users = sorted(
[
user_id
for user_id, likes in self.liked_by_user.items()
if len(likes) >= 10
],
key=int,
)[:40]
personalized = {}
for user_id in eligible_users:
response = self.recommend(
user_id,
n=n,
collaborative_available=True,
record_metrics=False,
)
if response["strategy"] == "hybrid_collaborative":
personalized[user_id] = response["recommendations"]How are comparison users selected?
The method starts with active users who have at least 10 liked movies. It keeps only users whose recommendation path successfully produces hybrid results.
- Continue compare_users() immediately below the personalized-results loop with this pair search:
user_ids = list(personalized)
best_pair = None
best_overlap = 1.0
for first_index, first_user in enumerate(user_ids):
first_ids = [
item["movie_id"] for item in personalized[first_user]
]
for second_user in user_ids[first_index + 1 :]:
second_ids = [
item["movie_id"] for item in personalized[second_user]
]
overlap = self.top_n_overlap(first_ids, second_ids)
if best_pair is None or overlap < best_overlap:
best_pair = (first_user, second_user)
best_overlap = overlap
if overlap == 0.0:
break
if best_overlap == 0.0:
break
if best_pair is None:
raise RuntimeError("Could not find two users with personalized results")Why search for a contrasting pair?
The dataset can change over time. Searching active users avoids relying on hard-coded user IDs or fixed recommendation titles.
The loop keeps the pair with the lowest personalized overlap. This makes the difference from the universal popularity list easy to inspect.
- Continue compare_users() with the baseline plus personalized list preparation:
user_a, user_b = best_pair
baseline_a = self.baseline(user_a, n)["recommendations"]
baseline_b = self.baseline(user_b, n)["recommendations"]
personalized_a = personalized[user_a]
personalized_b = personalized[user_b]
baseline_a_ids = [item["movie_id"] for item in baseline_a]
baseline_b_ids = [item["movie_id"] for item in baseline_b]
personalized_a_ids = [item["movie_id"] for item in personalized_a]
personalized_b_ids = [item["movie_id"] for item in personalized_b]What is being compared?
The same two users receive one popularity row plus one personalized row each. Their movie IDs feed the same Top-N overlap calculation.
- Finish compare_users() with this response object:
return {
"user_a": user_a,
"user_b": user_b,
"n": n,
"baseline_overlap": round(
self.top_n_overlap(baseline_a_ids, baseline_b_ids),
3,
),
"personalized_overlap": round(
self.top_n_overlap(personalized_a_ids, personalized_b_ids),
3,
),
"baseline_user_a_titles": [
item["title"] for item in baseline_a
],
"baseline_user_b_titles": [
item["title"] for item in baseline_b
],
"user_a_titles": [item["title"] for item in personalized_a],
"user_b_titles": [item["title"] for item in personalized_b],
}What does the comparison return?
The response keeps both baseline title lists for the original control. It adds personalized titles plus a second overlap score for the same users.
- Add metrics_snapshot() directly below compare_users() by pasting this method:
def metrics_snapshot(self):
request_count = self.metrics["request_count"]
average_latency = (
self.metrics["total_latency_ms"] / request_count
if request_count
else 0.0
)
return {
"request_count": request_count,
"fallback_count": self.metrics["fallback_count"],
"strategy_counts": dict(self.metrics["strategy_counts"]),
"average_latency_ms": round(average_latency, 3),
"last_candidate_count": self.metrics["last_candidate_count"],
"last_fallback_reason": self.metrics["last_fallback_reason"],
}Why prepare a metrics snapshot?
The snapshot turns internal counters into a JSON-ready dictionary. The next step exposes these values through the API while testing degraded behavior.
- Save recommender.py.
Before you restart the service, do you expect the personalized overlap to remain at 1.0 or fall below the baseline overlap?
- Restart StreamFlix from the VS Code terminal by running this command:
python3 app.pyWhat should the restart confirm?
You should see the StreamFlix listening message again. This confirms that the completed recommendation engine imports successfully.
- Refresh http://127.0.0.1:8000/compare?n=10 in your browser.
You should see baseline_overlap remain at 1.0. You should see personalized_overlap at a lower value.
The user_a_titles list should differ from user_b_titles. That visible split proves the collaborative signal now expresses different tastes.
Overlap did not fall?
Confirm that compare_users() calls recommend() with metric recording disabled. Check that it stores responses only when the strategy equals hybrid_collaborative.
Restart the server after every edit because the local process does not reload the module automatically.
Help me debug unchanged StreamFlix overlap.
✔️ Awesome, I've got everything!
Great work. Your service now produces inspectable personalized rows with lower overlap for contrasting users.
- Keep the StreamFlix server running for the fallback checks in the next step.
ⓧ I'd like to double check the full code
Compare your complete recommender.py with this version:
import json
import math
from collections import defaultdict
from pathlib import Path
from time import perf_counter
class RecommendationEngine:
def __init__(self, artifact_path="artifacts/recommendation_state.json"):
with Path(artifact_path).open("r", encoding="utf-8") as state_file:
self.state = json.load(state_file)
self.movies = self.state["movies"]
self.popularity = self.state["popularity"]
self.seen_by_user = self.state["seen_by_user"]
self.liked_by_user = self.state["liked_by_user"]
self.likers_by_movie = self.state["likers_by_movie"]
self.genre_preferences = self.state["genre_preferences"]
self.popularity_score = {
item["movie_id"]: item["score"] for item in self.popularity
}
self.max_popularity_score = max(self.popularity_score.values())
self.metrics = {
"request_count": 0,
"fallback_count": 0,
"strategy_counts": defaultdict(int),
"total_latency_ms": 0.0,
"last_candidate_count": 0,
"last_fallback_reason": None,
}
def _format_movie(self, movie_id, score, reason):
movie = self.movies[movie_id]
return {
"movie_id": movie_id,
"title": movie["title"],
"genres": movie["genres"],
"score": round(score, 4),
"reason": reason,
}
def baseline(self, user_id, n=10):
recommendations = []
for item in self.popularity[:n]:
recommendations.append(
self._format_movie(
item["movie_id"],
item["score"],
"global_popularity",
)
)
return {
"user_id": str(user_id),
"strategy": "popularity_baseline",
"recommendations": recommendations,
}
def _genre_affinity(self, user_id, movie_id):
preferences = self.genre_preferences.get(user_id, {})
if not preferences:
return 0.0
maximum = max(preferences.values())
genres = self.movies[movie_id]["genres"]
if not genres or maximum == 0:
return 0.0
return max(preferences.get(genre, 0) for genre in genres) / maximum
def _collaborative_candidates(self, user_id):
target_likes = set(self.liked_by_user[user_id])
seen = set(self.seen_by_user[user_id])
neighbor_ids = set()
for movie_id in target_likes:
neighbor_ids.update(self.likers_by_movie.get(movie_id, []))
neighbor_ids.discard(user_id)
neighbor_scores = []
for neighbor_id in neighbor_ids:
neighbor_likes = set(self.liked_by_user[neighbor_id])
common = len(target_likes & neighbor_likes)
denominator = math.sqrt(len(target_likes) * len(neighbor_likes))
similarity = common / denominator if denominator else 0.0
if similarity > 0:
neighbor_scores.append((neighbor_id, similarity))
neighbor_scores.sort(key=lambda pair: pair[1], reverse=True)
candidate_scores = defaultdict(float)
for neighbor_id, similarity in neighbor_scores[:50]:
for movie_id in self.liked_by_user[neighbor_id]:
if movie_id not in seen:
candidate_scores[movie_id] += similarity
return dict(candidate_scores)
def _rank_personalized(self, user_id, candidate_scores, n):
max_collaborative = max(candidate_scores.values())
strongest_candidates = sorted(
candidate_scores.items(),
key=lambda pair: pair[1],
reverse=True,
)[:250]
ranked = []
for movie_id, collaborative_score in strongest_candidates:
normalized_collaborative = collaborative_score / max_collaborative
normalized_popularity = (
self.popularity_score.get(movie_id, 0.0)
/ self.max_popularity_score
)
genre_affinity = self._genre_affinity(user_id, movie_id)
final_score = (
0.70 * normalized_collaborative
+ 0.20 * normalized_popularity
+ 0.10 * genre_affinity
)
ranked.append(
self._format_movie(
movie_id,
final_score,
{
"collaborative": round(normalized_collaborative, 4),
"popularity": round(normalized_popularity, 4),
"genre_affinity": round(genre_affinity, 4),
},
)
)
ranked.sort(key=lambda item: item["score"], reverse=True)
return ranked[:n]
def _fallback(self, user_id, n):
seen = set(self.seen_by_user.get(user_id, []))
ranked = []
for item in self.popularity:
movie_id = item["movie_id"]
if movie_id in seen:
continue
normalized_popularity = item["score"] / self.max_popularity_score
genre_affinity = self._genre_affinity(user_id, movie_id)
final_score = 0.85 * normalized_popularity + 0.15 * genre_affinity
ranked.append(
self._format_movie(
movie_id,
final_score,
{
"popularity": round(normalized_popularity, 4),
"genre_affinity": round(genre_affinity, 4),
},
)
)
ranked.sort(key=lambda item: item["score"], reverse=True)
return ranked[:n]
def _record_metrics(
self,
strategy,
fallback_reason,
elapsed_ms,
candidate_count,
):
self.metrics["request_count"] += 1
self.metrics["strategy_counts"][strategy] += 1
self.metrics["total_latency_ms"] += elapsed_ms
self.metrics["last_candidate_count"] = candidate_count
self.metrics["last_fallback_reason"] = fallback_reason
if fallback_reason is not None:
self.metrics["fallback_count"] += 1
def recommend(
self,
user_id,
n=10,
collaborative_available=True,
record_metrics=True,
):
started = perf_counter()
user_id = str(user_id)
fallback_reason = None
candidate_scores = {}
if user_id not in self.seen_by_user:
fallback_reason = "unknown_user"
elif not self.liked_by_user.get(user_id):
fallback_reason = "no_liked_movies"
elif not collaborative_available:
fallback_reason = "collaborative_unavailable"
else:
candidate_scores = self._collaborative_candidates(user_id)
if not candidate_scores:
fallback_reason = "no_usable_candidates"
if fallback_reason is None:
strategy = "hybrid_collaborative"
recommendations = self._rank_personalized(
user_id,
candidate_scores,
n,
)
else:
strategy = "popularity_genre_fallback"
recommendations = self._fallback(user_id, n)
elapsed_ms = (perf_counter() - started) * 1000
if record_metrics:
self._record_metrics(
strategy,
fallback_reason,
elapsed_ms,
len(candidate_scores),
)
return {
"user_id": user_id,
"strategy": strategy,
"fallback_reason": fallback_reason,
"candidate_count": len(candidate_scores),
"latency_ms": round(elapsed_ms, 3),
"recommendations": recommendations,
}
@staticmethod
def top_n_overlap(first_ids, second_ids):
denominator = min(len(first_ids), len(second_ids))
if denominator == 0:
return 0.0
return len(set(first_ids) & set(second_ids)) / denominator
def compare_users(self, n=10):
eligible_users = sorted(
[
user_id
for user_id, likes in self.liked_by_user.items()
if len(likes) >= 10
],
key=int,
)[:40]
personalized = {}
for user_id in eligible_users:
response = self.recommend(
user_id,
n=n,
collaborative_available=True,
record_metrics=False,
)
if response["strategy"] == "hybrid_collaborative":
personalized[user_id] = response["recommendations"]
user_ids = list(personalized)
best_pair = None
best_overlap = 1.0
for first_index, first_user in enumerate(user_ids):
first_ids = [
item["movie_id"] for item in personalized[first_user]
]
for second_user in user_ids[first_index + 1 :]:
second_ids = [
item["movie_id"] for item in personalized[second_user]
]
overlap = self.top_n_overlap(first_ids, second_ids)
if best_pair is None or overlap < best_overlap:
best_pair = (first_user, second_user)
best_overlap = overlap
if overlap == 0.0:
break
if best_overlap == 0.0:
break
if best_pair is None:
raise RuntimeError("Could not find two users with personalized results")
user_a, user_b = best_pair
baseline_a = self.baseline(user_a, n)["recommendations"]
baseline_b = self.baseline(user_b, n)["recommendations"]
personalized_a = personalized[user_a]
personalized_b = personalized[user_b]
baseline_a_ids = [item["movie_id"] for item in baseline_a]
baseline_b_ids = [item["movie_id"] for item in baseline_b]
personalized_a_ids = [item["movie_id"] for item in personalized_a]
personalized_b_ids = [item["movie_id"] for item in personalized_b]
return {
"user_a": user_a,
"user_b": user_b,
"n": n,
"baseline_overlap": round(
self.top_n_overlap(baseline_a_ids, baseline_b_ids),
3,
),
"personalized_overlap": round(
self.top_n_overlap(personalized_a_ids, personalized_b_ids),
3,
),
"baseline_user_a_titles": [
item["title"] for item in baseline_a
],
"baseline_user_b_titles": [
item["title"] for item in baseline_b
],
"user_a_titles": [item["title"] for item in personalized_a],
"user_b_titles": [item["title"] for item in personalized_b],
}
def metrics_snapshot(self):
request_count = self.metrics["request_count"]
average_latency = (
self.metrics["total_latency_ms"] / request_count
if request_count
else 0.0
)
return {
"request_count": request_count,
"fallback_count": self.metrics["fallback_count"],
"strategy_counts": dict(self.metrics["strategy_counts"]),
"average_latency_ms": round(average_latency, 3),
"last_candidate_count": self.metrics["last_candidate_count"],
"last_fallback_reason": self.metrics["last_fallback_reason"],
}Compare your complete app.py with this version:
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
from recommender import RecommendationEngine
HOST = "127.0.0.1"
PORT = 8000
ENGINE = RecommendationEngine()
class RecommendationHandler(BaseHTTPRequestHandler):
def send_json(self, payload, status=200):
body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
@staticmethod
def parse_n(parameters):
try:
value = int(parameters.get("n", ["10"])[0])
except ValueError:
value = 10
return max(1, min(value, 20))
def do_GET(self):
request = urlsplit(self.path)
parameters = parse_qs(request.query)
n = self.parse_n(parameters)
if request.path == "/health":
self.send_json({"status": "ok"})
return
if request.path == "/baseline":
user_id = parameters.get("user_id", ["1"])[0]
self.send_json(ENGINE.baseline(user_id, n))
return
if request.path == "/recommendations":
user_id = parameters.get("user_id", [""])[0]
if not user_id:
self.send_json({"error": "user_id is required"}, status=400)
return
simulate_failure = (
parameters.get("simulate_failure", ["0"])[0] == "1"
)
self.send_json(
ENGINE.recommend(
user_id,
n=n,
collaborative_available=not simulate_failure,
)
)
return
if request.path == "/compare":
self.send_json(ENGINE.compare_users(n))
return
self.send_json(
{
"error": "not_found",
"routes": [
"/health",
"/baseline?user_id=1&n=10",
"/recommendations?user_id=1&n=10",
"/recommendations?user_id=1&n=10&simulate_failure=1",
"/compare?n=10",
],
},
status=404,
)
if __name__ == "__main__":
server = ThreadingHTTPServer((HOST, PORT), RecommendationHandler)
print(f"StreamFlix listening at http://{HOST}:{PORT}")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopping StreamFlix")
finally:
server.server_close()You have replaced one universal popularity row with user-specific candidates plus an inspectable hybrid score. Next, you will make that personalized path resilient when user history or candidate generation is unavailable.
Degrade Gracefully and Map the Scale-Up
Your hybrid ranker now gives contrasting users different movie rows. The final serving layer has to remain useful when personalization data is missing.
A collaborative dependency can become unavailable. A new user can arrive without any rating history.
A deliberate fallback keeps the recommendation row useful in both cases. You will make that degraded mode observable before mapping the local design to production-scale components.
In this step, get ready to:
- Exercise the two fallback branches.
- Inspect the serving metrics after a simulated dependency failure.
- Map each local component to its production-scale counterpart.
Trigger both fallback paths
A cold start occurs when StreamFlix receives a user ID with no prepared history. The engine responds with popular unseen titles because collaborative candidates are unavailable.
A simulated dependency failure tests graceful degradation for a known user. Both conditions keep the request successful while exposing a different reason for the fallback.
Before you send the cold-start request, which strategy do you expect the engine to choose for an unknown user?
- Change the recommendation URL in your browser so its query includes user_id=999999.
- Press Enter to send the request.
You will see strategy set to popularity_genre_fallback. The fallback_reason is unknown_user.
- In app.py, locate the /recommendations route inside do_GET().
- Replace that complete route block with the code below:
if request.path == "/recommendations":
user_id = parameters.get("user_id", [""])[0]
if not user_id:
self.send_json({"error": "user_id is required"}, status=400)
return
simulate_failure = (
parameters.get("simulate_failure", ["0"])[0] == "1"
)
self.send_json(
ENGINE.recommend(
user_id,
n=n,
collaborative_available=not simulate_failure,
)
)
returnWhat does this route change?
- The simulate_failure value checks whether the request contains simulate_failure=1.
- The collaborative_available argument tells recommend() whether it can use collaborative candidates.
- A disabled collaborative path produces the collaborative_unavailable fallback reason.
- Save app.py.
- Switch back to the integrated terminal running StreamFlix.
- Press Control-C to stop the current server.
- Restart StreamFlix with the updated route by running:
python3 app.pyWhat does this command do?
This runs app.py with your verified Python interpreter. Restarting loads the updated request handler into the local server.
You will see StreamFlix listening at http://127.0.0.1:8000 in the terminal. Keep this process running while you test the endpoints.
Server not starting?
Confirm that the previous server process stopped before restarting it. A process still using the local port prevents the new server from listening there.
Make sure the terminal is still inside the streamflix folder.
help me restart the StreamFlix server after editing app.py
Before you trigger the simulated failure, do you expect the request to fail completely or return a degraded recommendation row?
- Open http://127.0.0.1:8000/recommendations?user_id=1&n=10&simulate_failure=1 in your browser.
You will still receive recommendations. The response uses popularity_genre_fallback with collaborative_unavailable as its fallback reason.
The response also includes candidate_count plus latency_ms. Those fields make the degraded state visible to clients.
Still seeing the hybrid strategy?
Check that the browser URL ends with simulate_failure=1.
Confirm that you saved app.py before restarting the server.
help me debug why simulate_failure still returns hybrid_collaborative
Inspect serving metrics
Observability turns a successful fallback into an operational signal. StreamFlix records the chosen strategy whenever recommend() handles a serving request.
The counters live inside the running Python process. Restarting the server resets them to their initial values.
- In app.py, locate the /compare route inside do_GET().
- Add the metrics route directly below the comparison route:
if request.path == "/metrics":
self.send_json(ENGINE.metrics_snapshot())
returnWhat does this route expose?
The /metrics route serializes the current snapshot from metrics_snapshot(). It gives you one place to inspect serving behavior after each test request.
- Save app.py.
- Switch back to the integrated terminal running StreamFlix.
- Press Control-C to stop the server.
- Restart StreamFlix with the metrics route by running:
python3 app.pyWhy restart again?
The running process still holds the previous version of app.py. Restarting activates the new /metrics route with a fresh metrics object.
You will see the local StreamFlix address in the terminal again. The fresh process starts with no recorded recommendation requests.
Before you trigger another fallback, which metric do you expect to increase when the request finishes?
- Open http://127.0.0.1:8000/recommendations?user_id=1&n=10&simulate_failure=1 in your browser.
The response again uses popularity_genre_fallback. This request records the simulated failure in the fresh metrics object.
Before you inspect the snapshot, what should fallback_count reveal about the request you just sent?
- Open http://127.0.0.1:8000/metrics in your browser.
You will see a non-zero fallback_count. The last_fallback_reason value shows collaborative_unavailable.
What can these metrics answer?
- request_count shows how many recommendation requests reached the engine.
- fallback_count reveals how often personalization degraded.
- strategy_counts shows the mix of personalized responses versus fallback responses.
- average_latency_ms tracks the average time spent generating recommendation responses.
- last_candidate_count reveals whether collaborative retrieval produced usable candidates.
- last_fallback_reason identifies the most recent cause of degraded serving.
Fallback count still zero?
Send the simulated-failure request after the latest server restart. Requests recorded before a restart are no longer in memory.
Refresh /metrics after the fallback response loads.
help me debug why StreamFlix metrics do not record my fallback request
✔️ Awesome, I've got everything!
Your app.py now exposes failure simulation plus the metrics snapshot.
ⓧ I'd like to double check the full code
Compare your app.py with this complete version.
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
from recommender import RecommendationEngine
HOST = "127.0.0.1"
PORT = 8000
ENGINE = RecommendationEngine()
class RecommendationHandler(BaseHTTPRequestHandler):
def send_json(self, payload, status=200):
body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
@staticmethod
def parse_n(parameters):
try:
value = int(parameters.get("n", ["10"])[0])
except ValueError:
value = 10
return max(1, min(value, 20))
def do_GET(self):
request = urlsplit(self.path)
parameters = parse_qs(request.query)
n = self.parse_n(parameters)
if request.path == "/health":
self.send_json({"status": "ok"})
return
if request.path == "/baseline":
user_id = parameters.get("user_id", ["1"])[0]
self.send_json(ENGINE.baseline(user_id, n))
return
if request.path == "/recommendations":
user_id = parameters.get("user_id", [""])[0]
if not user_id:
self.send_json({"error": "user_id is required"}, status=400)
return
simulate_failure = (
parameters.get("simulate_failure", ["0"])[0] == "1"
)
self.send_json(
ENGINE.recommend(
user_id,
n=n,
collaborative_available=not simulate_failure,
)
)
return
if request.path == "/compare":
self.send_json(ENGINE.compare_users(n))
return
if request.path == "/metrics":
self.send_json(ENGINE.metrics_snapshot())
return
self.send_json(
{
"error": "not_found",
"routes": [
"/health",
"/baseline?user_id=1&n=10",
"/recommendations?user_id=1&n=10",
"/recommendations?user_id=1&n=10&simulate_failure=1",
"/compare?n=10",
"/metrics",
],
},
status=404,
)
if __name__ == "__main__":
server = ThreadingHTTPServer((HOST, PORT), RecommendationHandler)
print(f"StreamFlix listening at http://{HOST}:{PORT}")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopping StreamFlix")
finally:
server.server_close()Map the production evolution
The local design already separates offline preparation from online serving. The architecture document makes each boundary explicit.
Each local component has a larger-scale counterpart. The mapping shows where pipelines, partitions, services, stores, and operational controls fit.
- Select ARCHITECTURE.md from the project file list.
- Replace its scaffold sentence with this first section:
# StreamFlix Architecture
## Local request flow
MovieLens -> prepare.py -> recommendation_state.json -> collaborative candidates -> seen-item filter -> hybrid ranker -> fallback policy -> HTTP API -> metrics
## Offline and online ownership
Offline preparation owns CSV parsing, popularity aggregation, user-like sets, inverse liker indexes, seen-item indexes, and genre preferences.
Online serving owns request validation, collaborative candidate lookup, filtering, hybrid ranking, fallback selection, response metadata, and latency measurement.
The API never scans raw ratings during a request. It loads the prepared state once when the process starts.What does this boundary capture?
MovieLens Latest Small supplies raw catalog data plus rating events. The offline process converts those files into keyed state before any request arrives.
The online path performs lookups plus ranking against that prepared state. Raw CSV parsing stays outside the latency-sensitive request path.
- Save ARCHITECTURE.md.
- Confirm the document ends with the sentence about loading prepared state once.
First section missing?
Check that the scaffold sentence was fully replaced. The file should begin with # StreamFlix Architecture.
help me compare the first section of my StreamFlix architecture document
- Find the final sentence in the offline ownership section.
- Add the scale-up mapping directly below it:
## Evolution toward large scale
| Local component | Scaled counterpart | Design reason |
| --- | --- | --- |
| MovieLens CSV files | Catalog, viewing, rating, and impression event pipelines | Separate durable facts from serving computation |
| `prepare.py` | Scheduled batch jobs plus nearline incremental updates | Precompute expensive state while controlling freshness |
| `recommendation_state.json` | Versioned feature store, candidate store, and cache | Support low-latency keyed reads and safe artifact rollout |
| User-keyed dictionaries | User-ID partitions with replication | Spread load and isolate failures while preserving locality |
| One collaborative function | Independent candidate-generation service | Scale retrieval separately and combine multiple candidate sources |
| One hybrid function | Independent ranking service | Iterate on scoring models without changing retrieval |
| One Top-N list | Multiple rows with row-specific candidate and ranking objectives | Model different user intents such as discovery and continuation |
| In-process counters | Central metrics, logs, traces, dashboards, and alerts | Detect latency, empty candidates, and fallback spikes |
| One local server | Stateless API replicas behind load balancing | Add horizontal capacity and remove a single serving process |How does the design scale?
The JSON artifact evolves into a versioned feature store plus a candidate store. A cache keeps frequently requested state close to serving.
User-keyed dictionaries become partitions. Candidate generation plus ranking become independently scalable services.
- Save ARCHITECTURE.md.
- Confirm the scale-up table contains nine local component rows.
Table not lining up?
Check that every table row begins with a vertical bar. Confirm that every row ends with one too.
help me fix the scale-up table in ARCHITECTURE.md
- Find the final row of the scale-up table.
- Add the hypothetical latency budget directly below it:
## Hypothetical latency budget
This is a StreamFlix design target, not a published Netflix value.
- Candidate-store read: 20 ms
- Candidate generation: 25 ms
- Filtering and ranking: 25 ms
- Fallback and response serialization: 10 ms
- Network and safety margin: 20 ms
- End-to-end target: 100 ms
When candidate generation exceeds its budget or returns no usable items, the serving layer should stop waiting, use a cached popularity or genre fallback, and increment fallback telemetry.Why define a latency budget?
A latency budget assigns a time allowance to each part of the serving path. It gives the fallback policy a concrete point for stopping slow personalization work.
The 100 ms total is a StreamFlix design target. No published Netflix latency value is claimed.
- Save ARCHITECTURE.md.
- Confirm the five budget entries add up to the 100 ms end-to-end target.
Budget total unclear?
Check the five component values against the complete block above. The end-to-end line should remain 100 ms.
help me understand the hypothetical StreamFlix latency budget
- Find the paragraph that ends with fallback telemetry.
- Add the production decision list directly below it:
## Next production decisions
- Version artifacts so an API instance reads one coherent snapshot.
- Keep a last-known-good index for rollback.
- Cache hot users and popular rows, but bound staleness explicitly.
- Partition user state by user ID and replicate catalog state broadly.
- Give each recommendation row an owner, objective, candidate sources, ranker, and fallback.
- Test offline metrics first, then validate product impact with controlled online experiments.What do these decisions protect?
Artifact versions prevent one API process from mixing incompatible snapshots. A last-known-good index gives the serving layer a rollback path.
Explicit row ownership keeps objectives plus fallback policies separate. Controlled experiments test whether offline improvements help real users.
- Save ARCHITECTURE.md.
- Confirm the final line refers to controlled online experiments.
Document ending early?
Confirm that the ## Next production decisions heading appears after the latency budget. Check that all six decision items are present.
help me complete the final production decisions in ARCHITECTURE.md
✔️ Awesome, I've got everything!
Your architecture now connects the local request path to a resilient scale-up design.
ⓧ I'd like to double check the full code
Compare your ARCHITECTURE.md with this complete version.
# StreamFlix Architecture
## Local request flow
MovieLens -> prepare.py -> recommendation_state.json -> collaborative candidates -> seen-item filter -> hybrid ranker -> fallback policy -> HTTP API -> metrics
## Offline and online ownership
Offline preparation owns CSV parsing, popularity aggregation, user-like sets, inverse liker indexes, seen-item indexes, and genre preferences.
Online serving owns request validation, collaborative candidate lookup, filtering, hybrid ranking, fallback selection, response metadata, and latency measurement.
The API never scans raw ratings during a request. It loads the prepared state once when the process starts.
## Evolution toward large scale
| Local component | Scaled counterpart | Design reason |
| --- | --- | --- |
| MovieLens CSV files | Catalog, viewing, rating, and impression event pipelines | Separate durable facts from serving computation |
| `prepare.py` | Scheduled batch jobs plus nearline incremental updates | Precompute expensive state while controlling freshness |
| `recommendation_state.json` | Versioned feature store, candidate store, and cache | Support low-latency keyed reads and safe artifact rollout |
| User-keyed dictionaries | User-ID partitions with replication | Spread load and isolate failures while preserving locality |
| One collaborative function | Independent candidate-generation service | Scale retrieval separately and combine multiple candidate sources |
| One hybrid function | Independent ranking service | Iterate on scoring models without changing retrieval |
| One Top-N list | Multiple rows with row-specific candidate and ranking objectives | Model different user intents such as discovery and continuation |
| In-process counters | Central metrics, logs, traces, dashboards, and alerts | Detect latency, empty candidates, and fallback spikes |
| One local server | Stateless API replicas behind load balancing | Add horizontal capacity and remove a single serving process |
## Hypothetical latency budget
This is a StreamFlix design target, not a published Netflix value.
- Candidate-store read: 20 ms
- Candidate generation: 25 ms
- Filtering and ranking: 25 ms
- Fallback and response serialization: 10 ms
- Network and safety margin: 20 ms
- End-to-end target: 100 ms
When candidate generation exceeds its budget or returns no usable items, the serving layer should stop waiting, use a cached popularity or genre fallback, and increment fallback telemetry.
## Next production decisions
- Version artifacts so an API instance reads one coherent snapshot.
- Keep a last-known-good index for rollback.
- Cache hot users and popular rows, but bound staleness explicitly.
- Partition user state by user ID and replicate catalog state broadly.
- Give each recommendation row an owner, objective, candidate sources, ranker, and fallback.
- Test offline metrics first, then validate product impact with controlled online experiments.Before the final check, do you expect another simulated failure to increase the fallback telemetry?
- Open http://127.0.0.1:8000/recommendations?user_id=1&n=10&simulate_failure=1 in your browser.
You will see popularity_genre_fallback with a populated recommendation list. The response includes fallback_reason, candidate_count, and latency_ms.
- Open http://127.0.0.1:8000/metrics in your browser.
You will see a non-zero fallback_count plus popularity_genre_fallback inside strategy_counts.
Strong finish: StreamFlix now exposes fallback telemetry. Your architecture maps every local component to a scaled counterpart.
Secret mission
Add Diversity-Aware Reranking
Cap each primary genre at two titles in the Top 10. Print both rows. Compare their dominant-genre shares.
Clean Up Your Resources
Clean Up Your Resources
Everything in StreamFlix runs on your Mac. The local HTTP service and project files create no ongoing costs.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- The streamflix folder containing your local project files.
- The MovieLens Latest Small archive at data/ml-latest-small.zip.
- The extracted dataset inside data/ml-latest-small/.
- The serving-ready index at artifacts/recommendation_state.json.
- The local API process listening at http://127.0.0.1:8000.
Keep everything running
No action is needed while you are actively experimenting with StreamFlix. Your API remains available from the local address.
- Keep using the API at http://127.0.0.1:8000 while you continue experimenting.
- Switch back to the integrated terminal when you finish working.
- Press Control-C to stop the local API.
- Retain the streamflix folder so every source file and local data file remains available.
Pause - I'll come back to this later
Shut down the local process to free port 8000. Delete only the generated artifact so you can rebuild the offline contract when you return.
- Switch back to the integrated terminal where the API is running.
- Press Control-C to stop the local API.
- Remove only the generated serving artifact by running this command:
rm artifacts/recommendation_state.jsonWhat does this command do?
This removes only the prepared recommendation state. Your source files and downloaded dataset remain inside streamflix.
- Confirm the artifact is gone by running this command:
ls artifactsWhat should I see?
You should no longer see recommendation_state.json in the output. This confirms that the generated serving state has been removed.
- Rebuild the offline recommendation index when you return by running this command:
python3 prepare.pyWhat should I see?
The terminal reports non-zero movie and user counts. It finishes with Wrote artifacts/recommendation_state.json.
You are ready to continue. The rebuilt artifact restores the contract used by the online service.
Delete - I don't want to use this again
Deleting the project folder is permanent. The command below targets only the streamflix folder from its parent folder.
- Switch back to the integrated terminal where the API is running.
- Press Control-C to stop the local API.
- Move to the folder above streamflix and remove the entire project by running these commands:
cd ..
rm -rf streamflixWhat do these commands do?
The first line moves the terminal to the folder containing streamflix. The second line removes the project folder with its source files.
The downloaded archive and extracted dataset are removed with the folder. The generated recommendation artifact is removed at the same time.
- Confirm the project folder is gone by running this command:
lsWhat should I see?
You should no longer see streamflix in the output. This confirms that every local project resource has been deleted.
Nice Work!
Nice Work!
You brought StreamFlix to life as a local hybrid recommendation service! It turns prepared MovieLens Latest Small data into personalized rows with visible fallback behavior.
You've learned how to:
- Build an offline preparation pipeline that turns MovieLens ratings into a serving-ready JSON index. Keep raw CSV scans out of each request by loading the artifact once at startup.
- Run a personalization experiment that proves the popularity baseline produces 100 percent Top-N overlap. Add collaborative candidate generation to find user-specific options. Apply hybrid ranking to reduce overlap while retaining stable popularity and genre signals.
- Add graceful degradation for unknown users. Reuse the fallback path when collaborative candidates are unavailable. Add serving observability for request health. Map the local design to a scalable architecture in ARCHITECTURE.md.
- Secret Mission: Added a diversity-aware post-ranker that caps each primary genre at two titles before backfilling. Measured the original and reranked dominant-genre shares.
Ready to quiz yourself?