""" Multi-Timeframe Context Features. Заменяет отдельные D1/W1 эксперты: вместо них вычисляет контекстные фичи из старших таймфреймов и добавляет их как дополнительные признаки к H1. Фичи: d1_trend_slope — наклон D1 Close за 20 дней (нормализованный) d1_trend_strength — D1 ADX (0-100) d1_rsi — D1 RSI (0-100) d1_vol_regime — D1 ATR процентиль: 0=low, 1=normal, 2=high w1_trend_slope — наклон W1 Close за 10 недель w1_trend_strength — W1 ADX w1_rsi — W1 RSI d1_h1_alignment — sign(D1 slope) × sign(H1 slope) ∈ {-1,0,+1} w1_h1_alignment — sign(W1 slope) × sign(H1 slope) w1_d1_alignment — sign(W1 slope) × sign(D1 slope) mtf_confluence — среднее всех alignment ∈ [0,1] mtf_regime — 0=range / 1=uptrend / 2=downtrend (по D1+W1) """ import numpy as np import pandas as pd from data.loader import load_dataframe from features.technical import engineer_features from features.directional import add_directional_signals MTF_CONTEXT_COLS = [ 'd1_trend_slope', 'd1_trend_strength', 'd1_rsi', 'd1_vol_regime', 'w1_trend_slope', 'w1_trend_strength', 'w1_rsi', 'd1_h1_alignment', 'w1_h1_alignment', 'w1_d1_alignment', 'mtf_confluence', 'mtf_regime', ] # --------------------------------------------------------------------------- # Вспомогательные функции для одного ТФ # --------------------------------------------------------------------------- def _compute_adx(df: pd.DataFrame, period: int = 14) -> np.ndarray: """Приближённый ADX: среднее направленное движение.""" high = df['High'].values low = df['Low'].values close = df['Close'].values up = np.diff(high) down = -np.diff(low) up[up < 0] = 0 down[down < 0] = 0 plus_dm = np.concatenate([[0], up]) minus_dm = np.concatenate([[0], down]) tr = np.maximum( high - low, np.maximum( np.abs(high - np.concatenate([[close[0]], close[:-1]])), np.abs(low - np.concatenate([[close[0]], close[:-1]])) ) ) atr = pd.Series(tr).rolling(period).mean().values plus_di = 100 * pd.Series(plus_dm).rolling(period).mean().values / (atr + 1e-10) minus_di = 100 * pd.Series(minus_dm).rolling(period).mean().values / (atr + 1e-10) dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di + 1e-10) adx = pd.Series(dx).rolling(period).mean().values return np.nan_to_num(adx, nan=20.0) def _compute_tf_features(df: pd.DataFrame, window: int = 20) -> pd.DataFrame: """ Вычисляет trend_slope, trend_strength (ADX), rsi, vol_regime для одного ТФ. Возвращает DataFrame с теми же индексами и дополнительными колонками. """ result = df.copy() # --- trend_slope: наклон Close за window баров --- closes = result['Close'].values slopes = np.full(len(result), 0.0) for i in range(window, len(result)): y = closes[i - window:i] x = np.arange(window) slope = np.polyfit(x, y, 1)[0] slopes[i] = slope / (closes[i] + 1e-10) # нормализация ценой result['trend_slope'] = slopes # --- trend_strength: ADX --- result['trend_strength'] = _compute_adx(result, period=min(14, window)) # --- RSI (уже есть в engineer_features, но на случай если нет) --- if 'rsi' not in result.columns: delta = result['Close'].diff() gain = delta.where(delta > 0, 0.0) loss = (-delta).where(delta < 0, 0.0) avg_g = gain.rolling(window=14, min_periods=1).mean() avg_l = loss.rolling(window=14, min_periods=1).mean() rs = avg_g / avg_l.replace(0, np.nan) result['rsi'] = 100 - (100 / (1 + rs)) result['rsi'] = result['rsi'].fillna(50) # --- vol_regime: ATR процентиль --- if 'atr_pct' in result.columns: atr_vals = result['atr_pct'].values else: # Считаем ATR tr = np.maximum( result['High'] - result['Low'], np.maximum( np.abs(result['High'] - result['Close'].shift(1)), np.abs(result['Low'] - result['Close'].shift(1)) ) ) atr_vals = (tr.rolling(14).mean() / result['Close']).values # Процентили: низкий < 30%, высокий > 70% vol_regime = np.zeros(len(result), dtype=int) for i in range(window, len(result)): window_vals = atr_vals[i - window:i] pct = np.sum(window_vals < atr_vals[i]) / len(window_vals) if pct < 0.3: vol_regime[i] = 0 # low elif pct > 0.7: vol_regime[i] = 2 # high else: vol_regime[i] = 1 # normal result['vol_regime'] = vol_regime return result # --------------------------------------------------------------------------- # Основная функция: добавляет MTF контекст к H1 # --------------------------------------------------------------------------- def add_mtf_context(ticker: str, df_h1: pd.DataFrame) -> pd.DataFrame: """ Загружает D1 и W1 для ticker, вычисляет контекстные фичи, и добавляет их в H1 DataFrame через asof merge. Возвращает df_h1 с новыми колонками из MTF_CONTEXT_COLS. """ result = df_h1.copy() result['timestamp'] = result['timestamp'].astype(int) # Флаг: какие ТФ доступны has_d1 = False has_w1 = False # Словарь для хранения merged фич extra_cols = pd.DataFrame({'timestamp': result['timestamp'].values}) # --- Обработка D1 --- try: df_d1 = load_dataframe(ticker, 'D1') if df_d1 is not None and len(df_d1) > 50: df_d1 = engineer_features(df_d1, skip_temporal=True) df_d1 = add_directional_signals(df_d1) df_d1 = _compute_tf_features(df_d1, window=20) d1_feats = df_d1[['timestamp', 'trend_slope', 'trend_strength', 'rsi', 'vol_regime']].copy() d1_feats.columns = ['timestamp', 'd1_trend_slope', 'd1_trend_strength', 'd1_rsi', 'd1_vol_regime'] d1_feats['timestamp'] = d1_feats['timestamp'].astype(int) # asof merge: берём последнюю D1 свечу для каждой H1 extra_cols = pd.merge_asof( extra_cols.sort_values('timestamp'), d1_feats.sort_values('timestamp'), on='timestamp', direction='backward', ) has_d1 = True except Exception as e: print(f' [MTF] D1 context error: {e}') # --- Обработка W1 --- try: df_w1 = load_dataframe(ticker, 'W1') if df_w1 is not None and len(df_w1) > 20: df_w1 = engineer_features(df_w1, skip_temporal=True) df_w1 = add_directional_signals(df_w1) df_w1 = _compute_tf_features(df_w1, window=10) w1_feats = df_w1[['timestamp', 'trend_slope', 'trend_strength', 'rsi']].copy() w1_feats.columns = ['timestamp', 'w1_trend_slope', 'w1_trend_strength', 'w1_rsi'] w1_feats['timestamp'] = w1_feats['timestamp'].astype(int) # Если D1 уже добавлен, мержим extra_cols, иначе создаём extra_cols = pd.merge_asof( extra_cols.sort_values('timestamp'), w1_feats.sort_values('timestamp'), on='timestamp', direction='backward', ) has_w1 = True except Exception as e: print(f' [MTF] W1 context error: {e}') # --- Объединяем с H1 --- # extra_cols содержит все MTF колонки; мержим обратно в result result = pd.merge( result, extra_cols, on='timestamp', how='left' ) # --- Вычисляем alignment фичи --- result['d1_h1_alignment'] = 0.0 result['w1_h1_alignment'] = 0.0 result['w1_d1_alignment'] = 0.0 result['mtf_confluence'] = 0.0 result['mtf_regime'] = 0 # H1 trend slope: используем return_20 как proxy h1_slope = result['return_20'].fillna(0).values # 20-period return ≈ trend if has_d1: d1_slope = result['d1_trend_slope'].fillna(0).values d1_strength = result['d1_trend_strength'].fillna(20).values # alignment result['d1_h1_alignment'] = np.sign(h1_slope) * np.sign(d1_slope) # mtf_regime based on D1 cond_up = (d1_slope > 0.001) & (d1_strength > 25) cond_down = (d1_slope < -0.001) & (d1_strength > 25) result.loc[cond_up, 'mtf_regime'] = 1 result.loc[cond_down, 'mtf_regime'] = 2 if has_w1: w1_slope = result['w1_trend_slope'].fillna(0).values w1_strength = result['w1_trend_strength'].fillna(20).values result['w1_h1_alignment'] = np.sign(h1_slope) * np.sign(w1_slope) result['w1_d1_alignment'] = np.sign(w1_slope) * np.sign( result['d1_trend_slope'].fillna(0).values ) # W1 regime переопределяет D1 regime (W1 имеет приоритет) cond_up = (w1_slope > 0.001) & (w1_strength > 25) cond_down = (w1_slope < -0.001) & (w1_strength > 25) result.loc[cond_up, 'mtf_regime'] = 1 result.loc[cond_down, 'mtf_regime'] = 2 # Если W1 слабый (ADX < 25) → regime = 0 (range) weak_w1 = (w1_strength < 25) & (d1_strength < 25) result.loc[weak_w1, 'mtf_regime'] = 0 # mtf_confluence: среднее абсолютных alignment'ов alignments = [] if 'd1_h1_alignment' in result.columns: alignments.append(result['d1_h1_alignment'].abs()) if 'w1_h1_alignment' in result.columns: alignments.append(result['w1_h1_alignment'].abs()) if 'w1_d1_alignment' in result.columns and has_w1 and has_d1: alignments.append(result['w1_d1_alignment'].abs()) if alignments: result['mtf_confluence'] = pd.concat(alignments, axis=1).mean(axis=1).fillna(0) # --- Заполняем пропуски --- mtf_cols = [c for c in MTF_CONTEXT_COLS if c in result.columns] result[mtf_cols] = result[mtf_cols].fillna(0) return result def has_mtf_context(result: pd.DataFrame) -> bool: """Проверяет, есть ли MTF фичи в DataFrame.""" return any(c in result.columns for c in MTF_CONTEXT_COLS)