bot/feed/BacktestFeed.py - dev - fgtmenow/tb

bot/feed/BacktestFeed.py
110 строк5 KB
Azarev Artem
plot results
17 апр 2025, 15:10
17 апр 2025, 15:10ba7c454
from _common.data.repository.BacktestRepository import getBacktestPeriodConfig
from _common.data.repository.KlineRepository import get_kline_data_frame
from _common.domain.util.Logger import print_message, print_header
from bot.Config import BACKTEST_PERIOD, TIME_FRAME, SHOW_PLOT
from bot.actions.NewTimeFrame import NewTimeFrame
from bot.feed.base.BaseFeed import BaseFeed


class BacktestFeed(BaseFeed):
    """
    Класс для работы с историческими данными в режиме бэктестинга.
    
    Загружает исторические данные и эмулирует их последовательную подачу
    для тестирования торговых стратегий на исторических данных.
    
    Атрибуты:
        result_showed (bool): Флаг, показывающий были ли выведены результаты бэктеста
    """

    def __init__(self):
        super().__init__()
        self.result_showed = False

    def load_data(self):
        print_message("Start load backtest data")
        period = getBacktestPeriodConfig(BACKTEST_PERIOD)
        datas = get_kline_data_frame(
            session=self.store.session,
            timeframe=TIME_FRAME,
            instruments=self.instrumentCache.instruments,
            period=period,
            cache=True
        )
        print_message(f"Finish load backtest data: {len(datas)}")
        return datas

    def start(self):
        """
        Запускает процесс бэктестинга:
        1. Загружает исторические данные для заданного периода
        2. Последовательно обрабатывает каждый временной фрейм
        3. Отправляет данные в обработчик событий
        """
        datas = self.load_data()

        start_idx = 0
        for data_index in range(0, len(datas)):
            symbol = list(self.instrumentCache.instruments.keys())[data_index]
            df_init = datas[data_index].iloc[:len(datas[data_index]) // 2].copy()
            self.store.model.set_initial(symbol, df_init)
            start_idx = len(df_init)

        print_header()
        row_count = len(datas[0])
        for timeframe_index in range(start_idx, row_count):
            for data_index in range(0, len(datas)):
                # on each frame send action
                symbol = list(self.instrumentCache.instruments.keys())[data_index]
                frame = datas[data_index].iloc[timeframe_index]
                self.processor.enqueue(
                    key=symbol,
                    action=NewTimeFrame(
                        symbol=symbol,
                        timestamp=frame['timestamp'],
                        high=frame['high'],
                        low=frame['low'],
                        open=frame['open'],
                        close=frame['close'],
                        volume=frame['volume'],
                    )
                )
        self.processor.register_empty_queue_callback(self.empty_queue_callback)
        self.processor.start()

    def empty_queue_callback(self):
        """
        Callback, вызываемый при завершении обработки всех данных.
        
        Закрывает все открытые позиции, останавливает процессор событий
        и выводит результаты бэктестинга, включая графики.
        """
        # close all positions
        positions = self.store.broker.positions.copy()
        if len(self.store.broker.positions) > 0:
            for symbol, position in positions.items():
                last_bar = self.store.model.get_bar(symbol, -1)
                self.store.broker.execute_market_order(
                    strategy=position.strategy,
                    datetime=last_bar.datetime(),
                    symbol=symbol,
                    size=position.size,
                    market_price=last_bar.close,
                    side=position.side.invert(),
                    leverage=position.leverage
                )
                self.stat.log_balance(
                    datetime=self.store.model.get_bar(symbol, -1).datetime(),
                    equity=self.store.broker.available,
                    available=self.store.broker.available
                )

        print()
        self.processor.stop()
        if not self.result_showed:
            pass
            self.result_showed = True

            self.stat.print_stat()
            if SHOW_PLOT:
                self.stat.plot(self.store.model.data_1m, self.store.model.data_5m)