import numpy as np import pandas as pd def add_directional_signals(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() # RSI сигнал: (-1) = перекупленность/медвежий, (+1) = перепроданность/бычий # Формула (50 - RSI) / 50 работает корректно для шкалы [0, 100] df['rsi_signal'] = np.clip((50 - df['rsi']) / 50, -1, 1) df['bb_signal'] = np.clip(1 - 2 * df['bb_position'], -1, 1) macd_std = df['macd_hist'].rolling(50).std().replace(0, np.nan) df['macd_signal'] = np.clip(df['macd_hist'] / macd_std, -1, 1) df['macd_signal'] = df['macd_signal'].fillna(0) sma_cols = [c for c in df.columns if c.startswith('close_to_sma_')] if sma_cols: weights = {'close_to_sma_5': 0.3, 'close_to_sma_10': 0.3, 'close_to_sma_20': 0.2, 'close_to_sma_50': 0.2} available_weights = {k: v for k, v in weights.items() if k in df.columns} if available_weights: w_sum = sum(available_weights.values()) df['trend_signal'] = sum( np.clip(df[k] * 10, -1, 1) * (v / w_sum) for k, v in available_weights.items() ) else: df['trend_signal'] = 0.0 else: df['trend_signal'] = 0.0 df['momentum_signal'] = np.clip(df['return_5'] * 20, -1, 1) # Volume signal × trend confluence (v12.7) # Wyckoff-style: high volume = reversal only if trend is weak. # In strong trend, high volume confirms direction. trend = df['trend_signal'].fillna(0).values vr = df['volume_ratio'].fillna(1).values vol_signal = np.zeros(len(df)) for i in range(len(df)): if vr[i] > 1.5: # High volume: confirm trend direction (not reversal) vol_signal[i] = np.clip(trend[i] * 0.5, -0.5, 0.5) elif vr[i] < 0.5: # Low volume: weak market, slight bullish bias (accumulation) vol_signal[i] = 0.3 else: vol_signal[i] = 0.0 df['volume_signal'] = vol_signal signals = ['rsi_signal', 'bb_signal', 'macd_signal', 'trend_signal', 'momentum_signal', 'volume_signal'] weights_sig = {'rsi_signal': 0.25, 'bb_signal': 0.15, 'macd_signal': 0.2, 'trend_signal': 0.2, 'momentum_signal': 0.1, 'volume_signal': 0.1} avail = [s for s in signals if s in df.columns] w_sum = sum(weights_sig[s] for s in avail) df['directional_bias'] = sum( df[s] * (weights_sig[s] / w_sum) for s in avail ) df['signal_strength'] = df['directional_bias'].abs() return df def add_short_specific_features(df: pd.DataFrame) -> pd.DataFrame: """Short/long-specific паттерн-признаки для улучшения precision recall. Добавляет: short_div_signal [0, +1] — Bearish RSI-price divergence short_vol_climax [0, +1] — Volume climax at price peaks (short) short_fail_break [0, +1] — Failed breakout above resistance long_vol_climax [0, +1] — Volume climax at price lows (long) """ df = df.copy() n = len(df) if n < 20: for col in ['short_div_signal', 'short_vol_climax', 'short_fail_break', 'long_vol_climax']: df[col] = 0.0 return df close = df['Close'].values.astype(np.float64) high = df['High'].values.astype(np.float64) low = df['Low'].values.astype(np.float64) volume = df['Volume'].values.astype(np.float64) # ── 1) Bearish divergence (RSI-price) ──────────────────────────────── # Сравниваем максимум цены и максимум RSI за окно 14 баров. # Если price HH, RSI LH → bearish divergence. if 'rsi' in df.columns: rsi = df['rsi'].values.astype(np.float64) else: rsi = None div = np.zeros(n, dtype=np.float64) if rsi is not None: for i in range(14, n): window_high = high[i-13:i+1] window_rsi = rsi[i-13:i+1] ph_idx = np.argmax(window_high) # peak price index in window rsi_max_idx = np.argmax(window_rsi) # peak RSI index in window rsi_at_ph = window_rsi[ph_idx] # RSI value at the price peak rsi_peak = window_rsi[rsi_max_idx] # max RSI in the window # Bearish divergence: price makes higher high than prev window, # but RSI at the price peak is lower than the window's RSI peak prev_high = high[i-14:i].max() if window_high[ph_idx] > prev_high and rsi_at_ph < rsi_peak: # Strength: how much RSI diverged rsi_diff = rsi_peak - rsi_at_ph div[i] = np.clip(rsi_diff / 20.0, 0, 1) # Normalize to [-1, +1] where +1 = strong bearish divergence df['short_div_signal'] = np.clip(div, 0, 1) # ── 2) Volume climax at peaks / lows ───────────────────────────────── vol_zscore = np.zeros(n, dtype=np.float64) vol_ma = np.zeros(n, dtype=np.float64) for i in range(20, n): vol_window = volume[i-19:i+1] vol_mean = vol_window.mean() vol_std = vol_window.std() if vol_std > 1e-8: vol_zscore[i] = (volume[i] - vol_mean) / vol_std else: vol_zscore[i] = 0.0 vol_ma[i] = vol_mean # Volume climax at peaks (short): price near window high + high volume vol_climax_short = np.zeros(n, dtype=np.float64) vol_climax_long = np.zeros(n, dtype=np.float64) for i in range(10, n): window_high_10 = high[i-9:i+1].max() window_low_10 = low[i-9:i+1].min() range_10 = window_high_10 - window_low_10 if range_10 > 1e-8: price_position = (close[i] - window_low_10) / range_10 # 0=bottom, 1=top if vol_zscore[i] > 2.0: # extreme volume # Near top → selling climax (bearish → short) if price_position > 0.7: vol_climax_short[i] = np.clip((vol_zscore[i] - 2.0) / 3.0, 0, 1) # Near bottom → buying climax (bullish → long) elif price_position < 0.3: vol_climax_long[i] = np.clip((vol_zscore[i] - 2.0) / 3.0, 0, 1) df['short_vol_climax'] = vol_climax_short df['long_vol_climax'] = vol_climax_long # ── 3) Failed breakout (short) — ТОЛЬКО ретроспективные данные ────── # На баре i проверяем: был ли пробой сопротивления в последние 3 бара # и не удалось ли цене удержаться выше (уже закрылась ниже). # Никакого заглядывания в будущее — вся информация доступна на момент i. fail_break = np.zeros(n, dtype=np.float64) for i in range(15, n): window_high_10 = high[i-10:i] # last 10 bars (excluding current) resistance = window_high_10.max() if resistance <= 0: continue # Ищем пробой в последние 3 бара (i-3 .. i-1) breakout_bar = None for k in range(max(14, i - 3), i): if close[k] > resistance * 1.005: breakout_bar = k break if breakout_bar is not None: # Пробой был. Проверяем, удержалась ли цена выше с тех пор. # Если цена сейчас ниже сопротивления → пробой failed. if close[i] < resistance: rejection = (close[breakout_bar] - resistance) / resistance fail_break[i] = np.clip(rejection * 100, 0, 1) df['short_fail_break'] = fail_break # Fill NaN for col in ['short_div_signal', 'short_vol_climax', 'short_fail_break', 'long_vol_climax']: if col in df.columns: df[col] = df[col].fillna(0.0) return df DIRECTIONAL_COLS = [ 'rsi_signal', 'bb_signal', 'macd_signal', 'trend_signal', 'momentum_signal', 'volume_signal', 'directional_bias', 'signal_strength', ] SHORT_SPECIFIC_COLS = [ 'short_div_signal', 'short_vol_climax', 'short_fail_break', 'long_vol_climax', ]