# SkoRZ1RStrategyBlockFix.py # Version: 0.2.0 (pending emulation, SCO+1 strict) from typing import Any, Dict, Optional from datetime import datetime, timezone import numpy as np import pandas as pd from freqtrade.strategy.interface import IStrategy from freqtrade.strategy import IntParameter, DecimalParameter, CategoricalParameter class SkoRZ1RStrategyBlockFix(IStrategy): timeframe = '5m' can_short = True minimal_roi = {"0": 10} stoploss = -0.99 trailing_stop = False process_only_new_candles = True # === параметры индикатора === zscore_lookback = IntParameter(50, 300, default=100, space='indicator', optimize=False) zscore_threshold = DecimalParameter(1.0, 4.0, default=2.0, decimals=1, space='indicator', optimize=False) # Сигнальная свеча = СКО + s signal_lag_bars = IntParameter(1, 1, default=1, space='indicator', optimize=False) # RR и порог минимального размера СКО (в % от цены) RR_MULT = DecimalParameter(1.0, 4.0, default=2.0, decimals=1, space='protection', optimize=False) MIN_R_PCT = DecimalParameter(0.05, 0.50, default=0.15, decimals=2, space='protection', optimize=False) # Эмуляция «отложки»: сколько свечей после сигнальной ждём касания триггера PENDING_BARS = IntParameter(1, 20, default=4, space='protection', optimize=False) # Что считаем сигналом (можно оба сразу) ENTRY_MODE = CategoricalParameter(["REVERSAL", "CONTINUATION", "BOTH"], default="BOTH", space='protection', optimize=False) DIAG_LOG = True # — визуализация — plot_config = { "main_plot": { # уровни «отложек» (горизонтали вперёд) "vis_trigger_long": {"type": "scatter", "color": "orange"}, "vis_trigger_short": {"type": "scatter", "color": "orange"}, # стоп/тейк уровни (на участке ожидания) "vis_sl_long": {"type": "scatter", "color": "blue"}, "vis_tp_long": {"type": "scatter", "color": "blue"}, "vis_sl_short": {"type": "scatter", "color": "purple"}, "vis_tp_short": {"type": "scatter", "color": "purple"}, # маркеры фактического входа (когда «взяли отложку») "mark_entry_long": {"type": "scatter", "color": "green"}, "mark_entry_short": {"type": "scatter", "color": "red"}, }, "subplots": { "SKO z": { "sko_zscore": {"type": "scatter"} }, "R%": { "sko_Rpct_signal": {"type": "scatter"} }, } } # ===== helpers ===== @staticmethod def _series_color(close: pd.Series, open_: pd.Series) -> pd.Series: return (close > open_) @staticmethod def _to_utc_series(x: pd.Series | pd.Index) -> pd.DatetimeIndex: dt = pd.to_datetime(x, utc=True, errors='coerce') if not isinstance(dt, pd.DatetimeIndex): dt = pd.DatetimeIndex(dt) return dt def _locate_entry_pos(self, df: pd.DataFrame, entry_dt: datetime) -> Optional[int]: if entry_dt.tzinfo is None: entry_ts = pd.Timestamp(entry_dt).tz_localize('UTC') else: entry_ts = pd.Timestamp(entry_dt).tz_convert('UTC') if isinstance(df.index, pd.DatetimeIndex): dt_index = df.index.tz_localize('UTC') if df.index.tz is None else df.index.tz_convert('UTC') elif 'date' in df.columns: dt_index = self._to_utc_series(df['date']) else: return None pos = dt_index.get_indexer([entry_ts], method='nearest')[0] pos = 0 if pos < 0 else (len(df) - 1 if pos >= len(df) else pos) return int(pos) # ===== indicators + pending emulation ===== def _compute_all(self, df: pd.DataFrame) -> pd.DataFrame: s = int(self.signal_lag_bars.value) rr = float(self.RR_MULT.value) pend = int(self.PENDING_BARS.value) mode = str(self.ENTRY_MODE.value) # --- SKO по z-score объёма --- vol = df["volume"].astype("float64") lb = int(self.zscore_lookback.value) mean = vol.rolling(lb, min_periods=lb).mean() std = vol.rolling(lb, min_periods=lb).std(ddof=0) z = (vol - mean) / (std.replace(0, np.nan)) df["sko_zscore"] = z thr = float(self.zscore_threshold.value) is_sko = (z > thr).fillna(False) # цвет свечи (зелёная/красная) green = self._series_color(df["close"], df["open"]) # цвет СИГНАЛЬНОЙ свечи (= СКО + s) — сдвиг строго ВПЕРЁД shifted = green.shift(s) sig_green = shifted.where(shifted.notna(), False).astype(bool) # диапазон СКО, точка входа/триггер и фильтр по размеру R = (df["high"] - df["low"]).astype("float64") # цена на сигнальной свече (СКО+1): open для CONT, close для REV sig_open = df["open"].shift(s).astype("float64") sig_close = df["close"].shift(s).astype("float64") # для графика и фильтра: entry_ref = sig_open.copy() # просто чтобы Rpct считать стабильно df["sko_Rpct_signal"] = (R / entry_ref * 100.0) min_r_pct = float(self.MIN_R_PCT.value) big_enough = (df["sko_Rpct_signal"] >= min_r_pct) & entry_ref.notna() valid_base = is_sko & big_enough # --- триггеры по режимам --- want_rev = (mode in ("REVERSAL", "BOTH")) want_cont = (mode in ("CONTINUATION", "BOTH")) # триггер-цена на СИГНАЛЬНОЙ свече: trig_long = np.where(sig_green, (sig_close if want_rev else sig_open), np.nan) trig_short = np.where(~sig_green, (sig_close if want_rev else sig_open), np.nan) # «база»: есть СКО в t и мы рисуем сигналы на t+1 base_mask = valid_base # сдвигаем на t+1 (там и лежит триггер) trig_long = pd.Series(trig_long, index=df.index).shift(+s) trig_short = pd.Series(trig_short, index=df.index).shift(+s) base_mask = base_mask.shift(+s).fillna(False) # --- эмуляция «отложки»: берём ПЕРВУЮ свечу в окне [t+1 .. t+pend] где коснулись триггера --- n = len(df) enter_long = pd.Series(False, index=df.index) enter_short = pd.Series(False, index=df.index) mark_entry_long = pd.Series(np.nan, index=df.index, dtype="float64") mark_entry_short = pd.Series(np.nan, index=df.index, dtype="float64") # уровни SL/TP на участке ожидания (для визуализации) vis_trigger_long = pd.Series(np.nan, index=df.index, dtype="float64") vis_trigger_short = pd.Series(np.nan, index=df.index, dtype="float64") vis_sl_long = pd.Series(np.nan, index=df.index, dtype="float64") vis_tp_long = pd.Series(np.nan, index=df.index, dtype="float64") vis_sl_short = pd.Series(np.nan, index=df.index, dtype="float64") vis_tp_short = pd.Series(np.nan, index=df.index, dtype="float64") highs = df["high"].to_numpy(dtype="float64") lows = df["low"].to_numpy(dtype="float64") tlong = trig_long.to_numpy(dtype="float64") tshort = trig_short.to_numpy(dtype="float64") base = base_mask.to_numpy() sko_low = df["low"].shift(+s - 1) # low СКО-бара (t) sko_high = df["high"].shift(+s - 1) # high СКО-бара (t) for i in range(n): if not base[i]: continue # окна ожидания: [i .. i+pend-1] j_to = min(i + pend, n) if np.isfinite(tlong[i]): trig = tlong[i] # проверяем первое касание для LONG: high >= trig hit_idx = None for j in range(i, j_to): if highs[j] >= trig: hit_idx = j break # рисуем полосы ожидания vis_trigger_long.iloc[i:j_to] = trig if pd.notna(sko_low.iloc[i]) and pd.notna(sko_high.iloc[i]): R_i = float(sko_high.iloc[i] - sko_low.iloc[i]) vis_sl_long.iloc[i:j_to] = float(sko_low.iloc[i]) vis_tp_long.iloc[i:j_to] = float(trig + R_i * float(self.RR_MULT.value)) # фиксируем вход на первом касании if hit_idx is not None: enter_long.iloc[hit_idx] = True mark_entry_long.iloc[hit_idx] = highs[hit_idx] # просто маркер на графике if np.isfinite(tshort[i]): trig = tshort[i] # проверяем первое касание для SHORT: low <= trig hit_idx = None for j in range(i, j_to): if lows[j] <= trig: hit_idx = j break vis_trigger_short.iloc[i:j_to] = trig if pd.notna(sko_low.iloc[i]) and pd.notna(sko_high.iloc[i]): R_i = float(sko_high.iloc[i] - sko_low.iloc[i]) vis_sl_short.iloc[i:j_to] = float(sko_high.iloc[i]) vis_tp_short.iloc[i:j_to] = float(trig - R_i * float(self.RR_MULT.value)) if hit_idx is not None: enter_short.iloc[hit_idx] = True mark_entry_short.iloc[hit_idx] = lows[hit_idx] # сохраним в df то, что рисуем df["vis_trigger_long"] = vis_trigger_long df["vis_trigger_short"] = vis_trigger_short df["vis_sl_long"] = vis_sl_long df["vis_tp_long"] = vis_tp_long df["vis_sl_short"] = vis_sl_short df["vis_tp_short"] = vis_tp_short df["mark_entry_long"] = mark_entry_long df["mark_entry_short"] = mark_entry_short # эти флаги пойдут в populate_entry_trend df["sig_enter_long"] = enter_long df["sig_enter_short"] = enter_short # сохраним также «по факту» уровни SL/TP для custom_exit (по цене триггера) # нам нужен R от СКО и реальная entry_price = триггер df["pending_trig_long"] = trig_long df["pending_trig_short"] = trig_short df["sko_low_at_sig"] = sko_low df["sko_high_at_sig"] = sko_high return df # ===== Freqtrade hooks ===== def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: return self._compute_all(dataframe.copy()) def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: df = dataframe df.loc[:, "enter_long"] = df["sig_enter_long"].fillna(False) df.loc[:, "enter_short"] = df["sig_enter_short"].fillna(False) df.loc[:, "enter_tag"] = np.where(df["enter_long"], "PENDING_LONG", np.where(df["enter_short"], "PENDING_SHORT", "")) return df def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: dataframe.loc[:, "exit_long"] = 0 dataframe.loc[:, "exit_short"] = 0 return dataframe # ===== custom_exit ===== def custom_exit(self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[str]: try: df, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) except Exception: return None if df is None or df.empty: return None entry_dt = trade.open_date_utc if entry_dt is None: return None entry_pos = self._locate_entry_pos(df, entry_dt) if entry_pos is None: return None # восстановим уровни из ближайшей строки row = df.iloc[entry_pos] is_long = (trade.is_short is False) # entry мы считаем триггером (ему соответствует mark_entry_*), SL/TP — от СКО if is_long: trig = float(row.get("pending_trig_long", np.nan)) sko_low = float(row.get("sko_low_at_sig", np.nan)) sko_high = float(row.get("sko_high_at_sig", np.nan)) if np.isnan(trig) or np.isnan(sko_low) or np.isnan(sko_high): return None R = max(sko_high - sko_low, 0.0) sl = sko_low tp = trig + R * float(self.RR_MULT.value) else: trig = float(row.get("pending_trig_short", np.nan)) sko_low = float(row.get("sko_low_at_sig", np.nan)) sko_high = float(row.get("sko_high_at_sig", np.nan)) if np.isnan(trig) or np.isnan(sko_low) or np.isnan(sko_high): return None R = max(sko_high - sko_low, 0.0) sl = sko_high tp = trig - R * float(self.RR_MULT.value) # текущая свеча last = df.iloc[-1] ch, cl = float(last["high"]), float(last["low"]) if is_long: if cl <= sl: return "stop_loss" if ch >= tp: return "take_profit" else: if ch >= sl: return "stop_loss" if cl <= tp: return "take_profit" return None