Spaces:

chen470
/

drl-trading-bot-dev2

Paused

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

DRL Trading Bot

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

fc115d5 8 months ago

12.1 kB

"""
Multi-Asset Correlation Engine
Fetches and analyzes correlation across multiple assets:
- USDT Dominance (risk-on/risk-off)
- BTC Dominance (altcoin season indicator)
- ETH/BTC ratio
- Total crypto market cap
"""

import numpy as np
import pandas as pd
from typing import Dict, Optional, Tuple
from datetime import datetime, timedelta
import logging
import os
import requests
import time

logger = logging.getLogger(__name__)


class CorrelationEngine:
    """
    Multi-Asset Correlation Engine for enhanced market analysis.
    
    Features derived:
    - USDT dominance trend and level
    - BTC dominance for altcoin rotation
    - ETH/BTC for ETH relative strength
    - Cross-asset divergences
    """
    
    BINANCE_BASE = os.environ.get("BINANCE_API_URL", "https://data-api.binance.vision/api/v3/klines")
    
    # Symbols for correlation analysis
    CORRELATION_PAIRS = {
        'ETH': 'ETHUSDT',
        'ETH_BTC': 'ETHBTC',  # ETH/BTC ratio
        'BNB': 'BNBUSDT',
        'SOL': 'SOLUSDT',
    }
    
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.session = requests.Session()
        self._cache = {}
        
    def fetch_correlation_data(
        self,
        primary_df: pd.DataFrame,
        timeframe: str = '1h',
    ) -> pd.DataFrame:
        """
        Fetch correlation assets and align with primary DataFrame.
        
        Args:
            primary_df: Primary asset DataFrame (e.g., BTC/USDT)
            timeframe: Timeframe to fetch
            
        Returns:
            DataFrame with correlation features added
        """
        result_df = primary_df.copy()
        
        # Get date range from primary data
        start_time = int(primary_df.index[0].timestamp() * 1000)
        end_time = int(primary_df.index[-1].timestamp() * 1000)
        
        # Fetch each correlation pair
        for name, symbol in self.CORRELATION_PAIRS.items():
            try:
                pair_df = self._fetch_asset(symbol, timeframe, start_time, end_time)
                if pair_df is not None and len(pair_df) > 0:
                    # Align with primary index
                    pair_df = pair_df.reindex(primary_df.index, method='ffill')
                    
                    # Add correlation features
                    result_df[f'{name}_close'] = pair_df['close']
                    result_df[f'{name}_return'] = pair_df['close'].pct_change()
                    
                    logger.debug(f"Added {name} correlation data")
            except Exception as e:
                logger.warning(f"Failed to fetch {name}: {e}")
                # Fill with NaN which will be handled later
                result_df[f'{name}_close'] = np.nan
                result_df[f'{name}_return'] = np.nan
                
        return result_df
    
    def _fetch_asset(
        self,
        symbol: str,
        timeframe: str,
        start_time: int,
        end_time: int,
    ) -> Optional[pd.DataFrame]:
        """Fetch single asset from Binance."""
        interval_map = {
            '1m': '1m', '5m': '5m', '15m': '15m', '30m': '30m',
            '1h': '1h', '4h': '4h', '1d': '1d',
        }
        
        interval = interval_map.get(timeframe, '1h')
        
        all_klines = []
        current_start = start_time
        
        while current_start < end_time:
            params = {
                'symbol': symbol,
                'interval': interval,
                'startTime': current_start,
                'endTime': end_time,
                'limit': 1000,
            }
            
            for attempt in range(self.max_retries):
                try:
                    response = self.session.get(self.BINANCE_BASE, params=params, timeout=30)
                    response.raise_for_status()
                    klines = response.json()
                    
                    if not klines:
                        break
                        
                    all_klines.extend(klines)
                    current_start = klines[-1][0] + 1
                    
                    time.sleep(0.1)  # Rate limiting
                    break
                    
                except Exception as e:
                    if attempt < self.max_retries - 1:
                        time.sleep(1 * (attempt + 1))
                    else:
                        raise
                        
            if not klines:
                break
                
        if not all_klines:
            return None
            
        # Convert to DataFrame
        df = pd.DataFrame(all_klines, 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')
        df.set_index('timestamp', inplace=True)
        
        for col in ['open', 'high', 'low', 'close', 'volume']:
            df[col] = df[col].astype(float)
            
        return df[['open', 'high', 'low', 'close', 'volume']]
    
    def compute_correlation_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]:
        """
        Compute correlation-based features from multi-asset data.
        
        Args:
            df: DataFrame with primary asset + correlation assets
            
        Returns:
            Dictionary of correlation features
        """
        features = {}
        primary_return = df['close'].pct_change()
        
        # ETH/BTC ratio features
        if 'ETH_BTC_close' in df.columns:
            eth_btc = df['ETH_BTC_close']
            features['corr_eth_btc_ratio'] = eth_btc / eth_btc.rolling(20).mean()
            features['corr_eth_btc_trend'] = np.sign(eth_btc.diff(5))
            features['corr_eth_strength'] = (eth_btc - eth_btc.rolling(50).min()) / (
                eth_btc.rolling(50).max() - eth_btc.rolling(50).min() + 1e-10
            )
        
        # ETH correlation
        if 'ETH_return' in df.columns:
            eth_return = df['ETH_return'].fillna(0)
            # Rolling correlation with BTC
            features['corr_eth_btc_corr'] = primary_return.rolling(20).corr(eth_return)
            # Divergence: BTC up, ETH down = potential weakness
            features['corr_eth_divergence'] = np.sign(primary_return) != np.sign(eth_return)
            features['corr_eth_divergence'] = features['corr_eth_divergence'].astype(float)
        
        # SOL correlation (high beta asset)
        if 'SOL_return' in df.columns:
            sol_return = df['SOL_return'].fillna(0)
            features['corr_sol_btc_corr'] = primary_return.rolling(20).corr(sol_return)
            # SOL beta: measures how much SOL moves relative to BTC
            features['corr_sol_beta'] = sol_return.rolling(20).cov(primary_return) / (
                primary_return.rolling(20).var() + 1e-10
            )
        
        # BNB correlation
        if 'BNB_return' in df.columns:
            bnb_return = df['BNB_return'].fillna(0)
            features['corr_bnb_btc_corr'] = primary_return.rolling(20).corr(bnb_return)
        
        # Cross-asset momentum
        altcoin_returns = []
        for col in ['ETH_return', 'SOL_return', 'BNB_return']:
            if col in df.columns:
                altcoin_returns.append(df[col].fillna(0))
                
        if altcoin_returns:
            avg_alt_return = pd.concat(altcoin_returns, axis=1).mean(axis=1)
            features['corr_alt_momentum'] = avg_alt_return.rolling(10).sum()
            # BTC vs Alts: BTC outperforming alts = BTC dominance rising
            features['corr_btc_vs_alts'] = primary_return.rolling(10).sum() - avg_alt_return.rolling(10).sum()
        
        # Simulate USDT dominance (approximation based on market behavior)
        # When BTC and alts dump, USDT.D rises (flight to safety)
        if len(altcoin_returns) > 0:
            avg_market_return = (primary_return + avg_alt_return) / 2
            features['corr_usdt_d_proxy'] = -avg_market_return.rolling(10).sum()  # Inverse of market
            features['corr_risk_off'] = (avg_market_return < -0.01).astype(float).rolling(5).mean()
        
        # Market regime based on correlations
        features['corr_regime'] = self._detect_regime(df, features)
        
        # Fill NaN values
        for key in features:
            if isinstance(features[key], pd.Series):
                features[key] = features[key].fillna(0)
                
        return features
    
    def _detect_regime(self, df: pd.DataFrame, features: Dict[str, pd.Series]) -> pd.Series:
        """
        Detect market regime:
        0 = Neutral
        1 = Risk-On (strong uptrend, high correlation)
        2 = Risk-Off (downtrend, flight to safety)
        3 = Rotation (BTC dominance changing)
        """
        regime = pd.Series(0, index=df.index)
        
        # Price trend
        sma20 = df['close'].rolling(20).mean()
        sma50 = df['close'].rolling(50).mean()
        
        uptrend = (sma20 > sma50) & (df['close'] > sma20)
        downtrend = (sma20 < sma50) & (df['close'] < sma20)
        
        regime[uptrend] = 1  # Risk-On
        regime[downtrend] = 2  # Risk-Off
        
        # Check for rotation (ETH/BTC changing significantly)
        if 'corr_eth_btc_trend' in features:
            eth_rotating = np.abs(features['corr_eth_btc_trend'].rolling(5).sum()) > 3
            regime[eth_rotating & ~uptrend & ~downtrend] = 3
            
        return regime
    
    def get_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]:
        """
        Main method to get all correlation features.
        
        Args:
            df: Primary asset DataFrame
            
        Returns:
            Dictionary of all correlation features
        """
        # First fetch correlation data
        enriched_df = self.fetch_correlation_data(df)
        
        # Then compute features
        return self.compute_correlation_features(enriched_df)


class SimulatedDominanceEngine:
    """
    Simulated Dominance metrics when real dominance data is unavailable.
    
    Approximates:
    - USDT.D behavior (inverse of risk sentiment)
    - BTC.D behavior (BTC vs alts performance)
    """
    
    def __init__(self):
        pass
        
    def compute_simulated_dominance(
        self,
        btc_df: pd.DataFrame,
        alt_returns: Optional[pd.Series] = None,
    ) -> Dict[str, pd.Series]:
        """Compute simulated dominance features."""
        features = {}
        
        btc_return = btc_df['close'].pct_change()
        
        # Simulated USDT Dominance
        # When market drops, USDT.D rises (flight to safety)
        market_momentum = btc_return.rolling(10).sum()
        features['sim_usdt_d'] = -market_momentum  # Inverse
        features['sim_usdt_d_rising'] = (features['sim_usdt_d'].diff(5) > 0).astype(float)
        
        # Simulated BTC Dominance
        if alt_returns is not None:
            # BTC.D rises when BTC outperforms alts
            btc_vs_alt = btc_return - alt_returns
            features['sim_btc_d'] = btc_vs_alt.rolling(10).sum()
            features['sim_btc_d_rising'] = (features['sim_btc_d'].diff(5) > 0).astype(float)
        else:
            # Without alt data, use volatility as proxy
            # High volatility periods often see BTC.D rise (flight to BTC)
            vol = btc_return.rolling(10).std()
            vol_ma = vol.rolling(20).mean()
            features['sim_btc_d'] = (vol > vol_ma).astype(float)
            features['sim_btc_d_rising'] = features['sim_btc_d'].diff(5) > 0
        
        # Risk sentiment
        features['sim_risk_sentiment'] = np.tanh(market_momentum * 10)  # -1 to 1
        features['sim_fear'] = (features['sim_risk_sentiment'] < -0.3).astype(float)
        features['sim_greed'] = (features['sim_risk_sentiment'] > 0.3).astype(float)
        
        return features

	"""
	Multi-Asset Correlation Engine

	Fetches and analyzes correlation across multiple assets:
	- USDT Dominance (risk-on/risk-off)
	- BTC Dominance (altcoin season indicator)
	- ETH/BTC ratio
	- Total crypto market cap
	"""

	import numpy as np
	import pandas as pd
	from typing import Dict, Optional, Tuple
	from datetime import datetime, timedelta
	import logging
	import os
	import requests
	import time

	logger = logging.getLogger(__name__)


	class CorrelationEngine:
	"""
	Multi-Asset Correlation Engine for enhanced market analysis.

	Features derived:
	- USDT dominance trend and level
	- BTC dominance for altcoin rotation
	- ETH/BTC for ETH relative strength
	- Cross-asset divergences
	"""

	BINANCE_BASE = os.environ.get("BINANCE_API_URL", "https://data-api.binance.vision/api/v3/klines")

	# Symbols for correlation analysis
	CORRELATION_PAIRS = {
	'ETH': 'ETHUSDT',
	'ETH_BTC': 'ETHBTC', # ETH/BTC ratio
	'BNB': 'BNBUSDT',
	'SOL': 'SOLUSDT',
	}

	def __init__(self, max_retries: int = 3):
	self.max_retries = max_retries
	self.session = requests.Session()
	self._cache = {}

	def fetch_correlation_data(
	self,
	primary_df: pd.DataFrame,
	timeframe: str = '1h',
	) -> pd.DataFrame:
	"""
	Fetch correlation assets and align with primary DataFrame.

	Args:
	primary_df: Primary asset DataFrame (e.g., BTC/USDT)
	timeframe: Timeframe to fetch

	Returns:
	DataFrame with correlation features added
	"""
	result_df = primary_df.copy()

	# Get date range from primary data
	start_time = int(primary_df.index[0].timestamp() * 1000)
	end_time = int(primary_df.index[-1].timestamp() * 1000)

	# Fetch each correlation pair
	for name, symbol in self.CORRELATION_PAIRS.items():
	try:
	pair_df = self._fetch_asset(symbol, timeframe, start_time, end_time)
	if pair_df is not None and len(pair_df) > 0:
	# Align with primary index
	pair_df = pair_df.reindex(primary_df.index, method='ffill')

	# Add correlation features
	result_df[f'{name}_close'] = pair_df['close']
	result_df[f'{name}_return'] = pair_df['close'].pct_change()

	logger.debug(f"Added {name} correlation data")
	except Exception as e:
	logger.warning(f"Failed to fetch {name}: {e}")
	# Fill with NaN which will be handled later
	result_df[f'{name}_close'] = np.nan
	result_df[f'{name}_return'] = np.nan

	return result_df

	def _fetch_asset(
	self,
	symbol: str,
	timeframe: str,
	start_time: int,
	end_time: int,
	) -> Optional[pd.DataFrame]:
	"""Fetch single asset from Binance."""
	interval_map = {
	'1m': '1m', '5m': '5m', '15m': '15m', '30m': '30m',
	'1h': '1h', '4h': '4h', '1d': '1d',
	}

	interval = interval_map.get(timeframe, '1h')

	all_klines = []
	current_start = start_time

	while current_start < end_time:
	params = {
	'symbol': symbol,
	'interval': interval,
	'startTime': current_start,
	'endTime': end_time,
	'limit': 1000,
	}

	for attempt in range(self.max_retries):
	try:
	response = self.session.get(self.BINANCE_BASE, params=params, timeout=30)
	response.raise_for_status()
	klines = response.json()

	if not klines:
	break

	all_klines.extend(klines)
	current_start = klines[-1][0] + 1

	time.sleep(0.1) # Rate limiting
	break

	except Exception as e:
	if attempt < self.max_retries - 1:
	time.sleep(1 * (attempt + 1))
	else:
	raise

	if not klines:
	break

	if not all_klines:
	return None

	# Convert to DataFrame
	df = pd.DataFrame(all_klines, 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')
	df.set_index('timestamp', inplace=True)

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

	return df[['open', 'high', 'low', 'close', 'volume']]

	def compute_correlation_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]:
	"""
	Compute correlation-based features from multi-asset data.

	Args:
	df: DataFrame with primary asset + correlation assets

	Returns:
	Dictionary of correlation features
	"""
	features = {}
	primary_return = df['close'].pct_change()

	# ETH/BTC ratio features
	if 'ETH_BTC_close' in df.columns:
	eth_btc = df['ETH_BTC_close']
	features['corr_eth_btc_ratio'] = eth_btc / eth_btc.rolling(20).mean()
	features['corr_eth_btc_trend'] = np.sign(eth_btc.diff(5))
	features['corr_eth_strength'] = (eth_btc - eth_btc.rolling(50).min()) / (
	eth_btc.rolling(50).max() - eth_btc.rolling(50).min() + 1e-10
	)

	# ETH correlation
	if 'ETH_return' in df.columns:
	eth_return = df['ETH_return'].fillna(0)
	# Rolling correlation with BTC
	features['corr_eth_btc_corr'] = primary_return.rolling(20).corr(eth_return)
	# Divergence: BTC up, ETH down = potential weakness
	features['corr_eth_divergence'] = np.sign(primary_return) != np.sign(eth_return)
	features['corr_eth_divergence'] = features['corr_eth_divergence'].astype(float)

	# SOL correlation (high beta asset)
	if 'SOL_return' in df.columns:
	sol_return = df['SOL_return'].fillna(0)
	features['corr_sol_btc_corr'] = primary_return.rolling(20).corr(sol_return)
	# SOL beta: measures how much SOL moves relative to BTC
	features['corr_sol_beta'] = sol_return.rolling(20).cov(primary_return) / (
	primary_return.rolling(20).var() + 1e-10
	)

	# BNB correlation
	if 'BNB_return' in df.columns:
	bnb_return = df['BNB_return'].fillna(0)
	features['corr_bnb_btc_corr'] = primary_return.rolling(20).corr(bnb_return)

	# Cross-asset momentum
	altcoin_returns = []
	for col in ['ETH_return', 'SOL_return', 'BNB_return']:
	if col in df.columns:
	altcoin_returns.append(df[col].fillna(0))

	if altcoin_returns:
	avg_alt_return = pd.concat(altcoin_returns, axis=1).mean(axis=1)
	features['corr_alt_momentum'] = avg_alt_return.rolling(10).sum()
	# BTC vs Alts: BTC outperforming alts = BTC dominance rising
	features['corr_btc_vs_alts'] = primary_return.rolling(10).sum() - avg_alt_return.rolling(10).sum()

	# Simulate USDT dominance (approximation based on market behavior)
	# When BTC and alts dump, USDT.D rises (flight to safety)
	if len(altcoin_returns) > 0:
	avg_market_return = (primary_return + avg_alt_return) / 2
	features['corr_usdt_d_proxy'] = -avg_market_return.rolling(10).sum() # Inverse of market
	features['corr_risk_off'] = (avg_market_return < -0.01).astype(float).rolling(5).mean()

	# Market regime based on correlations
	features['corr_regime'] = self._detect_regime(df, features)

	# Fill NaN values
	for key in features:
	if isinstance(features[key], pd.Series):
	features[key] = features[key].fillna(0)

	return features

	def _detect_regime(self, df: pd.DataFrame, features: Dict[str, pd.Series]) -> pd.Series:
	"""
	Detect market regime:
	0 = Neutral
	1 = Risk-On (strong uptrend, high correlation)
	2 = Risk-Off (downtrend, flight to safety)
	3 = Rotation (BTC dominance changing)
	"""
	regime = pd.Series(0, index=df.index)

	# Price trend
	sma20 = df['close'].rolling(20).mean()
	sma50 = df['close'].rolling(50).mean()

	uptrend = (sma20 > sma50) & (df['close'] > sma20)
	downtrend = (sma20 < sma50) & (df['close'] < sma20)

	regime[uptrend] = 1 # Risk-On
	regime[downtrend] = 2 # Risk-Off

	# Check for rotation (ETH/BTC changing significantly)
	if 'corr_eth_btc_trend' in features:
	eth_rotating = np.abs(features['corr_eth_btc_trend'].rolling(5).sum()) > 3
	regime[eth_rotating & ~uptrend & ~downtrend] = 3

	return regime

	def get_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]:
	"""
	Main method to get all correlation features.

	Args:
	df: Primary asset DataFrame

	Returns:
	Dictionary of all correlation features
	"""
	# First fetch correlation data
	enriched_df = self.fetch_correlation_data(df)

	# Then compute features
	return self.compute_correlation_features(enriched_df)


	class SimulatedDominanceEngine:
	"""
	Simulated Dominance metrics when real dominance data is unavailable.

	Approximates:
	- USDT.D behavior (inverse of risk sentiment)
	- BTC.D behavior (BTC vs alts performance)
	"""

	def __init__(self):
	pass

	def compute_simulated_dominance(
	self,
	btc_df: pd.DataFrame,
	alt_returns: Optional[pd.Series] = None,
	) -> Dict[str, pd.Series]:
	"""Compute simulated dominance features."""
	features = {}

	btc_return = btc_df['close'].pct_change()

	# Simulated USDT Dominance
	# When market drops, USDT.D rises (flight to safety)
	market_momentum = btc_return.rolling(10).sum()
	features['sim_usdt_d'] = -market_momentum # Inverse
	features['sim_usdt_d_rising'] = (features['sim_usdt_d'].diff(5) > 0).astype(float)

	# Simulated BTC Dominance
	if alt_returns is not None:
	# BTC.D rises when BTC outperforms alts
	btc_vs_alt = btc_return - alt_returns
	features['sim_btc_d'] = btc_vs_alt.rolling(10).sum()
	features['sim_btc_d_rising'] = (features['sim_btc_d'].diff(5) > 0).astype(float)
	else:
	# Without alt data, use volatility as proxy
	# High volatility periods often see BTC.D rise (flight to BTC)
	vol = btc_return.rolling(10).std()
	vol_ma = vol.rolling(20).mean()
	features['sim_btc_d'] = (vol > vol_ma).astype(float)
	features['sim_btc_d_rising'] = features['sim_btc_d'].diff(5) > 0

	# Risk sentiment
	features['sim_risk_sentiment'] = np.tanh(market_momentum * 10) # -1 to 1
	features['sim_fear'] = (features['sim_risk_sentiment'] < -0.3).astype(float)
	features['sim_greed'] = (features['sim_risk_sentiment'] > 0.3).astype(float)

	return features