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

bot/actions/NewTimeFrame.py
92 строки3 KB
Azarev Artem
add base bot sources
25 мар 2025, 20:11
25 мар 2025, 20:11184b0c6
from _common.domain.enums.Priority import Priority
from _common.domain.util.DateUtils import timestamp_to_datetime
from bot.actions.CheckStrategy import CheckStrategy
from bot.actions.base.BaseAction import BaseAction
from bot.objects.EventProcessor import EventProcessor

"""
Класс для обработки нового таймфрейма.
Отвечает за добавление новых данных в модель и обновление состояния брокера.

Attributes:
    symbol (str): Торговый символ
    timestamp (int): Временная метка
    high (float): Максимальная цена
    low (float): Минимальная цена
    open (float): Цена открытия
    close (float): Цена закрытия
    volume (float): Объем торгов
"""

class NewTimeFrame(BaseAction):

    def __init__(self,
                 symbol: str,
                 timestamp: int,
                 high: float,
                 low: float,
                 open: float,
                 close: float,
                 volume: float,
                 ):
        """
        Инициализация обработчика нового таймфрейма.
        
        Args:
            symbol (str): Торговый символ
            timestamp (int): Временная метка
            high (float): Максимальная цена
            low (float): Минимальная цена
            open (float): Цена открытия
            close (float): Цена закрытия
            volume (float): Объем торгов
        """
        super().__init__(self.__class__.__name__)
        self.symbol = symbol
        self.timestamp = timestamp
        self.high = high
        self.low = low
        self.open = open
        self.close = close
        self.volume = volume

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

    def work(self, processor: EventProcessor):
        """
        Обрабатывает новый таймфрейм:
        1. Добавляет данные в модель
        2. Обновляет состояние брокера
        3. Запускает проверку стратегий
        
        Args:
            processor (EventProcessor): Процессор событий
        """
        self.store.model.append_timeframe(
            symbol=self.symbol,
            open=self.open,
            high=self.high,
            low=self.low,
            close=self.close,
            volume=self.volume,
            timestamp=self.timestamp
        )
        self.store.broker.process_frame(
            symbol=self.symbol,
            datetime=timestamp_to_datetime(self.timestamp),
            open=self.open,
            close=self.close,
            high=self.high,
            low=self.low
        )
        processor.enqueue(key=self.symbol, action=CheckStrategy(self.symbol), priority=Priority.URGENT)