MCP Market Statistics Server

"""위험 평가 엔진 테스트""" import pytest import asyncio import time from datetime import datetime, timedelta from unittest.mock import AsyncMock, MagicMock, patch from typing import Dict, List, Any from src.ai.risk_assessment_engine import RiskAssessmentEngine from src.exceptions import ModelNotTrainedError, InsufficientDataError, PredictionError class TestRiskAssessmentEngine: """위험 평가 엔진 테스트""" @pytest.fixture def engine_config(self): """엔진 설정""" return { "risk_models": ["var", "cvar", "drawdown", "volatility", "beta"], "confidence_levels": [0.95, 0.99], "time_horizons": [1, 5, 10, 30], # days "lookback_period": 252, # trading days "monte_carlo_simulations": 10000, "risk_factors": { "market_risk": 0.4, "credit_risk": 0.2, "liquidity_risk": 0.15, "operational_risk": 0.15, "model_risk": 0.1 }, "stress_scenarios": ["market_crash", "interest_rate_shock", "currency_crisis"], "portfolio_constraints": { "max_position_size": 0.1, # 10% "max_sector_exposure": 0.3, # 30% "min_diversification_score": 0.7 } } @pytest.fixture def risk_engine(self, engine_config): """위험 평가 엔진 인스턴스""" return RiskAssessmentEngine(engine_config) @pytest.fixture def sample_portfolio_data(self): """샘플 포트폴리오 데이터""" return { "positions": [ { "symbol": "005930", # 삼성전자 "weight": 0.25, "market_value": 25000000, "sector": "technology", "beta": 1.2, "volatility": 0.25 }, { "symbol": "000660", # SK하이닉스 "weight": 0.20, "market_value": 20000000, "sector": "technology", "beta": 1.5, "volatility": 0.35 }, { "symbol": "035420", # NAVER "weight": 0.15, "market_value": 15000000, "sector": "internet", "beta": 1.1, "volatility": 0.30 }, { "symbol": "051910", # LG화학 "weight": 0.20, "market_value": 20000000, "sector": "chemicals", "beta": 0.9, "volatility": 0.28 }, { "symbol": "028260", # 삼성물산 "weight": 0.20, "market_value": 20000000, "sector": "construction", "beta": 0.8, "volatility": 0.22 } ], "total_value": 100000000, # 1억원 "currency": "KRW", "benchmark": "KOSPI", "creation_date": "2024-01-01" } @pytest.fixture def sample_market_data(self): """샘플 시장 데이터""" return { "005930": { "prices": [75000 + i * 100 for i in range(252)], # 1년치 데이터 "returns": [0.001 + i * 0.0001 for i in range(252)], "volumes": [1000000 + i * 1000 for i in range(252)] }, "000660": { "prices": [80000 + i * 150 for i in range(252)], "returns": [0.002 + i * 0.0002 for i in range(252)], "volumes": [800000 + i * 800 for i in range(252)] }, "KOSPI": { "prices": [2500 + i * 2 for i in range(252)], "returns": [0.0008 + i * 0.00008 for i in range(252)] } } def test_engine_initialization(self, risk_engine, engine_config): """엔진 초기화 테스트""" assert risk_engine.risk_models == engine_config["risk_models"] assert risk_engine.confidence_levels == engine_config["confidence_levels"] assert risk_engine.time_horizons == engine_config["time_horizons"] assert risk_engine.lookback_period == engine_config["lookback_period"] assert risk_engine.monte_carlo_simulations == engine_config["monte_carlo_simulations"] assert risk_engine.risk_factors == engine_config["risk_factors"] assert risk_engine.is_calibrated == False @pytest.mark.asyncio async def test_model_calibration(self, risk_engine, sample_market_data): """모델 캘리브레이션 테스트""" # 캘리브레이션 전 상태 assert risk_engine.is_calibrated == False # 모델 캘리브레이션 calibration_result = await risk_engine.calibrate_models(sample_market_data) # 캘리브레이션 후 상태 assert risk_engine.is_calibrated == True assert "calibrated_models" in calibration_result assert "calibration_metrics" in calibration_result assert "model_parameters" in calibration_result # 각 위험 모델이 캘리브레이션되었는지 확인 calibrated_models = calibration_result["calibrated_models"] for model in risk_engine.risk_models: assert model in calibrated_models @pytest.mark.asyncio async def test_var_calculation(self, risk_engine, sample_portfolio_data, sample_market_data): """VaR 계산 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # VaR 계산 var_results = await risk_engine.calculate_var( sample_portfolio_data, confidence_level=0.95, time_horizon=1 ) # 결과 검증 assert "var_amount" in var_results assert "var_percentage" in var_results assert "confidence_level" in var_results assert "time_horizon" in var_results assert "methodology" in var_results # VaR 값이 합리적인 범위인지 확인 var_amount = var_results["var_amount"] portfolio_value = sample_portfolio_data["total_value"] var_percentage = abs(var_amount) / portfolio_value assert 0 < var_percentage < 0.2 # 0-20% 범위 assert var_results["confidence_level"] == 0.95 assert var_results["time_horizon"] == 1 @pytest.mark.asyncio async def test_cvar_calculation(self, risk_engine, sample_portfolio_data, sample_market_data): """CVaR (Conditional VaR) 계산 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # CVaR 계산 cvar_results = await risk_engine.calculate_cvar( sample_portfolio_data, confidence_level=0.95, time_horizon=1 ) # 결과 검증 assert "cvar_amount" in cvar_results assert "cvar_percentage" in cvar_results assert "var_amount" in cvar_results # CVaR는 VaR도 포함 assert "expected_shortfall" in cvar_results # CVaR은 항상 VaR보다 크거나 같아야 함 assert abs(cvar_results["cvar_amount"]) >= abs(cvar_results["var_amount"]) @pytest.mark.asyncio async def test_maximum_drawdown_analysis(self, risk_engine, sample_portfolio_data, sample_market_data): """최대 낙폭 분석 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 최대 낙폭 분석 drawdown_results = await risk_engine.analyze_maximum_drawdown( sample_portfolio_data, sample_market_data ) # 결과 검증 assert "max_drawdown" in drawdown_results assert "max_drawdown_duration" in drawdown_results assert "recovery_time" in drawdown_results assert "drawdown_periods" in drawdown_results assert "current_drawdown" in drawdown_results # 낙폭은 음수이거나 0이어야 함 assert drawdown_results["max_drawdown"] <= 0 assert drawdown_results["max_drawdown_duration"] >= 0 @pytest.mark.asyncio async def test_volatility_analysis(self, risk_engine, sample_portfolio_data, sample_market_data): """변동성 분석 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 변동성 분석 volatility_results = await risk_engine.analyze_volatility( sample_portfolio_data, sample_market_data ) # 결과 검증 assert "portfolio_volatility" in volatility_results assert "annualized_volatility" in volatility_results assert "volatility_breakdown" in volatility_results assert "garch_forecast" in volatility_results assert "realized_volatility" in volatility_results # 변동성은 양수여야 함 assert volatility_results["portfolio_volatility"] > 0 assert volatility_results["annualized_volatility"] > 0 @pytest.mark.asyncio async def test_beta_analysis(self, risk_engine, sample_portfolio_data, sample_market_data): """베타 분석 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 베타 분석 beta_results = await risk_engine.analyze_beta( sample_portfolio_data, sample_market_data, benchmark="KOSPI" ) # 결과 검증 assert "portfolio_beta" in beta_results assert "systematic_risk" in beta_results assert "idiosyncratic_risk" in beta_results assert "correlation_with_market" in beta_results assert "r_squared" in beta_results # 베타 값이 합리적인 범위인지 확인 portfolio_beta = beta_results["portfolio_beta"] assert -2 < portfolio_beta < 3 # 일반적인 베타 범위 # R-squared는 0-1 범위 assert 0 <= beta_results["r_squared"] <= 1 @pytest.mark.asyncio async def test_stress_testing(self, risk_engine, sample_portfolio_data, sample_market_data): """스트레스 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 스트레스 테스트 실행 stress_results = await risk_engine.run_stress_tests( sample_portfolio_data, scenarios=["market_crash", "interest_rate_shock"] ) # 결과 검증 assert "stress_test_results" in stress_results assert "worst_case_scenario" in stress_results assert "scenario_rankings" in stress_results stress_test_results = stress_results["stress_test_results"] # 각 시나리오에 대한 결과 확인 for scenario in ["market_crash", "interest_rate_shock"]: assert scenario in stress_test_results scenario_result = stress_test_results[scenario] assert "portfolio_loss" in scenario_result assert "loss_percentage" in scenario_result assert "affected_positions" in scenario_result @pytest.mark.asyncio async def test_monte_carlo_simulation(self, risk_engine, sample_portfolio_data, sample_market_data): """몬테카를로 시뮬레이션 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 몬테카를로 시뮬레이션 mc_results = await risk_engine.run_monte_carlo_simulation( sample_portfolio_data, time_horizon=30, num_simulations=1000 # 테스트용으로 축소 ) # 결과 검증 assert "simulation_results" in mc_results assert "statistical_summary" in mc_results assert "percentiles" in mc_results assert "probability_of_loss" in mc_results # 시뮬레이션 결과 수 확인 simulation_results = mc_results["simulation_results"] assert len(simulation_results) == 1000 # 통계 요약 확인 statistical_summary = mc_results["statistical_summary"] assert "mean" in statistical_summary assert "std" in statistical_summary assert "min" in statistical_summary assert "max" in statistical_summary @pytest.mark.asyncio async def test_portfolio_optimization(self, risk_engine, sample_portfolio_data, sample_market_data): """포트폴리오 최적화 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 포트폴리오 최적화 optimization_results = await risk_engine.optimize_portfolio( sample_portfolio_data, objective="minimize_risk", constraints={"max_position_size": 0.3} ) # 결과 검증 assert "optimized_weights" in optimization_results assert "risk_reduction" in optimization_results assert "expected_return" in optimization_results assert "optimization_method" in optimization_results # 가중치 합이 1인지 확인 optimized_weights = optimization_results["optimized_weights"] total_weight = sum(optimized_weights.values()) assert abs(total_weight - 1.0) < 0.01 # 1에 가까움 # 제약조건 확인 for weight in optimized_weights.values(): assert weight <= 0.3 # 최대 포지션 크기 제약 @pytest.mark.asyncio async def test_risk_attribution(self, risk_engine, sample_portfolio_data, sample_market_data): """위험 기여도 분석 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 위험 기여도 분석 attribution_results = await risk_engine.analyze_risk_attribution( sample_portfolio_data, sample_market_data ) # 결과 검증 assert "position_contributions" in attribution_results assert "sector_contributions" in attribution_results assert "factor_contributions" in attribution_results assert "diversification_ratio" in attribution_results # 포지션별 기여도 확인 position_contributions = attribution_results["position_contributions"] for position in sample_portfolio_data["positions"]: symbol = position["symbol"] assert symbol in position_contributions assert "risk_contribution" in position_contributions[symbol] assert "marginal_risk" in position_contributions[symbol] @pytest.mark.asyncio async def test_liquidity_risk_assessment(self, risk_engine, sample_portfolio_data, sample_market_data): """유동성 위험 평가 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 유동성 위험 평가 liquidity_results = await risk_engine.assess_liquidity_risk( sample_portfolio_data, sample_market_data ) # 결과 검증 assert "liquidity_score" in liquidity_results assert "time_to_liquidate" in liquidity_results assert "liquidity_cost" in liquidity_results assert "position_liquidity" in liquidity_results # 유동성 점수는 0-1 범위 liquidity_score = liquidity_results["liquidity_score"] assert 0 <= liquidity_score <= 1 # 청산 시간은 양수 time_to_liquidate = liquidity_results["time_to_liquidate"] assert time_to_liquidate > 0 @pytest.mark.asyncio async def test_concentration_risk_analysis(self, risk_engine, sample_portfolio_data): """집중도 위험 분석 테스트""" # 집중도 위험 분석 concentration_results = await risk_engine.analyze_concentration_risk(sample_portfolio_data) # 결과 검증 assert "herfindahl_index" in concentration_results assert "max_position_weight" in concentration_results assert "sector_concentration" in concentration_results assert "concentration_score" in concentration_results assert "diversification_recommendations" in concentration_results # 허핀달 지수는 0-1 범위 herfindahl_index = concentration_results["herfindahl_index"] assert 0 < herfindahl_index <= 1 # 최대 포지션 비중 확인 max_position_weight = concentration_results["max_position_weight"] actual_max_weight = max(pos["weight"] for pos in sample_portfolio_data["positions"]) assert abs(max_position_weight - actual_max_weight) < 0.01 @pytest.mark.asyncio async def test_risk_reporting(self, risk_engine, sample_portfolio_data, sample_market_data): """위험 보고서 생성 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 종합 위험 보고서 생성 risk_report = await risk_engine.generate_risk_report( sample_portfolio_data, sample_market_data, report_date=datetime.now() ) # 보고서 구조 확인 assert "executive_summary" in risk_report assert "var_analysis" in risk_report assert "stress_test_results" in risk_report assert "risk_attribution" in risk_report assert "recommendations" in risk_report assert "report_metadata" in risk_report # 임원 요약 확인 executive_summary = risk_report["executive_summary"] assert "overall_risk_level" in executive_summary assert "key_risk_factors" in executive_summary assert "immediate_actions" in executive_summary @pytest.mark.asyncio async def test_real_time_risk_monitoring(self, risk_engine, sample_portfolio_data, sample_market_data): """실시간 위험 모니터링 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 실시간 시장 데이터 시뮬레이션 real_time_data = { "005930": {"price": 77000, "volume": 1500000, "timestamp": datetime.now().isoformat()}, "000660": {"price": 82000, "volume": 1200000, "timestamp": datetime.now().isoformat()} } # 실시간 위험 모니터링 monitoring_results = await risk_engine.monitor_real_time_risk( sample_portfolio_data, real_time_data ) # 결과 검증 assert "current_var" in monitoring_results assert "risk_alerts" in monitoring_results assert "position_changes" in monitoring_results assert "monitoring_timestamp" in monitoring_results # 위험 알림 구조 확인 risk_alerts = monitoring_results["risk_alerts"] for alert in risk_alerts: assert "alert_type" in alert assert "severity" in alert assert "message" in alert @pytest.mark.asyncio async def test_backtesting_validation(self, risk_engine, sample_portfolio_data, sample_market_data): """백테스팅 검증 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 백테스팅 실행 backtest_results = await risk_engine.run_backtest_validation( sample_portfolio_data, sample_market_data, start_date="2023-01-01", end_date="2023-12-31" ) # 결과 검증 assert "var_accuracy" in backtest_results assert "kupiec_test" in backtest_results assert "christoffersen_test" in backtest_results assert "model_performance" in backtest_results # VaR 정확도 확인 var_accuracy = backtest_results["var_accuracy"] assert "hit_rate" in var_accuracy assert "expected_exceptions" in var_accuracy assert "actual_exceptions" in var_accuracy @pytest.mark.asyncio async def test_error_handling(self, risk_engine): """오류 처리 테스트""" # 캘리브레이션되지 않은 모델로 계산 시도 with pytest.raises(ModelNotTrainedError): await risk_engine.calculate_var({}, confidence_level=0.95) # 불충분한 데이터로 캘리브레이션 시도 insufficient_data = {"symbol": [100, 101, 102]} # 너무 적은 데이터 with pytest.raises(InsufficientDataError): await risk_engine.calibrate_models(insufficient_data) # 잘못된 신뢰수준 await risk_engine.calibrate_models({"symbol": list(range(300))}) with pytest.raises(ValueError): await risk_engine.calculate_var({}, confidence_level=1.5) # > 1 @pytest.mark.asyncio async def test_performance_benchmarking(self, risk_engine, sample_portfolio_data, sample_market_data): """성능 벤치마킹 테스트""" # 모델 캘리브레이션 await risk_engine.calibrate_models(sample_market_data) # 대량 계산 성능 측정 start_time = time.time() # 여러 계산을 병렬로 실행 tasks = [ risk_engine.calculate_var(sample_portfolio_data, 0.95, 1), risk_engine.calculate_cvar(sample_portfolio_data, 0.95, 1), risk_engine.analyze_volatility(sample_portfolio_data, sample_market_data) ] results = await asyncio.gather(*tasks) end_time = time.time() # 성능 확인 processing_time = end_time - start_time assert processing_time < 5.0 # 5초 이내 assert len(results) == 3 # 각 결과가 유효한지 확인 for result in results: assert result is not None assert isinstance(result, dict) def test_risk_metrics_calculation(self, risk_engine): """위험 메트릭 계산 테스트""" # 샘플 수익률 데이터 returns = [0.01, -0.02, 0.015, -0.008, 0.003, -0.012, 0.007, -0.005] # 표준편차 계산 volatility = risk_engine._calculate_volatility(returns) assert volatility > 0 # VaR 계산 (히스토리컬 방법) var_95 = risk_engine._calculate_historical_var(returns, 0.95) var_99 = risk_engine._calculate_historical_var(returns, 0.99) # 99% VaR이 95% VaR보다 크거나 같아야 함 (절댓값) assert abs(var_99) >= abs(var_95) # 샤프 비율 계산 sharpe_ratio = risk_engine._calculate_sharpe_ratio(returns, risk_free_rate=0.02) assert isinstance(sharpe_ratio, (int, float)) def test_correlation_analysis(self, risk_engine): """상관관계 분석 테스트""" # 샘플 수익률 데이터 returns_a = [0.01, -0.02, 0.015, -0.008, 0.003] returns_b = [0.005, -0.015, 0.012, -0.006, 0.002] # 상관계수 계산 correlation = risk_engine._calculate_correlation(returns_a, returns_b) # 상관계수는 -1과 1 사이 assert -1 <= correlation <= 1 # 공분산 계산 covariance = risk_engine._calculate_covariance(returns_a, returns_b) assert isinstance(covariance, (int, float))