recommendation-engine

レコメンデーションエンジン

パーソナライズされたコンテンツや商品の提案を行うレコメンデーションシステムを構築します。

レコメンデーションのアプローチ

アプローチ	仕組み	メリット	デメリット
協調型	ユーザーとアイテムのインタラクション	隠れたパターンを発見できます	コールドスタート
コンテンツベース	アイテムの特徴	新しいアイテムにも対応できます	発見の幅が限られます
ハイブリッド	両方を組み合わせます	両方の長所を活かせます	複雑です

協調フィルタリング

import numpy as np
from scipy.sparse import csr_matrix
from sklearn.metrics.pairwise import cosine_similarity

class CollaborativeFilter:
    def __init__(self):
        self.user_similarity = None
        self.item_similarity = None

    def fit(self, user_item_matrix):
        # User-based similarity
        self.user_similarity = cosine_similarity(user_item_matrix)
        # Item-based similarity
        self.item_similarity = cosine_similarity(user_item_matrix.T)

    def recommend_for_user(self, user_id, n=10):
        scores = self.user_similarity[user_id].dot(self.user_item_matrix)
        # Exclude already interacted items
        already_interacted = self.user_item_matrix[user_id].nonzero()[0]
        scores[already_interacted] = -np.inf
        return np.argsort(scores)[-n:][::-1]

行列分解（SVD）

from sklearn.decomposition import TruncatedSVD

class MatrixFactorization:
    def __init__(self, n_factors=50):
        self.svd = TruncatedSVD(n_components=n_factors)

    def fit(self, user_item_matrix):
        self.user_factors = self.svd.fit_transform(user_item_matrix)
        self.item_factors = self.svd.components_.T

    def predict(self, user_id, item_id):
        return np.dot(self.user_factors[user_id], self.item_factors[item_id])

ハイブリッドレコメンダー

class HybridRecommender:
    def __init__(self, collab_weight=0.7, content_weight=0.3):
        self.collab = CollaborativeFilter()
        self.content = ContentBasedFilter()
        self.weights = (collab_weight, content_weight)

    def recommend(self, user_id, n=10):
        collab_scores = self.collab.score(user_id)
        content_scores = self.content.score(user_id)
        combined = self.weights[0] * collab_scores + self.weights[1] * content_scores
        return np.argsort(combined)[-n:][::-1]

評価指標

• Precision@K、Recall@K
• NDCG（ランキング品質）
• Coverage（カタログの多様性）
• A/B テストのコンバージョン率

コールドスタートの解決策

• 新規ユーザー：人気アイテム、オンボーディング時の好み、人口統計ベースの推薦
• 新規アイテム：コンテンツベースのブートストラップ、能動学習
• 探索戦略：ε-greedy、Thompson sampling バンディット

クイックスタート：5 ステップでレコメンダーを構築する

from scipy.sparse import csr_matrix
import numpy as np

# 1. Prepare user-item interaction matrix
# rows = users, cols = items, values = ratings/interactions
ratings_data = [(0, 5, 5), (0, 10, 4), (1, 5, 3), ...]  # (user, item, rating)
n_users, n_items = 1000, 5000

row_idx = [r[0] for r in ratings_data]
col_idx = [r[1] for r in ratings_data]
ratings = [r[2] for r in ratings_data]
user_item_matrix = csr_matrix((ratings, (row_idx, col_idx)), shape=(n_users, n_items))

# 2. Choose and train model
from recommendation_engine import ItemBasedCollaborativeFilter  # See references

model = ItemBasedCollaborativeFilter(similarity_metric='cosine', k_neighbors=20)
model.fit(user_item_matrix)

# 3. Generate recommendations
recommendations = model.recommend(user_id=42, n=10)
print(recommendations)  # [(item_id, score), ...]

# 4. Evaluate on test set
from evaluation_metrics import precision_at_k, recall_at_k

test_items = {42: {10, 25, 30}}  # True relevant items for user 42
rec_items = [item for item, score in recommendations]

precision = precision_at_k(rec_items, test_items[42], k=10)
recall = recall_at_k(rec_items, test_items[42], k=10)
print(f"Precision@10: {precision:.3f}, Recall@10: {recall:.3f}")

# 5. Handle cold start
from cold_start import PopularityRecommender

popularity_model = PopularityRecommender()
popularity_model.fit(interactions_with_timestamps)
new_user_recs = popularity_model.recommend(n=10)

既知の問題の予防

1. 人気度バイアス

問題：人気アイテムばかりを推薦し、ロングテールを無視します。多様性とセレンディピティが低下します。

解決策：人気度とパーソナライズのバランスを取り、多様性のために再ランキングを適用します：

def diversify_recommendations(
    recommendations: List[Tuple[int, float]],
    item_features: np.ndarray,
    diversity_weight: float = 0.3
) -> List[Tuple[int, float]]:
    """Re-rank to increase diversity while maintaining relevance."""
    from sklearn.metrics.pairwise import cosine_distances

    selected = []
    candidates = recommendations.copy()

    while len(selected) < len(recommendations) and candidates:
        if not selected:
            # First item: highest score
            selected.append(candidates.pop(0))
            continue

        # Compute diversity scores
        selected_features = item_features[[item for item, _ in selected]]
        diversity_scores = []

        for item, relevance in candidates:
            item_feature = item_features[item].reshape(1, -1)
            # Average distance to already selected items
            avg_distance = cosine_distances(item_feature, selected_features).mean()
            # Combined score: relevance + diversity
            combined = (1 - diversity_weight) * relevance + diversity_weight * avg_distance
            diversity_scores.append((item, relevance, combined))

        # Select item with best combined score
        best = max(diversity_scores, key=lambda x: x[2])
        selected.append((best[0], best[1]))
        candidates = [(i, s) for i, s, _ in diversity_scores if i != best[0]]

    return selected

2. データのスパース性（行列の 99% 超が空）

問題：ほとんどのユーザーがアイテムの 1% 未満しか評価していない場合、協調フィルタリングは機能しません。

解決策：メモリベースの CF ではなく、行列分解（SVD、ALS）を使用します：

# ❌ Bad: User-based CF on sparse data (fails to find similar users)
user_cf = UserBasedCollaborativeFilter()
user_cf.fit(sparse_matrix)  # Most users have <10 ratings

# ✅ Good: Matrix factorization handles sparsity
from sklearn.decomposition import TruncatedSVD

svd = TruncatedSVD(n_components=50)
user_factors = svd.fit_transform(sparse_matrix)
item_factors = svd.components_.T

# Predict rating: user_factors[u] @ item_factors[i]

3. フォールバックのないコールドスタート

問題：新規ユーザーや新規アイテムに対して、レコメンダーがクラッシュしたり、結果が空になったりします。

解決策：必ずフォールバックチェーンを実装します：

def recommend_with_fallback(user_id, n=10):
    """Graceful degradation through fallback chain."""
    try:
        # Try personalized recommendations
        if has_sufficient_history(user_id, min_interactions=5):
            return collaborative_filter.recommend(user_id, n)
    except Exception as e:
        logger.warning(f"CF failed for user {user_id}: {e}")

    # Fallback 1: Demographic-based
    if user_demographics_available(user_id):
        return demographic_recommender.recommend(user_id, n)

    # Fallback 2: Popularity
    return popularity_recommender.recommend(n)

4. すでにインタラクションしたアイテムを除外しない

問題：ユーザーがすでに購入または閲覧したアイテムを推薦すると、推薦枠が無駄になります。

解決策：必ずインタラクション済みのアイテムを除外します：

# ✅ Correct: Exclude interacted items
user_items = user_item_matrix[user_id].nonzero()[1]
scores[user_items] = -np.inf  # Ensure they don't appear in top-K
recommendations = np.argsort(scores)[-n:][::-1]

# ❌ Wrong: Forgetting to filter
recommendations = np.argsort(scores)[-n:][::-1]  # May include already purchased!

5. 暗黙的フィードバックの信頼度を無視する

問題：すべてのクリックや閲覧を同等に扱います。1 回の閲覧と 100 回の閲覧は同じではありません。

解決策：インタラクションの強度（閲覧回数、視聴時間など）に応じて重み付けします：

# For implicit feedback, use confidence weighting
confidence_matrix = 1 + alpha * np.log(1 + interaction_counts)

# In ALS: C_ui * (P_ui - X_ui)²
# Higher confidence for items with more interactions

6. ランキング品質を評価しない（精度のみを使用）

問題：予測精度（RMSE）が高くても、優れた上位 K 件の推薦になるとは限りません。

解決策：ランキング指標（NDCG、MAP@K）を使用します：

# ❌ Bad: Only RMSE
from sklearn.metrics import mean_squared_error
rmse = np.sqrt(mean_squared_error(y_true, y_pred))

# ✅ Good: Ranking metrics for top-K evaluation
from evaluation_metrics import ndcg_at_k, mean_average_precision_at_k

# NDCG rewards putting highly relevant items first
ndcg = ndcg_at_k(recommendations, relevance_scores, k=10)

# MAP@K considers precision at each relevant item position
map_score = mean_average_precision_at_k(all_recommendations, ground_truth, k=10)

7. フィルターバブル（探索不足）

問題：常に類似したアイテムを推薦すると、新しい発見が制限され、時間の経過とともにユーザーのエンゲージメントが低下します。

解決策：探索と活用の戦略を実装します：

class ExploreExploitRecommender:
    def __init__(self, base_model, epsilon=0.1):
        self.base_model = base_model
        self.epsilon = epsilon  # 10% exploration

    def recommend(self, user_id, n=10):
        # Exploit: Use trained model for most recommendations
        n_exploit = int(n * (1 - self.epsilon))
        exploitative_recs = self.base_model.recommend(user_id, n=n_exploit)

        # Explore: Add random diverse items
        n_explore = n - n_exploit
        explored_items = sample_diverse_items(n_explore)

        return exploitative_recs + explored_items

参照資料を読み込むタイミング

詳しい実装が必要な場合は、参照ファイルを読み込みます：

• 協調フィルタリング：ユーザーベースおよびアイテムベースの CF の完全な実装、類似度指標（コサイン、Pearson、Jaccard）、スケーラビリティ最適化（スパース行列、近似最近傍法）、エッジケース（コールドスタート、スパース性）への対応については、references/collaborative-filtering-deep-dive.md を読み込みます

• 行列分解：ハイパーパラメーター調整、暗黙的フィードバックへの対応、高度な手法（BPR、WARP）を含む SVD、ALS、NMF の実装については、references/matrix-factorization-methods.md を読み込みます

• 評価指標：Precision@K、Recall@K、NDCG、カバレッジ、多様性指標、交差検証戦略、統計的有意性検定（対応のある t 検定、ブートストラップ信頼区間）については、references/evaluation-metrics-implementation.md を読み込みます

• コールドスタートの解決策：新規ユーザー／アイテム向けの戦略（人気度ベース、オンボーディング、人口統計、コンテンツベースのブートストラップ、能動学習）、探索と活用のアプローチ（ε-greedy、Thompson sampling）、ハイブリッドなフォールバックチェーンについては、references/cold-start-strategies.md を読み込みます