협업 필터링

import torch
import torch.nn as nn
import numpy as np
from surprise import Dataset, Reader, SVD, KNNBasic
from surprise.model_selection import cross_validate
import pandas as pd

# 1. Surprise를 사용한 전통적 협업 필터링
def traditional_collaborative_filtering():
    """SVD와 KNN 기반 협업 필터링"""
    # 샘플 데이터 생성
    ratings_dict = {
        'user_id': [1, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4],
        'item_id': [1, 2, 3, 1, 2, 4, 2, 3, 4, 1, 3, 4],
        'rating': [5, 3, 4, 4, 2, 5, 3, 4, 4, 5, 2, 3]
    }
    df = pd.DataFrame(ratings_dict)

    reader = Reader(rating_scale=(1, 5))
    data = Dataset.load_from_df(df[['user_id', 'item_id', 'rating']], reader)

    # SVD (Matrix Factorization)
    svd = SVD(n_factors=50, n_epochs=20, random_state=42)
    cv_results = cross_validate(svd, data, measures=['RMSE', 'MAE'], cv=3, verbose=True)
    print(f"SVD RMSE: {np.mean(cv_results['test_rmse']):.4f}")

    # User-based KNN
    user_knn = KNNBasic(k=20, sim_options={'name': 'cosine', 'user_based': True})
    cross_validate(user_knn, data, measures=['RMSE'], cv=3, verbose=True)

    # Item-based KNN
    item_knn = KNNBasic(k=20, sim_options={'name': 'cosine', 'user_based': False})
    cross_validate(item_knn, data, measures=['RMSE'], cv=3, verbose=True)

    return svd

# 2. PyTorch로 Neural Collaborative Filtering (NCF) 구현
class NeuralCollaborativeFiltering(nn.Module):
    """NCF: GMF + MLP 결합 모델"""
    def __init__(
        self,
        num_users: int,
        num_items: int,
        mf_dim: int = 32,
        mlp_dims: list = [64, 32, 16]
    ):
        super().__init__()
        self.num_users = num_users
        self.num_items = num_items

        # GMF (Generalized Matrix Factorization) 임베딩
        self.user_embedding_gmf = nn.Embedding(num_users, mf_dim)
        self.item_embedding_gmf = nn.Embedding(num_items, mf_dim)

        # MLP 임베딩 (더 큰 차원)
        self.user_embedding_mlp = nn.Embedding(num_users, mlp_dims[0] // 2)
        self.item_embedding_mlp = nn.Embedding(num_items, mlp_dims[0] // 2)

        # MLP 레이어
        mlp_layers = []
        for i in range(len(mlp_dims) - 1):
            mlp_layers.append(nn.Linear(mlp_dims[i], mlp_dims[i+1]))
            mlp_layers.append(nn.ReLU())
            mlp_layers.append(nn.Dropout(0.2))
        self.mlp = nn.Sequential(*mlp_layers)

        # 최종 예측 레이어
        self.prediction = nn.Linear(mf_dim + mlp_dims[-1], 1)

        self._init_weights()

    def _init_weights(self):
        for module in self.modules():
            if isinstance(module, nn.Embedding):
                nn.init.normal_(module.weight, std=0.01)
            elif isinstance(module, nn.Linear):
                nn.init.xavier_uniform_(module.weight)

    def forward(self, user_ids: torch.Tensor, item_ids: torch.Tensor):
        # GMF 경로
        user_gmf = self.user_embedding_gmf(user_ids)
        item_gmf = self.item_embedding_gmf(item_ids)
        gmf_output = user_gmf * item_gmf  # Element-wise product

        # MLP 경로
        user_mlp = self.user_embedding_mlp(user_ids)
        item_mlp = self.item_embedding_mlp(item_ids)
        mlp_input = torch.cat([user_mlp, item_mlp], dim=-1)
        mlp_output = self.mlp(mlp_input)

        # 결합
        concat = torch.cat([gmf_output, mlp_output], dim=-1)
        prediction = self.prediction(concat)

        return prediction.squeeze()

# 3. Two-Tower 추천 모델 (대규모 서비스용)
class TwoTowerModel(nn.Module):
    """사용자/아이템 타워 분리 모델 - 효율적인 대규모 추천"""
    def __init__(
        self,
        num_users: int,
        num_items: int,
        embedding_dim: int = 64
    ):
        super().__init__()

        # User Tower
        self.user_embedding = nn.Embedding(num_users, embedding_dim)
        self.user_tower = nn.Sequential(
            nn.Linear(embedding_dim, 128),
            nn.ReLU(),
            nn.Linear(128, embedding_dim),
            nn.LayerNorm(embedding_dim)
        )

        # Item Tower
        self.item_embedding = nn.Embedding(num_items, embedding_dim)
        self.item_tower = nn.Sequential(
            nn.Linear(embedding_dim, 128),
            nn.ReLU(),
            nn.Linear(embedding_dim, embedding_dim),
            nn.LayerNorm(embedding_dim)
        )

    def get_user_embedding(self, user_ids: torch.Tensor):
        """사용자 임베딩 - 캐싱 가능"""
        x = self.user_embedding(user_ids)
        return self.user_tower(x)

    def get_item_embedding(self, item_ids: torch.Tensor):
        """아이템 임베딩 - 사전 계산 가능"""
        x = self.item_embedding(item_ids)
        return self.item_tower(x)

    def forward(self, user_ids: torch.Tensor, item_ids: torch.Tensor):
        user_emb = self.get_user_embedding(user_ids)
        item_emb = self.get_item_embedding(item_ids)
        # Dot product similarity
        return torch.sum(user_emb * item_emb, dim=-1)

# 4. 학습 및 추천
def train_ncf():
    """NCF 모델 학습 예시"""
    # 더미 데이터
    num_users, num_items = 1000, 5000
    train_users = torch.randint(0, num_users, (10000,))
    train_items = torch.randint(0, num_items, (10000,))
    train_ratings = torch.rand(10000) * 4 + 1  # 1-5 점수

    model = NeuralCollaborativeFiltering(num_users, num_items)
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    criterion = nn.MSELoss()

    # 학습
    model.train()
    for epoch in range(10):
        optimizer.zero_grad()
        predictions = model(train_users, train_items)
        loss = criterion(predictions, train_ratings)
        loss.backward()
        optimizer.step()

        if epoch % 2 == 0:
            print(f"Epoch {epoch}, Loss: {loss.item():.4f}")

    # 추천 생성
    model.eval()
    user_id = 0
    all_items = torch.arange(num_items)
    user_ids = torch.full((num_items,), user_id)

    with torch.no_grad():
        scores = model(user_ids, all_items)
        top_k_items = torch.argsort(scores, descending=True)[:10]

    print(f"\nUser {user_id}의 Top 10 추천: {top_k_items.tolist()}")
    return model

if __name__ == "__main__":
    print("=== 전통적 협업 필터링 (Surprise) ===")
    traditional_collaborative_filtering()

    print("\n=== Neural Collaborative Filtering ===")
    train_ncf()