""" 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', # v13 MTF divergence (audit 2026-08-03) 'mtf_h1_d1_rsi_div', 'mtf_h1_d1_divergence_flag', 'mtf_h1_w1_rsi_div', 'mtf_trend_confluence_3tf', ] # --------------------------------------------------------------------------- # Вспомогательные функции для одного ТФ # --------------------------------------------------------------------------- def _wilders_smooth(values: np.ndarray, period: int) -> np.ndarray: """Wilders Smoothing (SMMA) — метод Уайлдера для ADX/ATR. SMMA[i] = (SMMA[i-1] * (period - 1) + values[i]) / period SMMA[0:period] = mean(values[0:period]) — инициализация. """ n = len(values) result = np.zeros(n, dtype=values.dtype) if n < period: return result result[period - 1] = np.mean(values[:period]) for i in range(period, n): result[i] = (result[i - 1] * (period - 1) + values[i]) / period if period > 1: result[:period - 1] = result[period - 1] return result def _compute_adx(df: pd.DataFrame, period: int = 14) -> np.ndarray: """ADX по методу Уайлдера (1978) — SMMA для ATR, +DM, -DM, ADX. Consistent with experts.py _compute_adx — same Wilders smoothing. """ high = df['High'].values low = df['Low'].values close = df['Close'].values n = len(df) # True Range tr = np.zeros(n) for i in range(1, n): hl = high[i] - low[i] hc = abs(high[i] - close[i - 1]) lc = abs(low[i] - close[i - 1]) tr[i] = max(hl, hc, lc) # ATR через Wilders Smoothing atr = _wilders_smooth(tr, period) # Directional Movement up_move = np.zeros(n) down_move = np.zeros(n) for i in range(1, n): up_move[i] = max(high[i] - high[i - 1], 0) down_move[i] = max(low[i - 1] - low[i], 0) # +DM и -DM через Wilders Smoothing plus_dm = _wilders_smooth(up_move, period) minus_dm = _wilders_smooth(down_move, period) # +DI и -DI atr_safe = np.where(atr > 1e-10, atr, 1e-10) plus_di = 100 * plus_dm / atr_safe minus_di = 100 * minus_dm / atr_safe # DX и ADX (через Wilders Smoothing) dx = 100 * np.abs(plus_di - minus_di) / np.where(plus_di + minus_di > 1e-10, plus_di + minus_di, 1e-10) adx = _wilders_smooth(dx, period) 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, shift_mtf: bool = True) -> pd.DataFrame: """ Загружает D1 и W1 для ticker, вычисляет контекстные фичи, и добавляет их в H1 DataFrame через asof merge. Параметры: shift_mtf: True = сдвиг D1/W1 на 1 период назад (тренировка, защита от lookahead). False = без сдвига (инференс — берём последние актуальные данные D1/W1). Возвращает 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) # При shift_mtf=True: сдвиг D1 на 1 день вперёд, чтобы H1(23.07 06:00) # не видела текущую D1(23.07), а видела D1(22.07) — защита от lookahead. # При shift_mtf=False: используем последние данные D1 как есть (инференс). if shift_mtf: d1_feats['timestamp'] = d1_feats['timestamp'] + 86400 # +1 день # 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) # При shift_mtf=True: сдвиг W1 на 1 неделю вперёд, чтобы H1(текущая неделя) # не видела текущую W1, а видела предыдущую — защита от lookahead. # При shift_mtf=False: используем последние данные W1 как есть (инференс). if shift_mtf: w1_feats['timestamp'] = w1_feats['timestamp'] + 604800 # +7 дней # Если 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) # ── v13 MTF divergence features (audit 2026-08-03) ─────────────── # Цель: количественная мера расхождения между H1 и D1/W1, а не только # бинарный sign-product. Если H1 RSI > D1 RSI → локальная сила相对于 # дневному тренду; дивергенция часто предшествует развороту. if has_d1: d1_rsi = result['d1_rsi'].fillna(50).values # H1 RSI (вычислим простой, если нет в df_h1) if 'rsi' in result.columns: h1_rsi = result['rsi'].fillna(50).values else: h1_rsi = np.full(len(result), 50.0) # Нормированная дивергенция: (H1 - D1) / 100, в диапазоне ~[-0.5, +0.5] result['mtf_h1_d1_rsi_div'] = (h1_rsi - d1_rsi) / 100.0 # Бинарный флаг: H1 и D1 по разные стороны 50 (divergence) result['mtf_h1_d1_divergence_flag'] = ( (np.sign(h1_rsi - 50) != np.sign(d1_rsi - 50)) ).astype(float) if has_w1: w1_rsi = result['w1_rsi'].fillna(50).values if 'rsi' in result.columns: h1_rsi = result['rsi'].fillna(50).values else: h1_rsi = np.full(len(result), 50.0) result['mtf_h1_w1_rsi_div'] = (h1_rsi - w1_rsi) / 100.0 # Trend confluence: все 3 TF в одну сторону от 50 if has_d1: d1_rsi = result['d1_rsi'].fillna(50).values same_sign = ( (np.sign(h1_rsi - 50) == np.sign(d1_rsi - 50)) & (np.sign(d1_rsi - 50) == np.sign(w1_rsi - 50)) ) result['mtf_trend_confluence_3tf'] = same_sign.astype(float) # --- Заполняем пропуски --- 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)