추천 시스템

import pandas as pd
import numpy as np
from scipy.sparse import csr_matrix
from sklearn.metrics.pairwise import cosine_similarity
import warnings
warnings.filterwarnings('ignore')
 
# 재현 가능한 결과를 위한 시드 설정
np.random.seed(42)
 
# 사용자-상품 구매 데이터 생성
n_users = 200
n_products = 100
n_interactions = 3000
 
# 랜덤 상호작용 생성
user_ids = np.random.randint(1, n_users + 1, n_interactions)
product_ids = np.random.randint(1, n_products + 1, n_interactions)
 
# 구매 데이터프레임
interactions = pd.DataFrame({
    'user_id': user_ids,
    'product_id': product_ids,
    'purchase_count': np.random.randint(1, 5, n_interactions),
    'total_spent': np.random.exponential(50, n_interactions)
})
 
# 중복 제거 (동일 사용자-상품 합산)
interactions = interactions.groupby(['user_id', 'product_id']).agg({
    'purchase_count': 'sum',
    'total_spent': 'sum'
}).reset_index()
 
# 암묵적 피드백: 구매 여부 (0/1)
interactions['purchased'] = 1
 
print(f"총 상호작용: {len(interactions)}")
print(f"고유 사용자: {interactions['user_id'].nunique()}")
print(f"고유 상품: {interactions['product_id'].nunique()}")
print(f"\n상호작용 샘플:")
print(interactions.head())

# 피벗 테이블로 행렬 생성
user_item_matrix = interactions.pivot(
    index='user_id',
    columns='product_id',
    values='purchased'
).fillna(0)
 
# 희소성 계산
sparsity = 1 - (user_item_matrix.values.sum() / user_item_matrix.size)
 
print(f"행렬 크기: {user_item_matrix.shape}")
print(f"희소성: {sparsity:.2%}")
print(f"\n행렬 샘플 (처음 5명 x 처음 10개 상품):")
print(user_item_matrix.iloc[:5, :10])

# 사용자 간 유사도 계산 (코사인 유사도)
user_similarity = cosine_similarity(user_item_matrix)
user_similarity_df = pd.DataFrame(
    user_similarity,
    index=user_item_matrix.index,
    columns=user_item_matrix.index
)
 
# 특정 사용자와 가장 유사한 사용자 찾기
def find_similar_users(user_id, top_n=10):
    if user_id not in user_similarity_df.index:
        return pd.Series()
    similar = user_similarity_df[user_id].sort_values(ascending=False)[1:top_n+1]
    return similar
 
# 예시: 사용자 1과 유사한 사용자
target_user = 1
similar_users = find_similar_users(target_user)
print(f"사용자 {target_user}와 유사한 사용자:")
print(similar_users.head())

def recommend_user_based(user_id, top_n=10):
    """사용자 기반 협업 필터링 추천"""
    if user_id not in user_item_matrix.index:
        return []
 
    # 유사 사용자 찾기
    similar_users = find_similar_users(user_id, top_n=20)
 
    # 이미 구매한 상품
    purchased = set(user_item_matrix.loc[user_id][user_item_matrix.loc[user_id] > 0].index)
 
    # 유사 사용자들이 구매한 상품 점수 계산
    recommendations = {}
    for sim_user, similarity in similar_users.items():
        sim_user_purchases = user_item_matrix.loc[sim_user]
        for product, purchased_flag in sim_user_purchases.items():
            if purchased_flag > 0 and product not in purchased:
                if product not in recommendations:
                    recommendations[product] = 0
                recommendations[product] += similarity
 
    # 점수 순 정렬
    sorted_recs = sorted(recommendations.items(), key=lambda x: x[1], reverse=True)
    return sorted_recs[:top_n]
 
# 추천 실행
recs_user_based = recommend_user_based(target_user)
print(f"\n사용자 {target_user} 추천 상품 (User-Based CF):")
for product, score in recs_user_based[:5]:
    print(f"  상품 {product}: 점수 {score:.3f}")

# 아이템 간 유사도 계산
item_similarity = cosine_similarity(user_item_matrix.T)
item_similarity_df = pd.DataFrame(
    item_similarity,
    index=user_item_matrix.columns,
    columns=user_item_matrix.columns
)
 
def find_similar_items(product_id, top_n=10):
    """유사 아이템 찾기"""
    if product_id not in item_similarity_df.index:
        return pd.Series()
    similar = item_similarity_df[product_id].sort_values(ascending=False)[1:top_n+1]
    return similar
 
# 예시: 상품 3과 유사한 상품
target_product = 3
similar_items = find_similar_items(target_product)
print(f"상품 {target_product}와 유사한 상품:")
print(similar_items.head())

def recommend_item_based(user_id, top_n=10):
    """아이템 기반 협업 필터링 추천"""
    if user_id not in user_item_matrix.index:
        return []
 
    # 사용자가 구매한 상품
    user_purchases = user_item_matrix.loc[user_id]
    purchased = set(user_purchases[user_purchases > 0].index)
 
    # 구매한 상품과 유사한 상품 점수 계산
    recommendations = {}
    for product in purchased:
        similar = find_similar_items(product, top_n=20)
        for sim_product, similarity in similar.items():
            if sim_product not in purchased:
                if sim_product not in recommendations:
                    recommendations[sim_product] = 0
                recommendations[sim_product] += similarity
 
    sorted_recs = sorted(recommendations.items(), key=lambda x: x[1], reverse=True)
    return sorted_recs[:top_n]
 
# 추천 실행
recs_item_based = recommend_item_based(target_user)
print(f"\n아이템 기반 추천 (Item-Based CF):")
for product, score in recs_item_based[:5]:
    print(f"  상품 {product}: 점수 {score:.3f}")

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import OneHotEncoder
 
# 상품 메타데이터 생성
categories = ['Electronics', 'Clothing', 'Books', 'Home', 'Sports']
brands = ['BrandA', 'BrandB', 'BrandC', 'BrandD', 'BrandE']
 
products = pd.DataFrame({
    'product_id': range(1, n_products + 1),
    'name': [f'Product {i}' for i in range(1, n_products + 1)],
    'category': np.random.choice(categories, n_products),
    'brand': np.random.choice(brands, n_products),
    'price': np.random.exponential(50, n_products) + 10
})
 
print("상품 메타데이터 샘플:")
print(products.head(10))

# 범주형 특성: One-Hot Encoding
categorical_features = pd.get_dummies(products[['category', 'brand']])
 
# 가격 정규화
products['price_norm'] = (products['price'] - products['price'].min()) / \
                          (products['price'].max() - products['price'].min())
 
# 특성 결합
product_features = pd.concat([categorical_features, products[['price_norm']]], axis=1)
 
# 콘텐츠 유사도 계산
content_similarity = cosine_similarity(product_features)
content_similarity_df = pd.DataFrame(
    content_similarity,
    index=products['product_id'],
    columns=products['product_id']
)
 
def find_similar_items_content(product_id, top_n=10):
    """콘텐츠 기반 유사 아이템"""
    if product_id not in content_similarity_df.index:
        return pd.Series()
    similar = content_similarity_df.loc[product_id].sort_values(ascending=False)[1:top_n+1]
    return similar
 
# 예시
print(f"\n상품 3과 콘텐츠 기반 유사 상품:")
content_similar = find_similar_items_content(3)
for prod, sim in content_similar.head().items():
    prod_info = products[products['product_id'] == prod].iloc[0]
    print(f"  상품 {prod} ({prod_info['category']}, {prod_info['brand']}): {sim:.3f}")

def recommend_content_based(user_id, top_n=10):
    """콘텐츠 기반 추천"""
    if user_id not in user_item_matrix.index:
        return []
 
    # 사용자가 구매한 상품
    user_purchases = user_item_matrix.loc[user_id]
    purchased = list(user_purchases[user_purchases > 0].index)
 
    # 구매 상품과 유사한 콘텐츠 찾기
    recommendations = {}
    for product in purchased:
        if product in content_similarity_df.index:
            similar = find_similar_items_content(product, top_n=20)
            for sim_product, similarity in similar.items():
                if sim_product not in purchased:
                    if sim_product not in recommendations:
                        recommendations[sim_product] = 0
                    recommendations[sim_product] += similarity
 
    sorted_recs = sorted(recommendations.items(), key=lambda x: x[1], reverse=True)
    return sorted_recs[:top_n]
 
# 콘텐츠 기반 추천 실행
recs_content = recommend_content_based(target_user)
print(f"\n콘텐츠 기반 추천:")
for product, score in recs_content[:5]:
    prod_info = products[products['product_id'] == product].iloc[0]
    print(f"  상품 {product} ({prod_info['category']}): 점수 {score:.3f}")

def recommend_hybrid(user_id, cf_weight=0.6, content_weight=0.4, top_n=10):
    """하이브리드 추천 (CF + 콘텐츠)"""
 
    # 협업 필터링 점수
    cf_recs = recommend_item_based(user_id, top_n=50)
    cf_scores = {p: s for p, s in cf_recs}
 
    # 콘텐츠 기반 점수
    content_recs = recommend_content_based(user_id, top_n=50)
    content_scores = {p: s for p, s in content_recs}
 
    # 점수 정규화 및 결합
    all_products = set(cf_scores.keys()) | set(content_scores.keys())
 
    # Min-Max 정규화
    def normalize(scores):
        if not scores:
            return {}
        min_s, max_s = min(scores.values()), max(scores.values())
        if max_s == min_s:
            return {k: 0.5 for k in scores}
        return {k: (v - min_s) / (max_s - min_s) for k, v in scores.items()}
 
    cf_norm = normalize(cf_scores)
    content_norm = normalize(content_scores)
 
    # 가중 합산
    final_scores = {}
    for product in all_products:
        cf_s = cf_norm.get(product, 0)
        content_s = content_norm.get(product, 0)
        final_scores[product] = cf_weight * cf_s + content_weight * content_s
 
    sorted_recs = sorted(final_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_recs[:top_n]
 
# 하이브리드 추천 실행
hybrid_recs = recommend_hybrid(target_user)
print(f"하이브리드 추천 (CF 60% + 콘텐츠 40%):")
for product, score in hybrid_recs[:5]:
    prod_info = products[products['product_id'] == product].iloc[0]
    print(f"  상품 {product} ({prod_info['category']}): 점수 {score:.3f}")

def evaluate_recommendations(recommend_func, test_users, k=10):
    """추천 성능 평가"""
    precisions = []
    recalls = []
    hits = 0
 
    for user_id in test_users:
        # 사용자의 실제 구매 상품 (마지막 20%를 테스트로 가정)
        user_purchases = user_item_matrix.loc[user_id]
        all_purchased = list(user_purchases[user_purchases > 0].index)
 
        if len(all_purchased) < 5:
            continue
 
        # 테스트 세트: 마지막 20% 상품
        test_size = max(1, len(all_purchased) // 5)
        test_items = set(all_purchased[-test_size:])
 
        # 추천 상품
        recommendations = recommend_func(user_id, top_n=k)
        recommended = set([p for p, _ in recommendations])
 
        # Precision@K
        hits_count = len(recommended & test_items)
        precision = hits_count / k if k > 0 else 0
        precisions.append(precision)
 
        # Recall@K
        recall = hits_count / len(test_items) if len(test_items) > 0 else 0
        recalls.append(recall)
 
        # Hit Rate
        if hits_count > 0:
            hits += 1
 
    n_users = len(precisions)
    results = {
        'Precision@K': np.mean(precisions) if precisions else 0,
        'Recall@K': np.mean(recalls) if recalls else 0,
        'Hit Rate': hits / n_users if n_users > 0 else 0
    }
    return results
 
# 테스트 사용자 샘플
test_users = user_item_matrix.index[:50].tolist()
 
# 각 모델 평가
print("=== 추천 모델 성능 비교 (K=10) ===\n")
 
models = [
    ('User-Based CF', recommend_user_based),
    ('Item-Based CF', recommend_item_based),
    ('Content-Based', recommend_content_based),
    ('Hybrid', recommend_hybrid)
]
 
results_list = []
for name, func in models:
    results = evaluate_recommendations(func, test_users, k=10)
    results['모델'] = name
    results_list.append(results)
    print(f"{name}:")
    for metric, value in results.items():
        if metric != '모델':
            print(f"  {metric}: {value:.3f}")
    print()
 
# 결과 테이블
results_df = pd.DataFrame(results_list)[['모델', 'Precision@K', 'Recall@K', 'Hit Rate']]
print("=== 성능 요약 ===")
print(results_df.to_string(index=False))

def recommend_for_new_user(top_n=10):
    """신규 사용자 추천 (콜드 스타트)"""
    # 인기 상품 추천
    product_popularity = interactions.groupby('product_id')['user_id'].count()
    popular_products = product_popularity.sort_values(ascending=False).head(top_n)
    return list(popular_products.index)
 
def recommend_for_new_item(product_id, top_n=10):
    """신규 상품 추천 대상 사용자 찾기"""
    # 콘텐츠 기반으로 유사 상품 구매한 사용자 타겟
    similar = find_similar_items_content(product_id, top_n=5)
    target_users = set()
 
    for sim_product in similar.index:
        buyers = interactions[interactions['product_id'] == sim_product]['user_id'].unique()
        target_users.update(buyers)
 
    return list(target_users)[:top_n]
 
# 신규 사용자 추천
print("=== 신규 사용자 추천 (인기 상품) ===")
popular_recs = recommend_for_new_user()
for i, prod in enumerate(popular_recs[:5], 1):
    prod_info = products[products['product_id'] == prod].iloc[0]
    print(f"{i}. 상품 {prod} ({prod_info['category']})")

def diversify_recommendations(recommendations, top_n=10, diversity_weight=0.3):
    """추천 다양성 향상"""
    if not recommendations:
        return []
 
    selected = [recommendations[0]]
    remaining = list(recommendations[1:])
 
    while len(selected) < top_n and remaining:
        best_item = None
        best_score = -1
 
        for item, score in remaining:
            # 기존 선택 아이템과의 평균 거리 (다양성)
            if item in content_similarity_df.index:
                avg_sim = np.mean([
                    content_similarity_df.loc[item, s[0]]
                    for s in selected if s[0] in content_similarity_df.index
                ])
                diversity = 1 - avg_sim
            else:
                diversity = 0.5
 
            # 점수 + 다양성 가중합
            combined = (1 - diversity_weight) * score + diversity_weight * diversity
 
            if combined > best_score:
                best_score = combined
                best_item = (item, score)
 
        if best_item:
            selected.append(best_item)
            remaining.remove(best_item)
 
    return selected
 
# 다양성 적용 추천
original_recs = recommend_hybrid(target_user, top_n=10)
diverse_recs = diversify_recommendations(original_recs, diversity_weight=0.3)
 
print("=== 다양성 적용 전/후 비교 ===")
print("\n원본 추천:")
for prod, score in original_recs[:5]:
    prod_info = products[products['product_id'] == prod].iloc[0]
    print(f"  {prod} ({prod_info['category']})")
 
print("\n다양성 적용 후:")
for prod, score in diverse_recs[:5]:
    prod_info = products[products['product_id'] == prod].iloc[0]
    print(f"  {prod} ({prod_info['category']})")