""" Tests Unitaires - RegimeDetector. Tests de la détection de régimes de marché avec HMM. """ import pytest import pandas as pd import numpy as np from datetime import datetime, timedelta from src.ml.regime_detector import RegimeDetector class TestRegimeDetectorInitialization: """Tests d'initialisation du RegimeDetector.""" def test_initialization_default(self): """Test initialisation avec paramètres par défaut.""" detector = RegimeDetector() assert detector.n_regimes == 4 assert detector.random_state == 42 assert detector.is_fitted is False assert len(detector.feature_names) == 0 def test_initialization_custom_regimes(self): """Test initialisation avec nombre de régimes personnalisé.""" detector = RegimeDetector(n_regimes=3) assert detector.n_regimes == 3 def test_regime_names_defined(self): """Test que les noms de régimes sont définis.""" detector = RegimeDetector() assert len(detector.REGIME_NAMES) == 4 assert 'Trending Up' in detector.REGIME_NAMES.values() assert 'Trending Down' in detector.REGIME_NAMES.values() assert 'Ranging' in detector.REGIME_NAMES.values() assert 'High Volatility' in detector.REGIME_NAMES.values() class TestRegimeDetectorFitting: """Tests d'entraînement du modèle.""" @pytest.fixture def sample_data(self): """Génère des données de test.""" dates = pd.date_range(start='2024-01-01', periods=200, freq='1H') np.random.seed(42) returns = np.random.normal(0.0001, 0.01, 200) prices = 1.1000 * np.exp(np.cumsum(returns)) df = pd.DataFrame(index=dates) df['close'] = prices df['open'] = df['close'].shift(1).fillna(df['close'].iloc[0]) df['high'] = df[['open', 'close']].max(axis=1) * 1.001 df['low'] = df[['open', 'close']].min(axis=1) * 0.999 df['volume'] = np.random.randint(1000, 10000, 200) return df def test_fit_success(self, sample_data): """Test entraînement réussi.""" detector = RegimeDetector() detector.fit(sample_data) assert detector.is_fitted is True assert len(detector.feature_names) > 0 def test_fit_creates_features(self, sample_data): """Test que fit crée les features.""" detector = RegimeDetector() detector.fit(sample_data) # Vérifier que les features attendues sont créées expected_features = ['returns', 'volatility', 'trend', 'range', 'volume_change', 'momentum'] for feature in expected_features: assert feature in detector.feature_names def test_fit_with_insufficient_data(self): """Test avec données insuffisantes.""" detector = RegimeDetector() # Données trop courtes dates = pd.date_range(start='2024-01-01', periods=10, freq='1H') df = pd.DataFrame({ 'close': np.random.randn(10), 'open': np.random.randn(10), 'high': np.random.randn(10), 'low': np.random.randn(10), 'volume': np.random.randint(1000, 10000, 10) }, index=dates) # Devrait lever une erreur ou gérer gracieusement try: detector.fit(df) # Si pas d'erreur, vérifier que le modèle n'est pas fitted # ou qu'il y a un warning except Exception as e: # Acceptable pass class TestRegimeDetectorPrediction: """Tests de prédiction des régimes.""" @pytest.fixture def fitted_detector(self, sample_data): """Retourne un détecteur entraîné.""" detector = RegimeDetector() detector.fit(sample_data) return detector @pytest.fixture def sample_data(self): """Génère des données de test.""" dates = pd.date_range(start='2024-01-01', periods=200, freq='1H') np.random.seed(42) returns = np.random.normal(0.0001, 0.01, 200) prices = 1.1000 * np.exp(np.cumsum(returns)) df = pd.DataFrame(index=dates) df['close'] = prices df['open'] = df['close'].shift(1).fillna(df['close'].iloc[0]) df['high'] = df[['open', 'close']].max(axis=1) * 1.001 df['low'] = df[['open', 'close']].min(axis=1) * 0.999 df['volume'] = np.random.randint(1000, 10000, 200) return df def test_predict_regime_returns_array(self, fitted_detector, sample_data): """Test que predict_regime retourne un array.""" regimes = fitted_detector.predict_regime(sample_data) assert isinstance(regimes, np.ndarray) assert len(regimes) > 0 def test_predict_regime_values_valid(self, fitted_detector, sample_data): """Test que les régimes prédits sont valides.""" regimes = fitted_detector.predict_regime(sample_data) # Tous les régimes doivent être entre 0 et n_regimes-1 assert (regimes >= 0).all() assert (regimes < fitted_detector.n_regimes).all() def test_predict_current_regime(self, fitted_detector, sample_data): """Test prédiction du régime actuel.""" current_regime = fitted_detector.predict_current_regime(sample_data) assert isinstance(current_regime, (int, np.integer)) assert 0 <= current_regime < fitted_detector.n_regimes def test_predict_without_fitting(self, sample_data): """Test prédiction sans entraînement préalable.""" detector = RegimeDetector() with pytest.raises(ValueError, match="not fitted"): detector.predict_regime(sample_data) def test_get_regime_probabilities(self, fitted_detector, sample_data): """Test obtention des probabilités.""" probabilities = fitted_detector.get_regime_probabilities(sample_data) assert isinstance(probabilities, np.ndarray) assert probabilities.shape[1] == fitted_detector.n_regimes # Vérifier que les probabilités somment à 1 prob_sums = probabilities.sum(axis=1) np.testing.assert_array_almost_equal(prob_sums, np.ones(len(prob_sums)), decimal=5) class TestRegimeDetectorStatistics: """Tests des statistiques de régimes.""" @pytest.fixture def fitted_detector(self, sample_data): """Retourne un détecteur entraîné.""" detector = RegimeDetector() detector.fit(sample_data) return detector @pytest.fixture def sample_data(self): """Génère des données de test.""" dates = pd.date_range(start='2024-01-01', periods=200, freq='1H') np.random.seed(42) returns = np.random.normal(0.0001, 0.01, 200) prices = 1.1000 * np.exp(np.cumsum(returns)) df = pd.DataFrame(index=dates) df['close'] = prices df['open'] = df['close'].shift(1).fillna(df['close'].iloc[0]) df['high'] = df[['open', 'close']].max(axis=1) * 1.001 df['low'] = df[['open', 'close']].min(axis=1) * 0.999 df['volume'] = np.random.randint(1000, 10000, 200) return df def test_get_regime_name(self, fitted_detector): """Test récupération du nom d'un régime.""" for regime in range(fitted_detector.n_regimes): name = fitted_detector.get_regime_name(regime) assert isinstance(name, str) assert len(name) > 0 def test_get_regime_statistics(self, fitted_detector, sample_data): """Test calcul des statistiques.""" stats = fitted_detector.get_regime_statistics(sample_data) assert 'regime_counts' in stats assert 'regime_percentages' in stats assert 'current_regime' in stats assert 'current_regime_name' in stats # Vérifier que les pourcentages somment à 1 total_pct = sum(stats['regime_percentages'].values()) assert abs(total_pct - 1.0) < 0.01 class TestRegimeDetectorAdaptation: """Tests d'adaptation des paramètres.""" def test_adapt_strategy_parameters(self): """Test adaptation des paramètres selon le régime.""" detector = RegimeDetector() base_params = { 'min_confidence': 0.6, 'risk_per_trade': 0.02 } # Tester pour chaque régime for regime in range(4): adapted = detector.adapt_strategy_parameters(regime, base_params) assert 'min_confidence' in adapted assert 'risk_per_trade' in adapted # Les paramètres doivent être modifiés assert adapted != base_params def test_adapt_trending_up(self): """Test adaptation pour régime Trending Up.""" detector = RegimeDetector() base_params = { 'min_confidence': 0.6, 'risk_per_trade': 0.02 } adapted = detector.adapt_strategy_parameters(0, base_params) # 0 = Trending Up # Devrait être plus agressif assert adapted['min_confidence'] < base_params['min_confidence'] assert adapted['risk_per_trade'] > base_params['risk_per_trade'] def test_adapt_high_volatility(self): """Test adaptation pour régime High Volatility.""" detector = RegimeDetector() base_params = { 'min_confidence': 0.6, 'risk_per_trade': 0.02 } adapted = detector.adapt_strategy_parameters(3, base_params) # 3 = High Volatility # Devrait être plus conservateur assert adapted['min_confidence'] > base_params['min_confidence'] assert adapted['risk_per_trade'] < base_params['risk_per_trade'] def test_should_trade_in_regime(self): """Test décision de trading selon régime.""" detector = RegimeDetector() # Scalping devrait trader en Ranging assert detector.should_trade_in_regime(2, 'scalping') is True # Scalping ne devrait pas trader en High Volatility assert detector.should_trade_in_regime(3, 'scalping') is False # Intraday devrait trader en Trending assert detector.should_trade_in_regime(0, 'intraday') is True assert detector.should_trade_in_regime(1, 'intraday') is True # Intraday ne devrait pas trader en Ranging assert detector.should_trade_in_regime(2, 'intraday') is False class TestRegimeDetectorFeatures: """Tests de calcul des features.""" @pytest.fixture def sample_data(self): """Génère des données de test.""" dates = pd.date_range(start='2024-01-01', periods=200, freq='1H') np.random.seed(42) returns = np.random.normal(0.0001, 0.01, 200) prices = 1.1000 * np.exp(np.cumsum(returns)) df = pd.DataFrame(index=dates) df['close'] = prices df['open'] = df['close'].shift(1).fillna(df['close'].iloc[0]) df['high'] = df[['open', 'close']].max(axis=1) * 1.001 df['low'] = df[['open', 'close']].min(axis=1) * 0.999 df['volume'] = np.random.randint(1000, 10000, 200) return df def test_calculate_features(self, sample_data): """Test calcul des features.""" detector = RegimeDetector() features = detector._calculate_features(sample_data) assert isinstance(features, pd.DataFrame) assert len(features) > 0 # Vérifier présence des features expected_features = ['returns', 'volatility', 'trend', 'range', 'volume_change', 'momentum'] for feature in expected_features: assert feature in features.columns def test_features_no_nan(self, sample_data): """Test que les features n'ont pas de NaN après nettoyage.""" detector = RegimeDetector() features = detector._calculate_features(sample_data) # Après dropna, ne devrait pas y avoir de NaN assert features.isna().sum().sum() == 0 def test_normalize_features(self): """Test normalisation des features.""" detector = RegimeDetector() # Créer features test X = np.random.randn(100, 6) X_normalized = detector._normalize_features(X) # Vérifier que la moyenne est proche de 0 et std proche de 1 assert abs(X_normalized.mean()) < 0.1 assert abs(X_normalized.std() - 1.0) < 0.1 class TestRegimeDetectorEdgeCases: """Tests des cas limites.""" def test_with_missing_columns(self): """Test avec colonnes manquantes.""" detector = RegimeDetector() # DataFrame incomplet df = pd.DataFrame({ 'close': np.random.randn(100) # Manque open, high, low, volume }) with pytest.raises(KeyError): detector.fit(df) def test_with_constant_prices(self): """Test avec prix constants.""" detector = RegimeDetector() dates = pd.date_range(start='2024-01-01', periods=200, freq='1H') df = pd.DataFrame(index=dates) df['close'] = 1.1000 # Prix constant df['open'] = 1.1000 df['high'] = 1.1000 df['low'] = 1.1000 df['volume'] = 1000 # Devrait gérer gracieusement (ou lever erreur appropriée) try: detector.fit(df) # Si réussit, vérifier que le modèle est fitted assert detector.is_fitted except Exception: # Acceptable si erreur appropriée pass def test_regime_name_invalid(self): """Test avec numéro de régime invalide.""" detector = RegimeDetector() # Régime hors limites name = detector.get_regime_name(999) # Devrait retourner un nom par défaut assert 'Regime' in name class TestRegimeDetectorIntegration: """Tests d'intégration.""" @pytest.fixture def sample_data(self): """Génère des données de test.""" dates = pd.date_range(start='2024-01-01', periods=500, freq='1H') np.random.seed(42) # Créer différents régimes regimes = [] prices = [] base_price = 1.1000 for i in range(500): if i < 125: # Trending Up regime = 0 price = base_price * (1 + i * 0.0001) elif i < 250: # Trending Down regime = 1 price = base_price * (1 - (i - 125) * 0.0001) elif i < 375: # Ranging regime = 2 price = base_price + 0.001 * np.sin(i / 10) else: # High Volatility regime = 3 price = base_price + 0.01 * np.random.randn() regimes.append(regime) prices.append(price) df = pd.DataFrame(index=dates) df['close'] = prices df['open'] = df['close'].shift(1).fillna(df['close'].iloc[0]) df['high'] = df[['open', 'close']].max(axis=1) * 1.001 df['low'] = df[['open', 'close']].min(axis=1) * 0.999 df['volume'] = np.random.randint(1000, 10000, 500) return df def test_full_workflow(self, sample_data): """Test workflow complet.""" detector = RegimeDetector(n_regimes=4) # 1. Fit detector.fit(sample_data) assert detector.is_fitted # 2. Predict regimes = detector.predict_regime(sample_data) assert len(regimes) > 0 # 3. Current regime current = detector.predict_current_regime(sample_data) assert 0 <= current < 4 # 4. Statistics stats = detector.get_regime_statistics(sample_data) assert 'current_regime' in stats # 5. Adaptation adapted = detector.adapt_strategy_parameters(current, {'min_confidence': 0.6}) assert 'min_confidence' in adapted # 6. Should trade should_trade = detector.should_trade_in_regime(current, 'intraday') assert isinstance(should_trade, bool)