리드 데이터 사이언티스트 면접 질문

# 예시: 멘토링 추적 시스템
class MentorshipTracker:
    def __init__(self):
        self.mentees = {}
    
    def create_development_plan(self, mentee_name, current_level, target_level, timeline_months):
        """개인 개발 계획 생성"""
        plan = {
            'mentee': mentee_name,
            'current_level': current_level,
            'target_level': target_level,
            'timeline_months': timeline_months,
            'skills_to_develop': [],
            'projects': [],
            'milestones': [],
            'check_ins': []
        }
        self.mentees[mentee_name] = plan
        return plan
    
    def add_skill_goal(self, mentee_name, skill, current_proficiency, target_proficiency):
        """기술 개발 목표 추가"""
        goal = {
            'skill': skill,
            'current': current_proficiency,
            'target': target_proficiency,
            'resources': [],
            'progress': 0
        }
        self.mentees[mentee_name]['skills_to_develop'].append(goal)
    
    def log_check_in(self, mentee_name, date, notes, progress_rating):
        """멘토링 점검 기록"""
        check_in = {
            'date': date,
            'notes': notes,
            'progress_rating': progress_rating,
            'action_items': []
        }
        self.mentees[mentee_name]['check_ins'].append(check_in)
    
    def get_progress_summary(self, mentee_name):
        """멘티 진행 상황 요약 가져오기"""
        plan = self.mentees[mentee_name]
        avg_progress = sum(s['progress'] for s in plan['skills_to_develop']) / len(plan['skills_to_develop'])
        return {
            'mentee': mentee_name,
            'overall_progress': avg_progress,
            'skills_in_progress': len([s for s in plan['skills_to_develop'] if 0 < s['progress'] < 100]),
            'skills_completed': len([s for s in plan['skills_to_develop'] if s['progress'] == 100]),
            'total_check_ins': len(plan['check_ins'])
        }

# 예시 사용법
tracker = MentorshipTracker()
tracker.create_development_plan('Alice', '주니어 DS', '미드 레벨 DS', 12)
tracker.add_skill_goal('Alice', '딥러닝', current_proficiency=2, target_proficiency=4)
tracker.add_skill_goal('Alice', '모델 배포', current_proficiency=1, target_proficiency=3)
tracker.log_check_in('Alice', '2025-01-15', 'PyTorch 과정 완료', progress_rating=7)

summary = tracker.get_progress_summary('Alice')
print(f"진행 상황 요약: {summary}")

# 예시: ML 플랫폼 아키텍처 설계
class MLPlatformArchitecture:
    def __init__(self):
        self.components = {
            'data_layer': {
                'data_lake': 'S3/GCS/Azure Blob',
                'data_warehouse': 'Snowflake/BigQuery/Redshift',
                'feature_store': 'Feast/Tecton',
                'data_catalog': 'DataHub/Amundsen'
            },
            'compute_layer': {
                'training': 'SageMaker/Vertex AI/AzureML',
                'batch_inference': 'Spark/Dask',
                'real_time_serving': 'Kubernetes/SageMaker Endpoints'
            },
            'ml_ops': {
                'experiment_tracking': 'MLflow/W&B',
                'model_registry': 'MLflow/SageMaker',
                'pipeline_orchestration': 'Airflow/Kubeflow',
                'monitoring': 'Prometheus/Grafana'
            },
            'governance': {
                'version_control': 'Git/DVC',
                'access_control': 'IAM/RBAC',
                'audit_logging': 'CloudTrail/StackDriver'
            }
        }
    
    def design_for_scale(self, expected_models, expected_requests_per_sec):
        """규모 요구 사항에 따른 아키텍처 설계"""
        recommendations = []
        
        if expected_models > 100:
            recommendations.append("자동화된 모델 레지스트리 및 수명 주기 관리 구현")
            recommendations.append("중복성을 줄이기 위해 특징 저장소 사용")
        
        if expected_requests_per_sec > 1000:
            recommendations.append("자동 스케일링을 사용하여 Kubernetes에 모델 배포")
            recommendations.append("빈번한 예측을 위한 캐싱 레이어 구현")
            recommendations.append("모델 서빙 최적화 (TensorRT, ONNX) 사용")
        
        return recommendations
    
    def estimate_costs(self, team_size, data_volume_tb, model_count):
        """인프라 비용 추정"""
        # 간소화된 비용 추정
        costs = {
            'storage': data_volume_tb * 23,  # $23/TB/월
            'compute_training': model_count * 500,  # $500/모델/월
            'compute_serving': model_count * 200,  # $200/모델/월
            'tools_licenses': team_size * 100,  # $100/인/월
        }
        costs['total'] = sum(costs.values())
        return costs

# 예시 사용법
platform = MLPlatformArchitecture()
recommendations = platform.design_for_scale(expected_models=150, expected_requests_per_sec=2000)
print("아키텍처 권장 사항:")
for rec in recommendations:
    print(f"- {rec}")

costs = platform.estimate_costs(team_size=10, data_volume_tb=50, model_count=20)
print(f"\n예상 월별 비용: ${costs['total']:,.0f}")

import pandas as pd
import numpy as np

class ProjectPrioritization:
    def __init__(self):
        self.projects = []
    
    def add_project(self, name, business_value, feasibility, urgency, 
                   resource_requirement, strategic_alignment):
        """채점 기준을 사용하여 프로젝트 추가"""
        project = {
            'name': name,
            'business_value': business_value,  # 1-10
            'feasibility': feasibility,  # 1-10
            'urgency': urgency,  # 1-10
            'resource_requirement': resource_requirement,  # 1-10 (낮을수록 좋음)
            'strategic_alignment': strategic_alignment  # 1-10
        }
        
        # 우선 순위 점수 계산
        # 더 높은 비즈니스 가치, 타당성, 긴급성 및 정렬은 점수를 높입니다.
        # 더 높은 리소스 요구 사항은 점수를 낮춥니다.
        project['priority_score'] = (
            business_value * 0.3 +
            feasibility * 0.25 +
            urgency * 0.2 +
            strategic_alignment * 0.15 +
            (10 - resource_requirement) * 0.1
        )
        
        self.projects.append(project)
    
    def get_prioritized_projects(self):
        """우선 순위별로 정렬된 프로젝트 반환"""
        df = pd.DataFrame(self.projects)
        return df.sort_values('priority_score', ascending=False)
    
    def allocate_resources(self, available_capacity):
        """우선 순위 및 용량에 따라 리소스 할당"""
        df = self.get_prioritized_projects()
        allocated = []
        remaining_capacity = available_capacity
        
        for _, project in df.iterrows():
            if project['resource_requirement'] <= remaining_capacity:
                allocated.append(project['name'])
                remaining_capacity -= project['resource_requirement']
        
        return {
            'allocated_projects': allocated,
            'capacity_used': available_capacity - remaining_capacity,
            'capacity_remaining': remaining_capacity
        }

# 예시 사용법
prioritizer = ProjectPrioritization()

# 프로젝트 추가
prioritizer.add_project(
    name='고객 이탈 예측',
    business_value=9,
    feasibility=8,
    urgency=7,
    resource_requirement=5,
    strategic_alignment=9
)

prioritizer.add_project(
    name='추천 시스템',
    business_value=8,
    feasibility=6,
    urgency=5,
    resource_requirement=8,
    strategic_alignment=7
)

prioritizer.add_project(
    name='사기 탐지',
    business_value=10,
    feasibility=7,
    urgency=9,
    resource_requirement=6,
    strategic_alignment=10
)

prioritizer.add_project(
    name='수요 예측',
    business_value=7,
    feasibility=9,
    urgency=6,
    resource_requirement=4,
    strategic_alignment=6
)

# 우선 순위 목록 가져오기
prioritized = prioritizer.get_prioritized_projects()
print("우선 순위 지정된 프로젝트:")
print(prioritized[['name', 'priority_score', 'business_value', 'feasibility']].to_string(index=False))

# 리소스 할당 (용량 15 단위 가정)
allocation = prioritizer.allocate_resources(available_capacity=15)
print(f"\n할당된 프로젝트: {allocation['allocated_projects']}")
print(f"사용된 용량: {allocation['capacity_used']}/15")

class StakeholderPresentation:
    def __init__(self, project_name, audience_type):
        self.project_name = project_name
        self.audience_type = audience_type
    
    def create_executive_summary(self, business_problem, solution, impact, timeline, investment):
        """임원에게 친숙한 요약 생성"""
        summary = f"""
# {self.project_name}

## 비즈니스 문제
{business_problem}

## 제안된 솔루션
{solution}

## 예상되는 영향
{impact}

## 타임라인 및 투자
- 타임라인: {timeline}
- 필요한 투자: {investment}

## 주요 위험 및 완화
- 데이터 품질: 데이터 유효성 검사 파이프라인 구현
- 모델 정확도: 기준선 및 성공 지표 설정
- 채택: 사용자 교육 및 변경 관리 계획
"""
        return summary
    
    def create_technical_analogy(self, ml_concept):
        """ML 개념에 대한 비기술적 비유 생성"""
        analogies = {
            'supervised_learning': "플래시카드로 아이를 가르치는 것과 같습니다. 패턴을 배울 때까지 정답이 있는 예시를 보여줍니다.",
            'neural_network': "각 레이어가 다른 측면에 집중하고 다음 레이어로 정보를 전달하는 전문가 팀과 같습니다.",
            'overfitting': "개념을 이해하는 대신 답을 암기하는 것과 같습니다. 연습 시험에서는 작동하지만 새로운 질문에서는 실패합니다.",
            'feature_engineering': "레시피에 적합한 재료를 선택하는 것과 같습니다. 입력 품질이 출력 품질을 결정합니다.",
            'model_deployment': "새로운 제품을 출시하는 것과 같습니다. 철저히 테스트하고 점진적으로 출시하며 성능을 모니터링합니다."
        }
        return analogies.get(ml_concept, "개념을 찾을 수 없습니다.")
    
    def quantify_business_impact(self, metric_name, baseline, predicted, confidence_interval):
        """불확실성을 고려하여 비즈니스 영향 제시"""
        improvement = ((predicted - baseline) / baseline) * 100
        
        report = f"""
## 비즈니스 영향: {metric_name}

- **현재 상태**: {baseline:,.0f}
- **예측된 상태**: {predicted:,.0f}
- **예상되는 개선**: {improvement:.1f}%
- **신뢰 구간**: {confidence_interval[0]:.1f}% ~ {confidence_interval[1]:.1f}%

이 개선은 모델 불확실성을 고려하면서 상당한 비즈니스 가치로 이어집니다.
"""
        return report

# 예시 사용법
presentation = StakeholderPresentation("고객 이탈 예측", "임원")

summary = presentation.create_executive_summary(
    business_problem="연간 고객의 15%를 잃고 있으며, 이는 5백만 달러의 수익 손실을 초래합니다.",
    solution="이탈 위험을 3개월 전에 예측하여 사전 예방적 유지를 가능하게 하는 ML 모델",
    impact="이탈률을 20-30% 줄여 연간 1백만~150만 달러 절감",
    timeline="3개월 개발, 1개월 파일럿, 2개월 출시",
    investment="20만 달러 (팀 시간 + 인프라)"
)

print(summary)

analogy = presentation.create_technical_analogy('overfitting')
print(f"\n과적합 비유: {analogy}")

impact = presentation.quantify_business_impact(
    metric_name="고객 유지율",
    baseline=85,
    predicted=88,
    confidence_interval=(87.5, 88.5)
)
print(impact)

import numpy as np
from sklearn.metrics import confusion_matrix

class FairnessAuditor:
    def __init__(self, model, sensitive_attribute):
        self.model = model
        self.sensitive_attribute = sensitive_attribute
    
    def calculate_demographic_parity(self, X, y_pred, sensitive_feature):
        """인구 통계적 패리티 차이 계산"""
        groups = np.unique(sensitive_feature)
        positive_rates = {}
        
        for group in groups:
            group_mask = sensitive_feature == group
            positive_rate = np.mean(y_pred[group_mask])
            positive_rates[group] = positive_rate
        
        max_rate = max(positive_rates.values())
        min_rate = min(positive_rates.values())
        parity_difference = max_rate - min_rate
        
        return {
            'positive_rates': positive_rates,
            'parity_difference': parity_difference,
            'is_fair': parity_difference < 0.1  # 10% 임계값
        }
    
    def calculate_equal_opportunity(self, y_true, y_pred, sensitive_feature):
        """동등한 기회 차이 계산 (그룹 간 TPR)"""
        groups = np.unique(sensitive_feature)
        tpr_by_group = {}
        
        for group in groups:
            group_mask = sensitive_feature == group
            y_true_group = y_true[group_mask]
            y_pred_group = y_pred[group_mask]
            
            # 진양성률
            tp = np.sum((y_true_group == 1) & (y_pred_group == 1))
            fn = np.sum((y_true_group == 1) & (y_pred_group == 0))
            tpr = tp / (tp + fn) if (tp + fn) > 0 else 0
            
            tpr_by_group[group] = tpr
        
        max_tpr = max(tpr_by_group.values())
        min_tpr = min(tpr_by_group.values())
        eo_difference = max_tpr - min_tpr
        
        return {
            'tpr_by_group': tpr_by_group,
            'equal_opportunity_difference': eo_difference,
            'is_fair': eo_difference < 0.1
        }
    
    def generate_fairness_report(self, X, y_true, y_pred, sensitive_feature):
        """포괄적인 공정성 보고서 생성"""
        dp = self.calculate_demographic_parity(X, y_pred, sensitive_feature)
        eo = self.calculate_equal_opportunity(y_true, y_pred, sensitive_feature)
        
        report = f"""
# 공정성 감사 보고서

## 인구 통계적 패리티
- 그룹별 양성률: {dp['positive_rates']}
- 패리티 차이: {dp['parity_difference']:.3f}
- 공정함 (< 0.1): {dp['is_fair']}

## 동등한 기회
- 그룹별 TPR: {eo['tpr_by_group']}
- EO 차이: {eo['equal_opportunity_difference']:.3f}
- 공정함 (< 0.1): {eo['is_fair']}

## 권장 사항
"""
        if not dp['is_fair']:
            report += "- 재샘플링 또는 임계값 조정을 통해 인구 통계적 패리티 문제 해결\n"
        if not eo['is_fair']:
            report += "- 보정 또는 공정성 제약을 통해 동등한 기회 문제 해결\n"
        
        return report

# 예시 사용법
np.random.seed(42)
n_samples = 1000

# 편향이 있는 데이터 시뮬레이션
sensitive_feature = np.random.choice(['그룹 A', '그룹 B'], n_samples, p=[0.7, 0.3])
y_true = np.random.randint(0, 2, n_samples)

# 편향된 예측 시뮬레이션 (그룹 B는 양성률이 낮음)
y_pred = y_true.copy()
group_b_mask = sensitive_feature == '그룹 B'
y_pred[group_b_mask] = np.where(
    np.random.random(np.sum(group_b_mask)) < 0.3,
    1 - y_pred[group_b_mask],
    y_pred[group_b_mask]
)

auditor = FairnessAuditor(model=None, sensitive_attribute='group')
report = auditor.generate_fairness_report(None, y_true, y_pred, sensitive_feature)
print(report)

class DataScienceRoadmap:
    def __init__(self, company_name):
        self.company_name = company_name
        self.initiatives = {
            'quick_wins': [],
            'medium_term': [],
            'long_term': []
        }
        self.capabilities = []
    
    def add_initiative(self, name, timeline, business_value, dependencies, resources):
        """로드맵에 이니셔티브 추가"""
        initiative = {
            'name': name,
            'timeline': timeline,
            'business_value': business_value,
            'dependencies': dependencies,
            'resources': resources,
            'status': 'planned'
        }
        
        if timeline <= 6:
            self.initiatives['quick_wins'].append(initiative)
        elif timeline <= 12:
            self.initiatives['medium_term'].append(initiative)
        else:
            self.initiatives['long_term'].append(initiative)
    
    def add_capability(self, capability_name, current_level, target_level, timeline):
        """기능 개발 목표 추가"""
        capability = {
            'name': capability_name,
            'current_level': current_level,  # 1-5
            'target_level': target_level,  # 1-5
            'timeline': timeline,
            'gap': target_level - current_level
        }
        self.capabilities.append(capability)
    
    def generate_roadmap_summary(self):
        """로드맵 요약 생성"""
        summary = f"""
# {self.company_name} 데이터 과학 로드맵

## 빠른 성공 (0~6개월)
"""
        for init in self.initiatives['quick_wins']:
            summary += f"- **{init['name']}**: {init['business_value']}\n"
        
        summary += "\n## 중간 규모 (6~12개월)\n"
        for init in self.initiatives['medium_term']:
            summary += f"- **{init['name']}**: {init['business_value']}\n"
        
        summary += "\n## 장기 (12개월 이상)\n"
        for init in self.initiatives['long_term']:
            summary += f"- **{init['name']}**: {init['business_value']}\n"
        
        summary += "\n## 기능 개발\n"
        for cap in sorted(self.capabilities, key=lambda x: x['gap'], reverse=True):
            summary += f"- **{cap['name']}**: 레벨 {cap['current_level']} → {cap['target_level']} ({cap['timeline']}개월)\n"
        
        return summary

# 예시 사용법
roadmap = DataScienceRoadmap("TechCorp")

# 빠른 성공 추가
roadmap.add_initiative(
    name="고객 이탈 대시보드",
    timeline=3,
    business_value="이탈 동인에 대한 가시성, 사전 예방적 유지 가능",
    dependencies=[],
    resources="1 DS, 1 분석가"
)

# 중간 규모 이니셔티브 추가
roadmap.add_initiative(
    name="개인화 엔진",
    timeline=9,
    business_value="전환율 15% 증가, 사용자 경험 개선",
    dependencies=["특징 저장소", "A/B 테스트 프레임워크"],
    resources="2 DS, 1 MLE"
)

# 장기 이니셔티브 추가
roadmap.add_initiative(
    name="실시간 추천 시스템",
    timeline=18,
    business_value="실시간 개인화, 경쟁 우위",
    dependencies=["스트리밍 인프라", "모델 서빙 플랫폼"],
    resources="3 DS, 2 MLE, 1 DE"
)

# 기능 추가
roadmap.add_capability("MLOps 성숙도", current_level=2, target_level=4, timeline=12)
roadmap.add_capability("실시간 ML", current_level=1, target_level=3, timeline=18)
roadmap.add_capability("NLP 기능", current_level=2, target_level=4, timeline=15)

print(roadmap.generate_roadmap_summary())