Timeframe
N/A
Direction
Long Only
Stoploss
N/A
Trailing Stop
No
ROI
N/A
Interface Version
N/A
Startup Candles
N/A
Indicators
0
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# SkoRZ1RStrategyV2LogFix.py
# Version: 2.2.0
# ИСПРАВЛЕНО:
# - Правильная логика определения направления СКО-свечи
# - Корректный расчет уровней SL/TP относительно СКО-бара
# - Улучшенное логирование с детальной диагностикой
# - Очистка состояния трейдов после закрытия
from typing import Dict, Any, Optional
import os
import logging
import pandas as pd
import numpy as np
from freqtrade.strategy.interface import IStrategy
from SkoRZIndicatorV1 import SkoRZIndicatorV1, SkoConfig
class SkoRZ1RStrategyV2LogFix(IStrategy):
__version__ = "2.2.0"
can_short: bool = True
timeframe: str = "5m"
process_only_new_candles: bool = True
startup_candle_count: int = 100
minimal_roi: Dict[str, float] = {"0": 100}
use_custom_stoploss: bool = False
stoploss: float = -1.0
trailing_stop: bool = False
DEBUG_EXITS: bool = True
LOG_TO_FILE: bool = True
LOG_FILEPATH: str = "user_data/logs/sko_debug.log"
_trade_state: Dict[int, Dict[str, Any]] = {}
cfg = SkoConfig(
min_vol_ratio_prev=1.2,
min_vol_ratio_next=1.3,
zscore_lookback=100,
zscore_threshold=2.0,
use_log_volume=True,
signal_lag_bars=1,
)
_ind = SkoRZIndicatorV1(cfg)
plot_config = {
"main_plot": {
"sko_entry_long": {"type": "scatter", "color": "green"},
"sko_entry_short": {"type": "scatter", "color": "red"},
"sko_stop_long": {"type": "scatter", "color": "orange"},
"sko_take_long": {"type": "scatter", "color": "lime"},
"sko_stop_short": {"type": "scatter", "color": "orange"},
"sko_take_short": {"type": "scatter", "color": "pink"},
},
"subplots": {
"SKO z-score": {
"sko_zscore": {"type": "scatter", "color": "blue"}
},
},
}
def __init__(self, config: dict) -> None:
super().__init__(config)
self._file_logger: Optional[logging.Logger] = None
if self.LOG_TO_FILE:
try:
os.makedirs(os.path.dirname(self.LOG_FILEPATH), exist_ok=True)
lg = logging.getLogger(self.__class__.__name__ + ".file")
lg.setLevel(logging.INFO)
# Проверим, не добавлен ли уже наш хендлер
has_file = False
for h in lg.handlers:
if isinstance(h, logging.FileHandler) and getattr(h, "_sko_file", False):
has_file = True
break
if not has_file:
fh = logging.FileHandler(self.LOG_FILEPATH, mode="a", encoding="utf-8")
fh.setLevel(logging.INFO)
fmt = logging.Formatter(
"%(asctime)s | %(levelname)s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
fh.setFormatter(fmt)
setattr(fh, "_sko_file", True)
lg.addHandler(fh)
self._file_logger = lg
self._log_any(f"[INIT] Strategy {self.__version__} initialized. File logging: {self.LOG_FILEPATH}")
except Exception as e:
self._file_logger = None
print(f"Failed to setup file logger: {e}")
# --------- helpers ----------
def _log_any(self, msg: str) -> None:
"""Log to both file and strategy logger"""
if self._file_logger:
self._file_logger.info(msg)
try:
if hasattr(self, 'logger') and self.logger:
self.logger.info(msg)
except Exception:
pass
def _log_exit(self, msg: str) -> None:
"""Log exit-related messages"""
if not self.DEBUG_EXITS:
return
self._log_any(msg)
# --------- freqtrade interface ----------
def informative_pairs(self):
return []
def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
"""Compute SKO indicators"""
df = self._ind.compute(dataframe)
# Добавим диагностические колонки
if self.DEBUG_EXITS:
# Подсчитаем количество валидных СКО
sko_count = df["sko_is_valid"].sum()
self._log_any(f"[INDICATORS] Total SKO candles found: {sko_count}")
return df
def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
"""Mark entry signals based on SKO indicator"""
df = dataframe.copy()
# Установка сигналов входа
long_signals = df["sko_entry_long"].notna()
short_signals = df["sko_entry_short"].notna()
df.loc[long_signals, "enter_long"] = 1
df.loc[short_signals, "enter_short"] = 1
# Логирование количества сигналов
if self.DEBUG_EXITS:
n_long = long_signals.sum()
n_short = short_signals.sum()
self._log_any(f"[ENTRY_TREND] Long signals: {n_long}, Short signals: {n_short}")
return df
def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
"""Exit handled by custom_exit"""
return dataframe
# --------- levels calculation ----------
def _compute_levels_from_sko(
self,
df: pd.DataFrame,
entry_pos: int,
trade,
) -> Optional[Dict[str, Any]]:
"""
Вычисляем уровни SL/TP на основе СКО-свечи.
entry_pos - позиция бара входа (где открылась сделка)
СКО-бар = entry_pos - signal_lag_bars
"""
signal_lag = int(self.cfg.signal_lag_bars)
sko_pos = entry_pos - signal_lag
if sko_pos < 0 or entry_pos >= len(df):
self._log_exit(f"[LEVELS] Invalid positions: sko_pos={sko_pos}, entry_pos={entry_pos}, len={len(df)}")
return None
try:
# Данные бара входа
entry_row = df.iloc[entry_pos]
entry_open = float(entry_row["open"])
entry_time = entry_row.name
# Данные СКО-бара
sko_row = df.iloc[sko_pos]
sko_high = float(sko_row["high"])
sko_low = float(sko_row["low"])
sko_close = float(sko_row["close"])
sko_open = float(sko_row["open"])
sko_time = sko_row.name
# Определяем направление по СКО-свече
sko_is_green = sko_close > sko_open
# Проверяем, что СКО-свеча действительно валидная
if not sko_row.get("sko_is_valid", False):
self._log_exit(f"[LEVELS] Warning: SKO bar at {sko_pos} is not marked as valid!")
except Exception as e:
self._log_exit(f"[LEVELS] Error extracting data: {e}")
return None
# Диапазон СКО-свечи
rng = sko_high - sko_low
if rng <= 0:
self._log_exit(f"[LEVELS] Invalid range: high={sko_high}, low={sko_low}")
return None
# Определяем направление трейда
is_short_attr = getattr(trade, "is_short", None)
if is_short_attr is not None:
is_long = not trade.is_short
else:
is_long = trade.amount > 0
# Расчет SL/TP
if is_long:
sl = sko_low
tp = entry_open + rng
else:
sl = sko_high
tp = entry_open - rng
return {
"entry_open": entry_open,
"entry_time": entry_time,
"entry_pos": entry_pos,
"sko_high": sko_high,
"sko_low": sko_low,
"sko_close": sko_close,
"sko_open": sko_open,
"sko_time": sko_time,
"sko_pos": sko_pos,
"sko_is_green": sko_is_green,
"rng": rng,
"sl": sl,
"tp": tp,
"is_long": is_long,
}
# --------- custom exit ----------
def custom_exit(
self,
pair: str,
trade,
current_time: pd.Timestamp,
current_rate: float,
current_profit: float,
**kwargs: Any,
) -> Optional[str]:
"""
Проверяем условия выхода по SL/TP на каждом тике.
"""
df: pd.DataFrame = kwargs.get("dataframe", None)
candle = kwargs.get("current_candle", None)
if df is None or df.empty or candle is None:
return None
# Находим позицию бара входа
entry_time = trade.open_date_utc
try:
# Используем get_indexer с method='pad' для поиска ближайшего времени
entry_pos = df.index.get_indexer([entry_time], method="pad")[0]
if entry_pos < 0:
# Fallback на точный поиск
entry_pos = df.index.get_loc(entry_time)
except Exception as e:
self._log_exit(f"[EXIT] Cannot find entry position for {entry_time}: {e}")
return None
# Получаем или вычисляем уровни
trade_id = trade.trade_id
if trade_id not in self._trade_state:
# Первый тик этой сделки - вычисляем и сохраняем уровни
levels = self._compute_levels_from_sko(df, entry_pos, trade)
if not levels:
self._log_exit(f"[EXIT] Failed to compute levels for trade #{trade_id}")
return None
self._trade_state[trade_id] = {
"levels": levels,
"tick_count": 0,
"logged_passport": False,
}
state = self._trade_state[trade_id]
levels = state["levels"]
state["tick_count"] += 1
# Логируем паспорт сделки (один раз)
if not state["logged_passport"]:
state["logged_passport"] = True
direction = "LONG" if levels["is_long"] else "SHORT"
sko_color = "GREEN" if levels["sko_is_green"] else "RED"
self._log_exit(
f"\n{'='*80}\n"
f"[PASSPORT] Trade #{trade_id} | {pair} | {direction}\n"
f" Entry: time={levels['entry_time']} pos={levels['entry_pos']} open={levels['entry_open']:.2f}\n"
f" SKO: time={levels['sko_time']} pos={levels['sko_pos']} {sko_color} "
f"O={levels['sko_open']:.2f} C={levels['sko_close']:.2f} "
f"H={levels['sko_high']:.2f} L={levels['sko_low']:.2f}\n"
f" Range: R={levels['rng']:.2f} | SL={levels['sl']:.2f} TP={levels['tp']:.2f}\n"
f"{'='*80}"
)
# Текущие данные свечи
try:
chigh = float(candle["high"])
clow = float(candle["low"])
cclose = float(candle.get("close", current_rate))
except Exception as e:
self._log_exit(f"[EXIT] Error reading candle data: {e}")
return None
sl = levels["sl"]
tp = levels["tp"]
# Логируем тик
self._log_exit(
f"[TICK #{state['tick_count']}] Trade #{trade_id} @ {current_time} | "
f"Candle H={chigh:.2f} L={clow:.2f} C={cclose:.2f} | "
f"Check SL={sl:.2f} TP={tp:.2f}"
)
# Проверка условий выхода
exit_reason = None
exit_price = None
if levels["is_long"]:
# LONG: выход по SL если low <= SL
if clow <= sl:
exit_reason = "stop_loss"
exit_price = sl
self._log_exit(
f"[EXIT] Trade #{trade_id} LONG -> STOP LOSS triggered! "
f"Low={clow:.2f} <= SL={sl:.2f}"
)
# LONG: выход по TP если high >= TP
elif chigh >= tp:
exit_reason = "take_profit"
exit_price = tp
self._log_exit(
f"[EXIT] Trade #{trade_id} LONG -> TAKE PROFIT triggered! "
f"High={chigh:.2f} >= TP={tp:.2f}"
)
else:
# SHORT: выход по SL если high >= SL
if chigh >= sl:
exit_reason = "stop_loss"
exit_price = sl
self._log_exit(
f"[EXIT] Trade #{trade_id} SHORT -> STOP LOSS triggered! "
f"High={chigh:.2f} >= SL={sl:.2f}"
)
# SHORT: выход по TP если low <= TP
elif clow <= tp:
exit_reason = "take_profit"
exit_price = tp
self._log_exit(
f"[EXIT] Trade #{trade_id} SHORT -> TAKE PROFIT triggered! "
f"Low={clow:.2f} <= TP={tp:.2f}"
)
# Если сработал выход - очищаем состояние
if exit_reason:
if trade_id in self._trade_state:
del self._trade_state[trade_id]
self._log_exit(f"[EXIT] Trade #{trade_id} closed with {exit_reason} @ {exit_price:.2f}\n")
return exit_reason