"""AI 시스템 기본 테스트 (의존성 최소화)"""
import pytest
import asyncio
import json
import time
from datetime import datetime, timedelta
from unittest.mock import AsyncMock, MagicMock, patch
from typing import Dict, List, Any
from src.ai.price_predictor import PricePredictor
from src.exceptions import ModelNotTrainedError, InsufficientDataError, PredictionError
class MockDataFrame:
"""Pandas DataFrame 모킹"""
def __init__(self, data: Dict[str, List]):
self.data = data.copy() if data else {}
self._columns = list(self.data.keys())
self._length = len(next(iter(self.data.values()))) if self.data else 0
def __len__(self):
return self._length
def __getitem__(self, key):
if isinstance(key, str):
if key in self.data:
return MockSeries(self.data[key])
else:
return MockSeries([0] * self._length) # 기본값
return self
def __setitem__(self, key, value):
"""컬럼 할당 지원"""
if isinstance(value, MockSeries):
self.data[key] = value.data
elif isinstance(value, list):
self.data[key] = value
else:
self.data[key] = [value] * self._length
if key not in self._columns:
self._columns.append(key)
@property
def columns(self):
return self._columns
def copy(self):
return MockDataFrame(self.data.copy())
def dropna(self):
return self
def pct_change(self):
return MockSeries([0.01] * self._length)
def rolling(self, window):
return MockRolling(self.data)
def ewm(self, span):
return MockEWM(self.data)
@property
def iloc(self):
return MockILoc(self.data)
@property
def values(self):
return [[self.data[col][i] for col in self._columns] for i in range(self._length)]
class MockSeries:
"""Pandas Series 모킹"""
def __init__(self, data: List):
self.data = data
self._length = len(data)
def __len__(self):
return self._length
def __getitem__(self, index):
if isinstance(index, int):
return self.data[index] if 0 <= index < self._length else None
return MockSeries(self.data[index])
def __add__(self, other):
if isinstance(other, MockSeries):
return MockSeries([a + b for a, b in zip(self.data, other.data)])
else:
return MockSeries([a + other for a in self.data])
def __sub__(self, other):
if isinstance(other, MockSeries):
return MockSeries([a - b for a, b in zip(self.data, other.data)])
else:
return MockSeries([a - other for a in self.data])
def __mul__(self, other):
if isinstance(other, MockSeries):
return MockSeries([a * b for a, b in zip(self.data, other.data)])
else:
return MockSeries([a * other for a in self.data])
def __truediv__(self, other):
if isinstance(other, MockSeries):
return MockSeries([a / b if b != 0 else 0 for a, b in zip(self.data, other.data)])
else:
return MockSeries([a / other if other != 0 else 0 for a in self.data])
@property
def iloc(self):
return MockILoc({"data": self.data})["data"]
@property
def values(self):
return self.data
def pct_change(self):
return MockSeries([0.01] * self._length)
def rolling(self, window):
return MockRolling({"data": self.data})
def ewm(self, span):
return MockEWM({"data": self.data})
def diff(self):
return MockSeries([0.1] * self._length)
def where(self, condition, other):
return MockSeries([other if not condition else x for x in self.data])
class MockRolling:
"""Pandas Rolling 모킹"""
def __init__(self, data: Dict):
self.data = data
def mean(self):
if "data" in self.data:
return MockSeries([75000] * len(self.data["data"]))
key = next(iter(self.data.keys()))
return MockSeries([75000] * len(self.data[key]))
def std(self):
if "data" in self.data:
return MockSeries([1000] * len(self.data["data"]))
key = next(iter(self.data.keys()))
return MockSeries([1000] * len(self.data[key]))
class MockEWM:
"""Pandas EWM 모킹"""
def __init__(self, data: Dict):
self.data = data
def mean(self):
if "data" in self.data:
return MockSeries([75000] * len(self.data["data"]))
key = next(iter(self.data.keys()))
return MockSeries([75000] * len(self.data[key]))
class MockILoc:
"""Pandas iloc 모킹"""
def __init__(self, data: Dict):
self.data = data
def __getitem__(self, key):
if isinstance(key, int):
# 단일 인덱스
if self.data:
first_key = next(iter(self.data.keys()))
if key >= 0 and key < len(self.data[first_key]):
return self.data[first_key][key]
elif key < 0 and abs(key) <= len(self.data[first_key]):
return self.data[first_key][key] # 음수 인덱스 지원
return 0 # 기본값
elif isinstance(key, slice):
# 슬라이스
result_data = {}
for col, values in self.data.items():
result_data[col] = values[key]
return MockDataFrame(result_data)
return self
class TestPricePredictor:
"""가격 예측 모델 테스트 (기본)"""
@pytest.fixture
def model_config(self):
"""모델 설정"""
return {
"model_type": "lstm",
"sequence_length": 60,
"prediction_horizon": 5,
"features": ["price", "volume", "rsi", "macd", "bollinger_bands"],
"epochs": 100,
"batch_size": 32,
"validation_split": 0.2,
"early_stopping": True,
"learning_rate": 0.001
}
@pytest.fixture
def price_predictor(self, model_config):
"""가격 예측기 인스턴스"""
return PricePredictor(model_config)
@pytest.fixture
def sample_market_data(self):
"""샘플 시장 데이터 (Mock DataFrame)"""
data_length = 100
return MockDataFrame({
'timestamp': [f"2024-01-{i+1:02d}" for i in range(data_length)],
'symbol': ['005930'] * data_length,
'price': [75000 + i * 10 for i in range(data_length)],
'volume': [1000000 + i * 1000 for i in range(data_length)],
'high': [75500 + i * 10 for i in range(data_length)],
'low': [74500 + i * 10 for i in range(data_length)],
'open': [75000 + i * 10 for i in range(data_length)],
'close': [75000 + i * 10 for i in range(data_length)]
})
def test_predictor_initialization(self, price_predictor, model_config):
"""예측기 초기화 테스트"""
assert price_predictor.model_type == model_config["model_type"]
assert price_predictor.sequence_length == model_config["sequence_length"]
assert price_predictor.prediction_horizon == model_config["prediction_horizon"]
assert price_predictor.features == model_config["features"]
assert price_predictor.is_trained == False
assert price_predictor.model is None
@pytest.mark.asyncio
async def test_feature_engineering(self, price_predictor, sample_market_data):
"""피처 엔지니어링 테스트"""
features = await price_predictor.engineer_features(sample_market_data)
# 기본 피처 확인
assert 'price' in features.columns
assert 'volume' in features.columns
# 기술적 지표 확인
assert 'rsi' in features.columns
assert 'macd' in features.columns
assert 'bollinger_upper' in features.columns
assert 'bollinger_lower' in features.columns
assert 'sma_20' in features.columns
assert 'ema_12' in features.columns
# 데이터 길이 확인
assert len(features) > 0
@pytest.mark.asyncio
async def test_data_preprocessing(self, price_predictor, sample_market_data):
"""데이터 전처리 테스트"""
features = await price_predictor.engineer_features(sample_market_data)
X, y = await price_predictor.preprocess_data(features)
# 시퀀스 데이터 형태 확인
assert len(X) > 0
assert len(y) > 0
assert len(X) == len(y)
# 시퀀스 길이 확인
assert len(X[0]) == price_predictor.sequence_length
assert len(y[0]) == price_predictor.prediction_horizon
# 피처 수 확인 (available features 수와 매칭)
assert len(X[0][0]) > 0 # 최소 1개 이상의 피처
assert len(X[0][0]) >= len([f for f in price_predictor.features if f != "bollinger_bands"]) + 2 # bollinger_bands는 2개로 분리
@pytest.mark.asyncio
async def test_model_training(self, price_predictor, sample_market_data):
"""모델 훈련 테스트"""
features = await price_predictor.engineer_features(sample_market_data)
X, y = await price_predictor.preprocess_data(features)
# 훈련 전 상태 확인
assert price_predictor.is_trained == False
# 모델 훈련
training_history = await price_predictor.train(X, y)
# 훈련 후 상태 확인
assert price_predictor.is_trained == True
assert price_predictor.model is not None
assert 'history' in training_history
assert 'loss' in training_history['history']
assert 'val_loss' in training_history['history']
# 훈련 품질 확인 (손실 감소)
loss_history = training_history['history']['loss']
assert loss_history[0] > loss_history[-1] # 손실 감소
@pytest.mark.asyncio
async def test_price_prediction(self, price_predictor, sample_market_data):
"""가격 예측 테스트"""
# 모델 훈련
features = await price_predictor.engineer_features(sample_market_data)
X, y = await price_predictor.preprocess_data(features)
await price_predictor.train(X, y)
# 최근 데이터로 예측
recent_data = MockDataFrame({
'price': [76000, 76100, 76200, 76300, 76400],
'volume': [1100000, 1110000, 1120000, 1130000, 1140000]
})
predictions = await price_predictor.predict(recent_data)
# 예측 결과 확인
assert 'predictions' in predictions
assert 'confidence_intervals' in predictions
assert 'probability_distribution' in predictions
pred_values = predictions['predictions']
assert len(pred_values) == price_predictor.prediction_horizon
assert all(isinstance(p, (int, float)) for p in pred_values)
# 신뢰구간 확인
ci = predictions['confidence_intervals']
assert 'lower' in ci and 'upper' in ci
assert len(ci['lower']) == price_predictor.prediction_horizon
assert all(ci['lower'][i] <= pred_values[i] <= ci['upper'][i] for i in range(len(pred_values)))
@pytest.mark.asyncio
async def test_batch_prediction(self, price_predictor, sample_market_data):
"""배치 예측 테스트"""
# 모델 훈련
features = await price_predictor.engineer_features(sample_market_data)
X, y = await price_predictor.preprocess_data(features)
await price_predictor.train(X, y)
# 여러 심볼 데이터
symbols = ['005930', '000660', '035420']
batch_data = []
for symbol in symbols:
symbol_data = MockDataFrame({
'symbol': [symbol] * 10,
'price': [75000 + i * 10 for i in range(10)],
'volume': [1000000 + i * 1000 for i in range(10)]
})
batch_data.append(symbol_data)
# 배치 예측
batch_predictions = await price_predictor.predict_batch(batch_data)
# 결과 확인
assert len(batch_predictions) == len(symbols)
for symbol, prediction in batch_predictions.items():
assert symbol in symbols
assert 'predictions' in prediction
assert len(prediction['predictions']) == price_predictor.prediction_horizon
@pytest.mark.asyncio
async def test_model_evaluation(self, price_predictor, sample_market_data):
"""모델 평가 테스트"""
features = await price_predictor.engineer_features(sample_market_data)
X, y = await price_predictor.preprocess_data(features)
# 훈련/테스트 분할
split_idx = int(len(X) * 0.8)
X_train, X_test = X[:split_idx], X[split_idx:]
y_train, y_test = y[:split_idx], y[split_idx:]
# 모델 훈련
await price_predictor.train(X_train, y_train)
# 모델 평가
evaluation_metrics = await price_predictor.evaluate(X_test, y_test)
# 평가 지표 확인
assert 'mse' in evaluation_metrics
assert 'mae' in evaluation_metrics
assert 'rmse' in evaluation_metrics
assert 'mape' in evaluation_metrics
assert 'directional_accuracy' in evaluation_metrics
# 지표 범위 확인
assert evaluation_metrics['mse'] >= 0
assert evaluation_metrics['mae'] >= 0
assert evaluation_metrics['rmse'] >= 0
assert 0 <= evaluation_metrics['directional_accuracy'] <= 1
@pytest.mark.asyncio
async def test_model_save_load(self, price_predictor, sample_market_data, tmp_path):
"""모델 저장/로드 테스트"""
# 모델 훈련
features = await price_predictor.engineer_features(sample_market_data)
X, y = await price_predictor.preprocess_data(features)
await price_predictor.train(X, y)
# 모델 저장
model_path = tmp_path / "test_model.json"
await price_predictor.save_model(str(model_path))
assert model_path.exists()
# 새 예측기 인스턴스 생성
new_predictor = PricePredictor(price_predictor.config)
assert new_predictor.is_trained == False
# 모델 로드
await new_predictor.load_model(str(model_path))
assert new_predictor.is_trained == True
# 모델 구조 확인
assert new_predictor.model['type'] == price_predictor.model['type']
@pytest.mark.asyncio
async def test_insufficient_data_error(self, price_predictor):
"""데이터 부족 에러 테스트"""
# 매우 적은 데이터
small_data = MockDataFrame({
'price': [100] * 5,
'volume': [1000] * 5
})
with pytest.raises(InsufficientDataError):
features = await price_predictor.engineer_features(small_data)
X, y = await price_predictor.preprocess_data(features)
@pytest.mark.asyncio
async def test_untrained_model_error(self, price_predictor, sample_market_data):
"""훈련되지 않은 모델 에러 테스트"""
recent_data = MockDataFrame({
'price': [76000, 76100, 76200],
'volume': [1100000, 1110000, 1120000]
})
with pytest.raises(ModelNotTrainedError):
await price_predictor.predict(recent_data)
def test_model_architecture_creation(self, price_predictor):
"""모델 아키텍처 생성 테스트"""
input_shape = (price_predictor.sequence_length, len(price_predictor.features))
output_size = price_predictor.prediction_horizon
model = price_predictor._create_model(input_shape, output_size)
# 모델 구조 확인
assert model is not None
assert 'layers' in model
assert len(model['layers']) > 0
assert model['input_shape'] == input_shape
assert model['output_shape'] == (output_size,)
class TestTechnicalIndicators:
"""기술적 지표 계산 테스트"""
@pytest.fixture
def price_predictor(self):
"""예측기 인스턴스"""
config = {"model_type": "lstm"}
return PricePredictor(config)
def test_rsi_calculation(self, price_predictor):
"""RSI 계산 테스트"""
prices = MockSeries([100, 102, 105, 103, 107, 110, 108, 112, 115, 113])
rsi = price_predictor._calculate_rsi(prices)
# RSI 결과 확인
assert len(rsi) == len(prices)
# RSI는 보통 0-100 범위이지만 mock에서는 50 고정값
assert all(isinstance(val, (int, float)) for val in rsi.data)
def test_macd_calculation(self, price_predictor):
"""MACD 계산 테스트"""
prices = MockSeries([100, 102, 105, 103, 107, 110, 108, 112, 115, 113])
macd_line, signal_line = price_predictor._calculate_macd(prices)
# MACD 결과 확인
assert len(macd_line) == len(prices)
assert len(signal_line) == len(prices)
assert isinstance(macd_line.data[0], (int, float))
assert isinstance(signal_line.data[0], (int, float))
