bot/objects/BacktestStat.py - dev - fgtmenow/tb

bot/objects/BacktestStat.py
422 строки18 KB
Azarev Artem
plot results
17 апр 2025, 15:10
17 апр 2025, 15:10ba7c454
import datetime as dt
from statistics import fmean

import numpy as np
import pandas as pd
import plotly.graph_objects as go
from pandas import DataFrame
from plotly.subplots import make_subplots

from _common.domain.enums.Side import Side
from _common.domain.util.SignletonMeta import SingletonMeta
from bot.data.market_structure import market_structure
from bot.objects.InstrumentCache import InstrumentCache

# Определяем константы для цветов
BULLISH_COLOR: str = '#089981'  # Зеленый цвет для бычьих паттернов
BEARISH_COLOR: str = '#F23645'  # Красный цвет для медвежьих паттернов
SWING_COLOR: str = '#2157f3'  # Синий для swing points


class BacktestStat(metaclass=SingletonMeta):

    def __init__(self):
        self.task_avg: dict[str, list[int]] = {}
        self.balances = pd.DataFrame(columns=['timestamp', 'equity', 'available'])
        self.trade_profits: dict[str, list[float]] = {}
        self.trades: dict[str, pd.DataFrame] = {}
        self.instrumentCache = InstrumentCache()

    def log_task(self, key: str, time_ms: int):
        if self.task_avg.__contains__(key):
            self.task_avg[key].append(time_ms)
        else:
            self.task_avg[key] = [time_ms]

    def log_balance(self, datetime: dt.datetime, equity: float, available: float):
        """
        Логирование баланса кошелька
        :param datetime: Время логирования
        :param equity: Общий баланс кошелька
        :param available: Доступный баланс кошелька
        """
        new_entry = pd.DataFrame({
            'timestamp': [datetime],
            'equity': [equity],
            'available': [available]
        })
        if len(self.balances) == 0:
            self.balances = new_entry
        else:
            self.balances = pd.concat([self.balances, new_entry], ignore_index=True)

    def log_trades(self, symbol: str, datetime: dt.datetime, side: Side, price: float):
        """
        Логирование торгов
        :param symbol: Символ
        :param datetime: Время действия
        :param side: Тип сделки
        :param price: Цена действия
        """
        new_trade = pd.DataFrame({
            'datetime': [datetime],
            'side': [side],
            'price': [price]
        })
        if symbol in self.trades:
            self.trades[symbol] = pd.concat([self.trades[symbol], new_trade], ignore_index=True)
        else:
            self.trades[symbol] = new_trade

    def log_trade_profit(self,
                         symbol: str,
                         profit: float
                         ):
        """
        Логирование результатов торгов
        :param symbol: Символ
        :param profit: прибыль
        """
        if symbol in self.trade_profits:
            self.trade_profits[symbol].append(profit)
        else:
            self.trade_profits[symbol] = [profit]

    def print_stat(self):
        print()
        print("Task time avg:")
        for key in sorted(self.task_avg.keys()):
            time = self.task_avg[key]
            print('{:<{}}: {} ms'.format(key, 20, round(fmean(time), 1)))
        print()

        start_eq = self.balances.iloc[0]["equity"]
        end_eq = self.balances.iloc[-1]["equity"]

        print("Trade stat:")
        print('{:<{}}: {}$'.format("Start equity:", 20, round(start_eq, 2)))
        print('{:<{}}: {}$'.format("End equity:", 20, round(end_eq, 2)))
        print('{:<{}}: {}$'.format("Profit:", 20, round(end_eq - start_eq, 2)))
        print('{:<{}}: {}%'.format("ROI:", 20, round((end_eq - start_eq) / start_eq * 100, 2)))
        print()

        total_trades = sum([len(trades) for trades in self.trade_profits.values()])

        if total_trades > 0:
            win_trades = 0
            total_profit = 0
            total_loss = 0

            for symbol, profits in self.trade_profits.items():
                for profit in profits:
                    if profit > 0:
                        win_trades += 1
                        total_profit += profit
                    else:
                        total_loss += profit

            win_rate = (win_trades / total_trades) * 100

            print("Trade:")
            print('{:<{}}: {}'.format("Wins:", 20, round(win_trades)))
            print('{:<{}}: {}'.format("Loses:", 20, round(total_trades - win_trades)))
            print('{:<{}}: {}'.format("Total:", 20, round(total_trades)))
            print('{:<{}}: {}%'.format("WinRate:", 20, round(win_rate, 2)))
            print()
            print('{:<{}}: {}$'.format("Win:", 20, round(total_profit, 2)))
            print('{:<{}}: {}$'.format("Loss:", 20, round(total_loss, 2)))
            print('{:<{}}: {}$'.format("Profit:", 20, round(total_profit + total_loss, 2)))

    def __map_swing_points_to_original(
            self,
            df_original: pd.DataFrame,
            df_resampled: pd.DataFrame,
            resample_period: str
    ) -> None:
        """
        Отображает найденные точки разворота с ресемплированного DataFrame на оригинальный.

        Args:
            df_original: Оригинальный DataFrame, куда будут отображены точки
            df_resampled: Ресемплированный DataFrame с найденными точками разворота
            resample_period: Период ресемплирования, использованный для df_resampled
        """
        # Инициализируем колонки в оригинальном DataFrame, если они еще не существуют
        if 'swing_point' not in df_original.columns:
            df_original.loc[:, 'swing_point'] = np.nan
        if 'swing_level' not in df_original.columns:
            df_original.loc[:, 'swing_level'] = np.nan
        if 'swing_retracement' not in df_original.columns:
            df_original.loc[:, 'swing_retracement'] = np.nan

        # Отображаем точки максимумов (highs) обратно на оригинальный DataFrame
        high_swings = df_resampled[df_resampled['swing_point'] == 1]
        for _, row in high_swings.iterrows():
            mask = (df_original.index >= row.name) & (df_original.index < row.name + pd.Timedelta(resample_period))
            matching_rows = df_original[mask & (df_original['high'] == row['high'])]
            if not matching_rows.empty:
                idx = matching_rows.index[0]
                df_original.loc[idx, 'swing_point'] = 1
                df_original.loc[idx, 'swing_level'] = row['high']
                df_original.loc[idx, 'swing_retracement'] = row['swing_retracement']

        # Отображаем точки минимумов (lows) обратно на оригинальный DataFrame
        low_swings = df_resampled[df_resampled['swing_point'] == -1]
        for _, row in low_swings.iterrows():
            mask = (df_original.index >= row.name) & (df_original.index < row.name + pd.Timedelta(resample_period))
            matching_rows = df_original[mask & (df_original['low'] == row['low'])]
            if not matching_rows.empty:
                idx = matching_rows.index[0]
                df_original.loc[idx, 'swing_point'] = -1
                df_original.loc[idx, 'swing_level'] = row['low']
                df_original.loc[idx, 'swing_retracement'] = row['swing_retracement']

    def plot(self, data_1m: dict[str, DataFrame], data_5m: dict[str, DataFrame]):
        """
        Отображение результатов с использованием Plotly
        """
        # Создание субплотов
        for symbol, df in data_1m.items():
            # Синхронизация с 5-минутным таймфреймом
            df = df.copy()
            df_5m = data_5m[symbol].copy()
            self.__map_swing_points_to_original(df, df_5m, "5min")

            market_structure(df)

            fig = make_subplots(
                shared_xaxes=True,
                vertical_spacing=0.005,
                horizontal_spacing=0.005,
                row_width=[1]
            )

            fig.add_trace(
                go.Candlestick(
                    name=symbol,
                    x=df.index,
                    open=df['open'],
                    high=df['high'],
                    low=df['low'],
                    close=df['close']
                )
            )

            # Находим свинг-точки для динамического графика
            df_highs = df.index[df['swing_point'] == 1]
            df_lows = df.index[df['swing_point'] == -1]

            # Добавляем точки максимумов (HH) для динамического графика
            fig.add_trace(go.Scatter(
                x=df_highs,
                y=df.loc[df_highs, 'high'],
                mode='markers+text',
                text=[f"{retr:.1f}%" if not pd.isna(retr) else "" for retr in df.loc[df_highs, 'swing_retracement']],
                textposition="top center",
                marker=dict(
                    color=SWING_COLOR,
                    size=6,
                ),
                name='HH'
            ))

            # Добавляем точки минимумов (LL) для динамического графика
            fig.add_trace(go.Scatter(
                x=df_lows,
                y=df.loc[df_lows, 'low'],
                mode='markers+text',
                text=[f"{retr:.1f}%" if not pd.isna(retr) else "" for retr in df.loc[df_lows, 'swing_retracement']],
                textposition="bottom center",
                marker=dict(
                    color=SWING_COLOR,
                    size=6,
                ),
                name='LL'
            ))

            # Добавление сделок на OHLC график
            if self.trades is not None:
                if symbol in self.trades:
                    trades = self.trades[symbol]
                    buy_trades = trades[trades['side'] == Side.LONG]
                    sell_trades = trades[trades['side'] == Side.SHORT]

                    fig.add_trace(
                        go.Scatter(
                            x=buy_trades['datetime'],
                            y=buy_trades['price'],
                            mode='markers',
                            marker=dict(color='green', size=10, symbol='triangle-up'),
                            name='Buy Trades',
                            showlegend=False
                        )
                    )

                    fig.add_trace(
                        go.Scatter(
                            x=sell_trades['datetime'],
                            y=sell_trades['price'],
                            mode='markers',
                            marker=dict(color='red', size=10, symbol='triangle-down'),
                            name='Sell Trades',
                            showlegend=False
                        )
                    )

            for idx in df.index:
                # Пропускаем строки где нет BOS, CHoCH или Confirm
                if pd.isna(df.loc[idx, "bos"]) and pd.isna(df.loc[idx, "choch"]) and pd.isna(df.loc[idx, "confirm"]):
                    continue

                # Получаем общие данные для строки
                if pd.isna(df.loc[idx, "struct_start_idx"]) or pd.isna(df.loc[idx, "struct_broken_idx"]) or pd.isna(
                        df.loc[idx, "struct_level"]):
                    continue

                # Получаем даты напрямую из индекса DataFrame
                start_idx = df.loc[idx, "struct_start_idx"]
                broken_idx = df.loc[idx, "struct_broken_idx"]
                level_value: float = df.loc[idx, "struct_level"]
                is_bullish: bool = df.loc[idx, "struct_direction"] == 1

                # Используем индексы напрямую для построения графика
                x_start = start_idx
                x_end = broken_idx
                x_mid = x_start + (x_end - x_start) / 2

                # Обработка BOS
                if not pd.isna(df.loc[idx, "bos"]):
                    bos_color: str = BULLISH_COLOR if is_bullish else BEARISH_COLOR
                    text_offset: float = level_value * 0.0001  # 0.05% от значения уровня

                    # Горизонтальная линия
                    fig.add_trace(
                        go.Scatter(
                            x=[x_start, x_end],
                            y=[level_value, level_value],
                            mode="lines",
                            line=dict(color=bos_color, width=1),
                            showlegend=False
                        )
                    )

                    # Текст с отступом
                    fig.add_trace(
                        go.Scatter(
                            x=[x_mid],
                            y=[level_value + text_offset if is_bullish else level_value - text_offset],
                            mode="text",
                            text="BOS",
                            textposition="top center" if is_bullish else "bottom center",
                            textfont=dict(color=bos_color, size=8),
                            showlegend=False
                        )
                    )

                # Обработка CHoCH
                if not pd.isna(df.loc[idx, "choch"]):
                    choch_color: str = BULLISH_COLOR if is_bullish else BEARISH_COLOR
                    text_offset: float = level_value * 0.0001  # 0.05% от значения уровня

                    # Горизонтальная линия
                    fig.add_trace(
                        go.Scatter(
                            x=[x_start, x_end],
                            y=[level_value, level_value],
                            mode="lines",
                            line=dict(color=choch_color, width=2),
                            showlegend=False
                        )
                    )

                    # Текст с отступом
                    fig.add_trace(
                        go.Scatter(
                            x=[x_mid],
                            y=[level_value + text_offset if is_bullish else level_value - text_offset],
                            mode="text",
                            text="CHoCH",
                            textposition="top center" if is_bullish else "bottom center",
                            textfont=dict(color=choch_color, size=8),
                            showlegend=False
                        )
                    )

                # Обработка Confirm
                if not pd.isna(df.loc[idx, "confirm"]):
                    confirm_color: str = BULLISH_COLOR if is_bullish else BEARISH_COLOR
                    text_offset: float = level_value * 0.0001  # 0.05% от значения уровня

                    # Горизонтальная линия - используем пунктирную линию для отличия от BOS и CHoCH
                    fig.add_trace(
                        go.Scatter(
                            x=[x_start, x_end],
                            y=[level_value, level_value],
                            mode="lines",
                            line=dict(color=confirm_color, width=2),
                            showlegend=False
                        )
                    )

                    # Текст с отступом
                    fig.add_trace(
                        go.Scatter(
                            x=[x_mid],
                            y=[level_value + text_offset if is_bullish else level_value - text_offset],
                            mode="text",
                            text="Confirm",
                            textposition="top center" if is_bullish else "bottom center",
                            textfont=dict(color=confirm_color, size=8),
                            showlegend=False
                        )
                    )

                # Убеждаемся, что столбец datetime существует
            if 'datetime' not in df.columns:
                df['datetime'] = [dt.datetime.fromtimestamp(int(t) / 1000) for t in df['timestamp']]

                # Находим диапазоны с непрерывными значениями тренда
            trend_changes = df['trend'].ne(df['trend'].shift()).cumsum()
            trend_groups = df.groupby([trend_changes, 'trend'])

            # Перебираем каждую группу тренда
            for (_, trend_value), group in trend_groups:
                # Пропускаем группы без тренда
                if pd.isna(trend_value):
                    continue

                # Получаем начало и конец интервала тренда
                start_idx = group.index[0]
                end_idx = group.index[-1]

                # Получаем диапазон цен для прямоугольника
                y_min = group['low'].min() * 0.999  # Добавляем небольшой отступ
                y_max = group['high'].max() * 1.001  # Добавляем небольшой отступ

                # Устанавливаем цвет в зависимости от направления тренда
                color = BULLISH_COLOR if trend_value == 1 else BEARISH_COLOR

                # Добавляем прямоугольник для интервала тренда
                fig.add_shape(
                    type="rect",
                    x0=start_idx,
                    x1=end_idx,
                    y0=y_min,
                    y1=y_max,
                    fillcolor=color,
                    opacity=0.1,
                    layer="below",
                    line=dict(width=0)
                )

            fig.update_xaxes(rangeslider={'visible': False})
            # show
            fig.update_layout(
                xaxis_rangeslider_visible=False, title=symbol,
                showlegend=False,
                yaxis=dict(
                    tickformat=' ',
                    separatethousands=True
                )
            )
            fig.show()