추천 시스템

# 협업 필터링 추천 시스템 (Surprise 라이브러리)
from surprise import Dataset, Reader, SVD, KNNBasic
from surprise.model_selection import cross_validate, train_test_split
from surprise import accuracy
import pandas as pd

# 1. 샘플 데이터 준비 (user, item, rating)
data = {
    'user_id': [1,1,1,2,2,2,3,3,3,4,4,4,5,5,5],
    'item_id': ['A','B','C','A','B','D','B','C','D','A','C','E','C','D','E'],
    'rating': [5,4,3,4,5,4,3,4,5,4,3,4,5,4,3]
}
df = pd.DataFrame(data)

# Surprise 데이터 형식으로 변환
reader = Reader(rating_scale=(1, 5))
dataset = Dataset.load_from_df(df[['user_id', 'item_id', 'rating']], reader)

# 2. 모델 선택 및 학습
# SVD (Matrix Factorization 기반)
svd_model = SVD(n_factors=50, n_epochs=20, lr_all=0.005, reg_all=0.02)

# 교차 검증
cv_results = cross_validate(svd_model, dataset, measures=['RMSE', 'MAE'], cv=3, verbose=True)
print(f"\n평균 RMSE: {cv_results['test_rmse'].mean():.4f}")
print(f"평균 MAE: {cv_results['test_mae'].mean():.4f}")

# 3. 전체 데이터로 학습 후 예측
trainset = dataset.build_full_trainset()
svd_model.fit(trainset)

# 특정 사용자에게 아이템 추천
def get_top_n_recommendations(model, user_id, all_items, n=5):
    """사용자에게 상위 N개 아이템 추천"""
    predictions = []
    for item_id in all_items:
        pred = model.predict(user_id, item_id)
        predictions.append((item_id, pred.est))

    # 예측 점수로 정렬
    predictions.sort(key=lambda x: x[1], reverse=True)
    return predictions[:n]

all_items = df['item_id'].unique()
user_1_recs = get_top_n_recommendations(svd_model, 1, all_items, n=3)
print(f"\n사용자 1 추천: {user_1_recs}")

# Two-Tower 추천 모델 (PyTorch)
import torch
import torch.nn as nn
import torch.nn.functional as F

class TwoTowerModel(nn.Module):
    """
    Two-Tower Architecture for Recommendation
    - User Tower: 사용자 특성 임베딩
    - Item Tower: 아이템 특성 임베딩
    - 두 타워의 내적으로 유사도 계산
    """
    def __init__(self, num_users, num_items, embedding_dim=64, hidden_dim=128):
        super().__init__()

        # User Tower
        self.user_embedding = nn.Embedding(num_users, embedding_dim)
        self.user_tower = nn.Sequential(
            nn.Linear(embedding_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 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, hidden_dim),
            nn.ReLU(),
            nn.Linear(embedding_dim, embedding_dim),
            nn.LayerNorm(embedding_dim)
        )

    def forward(self, user_ids, item_ids):
        # User representation
        user_emb = self.user_embedding(user_ids)
        user_repr = self.user_tower(user_emb)

        # Item representation
        item_emb = self.item_embedding(item_ids)
        item_repr = self.item_tower(item_emb)

        # 코사인 유사도 계산
        user_repr = F.normalize(user_repr, dim=1)
        item_repr = F.normalize(item_repr, dim=1)
        similarity = (user_repr * item_repr).sum(dim=1)

        return similarity

    def get_user_embedding(self, user_ids):
        """실시간 추천을 위한 사용자 임베딩 추출"""
        user_emb = self.user_embedding(user_ids)
        return F.normalize(self.user_tower(user_emb), dim=1)

    def get_item_embedding(self, item_ids):
        """인덱싱을 위한 아이템 임베딩 추출"""
        item_emb = self.item_embedding(item_ids)
        return F.normalize(self.item_tower(item_emb), dim=1)

# 모델 초기화 및 학습
model = TwoTowerModel(num_users=10000, num_items=50000, embedding_dim=128)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

# Contrastive Loss (In-batch Negatives)
def contrastive_loss(user_repr, item_repr, temperature=0.1):
    logits = torch.matmul(user_repr, item_repr.T) / temperature
    labels = torch.arange(len(user_repr), device=logits.device)
    return F.cross_entropy(logits, labels)

print(f"모델 파라미터: {sum(p.numel() for p in model.parameters()):,}")