Разработка системы сравнения стратегий

Разработка системы сравнения стратегий

Система сравнения стратегий позволяет объективно оценить несколько торговых стратегий по единому набору метрик, выявить их сильные и слабые стороны, выбрать лучшую или оптимальную комбинацию для портфеля.

Стандартизированный формат результатов

from dataclasses import dataclass
import pandas as pd
import numpy as np

@dataclass
class StrategyResult:
    name: str
    params: dict
    equity_curve: pd.Series
    trades: pd.DataFrame

    # Вычисленные метрики
    sharpe_ratio: float
    sortino_ratio: float
    calmar_ratio: float
    annual_return_pct: float
    max_drawdown_pct: float
    win_rate: float
    profit_factor: float
    total_trades: int
    avg_trade_duration_hours: float
    total_commission_pct: float

Comparison Engine

class StrategyComparator:
    def __init__(self, backtester, benchmark_data: pd.Series = None):
        self.backtester = backtester
        self.benchmark = benchmark_data  # Buy & Hold BTC для сравнения

    def compare(self, strategies: list[dict], data: pd.DataFrame) -> ComparisonReport:
        results = []

        for strategy_config in strategies:
            result = self.backtester.run(
                strategy_class=strategy_config['class'],
                params=strategy_config['params'],
                data=data,
                name=strategy_config['name'],
            )
            results.append(result)

        # Добавляем Buy & Hold как benchmark
        if self.benchmark is not None:
            bh_return = (self.benchmark.iloc[-1] / self.benchmark.iloc[0] - 1)
            results.append(self._create_buyhold_result(self.benchmark))

        return self.build_report(results)

    def build_report(self, results: list[StrategyResult]) -> ComparisonReport:
        comparison_df = pd.DataFrame([{
            'Strategy': r.name,
            'Annual Return %': round(r.annual_return_pct, 2),
            'Sharpe Ratio': round(r.sharpe_ratio, 3),
            'Sortino Ratio': round(r.sortino_ratio, 3),
            'Max Drawdown %': round(r.max_drawdown_pct, 2),
            'Calmar Ratio': round(r.calmar_ratio, 3),
            'Win Rate %': round(r.win_rate * 100, 1),
            'Profit Factor': round(r.profit_factor, 2),
            'Total Trades': r.total_trades,
            'Avg Trade Hours': round(r.avg_trade_duration_hours, 1),
            'Commission Drag %': round(r.total_commission_pct, 2),
        } for r in results])

        # Ранжирование по каждой метрике
        rankings = self._compute_rankings(comparison_df)

        # Корреляция между стратегиями
        correlations = self._compute_correlations(results)

        return ComparisonReport(
            summary=comparison_df,
            rankings=rankings,
            correlations=correlations,
            strategies=results,
        )

    def _compute_rankings(self, df: pd.DataFrame) -> pd.DataFrame:
        """Ранжируем стратегии по каждой метрике"""
        rankings = pd.DataFrame({'Strategy': df['Strategy']})

        for metric, ascending in [
            ('Annual Return %', False),
            ('Sharpe Ratio', False),
            ('Max Drawdown %', True),  # меньше = лучше
            ('Profit Factor', False),
            ('Win Rate %', False),
        ]:
            if metric in df.columns:
                rankings[f'Rank: {metric}'] = df[metric].rank(ascending=ascending).astype(int)

        # Общий rank (среднее по всем метрикам)
        rank_cols = [c for c in rankings.columns if c.startswith('Rank:')]
        rankings['Overall Rank'] = rankings[rank_cols].mean(axis=1).rank().astype(int)

        return rankings.sort_values('Overall Rank')

    def _compute_correlations(self, results: list[StrategyResult]) -> pd.DataFrame:
        """Корреляция доходностей между стратегиями"""
        returns_dict = {
            r.name: r.equity_curve.pct_change().dropna()
            for r in results
        }
        returns_df = pd.DataFrame(returns_dict).dropna()
        return returns_df.corr()

Визуализация сравнения

def plot_comparison(report: ComparisonReport):
    import plotly.graph_objects as go
    from plotly.subplots import make_subplots

    fig = make_subplots(
        rows=2, cols=2,
        subplot_titles=[
            'Equity Curves',
            'Risk-Return Scatter',
            'Monthly Returns Distribution',
            'Drawdown Comparison',
        ]
    )

    colors = ['#00C853', '#2196F3', '#FF9800', '#E91E63', '#9C27B0']

    for i, strat in enumerate(report.strategies):
        color = colors[i % len(colors)]

        # Equity curves
        fig.add_trace(go.Scatter(
            x=strat.equity_curve.index,
            y=strat.equity_curve / strat.equity_curve.iloc[0] * 100,  # нормализованы к 100
            name=strat.name,
            line=dict(color=color),
        ), row=1, col=1)

        # Risk-Return scatter
        fig.add_trace(go.Scatter(
            x=[abs(strat.max_drawdown_pct)],
            y=[strat.annual_return_pct],
            mode='markers+text',
            marker=dict(size=12, color=color),
            text=[strat.name],
            textposition='top center',
            showlegend=False,
        ), row=1, col=2)

        # Drawdown
        rolling_max = strat.equity_curve.cummax()
        drawdown = (strat.equity_curve - rolling_max) / rolling_max * 100
        fig.add_trace(go.Scatter(
            x=drawdown.index,
            y=drawdown,
            fill='tozeroy',
            name=strat.name,
            line=dict(color=color),
            showlegend=False,
            opacity=0.6,
        ), row=2, col=2)

    fig.update_layout(
        title='Strategy Comparison Report',
        height=900,
        template='plotly_dark',
    )
    return fig

Portfolio Combination Analysis

Низкая корреляция между стратегиями — возможность диверсификации:

def analyze_portfolio_combination(
    strategy_returns: dict[str, pd.Series],
    target_sharpe: float = 2.0,
) -> dict:
    """Находим оптимальные веса для портфеля из нескольких стратегий"""
    from scipy.optimize import minimize

    returns_df = pd.DataFrame(strategy_returns).dropna()
    mean_returns = returns_df.mean()
    cov_matrix = returns_df.cov()

    def neg_sharpe(weights):
        portfolio_return = np.dot(weights, mean_returns) * 252
        portfolio_vol = np.sqrt(np.dot(weights.T, np.dot(cov_matrix * 252, weights)))
        return -portfolio_return / portfolio_vol if portfolio_vol > 0 else 999

    n = len(strategy_returns)
    constraints = [{'type': 'eq', 'fun': lambda w: np.sum(w) - 1}]
    bounds = [(0, 1)] * n
    x0 = [1/n] * n

    result = minimize(neg_sharpe, x0, method='SLSQP', bounds=bounds, constraints=constraints)

    optimal_weights = dict(zip(strategy_returns.keys(), result.x))
    combined_returns = sum(ret * w for ret, w in zip(returns_df.values.T, result.x))
    combined_sharpe = -result.fun

    return {
        'optimal_weights': optimal_weights,
        'combined_sharpe': combined_sharpe,
        'improvement_vs_best_single': combined_sharpe - max(
            ret.mean() / ret.std() * np.sqrt(252) for ret in strategy_returns.values()
        ),
    }

Комбинация некоррелированных стратегий часто даёт лучший risk-adjusted результат, чем любая отдельная стратегия. Это работает как диверсификация в классических финансах, только на уровне торговых алгоритмов.