#!/usr/bin/env python3 """ Retention Analysis Script v2.0 던전 스토커즈 신규 유저 리텐션 분석 도구 사용법: python retention_analysis.py [CSV_파일_경로] 설정: retention_analysis_config.json """ import pandas as pd import numpy as np from scipy import stats from scipy.stats import spearmanr import sys import os import json import logging from datetime import datetime from sklearn.ensemble import RandomForestClassifier from tqdm import tqdm import warnings warnings.filterwarnings('ignore') class RetentionAnalyzer: def __init__(self, csv_path, config_path='retention_analysis_config.json'): """ 초기화 Parameters: ----------- csv_path : str 분석할 CSV 파일 경로 config_path : str 설정 파일 경로 """ self.csv_path = csv_path self.config = self.load_config(config_path) self.df = None self.correlation_results = {} self.feature_importance = {} self.console_log = [] # 출력 디렉토리 설정 self.output_dir = 'analysis_results' os.makedirs(self.output_dir, exist_ok=True) # 타임스탬프 self.timestamp = datetime.now().strftime(self.config['output_settings']['timestamp_format']) # 로깅 설정 self.setup_logging() def setup_logging(self): """로깅 시스템 초기화""" log_filename = f"{self.output_dir}/retention_analysis_{self.timestamp}.log" # 콘솔과 파일에 동시 로깅 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.FileHandler(log_filename, encoding='utf-8'), logging.StreamHandler(sys.stdout) ] ) self.logger = logging.getLogger(__name__) def load_config(self, config_path): """설정 파일 로드""" try: with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) return config except FileNotFoundError: print(f"Warning: {config_path} 파일을 찾을 수 없습니다. 기본 설정을 사용합니다.") return self.get_default_config() except json.JSONDecodeError as e: print(f"Error: {config_path} 파일 형식이 올바르지 않습니다: {e}") return self.get_default_config() def get_default_config(self): """기본 설정 반환""" return { "analysis_settings": { "correlation_threshold": 0.05, "top_n_features": 20, "min_sample_size": 10 }, "retention_groups": [ "Retained_d0", "Retained_d1", "Retained_d2", "Retained_d3", "Retained_d4", "Retained_d5", "Retained_d6", "Retained_d7+" ], "exclude_columns": [ "uid", "retention_status", "retention_encoded", "country", "nickname", "auth_id" ], "key_metrics_for_d0_analysis": [ "tutorial_complete", "level_up_count", "play_time_total", "COOP_entry_count", "Solo_entry_count", "death_Monster", "death_Trap", "death_PK", "item_gain_Equipment", "gold_gain_total" ], "output_settings": { "save_csv": True, "save_markdown": True, "save_console_log": True, "timestamp_format": "%Y%m%d_%H%M%S" } } def log_message(self, stage, message, level="INFO"): """구조화된 로그 메시지""" timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') formatted_msg = f"[{timestamp}] [{stage}] {level}: {message}" if level == "INFO": self.logger.info(f"[{stage}] {message}") elif level == "WARNING": self.logger.warning(f"[{stage}] {message}") elif level == "ERROR": self.logger.error(f"[{stage}] {message}") self.console_log.append(formatted_msg) def validate_csv(self): """CSV 파일 유효성 검사""" self.log_message("VALIDATION", "CSV 파일 유효성 검사 시작") if not os.path.exists(self.csv_path): raise FileNotFoundError(f"파일을 찾을 수 없습니다: {self.csv_path}") # 필수 컬럼 검사 try: sample_df = pd.read_csv(self.csv_path, nrows=5) required_cols = ['uid', 'retention_status'] missing_cols = [col for col in required_cols if col not in sample_df.columns] if missing_cols: raise ValueError(f"필수 컬럼이 없습니다: {missing_cols}") # retention_status 값 검사 retention_groups = set(self.config['retention_groups']) sample_retention = set(sample_df['retention_status'].dropna().unique()) if not sample_retention.issubset(retention_groups): invalid_values = sample_retention - retention_groups self.log_message("VALIDATION", f"예상치 못한 retention_status 값: {invalid_values}", "WARNING") self.log_message("VALIDATION", "CSV 파일 유효성 검사 완료 OK") except Exception as e: raise ValueError(f"CSV 파일 검증 실패: {str(e)}") def load_data(self): """CSV 데이터 로드 및 기본 전처리""" self.log_message("LOAD", "데이터 로딩 시작") # 파일 크기 확인 file_size = os.path.getsize(self.csv_path) / (1024 * 1024) # MB self.log_message("LOAD", f"파일 크기: {file_size:.1f} MB") # 진행률 바와 함께 데이터 로드 with tqdm(desc="데이터 로딩 중", unit="MB") as pbar: self.df = pd.read_csv(self.csv_path) pbar.update(file_size) self.log_message("LOAD", f"데이터 로드 완료: {len(self.df):,} rows, {len(self.df.columns)} columns") # 분석 기간 정보 추출 (create_time 기반) self.analysis_period_info = self.extract_analysis_period() # retention_status를 순서형 변수로 인코딩 retention_groups = self.config['retention_groups'] self.df['retention_encoded'] = self.df['retention_status'].map( {status: i for i, status in enumerate(retention_groups)} ) # 기본 정보 출력 self.log_message("STATS", "=== 데이터 기본 정보 ===") self.log_message("STATS", f"총 유저 수: {len(self.df):,}") # 분석 기간 정보 출력 if self.analysis_period_info['valid']: self.log_message("STATS", f"분석 기간: {self.analysis_period_info['start_date']} ~ {self.analysis_period_info['end_date']}") self.log_message("STATS", f"분석 대상 기간: {self.analysis_period_info['total_days']}일") self.log_message("STATS", f"일평균 신규 가입: {self.analysis_period_info['avg_daily_signups']:.1f}명") self.log_message("STATS", "") self.log_message("STATS", "Retention 분포:") retention_dist = self.df['retention_status'].value_counts() for status in retention_groups: if status in retention_dist.index: count = retention_dist[status] pct = count / len(self.df) * 100 self.log_message("STATS", f" {status}: {count:,} ({pct:.1f}%)") # D0 이탈률 강조 d0_rate = retention_dist.get('Retained_d0', 0) / len(self.df) * 100 self.log_message("STATS", "") self.log_message("STATS", f"CRITICAL: D0 이탈률: {d0_rate:.1f}%", "WARNING") return self.df def extract_analysis_period(self): """분석 기간 정보 추출 (ISO 8601 형식 create_time 처리)""" try: # create_time 컬럼이 있는지 확인 if 'create_time' not in self.df.columns: self.log_message("PERIOD", "create_time 컬럼이 없습니다. 기간 분석을 생략합니다.", "WARNING") return {'valid': False} # ISO 8601 형식 (2025-08-13T20:14:43+09:00)을 datetime으로 변환 # 빈 값이 없다고 했으므로 errors='raise'로 설정하여 문제 발견 시 즉시 알림 create_times = pd.to_datetime(self.df['create_time'], format='ISO8601') # 기간 정보 계산 start_date = create_times.min() end_date = create_times.max() total_days = (end_date - start_date).days + 1 # 일별 가입자 분포 (한국 시간 기준으로 날짜 추출) korea_dates = create_times.dt.tz_convert('Asia/Seoul').dt.date daily_signups = korea_dates.value_counts().sort_index() avg_daily_signups = len(create_times) / total_days if total_days > 0 else 0 # 시간대별 가입 패턴 분석 (추가 인사이트) korea_hours = create_times.dt.tz_convert('Asia/Seoul').dt.hour hourly_signups = korea_hours.value_counts().sort_index() peak_hour = hourly_signups.idxmax() # 요일별 가입 패턴 (0=월요일, 6=일요일) korea_weekdays = create_times.dt.tz_convert('Asia/Seoul').dt.dayofweek weekday_signups = korea_weekdays.value_counts().sort_index() weekday_names = ['월', '화', '수', '목', '금', '토', '일'] self.log_message("PERIOD", f"가입 집중 시간: {peak_hour}시 ({hourly_signups[peak_hour]}명)") return { 'valid': True, 'start_date': start_date.strftime('%Y-%m-%d'), 'end_date': end_date.strftime('%Y-%m-%d'), 'start_datetime': start_date, 'end_datetime': end_date, 'total_days': total_days, 'total_users': len(create_times), 'avg_daily_signups': avg_daily_signups, 'valid_ratio': 100.0, # 빈 값이 없다고 했으므로 항상 100% 'daily_signups': daily_signups, 'hourly_signups': hourly_signups, 'weekday_signups': weekday_signups, 'peak_hour': peak_hour, 'peak_hour_count': hourly_signups[peak_hour], 'weekday_names': weekday_names } except Exception as e: self.log_message("PERIOD", f"분석 기간 추출 중 오류: {str(e)}", "ERROR") # ISO 형식 파싱 실패 시 일반적인 방법으로 재시도 try: create_times = pd.to_datetime(self.df['create_time'], errors='coerce') valid_times = create_times.dropna() if len(valid_times) == 0: return {'valid': False} start_date = valid_times.min() end_date = valid_times.max() total_days = (end_date - start_date).days + 1 daily_signups = valid_times.dt.date.value_counts().sort_index() avg_daily_signups = len(valid_times) / total_days if total_days > 0 else 0 return { 'valid': True, 'start_date': start_date.strftime('%Y-%m-%d'), 'end_date': end_date.strftime('%Y-%m-%d'), 'start_datetime': start_date, 'end_datetime': end_date, 'total_days': total_days, 'total_users': len(valid_times), 'avg_daily_signups': avg_daily_signups, 'valid_ratio': len(valid_times) / len(self.df) * 100, 'daily_signups': daily_signups } except: return {'valid': False} def analyze_correlations(self): """모든 지표와 retention_status 간의 상관관계 분석""" self.log_message("CORRELATION", "상관관계 분석 시작") # 분석할 컬럼 선택 exclude_cols = self.config['exclude_columns'] all_numeric_cols = self.df.select_dtypes(include=[np.number]).columns.tolist() numeric_cols = [col for col in all_numeric_cols if col not in exclude_cols] self.log_message("CORRELATION", f"분석 대상 지표: {len(numeric_cols)}개") correlations = [] # 진행률 바와 함께 상관관계 분석 for col in tqdm(numeric_cols, desc="상관관계 분석 중"): try: # NULL 값 제거 valid_data = self.df[[col, 'retention_encoded']].dropna() if len(valid_data) < self.config['analysis_settings']['min_sample_size']: continue # Spearman 상관계수 계산 corr, p_value = spearmanr(valid_data[col], valid_data['retention_encoded']) correlations.append({ 'metric': col, 'correlation': corr, 'p_value': p_value, 'significant': p_value < self.config['analysis_settings']['correlation_threshold'], 'abs_correlation': abs(corr) }) except Exception as e: self.log_message("CORRELATION", f"Warning: {col} 분석 실패 - {str(e)}", "WARNING") continue # 결과 정리 self.correlation_results = pd.DataFrame(correlations) self.correlation_results = self.correlation_results.sort_values('abs_correlation', ascending=False) # 상위 결과 출력 self.log_message("CORRELATION", "") self.log_message("CORRELATION", "[+] Retention에 긍정적 영향 (Top 15):") positive = self.correlation_results[ (self.correlation_results['correlation'] > 0) & (self.correlation_results['significant']) ].head(15) for _, row in positive.iterrows(): self.log_message("CORRELATION", f" {row['metric']}: {row['correlation']:.3f}") self.log_message("CORRELATION", "") self.log_message("CORRELATION", "[-] Retention에 부정적 영향 (Top 15):") negative = self.correlation_results[ (self.correlation_results['correlation'] < 0) & (self.correlation_results['significant']) ].head(15) for _, row in negative.iterrows(): self.log_message("CORRELATION", f" {row['metric']}: {row['correlation']:.3f}") # 결과 저장 if self.config['output_settings']['save_csv']: output_file = f"{self.output_dir}/correlation_analysis_{self.timestamp}.csv" self.correlation_results.to_csv(output_file, index=False) self.log_message("CORRELATION", f"결과 저장: {output_file}") return self.correlation_results def analyze_d0_churners(self): """D0 이탈 유저 특성 분석""" self.log_message("D0_ANALYSIS", "D0 이탈 유저 특성 분석 시작") d0_users = self.df[self.df['retention_status'] == 'Retained_d0'] retained_users = self.df[self.df['retention_status'] != 'Retained_d0'] self.log_message("D0_ANALYSIS", f"D0 이탈 유저: {len(d0_users):,}명") self.log_message("D0_ANALYSIS", f"잔존 유저: {len(retained_users):,}명") # 주요 지표 비교 key_metrics = self.config['key_metrics_for_d0_analysis'] comparison = [] self.log_message("D0_ANALYSIS", "") self.log_message("D0_ANALYSIS", "=== D0 이탈 vs 잔존 유저 주요 차이 ===") for metric in tqdm(key_metrics, desc="D0 분석 중"): if metric in self.df.columns: try: d0_data = d0_users[metric].dropna() retained_data = retained_users[metric].dropna() if len(d0_data) < 10 or len(retained_data) < 10: continue d0_mean = d0_data.mean() retained_mean = retained_data.mean() # t-test t_stat, p_value = stats.ttest_ind(d0_data, retained_data, equal_var=False) diff_pct = ((retained_mean - d0_mean) / (d0_mean + 0.0001)) * 100 comparison.append({ 'metric': metric, 'd0_mean': d0_mean, 'retained_mean': retained_mean, 'difference_%': diff_pct, 'p_value': p_value, 'significant': p_value < 0.05 }) except Exception as e: self.log_message("D0_ANALYSIS", f"Warning: {metric} 분석 실패 - {str(e)}", "WARNING") continue comparison_df = pd.DataFrame(comparison) comparison_df = comparison_df.sort_values('difference_%', ascending=False) # 클래스 변수로 저장 (HTML 리포트에서 사용) self.d0_comparison = comparison_df # 결과 출력 for _, row in comparison_df.head(10).iterrows(): if row['significant']: self.log_message("D0_ANALYSIS", f" {row['metric']}:") self.log_message("D0_ANALYSIS", f" D0 평균: {row['d0_mean']:.2f}") self.log_message("D0_ANALYSIS", f" 잔존 평균: {row['retained_mean']:.2f}") self.log_message("D0_ANALYSIS", f" 차이: {row['difference_%']:.1f}%") # 결과 저장 if self.config['output_settings']['save_csv']: output_file = f"{self.output_dir}/d0_churn_analysis_{self.timestamp}.csv" comparison_df.to_csv(output_file, index=False) self.log_message("D0_ANALYSIS", f"결과 저장: {output_file}") return comparison_df def feature_importance_analysis(self): """Random Forest를 이용한 Feature Importance 분석""" self.log_message("FEATURE_IMPORTANCE", "Feature Importance 분석 시작") # 데이터 준비 exclude_cols = self.config['exclude_columns'] numeric_cols = self.df.select_dtypes(include=[np.number]).columns.tolist() feature_cols = [col for col in numeric_cols if col not in exclude_cols] X = self.df[feature_cols].fillna(0) y = self.df['retention_encoded'] self.log_message("FEATURE_IMPORTANCE", f"Feature 수: {len(feature_cols)}") self.log_message("FEATURE_IMPORTANCE", f"샘플 수: {len(X)}") # Random Forest 모델 학습 (진행률 표시) with tqdm(desc="Random Forest 학습 중") as pbar: rf = RandomForestClassifier(n_estimators=100, random_state=42, n_jobs=-1) rf.fit(X, y) pbar.update(1) # Feature Importance 추출 importance_df = pd.DataFrame({ 'feature': feature_cols, 'importance': rf.feature_importances_ }).sort_values('importance', ascending=False) self.log_message("FEATURE_IMPORTANCE", "") self.log_message("FEATURE_IMPORTANCE", f"=== Top {self.config['analysis_settings']['top_n_features']} 중요 지표 ===") top_n = self.config['analysis_settings']['top_n_features'] for idx, row in importance_df.head(top_n).iterrows(): self.log_message("FEATURE_IMPORTANCE", f" {row['feature']}: {row['importance']:.4f}") # 결과 저장 if self.config['output_settings']['save_csv']: output_file = f"{self.output_dir}/feature_importance_{self.timestamp}.csv" importance_df.to_csv(output_file, index=False) self.log_message("FEATURE_IMPORTANCE", f"결과 저장: {output_file}") self.feature_importance = importance_df return importance_df def generate_insights_report(self): """HTML 형식의 종합 인사이트 리포트 생성""" self.log_message("REPORT", "HTML 인사이트 리포트 생성 시작") # 기본 통계 계산 d0_rate = len(self.df[self.df['retention_status'] == 'Retained_d0']) / len(self.df) * 100 d1_rate = len(self.df[self.df['retention_status'] == 'Retained_d1']) / len(self.df) * 100 d2_rate = len(self.df[self.df['retention_status'] == 'Retained_d2']) / len(self.df) * 100 d3_rate = len(self.df[self.df['retention_status'] == 'Retained_d3']) / len(self.df) * 100 d3_plus_rate = len(self.df[self.df['retention_encoded'] >= 3]) / len(self.df) * 100 d7_plus_rate = len(self.df[self.df['retention_status'] == 'Retained_d7+']) / len(self.df) * 100 # 리텐션 분포 데이터 retention_dist = self.df['retention_status'].value_counts().sort_index() # 상관관계 분석 결과 준비 (HTML 전체에서 사용하기 위해 미리 준비) positive = pd.DataFrame() # 기본값 negative = pd.DataFrame() # 기본값 if not self.correlation_results.empty: positive = self.correlation_results[ (self.correlation_results['correlation'] > 0) & (self.correlation_results['significant']) ].head(15) negative = self.correlation_results[ (self.correlation_results['correlation'] < 0) & (self.correlation_results['significant']) ].head(15) # HTML 생성 html_content = f""" 던전 스토커즈 - 신규 유저 리텐션 분석 리포트

던전 스토커즈

신규 유저 리텐션 분석 리포트

분석 일시: {datetime.now().strftime('%Y년 %m월 %d일 %H:%M:%S')}

📊 분석 개요

데이터 파일: {os.path.basename(self.csv_path)}
분석 대상: {len(self.df):,}명의 신규 유저
""" # 분석 기간 정보 추가 if hasattr(self, 'analysis_period_info') and self.analysis_period_info['valid']: period_info = self.analysis_period_info html_content += f""" 분석 기간: {period_info['start_date']} ~ {period_info['end_date']} ({period_info['total_days']}일)
일평균 신규 가입: {period_info['avg_daily_signups']:.1f}명
데이터 품질: {period_info['valid_ratio']:.1f}% (유효한 가입일 데이터)
""" html_content += f""" 분석 범위: D0 ~ D7+ 리텐션 분석
통계 방법: Spearman 상관분석, Welch's t-test, Random Forest 특성 중요도
""" # 분석 기간 상세 정보 (별도 섹션) if hasattr(self, 'analysis_period_info') and self.analysis_period_info['valid']: period_info = self.analysis_period_info # 최근 7일, 최근 30일 가입자 수 계산 (가능한 경우) recent_stats = "" try: end_datetime = period_info['end_datetime'] recent_7days = len(self.df[pd.to_datetime(self.df['create_time'], errors='coerce') >= (end_datetime - pd.Timedelta(days=7))]) recent_30days = len(self.df[pd.to_datetime(self.df['create_time'], errors='coerce') >= (end_datetime - pd.Timedelta(days=30))]) recent_stats = f"""
{recent_7days:,}
최근 7일 가입
{recent_30days:,}
최근 30일 가입
{period_info['total_days']}
전체 분석 일수
{period_info['avg_daily_signups']:.1f}
일평균 신규가입
""" except: pass # 시간대/요일별 패턴 정보 추가 pattern_info = "" if 'peak_hour' in period_info and 'weekday_signups' in period_info: # 가장 활발한 요일 찾기 max_weekday_idx = period_info['weekday_signups'].idxmax() max_weekday_name = period_info['weekday_names'][max_weekday_idx] max_weekday_count = period_info['weekday_signups'][max_weekday_idx] pattern_info = f"""

🕒 가입 패턴 분석

가입 집중 시간대: {period_info['peak_hour']}시 ({period_info['peak_hour_count']:,}명)
가입 집중 요일: {max_weekday_name}요일 ({max_weekday_count:,}명)
시간대 분석: 한국 시간(KST) 기준으로 분석됨
""" html_content += f"""

📅 분석 기간 상세

데이터 수집 기간: {period_info['start_date']} ~ {period_info['end_date']}
전체 분석 대상: {period_info['total_users']:,}명 (ISO 8601 표준 형식)
일평균 신규 가입자: {period_info['avg_daily_signups']:.1f}명
분석 대상 기간: {period_info['total_days']}일간의 신규 가입자 행동 패턴
데이터 품질: {period_info['valid_ratio']:.1f}% (표준 시간 형식으로 완전한 데이터)
{pattern_info} {recent_stats}""" html_content += f"""

🎯 전체 리텐션 현황

{d0_rate:.1f}%
D0 이탈률
{d1_rate:.1f}%
D1 유지율
{d3_plus_rate:.1f}%
D3+ 잔존율
{d7_plus_rate:.1f}%
D7+ 잔존율

⚠️ Critical Alert

D0 이탈률 {d0_rate:.1f}%는 게임 업계 평균(25-35%)을 크게 상회하는 매우 심각한 수준입니다.

즉시 개선 조치가 필요합니다.

일별 리텐션 분포

""" # 리텐션 분포 HTML 생성 for status in self.config['retention_groups']: if status in retention_dist.index: count = retention_dist[status] pct = count / len(self.df) * 100 html_content += f"""
{status.replace('Retained_', '')}
{count:,}
{pct:.1f}%
""" html_content += """
""" # 상관관계 분석 결과 if not self.correlation_results.empty: html_content += f"""

🔍 리텐션 영향 요인 분석

분석 방법: Spearman 순위 상관계수 (ρ)
유의수준: p < 0.05
분석 지표 수: {len(self.correlation_results)}개
유의한 지표 수: {len(self.correlation_results[self.correlation_results['significant']])}개

📈 리텐션 향상 요인 (긍정적 상관관계)

""" for idx, (_, row) in enumerate(positive.iterrows(), 1): interpretation = "강한 긍정적 관계" if row['correlation'] > 0.3 else "중간 긍정적 관계" if row['correlation'] > 0.1 else "약한 긍정적 관계" html_content += f""" """ html_content += """
순위 지표명 상관계수 (ρ) 유의확률 (p) 해석
{idx} {row['metric']} +{row['correlation']:.3f} {'< 0.001' if row['p_value'] < 0.001 else f'{row["p_value"]:.3f}'} {interpretation}
""" if len(negative) > 0: html_content += """

📉 리텐션 저해 요인 (부정적 상관관계)

""" for idx, (_, row) in enumerate(negative.iterrows(), 1): interpretation = "강한 부정적 관계" if abs(row['correlation']) > 0.3 else "중간 부정적 관계" if abs(row['correlation']) > 0.1 else "약한 부정적 관계" html_content += f""" """ html_content += """
순위 지표명 상관계수 (ρ) 유의확률 (p) 해석
{idx} {row['metric']} {row['correlation']:.3f} {'< 0.001' if row['p_value'] < 0.001 else f'{row["p_value"]:.3f}'} {interpretation}
""" html_content += """
""" # D0 분석 결과가 있다면 추가 if hasattr(self, 'd0_comparison') and not self.d0_comparison.empty: html_content += f"""

🎯 D0 이탈 vs 잔존 유저 비교 분석

분석 대상:
D0 이탈 유저: {len(self.df[self.df['retention_status'] == 'Retained_d0']):,}명
잔존 유저: {len(self.df[self.df['retention_status'] != 'Retained_d0']):,}명

핵심 지표별 차이 분석

""" significant_diffs = self.d0_comparison[self.d0_comparison['significant']].head(15) for _, row in significant_diffs.iterrows(): if row['difference_%'] > 100: interpretation = "매우 큰 차이 (즉시 개선 필요)" elif row['difference_%'] > 50: interpretation = "큰 차이 (우선 개선)" elif row['difference_%'] > 20: interpretation = "중간 차이 (개선 고려)" else: interpretation = "작은 차이" diff_class = "positive" if row['difference_%'] > 0 else "negative" html_content += f""" """ html_content += """
지표명 D0 이탈자 평균 잔존자 평균 차이율 유의확률 (p) 해석
{row['metric']} {row['d0_mean']:.2f} {row['retained_mean']:.2f} {row['difference_%']:+.1f}% {'< 0.001' if row['p_value'] < 0.001 else f'{row["p_value"]:.3f}'} {interpretation}
""" # Feature Importance 결과 if hasattr(self, 'feature_importance') and not self.feature_importance.empty: top_features = self.feature_importance.head(20) html_content += f"""

🎖️ 리텐션 예측 중요도 분석

분석 방법: Random Forest Classifier
모델 설정: 100개 트리, 교차 검증
분석 특성 수: {len(self.feature_importance)}개

Top 20 중요 특성

""" for idx, (_, row) in enumerate(top_features.iterrows(), 1): importance = row['importance'] if importance >= 0.1: grade = "매우 높음" priority = "즉시 액션" grade_class = "critical" elif importance >= 0.05: grade = "높음" priority = "우선 개선" grade_class = "warning" elif importance >= 0.01: grade = "보통" priority = "모니터링" grade_class = "good" else: grade = "낮음" priority = "후순위" grade_class = "" html_content += f""" """ html_content += """
순위 지표명 중요도 점수 중요도 등급 액션 우선순위
{idx} {row['feature']} {importance:.4f} {grade} {priority}
""" # 개선 권고사항 html_content += f"""

🚀 개선 권고사항

🔥 즉시 실행 (Critical)

  • D0 이탈률 {d0_rate:.1f}% - 업계 평균의 2배 수준으로 즉시 개선 필요
    • """ if not self.correlation_results.empty and len(positive) > 0: top_positive = positive.iloc[0] html_content += f"""
  • {top_positive['metric']} 지표 강화 - 최고 긍정 요인 (상관계수: {top_positive['correlation']:.3f})
    • """ html_content += """
  • 신규 유저 온보딩 프로세스 전면 재검토
  • 첫날 경험 최적화 프로젝트 착수

⚡ 우선 개선 (High Priority)

    """ if not self.correlation_results.empty: for _, row in positive.head(3).iterrows(): html_content += f"""
  • {row['metric']} 개선 - 리텐션 향상 효과 기대 (ρ = {row['correlation']:.3f})
    • """ html_content += """
  • D1-D3 전환율 개선 프로그램 도입
  • 이탈 위험 유저 조기 탐지 시스템 구축

📊 지속 모니터링 (Monitoring)

    """ if len(negative) > 0: for _, row in negative.head(3).iterrows(): html_content += f"""
  • {row['metric']} 모니터링 - 부정적 영향 최소화 (ρ = {row['correlation']:.3f})
    • """ html_content += f"""
  • 주간 리텐션 대시보드 구축
  • A/B 테스트를 통한 개선 효과 검증
  • 세그먼트별 리텐션 패턴 심층 분석

📝 종합 결론

현재 상황 진단

던전 스토커즈의 D0 이탈률 {d0_rate:.1f}%는 게임 업계 평균(25-35%)을 크게 상회하는 Critical 수준입니다. 이는 서비스 지속성에 직접적인 위협이 되는 상황으로, 즉시 종합적인 개선 조치가 필요합니다.

핵심 개선 전략

  1. 첫날 경험 최적화: 온보딩, 튜토리얼, 초기 보상 시스템 전면 재설계
  2. 데이터 기반 개선: 긍정적 상관관계 지표들의 초기 경험 강화
  3. 위험 요소 제거: 부정적 영향 지표들의 영향도 최소화
  4. 지속적 모니터링: 실시간 리텐션 추적 및 조기 경보 시스템

기대 효과

본 분석 결과를 바탕으로 한 개선 조치 시행 시, D0 이탈률을 업계 평균 수준인 30% 이하로 개선할 수 있을 것으로 예상됩니다. 이는 월간 신규 유저 중 약 35% 추가 확보를 의미하며, 장기적으로 서비스 성장에 크게 기여할 것입니다.

""" # HTML 리포트 저장 output_file = f"{self.output_dir}/insights_report_{self.timestamp}.html" with open(output_file, 'w', encoding='utf-8') as f: f.write(html_content) self.log_message("REPORT", f"HTML 리포트 저장: {output_file}") # 콘솔에 요약만 출력 self.log_message("REPORT", "") self.log_message("REPORT", "="*60) self.log_message("REPORT", "RETENTION 분석 완료 - 주요 결과") self.log_message("REPORT", "="*60) self.log_message("REPORT", f"D0 이탈률: {d0_rate:.1f}% (매우 심각)") self.log_message("REPORT", f"D7+ 잔존률: {d7_plus_rate:.1f}%") if not self.correlation_results.empty: positive = self.correlation_results[ (self.correlation_results['correlation'] > 0) & (self.correlation_results['significant']) ].head(1) if len(positive) > 0: top_positive = positive.iloc[0] self.log_message("REPORT", f"최고 긍정 지표: {top_positive['metric']} ({top_positive['correlation']:.3f})") self.log_message("REPORT", "="*60) return html_content def run_full_analysis(self): """전체 분석 실행""" self.log_message("MAIN", "Retention 분석 시작") try: # 1. CSV 유효성 검사 self.validate_csv() # 2. 데이터 로드 self.load_data() # 3. 상관관계 분석 self.analyze_correlations() # 4. D0 이탈 분석 self.analyze_d0_churners() # 5. Feature Importance self.feature_importance_analysis() # 6. 인사이트 리포트 self.generate_insights_report() self.log_message("MAIN", "") self.log_message("MAIN", "SUCCESS: 모든 분석 완료!") self.log_message("MAIN", f"결과 파일들이 {self.output_dir}/ 디렉토리에 저장되었습니다.") except Exception as e: self.log_message("MAIN", f"분석 중 오류 발생: {str(e)}", "ERROR") raise def get_csv_path(): """사용자로부터 CSV 파일 경로 입력받기""" if len(sys.argv) > 1: return sys.argv[1] print("="*60) print("Retention Analysis Tool") print("="*60) print() while True: csv_path = input(">> 분석할 CSV 파일 경로를 입력하세요: ").strip() if not csv_path: print("ERROR: 경로를 입력해주세요.") continue # 따옴표 제거 csv_path = csv_path.strip('"\'') if not os.path.exists(csv_path): print(f"ERROR: 파일을 찾을 수 없습니다: {csv_path}") continue if not csv_path.lower().endswith('.csv'): print("ERROR: CSV 파일만 지원됩니다.") continue return csv_path def main(): """메인 실행 함수""" try: csv_path = get_csv_path() print(f"\n=== 분석 시작: {os.path.basename(csv_path)} ===") print("="*60) analyzer = RetentionAnalyzer(csv_path) analyzer.run_full_analysis() except KeyboardInterrupt: print("\n\nWARNING: 사용자에 의해 중단되었습니다.") except Exception as e: print(f"\n\nERROR: 오류 발생: {str(e)}") sys.exit(1) if __name__ == "__main__": main()