diff --git a/README.md b/README.md
index 615c55c..a41074c 100644
--- a/README.md
+++ b/README.md
@@ -1,270 +1,403 @@
-# DeepSequence Recommender
-
-A versioned, evaluated sequential recommender with a reproducible training-to-serving contract.
-
-
- 
- 
- 
- 
- 
-
-
-
- 
- 
- 
- 
- 
- 
-
-DeepSequence Recommender learns next-item behavior from timestamped interactions. One connected
-path now covers schema validation, chronological splits, training, baseline comparison, model
-packaging, checksum verification, serving, feedback capture, and monitoring.
-
-The repository does **not** ship a trained production model or claim production availability.
-Production fails closed unless a verified bundle is mounted. Development uses a deterministic
-untrained model so API integration can be tested without presenting random ranking as AI quality.
-
-## Engineering features
-
-- Padding-aware attention, aligned item/output IDs, excluded padding class, and bounded top-k.
-- Versioned events, temporal splits, causal targets, deterministic seeds, and popularity baseline.
-- Recall@K, NDCG@K, MRR@K, catalogue coverage, and machine-readable evaluation reports.
-- Immutable bundles containing weights, vocabulary, architecture, lineage, metrics, and checksums.
-- Fail-closed production startup for absent, corrupted, or incompatible artifacts.
-- API-key option, rate limiting, admission control, latency fallback, and version-aware caching.
-- Privacy-minimized feedback events for impressions, clicks, skips, carts, purchases, and dislikes.
-- Prometheus inference, request, cache, fallback, saturation, and feedback signals.
-- Non-root/read-only container and hardened Kubernetes security/resource configuration.
-- CI that cannot hide test, schema, audit, or benchmark failures with shell fallbacks.
-
-## Architecture
-
-```text
-Events -> contract -> chronological split -> causal examples
- | |
- | popularity baseline
- v v
-PyTorch training --------------------> ranking evaluation
- |
- v
-manifest + vocabulary + weights + metrics + checksums
- |
- v
-startup verification -> API key -> rate limit -> cache -> admission -> model
- | |
- +-> fallback
- |
- Prometheus + feedback events
-```
-
-## Quick start
-
-```bash
-git clone https://github.com/CoreyLeath-code/DeepSequence-Recommender.git
-cd DeepSequence-Recommender
-python -m venv .venv
-source .venv/bin/activate # Windows: .venv\Scripts\activate
-pip install -r requirements.txt
-pip install pytest pytest-cov ruff
-pytest -q
-ENVIRONMENT=development uvicorn app.main:app --reload
-```
-
-Development responses identify the model as `development-untrained`.
-
-## Train, evaluate, and package
-
-```bash
-python -m scripts.generate_demo_data --output data/demo-events.jsonl
-python -m src.training.train \
- --dataset data/demo-events.jsonl \
- --output models/candidate \
- --epochs 3 \
- --top-k 10 \
- --seed 7
-```
-
-The candidate contains:
-
-```text
-manifest.json version, architecture, dataset lineage, metrics, checksums
-model.pt PyTorch state dictionary
-vocabulary.json exact training/serving item mapping
-evaluation.json neural and popularity-baseline results
-```
-
-Demo data validates the pipeline only. Use a leakage-reviewed production snapshot before making
-ranking claims. Promote the complete directory through the model registry; never mix files.
-
-Export the exact verified model—not a mock architecture—to ONNX:
-
-```bash
-python -m src.serving.onnx_exporter --bundle models/current --output models/model.onnx
-```
-
-## Production serving
-
-```bash
-ENVIRONMENT=production \
-MODEL_BUNDLE_PATH=models/current \
-API_KEY='replace-me' \
-uvicorn app.main:app --host 0.0.0.0 --port 8000
-```
-
-```bash
-curl -X POST http://localhost:8000/recommendations/ \
- -H 'Content-Type: application/json' \
- -H 'X-API-Key: replace-me' \
- -d '{"user_id":"user-42","item_sequence":["item-2","item-7"],"top_k":5}'
-```
-
-Feedback is accepted at `POST /recommendations/feedback`. Raw user IDs are hashed before events
-are logged. Production should route structured logs to a governed event bus with retention,
-consent, deletion, and delivery guarantees.
-
-
-
-| Concern | Decision | Tradeoff / boundary |
-|---|---|---|
-| Training-serving parity | One vocabulary and architecture manifest drives serving and ONNX | Bundle migrations need explicit compatibility handling |
-| Leakage | Global chronological split and causal prefix targets | Production also needs cold-start and cohort evaluation |
-| Promotion | Neural NDCG is compared with popularity | Registry automation remains deployment-specific |
-| Integrity | SHA-256 plus strict state-dict loading | Add KMS/Sigstore for signed provenance |
-| Overload | Non-blocking admission and deterministic fallback | Hard cancellation needs a worker/model-server boundary |
-| Cache | Key includes model version, history, and top-k | In-memory state is replica-local |
-| Rate limit | Per-process fixed window | Authoritative distributed limits belong at the gateway |
-| Authentication | Constant-time optional API key | Multi-tenant systems need workload identity/OAuth and policy |
-| Feedback | Pseudonymized structured events | Event bus, retention, and deletion are platform integrations |
-| Kubernetes | Immutable tag, read-only FS, non-root, seccomp, resource bounds | Registry, secrets, PVC, and rollout controller are external |
-
-## Research metrics and benchmarks
-
-Every real candidate reports Recall@K, NDCG@K, MRR@K, and coverage beside popularity. No quality
-number is published here because a representative production dataset is not included.
-
-Reference systems microbenchmark:
-
-```bash
-python -m benchmarks.inference --iterations 200 --warmup 20
-```
-
-| Metric | Observation |
-|---|---:|
-| Mean | 0.573 ms |
-| p50 | 0.556 ms |
-| p95 | 0.755 ms |
-| p99 | 1.118 ms |
-| Sequential rate | 1,743.0 inferences/s |
-
-Environment: Python 3.12.13, PyTorch 2.13.0 CPU, Windows 11 build 26200, AMD64 Family 25 Model
-97; batch 1, 500 items, padded length 50, top-k 10, 20 warmups, 200 measured samples, collected
-July 22, 2026.
-
-This excludes HTTP, serialization, queueing, network, concurrency, replicas, cold starts, and
-real catalogue scale. It is a regression baseline—not a production SLO or availability claim.
-See [benchmark methodology](docs/BENCHMARKS.md).
-
-## Production-readiness status
-
-- [x] Correct ranking IDs and padding behavior
-- [x] Strict request and event contracts
-- [x] Temporal evaluation and baseline comparison
-- [x] Versioned checksummed model bundle
-- [x] Fail-closed production loading and model readiness
-- [x] Admission, fallback, cache, rate limit, and API-key option
-- [x] Feedback contract and user-ID minimization
-- [x] Enforced CI, schema, security, and benchmark evidence
-- [x] Hardened container and Kubernetes manifests
-- [ ] Governed representative production dataset
-- [ ] External registry, signed provenance, and automated promotion
-- [ ] Distributed gateway limits and durable event bus
-- [ ] Target-environment multi-replica load test
-- [ ] Shadow/canary rollout with automated rollback
-- [ ] Online experiment proving user or business impact
-
-Unchecked items require real data or deployment infrastructure and cannot be honestly completed
-inside a standalone public repository.
-
-## Extended recruiter Q&A
-
-### Is this genuinely AI-powered?
-
-Yes when a trained bundle is supplied. It trains a bidirectional LSTM with padding-aware attention
-on causal next-item examples. Production cannot silently substitute random weights.
-
-### Why was the old demo not production AI?
-
-A neural class with random weights is not learned recommendation. Production requires data,
-evaluation, lineage, versioned artifacts, readiness, rollback, and operational evidence.
-
-### Why use a popularity baseline?
-
-A complex model should earn its cost. Popularity is cheap and often strong; a neural candidate
-that cannot beat it on leakage-safe ranking metrics should not be promoted.
-
-### Why split chronologically?
-
-Random splits leak future behavior. Chronological evaluation asks the real question: can the model
-rank behavior that occurs later than its training evidence?
-
-### Why package vocabulary with weights?
-
-Embedding rows and output classes only mean something under the exact training mapping. A different
-mapping can return plausible but incorrect item IDs without producing a tensor shape error.
-
-### Why verify checksums?
-
-Models are usually deployed outside Git. Checksums detect partial uploads, accidental bundle
-mixing, and corruption. Signing/provenance is the next layer for regulated environments.
-
-### Why retain a development model?
-
-It keeps smoke tests self-contained. It is seeded, visibly labeled untrained, and forbidden in
-production, so convenience cannot become an invisible ranking failure.
-
-### Is the latency fallback a hard timeout?
-
-No. It replaces a result after an in-process kernel finishes. Hard cancellation requires a process
-or model-server boundary with load shedding and cancelable requests.
-
-### Why not use an LLM as the ranker?
-
-LLMs help with intent, catalogue enrichment, and grounded explanations. Core ranking still needs
-measurable retrieval, temporal evaluation, deterministic policy, and controlled cost.
-
-### How would this scale to millions of items?
-
-Use a two-stage system: two-tower/ANN candidate retrieval followed by sequence re-ranking. A dense
-projection over the entire catalogue is not economical at very large scale.
-
-### How would you handle cold start?
-
-Initialize new items from text/image/category embeddings and use contextual popularity until
-behavior accumulates. Report new-user and new-item cohorts separately.
-
-### What should be monitored?
-
-Latency, errors, saturation, cache/fallback rates, model version, unknown-item rate, feature and
-prediction drift, delayed-label ranking quality, coverage, diversity, bias, and experiment goals.
-
-### What is the rollout strategy?
-
-Shadow first, compare outputs/resources, canary a small percentage, enforce guardrails, then ramp
-with automatic rollback to the previous immutable image-and-bundle pair.
-
-### What demonstrates senior engineering here?
-
-The model is treated as one component. Data contracts, evaluation, lineage, serving safety,
-privacy, observability, delivery, and honest claims form the production system around it.
-
-## Documentation
-
-- [Benchmark methodology](docs/BENCHMARKS.md)
-- [Model card](docs/MODEL_CARD.md)
-- [Operations runbook](docs/OPERATIONS.md)
-- [Privacy policy](docs/PRIVACY.md)
-- [Issue #8](https://github.com/CoreyLeath-code/DeepSequence-Recommender/issues/8)
-
-MIT licensed. See [LICENSE](LICENSE).
+# DeepSequence Recommender
+
+
A versioned, evaluated sequential recommender with a reproducible training-to-serving contract.
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+DeepSequence Recommender learns next-item behavior from timestamped interactions. One connected
+path now covers schema validation, chronological splits, training, baseline comparison, model
+packaging, checksum verification, serving, feedback capture, and monitoring.
+
+The repository does **not** ship a trained production model or claim production availability.
+Production fails closed unless a verified bundle is mounted. Development uses a deterministic
+untrained model so API integration can be tested without presenting random ranking as AI quality.
+
+
+## Production Readiness Guide
+
+> This section is the portfolio audit entry point for **DeepSequence-Recommender**. It describes an engineering promotion path; it is not a claim that the repository is already production-authorized.
+
+[](https://github.com/CoreyLeath-code/DeepSequence-Recommender/actions) [](https://github.com/CoreyLeath-code/DeepSequence-Recommender/blob/main/LICENSE)
+
+### Architecture flowchart
+
+```mermaid
+flowchart LR
+ Client --> Gateway --> Services[API + workers] --> Events[(Event bus)] --> Store[(State)]
+```
+
+### Quickstart and local validation
+
+The supported local path should be reproducible from a clean checkout. The inferred stack for this repository is **Python/platform services**.
+
+```bash
+python -m venv .venv && source .venv/bin/activate && pip install -r requirements.txt
+pytest -q
+```
+
+If the project uses external services, model artifacts, cloud credentials, or private data, start them through documented local fixtures or mocks. Never place secrets or identifiable records in the repository.
+
+### Research-style metrics and benchmarks
+
+| Evidence | Required record |
+|---|---|
+| Correctness | Test command, commit SHA, runtime, and pass/fail result |
+| Performance | Warm-up, sample count, concurrency, median, p95, p99, throughput, and memory |
+| Data/model quality | Dataset version, split strategy, leakage controls, calibration, subgroup results, and uncertainty |
+| Runtime | Image digest, health-check latency, resource limits, and rollback target |
+| Security | Dependency, secret, SAST, container, and SBOM results |
+
+A benchmark number belongs in a versioned artifact tied to a commit and hardware/runtime description. Engineering benchmarks must not be presented as clinical, financial, safety, or model-quality validation without the appropriate domain evidence.
+
+### Extended Q&A
+
+**What is production-ready for this repository?**
+A reproducible build, tested public contract, controlled configuration, observable runtime, documented security boundary, versioned artifacts, and a tested rollback path.
+
+**What must remain explicit?**
+The intended use, excluded use, data/credential handling, model or algorithm limitations, and which metrics are measured versus aspirational.
+
+**What should be completed next?**
+Use the linked production-readiness issue for this repository as the checklist. Resolve missing tests, deployment instructions, observability, supply-chain controls, and release evidence before attaching a production claim.
+
+
+## Engineering features
+
+- Padding-aware attention, aligned item/output IDs, excluded padding class, and bounded top-k.
+- Versioned events, temporal splits, causal targets, deterministic seeds, and popularity baseline.
+- Recall@K, NDCG@K, MRR@K, catalogue coverage, and machine-readable evaluation reports.
+- Immutable bundles containing weights, vocabulary, architecture, lineage, metrics, and checksums.
+- Fail-closed production startup for absent, corrupted, or incompatible artifacts.
+- API-key option, rate limiting, admission control, latency fallback, and version-aware caching.
+- Privacy-minimized feedback events for impressions, clicks, skips, carts, purchases, and dislikes.
+- Prometheus inference, request, cache, fallback, saturation, and feedback signals.
+- Non-root/read-only container and hardened Kubernetes security/resource configuration.
+- CI that cannot hide test, schema, audit, or benchmark failures with shell fallbacks.
+
+## Architecture
+
+```text
+Events -> contract -> chronological split -> causal examples
+ | |
+ | popularity baseline
+ v v
+PyTorch training --------------------> ranking evaluation
+ |
+ v
+manifest + vocabulary + weights + metrics + checksums
+ |
+ v
+startup verification -> API key -> rate limit -> cache -> admission -> model
+ | |
+ +-> fallback
+ |
+ Prometheus + feedback events
+```
+
+## Quick start
+
+```bash
+git clone https://github.com/CoreyLeath-code/DeepSequence-Recommender.git
+cd DeepSequence-Recommender
+python -m venv .venv
+source .venv/bin/activate # Windows: .venv\Scripts\activate
+pip install -r requirements.txt
+pip install pytest pytest-cov ruff
+pytest -q
+ENVIRONMENT=development uvicorn app.main:app --reload
+```
+
+Development responses identify the model as `development-untrained`.
+
+## Train, evaluate, and package
+
+```bash
+python -m scripts.generate_demo_data --output data/demo-events.jsonl
+python -m src.training.train \
+ --dataset data/demo-events.jsonl \
+ --output models/candidate \
+ --epochs 3 \
+ --top-k 10 \
+ --seed 7
+```
+
+The candidate contains:
+
+```text
+manifest.json version, architecture, dataset lineage, metrics, checksums
+model.pt PyTorch state dictionary
+vocabulary.json exact training/serving item mapping
+evaluation.json neural and popularity-baseline results
+```
+
+Demo data validates the pipeline only. Use a leakage-reviewed production snapshot before making
+ranking claims. Promote the complete directory through the model registry; never mix files.
+
+Export the exact verified model—not a mock architecture—to ONNX:
+
+```bash
+python -m src.serving.onnx_exporter --bundle models/current --output models/model.onnx
+```
+
+## Production serving
+
+```bash
+ENVIRONMENT=production \
+MODEL_BUNDLE_PATH=models/current \
+API_KEY='replace-me' \
+uvicorn app.main:app --host 0.0.0.0 --port 8000
+```
+
+```bash
+curl -X POST http://localhost:8000/recommendations/ \
+ -H 'Content-Type: application/json' \
+ -H 'X-API-Key: replace-me' \
+ -d '{"user_id":"user-42","item_sequence":["item-2","item-7"],"top_k":5}'
+```
+
+Feedback is accepted at `POST /recommendations/feedback`. Raw user IDs are hashed before events
+are logged. Production should route structured logs to a governed event bus with retention,
+consent, deletion, and delivery guarantees.
+
+
+
+| Concern | Decision | Tradeoff / boundary |
+|---|---|---|
+| Training-serving parity | One vocabulary and architecture manifest drives serving and ONNX | Bundle migrations need explicit compatibility handling |
+| Leakage | Global chronological split and causal prefix targets | Production also needs cold-start and cohort evaluation |
+| Promotion | Neural NDCG is compared with popularity | Registry automation remains deployment-specific |
+| Integrity | SHA-256 plus strict state-dict loading | Add KMS/Sigstore for signed provenance |
+| Overload | Non-blocking admission and deterministic fallback | Hard cancellation needs a worker/model-server boundary |
+| Cache | Key includes model version, history, and top-k | In-memory state is replica-local |
+| Rate limit | Per-process fixed window | Authoritative distributed limits belong at the gateway |
+| Authentication | Constant-time optional API key | Multi-tenant systems need workload identity/OAuth and policy |
+| Feedback | Pseudonymized structured events | Event bus, retention, and deletion are platform integrations |
+| Kubernetes | Immutable tag, read-only FS, non-root, seccomp, resource bounds | Registry, secrets, PVC, and rollout controller are external |
+
+## Research metrics and benchmarks
+
+Every real candidate reports Recall@K, NDCG@K, MRR@K, and coverage beside popularity. No quality
+number is published here because a representative production dataset is not included.
+
+Reference systems microbenchmark:
+
+```bash
+python -m benchmarks.inference --iterations 200 --warmup 20
+```
+
+| Metric | Observation |
+|---|---:|
+| Mean | 0.573 ms |
+| p50 | 0.556 ms |
+| p95 | 0.755 ms |
+| p99 | 1.118 ms |
+| Sequential rate | 1,743.0 inferences/s |
+
+Environment: Python 3.12.13, PyTorch 2.13.0 CPU, Windows 11 build 26200, AMD64 Family 25 Model
+97; batch 1, 500 items, padded length 50, top-k 10, 20 warmups, 200 measured samples, collected
+July 22, 2026.
+
+This excludes HTTP, serialization, queueing, network, concurrency, replicas, cold starts, and
+real catalogue scale. It is a regression baseline—not a production SLO or availability claim.
+See [benchmark methodology](docs/BENCHMARKS.md).
+
+## Production-readiness status
+
+- [x] Correct ranking IDs and padding behavior
+- [x] Strict request and event contracts
+- [x] Temporal evaluation and baseline comparison
+- [x] Versioned checksummed model bundle
+- [x] Fail-closed production loading and model readiness
+- [x] Admission, fallback, cache, rate limit, and API-key option
+- [x] Feedback contract and user-ID minimization
+- [x] Enforced CI, schema, security, and benchmark evidence
+- [x] Hardened container and Kubernetes manifests
+- [ ] Governed representative production dataset
+- [ ] External registry, signed provenance, and automated promotion
+- [ ] Distributed gateway limits and durable event bus
+- [ ] Target-environment multi-replica load test
+- [ ] Shadow/canary rollout with automated rollback
+- [ ] Online experiment proving user or business impact
+
+Unchecked items require real data or deployment infrastructure and cannot be honestly completed
+inside a standalone public repository.
+
+## Extended recruiter Q&A
+
+### Is this genuinely AI-powered?
+
+Yes when a trained bundle is supplied. It trains a bidirectional LSTM with padding-aware attention
+on causal next-item examples. Production cannot silently substitute random weights.
+
+### Why was the old demo not production AI?
+
+A neural class with random weights is not learned recommendation. Production requires data,
+evaluation, lineage, versioned artifacts, readiness, rollback, and operational evidence.
+
+### Why use a popularity baseline?
+
+A complex model should earn its cost. Popularity is cheap and often strong; a neural candidate
+that cannot beat it on leakage-safe ranking metrics should not be promoted.
+
+### Why split chronologically?
+
+Random splits leak future behavior. Chronological evaluation asks the real question: can the model
+rank behavior that occurs later than its training evidence?
+
+### Why package vocabulary with weights?
+
+Embedding rows and output classes only mean something under the exact training mapping. A different
+mapping can return plausible but incorrect item IDs without producing a tensor shape error.
+
+### Why verify checksums?
+
+Models are usually deployed outside Git. Checksums detect partial uploads, accidental bundle
+mixing, and corruption. Signing/provenance is the next layer for regulated environments.
+
+### Why retain a development model?
+
+It keeps smoke tests self-contained. It is seeded, visibly labeled untrained, and forbidden in
+production, so convenience cannot become an invisible ranking failure.
+
+### Is the latency fallback a hard timeout?
+
+No. It replaces a result after an in-process kernel finishes. Hard cancellation requires a process
+or model-server boundary with load shedding and cancelable requests.
+
+### Why not use an LLM as the ranker?
+
+LLMs help with intent, catalogue enrichment, and grounded explanations. Core ranking still needs
+measurable retrieval, temporal evaluation, deterministic policy, and controlled cost.
+
+### How would this scale to millions of items?
+
+Use a two-stage system: two-tower/ANN candidate retrieval followed by sequence re-ranking. A dense
+projection over the entire catalogue is not economical at very large scale.
+
+### How would you handle cold start?
+
+Initialize new items from text/image/category embeddings and use contextual popularity until
+behavior accumulates. Report new-user and new-item cohorts separately.
+
+### What should be monitored?
+
+Latency, errors, saturation, cache/fallback rates, model version, unknown-item rate, feature and
+prediction drift, delayed-label ranking quality, coverage, diversity, bias, and experiment goals.
+
+### What is the rollout strategy?
+
+Shadow first, compare outputs/resources, canary a small percentage, enforce guardrails, then ramp
+with automatic rollback to the previous immutable image-and-bundle pair.
+
+### What demonstrates senior engineering here?
+
+The model is treated as one component. Data contracts, evaluation, lineage, serving safety,
+privacy, observability, delivery, and honest claims form the production system around it.
+
+## Documentation
+
+- [Benchmark methodology](docs/BENCHMARKS.md)
+- [Model card](docs/MODEL_CARD.md)
+- [Operations runbook](docs/OPERATIONS.md)
+- [Privacy policy](docs/PRIVACY…4021 tokens truncated…Scaled dot-product self-attention over LSTM hidden states."""
+
+ def __init__(self, hidden_dim: int) -> None:
+ super().__init__()
+ self.attn = nn.Linear(hidden_dim * 2, 1)
+
+ def forward(self, lstm_out: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
+ # lstm_out: (batch, seq_len, hidden_dim*2)
+ scores = self.attn(lstm_out).squeeze(-1) # (batch, seq_len)
+ scores = scores.masked_fill(~mask, -1e9)
+ weights = F.softmax(scores, dim=-1) * mask
+ weights = weights / weights.sum(dim=-1, keepdim=True).clamp_min(1e-9)
+ weights = weights.unsqueeze(-1) # (batch, seq_len, 1)
+ context = (lstm_out * weights).sum(dim=1) # (batch, hidden_dim*2)
+ return context
+
+
+class DeepSequenceModel(nn.Module):
+ """Bidirectional LSTM + attention sequence recommender."""
+
+ def __init__(
+ self,
+ num_items: int,
+ embedding_dim: int = 64,
+ hidden_dim: int = 128,
+ num_layers: int = 2,
+ dropout: float = 0.3,
+ padding_idx: int = 0,
+ ) -> None:
+ super().__init__()
+ self.embedding = nn.Embedding(num_items + 1, embedding_dim, padding_idx=padding_idx)
+ self.lstm = nn.LSTM(
+ input_size=embedding_dim,
+ hidden_size=hidden_dim,
+ num_layers=num_layers,
+ batch_first=True,
+ bidirectional=True,
+ dropout=dropout if num_layers > 1 else 0.0,
+ )
+ self.attention = AttentionLayer(hidden_dim)
+ self.dropout = nn.Dropout(dropout)
+ self.num_items = num_items
+ self.padding_idx = padding_idx
+ self.output_proj = nn.Linear(hidden_dim * 2, num_items + 1)
+
+ def forward(self, item_seq: torch.Tensor) -> torch.Tensor:
+ """Return logits over the item catalogue.
+
+ Parameters
+ ----------
+ item_seq:
+ LongTensor of shape ``(batch_size, seq_len)`` containing item IDs.
+
+ Returns
+ -------
+ torch.Tensor
+ Logit tensor of shape ``(batch_size, num_items + 1)``. Index zero
+ is reserved for padding and is never eligible for recommendation.
+ """
+ emb = self.dropout(self.embedding(item_seq)) # (B, L, E)
+ lstm_out, _ = self.lstm(emb) # (B, L, H*2)
+ mask = item_seq.ne(self.padding_idx)
+ context = self.attention(lstm_out, mask) # (B, H*2)
+ logits = self.output_proj(self.dropout(context)) # (B, num_items + 1)
+ logits[:, self.padding_idx] = float("-inf")
+ return logits
+
+ @torch.no_grad()
+ def recommend(
+ self,
+ item_seq: torch.Tensor,
+ top_k: int = 10,
+ exclude_ids: list[int] | None = None,
+ ) -> list[int]:
+ """Return top-k recommended item IDs for a single sequence."""
+ self.eval()
+ if item_seq.ndim != 2 or not item_seq.ne(self.padding_idx).any():
+ raise ValueError("A recommendation requires at least one known item")
+ if not 1 <= top_k <= self.num_items:
+ raise ValueError(f"top_k must be between 1 and {self.num_items}")
+ excluded = {idx for idx in (exclude_ids or []) if 1 <= idx <= self.num_items}
+ if top_k > self.num_items - len(excluded):
+ raise ValueError("top_k exceeds the remaining eligible catalogue")
+ logits = self.forward(item_seq)
+ for idx in excluded:
+ logits[:, idx] = float("-inf")
+ scores = torch.topk(logits, k=top_k, dim=-1)
+ return scores.indices[0].tolist()
diff --git a/app/api/routes.py b/app/api/routes.py
index 55f518e..af467da 100644
--- a/app/api/routes.py
+++ b/app/api/routes.py
@@ -1,197 +1,195 @@
-"""Validated recommendation routes with bounded inference and fallback behavior."""
-
-from __future__ import annotations
-
-import hashlib
-import json
-import logging
-import time
-from dataclasses import dataclass
-from typing import Literal
-
-from fastapi import APIRouter, Header, HTTPException
-from pydantic import BaseModel, Field
-
-from app.core.config import settings
-from app.core.data_processor import SequenceProcessor
-from app.core.metrics import (
- active_requests,
- cache_hits_total,
- cache_misses_total,
- feedback_events_total,
- model_inference_latency,
- recommendation_latency,
- recommendations_total,
-)
-from app.core.model import DeepSequenceModel
-from app.core.security import api_key_is_valid
-from app.core.serving import AdmissionController, RateLimiter, RecommendationCache
-
-router = APIRouter(prefix="/recommendations", tags=["recommendations"])
-
-
-@dataclass
-class ModelRuntime:
- processor: SequenceProcessor
- model: DeepSequenceModel
- model_version: str
- trained: bool
- popular_items: list[str]
-
-
-_runtime: ModelRuntime | None = None
-_admission = AdmissionController(settings.max_concurrent_inferences)
-_cache = RecommendationCache(settings.cache_ttl_seconds)
-_rate_limiter = RateLimiter(settings.requests_per_minute)
-logger = logging.getLogger(__name__)
-
-
-def init_model(
- processor: SequenceProcessor,
- model: DeepSequenceModel,
- *,
- model_version: str,
- trained: bool,
- popular_items: list[str] | None = None,
-) -> None:
- global _runtime
- _runtime = ModelRuntime(
- processor=processor,
- model=model,
- model_version=model_version,
- trained=trained,
- popular_items=popular_items or list(processor.export_vocabulary())[: settings.max_top_k],
- )
-
-
-class RecommendRequest(BaseModel):
- user_id: str = Field(min_length=1, max_length=128)
- item_sequence: list[str] = Field(min_length=1, max_length=500)
- top_k: int = Field(default=settings.top_k, ge=1, le=settings.max_top_k)
-
-
-class RecommendResponse(BaseModel):
- user_id: str
- recommendations: list[str]
- latency_ms: float
- model_version: str
- fallback: bool = False
- cache_hit: bool = False
-
-
-class FeedbackRequest(BaseModel):
- impression_id: str = Field(min_length=1, max_length=128)
- user_id: str = Field(min_length=1, max_length=128)
- item_id: str = Field(min_length=1, max_length=256)
- event_type: Literal["impression", "click", "skip", "cart", "purchase", "dislike"]
- position: int | None = Field(default=None, ge=0, le=10_000)
- model_version: str = Field(min_length=1, max_length=128)
-
-
-def _authorize(api_key: str | None) -> None:
- if not api_key_is_valid(api_key, settings.api_key):
- raise HTTPException(status_code=401, detail="Invalid API key")
-
-
-def _response(
- request: RecommendRequest,
- recommendations: list[str],
- started: float,
- *,
- fallback: bool = False,
- cache_hit: bool = False,
-) -> RecommendResponse:
- assert _runtime is not None
- return RecommendResponse(
- user_id=request.user_id,
- recommendations=recommendations,
- latency_ms=(time.perf_counter() - started) * 1_000,
- model_version=_runtime.model_version,
- fallback=fallback,
- cache_hit=cache_hit,
- )
-
-
-@router.post("/", response_model=RecommendResponse, summary="Generate recommendations")
-def recommend(
- req: RecommendRequest, x_api_key: str | None = Header(default=None)
-) -> RecommendResponse:
- _authorize(x_api_key)
- if not _rate_limiter.allow(req.user_id):
- raise HTTPException(status_code=429, detail="Recommendation rate limit exceeded")
- if _runtime is None:
- raise HTTPException(status_code=503, detail="Model not initialised")
- if req.top_k > _runtime.processor.vocab_size:
- raise HTTPException(status_code=422, detail="top_k exceeds catalogue size")
-
- known_items = [
- item for item in req.item_sequence if _runtime.processor.item_to_idx(item) != 0
- ]
- if not known_items:
- raise HTTPException(status_code=422, detail="Sequence contains no known catalogue items")
- remaining_items = _runtime.processor.vocab_size - len(set(known_items))
- if req.top_k > remaining_items:
- raise HTTPException(status_code=422, detail="top_k exceeds remaining eligible items")
-
- started = time.perf_counter()
- cache_key = _cache.key(_runtime.model_version, known_items, req.top_k)
- cached = _cache.get(cache_key)
- if cached is not None:
- cache_hits_total.inc()
- recommendations_total.labels(status="cache_hit").inc()
- return _response(req, cached, started, cache_hit=True)
- cache_misses_total.inc()
-
- if not _admission.acquire():
- recommendations_total.labels(status="fallback_overload").inc()
- return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True)
-
- active_requests.inc()
- try:
- tensor = _runtime.processor.to_tensor(known_items)
- infer_started = time.perf_counter()
- indices = _runtime.model.recommend(
- tensor,
- top_k=req.top_k,
- exclude_ids=[_runtime.processor.item_to_idx(item) for item in known_items],
- )
- inference_ms = (time.perf_counter() - infer_started) * 1_000
- model_inference_latency.observe(inference_ms / 1_000)
- if inference_ms > settings.max_inference_ms:
- recommendations_total.labels(status="fallback_latency").inc()
- return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True)
-
- decoded = _runtime.processor.decode_recommendations(indices)
- recommendations = [item for item in decoded if item is not None]
- _cache.put(cache_key, recommendations)
- recommendations_total.labels(status="success").inc()
- return _response(req, recommendations, started)
- except (ValueError, RuntimeError, KeyError) as exc:
- recommendations_total.labels(status="error").inc()
- raise HTTPException(status_code=500, detail="Recommendation inference failed") from exc
- finally:
- recommendation_latency.observe(time.perf_counter() - started)
- active_requests.dec()
- _admission.release()
-
-
-@router.post("/feedback", status_code=202, summary="Capture recommendation feedback")
-def feedback(req: FeedbackRequest, x_api_key: str | None = Header(default=None)) -> dict:
- """Emit a privacy-minimized event for collection by the platform log pipeline."""
- _authorize(x_api_key)
- anonymized_user = hashlib.sha256(req.user_id.encode()).hexdigest()[:16]
- event = req.model_dump(exclude={"user_id"}) | {"anonymous_user_id": anonymized_user}
- logger.info("recommendation_feedback=%s", json.dumps(event, sort_keys=True))
- feedback_events_total.labels(event_type=req.event_type).inc()
- return {"accepted": True, "impression_id": req.impression_id}
-
-
-@router.get("/health", summary="Model readiness check")
-def health() -> dict:
- return {
- "status": "ready" if _runtime is not None else "not_ready",
- "model_loaded": _runtime is not None,
- "trained_model": _runtime.trained if _runtime else False,
- "model_version": _runtime.model_version if _runtime else None,
- "vocab_size": _runtime.processor.vocab_size if _runtime else 0,
- }
+"""Validated recommendation routes with bounded inference and fallback behavior."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import time
+from dataclasses import dataclass
+from typing import Literal
+
+from fastapi import APIRouter, Header, HTTPException
+from pydantic import BaseModel, Field
+
+from app.core.config import settings
+from app.core.data_processor import SequenceProcessor
+from app.core.metrics import (
+ active_requests,
+ cache_hits_total,
+ cache_misses_total,
+ feedback_events_total,
+ model_inference_latency,
+ recommendation_latency,
+ recommendations_total,
+)
+from app.core.model import DeepSequenceModel
+from app.core.security import api_key_is_valid
+from app.core.serving import AdmissionController, RateLimiter, RecommendationCache
+
+router = APIRouter(prefix="/recommendations", tags=["recommendations"])
+
+
+@dataclass
+class ModelRuntime:
+ processor: SequenceProcessor
+ model: DeepSequenceModel
+ model_version: str
+ trained: bool
+ popular_items: list[str]
+
+
+_runtime: ModelRuntime | None = None
+_admission = AdmissionController(settings.max_concurrent_inferences)
+_cache = RecommendationCache(settings.cache_ttl_seconds)
+_rate_limiter = RateLimiter(settings.requests_per_minute)
+logger = logging.getLogger(__name__)
+
+
+def init_model(
+ processor: SequenceProcessor,
+ model: DeepSequenceModel,
+ *,
+ model_version: str,
+ trained: bool,
+ popular_items: list[str] | None = None,
+) -> None:
+ global _runtime
+ _runtime = ModelRuntime(
+ processor=processor,
+ model=model,
+ model_version=model_version,
+ trained=trained,
+ popular_items=popular_items or list(processor.export_vocabulary())[: settings.max_top_k],
+ )
+
+
+class RecommendRequest(BaseModel):
+ user_id: str = Field(min_length=1, max_length=128)
+ item_sequence: list[str] = Field(min_length=1, max_length=500)
+ top_k: int = Field(default=settings.top_k, ge=1, le=settings.max_top_k)
+
+
+class RecommendResponse(BaseModel):
+ user_id: str
+ recommendations: list[str]
+ latency_ms: float
+ model_version: str
+ fallback: bool = False
+ cache_hit: bool = False
+
+
+class FeedbackRequest(BaseModel):
+ impression_id: str = Field(min_length=1, max_length=128)
+ user_id: str = Field(min_length=1, max_length=128)
+ item_id: str = Field(min_length=1, max_length=256)
+ event_type: Literal["impression", "click", "skip", "cart", "purchase", "dislike"]
+ position: int | None = Field(default=None, ge=0, le=10_000)
+ model_version: str = Field(min_length=1, max_length=128)
+
+
+def _authorize(api_key: str | None) -> None:
+ if not api_key_is_valid(api_key, settings.api_key):
+ raise HTTPException(status_code=401, detail="Invalid API key")
+
+
+def _response(
+ request: RecommendRequest,
+ recommendations: list[str],
+ started: float,
+ *,
+ fallback: bool = False,
+ cache_hit: bool = False,
+) -> RecommendResponse:
+ assert _runtime is not None
+ return RecommendResponse(
+ user_id=request.user_id,
+ recommendations=recommendations,
+ latency_ms=(time.perf_counter() - started) * 1_000,
+ model_version=_runtime.model_version,
+ fallback=fallback,
+ cache_hit=cache_hit,
+ )
+
+
+@router.post("/", response_model=RecommendResponse, summary="Generate recommendations")
+def recommend(
+ req: RecommendRequest, x_api_key: str | None = Header(default=None)
+) -> RecommendResponse:
+ _authorize(x_api_key)
+ if not _rate_limiter.allow(req.user_id):
+ raise HTTPException(status_code=429, detail="Recommendation rate limit exceeded")
+ if _runtime is None:
+ raise HTTPException(status_code=503, detail="Model not initialised")
+ if req.top_k > _runtime.processor.vocab_size:
+ raise HTTPException(status_code=422, detail="top_k exceeds catalogue size")
+
+ known_items = [item for item in req.item_sequence if _runtime.processor.item_to_idx(item) != 0]
+ if not known_items:
+ raise HTTPException(status_code=422, detail="Sequence contains no known catalogue items")
+ remaining_items = _runtime.processor.vocab_size - len(set(known_items))
+ if req.top_k > remaining_items:
+ raise HTTPException(status_code=422, detail="top_k exceeds remaining eligible items")
+
+ started = time.perf_counter()
+ cache_key = _cache.key(_runtime.model_version, known_items, req.top_k)
+ cached = _cache.get(cache_key)
+ if cached is not None:
+ cache_hits_total.inc()
+ recommendations_total.labels(status="cache_hit").inc()
+ return _response(req, cached, started, cache_hit=True)
+ cache_misses_total.inc()
+
+ if not _admission.acquire():
+ recommendations_total.labels(status="fallback_overload").inc()
+ return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True)
+
+ active_requests.inc()
+ try:
+ tensor = _runtime.processor.to_tensor(known_items)
+ infer_started = time.perf_counter()
+ indices = _runtime.model.recommend(
+ tensor,
+ top_k=req.top_k,
+ exclude_ids=[_runtime.processor.item_to_idx(item) for item in known_items],
+ )
+ inference_ms = (time.perf_counter() - infer_started) * 1_000
+ model_inference_latency.observe(inference_ms / 1_000)
+ if inference_ms > settings.max_inference_ms:
+ recommendations_total.labels(status="fallback_latency").inc()
+ return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True)
+
+ decoded = _runtime.processor.decode_recommendations(indices)
+ recommendations = [item for item in decoded if item is not None]
+ _cache.put(cache_key, recommendations)
+ recommendations_total.labels(status="success").inc()
+ return _response(req, recommendations, started)
+ except (ValueError, RuntimeError, KeyError) as exc:
+ recommendations_total.labels(status="error").inc()
+ raise HTTPException(status_code=500, detail="Recommendation inference failed") from exc
+ finally:
+ recommendation_latency.observe(time.perf_counter() - started)
+ active_requests.dec()
+ _admission.release()
+
+
+@router.post("/feedback", status_code=202, summary="Capture recommendation feedback")
+def feedback(req: FeedbackRequest, x_api_key: str | None = Header(default=None)) -> dict:
+ """Emit a privacy-minimized event for collection by the platform log pipeline."""
+ _authorize(x_api_key)
+ anonymized_user = hashlib.sha256(req.user_id.encode()).hexdigest()[:16]
+ event = req.model_dump(exclude={"user_id"}) | {"anonymous_user_id": anonymized_user}
+ logger.info("recommendation_feedback=%s", json.dumps(event, sort_keys=True))
+ feedback_events_total.labels(event_type=req.event_type).inc()
+ return {"accepted": True, "impression_id": req.impression_id}
+
+
+@router.get("/health", summary="Model readiness check")
+def health() -> dict:
+ return {
+ "status": "ready" if _runtime is not None else "not_ready",
+ "model_loaded": _runtime is not None,
+ "trained_model": _runtime.trained if _runtime else False,
+ "model_version": _runtime.model_version if _runtime else None,
+ "vocab_size": _runtime.processor.vocab_size if _runtime else 0,
+ }
diff --git a/app/core/artifacts.py b/app/core/artifacts.py
index 553bf88..47abad4 100644
--- a/app/core/artifacts.py
+++ b/app/core/artifacts.py
@@ -1,85 +1,83 @@
-"""Versioned, checksummed model bundles shared by training and serving."""
-
-from __future__ import annotations
-
-import hashlib
-import json
-from pathlib import Path
-from typing import Any
-
-import torch
-from pydantic import BaseModel, Field
-
-from app.core.data_processor import SequenceProcessor
-from app.core.model import DeepSequenceModel
-
-
-class ModelManifest(BaseModel):
- model_version: str
- created_at: str
- architecture: str = "bidirectional-lstm-attention"
- architecture_config: dict[str, Any]
- metrics: dict[str, float] = Field(default_factory=dict)
- dataset_id: str
- weights_sha256: str
- vocabulary_sha256: str
-
-
-def _sha256(path: Path) -> str:
- digest = hashlib.sha256()
- with path.open("rb") as handle:
- for block in iter(lambda: handle.read(1024 * 1024), b""):
- digest.update(block)
- return digest.hexdigest()
-
-
-def save_bundle(
- directory: str | Path,
- model: DeepSequenceModel,
- processor: SequenceProcessor,
- manifest: ModelManifest,
-) -> ModelManifest:
- target = Path(directory)
- target.mkdir(parents=True, exist_ok=True)
- weights_path = target / "model.pt"
- vocabulary_path = target / "vocabulary.json"
- torch.save(model.state_dict(), weights_path)
- vocabulary_path.write_text(
- json.dumps(processor.export_vocabulary(), indent=2, sort_keys=True),
- encoding="utf-8",
- )
- completed = manifest.model_copy(
- update={
- "weights_sha256": _sha256(weights_path),
- "vocabulary_sha256": _sha256(vocabulary_path),
- }
- )
- (target / "manifest.json").write_text(
- completed.model_dump_json(indent=2), encoding="utf-8"
- )
- return completed
-
-
-def load_bundle(
- directory: str | Path,
-) -> tuple[SequenceProcessor, DeepSequenceModel, ModelManifest]:
- target = Path(directory)
- manifest = ModelManifest.model_validate_json(
- (target / "manifest.json").read_text(encoding="utf-8")
- )
- weights_path = target / "model.pt"
- vocabulary_path = target / "vocabulary.json"
- if _sha256(weights_path) != manifest.weights_sha256:
- raise ValueError("Model weights checksum does not match the manifest")
- if _sha256(vocabulary_path) != manifest.vocabulary_sha256:
- raise ValueError("Vocabulary checksum does not match the manifest")
-
- vocabulary = json.loads(vocabulary_path.read_text(encoding="utf-8"))
- config = dict(manifest.architecture_config)
- max_length = int(config.pop("max_sequence_length"))
- processor = SequenceProcessor.from_vocabulary(vocabulary, max_length=max_length)
- model = DeepSequenceModel(num_items=processor.vocab_size, **config)
- state = torch.load(weights_path, map_location="cpu", weights_only=True)
- model.load_state_dict(state, strict=True)
- model.eval()
- return processor, model, manifest
+"""Versioned, checksummed model bundles shared by training and serving."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+import torch
+from pydantic import BaseModel, Field
+
+from app.core.data_processor import SequenceProcessor
+from app.core.model import DeepSequenceModel
+
+
+class ModelManifest(BaseModel):
+ model_version: str
+ created_at: str
+ architecture: str = "bidirectional-lstm-attention"
+ architecture_config: dict[str, Any]
+ metrics: dict[str, float] = Field(default_factory=dict)
+ dataset_id: str
+ weights_sha256: str
+ vocabulary_sha256: str
+
+
+def _sha256(path: Path) -> str:
+ digest = hashlib.sha256()
+ with path.open("rb") as handle:
+ for block in iter(lambda: handle.read(1024 * 1024), b""):
+ digest.update(block)
+ return digest.hexdigest()
+
+
+def save_bundle(
+ directory: str | Path,
+ model: DeepSequenceModel,
+ processor: SequenceProcessor,
+ manifest: ModelManifest,
+) -> ModelManifest:
+ target = Path(directory)
+ target.mkdir(parents=True, exist_ok=True)
+ weights_path = target / "model.pt"
+ vocabulary_path = target / "vocabulary.json"
+ torch.save(model.state_dict(), weights_path)
+ vocabulary_path.write_text(
+ json.dumps(processor.export_vocabulary(), indent=2, sort_keys=True),
+ encoding="utf-8",
+ )
+ completed = manifest.model_copy(
+ update={
+ "weights_sha256": _sha256(weights_path),
+ "vocabulary_sha256": _sha256(vocabulary_path),
+ }
+ )
+ (target / "manifest.json").write_text(completed.model_dump_json(indent=2), encoding="utf-8")
+ return completed
+
+
+def load_bundle(
+ directory: str | Path,
+) -> tuple[SequenceProcessor, DeepSequenceModel, ModelManifest]:
+ target = Path(directory)
+ manifest = ModelManifest.model_validate_json(
+ (target / "manifest.json").read_text(encoding="utf-8")
+ )
+ weights_path = target / "model.pt"
+ vocabulary_path = target / "vocabulary.json"
+ if _sha256(weights_path) != manifest.weights_sha256:
+ raise ValueError("Model weights checksum does not match the manifest")
+ if _sha256(vocabulary_path) != manifest.vocabulary_sha256:
+ raise ValueError("Vocabulary checksum does not match the manifest")
+
+ vocabulary = json.loads(vocabulary_path.read_text(encoding="utf-8"))
+ config = dict(manifest.architecture_config)
+ max_length = int(config.pop("max_sequence_length"))
+ processor = SequenceProcessor.from_vocabulary(vocabulary, max_length=max_length)
+ model = DeepSequenceModel(num_items=processor.vocab_size, **config)
+ state = torch.load(weights_path, map_location="cpu", weights_only=True)
+ model.load_state_dict(state, strict=True)
+ model.eval()
+ return processor, model, manifest
diff --git a/app/core/config.py b/app/core/config.py
index 9918b80..a59a710 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -1,25 +1,26 @@
-"""Application settings loaded from environment variables."""
-
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class Settings(BaseSettings):
- model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
- environment: str = "development"
- log_level: str = "INFO"
-
- # Model hyper-parameters
- embedding_dim: int = 64
- hidden_dim: int = 128
- num_layers: int = 2
- max_sequence_length: int = 50
- top_k: int = 10
- max_top_k: int = 50
- model_bundle_path: str = "models/current"
- max_inference_ms: float = 250.0
- max_concurrent_inferences: int = 8
- cache_ttl_seconds: int = 30
- requests_per_minute: int = 120
- api_key: str | None = None
-
-settings = Settings()
+"""Application settings loaded from environment variables."""
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+ model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
+ environment: str = "development"
+ log_level: str = "INFO"
+
+ # Model hyper-parameters
+ embedding_dim: int = 64
+ hidden_dim: int = 128
+ num_layers: int = 2
+ max_sequence_length: int = 50
+ top_k: int = 10
+ max_top_k: int = 50
+ model_bundle_path: str = "models/current"
+ max_inference_ms: float = 250.0
+ max_concurrent_inferences: int = 8
+ cache_ttl_seconds: int = 30
+ requests_per_minute: int = 120
+ api_key: str | None = None
+
+
+settings = Settings()
diff --git a/app/core/data_processor.py b/app/core/data_processor.py
index 0beb6fe..e569c7e 100644
--- a/app/core/data_processor.py
+++ b/app/core/data_processor.py
@@ -1,86 +1,82 @@
-"""Data pre-processing utilities for sequence recommendation."""
-
-from __future__ import annotations
-
-from typing import Dict, List, Optional
-
-import torch
-
-
-class SequenceProcessor:
- """Encode raw item interaction histories into padded tensors."""
-
- def __init__(self, max_length: int = 50) -> None:
- self.max_length = max_length
- self._item2idx: Dict[str, int] = {}
- self._idx2item: Dict[int, str] = {}
- self._next_idx: int = 1 # 0 reserved for padding
-
- # ------------------------------------------------------------------
- # Vocabulary management
- # ------------------------------------------------------------------
-
- def fit(self, sequences: List[List[str]]) -> "SequenceProcessor":
- """Build item vocabulary from a list of interaction sequences."""
- for seq in sequences:
- for item in seq:
- if item not in self._item2idx:
- self._item2idx[item] = self._next_idx
- self._idx2item[self._next_idx] = item
- self._next_idx += 1
- return self
-
- @property
- def vocab_size(self) -> int:
- """Number of known items (excluding padding index 0)."""
- return len(self._item2idx)
-
- def item_to_idx(self, item: str) -> int:
- """Return numeric index for an item; 0 for unknown items."""
- return self._item2idx.get(item, 0)
-
- def idx_to_item(self, idx: int) -> Optional[str]:
- """Return item identifier for a numeric index."""
- return self._idx2item.get(idx)
-
- # ------------------------------------------------------------------
- # Encoding
- # ------------------------------------------------------------------
-
- def encode(self, sequence: List[str]) -> List[int]:
- """Convert a list of item IDs to a list of numeric indices."""
- return [self.item_to_idx(item) for item in sequence]
-
- def pad_sequence(self, indices: List[int]) -> List[int]:
- """Truncate to max_length and left-pad with zeros."""
- indices = indices[-self.max_length :]
- padded = [0] * (self.max_length - len(indices)) + indices
- return padded
-
- def to_tensor(self, sequence: List[str]) -> torch.Tensor:
- """Encode and pad a single sequence, returning shape ``(1, max_length)``."""
- indices = self.encode(sequence)
- padded = self.pad_sequence(indices)
- return torch.tensor([padded], dtype=torch.long)
-
- def decode_recommendations(self, indices: List[int]) -> List[Optional[str]]:
- """Map numeric recommendation indices back to item identifiers."""
- return [self.idx_to_item(idx) for idx in indices]
-
- def export_vocabulary(self) -> Dict[str, int]:
- """Return a stable copy of the item-to-index mapping."""
- return dict(self._item2idx)
-
- @classmethod
- def from_vocabulary(
- cls, vocabulary: Dict[str, int], max_length: int = 50
- ) -> "SequenceProcessor":
- """Restore and validate a persisted vocabulary."""
- expected = set(range(1, len(vocabulary) + 1))
- if set(vocabulary.values()) != expected:
- raise ValueError("Vocabulary indices must be contiguous and start at one")
- processor = cls(max_length=max_length)
- processor._item2idx = dict(vocabulary)
- processor._idx2item = {index: item for item, index in vocabulary.items()}
- processor._next_idx = len(vocabulary) + 1
- return processor
+"""Data pre-processing utilities for sequence recommendation."""
+
+from __future__ import annotations
+
+import torch
+
+
+class SequenceProcessor:
+ """Encode raw item interaction histories into padded tensors."""
+
+ def __init__(self, max_length: int = 50) -> None:
+ self.max_length = max_length
+ self._item2idx: dict[str, int] = {}
+ self._idx2item: dict[int, str] = {}
+ self._next_idx: int = 1 # 0 reserved for padding
+
+ # ------------------------------------------------------------------
+ # Vocabulary management
+ # ------------------------------------------------------------------
+
+ def fit(self, sequences: list[list[str]]) -> SequenceProcessor:
+ """Build item vocabulary from a list of interaction sequences."""
+ for seq in sequences:
+ for item in seq:
+ if item not in self._item2idx:
+ self._item2idx[item] = self._next_idx
+ self._idx2item[self._next_idx] = item
+ self._next_idx += 1
+ return self
+
+ @property
+ def vocab_size(self) -> int:
+ """Number of known items (excluding padding index 0)."""
+ return len(self._item2idx)
+
+ def item_to_idx(self, item: str) -> int:
+ """Return numeric index for an item; 0 for unknown items."""
+ return self._item2idx.get(item, 0)
+
+ def idx_to_item(self, idx: int) -> str | None:
+ """Return item identifier for a numeric index."""
+ return self._idx2item.get(idx)
+
+ # ------------------------------------------------------------------
+ # Encoding
+ # ------------------------------------------------------------------
+
+ def encode(self, sequence: list[str]) -> list[int]:
+ """Convert a list of item IDs to a list of numeric indices."""
+ return [self.item_to_idx(item) for item in sequence]
+
+ def pad_sequence(self, indices: list[int]) -> list[int]:
+ """Truncate to max_length and left-pad with zeros."""
+ indices = indices[-self.max_length :]
+ padded = [0] * (self.max_length - len(indices)) + indices
+ return padded
+
+ def to_tensor(self, sequence: list[str]) -> torch.Tensor:
+ """Encode and pad a single sequence, returning shape ``(1, max_length)``."""
+ indices = self.encode(sequence)
+ padded = self.pad_sequence(indices)
+ return torch.tensor([padded], dtype=torch.long)
+
+ def decode_recommendations(self, indices: list[int]) -> list[str | None]:
+ """Map numeric recommendation indices back to item identifiers."""
+ return [self.idx_to_item(idx) for idx in indices]
+
+ def export_vocabulary(self) -> dict[str, int]:
+ """Return a stable copy of the item-to-index mapping."""
+ return dict(self._item2idx)
+
+ @classmethod
+ def from_vocabulary(cls, vocabulary: dict[str, int], max_length: int = 50) -> SequenceProcessor:
+ """Restore and validate a persisted vocabulary."""
+ expected = set(range(1, len(vocabulary) + 1))
+ if set(vocabulary.values()) != expected:
+ raise ValueError("Vocabulary indices must be contiguous and start at one")
+ processor = cls(max_length=max_length)
+ processor._item2idx = dict(vocabulary)
+ processor._idx2item = {index: item for item, index in vocabulary.items()}
+ processor._next_idx = len(vocabulary) + 1
+ return processor
diff --git a/app/core/model.py b/app/core/model.py
index 00a1ebd..a4de21b 100644
--- a/app/core/model.py
+++ b/app/core/model.py
@@ -1,108 +1,106 @@
-"""Deep sequence recommender model.
-
-Architecture
-------------
-- Item embedding layer
-- Bidirectional LSTM encoder
-- Dot-product attention over encoder outputs
-- Linear projection to item vocabulary (logits)
-"""
-
-from __future__ import annotations
-
-from typing import List, Optional
-
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-
-
-class AttentionLayer(nn.Module):
- """Scaled dot-product self-attention over LSTM hidden states."""
-
- def __init__(self, hidden_dim: int) -> None:
- super().__init__()
- self.attn = nn.Linear(hidden_dim * 2, 1)
-
- def forward(self, lstm_out: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
- # lstm_out: (batch, seq_len, hidden_dim*2)
- scores = self.attn(lstm_out).squeeze(-1) # (batch, seq_len)
- scores = scores.masked_fill(~mask, -1e9)
- weights = F.softmax(scores, dim=-1) * mask
- weights = weights / weights.sum(dim=-1, keepdim=True).clamp_min(1e-9)
- weights = weights.unsqueeze(-1) # (batch, seq_len, 1)
- context = (lstm_out * weights).sum(dim=1) # (batch, hidden_dim*2)
- return context
-
-
-class DeepSequenceModel(nn.Module):
- """Bidirectional LSTM + attention sequence recommender."""
-
- def __init__(
- self,
- num_items: int,
- embedding_dim: int = 64,
- hidden_dim: int = 128,
- num_layers: int = 2,
- dropout: float = 0.3,
- padding_idx: int = 0,
- ) -> None:
- super().__init__()
- self.embedding = nn.Embedding(num_items + 1, embedding_dim, padding_idx=padding_idx)
- self.lstm = nn.LSTM(
- input_size=embedding_dim,
- hidden_size=hidden_dim,
- num_layers=num_layers,
- batch_first=True,
- bidirectional=True,
- dropout=dropout if num_layers > 1 else 0.0,
- )
- self.attention = AttentionLayer(hidden_dim)
- self.dropout = nn.Dropout(dropout)
- self.num_items = num_items
- self.padding_idx = padding_idx
- self.output_proj = nn.Linear(hidden_dim * 2, num_items + 1)
-
- def forward(self, item_seq: torch.Tensor) -> torch.Tensor:
- """Return logits over the item catalogue.
-
- Parameters
- ----------
- item_seq:
- LongTensor of shape ``(batch_size, seq_len)`` containing item IDs.
-
- Returns
- -------
- torch.Tensor
- Logit tensor of shape ``(batch_size, num_items + 1)``. Index zero
- is reserved for padding and is never eligible for recommendation.
- """
- emb = self.dropout(self.embedding(item_seq)) # (B, L, E)
- lstm_out, _ = self.lstm(emb) # (B, L, H*2)
- mask = item_seq.ne(self.padding_idx)
- context = self.attention(lstm_out, mask) # (B, H*2)
- logits = self.output_proj(self.dropout(context)) # (B, num_items + 1)
- logits[:, self.padding_idx] = float("-inf")
- return logits
-
- @torch.no_grad()
- def recommend(
- self,
- item_seq: torch.Tensor,
- top_k: int = 10,
- exclude_ids: Optional[List[int]] = None,
- ) -> List[int]:
- """Return top-k recommended item IDs for a single sequence."""
- self.eval()
- if item_seq.ndim != 2 or not item_seq.ne(self.padding_idx).any():
- raise ValueError("A recommendation requires at least one known item")
- if not 1 <= top_k <= self.num_items:
- raise ValueError(f"top_k must be between 1 and {self.num_items}")
- excluded = {idx for idx in (exclude_ids or []) if 1 <= idx <= self.num_items}
- if top_k > self.num_items - len(excluded):
- raise ValueError("top_k exceeds the remaining eligible catalogue")
- logits = self.forward(item_seq)
- for idx in excluded:
- logits[:, idx] = float("-inf")
- scores = torch.topk(logits, k=top_k, dim=-1)
- return scores.indices[0].tolist()
+"""Deep sequence recommender model.
+
+Architecture
+------------
+- Item embedding layer
+- Bidirectional LSTM encoder
+- Dot-product attention over encoder outputs
+- Linear projection to item vocabulary (logits)
+"""
+
+from __future__ import annotations
+
+import torch
+import torch.nn.functional as F
+from torch import nn
+
+
+class AttentionLayer(nn.Module):
+ """Scaled dot-product self-attention over LSTM hidden states."""
+
+ def __init__(self, hidden_dim: int) -> None:
+ super().__init__()
+ self.attn = nn.Linear(hidden_dim * 2, 1)
+
+ def forward(self, lstm_out: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
+ # lstm_out: (batch, seq_len, hidden_dim*2)
+ scores = self.attn(lstm_out).squeeze(-1) # (batch, seq_len)
+ scores = scores.masked_fill(~mask, -1e9)
+ weights = F.softmax(scores, dim=-1) * mask
+ weights = weights / weights.sum(dim=-1, keepdim=True).clamp_min(1e-9)
+ weights = weights.unsqueeze(-1) # (batch, seq_len, 1)
+ context = (lstm_out * weights).sum(dim=1) # (batch, hidden_dim*2)
+ return context
+
+
+class DeepSequenceModel(nn.Module):
+ """Bidirectional LSTM + attention sequence recommender."""
+
+ def __init__(
+ self,
+ num_items: int,
+ embedding_dim: int = 64,
+ hidden_dim: int = 128,
+ num_layers: int = 2,
+ dropout: float = 0.3,
+ padding_idx: int = 0,
+ ) -> None:
+ super().__init__()
+ self.embedding = nn.Embedding(num_items + 1, embedding_dim, padding_idx=padding_idx)
+ self.lstm = nn.LSTM(
+ input_size=embedding_dim,
+ hidden_size=hidden_dim,
+ num_layers=num_layers,
+ batch_first=True,
+ bidirectional=True,
+ dropout=dropout if num_layers > 1 else 0.0,
+ )
+ self.attention = AttentionLayer(hidden_dim)
+ self.dropout = nn.Dropout(dropout)
+ self.num_items = num_items
+ self.padding_idx = padding_idx
+ self.output_proj = nn.Linear(hidden_dim * 2, num_items + 1)
+
+ def forward(self, item_seq: torch.Tensor) -> torch.Tensor:
+ """Return logits over the item catalogue.
+
+ Parameters
+ ----------
+ item_seq:
+ LongTensor of shape ``(batch_size, seq_len)`` containing item IDs.
+
+ Returns
+ -------
+ torch.Tensor
+ Logit tensor of shape ``(batch_size, num_items + 1)``. Index zero
+ is reserved for padding and is never eligible for recommendation.
+ """
+ emb = self.dropout(self.embedding(item_seq)) # (B, L, E)
+ lstm_out, _ = self.lstm(emb) # (B, L, H*2)
+ mask = item_seq.ne(self.padding_idx)
+ context = self.attention(lstm_out, mask) # (B, H*2)
+ logits = self.output_proj(self.dropout(context)) # (B, num_items + 1)
+ logits[:, self.padding_idx] = float("-inf")
+ return logits
+
+ @torch.no_grad()
+ def recommend(
+ self,
+ item_seq: torch.Tensor,
+ top_k: int = 10,
+ exclude_ids: list[int] | None = None,
+ ) -> list[int]:
+ """Return top-k recommended item IDs for a single sequence."""
+ self.eval()
+ if item_seq.ndim != 2 or not item_seq.ne(self.padding_idx).any():
+ raise ValueError("A recommendation requires at least one known item")
+ if not 1 <= top_k <= self.num_items:
+ raise ValueError(f"top_k must be between 1 and {self.num_items}")
+ excluded = {idx for idx in (exclude_ids or []) if 1 <= idx <= self.num_items}
+ if top_k > self.num_items - len(excluded):
+ raise ValueError("top_k exceeds the remaining eligible catalogue")
+ logits = self.forward(item_seq)
+ for idx in excluded:
+ logits[:, idx] = float("-inf")
+ scores = torch.topk(logits, k=top_k, dim=-1)
+ return scores.indices[0].tolist()
diff --git a/benchmarks/inference.py b/benchmarks/inference.py
index a97cc32..b4bf3e0 100644
--- a/benchmarks/inference.py
+++ b/benchmarks/inference.py
@@ -1,79 +1,77 @@
-"""Deterministic single-process inference benchmark."""
-
-from __future__ import annotations
-
-import argparse
-import json
-import math
-import platform
-import statistics
-import time
-
-import torch
-
-from app.core.data_processor import SequenceProcessor
-from app.core.model import DeepSequenceModel
-
-
-def _percentile(values: list[float], fraction: float) -> float:
- return sorted(values)[max(0, math.ceil(len(values) * fraction) - 1)]
-
-
-def run_benchmark(iterations: int = 200, warmup: int = 20) -> dict:
- if iterations < 1 or warmup < 0:
- raise ValueError("iterations must be positive and warmup non-negative")
- torch.manual_seed(7)
- processor = SequenceProcessor(max_length=50).fit(
- [[f"item_{index}" for index in range(500)]]
- )
- model = DeepSequenceModel(
- num_items=processor.vocab_size,
- embedding_dim=32,
- hidden_dim=64,
- num_layers=1,
- ).eval()
- tensor = processor.to_tensor([f"item_{index}" for index in range(20)])
- for _ in range(warmup):
- model.recommend(tensor, top_k=10)
- latencies = []
- started = time.perf_counter()
- for _ in range(iterations):
- call_started = time.perf_counter()
- model.recommend(tensor, top_k=10)
- latencies.append((time.perf_counter() - call_started) * 1_000)
- duration = time.perf_counter() - started
- return {
- "scope": "single-process CPU PyTorch inference; batch=1; no HTTP or network",
- "environment": {
- "python": platform.python_version(),
- "pytorch": torch.__version__,
- "platform": platform.platform(),
- "processor": platform.processor() or "not reported",
- },
- "workload": {
- "iterations": iterations,
- "warmup": warmup,
- "catalogue_size": processor.vocab_size,
- "sequence_length": 50,
- "top_k": 10,
- },
- "results": {
- "latency_ms_mean": round(statistics.fmean(latencies), 3),
- "latency_ms_p50": round(statistics.median(latencies), 3),
- "latency_ms_p95": round(_percentile(latencies, 0.95), 3),
- "latency_ms_p99": round(_percentile(latencies, 0.99), 3),
- "sequential_inferences_per_second": round(iterations / duration, 3),
- },
- }
-
-
-def main() -> None:
- parser = argparse.ArgumentParser(description=__doc__)
- parser.add_argument("--iterations", type=int, default=200)
- parser.add_argument("--warmup", type=int, default=20)
- args = parser.parse_args()
- print(json.dumps(run_benchmark(args.iterations, args.warmup), indent=2))
-
-
-if __name__ == "__main__":
- main()
+"""Deterministic single-process inference benchmark."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import platform
+import statistics
+import time
+
+import torch
+
+from app.core.data_processor import SequenceProcessor
+from app.core.model import DeepSequenceModel
+
+
+def _percentile(values: list[float], fraction: float) -> float:
+ return sorted(values)[max(0, math.ceil(len(values) * fraction) - 1)]
+
+
+def run_benchmark(iterations: int = 200, warmup: int = 20) -> dict:
+ if iterations < 1 or warmup < 0:
+ raise ValueError("iterations must be positive and warmup non-negative")
+ torch.manual_seed(7)
+ processor = SequenceProcessor(max_length=50).fit([[f"item_{index}" for index in range(500)]])
+ model = DeepSequenceModel(
+ num_items=processor.vocab_size,
+ embedding_dim=32,
+ hidden_dim=64,
+ num_layers=1,
+ ).eval()
+ tensor = processor.to_tensor([f"item_{index}" for index in range(20)])
+ for _ in range(warmup):
+ model.recommend(tensor, top_k=10)
+ latencies = []
+ started = time.perf_counter()
+ for _ in range(iterations):
+ call_started = time.perf_counter()
+ model.recommend(tensor, top_k=10)
+ latencies.append((time.perf_counter() - call_started) * 1_000)
+ duration = time.perf_counter() - started
+ return {
+ "scope": "single-process CPU PyTorch inference; batch=1; no HTTP or network",
+ "environment": {
+ "python": platform.python_version(),
+ "pytorch": torch.__version__,
+ "platform": platform.platform(),
+ "processor": platform.processor() or "not reported",
+ },
+ "workload": {
+ "iterations": iterations,
+ "warmup": warmup,
+ "catalogue_size": processor.vocab_size,
+ "sequence_length": 50,
+ "top_k": 10,
+ },
+ "results": {
+ "latency_ms_mean": round(statistics.fmean(latencies), 3),
+ "latency_ms_p50": round(statistics.median(latencies), 3),
+ "latency_ms_p95": round(_percentile(latencies, 0.95), 3),
+ "latency_ms_p99": round(_percentile(latencies, 0.99), 3),
+ "sequential_inferences_per_second": round(iterations / duration, 3),
+ },
+ }
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--iterations", type=int, default=200)
+ parser.add_argument("--warmup", type=int, default=20)
+ args = parser.parse_args()
+ print(json.dumps(run_benchmark(args.iterations, args.warmup), indent=2))
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/generate_demo_data.py b/scripts/generate_demo_data.py
index 4bcab56..60bf0bc 100644
--- a/scripts/generate_demo_data.py
+++ b/scripts/generate_demo_data.py
@@ -1,42 +1,42 @@
-"""Generate a deterministic interaction dataset for local pipeline validation."""
-
-from __future__ import annotations
-
-import argparse
-from datetime import datetime, timedelta, timezone
-
-from src.training.data import InteractionEvent, write_jsonl
-
-
-def generate_events(sessions: int = 20, events_per_session: int = 6) -> list[InteractionEvent]:
- events = []
- started = datetime(2026, 1, 1, tzinfo=timezone.utc)
- index = 0
- for session_index in range(sessions):
- for position in range(events_per_session):
- item_index = (session_index + position) % 12
- events.append(
- InteractionEvent(
- event_id=f"event-{index}",
- user_id=f"user-{session_index % 5}",
- session_id=f"session-{session_index}",
- item_id=f"item-{item_index}",
- event_type="purchase" if position == events_per_session - 1 else "click",
- timestamp=started + timedelta(minutes=index),
- )
- )
- index += 1
- return events
-
-
-def main() -> None:
- parser = argparse.ArgumentParser(description=__doc__)
- parser.add_argument("--output", default="data/demo-events.jsonl")
- args = parser.parse_args()
- events = generate_events()
- write_jsonl(args.output, events)
- print(f"Wrote {len(events)} events to {args.output}")
-
-
-if __name__ == "__main__":
- main()
+"""Generate a deterministic interaction dataset for local pipeline validation."""
+
+from __future__ import annotations
+
+import argparse
+from datetime import UTC, datetime, timedelta
+
+from src.training.data import InteractionEvent, write_jsonl
+
+
+def generate_events(sessions: int = 20, events_per_session: int = 6) -> list[InteractionEvent]:
+ events = []
+ started = datetime(2026, 1, 1, tzinfo=UTC)
+ index = 0
+ for session_index in range(sessions):
+ for position in range(events_per_session):
+ item_index = (session_index + position) % 12
+ events.append(
+ InteractionEvent(
+ event_id=f"event-{index}",
+ user_id=f"user-{session_index % 5}",
+ session_id=f"session-{session_index}",
+ item_id=f"item-{item_index}",
+ event_type="purchase" if position == events_per_session - 1 else "click",
+ timestamp=started + timedelta(minutes=index),
+ )
+ )
+ index += 1
+ return events
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--output", default="data/demo-events.jsonl")
+ args = parser.parse_args()
+ events = generate_events()
+ write_jsonl(args.output, events)
+ print(f"Wrote {len(events)} events to {args.output}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/src/training/baselines.py b/src/training/baselines.py
index 501c724..e5b124d 100644
--- a/src/training/baselines.py
+++ b/src/training/baselines.py
@@ -1,19 +1,19 @@
-"""Auditable non-neural baselines used as model promotion gates."""
-
-from __future__ import annotations
-
-from collections import Counter
-
-
-class PopularityBaseline:
- def __init__(self) -> None:
- self.ranking: list[str] = []
-
- def fit(self, sequences: list[list[str]]) -> "PopularityBaseline":
- counts = Counter(item for sequence in sequences for item in sequence)
- self.ranking = [item for item, _ in counts.most_common()]
- return self
-
- def recommend(self, seen: list[str], top_k: int) -> list[str]:
- excluded = set(seen)
- return [item for item in self.ranking if item not in excluded][:top_k]
+"""Auditable non-neural baselines used as model promotion gates."""
+
+from __future__ import annotations
+
+from collections import Counter
+
+
+class PopularityBaseline:
+ def __init__(self) -> None:
+ self.ranking: list[str] = []
+
+ def fit(self, sequences: list[list[str]]) -> PopularityBaseline:
+ counts = Counter(item for sequence in sequences for item in sequence)
+ self.ranking = [item for item, _ in counts.most_common()]
+ return self
+
+ def recommend(self, seen: list[str], top_k: int) -> list[str]:
+ excluded = set(seen)
+ return [item for item in self.ranking if item not in excluded][:top_k]
diff --git a/src/training/metrics.py b/src/training/metrics.py
index b4cfd17..cb10577 100644
--- a/src/training/metrics.py
+++ b/src/training/metrics.py
@@ -1,45 +1,43 @@
-"""Ranking metrics with explicit definitions and no framework dependency."""
-
-from __future__ import annotations
-
-import math
-
-
-def recall_at_k(predictions: list[list[str]], targets: list[str]) -> float:
- return sum(target in predicted for predicted, target in zip(predictions, targets)) / len(targets)
-
-
-def mrr_at_k(predictions: list[list[str]], targets: list[str]) -> float:
- reciprocal_ranks = []
- for predicted, target in zip(predictions, targets):
- reciprocal_ranks.append(
- 1 / (predicted.index(target) + 1) if target in predicted else 0.0
- )
- return sum(reciprocal_ranks) / len(targets)
-
-
-def ndcg_at_k(predictions: list[list[str]], targets: list[str]) -> float:
- gains = []
- for predicted, target in zip(predictions, targets):
- gains.append(
- 1 / math.log2(predicted.index(target) + 2) if target in predicted else 0.0
- )
- return sum(gains) / len(targets)
-
-
-def catalogue_coverage(predictions: list[list[str]], catalogue: set[str]) -> float:
- recommended = {item for prediction in predictions for item in prediction}
- return len(recommended & catalogue) / len(catalogue) if catalogue else 0.0
-
-
-def evaluate_ranking(
- predictions: list[list[str]], targets: list[str], catalogue: set[str]
-) -> dict[str, float]:
- if not targets or len(predictions) != len(targets):
- raise ValueError("Predictions and non-empty targets must have equal lengths")
- return {
- "recall_at_k": recall_at_k(predictions, targets),
- "ndcg_at_k": ndcg_at_k(predictions, targets),
- "mrr_at_k": mrr_at_k(predictions, targets),
- "catalogue_coverage": catalogue_coverage(predictions, catalogue),
- }
+"""Ranking metrics with explicit definitions and no framework dependency."""
+
+from __future__ import annotations
+
+import math
+
+
+def recall_at_k(predictions: list[list[str]], targets: list[str]) -> float:
+ return sum(target in predicted for predicted, target in zip(predictions, targets)) / len(
+ targets
+ )
+
+
+def mrr_at_k(predictions: list[list[str]], targets: list[str]) -> float:
+ reciprocal_ranks = []
+ for predicted, target in zip(predictions, targets):
+ reciprocal_ranks.append(1 / (predicted.index(target) + 1) if target in predicted else 0.0)
+ return sum(reciprocal_ranks) / len(targets)
+
+
+def ndcg_at_k(predictions: list[list[str]], targets: list[str]) -> float:
+ gains = []
+ for predicted, target in zip(predictions, targets):
+ gains.append(1 / math.log2(predicted.index(target) + 2) if target in predicted else 0.0)
+ return sum(gains) / len(targets)
+
+
+def catalogue_coverage(predictions: list[list[str]], catalogue: set[str]) -> float:
+ recommended = {item for prediction in predictions for item in prediction}
+ return len(recommended & catalogue) / len(catalogue) if catalogue else 0.0
+
+
+def evaluate_ranking(
+ predictions: list[list[str]], targets: list[str], catalogue: set[str]
+) -> dict[str, float]:
+ if not targets or len(predictions) != len(targets):
+ raise ValueError("Predictions and non-empty targets must have equal lengths")
+ return {
+ "recall_at_k": recall_at_k(predictions, targets),
+ "ndcg_at_k": ndcg_at_k(predictions, targets),
+ "mrr_at_k": mrr_at_k(predictions, targets),
+ "catalogue_coverage": catalogue_coverage(predictions, catalogue),
+ }
diff --git a/src/training/train.py b/src/training/train.py
index b724b7a..a3db9ef 100644
--- a/src/training/train.py
+++ b/src/training/train.py
@@ -1,189 +1,192 @@
-"""Train, evaluate, and package the production sequence recommender."""
-
-from __future__ import annotations
-
-import argparse
-import json
-import random
-from datetime import datetime, timezone
-from pathlib import Path
-
-import numpy as np
-import torch
-import torch.nn.functional as F
-
-from app.core.artifacts import ModelManifest, save_bundle
-from app.core.data_processor import SequenceProcessor
-from app.core.model import DeepSequenceModel
-from src.training.baselines import PopularityBaseline
-from src.training.data import (
- dataset_id,
- load_jsonl,
- next_item_examples,
- session_sequences,
- temporal_split,
-)
-from src.training.metrics import evaluate_ranking
-
-
-def seed_everything(seed: int) -> None:
- random.seed(seed)
- np.random.seed(seed)
- torch.manual_seed(seed)
-
-
-def train_model(
- train_sequences: list[list[str]],
- *,
- embedding_dim: int = 32,
- hidden_dim: int = 64,
- num_layers: int = 1,
- max_sequence_length: int = 50,
- epochs: int = 3,
- learning_rate: float = 1e-3,
- seed: int = 7,
-) -> tuple[SequenceProcessor, DeepSequenceModel]:
- seed_everything(seed)
- processor = SequenceProcessor(max_length=max_sequence_length).fit(train_sequences)
- if processor.vocab_size < 2:
- raise ValueError("Training requires at least two catalogue items")
- examples = next_item_examples(train_sequences)
- model = DeepSequenceModel(
- num_items=processor.vocab_size,
- embedding_dim=embedding_dim,
- hidden_dim=hidden_dim,
- num_layers=num_layers,
- )
- optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)
- model.train()
- for _ in range(epochs):
- for history, target in examples:
- target_index = processor.item_to_idx(target)
- if target_index == 0:
- continue
- optimizer.zero_grad()
- logits = model(processor.to_tensor(history))
- loss = F.cross_entropy(logits, torch.tensor([target_index]))
- loss.backward()
- optimizer.step()
- model.eval()
- return processor, model
-
-
-def model_predictions(
- model: DeepSequenceModel,
- processor: SequenceProcessor,
- examples: list[tuple[list[str], str]],
- top_k: int,
-) -> tuple[list[list[str]], list[str]]:
- predictions: list[list[str]] = []
- targets: list[str] = []
- for history, target in examples:
- if processor.item_to_idx(target) == 0:
- continue
- known_history = [item for item in history if processor.item_to_idx(item) != 0]
- if not known_history:
- continue
- eligible_top_k = min(
- top_k, processor.vocab_size - len(set(known_history))
- )
- if eligible_top_k < 1:
- continue
- indices = model.recommend(
- processor.to_tensor(known_history),
- top_k=eligible_top_k,
- exclude_ids=[processor.item_to_idx(item) for item in known_history],
- )
- predictions.append(
- [item for item in processor.decode_recommendations(indices) if item is not None]
- )
- targets.append(target)
- return predictions, targets
-
-
-def run_training(
- dataset_path: str | Path,
- output_directory: str | Path,
- *,
- epochs: int = 3,
- top_k: int = 10,
- seed: int = 7,
-) -> dict[str, object]:
- events = load_jsonl(dataset_path)
- train_events, validation_events, test_events = temporal_split(events)
- train_sequences = session_sequences(train_events)
- validation_examples = next_item_examples(session_sequences(validation_events))
- test_examples = next_item_examples(session_sequences(test_events))
- processor, model = train_model(train_sequences, epochs=epochs, seed=seed)
-
- baseline = PopularityBaseline().fit(train_sequences)
- catalogue = set(processor.export_vocabulary())
- eligible_validation = [
- (history, target)
- for history, target in validation_examples
- if processor.item_to_idx(target) != 0
- and any(processor.item_to_idx(item) != 0 for item in history)
- ]
- validation_predictions, validation_targets = model_predictions(
- model, processor, eligible_validation, top_k
- )
- if not validation_targets:
- raise ValueError("Validation split has no evaluable known-item examples")
- neural_metrics = evaluate_ranking(validation_predictions, validation_targets, catalogue)
- baseline_predictions = [
- baseline.recommend(history, top_k) for history, _target in eligible_validation
- ]
- baseline_metrics = evaluate_ranking(
- baseline_predictions, validation_targets, catalogue
- )
-
- test_predictions, test_targets = model_predictions(model, processor, test_examples, top_k)
- test_metrics = (
- evaluate_ranking(test_predictions, test_targets, catalogue) if test_targets else {}
- )
- version = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
- config = {
- "embedding_dim": 32,
- "hidden_dim": 64,
- "num_layers": 1,
- "dropout": 0.3,
- "padding_idx": 0,
- "max_sequence_length": 50,
- }
- manifest = ModelManifest(
- model_version=version,
- created_at=datetime.now(timezone.utc).isoformat(),
- architecture_config=config,
- metrics={f"validation_{key}": value for key, value in neural_metrics.items()},
- dataset_id=dataset_id(events),
- weights_sha256="pending",
- vocabulary_sha256="pending",
- )
- completed = save_bundle(output_directory, model, processor, manifest)
- report = {
- "model_version": version,
- "dataset_id": completed.dataset_id,
- "validation": neural_metrics,
- "popularity_baseline": baseline_metrics,
- "test": test_metrics,
- "promotion_eligible": neural_metrics["ndcg_at_k"] >= baseline_metrics["ndcg_at_k"],
- }
- (Path(output_directory) / "evaluation.json").write_text(
- json.dumps(report, indent=2), encoding="utf-8"
- )
- return report
-
-
-def main() -> None:
- parser = argparse.ArgumentParser(description=__doc__)
- parser.add_argument("--dataset", required=True)
- parser.add_argument("--output", default="models/candidate")
- parser.add_argument("--epochs", type=int, default=3)
- parser.add_argument("--top-k", type=int, default=10)
- parser.add_argument("--seed", type=int, default=7)
- args = parser.parse_args()
- print(json.dumps(run_training(args.dataset, args.output, epochs=args.epochs, top_k=args.top_k, seed=args.seed), indent=2))
-
-
-if __name__ == "__main__":
- main()
+"""Train, evaluate, and package the production sequence recommender."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import random
+from datetime import UTC, datetime
+from pathlib import Path
+
+import numpy as np
+import torch
+import torch.nn.functional as F
+
+from app.core.artifacts import ModelManifest, save_bundle
+from app.core.data_processor import SequenceProcessor
+from app.core.model import DeepSequenceModel
+from src.training.baselines import PopularityBaseline
+from src.training.data import (
+ dataset_id,
+ load_jsonl,
+ next_item_examples,
+ session_sequences,
+ temporal_split,
+)
+from src.training.metrics import evaluate_ranking
+
+
+def seed_everything(seed: int) -> None:
+ random.seed(seed)
+ np.random.seed(seed)
+ torch.manual_seed(seed)
+
+
+def train_model(
+ train_sequences: list[list[str]],
+ *,
+ embedding_dim: int = 32,
+ hidden_dim: int = 64,
+ num_layers: int = 1,
+ max_sequence_length: int = 50,
+ epochs: int = 3,
+ learning_rate: float = 1e-3,
+ seed: int = 7,
+) -> tuple[SequenceProcessor, DeepSequenceModel]:
+ seed_everything(seed)
+ processor = SequenceProcessor(max_length=max_sequence_length).fit(train_sequences)
+ if processor.vocab_size < 2:
+ raise ValueError("Training requires at least two catalogue items")
+ examples = next_item_examples(train_sequences)
+ model = DeepSequenceModel(
+ num_items=processor.vocab_size,
+ embedding_dim=embedding_dim,
+ hidden_dim=hidden_dim,
+ num_layers=num_layers,
+ )
+ optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)
+ model.train()
+ for _ in range(epochs):
+ for history, target in examples:
+ target_index = processor.item_to_idx(target)
+ if target_index == 0:
+ continue
+ optimizer.zero_grad()
+ logits = model(processor.to_tensor(history))
+ loss = F.cross_entropy(logits, torch.tensor([target_index]))
+ loss.backward()
+ optimizer.step()
+ model.eval()
+ return processor, model
+
+
+def model_predictions(
+ model: DeepSequenceModel,
+ processor: SequenceProcessor,
+ examples: list[tuple[list[str], str]],
+ top_k: int,
+) -> tuple[list[list[str]], list[str]]:
+ predictions: list[list[str]] = []
+ targets: list[str] = []
+ for history, target in examples:
+ if processor.item_to_idx(target) == 0:
+ continue
+ known_history = [item for item in history if processor.item_to_idx(item) != 0]
+ if not known_history:
+ continue
+ eligible_top_k = min(top_k, processor.vocab_size - len(set(known_history)))
+ if eligible_top_k < 1:
+ continue
+ indices = model.recommend(
+ processor.to_tensor(known_history),
+ top_k=eligible_top_k,
+ exclude_ids=[processor.item_to_idx(item) for item in known_history],
+ )
+ predictions.append(
+ [item for item in processor.decode_recommendations(indices) if item is not None]
+ )
+ targets.append(target)
+ return predictions, targets
+
+
+def run_training(
+ dataset_path: str | Path,
+ output_directory: str | Path,
+ *,
+ epochs: int = 3,
+ top_k: int = 10,
+ seed: int = 7,
+) -> dict[str, object]:
+ events = load_jsonl(dataset_path)
+ train_events, validation_events, test_events = temporal_split(events)
+ train_sequences = session_sequences(train_events)
+ validation_examples = next_item_examples(session_sequences(validation_events))
+ test_examples = next_item_examples(session_sequences(test_events))
+ processor, model = train_model(train_sequences, epochs=epochs, seed=seed)
+
+ baseline = PopularityBaseline().fit(train_sequences)
+ catalogue = set(processor.export_vocabulary())
+ eligible_validation = [
+ (history, target)
+ for history, target in validation_examples
+ if processor.item_to_idx(target) != 0
+ and any(processor.item_to_idx(item) != 0 for item in history)
+ ]
+ validation_predictions, validation_targets = model_predictions(
+ model, processor, eligible_validation, top_k
+ )
+ if not validation_targets:
+ raise ValueError("Validation split has no evaluable known-item examples")
+ neural_metrics = evaluate_ranking(validation_predictions, validation_targets, catalogue)
+ baseline_predictions = [
+ baseline.recommend(history, top_k) for history, _target in eligible_validation
+ ]
+ baseline_metrics = evaluate_ranking(baseline_predictions, validation_targets, catalogue)
+
+ test_predictions, test_targets = model_predictions(model, processor, test_examples, top_k)
+ test_metrics = (
+ evaluate_ranking(test_predictions, test_targets, catalogue) if test_targets else {}
+ )
+ version = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
+ config = {
+ "embedding_dim": 32,
+ "hidden_dim": 64,
+ "num_layers": 1,
+ "dropout": 0.3,
+ "padding_idx": 0,
+ "max_sequence_length": 50,
+ }
+ manifest = ModelManifest(
+ model_version=version,
+ created_at=datetime.now(UTC).isoformat(),
+ architecture_config=config,
+ metrics={f"validation_{key}": value for key, value in neural_metrics.items()},
+ dataset_id=dataset_id(events),
+ weights_sha256="pending",
+ vocabulary_sha256="pending",
+ )
+ completed = save_bundle(output_directory, model, processor, manifest)
+ report = {
+ "model_version": version,
+ "dataset_id": completed.dataset_id,
+ "validation": neural_metrics,
+ "popularity_baseline": baseline_metrics,
+ "test": test_metrics,
+ "promotion_eligible": neural_metrics["ndcg_at_k"] >= baseline_metrics["ndcg_at_k"],
+ }
+ (Path(output_directory) / "evaluation.json").write_text(
+ json.dumps(report, indent=2), encoding="utf-8"
+ )
+ return report
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--dataset", required=True)
+ parser.add_argument("--output", default="models/candidate")
+ parser.add_argument("--epochs", type=int, default=3)
+ parser.add_argument("--top-k", type=int, default=10)
+ parser.add_argument("--seed", type=int, default=7)
+ args = parser.parse_args()
+ print(
+ json.dumps(
+ run_training(
+ args.dataset, args.output, epochs=args.epochs, top_k=args.top_k, seed=args.seed
+ ),
+ indent=2,
+ )
+ )
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/test_artifacts.py b/tests/test_artifacts.py
index 808a36e..b85eeb1 100644
--- a/tests/test_artifacts.py
+++ b/tests/test_artifacts.py
@@ -1,59 +1,57 @@
-"""Model-bundle integrity and ranking-evaluation tests."""
-
-from datetime import datetime, timezone
-
-import pytest
-import onnx
-
-from app.core.artifacts import ModelManifest, load_bundle, save_bundle
-from app.core.data_processor import SequenceProcessor
-from app.core.model import DeepSequenceModel
-from src.training.metrics import evaluate_ranking
-from src.serving.onnx_exporter import export_bundle_to_onnx
-
-
-def test_model_bundle_roundtrip_and_checksum(tmp_path) -> None:
- processor = SequenceProcessor(max_length=5).fit([["a", "b", "c"]])
- model = DeepSequenceModel(
- num_items=processor.vocab_size,
- embedding_dim=8,
- hidden_dim=8,
- num_layers=1,
- )
- manifest = ModelManifest(
- model_version="test-v1",
- created_at=datetime.now(timezone.utc).isoformat(),
- architecture_config={
- "embedding_dim": 8,
- "hidden_dim": 8,
- "num_layers": 1,
- "dropout": 0.3,
- "padding_idx": 0,
- "max_sequence_length": 5,
- },
- dataset_id="dataset-test",
- weights_sha256="pending",
- vocabulary_sha256="pending",
- )
- save_bundle(tmp_path, model, processor, manifest)
-
- restored_processor, restored_model, restored_manifest = load_bundle(tmp_path)
- assert restored_manifest.model_version == "test-v1"
- assert restored_processor.export_vocabulary() == processor.export_vocabulary()
- assert restored_model.recommend(restored_processor.to_tensor(["a"]), top_k=2)
-
- onnx_path = export_bundle_to_onnx(tmp_path, tmp_path / "model.onnx")
- onnx.checker.check_model(onnx.load(onnx_path))
-
- (tmp_path / "vocabulary.json").write_text("{}", encoding="utf-8")
- with pytest.raises(ValueError, match="checksum"):
- load_bundle(tmp_path)
-
-
-def test_ranking_metrics_have_known_values() -> None:
- metrics = evaluate_ranking(
- [["a", "b"], ["c", "d"]], ["a", "d"], {"a", "b", "c", "d"}
- )
- assert metrics["recall_at_k"] == 1.0
- assert metrics["mrr_at_k"] == 0.75
- assert metrics["catalogue_coverage"] == 1.0
+"""Model-bundle integrity and ranking-evaluation tests."""
+
+from datetime import UTC, datetime
+
+import onnx
+import pytest
+
+from app.core.artifacts import ModelManifest, load_bundle, save_bundle
+from app.core.data_processor import SequenceProcessor
+from app.core.model import DeepSequenceModel
+from src.serving.onnx_exporter import export_bundle_to_onnx
+from src.training.metrics import evaluate_ranking
+
+
+def test_model_bundle_roundtrip_and_checksum(tmp_path) -> None:
+ processor = SequenceProcessor(max_length=5).fit([["a", "b", "c"]])
+ model = DeepSequenceModel(
+ num_items=processor.vocab_size,
+ embedding_dim=8,
+ hidden_dim=8,
+ num_layers=1,
+ )
+ manifest = ModelManifest(
+ model_version="test-v1",
+ created_at=datetime.now(UTC).isoformat(),
+ architecture_config={
+ "embedding_dim": 8,
+ "hidden_dim": 8,
+ "num_layers": 1,
+ "dropout": 0.3,
+ "padding_idx": 0,
+ "max_sequence_length": 5,
+ },
+ dataset_id="dataset-test",
+ weights_sha256="pending",
+ vocabulary_sha256="pending",
+ )
+ save_bundle(tmp_path, model, processor, manifest)
+
+ restored_processor, restored_model, restored_manifest = load_bundle(tmp_path)
+ assert restored_manifest.model_version == "test-v1"
+ assert restored_processor.export_vocabulary() == processor.export_vocabulary()
+ assert restored_model.recommend(restored_processor.to_tensor(["a"]), top_k=2)
+
+ onnx_path = export_bundle_to_onnx(tmp_path, tmp_path / "model.onnx")
+ onnx.checker.check_model(onnx.load(onnx_path))
+
+ (tmp_path / "vocabulary.json").write_text("{}", encoding="utf-8")
+ with pytest.raises(ValueError, match="checksum"):
+ load_bundle(tmp_path)
+
+
+def test_ranking_metrics_have_known_values() -> None:
+ metrics = evaluate_ranking([["a", "b"], ["c", "d"]], ["a", "d"], {"a", "b", "c", "d"})
+ assert metrics["recall_at_k"] == 1.0
+ assert metrics["mrr_at_k"] == 0.75
+ assert metrics["catalogue_coverage"] == 1.0
diff --git a/tests/test_schema_contracts.py b/tests/test_schema_contracts.py
index b16be78..6dbd948 100644
--- a/tests/test_schema_contracts.py
+++ b/tests/test_schema_contracts.py
@@ -1,42 +1,47 @@
-"""Versioned event and temporal-split contract tests."""
-
-from datetime import datetime, timedelta, timezone
-
-import pytest
-from pydantic import ValidationError
-
-from src.training.data import InteractionEvent, next_item_examples, session_sequences, temporal_split
-
-
-def event(index: int, session: str = "s1") -> InteractionEvent:
- return InteractionEvent(
- event_id=f"event-{index}",
- user_id="user-1",
- session_id=session,
- item_id=f"item-{index % 3}",
- event_type="click",
- timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(minutes=index),
- )
-
-
-def test_event_schema_rejects_unknown_event_type() -> None:
- with pytest.raises(ValidationError):
- InteractionEvent(
- event_id="e",
- user_id="u",
- session_id="s",
- item_id="i",
- event_type="share",
- timestamp=datetime.now(timezone.utc),
- )
-
-
-def test_temporal_split_preserves_order() -> None:
- train, validation, test = temporal_split([event(index) for index in range(20)])
- assert max(item.timestamp for item in train) < min(item.timestamp for item in validation)
- assert max(item.timestamp for item in validation) < min(item.timestamp for item in test)
-
-
-def test_next_item_examples_are_causal() -> None:
- examples = next_item_examples(session_sequences([event(index) for index in range(4)]))
- assert examples[0] == (["item-0"], "item-1")
+"""Versioned event and temporal-split contract tests."""
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from pydantic import ValidationError
+
+from src.training.data import (
+ InteractionEvent,
+ next_item_examples,
+ session_sequences,
+ temporal_split,
+)
+
+
+def event(index: int, session: str = "s1") -> InteractionEvent:
+ return InteractionEvent(
+ event_id=f"event-{index}",
+ user_id="user-1",
+ session_id=session,
+ item_id=f"item-{index % 3}",
+ event_type="click",
+ timestamp=datetime(2026, 1, 1, tzinfo=UTC) + timedelta(minutes=index),
+ )
+
+
+def test_event_schema_rejects_unknown_event_type() -> None:
+ with pytest.raises(ValidationError):
+ InteractionEvent(
+ event_id="e",
+ user_id="u",
+ session_id="s",
+ item_id="i",
+ event_type="share",
+ timestamp=datetime.now(UTC),
+ )
+
+
+def test_temporal_split_preserves_order() -> None:
+ train, validation, test = temporal_split([event(index) for index in range(20)])
+ assert max(item.timestamp for item in train) < min(item.timestamp for item in validation)
+ assert max(item.timestamp for item in validation) < min(item.timestamp for item in test)
+
+
+def test_next_item_examples_are_causal() -> None:
+ examples = next_item_examples(session_sequences([event(index) for index in range(4)]))
+ assert examples[0] == (["item-0"], "item-1")