Spaces:

chen470
/

drl-trading-bot-dev2

Paused

drl-trading-bot-dev2 / src /features /mtf_analyzer.py

DRL Trading Bot

Feature: HTF Agent integration — live trading, API endpoints, UI tab

fc115d5 8 months ago

12.5 kB

"""
Multi-Timeframe Confluence Analyzer
Requires alignment across multiple timeframes before trading:
- 4H: Overall trend direction (higher timeframe bias)
- 1H: Primary trading timeframe (model operates here)
- 15m: Entry timing (fine-tune entries)
Only trades when all timeframes agree on direction.
"""

import numpy as np
import pandas as pd
from typing import Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import logging
import requests
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)


class TrendDirection(Enum):
    BULLISH = "bullish"
    BEARISH = "bearish"
    NEUTRAL = "neutral"


@dataclass
class TimeframeSignal:
    """Signal from a single timeframe."""
    timeframe: str
    direction: TrendDirection
    strength: float  # 0-1
    price: float
    ema_fast: float
    ema_slow: float
    rsi: float


@dataclass
class ConfluenceResult:
    """Result of multi-timeframe analysis."""
    aligned: bool
    direction: TrendDirection
    strength: float  # Average strength across timeframes
    signals: Dict[str, TimeframeSignal]
    recommendation: str


class MultiTimeframeAnalyzer:
    """
    Analyze multiple timeframes for trade confluence.
    
    Strategy:
    - 4H determines overall bias (trend direction)
    - 1H is the primary trading timeframe
    - 15m provides entry timing confirmation
    
    Only enter trades when all timeframes align.
    """
    
    def __init__(
        self,
        symbol: str = "BTCUSDT",
        ema_fast: int = 9,
        ema_slow: int = 21,
        rsi_period: int = 14,
        rsi_overbought: float = 70,
        rsi_oversold: float = 30
    ):
        """
        Initialize multi-timeframe analyzer.
        
        Args:
            symbol: Trading symbol
            ema_fast: Fast EMA period
            ema_slow: Slow EMA period
            rsi_period: RSI period
            rsi_overbought: RSI overbought level
            rsi_oversold: RSI oversold level
        """
        self.symbol = symbol
        self.ema_fast = ema_fast
        self.ema_slow = ema_slow
        self.rsi_period = rsi_period
        self.rsi_overbought = rsi_overbought
        self.rsi_oversold = rsi_oversold
        
        # Timeframes to analyze
        self.timeframes = ["4h", "1h", "15m"]
        
        # Cache for fetched data
        self._data_cache: Dict[str, pd.DataFrame] = {}
        self._cache_time: Dict[str, datetime] = {}
        
        logger.info(f"📊 MTF Analyzer initialized for {symbol} ({', '.join(self.timeframes)})")
    
    def _fetch_klines(self, timeframe: str, limit: int = 100) -> Optional[pd.DataFrame]:
        """Fetch klines from Binance for a specific timeframe."""
        # Check cache (valid for 1 minute)
        cache_key = f"{self.symbol}_{timeframe}"
        if cache_key in self._cache_time:
            if datetime.now() - self._cache_time[cache_key] < timedelta(minutes=1):
                return self._data_cache.get(cache_key)
        
        try:
            import os
            url = os.environ.get("BINANCE_FUTURES_URL", "https://data-api.binance.vision") + "/api/v3/klines"
            params = {
                "symbol": self.symbol,
                "interval": timeframe,
                "limit": limit
            }
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            df = pd.DataFrame(data, columns=[
                'timestamp', 'open', 'high', 'low', 'close', 'volume',
                'close_time', 'quote_volume', 'trades', 'taker_buy_base',
                'taker_buy_quote', 'ignore'
            ])
            
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            for col in ['open', 'high', 'low', 'close', 'volume']:
                df[col] = df[col].astype(float)
            
            # Cache the result
            self._data_cache[cache_key] = df
            self._cache_time[cache_key] = datetime.now()
            
            return df
            
        except Exception as e:
            logger.error(f"Failed to fetch {timeframe} klines: {e}")
            return None
    
    def _calculate_ema(self, series: pd.Series, period: int) -> pd.Series:
        """Calculate Exponential Moving Average."""
        return series.ewm(span=period, adjust=False).mean()
    
    def _calculate_rsi(self, series: pd.Series, period: int = 14) -> pd.Series:
        """Calculate Relative Strength Index."""
        delta = series.diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
        rs = gain / (loss + 1e-10)
        return 100 - (100 / (1 + rs))
    
    def analyze_timeframe(self, timeframe: str, df: pd.DataFrame = None) -> Optional[TimeframeSignal]:
        """
        Analyze a single timeframe.
        
        Args:
            timeframe: Timeframe to analyze (e.g., "4h", "1h", "15m")
            df: Optional pre-fetched dataframe
            
        Returns:
            TimeframeSignal with direction and strength
        """
        if df is None:
            df = self._fetch_klines(timeframe)
        
        if df is None or len(df) < self.ema_slow + 5:
            return None
        
        close = df['close']
        current_price = close.iloc[-1]
        
        # Calculate indicators
        ema_fast = self._calculate_ema(close, self.ema_fast)
        ema_slow = self._calculate_ema(close, self.ema_slow)
        rsi = self._calculate_rsi(close, self.rsi_period)
        
        current_ema_fast = ema_fast.iloc[-1]
        current_ema_slow = ema_slow.iloc[-1]
        current_rsi = rsi.iloc[-1]
        
        # Determine direction
        direction = TrendDirection.NEUTRAL
        strength = 0.0
        
        # EMA crossover direction
        ema_diff_pct = (current_ema_fast - current_ema_slow) / current_ema_slow * 100
        
        if ema_diff_pct > 0.1:  # Fast EMA above slow = bullish
            direction = TrendDirection.BULLISH
            strength = min(abs(ema_diff_pct) / 2, 1.0)  # Cap at 1.0
        elif ema_diff_pct < -0.1:  # Fast EMA below slow = bearish
            direction = TrendDirection.BEARISH
            strength = min(abs(ema_diff_pct) / 2, 1.0)
        
        # RSI confirmation
        if current_rsi > self.rsi_overbought:
            if direction == TrendDirection.BEARISH:
                strength = min(strength * 1.3, 1.0)  # Boost bearish
            elif direction == TrendDirection.BULLISH:
                strength *= 0.7  # Reduce bullish (overbought warning)
        elif current_rsi < self.rsi_oversold:
            if direction == TrendDirection.BULLISH:
                strength = min(strength * 1.3, 1.0)  # Boost bullish
            elif direction == TrendDirection.BEARISH:
                strength *= 0.7  # Reduce bearish (oversold warning)
        
        return TimeframeSignal(
            timeframe=timeframe,
            direction=direction,
            strength=strength,
            price=current_price,
            ema_fast=current_ema_fast,
            ema_slow=current_ema_slow,
            rsi=current_rsi
        )
    
    def get_confluence(self, primary_df: pd.DataFrame = None) -> ConfluenceResult:
        """
        Analyze all timeframes and check for confluence.
        
        Args:
            primary_df: Optional 1H dataframe if already fetched
            
        Returns:
            ConfluenceResult with alignment status and recommendation
        """
        signals: Dict[str, TimeframeSignal] = {}
        
        # Analyze each timeframe
        for tf in self.timeframes:
            df = primary_df if tf == "1h" else None
            signal = self.analyze_timeframe(tf, df)
            if signal:
                signals[tf] = signal
        
        # Check if we have all signals
        if len(signals) < len(self.timeframes):
            missing = set(self.timeframes) - set(signals.keys())
            return ConfluenceResult(
                aligned=False,
                direction=TrendDirection.NEUTRAL,
                strength=0,
                signals=signals,
                recommendation=f"Missing data for: {', '.join(missing)}"
            )
        
        # Count directions
        bullish_count = sum(1 for s in signals.values() if s.direction == TrendDirection.BULLISH)
        bearish_count = sum(1 for s in signals.values() if s.direction == TrendDirection.BEARISH)
        
        # All aligned?
        all_bullish = bullish_count == len(self.timeframes)
        all_bearish = bearish_count == len(self.timeframes)
        aligned = all_bullish or all_bearish
        
        # Determine overall direction
        if all_bullish:
            direction = TrendDirection.BULLISH
        elif all_bearish:
            direction = TrendDirection.BEARISH
        elif bullish_count > bearish_count:
            direction = TrendDirection.BULLISH
        elif bearish_count > bullish_count:
            direction = TrendDirection.BEARISH
        else:
            direction = TrendDirection.NEUTRAL
        
        # Calculate average strength
        avg_strength = np.mean([s.strength for s in signals.values()])
        
        # Generate recommendation
        if aligned:
            if direction == TrendDirection.BULLISH:
                recommendation = f"✅ ALL BULLISH - Strong LONG setup (strength: {avg_strength:.0%})"
            else:
                recommendation = f"✅ ALL BEARISH - Strong SHORT setup (strength: {avg_strength:.0%})"
        else:
            # Describe the conflict
            tf_dirs = [f"{tf}={s.direction.value}" for tf, s in signals.items()]
            recommendation = f"⚠️ CONFLICTING: {', '.join(tf_dirs)}"
        
        result = ConfluenceResult(
            aligned=aligned,
            direction=direction,
            strength=avg_strength,
            signals=signals,
            recommendation=recommendation
        )
        
        # Log the analysis
        emoji = "✅" if aligned else "⚠️"
        logger.info(
            f"📊 MTF Analysis: {emoji} {direction.value.upper()} | "
            f"4H={signals['4h'].direction.value if '4h' in signals else '?'} "
            f"1H={signals['1h'].direction.value if '1h' in signals else '?'} "
            f"15m={signals['15m'].direction.value if '15m' in signals else '?'}"
        )
        
        return result
    
    def should_trade(self, trade_type: str, primary_df: pd.DataFrame = None) -> Tuple[bool, str]:
        """
        Check if trade is allowed based on MTF confluence.
        
        Args:
            trade_type: "long" or "short"
            primary_df: Optional 1H dataframe
            
        Returns:
            Tuple of (should_trade, reason)
        """
        result = self.get_confluence(primary_df)
        
        # If not aligned, don't trade
        if not result.aligned:
            return False, f"MTF not aligned: {result.recommendation}"
        
        # Check direction matches trade type
        if trade_type == "long" and result.direction == TrendDirection.BULLISH:
            return True, f"MTF confirms LONG: {result.recommendation}"
        elif trade_type == "short" and result.direction == TrendDirection.BEARISH:
            return True, f"MTF confirms SHORT: {result.recommendation}"
        else:
            return False, f"MTF opposes {trade_type.upper()}: Market is {result.direction.value}"
    
    def get_summary(self, primary_df: pd.DataFrame = None) -> Dict:
        """Get a summary dict for UI/logging."""
        result = self.get_confluence(primary_df)
        
        signal_summary = {}
        for tf, signal in result.signals.items():
            signal_summary[tf] = {
                'direction': signal.direction.value,
                'strength': signal.strength,
                'rsi': signal.rsi,
                'emoji': '📈' if signal.direction == TrendDirection.BULLISH else 
                         '📉' if signal.direction == TrendDirection.BEARISH else '➡️'
            }
        
        return {
            'aligned': result.aligned,
            'direction': result.direction.value,
            'strength': result.strength,
            'signals': signal_summary,
            'recommendation': result.recommendation
        }

	"""
	Multi-Timeframe Confluence Analyzer

	Requires alignment across multiple timeframes before trading:
	- 4H: Overall trend direction (higher timeframe bias)
	- 1H: Primary trading timeframe (model operates here)
	- 15m: Entry timing (fine-tune entries)

	Only trades when all timeframes agree on direction.
	"""

	import numpy as np
	import pandas as pd
	from typing import Dict, Tuple, Optional
	from dataclasses import dataclass
	from enum import Enum
	import logging
	import requests
	from datetime import datetime, timedelta

	logger = logging.getLogger(__name__)


	class TrendDirection(Enum):
	BULLISH = "bullish"
	BEARISH = "bearish"
	NEUTRAL = "neutral"


	@dataclass
	class TimeframeSignal:
	"""Signal from a single timeframe."""
	timeframe: str
	direction: TrendDirection
	strength: float # 0-1
	price: float
	ema_fast: float
	ema_slow: float
	rsi: float


	@dataclass
	class ConfluenceResult:
	"""Result of multi-timeframe analysis."""
	aligned: bool
	direction: TrendDirection
	strength: float # Average strength across timeframes
	signals: Dict[str, TimeframeSignal]
	recommendation: str


	class MultiTimeframeAnalyzer:
	"""
	Analyze multiple timeframes for trade confluence.

	Strategy:
	- 4H determines overall bias (trend direction)
	- 1H is the primary trading timeframe
	- 15m provides entry timing confirmation

	Only enter trades when all timeframes align.
	"""

	def __init__(
	self,
	symbol: str = "BTCUSDT",
	ema_fast: int = 9,
	ema_slow: int = 21,
	rsi_period: int = 14,
	rsi_overbought: float = 70,
	rsi_oversold: float = 30
	):
	"""
	Initialize multi-timeframe analyzer.

	Args:
	symbol: Trading symbol
	ema_fast: Fast EMA period
	ema_slow: Slow EMA period
	rsi_period: RSI period
	rsi_overbought: RSI overbought level
	rsi_oversold: RSI oversold level
	"""
	self.symbol = symbol
	self.ema_fast = ema_fast
	self.ema_slow = ema_slow
	self.rsi_period = rsi_period
	self.rsi_overbought = rsi_overbought
	self.rsi_oversold = rsi_oversold

	# Timeframes to analyze
	self.timeframes = ["4h", "1h", "15m"]

	# Cache for fetched data
	self._data_cache: Dict[str, pd.DataFrame] = {}
	self._cache_time: Dict[str, datetime] = {}

	logger.info(f"📊 MTF Analyzer initialized for {symbol} ({', '.join(self.timeframes)})")

	def _fetch_klines(self, timeframe: str, limit: int = 100) -> Optional[pd.DataFrame]:
	"""Fetch klines from Binance for a specific timeframe."""
	# Check cache (valid for 1 minute)
	cache_key = f"{self.symbol}_{timeframe}"
	if cache_key in self._cache_time:
	if datetime.now() - self._cache_time[cache_key] < timedelta(minutes=1):
	return self._data_cache.get(cache_key)

	try:
	import os
	url = os.environ.get("BINANCE_FUTURES_URL", "https://data-api.binance.vision") + "/api/v3/klines"
	params = {
	"symbol": self.symbol,
	"interval": timeframe,
	"limit": limit
	}
	response = requests.get(url, params=params, timeout=10)
	response.raise_for_status()
	data = response.json()

	df = pd.DataFrame(data, columns=[
	'timestamp', 'open', 'high', 'low', 'close', 'volume',
	'close_time', 'quote_volume', 'trades', 'taker_buy_base',
	'taker_buy_quote', 'ignore'
	])

	df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
	for col in ['open', 'high', 'low', 'close', 'volume']:
	df[col] = df[col].astype(float)

	# Cache the result
	self._data_cache[cache_key] = df
	self._cache_time[cache_key] = datetime.now()

	return df

	except Exception as e:
	logger.error(f"Failed to fetch {timeframe} klines: {e}")
	return None

	def _calculate_ema(self, series: pd.Series, period: int) -> pd.Series:
	"""Calculate Exponential Moving Average."""
	return series.ewm(span=period, adjust=False).mean()

	def _calculate_rsi(self, series: pd.Series, period: int = 14) -> pd.Series:
	"""Calculate Relative Strength Index."""
	delta = series.diff()
	gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
	loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
	rs = gain / (loss + 1e-10)
	return 100 - (100 / (1 + rs))

	def analyze_timeframe(self, timeframe: str, df: pd.DataFrame = None) -> Optional[TimeframeSignal]:
	"""
	Analyze a single timeframe.

	Args:
	timeframe: Timeframe to analyze (e.g., "4h", "1h", "15m")
	df: Optional pre-fetched dataframe

	Returns:
	TimeframeSignal with direction and strength
	"""
	if df is None:
	df = self._fetch_klines(timeframe)

	if df is None or len(df) < self.ema_slow + 5:
	return None

	close = df['close']
	current_price = close.iloc[-1]

	# Calculate indicators
	ema_fast = self._calculate_ema(close, self.ema_fast)
	ema_slow = self._calculate_ema(close, self.ema_slow)
	rsi = self._calculate_rsi(close, self.rsi_period)

	current_ema_fast = ema_fast.iloc[-1]
	current_ema_slow = ema_slow.iloc[-1]
	current_rsi = rsi.iloc[-1]

	# Determine direction
	direction = TrendDirection.NEUTRAL
	strength = 0.0

	# EMA crossover direction
	ema_diff_pct = (current_ema_fast - current_ema_slow) / current_ema_slow * 100

	if ema_diff_pct > 0.1: # Fast EMA above slow = bullish
	direction = TrendDirection.BULLISH
	strength = min(abs(ema_diff_pct) / 2, 1.0) # Cap at 1.0
	elif ema_diff_pct < -0.1: # Fast EMA below slow = bearish
	direction = TrendDirection.BEARISH
	strength = min(abs(ema_diff_pct) / 2, 1.0)

	# RSI confirmation
	if current_rsi > self.rsi_overbought:
	if direction == TrendDirection.BEARISH:
	strength = min(strength * 1.3, 1.0) # Boost bearish
	elif direction == TrendDirection.BULLISH:
	strength *= 0.7 # Reduce bullish (overbought warning)
	elif current_rsi < self.rsi_oversold:
	if direction == TrendDirection.BULLISH:
	strength = min(strength * 1.3, 1.0) # Boost bullish
	elif direction == TrendDirection.BEARISH:
	strength *= 0.7 # Reduce bearish (oversold warning)

	return TimeframeSignal(
	timeframe=timeframe,
	direction=direction,
	strength=strength,
	price=current_price,
	ema_fast=current_ema_fast,
	ema_slow=current_ema_slow,
	rsi=current_rsi
	)

	def get_confluence(self, primary_df: pd.DataFrame = None) -> ConfluenceResult:
	"""
	Analyze all timeframes and check for confluence.

	Args:
	primary_df: Optional 1H dataframe if already fetched

	Returns:
	ConfluenceResult with alignment status and recommendation
	"""
	signals: Dict[str, TimeframeSignal] = {}

	# Analyze each timeframe
	for tf in self.timeframes:
	df = primary_df if tf == "1h" else None
	signal = self.analyze_timeframe(tf, df)
	if signal:
	signals[tf] = signal

	# Check if we have all signals
	if len(signals) < len(self.timeframes):
	missing = set(self.timeframes) - set(signals.keys())
	return ConfluenceResult(
	aligned=False,
	direction=TrendDirection.NEUTRAL,
	strength=0,
	signals=signals,
	recommendation=f"Missing data for: {', '.join(missing)}"
	)

	# Count directions
	bullish_count = sum(1 for s in signals.values() if s.direction == TrendDirection.BULLISH)
	bearish_count = sum(1 for s in signals.values() if s.direction == TrendDirection.BEARISH)

	# All aligned?
	all_bullish = bullish_count == len(self.timeframes)
	all_bearish = bearish_count == len(self.timeframes)
	aligned = all_bullish or all_bearish

	# Determine overall direction
	if all_bullish:
	direction = TrendDirection.BULLISH
	elif all_bearish:
	direction = TrendDirection.BEARISH
	elif bullish_count > bearish_count:
	direction = TrendDirection.BULLISH
	elif bearish_count > bullish_count:
	direction = TrendDirection.BEARISH
	else:
	direction = TrendDirection.NEUTRAL

	# Calculate average strength
	avg_strength = np.mean([s.strength for s in signals.values()])

	# Generate recommendation
	if aligned:
	if direction == TrendDirection.BULLISH:
	recommendation = f"✅ ALL BULLISH - Strong LONG setup (strength: {avg_strength:.0%})"
	else:
	recommendation = f"✅ ALL BEARISH - Strong SHORT setup (strength: {avg_strength:.0%})"
	else:
	# Describe the conflict
	tf_dirs = [f"{tf}={s.direction.value}" for tf, s in signals.items()]
	recommendation = f"⚠️ CONFLICTING: {', '.join(tf_dirs)}"

	result = ConfluenceResult(
	aligned=aligned,
	direction=direction,
	strength=avg_strength,
	signals=signals,
	recommendation=recommendation
	)

	# Log the analysis
	emoji = "✅" if aligned else "⚠️"
	logger.info(
	f"📊 MTF Analysis: {emoji} {direction.value.upper()} \| "
	f"4H={signals['4h'].direction.value if '4h' in signals else '?'} "
	f"1H={signals['1h'].direction.value if '1h' in signals else '?'} "
	f"15m={signals['15m'].direction.value if '15m' in signals else '?'}"
	)

	return result

	def should_trade(self, trade_type: str, primary_df: pd.DataFrame = None) -> Tuple[bool, str]:
	"""
	Check if trade is allowed based on MTF confluence.

	Args:
	trade_type: "long" or "short"
	primary_df: Optional 1H dataframe

	Returns:
	Tuple of (should_trade, reason)
	"""
	result = self.get_confluence(primary_df)

	# If not aligned, don't trade
	if not result.aligned:
	return False, f"MTF not aligned: {result.recommendation}"

	# Check direction matches trade type
	if trade_type == "long" and result.direction == TrendDirection.BULLISH:
	return True, f"MTF confirms LONG: {result.recommendation}"
	elif trade_type == "short" and result.direction == TrendDirection.BEARISH:
	return True, f"MTF confirms SHORT: {result.recommendation}"
	else:
	return False, f"MTF opposes {trade_type.upper()}: Market is {result.direction.value}"

	def get_summary(self, primary_df: pd.DataFrame = None) -> Dict:
	"""Get a summary dict for UI/logging."""
	result = self.get_confluence(primary_df)

	signal_summary = {}
	for tf, signal in result.signals.items():
	signal_summary[tf] = {
	'direction': signal.direction.value,
	'strength': signal.strength,
	'rsi': signal.rsi,
	'emoji': '📈' if signal.direction == TrendDirection.BULLISH else
	'📉' if signal.direction == TrendDirection.BEARISH else '➡️'
	}

	return {
	'aligned': result.aligned,
	'direction': result.direction.value,
	'strength': result.strength,
	'signals': signal_summary,
	'recommendation': result.recommendation
	}