import numpy as np import pandas as pd from typing import Optional def _scan_outcome( closes: np.ndarray, highs: np.ndarray, lows: np.ndarray, atr_values: np.ndarray, atr_mult_sl: float, atr_mult_tp: float, max_bars: int, direction: int, sl_levels: Optional[np.ndarray] = None, tp_levels: Optional[np.ndarray] = None, ) -> np.ndarray: """Outcome scan: TP/SL hit detection within max_bars forward bars. Outcome: 1 = TP hit first, 0 = SL hit first, 0 = no hit / unresolved. Если sl_levels/tp_levels переданы — использует их вместо ATR-based расчёта. Args: sl_levels: Кастомные SL уровни (для swing-based стратегии). tp_levels: Кастомные TP уровни. """ n = len(closes) outcomes = np.zeros(n, dtype=np.int32) valid = (atr_values > 1e-10) & (~np.isnan(atr_values)) # TP/SL levels if sl_levels is not None and tp_levels is not None: sl = sl_levels tp = tp_levels else: if direction == 1: tp = closes + atr_values * atr_mult_tp sl = closes - atr_values * atr_mult_sl else: tp = closes - atr_values * atr_mult_tp sl = closes + atr_values * atr_mult_sl for i in range(n - 1): if not valid[i]: continue tp_level = tp[i] sl_level = sl[i] limit = min(n, i + max_bars + 1) for j in range(i + 1, limit): if direction == 1: tp_hit = highs[j] >= tp_level sl_hit = lows[j] <= sl_level else: tp_hit = lows[j] <= tp_level sl_hit = highs[j] >= sl_level if tp_hit: outcomes[i] = 1 break if sl_hit: break return outcomes def compute_trade_outcome( df: pd.DataFrame, atr_mult_sl: float = 1.5, atr_mult_tp: float = 3.0, max_bars: int = 100, ) -> pd.DataFrame: df = df.copy() if 'atr' not in df.columns: raise ValueError("add_atr() must be called before compute_trade_outcome()") closes = df['Close'].values highs = df['High'].values lows = df['Low'].values atr_values = df['atr'].values n = len(df) outcomes = np.zeros(n, dtype=np.int32) for i in range(n - 1): if np.isnan(atr_values[i]) or atr_values[i] == 0: continue entry = closes[i] tp = entry + atr_values[i] * atr_mult_tp sl = entry - atr_values[i] * atr_mult_sl limit = min(n, i + max_bars + 1) for j in range(i + 1, limit): if highs[j] >= tp: outcomes[i] = 1 break if lows[j] <= sl: outcomes[i] = 2 break df['outcome'] = outcomes return df def compute_dual_outcomes( df: pd.DataFrame, atr_mult_sl: float = 1.5, atr_mult_tp: float = 3.0, max_bars: int = 100, disable_short: bool = False, ) -> pd.DataFrame: """Вычисляет SL/TP outcomes (long и опционально short). Args: disable_short: Если True — outcome_short НЕ вычисляется (для MOEX-тикеров, где short-selling недоступен). """ df = df.copy() if 'atr' not in df.columns: raise ValueError("add_atr() must be called before compute_dual_outcomes()") arr = ( df['Close'].values, df['High'].values, df['Low'].values, df['atr'].values, ) df['outcome_long'] = _scan_outcome(*arr, atr_mult_sl, atr_mult_tp, max_bars, direction=1) if not disable_short: df['outcome_short'] = _scan_outcome(*arr, atr_mult_sl, atr_mult_tp, max_bars, direction=-1) # Rolling WR использует упрощённый outcome с меньшим max_bars (30), # чтобы разрешённые строки были ближе к настоящему моменту df['outcome_long_fast'] = _scan_outcome(*arr, atr_mult_sl, atr_mult_tp, 30, direction=1) if not disable_short: df['outcome_short_fast'] = _scan_outcome(*arr, atr_mult_sl, atr_mult_tp, 30, direction=-1) return df # --------------------------------------------------------------------------- # Улучшенные таргеты: Swing-based SL + ATR-based TP # --------------------------------------------------------------------------- def _compute_swing_sl_levels( closes: np.ndarray, highs: np.ndarray, lows: np.ndarray, atr_values: np.ndarray, swing_lookback: int, atr_mult_sl: float, atr_mult_tp: float, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: """ Вычисляет SL на основе свингов (support/resistance) и ATR-based TP. Для long: swing_sl = min(low[i-swing_lookback : i]) — поддержка atr_sl = close - atr * atr_mult_sl — ATR-minimal distance SL = min(swing_sl, atr_sl) — более консервативный (ближе) TP = close + atr * atr_mult_tp Для short: swing_sl = max(high[i-swing_lookback : i]) — сопротивление atr_sl = close + atr * atr_mult_sl SL = max(swing_sl, atr_sl) TP = close - atr * atr_mult_tp Гарантии: - SL расстояние ≥ 0.5×ATR (не слишком близко) - SL расстояние ≤ 5.0×ATR (не слишком далеко) Returns: (sl_long, tp_long, sl_short, tp_short) — np.ndarray уровней. """ n = len(closes) sl_long = np.zeros(n) tp_long = np.zeros(n) sl_short = np.zeros(n) tp_short = np.zeros(n) for i in range(n): if np.isnan(atr_values[i]) or atr_values[i] < 1e-10: continue atr = atr_values[i] entry = closes[i] # --- Swing support/resistance до входа --- lookback_start = max(0, i - swing_lookback) # Минимальный лоу и максимальный хай до входа swing_support = np.min(lows[lookback_start:i+1]) if i > lookback_start else entry - atr * 2 swing_resist = np.max(highs[lookback_start:i+1]) if i > lookback_start else entry + atr * 2 # --- Long: SL = swing support (но не ближе 0.5×ATR, не дальше 5×ATR) --- atr_sl = entry - atr * atr_mult_sl # SL = поддержка, но не дальше entry - 5×ATR и не ближе entry - 0.5×ATR candidate_sl = min(swing_support, atr_sl) min_sl = entry - atr * 5.0 max_sl = entry - atr * 0.5 sl_long[i] = np.clip(candidate_sl, min_sl, max_sl) tp_long[i] = entry + atr * atr_mult_tp # --- Short: SL = swing resistance --- atr_sl_short = entry + atr * atr_mult_sl candidate_sl_short = max(swing_resist, atr_sl_short) min_sl_short = entry + atr * 0.5 max_sl_short = entry + atr * 5.0 sl_short[i] = np.clip(candidate_sl_short, min_sl_short, max_sl_short) tp_short[i] = entry - atr * atr_mult_tp return sl_long, tp_long, sl_short, tp_short def compute_swing_dual_outcomes( df: pd.DataFrame, atr_mult_sl: float = 3.0, atr_mult_tp: float = 6.0, max_bars: int = 200, swing_lookback: int = 20, disable_short: bool = False, ) -> pd.DataFrame: """ Dual-Outcome таргеты со Swing-based SL вместо чисто ATR-based. SL размещается на ближайшем уровне поддержки/сопротивления (минимум/максимум за последние swing_lookback баров до входа), что делает стоп-лосс рыночно-обоснованным. TP остаётся ATR-based (close + atr_mult_tp × ATR). Args: df: DataFrame с колонкой 'atr'. atr_mult_sl: Множитель ATR для fallback SL. atr_mult_tp: Множитель ATR для TP. max_bars: Максимум баров для сканирования forward. swing_lookback: Размер окна для поиска support/resistance. disable_short: Если True — outcome_short НЕ вычисляется (для MOEX-тикеров, где short-selling недоступен). Returns: DataFrame с колонками outcome_long/outcome_short + _fast. """ df = df.copy() if 'atr' not in df.columns: raise ValueError("add_atr() must be called before compute_swing_dual_outcomes()") closes = df['Close'].values highs = df['High'].values lows = df['Low'].values atr_values = df['atr'].values sl_long, tp_long, sl_short, tp_short = _compute_swing_sl_levels( closes, highs, lows, atr_values, swing_lookback, atr_mult_sl, atr_mult_tp, ) # Scan outcome с кастомными SL/TP уровнями df['outcome_long'] = _scan_outcome( closes, highs, lows, atr_values, atr_mult_sl, atr_mult_tp, max_bars, direction=1, sl_levels=sl_long, tp_levels=tp_long, ) if not disable_short: df['outcome_short'] = _scan_outcome( closes, highs, lows, atr_values, atr_mult_sl, atr_mult_tp, max_bars, direction=-1, sl_levels=sl_short, tp_levels=tp_short, ) # Fast versions для rolling WR (30 bars) df['outcome_long_fast'] = _scan_outcome( closes, highs, lows, atr_values, atr_mult_sl, atr_mult_tp, 30, direction=1, sl_levels=sl_long, tp_levels=tp_long, ) if not disable_short: df['outcome_short_fast'] = _scan_outcome( closes, highs, lows, atr_values, atr_mult_sl, atr_mult_tp, 30, direction=-1, sl_levels=sl_short, tp_levels=tp_short, ) # Store actual SL/TP levels for PnL calculation in threshold search df['swing_sl_long'] = sl_long df['swing_tp_long'] = tp_long if not disable_short: df['swing_sl_short'] = sl_short df['swing_tp_short'] = tp_short return df def _find_valid_peak_high( high: np.ndarray, low: np.ndarray, start: int, end: int, ) -> float: """ Ищет ближайший валидный пик в окне [start, end). Критерии пика: 1. High[j] — локальный максимум (High[j] >= High[j-1] и High[j] >= High[j+1]) 2. Пик центрирован: в симметричном окне [j-r : j+r], где r = min(j-start, end-1-j), нет High выше, чем High[j] (т.е. j — истинная вершина на этом участке графика) 3. Выбирается первый (ближайший к start) валидный пик Если валидный пик не найден — возвращает глобальный максимум окна (фолбэк). """ # Сначала проверяем глобальный максимум global_max = np.max(high[start:end]) argmax = start + np.argmax(high[start:end]) # Проверяем центрированность глобального максимума left_dist = argmax - start right_dist = end - 1 - argmax radius = min(left_dist, right_dist) if radius >= 2: check_l = argmax - radius check_r = argmax + radius + 1 if high[argmax] >= np.max(high[check_l:check_r]): return global_max # глобальный максимум — валидный центрированный пик # Ищем первый валидный центрированный пик, сканируя слева направо for j in range(start + 1, end - 1): # Проверка: локальный максимум if high[j] >= high[j - 1] and high[j] >= high[j + 1]: left_d = j - start right_d = end - 1 - j r = min(left_d, right_d) if r >= 2: check_l = j - r check_r = j + r + 1 if high[j] >= np.max(high[check_l:check_r]): return high[j] # первый валидный пик # Фолбэк: глобальный максимум return global_max def _find_valid_trough_low( high: np.ndarray, low: np.ndarray, start: int, end: int, ) -> float: """ Ищет ближайшую валидную впадину в окне [start, end). Критерии впадины: 1. Low[j] — локальный минимум (Low[j] <= Low[j-1] и Low[j] <= Low[j+1]) 2. Впадина центрирована: в симметричном окне [j-r : j+r], где r = min(j-start, end-1-j), нет Low ниже, чем Low[j] 3. Выбирается первая (ближайшая к start) валидная впадина Если валидная впадина не найдена — возвращает глобальный минимум окна (фолбэк). """ # Сначала проверяем глобальный минимум global_min = np.min(low[start:end]) argmin = start + np.argmin(low[start:end]) left_dist = argmin - start right_dist = end - 1 - argmin radius = min(left_dist, right_dist) if radius >= 2: check_l = argmin - radius check_r = argmin + radius + 1 if low[argmin] <= np.min(low[check_l:check_r]): return global_min # глобальный минимум — валидная центрированная впадина # Ищем первую валидную центрированную впадину, сканируя слева направо for j in range(start + 1, end - 1): if low[j] <= low[j - 1] and low[j] <= low[j + 1]: left_d = j - start right_d = end - 1 - j r = min(left_d, right_d) if r >= 2: check_l = j - r check_r = j + r + 1 if low[j] <= np.min(low[check_l:check_r]): return low[j] # первая валидная впадина # Фолбэк: глобальный минимум return global_min def compute_peak_trough_targets( df: pd.DataFrame, horizon: int = 100, ) -> pd.DataFrame: """ Для каждой свечи находит ближайший ВАЛИДНЫЙ пик и впадину в следующих `horizon` барах. Пик/впадина считаются валидными, если: - Это локальный экстремум (выше/ниже соседей) - Центрирован: в симметричном окне [pos-r : pos+r] нет точек выше/ниже Возвращает регрессионные цели: pct_to_peak: High[пик] / Close[i] - 1 — потенциал роста (+, %) pct_to_trough: Low[впадина] / Close[i] - 1 — потенциал падения (-, %) """ df = df.copy() close = df['Close'].values high = df['High'].values low = df['Low'].values n = len(df) pct_to_peak = np.full(n, np.nan, dtype=np.float64) pct_to_trough = np.full(n, np.nan, dtype=np.float64) for i in range(n - horizon): start = i + 1 end = i + 1 + horizon peak_high = _find_valid_peak_high(high, low, start, end) trough_low = _find_valid_trough_low(high, low, start, end) pct_to_peak[i] = peak_high / close[i] - 1.0 pct_to_trough[i] = trough_low / close[i] - 1.0 df['pct_to_peak'] = pct_to_peak df['pct_to_trough'] = pct_to_trough return df # --------------------------------------------------------------------------- # Улучшенные регрессионные цели (v2 — multi-horizon + ATR-norm + time-to-extreme) # --------------------------------------------------------------------------- def _find_valid_peak_with_pos( high: np.ndarray, low: np.ndarray, start: int, end: int, ) -> tuple[float, int]: """ Как _find_valid_peak_high, но возвращает (значение, позиция). """ global_max = float(np.max(high[start:end])) argmax = int(start + np.argmax(high[start:end])) left_dist = argmax - start right_dist = end - 1 - argmax radius = min(left_dist, right_dist) if radius >= 2: check_l = argmax - radius check_r = argmax + radius + 1 if high[argmax] >= np.max(high[check_l:check_r]): return float(global_max), argmax for j in range(start + 1, end - 1): if high[j] >= high[j - 1] and high[j] >= high[j + 1]: left_d = j - start right_d = end - 1 - j r = min(left_d, right_d) if r >= 2: check_l = j - r check_r = j + r + 1 if high[j] >= np.max(high[check_l:check_r]): return float(high[j]), j return float(global_max), argmax def _find_valid_trough_with_pos( high: np.ndarray, low: np.ndarray, start: int, end: int, ) -> tuple[float, int]: """ Как _find_valid_trough_low, но возвращает (значение, позиция). """ global_min = float(np.min(low[start:end])) argmin = int(start + np.argmin(low[start:end])) left_dist = argmin - start right_dist = end - 1 - argmin radius = min(left_dist, right_dist) if radius >= 2: check_l = argmin - radius check_r = argmin + radius + 1 if low[argmin] <= np.min(low[check_l:check_r]): return float(global_min), argmin for j in range(start + 1, end - 1): if low[j] <= low[j - 1] and low[j] <= low[j + 1]: left_d = j - start right_d = end - 1 - j r = min(left_d, right_d) if r >= 2: check_l = j - r check_r = j + r + 1 if low[j] <= np.min(low[check_l:check_r]): return float(low[j]), j return float(global_min), argmin REGRESSION_HORIZONS = [10, 30, 60] # короткий, средний, длинный DEFAULT_ATR_PERIOD = 14 def compute_enhanced_regression_targets( df: pd.DataFrame, horizons: list[int] | None = None, min_atr_move: float | dict[int, float] = 0.5, ) -> pd.DataFrame: """ Улучшенные регрессионные цели для MoERegression v2. Для каждого бара i и каждого горизонта h: - pct_to_peak_h: % до пика в окне h баров (всегда ≥ 0) - pct_to_trough_h: % до впадины в окне h баров (всегда ≤ 0) - peak_norm_h: pct_to_peak_h / atr_pct (в единицах ATR) - trough_norm_h: pct_to_trough_h / atr_pct (в единицах ATR) - bars_to_peak_h: кол-во баров до достижения пика - bars_to_trough_h: кол-во баров до достижения впадины - move_exists_h: 1 если есть движение > min_atr_move × ATR ATR-нормализация переводит цели в единую шкалу «сколько ATR», что позволяет модели одинаково хорошо работать в разных режимах волатильности. Время до экстремума (bars_to_*) позволяет модели различать «скоро будет пик» (через 3 бара) и «когда-нибудь будет пик» (через 55). Параметры: min_atr_move: множитель ATR для move_exists. Если float — применяется ко всем горизонтам. Если dict — {horizon: threshold} для каждого горизонта. """ df = df.copy() if horizons is None: horizons = REGRESSION_HORIZONS # ATR как фракция от close if 'atr_pct' in df.columns: atr_pct = df['atr_pct'].values else: tr = np.maximum( df['High'] - df['Low'], np.maximum( np.abs(df['High'] - df['Close'].shift(1)), np.abs(df['Low'] - df['Close'].shift(1)), ), ) atr_pct = (tr.rolling(DEFAULT_ATR_PERIOD).mean() / df['Close']).fillna(0).values close = df['Close'].values high = df['High'].values low = df['Low'].values n = len(df) # Resolve per-horizon thresholds if isinstance(min_atr_move, dict): move_thresholds = min_atr_move else: move_thresholds = {h: min_atr_move for h in horizons} for h in horizons: thr = move_thresholds.get(h, 0.5) # fallback 0.5 if horizon not in dict pct_peak = np.full(n, np.nan, dtype=np.float64) pct_trough = np.full(n, np.nan, dtype=np.float64) peak_norm = np.full(n, np.nan, dtype=np.float64) trough_norm = np.full(n, np.nan, dtype=np.float64) bars_peak = np.full(n, -1, dtype=np.int32) bars_trough = np.full(n, -1, dtype=np.int32) move_exists = np.zeros(n, dtype=np.float32) for i in range(n - h): start = i + 1 end = i + 1 + h p_val, p_pos = _find_valid_peak_with_pos(high, low, start, end) t_val, t_pos = _find_valid_trough_with_pos(high, low, start, end) peak_pct = p_val / close[i] - 1.0 trough_pct = t_val / close[i] - 1.0 # отрицательное pct_peak[i] = peak_pct pct_trough[i] = trough_pct # ATR-нормализация atr_i = max(atr_pct[i], 1e-6) peak_norm[i] = peak_pct / atr_i trough_norm[i] = trough_pct / atr_i # Время до экстремума bars_peak[i] = p_pos - i bars_trough[i] = t_pos - i # Движение существует, если хотя бы одна сторона > thr × ATR move_exists[i] = 1.0 if (peak_pct > thr * atr_i or abs(trough_pct) > thr * atr_i) else 0.0 df[f'pct_to_peak_{h}'] = pct_peak df[f'pct_to_trough_{h}'] = pct_trough df[f'peak_norm_{h}'] = peak_norm df[f'trough_norm_{h}'] = trough_norm df[f'bars_to_peak_{h}'] = bars_peak df[f'bars_to_trough_{h}'] = bars_trough df[f'move_exists_{h}'] = move_exists return df def outcome_distribution(df: pd.DataFrame) -> pd.Series: return df['outcome'].value_counts().sort_index()