bot/actions/CheckStrategy.py - dev - fgtmenow/tb

bot/actions/CheckStrategy.py
138 строк6 KB
Azarev Artem
Add project configuration and initial implementation of trading bot
17 янв 2026, 02:00
17 янв 2026, 02:005246123
from _common.domain.enums.Action import Action
from _common.domain.enums.OrderType import OrderType
from _common.domain.enums.Side import Side
from bot.Config import SKIP_FRAMES, LEVERAGE
from bot.actions.base.BaseAction import BaseAction
from bot.objects.EventProcessor import EventProcessor
from bot.strategy.Example import Example

"""
Класс для проверки и выполнения торговых стратегий.
Отвечает за проверку условий входа в позицию, выхода из позиции и управление существующими позициями.

Attributes:
    STRATEGIES (list): Список доступных торговых стратегий
    symbol (str): Торговый символ
"""


class CheckStrategy(BaseAction):
    STRATEGIES = [Example()]

    def __init__(self, symbol: str):
        """
        Инициализация проверки стратегий.
        
        Args:
            symbol (str): Торговый символ
        """
        super().__init__(self.__class__.__name__)
        self.symbol = symbol

    def __lt__(self, other):
        """
        Сравнение для сортировки по символу.
        
        Args:
            other: Другой объект для сравнения
            
        Returns:
            bool: Результат сравнения
        """
        return self.symbol < other.symbol

    def work(self, processor: EventProcessor):
        """
        Выполняет проверку стратегий и управление позициями.
        
        Проверяет:
        1. Достаточно ли данных для анализа
        2. Если есть открытая позиция - проверяет условия выхода или увеличения
        3. Если нет позиции - проверяет условия входа для всех стратегий
        
        Args:
            processor (EventProcessor): Процессор событий
        """
        if len(self.store.model.data_1m[self.symbol]) < SKIP_FRAMES:
            return
        bar = self.store.model.get_bar(symbol=self.symbol)

        if self.store.broker.is_in_position(self.symbol):
            # check exit for position strategy
            position = self.store.broker.get_position(self.symbol)
            st = next(s for s in self.STRATEGIES if s.name == position.strategy)
            action = st.check_bar(self.symbol)
            match action:
                case Action.CLOSE_POSITION:
                    self.store.broker.execute_market_order(
                        strategy=st.name,
                        datetime=bar.datetime(),
                        symbol=self.symbol,
                        size=position.size,
                        market_price=bar.close,
                        side=position.side.invert(),
                        leverage=LEVERAGE
                    )
                case Action.INCREASE_POSITION:
                    self.store.broker.execute_market_order(
                        strategy=st.name,
                        datetime=bar.datetime(),
                        symbol=self.symbol,
                        size=position.size,
                        market_price=bar.close,
                        side=Side.LONG,
                        leverage=LEVERAGE
                    )

                case Action.NOTHING:
                    pass

        else:
            if self.store.broker.orders.__contains__(self.symbol):
                return
            # check enter for all strategies
            for strategy in self.STRATEGIES:
                order_to_place = strategy.check_entry(self.symbol)
                if order_to_place is not None:
                    if order_to_place.type == OrderType.MARKET:
                        self.store.broker.execute_market_order(
                            strategy=order_to_place.strategy,
                            datetime=bar.datetime(),
                            symbol=order_to_place.symbol,
                            size=order_to_place.size,
                            market_price=bar.close,
                            side=order_to_place.side,
                            leverage=order_to_place.leverage
                        )
                        return
                    else:
                        self.store.broker.place_limit_order(
                            strategy=order_to_place.strategy,
                            datetime=bar.datetime(),
                            symbol=order_to_place.symbol,
                            size=order_to_place.size,
                            price=order_to_place.entry_price,
                            side=order_to_place.side,
                            leverage=order_to_place.leverage
                        )
                        if order_to_place.target_price is not None:
                            self.store.broker.place_limit_order(
                                strategy=order_to_place.strategy,
                                datetime=bar.datetime(),
                                symbol=order_to_place.symbol,
                                size=order_to_place.size,
                                price=order_to_place.target_price,
                                side=order_to_place.side.invert(),
                                leverage=order_to_place.leverage
                            )
                        if order_to_place.stop_price is not None:
                            self.store.broker.place_limit_order(
                                strategy=order_to_place.strategy,
                                datetime=bar.datetime(),
                                symbol=order_to_place.symbol,
                                size=order_to_place.size,
                                price=order_to_place.stop_price,
                                side=order_to_place.side.invert(),
                                leverage=order_to_place.leverage
                            )
                        return