""" Ultimate Feature Engine Comprehensive feature engineering for professional-grade trading. Features: - Wyckoff Phase Detection - Smart Money Concepts (Order Blocks, FVG, BOS, CHOCH) - Market Structure Analysis - Multi-Timeframe Analysis - Volume Profile Analysis """ import numpy as np import pandas as pd from typing import Dict, List, Tuple, Optional import logging logger = logging.getLogger(__name__) class WyckoffAnalyzer: """ Wyckoff Phase Analysis Detects accumulation/distribution phases and key Wyckoff events: - Selling Climax (SC) / Buying Climax (BC) - Automatic Rally (AR) / Automatic Reaction - Secondary Test (ST) - Spring / Upthrust - Sign of Strength (SOS) / Sign of Weakness (SOW) - Last Point of Support (LPS) / Last Point of Supply (LPSY) """ def __init__(self, lookback: int = 50): self.lookback = lookback def detect_climax(self, df: pd.DataFrame) -> pd.Series: """Detect Selling/Buying Climax based on volume and price action.""" volume_ma = df['volume'].rolling(20).mean() volume_std = df['volume'].rolling(20).std() # High volume spike (> 2 std above mean) volume_spike = df['volume'] > (volume_ma + 2 * volume_std) # Price range expansion price_range = df['high'] - df['low'] range_ma = price_range.rolling(20).mean() wide_range = price_range > (range_ma * 1.5) # Selling climax: high volume + wide range + closes near low close_near_low = (df['close'] - df['low']) / (df['high'] - df['low'] + 1e-10) < 0.3 selling_climax = volume_spike & wide_range & close_near_low # Buying climax: high volume + wide range + closes near high close_near_high = (df['close'] - df['low']) / (df['high'] - df['low'] + 1e-10) > 0.7 buying_climax = volume_spike & wide_range & close_near_high # 1 = selling climax, -1 = buying climax, 0 = none climax = pd.Series(0, index=df.index) climax[selling_climax] = 1 climax[buying_climax] = -1 return climax def detect_spring_upthrust(self, df: pd.DataFrame) -> Tuple[pd.Series, pd.Series]: """Detect Spring (false breakdown) and Upthrust (false breakout).""" # Find local lows and highs window = 10 local_low = df['low'].rolling(window, center=True).min() local_high = df['high'].rolling(window, center=True).max() # Support/Resistance as rolling min/max of last N periods support = df['low'].rolling(self.lookback).min() resistance = df['high'].rolling(self.lookback).max() # Spring: breaks below support but closes above it spring = (df['low'] < support.shift(1)) & (df['close'] > support.shift(1)) # Upthrust: breaks above resistance but closes below it upthrust = (df['high'] > resistance.shift(1)) & (df['close'] < resistance.shift(1)) return spring.astype(float), upthrust.astype(float) def detect_phase(self, df: pd.DataFrame) -> pd.Series: """ Classify Wyckoff phase (simplified): 0 = Unknown/Ranging 1 = Accumulation (Phase A-C) 2 = Markup 3 = Distribution (Phase A-C) 4 = Markdown """ # Calculate trend sma20 = df['close'].rolling(20).mean() sma50 = df['close'].rolling(50).mean() # Volume analysis vol_ma = df['volume'].rolling(20).mean() vol_increasing = df['volume'] > vol_ma # Price momentum roc = df['close'].pct_change(10) # Volatility (shrinking = accumulation/distribution) atr = self._calculate_atr(df, 14) atr_ma = atr.rolling(20).mean() low_volatility = atr < atr_ma * 0.8 phase = pd.Series(0, index=df.index) # Markup: uptrend with expanding volume markup = (sma20 > sma50) & (roc > 0.02) phase[markup] = 2 # Markdown: downtrend markdown = (sma20 < sma50) & (roc < -0.02) phase[markdown] = 4 # Accumulation: low volatility after downtrend, volume increasing accum = low_volatility & (sma20 < sma50) & vol_increasing phase[accum] = 1 # Distribution: low volatility after uptrend distrib = low_volatility & (sma20 > sma50) & vol_increasing phase[distrib] = 3 return phase def _calculate_atr(self, df: pd.DataFrame, period: int = 14) -> pd.Series: """Calculate Average True Range.""" high_low = df['high'] - df['low'] high_close = np.abs(df['high'] - df['close'].shift()) low_close = np.abs(df['low'] - df['close'].shift()) tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) atr = tr.rolling(period).mean() return atr def get_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Get all Wyckoff features.""" climax = self.detect_climax(df) spring, upthrust = self.detect_spring_upthrust(df) phase = self.detect_phase(df) return { 'wyckoff_climax': climax, 'wyckoff_spring': spring, 'wyckoff_upthrust': upthrust, 'wyckoff_phase': phase, 'wyckoff_accumulation': (phase == 1).astype(float), 'wyckoff_distribution': (phase == 3).astype(float), 'wyckoff_markup': (phase == 2).astype(float), 'wyckoff_markdown': (phase == 4).astype(float), } class SMCAnalyzer: """ Smart Money Concepts (SMC) Analysis Detects: - Order Blocks (demand/supply zones) - Fair Value Gaps (FVG) - Break of Structure (BOS) - Change of Character (CHOCH) - Liquidity pools """ def __init__(self, swing_lookback: int = 5): self.swing_lookback = swing_lookback def detect_swing_points(self, df: pd.DataFrame) -> Tuple[pd.Series, pd.Series]: """Detect swing highs and lows.""" window = self.swing_lookback # Swing high: high is highest in window swing_high = df['high'] == df['high'].rolling(window * 2 + 1, center=True).max() # Swing low: low is lowest in window swing_low = df['low'] == df['low'].rolling(window * 2 + 1, center=True).min() return swing_high.astype(float), swing_low.astype(float) def detect_order_blocks(self, df: pd.DataFrame) -> Tuple[pd.Series, pd.Series]: """ Detect Order Blocks (OB): - Bullish OB: last bearish candle before impulsive bullish move - Bearish OB: last bullish candle before impulsive bearish move """ # Calculate candle direction bullish = df['close'] > df['open'] bearish = df['close'] < df['open'] # Impulsive move: large body relative to ATR body = np.abs(df['close'] - df['open']) atr = self._calculate_atr(df, 14) impulsive = body > atr * 1.5 # Bullish OB: bearish candle followed by impulsive bullish bullish_ob = bearish.shift(1) & bullish & impulsive # Bearish OB: bullish candle followed by impulsive bearish bearish_ob = bullish.shift(1) & bearish & impulsive return bullish_ob.astype(float), bearish_ob.astype(float) def detect_fvg(self, df: pd.DataFrame) -> Tuple[pd.Series, pd.Series]: """ Detect Fair Value Gaps (FVG): - Bullish FVG: gap between candle 1 high and candle 3 low - Bearish FVG: gap between candle 1 low and candle 3 high """ # Bullish FVG: candle[i-2].high < candle[i].low bullish_fvg = df['high'].shift(2) < df['low'] # Bearish FVG: candle[i-2].low > candle[i].high bearish_fvg = df['low'].shift(2) > df['high'] return bullish_fvg.astype(float), bearish_fvg.astype(float) def detect_bos_choch(self, df: pd.DataFrame) -> Tuple[pd.Series, pd.Series]: """ Detect Break of Structure (BOS) and Change of Character (CHOCH): - BOS: break of previous swing high/low in trend direction - CHOCH: break of previous swing in opposite direction (trend reversal) """ swing_high, swing_low = self.detect_swing_points(df) # Track previous swing levels prev_swing_high = df['high'].where(swing_high.astype(bool)).ffill() prev_swing_low = df['low'].where(swing_low.astype(bool)).ffill() # BOS bullish: break above previous swing high bos_bullish = df['close'] > prev_swing_high.shift(1) # BOS bearish: break below previous swing low bos_bearish = df['close'] < prev_swing_low.shift(1) # Simple trend tracking sma20 = df['close'].rolling(20).mean() uptrend = df['close'] > sma20 downtrend = df['close'] < sma20 # CHOCH: BOS in opposite direction of trend choch_bullish = bos_bullish & downtrend.shift(1) choch_bearish = bos_bearish & uptrend.shift(1) bos = pd.Series(0, index=df.index) bos[bos_bullish] = 1 bos[bos_bearish] = -1 choch = pd.Series(0, index=df.index) choch[choch_bullish] = 1 choch[choch_bearish] = -1 return bos, choch def detect_liquidity(self, df: pd.DataFrame) -> Tuple[pd.Series, pd.Series]: """Detect liquidity pools (clusters of swing highs/lows).""" swing_high, swing_low = self.detect_swing_points(df) # Count nearby swing points window = 20 liquidity_above = swing_high.rolling(window).sum() liquidity_below = swing_low.rolling(window).sum() return liquidity_above, liquidity_below def _calculate_atr(self, df: pd.DataFrame, period: int = 14) -> pd.Series: """Calculate Average True Range.""" high_low = df['high'] - df['low'] high_close = np.abs(df['high'] - df['close'].shift()) low_close = np.abs(df['low'] - df['close'].shift()) tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) atr = tr.rolling(period).mean() return atr def get_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Get all SMC features.""" swing_high, swing_low = self.detect_swing_points(df) bullish_ob, bearish_ob = self.detect_order_blocks(df) bullish_fvg, bearish_fvg = self.detect_fvg(df) bos, choch = self.detect_bos_choch(df) liq_above, liq_below = self.detect_liquidity(df) return { 'smc_swing_high': swing_high, 'smc_swing_low': swing_low, 'smc_bullish_ob': bullish_ob, 'smc_bearish_ob': bearish_ob, 'smc_bullish_fvg': bullish_fvg, 'smc_bearish_fvg': bearish_fvg, 'smc_bos': bos, 'smc_choch': choch, 'smc_liquidity_above': liq_above, 'smc_liquidity_below': liq_below, } class MarketStructureAnalyzer: """ Market Structure Analysis Detects: - Higher Highs / Higher Lows (HH/HL) - Uptrend - Lower Highs / Lower Lows (LH/LL) - Downtrend - Support / Resistance levels - Trend strength """ def __init__(self, pivot_lookback: int = 5): self.pivot_lookback = pivot_lookback def detect_structure(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Detect market structure (HH, HL, LH, LL).""" window = self.pivot_lookback # Find pivot points pivot_high = df['high'] == df['high'].rolling(window * 2 + 1, center=True).max() pivot_low = df['low'] == df['low'].rolling(window * 2 + 1, center=True).min() # Track previous pivots prev_pivot_high = df['high'].where(pivot_high).ffill() prev_pivot_low = df['low'].where(pivot_low).ffill() # Higher High: current pivot high > previous pivot high hh = pivot_high & (df['high'] > prev_pivot_high.shift(1)) # Higher Low: current pivot low > previous pivot low hl = pivot_low & (df['low'] > prev_pivot_low.shift(1)) # Lower High: current pivot high < previous pivot high lh = pivot_high & (df['high'] < prev_pivot_high.shift(1)) # Lower Low: current pivot low < previous pivot low ll = pivot_low & (df['low'] < prev_pivot_low.shift(1)) return { 'structure_hh': hh.astype(float), 'structure_hl': hl.astype(float), 'structure_lh': lh.astype(float), 'structure_ll': ll.astype(float), } def detect_trend(self, df: pd.DataFrame) -> pd.Series: """ Detect trend based on structure: 1 = Uptrend (HH + HL) -1 = Downtrend (LH + LL) 0 = Ranging """ structure = self.detect_structure(df) # Count HH/HL vs LH/LL in rolling window window = 20 bullish_count = ( structure['structure_hh'].rolling(window).sum() + structure['structure_hl'].rolling(window).sum() ) bearish_count = ( structure['structure_lh'].rolling(window).sum() + structure['structure_ll'].rolling(window).sum() ) trend = pd.Series(0, index=df.index) trend[bullish_count > bearish_count + 1] = 1 trend[bearish_count > bullish_count + 1] = -1 return trend def detect_sr_levels(self, df: pd.DataFrame) -> Tuple[pd.Series, pd.Series]: """Detect Support/Resistance levels.""" window = 50 # Rolling support/resistance support = df['low'].rolling(window).min() resistance = df['high'].rolling(window).max() # Distance to S/R (normalized) range_size = resistance - support + 1e-10 dist_to_support = (df['close'] - support) / range_size dist_to_resistance = (resistance - df['close']) / range_size return dist_to_support, dist_to_resistance def get_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Get all market structure features.""" structure = self.detect_structure(df) trend = self.detect_trend(df) dist_support, dist_resistance = self.detect_sr_levels(df) features = structure.copy() features['structure_trend'] = trend features['structure_dist_support'] = dist_support features['structure_dist_resistance'] = dist_resistance # Trend strength sma_short = df['close'].rolling(10).mean() sma_long = df['close'].rolling(50).mean() features['structure_trend_strength'] = (sma_short - sma_long) / sma_long return features class VolumeProfileAnalyzer: """ Volume Profile Analysis Features: - Point of Control (POC) - Value Area High/Low (VAH/VAL) - VWAP - Volume at price zones """ def __init__(self, lookback: int = 50, num_bins: int = 20): self.lookback = lookback self.num_bins = num_bins def calculate_vwap(self, df: pd.DataFrame) -> pd.Series: """Calculate VWAP.""" typical_price = (df['high'] + df['low'] + df['close']) / 3 cumulative_tp_vol = (typical_price * df['volume']).cumsum() cumulative_vol = df['volume'].cumsum() vwap = cumulative_tp_vol / cumulative_vol return vwap def calculate_volume_profile(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Calculate volume profile metrics.""" vwap = self.calculate_vwap(df) # Price position relative to VWAP vwap_distance = (df['close'] - vwap) / vwap above_vwap = (df['close'] > vwap).astype(float) # Volume moving averages vol_sma = df['volume'].rolling(20).mean() vol_ratio = df['volume'] / vol_sma # On-balance volume obv = (np.sign(df['close'].diff()) * df['volume']).cumsum() obv_normalized = (obv - obv.rolling(50).mean()) / (obv.rolling(50).std() + 1e-10) # Volume price trend vpt = (df['close'].pct_change() * df['volume']).cumsum() vpt_normalized = (vpt - vpt.rolling(50).mean()) / (vpt.rolling(50).std() + 1e-10) # Accumulation/Distribution Line mfm = ((df['close'] - df['low']) - (df['high'] - df['close'])) / (df['high'] - df['low'] + 1e-10) adl = (mfm * df['volume']).cumsum() adl_normalized = (adl - adl.rolling(50).mean()) / (adl.rolling(50).std() + 1e-10) return { 'volume_vwap_distance': vwap_distance, 'volume_above_vwap': above_vwap, 'volume_ratio': vol_ratio.clip(0, 5), # Cap extreme values 'volume_obv': obv_normalized, 'volume_vpt': vpt_normalized, 'volume_adl': adl_normalized, } def get_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Get all volume profile features.""" return self.calculate_volume_profile(df) class UltimateFeatureEngine: """ Ultimate Feature Engine combining all analysis methods. Total features: ~150+ """ def __init__(self, offline_mode: bool = False): """ Args: offline_mode: If True, skip any features that require live API calls (cross-chain whale flow). Use this during training to avoid fetching live Binance data for every environment instance. """ self.offline_mode = offline_mode self.wyckoff = WyckoffAnalyzer() self.smc = SMCAnalyzer() self.structure = MarketStructureAnalyzer() self.volume = VolumeProfileAnalyzer() def compute_features(self, df: pd.DataFrame) -> np.ndarray: """Compute all features and return as numpy array.""" features_dict = self.get_all_features(df) # Convert to DataFrame and then numpy array features_df = pd.DataFrame(features_dict) # Fill NaN with 0 features_df = features_df.fillna(0) # Replace inf with large values features_df = features_df.replace([np.inf, -np.inf], 0) return features_df.values def get_all_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Get all features as dictionary.""" all_features = {} # 1. Basic price features all_features.update(self._get_price_features(df)) # 2. Technical indicators all_features.update(self._get_technical_features(df)) # 3. Wyckoff features all_features.update(self.wyckoff.get_features(df)) # 4. SMC features all_features.update(self.smc.get_features(df)) # 5. Market structure features all_features.update(self.structure.get_features(df)) # 6. Volume profile features all_features.update(self.volume.get_features(df)) # 7. Whale-proxy features (approximate institutional behavior) all_features.update(self._get_whale_proxy_features(df)) # 8. NEW: Explicit Whale Action Vectors (On-Chain Proxies) all_features.update(self._get_whale_action_vectors(df)) # 9. Cross-Chain Whale Flow Features all_features.update(self._get_cross_chain_features(df)) return all_features def _get_whale_action_vectors(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """ Calculates explicit Whale Action Vectors simulating on-chain network data. These directly mimic the signals from the WhalePatternPredictor model. """ features = {} vol_ma = df['volume'].rolling(20).mean() price_range = df['high'] - df['low'] # 1. Stealth Accumulation (Whales buying quietly without pumping price) # Condition: High volume, but very small price range and close near open small_body = np.abs(df['close'] - df['open']) < (price_range * 0.2) stealth_vol = df['volume'] > vol_ma features['whale_stealth_accumulation'] = (small_body & stealth_vol).astype(float).rolling(10).mean() # NOTE: Removed whale_capitulation_index (Sharpe -8.5, toxic feature) # NOTE: Removed whale_fomo_index (Sharpe -5.4, toxic feature) # 4. Large Tx Ratio Proxy (Percentage of volume happening in macro-moves) large_move = price_range > price_range.rolling(20).mean() * 1.5 macro_vol = df['volume'].where(large_move, 0) features['whale_large_tx_ratio'] = (macro_vol.rolling(10).sum() / (df['volume'].rolling(10).sum() + 1e-10)).clip(0, 1) # 5. Net Flow Proxy (Institutional Net Buying/Selling Pressure) # Buy volume proxy = total volume * (close - low) / range buy_vol_proxy = df['volume'] * ((df['close'] - df['low']) / (price_range + 1e-10)) sell_vol_proxy = df['volume'] * ((df['high'] - df['close']) / (price_range + 1e-10)) net_flow_raw = buy_vol_proxy - sell_vol_proxy features['whale_net_flow_proxy'] = (net_flow_raw - net_flow_raw.rolling(50).mean()) / (net_flow_raw.rolling(50).std() + 1e-10) return features def _get_cross_chain_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """ Cross-chain whale flow features. These features capture capital rotation and consensus across chains. They are computed once and broadcast to all timesteps in the dataframe. In offline_mode (training), returns zeros immediately without making any live API calls. The feature names are preserved so the observation space stays identical between training and live inference. """ features = {} n_rows = len(df) if self.offline_mode: features['whale_eth_sol_rotation'] = pd.Series([0.0] * n_rows, index=df.index) features['whale_stablecoin_flow'] = pd.Series([0.0] * n_rows, index=df.index) features['whale_cross_chain_consensus'] = pd.Series([0.0] * n_rows, index=df.index) features['whale_unified_signal'] = pd.Series([0.0] * n_rows, index=df.index) return features try: from src.features.cross_chain_whale_flow import get_cross_chain_analyzer # Get cross-chain analyzer analyzer = get_cross_chain_analyzer() # Compute cross-chain features (returns dict with scalar values) cc_features = analyzer.compute_cross_chain_features() # Broadcast scalar values to match dataframe length n_rows = len(df) # Add as constant features (same value for all timesteps) features['whale_eth_sol_rotation'] = pd.Series([cc_features.get('eth_to_sol_flow_ratio', 0.0)] * n_rows, index=df.index) features['whale_stablecoin_flow'] = pd.Series([cc_features.get('total_stablecoin_flow', 0.0)] * n_rows, index=df.index) features['whale_cross_chain_consensus'] = pd.Series([cc_features.get('cross_chain_consensus', 0.0)] * n_rows, index=df.index) features['whale_unified_signal'] = pd.Series([cc_features.get('unified_signal', 0.0)] * n_rows, index=df.index) logger.debug( f"Cross-chain features: " f"rotation={cc_features.get('eth_to_sol_flow_ratio', 0):+.2f}, " f"consensus={cc_features.get('cross_chain_consensus', 0):.2f}, " f"unified={cc_features.get('unified_signal', 0):+.2f}" ) except Exception as e: # If cross-chain analyzer fails, return zero features logger.warning(f"Cross-chain features unavailable: {e}") n_rows = len(df) features['whale_eth_sol_rotation'] = pd.Series([0.0] * n_rows, index=df.index) features['whale_stablecoin_flow'] = pd.Series([0.0] * n_rows, index=df.index) features['whale_cross_chain_consensus'] = pd.Series([0.0] * n_rows, index=df.index) features['whale_unified_signal'] = pd.Series([0.0] * n_rows, index=df.index) return features def _get_whale_proxy_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """ Whale-proxy features that approximate institutional/whale behavior. These are derived from OHLCV data to identify large player activity. """ features = {} # 1. Whale Volume Spike - large player activity detection vol_ma = df['volume'].rolling(20).mean() vol_std = df['volume'].rolling(20).std() features['whale_volume_spike'] = (df['volume'] / (vol_ma + 1e-10)).clip(0, 5) features['whale_volume_zscore'] = ((df['volume'] - vol_ma) / (vol_std + 1e-10)).clip(-3, 3) # 2. Accumulation/Distribution Detection # Accumulation: close > open on high volume (institutional buying) # Distribution: close < open on high volume (institutional selling) body_up = (df['close'] > df['open']).astype(float) body_down = (df['close'] < df['open']).astype(float) high_volume = (df['volume'] > vol_ma * 1.5).astype(float) # NOTE: Removed whale_accumulation (redundant) and whale_distribution (Sharpe -6.4, toxic) # Compute accumulation_dist_ratio directly (this feature is kept - statistically significant) whale_acc = (body_up * high_volume).rolling(5).sum() / 5 whale_dist = (body_down * high_volume).rolling(5).sum() / 5 features['whale_accumulation_dist_ratio'] = (whale_acc - whale_dist).clip(-1, 1) # 3. Crowd Sentiment Proxy (approximates Fear & Greed) # Combines RSI extremes with volume for sentiment signal rsi = self._calculate_rsi(df['close'], 14) rsi_extreme_greed = (rsi > 70).astype(float) # NOTE: Removed whale_crowd_fear (Sharpe -7.4, toxic feature) # Keep whale_crowd_greed - Win Rate 54.5% (best whale feature!) features['whale_crowd_greed'] = (rsi_extreme_greed * high_volume).rolling(5).mean() # Sentiment score: -1 (fear) to +1 (greed) features['whale_sentiment_score'] = ( (rsi / 100 - 0.5) * 2 ).rolling(10).mean().clip(-1, 1) # 4. Smart Money Divergence # Price vs volume divergence detection price_direction = np.sign(df['close'].diff(5)) volume_direction = np.sign(df['volume'].diff(5)) # Positive divergence: price down but volume up (smart money buying) features['whale_positive_divergence'] = ( ((price_direction < 0) & (volume_direction > 0)).astype(float) ).rolling(5).mean() # Negative divergence: price up but volume down (distribution) features['whale_negative_divergence'] = ( ((price_direction > 0) & (volume_direction < 0)).astype(float) ).rolling(5).mean() # 5. Open Interest Proxy (volatility expansion/contraction) # High vol expansion = positions being opened, contraction = being closed returns = df['close'].pct_change() volatility = returns.rolling(14).std() vol_expanding = volatility > volatility.rolling(50).mean() vol_contracting = volatility < volatility.rolling(50).mean() * 0.7 features['whale_oi_proxy_expanding'] = vol_expanding.astype(float) features['whale_oi_proxy_contracting'] = vol_contracting.astype(float) # 6. Long/Short Ratio Proxy # Based on candle close position and volume # Close in upper half = more longs, lower half = more shorts close_position = (df['close'] - df['low']) / (df['high'] - df['low'] + 1e-10) volume_weighted_position = close_position * df['volume'] features['whale_ls_ratio_proxy'] = ( volume_weighted_position.rolling(10).mean() / (df['volume'].rolling(10).mean() + 1e-10) - 0.5 ) * 2 # Scale to -1 to +1 # 7. Top Trader Proxy (large candle with high volume) # Big players move price significantly with volume price_range = df['high'] - df['low'] range_ma = price_range.rolling(20).mean() large_range = price_range > range_ma * 1.5 features['whale_large_player_activity'] = ( (large_range & (df['volume'] > vol_ma * 1.5)).astype(float) ).rolling(5).mean() # 8. Institutional vs Retail Flow Proxy # Large moves during "smart money" hours approximation # Using volume patterns as proxy features['whale_institutional_flow'] = ( (df['volume'] / (vol_ma + 1e-10)) * np.abs(df['close'] - df['open']) / (price_range + 1e-10) ).rolling(5).mean().clip(0, 3) # ===== FUNDING RATE PROXY FEATURES ===== # Funding rate correlates with price premium and market imbalance # 9. Price Premium Proxy (basis approximation) # When price is above recent average, funding tends to be positive ema_20 = df['close'].ewm(span=20).mean() ema_50 = df['close'].ewm(span=50).mean() features['funding_premium_proxy'] = ((df['close'] - ema_20) / ema_20 * 100).clip(-2, 2) # 10. Momentum-based funding proxy # Strong uptrends = positive funding, strong downtrends = negative momentum_10 = df['close'].pct_change(10) momentum_20 = df['close'].pct_change(20) features['funding_momentum_proxy'] = (momentum_10 + momentum_20).clip(-0.1, 0.1) * 10 # 11. Funding rate extreme detector # Extreme premium = potential reversal extreme_premium = np.abs(features['funding_premium_proxy']) > 1.5 features['funding_extreme'] = extreme_premium.astype(float) # ===== ORDER FLOW PROXY FEATURES ===== # 12. CVD Proxy from candle analysis body = df['close'] - df['open'] range_total = df['high'] - df['low'] body_ratio = body / (range_total + 1e-10) raw_cvd = (body_ratio * df['volume']).cumsum() # Normalize to recent range features['orderflow_cvd'] = ((raw_cvd - raw_cvd.rolling(20).mean()) / (raw_cvd.rolling(20).std() + 1e-10)).clip(-3, 3) # 13. Buying/Selling Pressure upper_wick = df['high'] - df[['close', 'open']].max(axis=1) lower_wick = df[['close', 'open']].min(axis=1) - df['low'] # More lower wick = buying pressure (rejected lows) # More upper wick = selling pressure (rejected highs) features['orderflow_buy_pressure'] = (lower_wick / (range_total + 1e-10)).rolling(5).mean() features['orderflow_sell_pressure'] = (upper_wick / (range_total + 1e-10)).rolling(5).mean() features['orderflow_pressure_diff'] = (features['orderflow_buy_pressure'] - features['orderflow_sell_pressure']).clip(-0.5, 0.5) # 14. Large Order Proxy (volume spikes with directional moves) vol_spike = df['volume'] > vol_ma * 2 up_move = df['close'] > df['open'] down_move = df['close'] < df['open'] features['orderflow_large_buys'] = (vol_spike & up_move).astype(float).rolling(10).mean() features['orderflow_large_sells'] = (vol_spike & down_move).astype(float).rolling(10).mean() features['orderflow_large_bias'] = (features['orderflow_large_buys'] - features['orderflow_large_sells']).clip(-1, 1) # 15. Market Regime Confidence # Combine multiple signals for overall market state trend_signal = (ema_20 > ema_50).astype(float) * 2 - 1 # 1 or -1 volume_confirms = (df['volume'] > vol_ma).astype(float) features['regime_confidence'] = (trend_signal * volume_confirms * np.abs(features['funding_momentum_proxy'])).clip(-1, 1) return features def _get_price_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Basic price-based features.""" features = {} # Returns for period in [1, 5, 10, 20]: features[f'return_{period}'] = df['close'].pct_change(period) # Log returns features['log_return'] = np.log(df['close'] / df['close'].shift(1)) # Volatility features['volatility_10'] = df['close'].pct_change().rolling(10).std() features['volatility_20'] = df['close'].pct_change().rolling(20).std() # Price position in range features['price_position'] = (df['close'] - df['low']) / (df['high'] - df['low'] + 1e-10) # Body ratio features['body_ratio'] = np.abs(df['close'] - df['open']) / (df['high'] - df['low'] + 1e-10) # Candle direction features['candle_direction'] = np.sign(df['close'] - df['open']) # Gap features['gap'] = (df['open'] - df['close'].shift(1)) / df['close'].shift(1) return features def _get_technical_features(self, df: pd.DataFrame) -> Dict[str, pd.Series]: """Technical indicator features.""" features = {} # RSI for period in [7, 14, 21]: features[f'rsi_{period}'] = self._calculate_rsi(df['close'], period) / 100 # MACD macd, signal, hist = self._calculate_macd(df['close']) features['macd'] = macd / df['close'] # Normalize features['macd_signal'] = signal / df['close'] features['macd_hist'] = hist / df['close'] # Bollinger Bands bb_upper, bb_middle, bb_lower = self._calculate_bollinger(df['close']) features['bb_position'] = (df['close'] - bb_lower) / (bb_upper - bb_lower + 1e-10) features['bb_width'] = (bb_upper - bb_lower) / bb_middle # ATR atr = self._calculate_atr(df) features['atr_normalized'] = atr / df['close'] # Stochastic stoch_k, stoch_d = self._calculate_stochastic(df) features['stoch_k'] = stoch_k / 100 features['stoch_d'] = stoch_d / 100 # Moving averages for period in [10, 20, 50, 100]: sma = df['close'].rolling(period).mean() features[f'sma_{period}_dist'] = (df['close'] - sma) / sma # EMA crossovers ema_12 = df['close'].ewm(span=12).mean() ema_26 = df['close'].ewm(span=26).mean() features['ema_cross'] = (ema_12 > ema_26).astype(float) # ADX (trend strength) features['adx'] = self._calculate_adx(df) / 100 # CCI features['cci'] = self._calculate_cci(df) / 200 # Normalize around 0 return features def _calculate_rsi(self, prices: pd.Series, period: int = 14) -> pd.Series: """Calculate RSI.""" delta = prices.diff() gain = delta.where(delta > 0, 0).rolling(period).mean() loss = (-delta.where(delta < 0, 0)).rolling(period).mean() rs = gain / (loss + 1e-10) rsi = 100 - (100 / (1 + rs)) return rsi def _calculate_macd(self, prices: pd.Series) -> Tuple[pd.Series, pd.Series, pd.Series]: """Calculate MACD.""" ema_12 = prices.ewm(span=12).mean() ema_26 = prices.ewm(span=26).mean() macd = ema_12 - ema_26 signal = macd.ewm(span=9).mean() hist = macd - signal return macd, signal, hist def _calculate_bollinger(self, prices: pd.Series, period: int = 20, std: float = 2.0) -> Tuple[pd.Series, pd.Series, pd.Series]: """Calculate Bollinger Bands.""" middle = prices.rolling(period).mean() std_dev = prices.rolling(period).std() upper = middle + std * std_dev lower = middle - std * std_dev return upper, middle, lower def _calculate_atr(self, df: pd.DataFrame, period: int = 14) -> pd.Series: """Calculate ATR.""" high_low = df['high'] - df['low'] high_close = np.abs(df['high'] - df['close'].shift()) low_close = np.abs(df['low'] - df['close'].shift()) tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) atr = tr.rolling(period).mean() return atr def _calculate_stochastic(self, df: pd.DataFrame, k_period: int = 14, d_period: int = 3) -> Tuple[pd.Series, pd.Series]: """Calculate Stochastic oscillator.""" lowest_low = df['low'].rolling(k_period).min() highest_high = df['high'].rolling(k_period).max() stoch_k = 100 * (df['close'] - lowest_low) / (highest_high - lowest_low + 1e-10) stoch_d = stoch_k.rolling(d_period).mean() return stoch_k, stoch_d def _calculate_adx(self, df: pd.DataFrame, period: int = 14) -> pd.Series: """Calculate ADX.""" plus_dm = df['high'].diff() minus_dm = -df['low'].diff() plus_dm[plus_dm < 0] = 0 minus_dm[minus_dm < 0] = 0 tr = self._calculate_atr(df, 1) plus_di = 100 * (plus_dm.rolling(period).mean() / (tr.rolling(period).mean() + 1e-10)) minus_di = 100 * (minus_dm.rolling(period).mean() / (tr.rolling(period).mean() + 1e-10)) dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di + 1e-10) adx = dx.rolling(period).mean() return adx def _calculate_cci(self, df: pd.DataFrame, period: int = 20) -> pd.Series: """Calculate CCI.""" typical_price = (df['high'] + df['low'] + df['close']) / 3 sma = typical_price.rolling(period).mean() mad = typical_price.rolling(period).apply(lambda x: np.abs(x - x.mean()).mean()) cci = (typical_price - sma) / (0.015 * mad + 1e-10) return cci def get_feature_names(self) -> List[str]: """Get list of feature names.""" # Create a dummy DataFrame to get feature names dummy_df = pd.DataFrame({ 'open': np.random.rand(100), 'high': np.random.rand(100), 'low': np.random.rand(100), 'close': np.random.rand(100), 'volume': np.random.rand(100), }) dummy_df['high'] = dummy_df[['open', 'close', 'high']].max(axis=1) dummy_df['low'] = dummy_df[['open', 'close', 'low']].min(axis=1) features = self.get_all_features(dummy_df) return list(features.keys())