bot/strategy/StrategyTemplate.py - dev - fgtmenow/tb

bot/strategy/StrategyTemplate.py
466 строк20 KB
Azarev Artem
Добавлена база знаний по стратегии Smart Money Trading, включая архитектуру, терминологию и торговые стратегии. Реализованы интерфейсы для типизации компонентов торгового бота и шаблон для торговых стратегий.
17 янв 2026, 03:39
17 янв 2026, 03:39ab5fefa
"""
Шаблон торговой стратегии.

Этот файл служит основой для создания новых торговых стратегий.
Скопируйте его и переименуйте под вашу стратегию.

Инструкция:
1. Скопируйте этот файл: cp StrategyTemplate.py MyStrategy.py
2. Переименуйте класс StrategyTemplate -> MyStrategy
3. Реализуйте логику в check_entry() и check_bar()
4. Добавьте стратегию в CheckStrategy.STRATEGIES

Пример добавления в CheckStrategy.py:
    from bot.strategy.MyStrategy import MyStrategy
    
    class CheckStrategy(BaseAction):
        STRATEGIES = [Example(), MyStrategy()]  # Добавьте сюда
"""

from _common.domain.enums.Action import Action
from _common.domain.enums.OrderType import OrderType
from _common.domain.enums.Side import Side
from _common.domain.model.OrderToPlace import OrderToPlace
from _common.domain.util.Logger import print_messaged
from _common.domain.util.ProfitCalc import calc_closed_pl
from bot.strategy.base.BaseStrategy import BaseStrategy


class StrategyTemplate(BaseStrategy):
    """
    Шаблон торговой стратегии.
    
    Наследуется от BaseStrategy и реализует два ключевых метода:
    - check_entry(): Проверка условий входа в позицию
    - check_bar(): Управление открытой позицией
    
    Доступные атрибуты (из BaseStrategy):
        self.name: str - Имя стратегии (автоматически = имя класса)
        self.store: Store - Глобальное хранилище (broker, model, session)
        self.model: DataModel - Доступ к ценовым данным
        self.instrumentCache: InstrumentCache - Информация об инструментах
    
    Доступные данные:
        self.model.data_1m[symbol]: DataFrame - Минутные данные
        self.model.data_5m[symbol]: DataFrame - 5-минутные данные
        self.model.get_bar(symbol): Bar - Последний бар
        self.model.get_bar(symbol, -2): Bar - Предпоследний бар
    
    Колонки DataFrame после анализа:
        - timestamp, open, high, low, close, volume: OHLCV
        - swing_point: 1 (high), -1 (low), NaN (нет)
        - swing_level: Цена swing point
        - trend: 1 (бычий), -1 (медвежий), NaN (нет)
        - bos: 1 (пробой структуры)
        - choch: 1 (смена тренда)
        - confirm: 1 (подтверждение)
    """
    
    def __init__(self):
        """
        Инициализация стратегии.
        
        Вызовите super().__init__(self.__class__.__name__) для автоматического
        присвоения имени стратегии (имя класса).
        
        Здесь инициализируйте:
        - Параметры стратегии
        - Внутреннее состояние
        - Конфигурационные значения
        """
        # Обязательно вызываем родительский конструктор
        super().__init__(self.__class__.__name__)
        
        # === Параметры стратегии ===
        # Настройте эти значения под вашу стратегию
        
        self.leverage: float = 10.0         # Торговое плечо
        self.position_size: float = 0.001   # Размер позиции (в базовой валюте)
        
        # Уровни Фибоначчи для входа/выхода
        self.entry_fibo: float = 0.382      # Уровень входа (38.2%)
        self.target_fibo: float = 1.618     # Уровень тейк-профита (161.8%)
        self.stop_fibo: float = 0.0         # Уровень стоп-лосса (0% = swing low)
        
        # Требования к тренду
        self.require_trend: bool = True     # Требовать подтверждённый тренд
        self.require_confirm: bool = True   # Требовать confirm (не только choch)
        
        # === Внутреннее состояние ===
        # Используйте для отслеживания между барами
        
        self.last_signal_bar: dict[str, int] = {}  # Последний бар с сигналом
        self.cooldown_bars: int = 5                 # Баров между сигналами
    
    def check_entry(self, symbol: str) -> OrderToPlace | None:
        """
        Проверяет условия входа в позицию.
        
        Вызывается на каждом новом баре, если нет открытой позиции.
        
        Аргументы:
            symbol: Торговый символ (например, 'BTCUSDT')
        
        Возвращает:
            OrderToPlace: Ордер для открытия позиции
            None: Если нет сигнала на вход
        
        Типы ордеров:
            OrderType.MARKET - Исполнить немедленно по рыночной цене
            OrderType.LIMIT - Исполнить при достижении указанной цены
        """
        # === Шаг 1: Проверка готовности данных ===
        
        # Минимальное количество баров для анализа
        if self.model.data_length(symbol) < 10:
            return None
        
        # Проверяем, что данные готовы (индикаторы рассчитаны)
        if not self.model.check_ready(symbol):
            return None
        
        # === Шаг 2: Получение данных ===
        
        # Получаем текущий бар
        bar = self.model.get_bar(symbol)
        
        # Получаем DataFrame с 5-минутными данными (там swing points и тренд)
        df_5m = self.model.data_5m[symbol]
        
        # Последний бар 5-минутки
        last_5m = df_5m.iloc[-1]
        
        # Текущий тренд (1 = бычий, -1 = медвежий, NaN = нет)
        trend = last_5m['trend']
        
        # Последний swing point (1 = high, -1 = low)
        swing_point = last_5m['swing_point']
        
        # Подтверждение тренда
        confirm = last_5m['confirm']
        
        # === Шаг 3: Проверка cooldown ===
        
        current_bar_idx = len(self.model.data_1m[symbol])
        last_signal = self.last_signal_bar.get(symbol, 0)
        
        if current_bar_idx - last_signal < self.cooldown_bars:
            return None  # Ждём cooldown
        
        # === Шаг 4: Логика входа в LONG ===
        
        # Пример: Вход в LONG при бычьем тренде после swing low
        if trend == 1:  # Бычий тренд
            if self.require_confirm and confirm != 1:
                return None  # Нет подтверждения тренда
            
            if swing_point == -1:  # Swing low (потенциальная точка входа)
                # Находим последние HH и LL для расчёта уровней
                hh_idx = df_5m.index[df_5m['swing_point'] == 1]
                ll_idx = df_5m.index[df_5m['swing_point'] == -1]
                
                if len(hh_idx) == 0 or len(ll_idx) == 0:
                    return None
                
                hh = df_5m.loc[hh_idx[-1], 'high']  # Последний HH
                ll = df_5m.loc[ll_idx[-1], 'low']   # Последний LL
                
                # Расчёт уровней по Фибоначчи
                range_size = hh - ll
                entry_price = ll + range_size * self.entry_fibo
                target_price = ll + range_size * self.target_fibo
                stop_price = ll - range_size * 0.1  # 10% ниже LL
                
                # Логирование (опционально)
                print_messaged(
                    bar.datetime(),
                    f"[{self.name}] LONG signal: {symbol} "
                    f"entry={entry_price:.2f} target={target_price:.2f} stop={stop_price:.2f}"
                )
                
                # Запоминаем бар сигнала
                self.last_signal_bar[symbol] = current_bar_idx
                
                # Возвращаем ордер
                return OrderToPlace(
                    type=OrderType.LIMIT,      # Лимитный ордер
                    symbol=symbol,
                    side=Side.LONG,
                    size=self.position_size,
                    entry_price=entry_price,
                    leverage=self.leverage,
                    strategy=self.name,
                    target_price=target_price,  # Take Profit
                    stop_price=stop_price,      # Stop Loss
                )
        
        # === Шаг 5: Логика входа в SHORT ===
        
        # Пример: Вход в SHORT при медвежьем тренде после swing high
        if trend == -1:  # Медвежий тренд
            if self.require_confirm and confirm != 1:
                return None
            
            if swing_point == 1:  # Swing high
                hh_idx = df_5m.index[df_5m['swing_point'] == 1]
                ll_idx = df_5m.index[df_5m['swing_point'] == -1]
                
                if len(hh_idx) == 0 or len(ll_idx) == 0:
                    return None
                
                hh = df_5m.loc[hh_idx[-1], 'high']
                ll = df_5m.loc[ll_idx[-1], 'low']
                
                range_size = hh - ll
                entry_price = hh - range_size * self.entry_fibo
                target_price = hh - range_size * self.target_fibo
                stop_price = hh + range_size * 0.1
                
                print_messaged(
                    bar.datetime(),
                    f"[{self.name}] SHORT signal: {symbol} "
                    f"entry={entry_price:.2f} target={target_price:.2f} stop={stop_price:.2f}"
                )
                
                self.last_signal_bar[symbol] = current_bar_idx
                
                return OrderToPlace(
                    type=OrderType.LIMIT,
                    symbol=symbol,
                    side=Side.SHORT,
                    size=self.position_size,
                    entry_price=entry_price,
                    leverage=self.leverage,
                    strategy=self.name,
                    target_price=target_price,
                    stop_price=stop_price,
                )
        
        # Нет сигнала
        return None
    
    def check_bar(self, symbol: str) -> Action:
        """
        Проверяет условия управления открытой позицией.
        
        Вызывается на каждом новом баре при наличии открытой позиции.
        
        Аргументы:
            symbol: Торговый символ
        
        Возвращает:
            Action.CLOSE_POSITION - Закрыть позицию
            Action.INCREASE_POSITION - Увеличить позицию (добавить к существующей)
            Action.NOTHING - Ничего не делать (удерживать позицию)
        
        Доступ к позиции:
            pos = self.store.broker.get_position(symbol)
            pos.entry_price - цена входа
            pos.size - размер
            pos.side - направление (LONG/SHORT)
            pos.leverage - плечо
        """
        # === Получаем информацию о позиции ===
        
        pos = self.store.broker.get_position(symbol)
        bar = self.model.get_bar(symbol)
        df_5m = self.model.data_5m[symbol]
        
        # Текущий тренд
        current_trend = df_5m.iloc[-1]['trend']
        
        # === Расчёт текущего P&L ===
        
        pl = calc_closed_pl(
            entry_commission_perc=pos.entry_commission,
            exit_commission_perc=self.store.broker.market_fee,
            funding_rate_perc=self.instrumentCache.get_funding_rate(symbol),
            side=pos.side,
            position_size=pos.size,
            entry_price=pos.entry_price,
            exit_price=bar.close,
            leverage=pos.leverage
        )
        
        # === Условия выхода ===
        
        # 1. Take Profit по прибыли
        # Закрываем если прибыль > 0.5% от позиции
        if pl > 0.5:
            print_messaged(
                bar.datetime(),
                f"[{self.name}] Take Profit: {symbol} P&L={pl:.2f}%"
            )
            return Action.CLOSE_POSITION
        
        # 2. Stop Loss по убытку
        # Закрываем если убыток > 0.3%
        if pl < -0.3:
            print_messaged(
                bar.datetime(),
                f"[{self.name}] Stop Loss: {symbol} P&L={pl:.2f}%"
            )
            return Action.CLOSE_POSITION
        
        # 3. Выход при смене тренда (CHoCH)
        # LONG позиция при развороте в медвежий тренд
        if pos.side == Side.LONG and current_trend == -1:
            print_messaged(
                bar.datetime(),
                f"[{self.name}] Trend Reversal Exit: {symbol}"
            )
            return Action.CLOSE_POSITION
        
        # SHORT позиция при развороте в бычий тренд
        if pos.side == Side.SHORT and current_trend == 1:
            print_messaged(
                bar.datetime(),
                f"[{self.name}] Trend Reversal Exit: {symbol}"
            )
            return Action.CLOSE_POSITION
        
        # === Условия увеличения позиции ===
        
        # Пример: Увеличиваем позицию при подтверждении тренда
        # confirm = df_5m.iloc[-1]['confirm']
        # if confirm == 1 and pl > 0.1:
        #     return Action.INCREASE_POSITION
        
        # === По умолчанию: удерживаем позицию ===
        
        return Action.NOTHING


# =============================================================================
# ДОПОЛНИТЕЛЬНЫЕ ПРИМЕРЫ СТРАТЕГИЙ
# =============================================================================

class SimpleMomentumStrategy(BaseStrategy):
    """
    Пример простой моментум-стратегии.
    
    Входит в LONG когда цена растёт N баров подряд.
    Выходит при первом красном баре.
    """
    
    def __init__(self):
        super().__init__(self.__class__.__name__)
        self.green_bars_required = 3
        self.position_size = 0.001
        self.leverage = 10
    
    def check_entry(self, symbol: str) -> OrderToPlace | None:
        if self.model.data_length(symbol) < self.green_bars_required + 1:
            return None
        
        # Считаем зелёные бары подряд
        green_count = 0
        for i in range(-1, -self.green_bars_required - 1, -1):
            bar = self.model.get_bar(symbol, i)
            if bar.close > bar.open:
                green_count += 1
            else:
                break
        
        if green_count >= self.green_bars_required:
            bar = self.model.get_bar(symbol)
            return OrderToPlace(
                type=OrderType.MARKET,
                symbol=symbol,
                side=Side.LONG,
                size=self.position_size,
                entry_price=bar.close,
                leverage=self.leverage,
                strategy=self.name,
                target_price=None,
                stop_price=None,
            )
        
        return None
    
    def check_bar(self, symbol: str) -> Action:
        bar = self.model.get_bar(symbol)
        
        # Выходим при красном баре
        if bar.close < bar.open:
            return Action.CLOSE_POSITION
        
        return Action.NOTHING


class MeanReversionStrategy(BaseStrategy):
    """
    Пример стратегии возврата к среднему.
    
    Входит в LONG когда цена значительно ниже средней.
    Выходит при возврате к средней.
    """
    
    def __init__(self):
        super().__init__(self.__class__.__name__)
        self.lookback = 20          # Период для расчёта средней
        self.entry_deviation = 2.0  # Стандартных отклонений для входа
        self.position_size = 0.001
        self.leverage = 5
    
    def check_entry(self, symbol: str) -> OrderToPlace | None:
        if self.model.data_length(symbol) < self.lookback + 1:
            return None
        
        df = self.model.data_1m[symbol]
        closes = df['close'].tail(self.lookback)
        
        mean = closes.mean()
        std = closes.std()
        
        current_price = self.model.close(symbol)
        
        # Z-score
        z_score = (current_price - mean) / std if std > 0 else 0
        
        # Вход в LONG при сильном отклонении вниз
        if z_score < -self.entry_deviation:
            return OrderToPlace(
                type=OrderType.MARKET,
                symbol=symbol,
                side=Side.LONG,
                size=self.position_size,
                entry_price=current_price,
                leverage=self.leverage,
                strategy=self.name,
                target_price=mean,  # Target = средняя
                stop_price=None,
            )
        
        # Вход в SHORT при сильном отклонении вверх
        if z_score > self.entry_deviation:
            return OrderToPlace(
                type=OrderType.MARKET,
                symbol=symbol,
                side=Side.SHORT,
                size=self.position_size,
                entry_price=current_price,
                leverage=self.leverage,
                strategy=self.name,
                target_price=mean,
                stop_price=None,
            )
        
        return None
    
    def check_bar(self, symbol: str) -> Action:
        df = self.model.data_1m[symbol]
        closes = df['close'].tail(self.lookback)
        mean = closes.mean()
        
        pos = self.store.broker.get_position(symbol)
        current_price = self.model.close(symbol)
        
        # Выходим при возврате к средней
        if pos.side == Side.LONG and current_price >= mean:
            return Action.CLOSE_POSITION
        
        if pos.side == Side.SHORT and current_price <= mean:
            return Action.CLOSE_POSITION
        
        return Action.NOTHING