386 lines
13 KiB
Python
386 lines
13 KiB
Python
"""
|
|
Scalping Strategy - Stratégie de Scalping Mean Reversion.
|
|
|
|
Cette stratégie exploite les micro-mouvements du marché en utilisant
|
|
le retour à la moyenne sur des timeframes très courts (1-5 minutes).
|
|
|
|
Indicateurs:
|
|
- Bollinger Bands: Détection zones oversold/overbought
|
|
- RSI: Confirmation conditions extrêmes
|
|
- MACD: Validation momentum reversal
|
|
- Volume: Confirmation force du mouvement
|
|
- ATR: Calcul stop-loss/take-profit dynamiques
|
|
|
|
Conditions LONG:
|
|
- Prix proche Bollinger Band inférieure (< 20%)
|
|
- RSI < 30 (oversold)
|
|
- MACD histogram positif (reversal)
|
|
- Volume > 1.5x moyenne
|
|
- Confiance >= seuil minimum
|
|
|
|
Conditions SHORT:
|
|
- Prix proche Bollinger Band supérieure (> 80%)
|
|
- RSI > 70 (overbought)
|
|
- MACD histogram négatif (reversal)
|
|
- Volume > 1.5x moyenne
|
|
- Confiance >= seuil minimum
|
|
"""
|
|
|
|
from typing import Optional
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
from src.strategies.base_strategy import BaseStrategy, Signal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ScalpingStrategy(BaseStrategy):
|
|
"""
|
|
Stratégie de scalping basée sur mean reversion.
|
|
|
|
Timeframe: 1-5 minutes
|
|
Holding time: 5-30 minutes
|
|
Risk per trade: 0.5-1%
|
|
Win rate target: 60-70%
|
|
Profit target: 0.3-0.5% par trade
|
|
|
|
Usage:
|
|
strategy = ScalpingStrategy(config)
|
|
signal = strategy.analyze(market_data)
|
|
"""
|
|
|
|
def __init__(self, config: dict):
|
|
"""
|
|
Initialise la stratégie de scalping.
|
|
|
|
Args:
|
|
config: Configuration de la stratégie
|
|
"""
|
|
# Aligner risk_per_trade avec la limite RiskManager pour scalping (0.5%)
|
|
config.setdefault('risk_per_trade', 0.005)
|
|
super().__init__(config)
|
|
|
|
# Paramètres par défaut si non fournis
|
|
self.parameters.setdefault('bb_period', 20)
|
|
self.parameters.setdefault('bb_std', 2.0)
|
|
self.parameters.setdefault('rsi_period', 14)
|
|
self.parameters.setdefault('rsi_oversold', 30)
|
|
self.parameters.setdefault('rsi_overbought', 70)
|
|
self.parameters.setdefault('volume_threshold', 1.5)
|
|
self.parameters.setdefault('min_confidence', 0.65)
|
|
|
|
logger.info(f"Scalping Strategy initialized with params: {self.parameters}")
|
|
|
|
def calculate_indicators(self, data: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Calcule tous les indicateurs nécessaires pour le scalping.
|
|
|
|
Args:
|
|
data: DataFrame avec colonnes OHLCV
|
|
|
|
Returns:
|
|
DataFrame avec indicateurs ajoutés
|
|
"""
|
|
df = data.copy()
|
|
|
|
# Bollinger Bands
|
|
bb_period = int(self.parameters['bb_period'])
|
|
bb_std = float(self.parameters['bb_std'])
|
|
|
|
df['bb_middle'] = df['close'].rolling(bb_period).mean()
|
|
df['bb_std'] = df['close'].rolling(bb_period).std()
|
|
df['bb_upper'] = df['bb_middle'] + (bb_std * df['bb_std'])
|
|
df['bb_lower'] = df['bb_middle'] - (bb_std * df['bb_std'])
|
|
|
|
# Position dans les Bollinger Bands (0 = lower, 1 = upper)
|
|
df['bb_position'] = (df['close'] - df['bb_lower']) / (df['bb_upper'] - df['bb_lower'])
|
|
|
|
# RSI (Relative Strength Index)
|
|
rsi_period = int(self.parameters['rsi_period'])
|
|
delta = df['close'].diff()
|
|
|
|
gain = delta.where(delta > 0, 0).rolling(rsi_period).mean()
|
|
loss = (-delta.where(delta < 0, 0)).rolling(rsi_period).mean()
|
|
|
|
rs = gain / loss
|
|
df['rsi'] = 100 - (100 / (1 + rs))
|
|
|
|
# MACD (Moving Average Convergence Divergence)
|
|
df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean()
|
|
df['ema_26'] = df['close'].ewm(span=26, adjust=False).mean()
|
|
df['macd'] = df['ema_12'] - df['ema_26']
|
|
df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
|
|
df['macd_hist'] = df['macd'] - df['macd_signal']
|
|
|
|
# Volume (désactivé si données volume non fiables, ex: forex Yahoo Finance)
|
|
df['volume_ma'] = df['volume'].rolling(20).mean()
|
|
if df['volume'].sum() == 0:
|
|
df['volume_ratio'] = 2.0 # Volume fictif >= seuil pour ne pas bloquer
|
|
else:
|
|
df['volume_ratio'] = df['volume'] / df['volume_ma']
|
|
|
|
# ATR (Average True Range) pour stop-loss/take-profit
|
|
df['high_low'] = df['high'] - df['low']
|
|
df['high_close'] = abs(df['high'] - df['close'].shift(1))
|
|
df['low_close'] = abs(df['low'] - df['close'].shift(1))
|
|
|
|
df['tr'] = df[['high_low', 'high_close', 'low_close']].max(axis=1)
|
|
df['atr'] = df['tr'].rolling(14).mean()
|
|
|
|
return df
|
|
|
|
def analyze(self, market_data: pd.DataFrame) -> Optional[Signal]:
|
|
"""
|
|
Analyse le marché et génère un signal de scalping.
|
|
|
|
Args:
|
|
market_data: DataFrame avec données OHLCV
|
|
|
|
Returns:
|
|
Signal si opportunité détectée, None sinon
|
|
"""
|
|
# Calculer indicateurs
|
|
df = self.calculate_indicators(market_data)
|
|
|
|
# Besoin d'au moins 50 barres pour indicateurs fiables
|
|
if len(df) < 50:
|
|
logger.debug("Not enough data for analysis")
|
|
return None
|
|
|
|
# Données actuelles et précédentes
|
|
current = df.iloc[-1]
|
|
prev = df.iloc[-2]
|
|
|
|
# Vérifier que tous les indicateurs sont calculés
|
|
if pd.isna(current['bb_position']) or pd.isna(current['rsi']) or pd.isna(current['macd_hist']):
|
|
logger.debug("Indicators not fully calculated")
|
|
return None
|
|
|
|
# Vérifier volume suffisant
|
|
if current['volume_ratio'] < self.parameters['volume_threshold']:
|
|
logger.debug(f"Volume too low: {current['volume_ratio']:.2f}")
|
|
return None
|
|
|
|
# Détecter signal LONG (oversold reversal)
|
|
if self._check_long_conditions(current, prev):
|
|
confidence = self._calculate_confidence(df, 'LONG')
|
|
|
|
if confidence >= self.parameters['min_confidence']:
|
|
return self._create_long_signal(current, confidence)
|
|
|
|
# Détecter signal SHORT (overbought reversal)
|
|
elif self._check_short_conditions(current, prev):
|
|
confidence = self._calculate_confidence(df, 'SHORT')
|
|
|
|
if confidence >= self.parameters['min_confidence']:
|
|
return self._create_short_signal(current, confidence)
|
|
|
|
return None
|
|
|
|
def _check_long_conditions(self, current: pd.Series, prev: pd.Series) -> bool:
|
|
"""
|
|
Vérifie les conditions pour un signal LONG.
|
|
|
|
Args:
|
|
current: Barre actuelle
|
|
prev: Barre précédente
|
|
|
|
Returns:
|
|
True si conditions remplies
|
|
"""
|
|
return (
|
|
# Prix proche Bollinger Band inférieure
|
|
current['bb_position'] < 0.2 and
|
|
|
|
# RSI oversold
|
|
current['rsi'] < self.parameters['rsi_oversold'] and
|
|
|
|
# MACD momentum haussier (histogram en hausse — pas besoin de croiser zéro)
|
|
current['macd_hist'] > prev['macd_hist'] and
|
|
|
|
# Volume confirmation
|
|
current['volume_ratio'] > self.parameters['volume_threshold']
|
|
)
|
|
|
|
def _check_short_conditions(self, current: pd.Series, prev: pd.Series) -> bool:
|
|
"""
|
|
Vérifie les conditions pour un signal SHORT.
|
|
|
|
Args:
|
|
current: Barre actuelle
|
|
prev: Barre précédente
|
|
|
|
Returns:
|
|
True si conditions remplies
|
|
"""
|
|
return (
|
|
# Prix proche Bollinger Band supérieure
|
|
current['bb_position'] > 0.8 and
|
|
|
|
# RSI overbought
|
|
current['rsi'] > self.parameters['rsi_overbought'] and
|
|
|
|
# MACD momentum baissier (histogram en baisse — pas besoin de croiser zéro)
|
|
current['macd_hist'] < prev['macd_hist'] and
|
|
|
|
# Volume confirmation
|
|
current['volume_ratio'] > self.parameters['volume_threshold']
|
|
)
|
|
|
|
def _create_long_signal(self, current: pd.Series, confidence: float) -> Signal:
|
|
"""
|
|
Crée un signal LONG.
|
|
|
|
Args:
|
|
current: Barre actuelle
|
|
confidence: Confiance du signal
|
|
|
|
Returns:
|
|
Signal LONG
|
|
"""
|
|
entry_price = current['close']
|
|
atr = current['atr']
|
|
|
|
# Stop-loss à 2 ATR en dessous
|
|
stop_loss = entry_price - (2.0 * atr)
|
|
|
|
# Take-profit à 4 ATR au-dessus (R:R 2:1)
|
|
take_profit = entry_price + (4.0 * atr)
|
|
|
|
signal = Signal(
|
|
symbol=current.name if hasattr(current, 'name') else 'UNKNOWN',
|
|
direction='LONG',
|
|
entry_price=entry_price,
|
|
stop_loss=stop_loss,
|
|
take_profit=take_profit,
|
|
confidence=confidence,
|
|
timestamp=datetime.now(),
|
|
strategy='scalping',
|
|
metadata={
|
|
'rsi': float(current['rsi']),
|
|
'bb_position': float(current['bb_position']),
|
|
'macd_hist': float(current['macd_hist']),
|
|
'volume_ratio': float(current['volume_ratio']),
|
|
'atr': float(atr)
|
|
}
|
|
)
|
|
|
|
logger.info(f"LONG signal generated - Confidence: {confidence:.2%}")
|
|
|
|
return signal
|
|
|
|
def _create_short_signal(self, current: pd.Series, confidence: float) -> Signal:
|
|
"""
|
|
Crée un signal SHORT.
|
|
|
|
Args:
|
|
current: Barre actuelle
|
|
confidence: Confiance du signal
|
|
|
|
Returns:
|
|
Signal SHORT
|
|
"""
|
|
entry_price = current['close']
|
|
atr = current['atr']
|
|
|
|
# Stop-loss à 2 ATR au-dessus
|
|
stop_loss = entry_price + (2.0 * atr)
|
|
|
|
# Take-profit à 4 ATR en dessous (R:R 2:1)
|
|
take_profit = entry_price - (4.0 * atr)
|
|
|
|
signal = Signal(
|
|
symbol=current.name if hasattr(current, 'name') else 'UNKNOWN',
|
|
direction='SHORT',
|
|
entry_price=entry_price,
|
|
stop_loss=stop_loss,
|
|
take_profit=take_profit,
|
|
confidence=confidence,
|
|
timestamp=datetime.now(),
|
|
strategy='scalping',
|
|
metadata={
|
|
'rsi': float(current['rsi']),
|
|
'bb_position': float(current['bb_position']),
|
|
'macd_hist': float(current['macd_hist']),
|
|
'volume_ratio': float(current['volume_ratio']),
|
|
'atr': float(atr)
|
|
}
|
|
)
|
|
|
|
logger.info(f"SHORT signal generated - Confidence: {confidence:.2%}")
|
|
|
|
return signal
|
|
|
|
def _calculate_confidence(self, df: pd.DataFrame, direction: str) -> float:
|
|
"""
|
|
Calcule la confiance du signal (0.0 à 1.0).
|
|
|
|
Facteurs:
|
|
- Force de l'oversold/overbought (RSI)
|
|
- Position dans Bollinger Bands
|
|
- Force du volume
|
|
- Historique win rate
|
|
|
|
Args:
|
|
df: DataFrame avec indicateurs
|
|
direction: 'LONG' ou 'SHORT'
|
|
|
|
Returns:
|
|
Confiance entre 0.0 et 1.0
|
|
"""
|
|
current = df.iloc[-1]
|
|
|
|
# Confiance de base
|
|
confidence = 0.5
|
|
|
|
if direction == 'LONG':
|
|
# Force RSI oversold (plus c'est bas, plus c'est fort)
|
|
rsi_strength = (30 - current['rsi']) / 30
|
|
confidence += 0.2 * max(0, rsi_strength)
|
|
|
|
# Position Bollinger Bands (plus c'est bas, plus c'est fort)
|
|
bb_strength = (0.2 - current['bb_position']) / 0.2
|
|
confidence += 0.15 * max(0, bb_strength)
|
|
|
|
else: # SHORT
|
|
# Force RSI overbought (plus c'est haut, plus c'est fort)
|
|
rsi_strength = (current['rsi'] - 70) / 30
|
|
confidence += 0.2 * max(0, rsi_strength)
|
|
|
|
# Position Bollinger Bands (plus c'est haut, plus c'est fort)
|
|
bb_strength = (current['bb_position'] - 0.8) / 0.2
|
|
confidence += 0.15 * max(0, bb_strength)
|
|
|
|
# Force du volume
|
|
volume_strength = min((current['volume_ratio'] - 1.5) / 1.5, 1.0)
|
|
confidence += 0.15 * max(0, volume_strength)
|
|
|
|
# Historique win rate (bonus si bonne performance)
|
|
if self.win_rate > 0.5:
|
|
confidence += 0.1 * (self.win_rate - 0.5)
|
|
|
|
# Limiter entre 0 et 1
|
|
return np.clip(confidence, 0.0, 1.0)
|
|
|
|
def get_strategy_info(self) -> dict:
|
|
"""
|
|
Retourne les informations de la stratégie.
|
|
|
|
Returns:
|
|
Dictionnaire avec informations
|
|
"""
|
|
return {
|
|
'name': 'Scalping Mean Reversion',
|
|
'type': 'scalping',
|
|
'timeframe': '1-5min',
|
|
'indicators': ['Bollinger Bands', 'RSI', 'MACD', 'Volume', 'ATR'],
|
|
'risk_per_trade': '0.5-1%',
|
|
'target_win_rate': '60-70%',
|
|
'target_profit': '0.3-0.5%',
|
|
'parameters': self.parameters,
|
|
'statistics': self.get_statistics()
|
|
}
|