Catalog Coverage in Machine Learning

import numpy as np
from typing import Dict, List, Set, Tuple


def catalog_coverage(
    recommendations: Dict[str, List[str]],
    catalog: Set[str]
) -> float:
    """Compute catalog coverage: fraction of catalog items recommended.

    Args:
        recommendations: {user_id: [item_id, ...]} for all users
        catalog: set of all item IDs in the catalog

    Returns:
        Coverage score between 0 and 1
    """
    recommended_items = set()
    for user_id, items in recommendations.items():
        recommended_items.update(items)

    # Only count items that are actually in the catalog
    covered = recommended_items & catalog
    return len(covered) / len(catalog) if catalog else 0.0


def gini_index(item_frequencies: np.ndarray) -> float:
    """Compute Gini index of item recommendation frequencies.

    Args:
        item_frequencies: array of recommendation counts per item
                          (including zeros for unrecommended items)

    Returns:
        Gini index between 0 (perfect equality) and 1 (max inequality)
    """
    sorted_freq = np.sort(item_frequencies)
    n = len(sorted_freq)
    if n == 0 or sorted_freq.sum() == 0:
        return 0.0
    index = np.arange(1, n + 1)
    return (2 * np.sum(index * sorted_freq) - (n + 1) * np.sum(sorted_freq)) / (
        n * np.sum(sorted_freq)
    )


def shannon_entropy(item_frequencies: np.ndarray) -> Tuple[float, float]:
    """Compute Shannon entropy of item exposure distribution.

    Args:
        item_frequencies: array of recommendation counts per item

    Returns:
        (entropy, normalized_entropy) where normalized is in [0, 1]
    """
    total = item_frequencies.sum()
    if total == 0:
        return 0.0, 0.0
    probs = item_frequencies[item_frequencies > 0] / total
    entropy = -np.sum(probs * np.log2(probs))
    max_entropy = np.log2(len(item_frequencies)) if len(item_frequencies) > 1 else 1.0
    return entropy, entropy / max_entropy


# ── Example Usage ───────────────────────────────────────────────────
catalog_items = {f"item_{i}" for i in range(1000)}  # 1000-item catalog

# Simulated recommendations: 500 users, top-10 each
np.random.seed(42)
popular_items = [f"item_{i}" for i in range(50)]   # 50 popular items
long_tail = [f"item_{i}" for i in range(50, 1000)]  # 950 long-tail items

recs = {}
for u in range(500):
    # 80% chance of picking popular, 20% long-tail (realistic skew)
    user_recs = []
    for _ in range(10):
        if np.random.random() < 0.8:
            user_recs.append(np.random.choice(popular_items))
        else:
            user_recs.append(np.random.choice(long_tail))
    recs[f"user_{u}"] = user_recs

coverage = catalog_coverage(recs, catalog_items)
print(f"Catalog Coverage: {coverage:.2%}")
# Output: Catalog Coverage: ~58.60%

# Compute item frequencies for Gini
from collections import Counter
all_recs = [item for items in recs.values() for item in items]
freq_counter = Counter(all_recs)
freqs = np.array([freq_counter.get(f"item_{i}", 0) for i in range(1000)])

gini = gini_index(freqs)
print(f"Gini Index: {gini:.4f}")
# Output: Gini Index: ~0.8753 (highly unequal)

entropy, norm_entropy = shannon_entropy(freqs)
print(f"Shannon Entropy: {entropy:.2f} bits (normalized: {norm_entropy:.4f})")

# pip install recmetrics
import recmetrics
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# Simulated recommendation data
np.random.seed(42)

# Full catalog of 5000 items
catalog = list(range(5000))

# Generate recommendations for 1000 users (top-10 lists)
# Skewed toward popular items (items 0-99 are popular)
recommendations = []
for user_id in range(1000):
    user_recs = []
    for _ in range(10):
        if np.random.random() < 0.7:
            user_recs.append(np.random.randint(0, 100))       # popular
        else:
            user_recs.append(np.random.randint(100, 5000))    # long-tail
    recommendations.append(user_recs)

# Compute prediction coverage using recmetrics
# prediction_coverage: % of catalog that appears in recommendations
coverage = recmetrics.prediction_coverage(
    predicted=recommendations,
    catalog=catalog
)
print(f"Prediction Coverage: {coverage:.2f}%")
# Output: ~48.2%

# Compute catalog coverage (same as prediction_coverage in this lib)
cat_cov = recmetrics.catalog_coverage(
    predicted=recommendations,
    catalog=catalog,
    k=10
)
print(f"Catalog Coverage @10: {cat_cov:.2f}%")

# Visualize the long-tail distribution
all_recommended = [item for user_recs in recommendations for item in user_recs]
item_counts = pd.Series(all_recommended).value_counts().sort_values(ascending=False)

plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.plot(range(len(item_counts)), item_counts.values)
plt.xlabel("Item Rank (by frequency)")
plt.ylabel("Recommendation Count")
plt.title("Long-Tail Distribution of Recommendations")
plt.yscale("log")

plt.subplot(1, 2, 2)
recmetrics.long_tail_plot(
    test=pd.DataFrame({"item": all_recommended}),
    column="item",
    percentage=0.33,
    title="Long Tail of Recommendations"
)
plt.tight_layout()
plt.savefig("coverage_analysis.png", dpi=150)
print("Saved coverage_analysis.png")

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from datetime import datetime, timedelta

spark = SparkSession.builder \
    .appName("CoverageMonitoring") \
    .getOrCreate()

# ── Load recommendation logs ──────────────────────────────────────
# Schema: user_id, item_id, position, timestamp, model_version
rec_logs = spark.read.parquet("s3://data/recommendation_logs/")

# Filter to last 7 days
window_start = datetime.now() - timedelta(days=7)
recent_recs = rec_logs.filter(
    F.col("timestamp") >= F.lit(window_start)
)

# ── Load active catalog ───────────────────────────────────────────
catalog = spark.read.parquet("s3://data/item_catalog/") \
    .filter(F.col("status") == "active")

total_items = catalog.count()
print(f"Active catalog size: {total_items:,}")

# ── Compute overall catalog coverage ─────────────────────────────
recommended_items = recent_recs.select("item_id").distinct()
covered_items = recommended_items.join(
    catalog.select("item_id"),
    on="item_id",
    how="inner"
)
coverage = covered_items.count() / total_items
print(f"Catalog Coverage (7-day): {coverage:.2%}")

# ── Per-category coverage ─────────────────────────────────────────
category_coverage = catalog.groupBy("category").agg(
    F.countDistinct("item_id").alias("total_items")
).join(
    covered_items.join(
        catalog.select("item_id", "category"),
        on="item_id"
    ).groupBy("category").agg(
        F.countDistinct("item_id").alias("covered_items")
    ),
    on="category",
    how="left"
).fillna(0, subset=["covered_items"]) \
 .withColumn("coverage", F.col("covered_items") / F.col("total_items")) \
 .orderBy("coverage")

print("\nPer-Category Coverage (ascending):")
category_coverage.show(20, truncate=False)

# ── Gini index via Spark ──────────────────────────────────────────
item_freq = recent_recs.groupBy("item_id").agg(
    F.count("*").alias("freq")
)

# Include zero-frequency items from catalog
all_item_freq = catalog.select("item_id").join(
    item_freq, on="item_id", how="left"
).fillna(0, subset=["freq"])

# Compute Gini using window function
windowed = all_item_freq.withColumn(
    "rank", F.row_number().over(Window.orderBy("freq"))
)
n = total_items
total_freq = all_item_freq.agg(F.sum("freq")).collect()[0][0]

gini_num = windowed.agg(
    F.sum(F.col("rank") * F.col("freq") * 2 - (n + 1) * F.col("freq"))
).collect()[0][0]

gini = gini_num / (n * total_freq) if total_freq > 0 else 0
print(f"\nGini Index: {gini:.4f}")

# ── Write metrics for monitoring ──────────────────────────────────
metrics = {
    "date": datetime.now().isoformat(),
    "window_days": 7,
    "catalog_size": total_items,
    "covered_items": covered_items.count(),
    "coverage": coverage,
    "gini_index": gini
}

metrics_df = spark.createDataFrame([metrics])
metrics_df.write.mode("append").parquet("s3://data/coverage_metrics/")
print("Metrics written to S3")

import numpy as np
from collections import Counter
from typing import List, Dict, Tuple


class CoverageAwareReranker:
    """Post-processing re-ranker that boosts long-tail items to improve
    catalog coverage while maintaining acceptable accuracy.

    Uses a simple interpolation between relevance score and item rarity.
    """

    def __init__(
        self,
        item_frequencies: Dict[str, int],
        catalog_size: int,
        lambda_coverage: float = 0.3,  # Weight for coverage boost
    ):
        """
        Args:
            item_frequencies: {item_id: count} of past recommendations
            catalog_size: total number of items in catalog
            lambda_coverage: interpolation weight (0 = pure accuracy,
                             1 = pure coverage optimization)
        """
        self.item_freq = item_frequencies
        self.catalog_size = catalog_size
        self.lam = lambda_coverage

        # Compute max frequency for normalization
        self.max_freq = max(item_frequencies.values()) if item_frequencies else 1

    def rarity_score(self, item_id: str) -> float:
        """Score inversely proportional to recommendation frequency.
        Never-recommended items get the highest rarity score."""
        freq = self.item_freq.get(item_id, 0)
        return 1.0 - (freq / self.max_freq)

    def rerank(
        self,
        candidates: List[Tuple[str, float]],
        top_k: int = 10
    ) -> List[Tuple[str, float]]:
        """Re-rank candidates by interpolating relevance and rarity.

        Args:
            candidates: [(item_id, relevance_score), ...]
            top_k: number of items to return

        Returns:
            Re-ranked list of (item_id, combined_score) tuples
        """
        reranked = []
        for item_id, relevance in candidates:
            rarity = self.rarity_score(item_id)
            combined = (1 - self.lam) * relevance + self.lam * rarity
            reranked.append((item_id, combined))

        reranked.sort(key=lambda x: x[1], reverse=True)
        return reranked[:top_k]


# ── Example Usage ───────────────────────────────────────────────────
# Historical recommendation frequencies (popular items have high counts)
hist_freq = {f"item_{i}": 1000 - i * 10 for i in range(100)}

reranker = CoverageAwareReranker(
    item_frequencies=hist_freq,
    catalog_size=10000,
    lambda_coverage=0.3  # 30% weight on coverage
)

# Candidates from the base model (sorted by relevance)
candidates = [
    ("item_0", 0.95),   # Very popular, high relevance
    ("item_1", 0.90),   # Popular, high relevance
    ("item_50", 0.85),  # Medium popularity
    ("item_99", 0.80),  # Less popular
    ("item_500", 0.75), # Long-tail (not in history)
    ("item_999", 0.70), # Long-tail
]

original = [(item, score) for item, score in candidates]
reranked = reranker.rerank(candidates, top_k=5)

print("Original ranking:")
for item, score in original[:5]:
    print(f"  {item}: relevance={score:.2f}")

print("\nRe-ranked (lambda=0.3):")
for item, score in reranked:
    print(f"  {item}: combined={score:.2f}, rarity={reranker.rarity_score(item):.2f}")

# Coverage monitoring configuration (YAML)
coverage_monitoring:
  enabled: true
  window_days: 7
  
  # Thresholds for alerting
  thresholds:
    overall_coverage_min: 0.15        # Alert if <15% catalog covered
    category_coverage_min: 0.05       # Alert if any category <5%
    gini_index_max: 0.95              # Alert if Gini >0.95
    
  # Segmentation dimensions
  segments:
    - category
    - price_tier
    - seller_tier
    - item_age_bucket   # new, medium, old
  
  # Re-ranking configuration
  reranking:
    enabled: true
    lambda_coverage: 0.25             # Coverage boost weight
    min_long_tail_fraction: 0.2       # At least 20% of recs from long-tail
    long_tail_threshold_percentile: 80  # Items below 80th percentile popularity
  
  # Reporting
  schedule: "0 6 * * *"              # Daily at 6 AM
  output_path: "s3://metrics/coverage/"
  dashboard_url: "https://grafana.internal/d/rec-coverage"