Разработка системы бэктестинга портфельных стратегий

Разработка системы бэктестинга портфельных стратегий

Портфельный бэктестинг отличается от одноинструментального: вместо анализа одной пары отслеживается распределение капитала между множеством активов, корреляции между позициями и совокупный риск портфеля.

Ключевые отличия от single-asset бэктеста

Capital allocation — как распределяется капитал между сигналами? Equal weight, risk-parity, momentum-weighted? Решение принимается динамически каждый ребалансировочный период.

Correlated positions — BTC и ETH движутся вместе. Держать обе позиции — не диверсификация. Нужно учитывать корреляцию при расчёте реального риска.

Transaction costs — при ребалансировке портфеля из 20 активов каждые 2 недели комиссии существенно съедают прибыль.

Slippage при масштабировании — крупный портфель не может войти в малоликвидный альткоин по рыночной цене без значительного impact.

Структура данных портфеля

from dataclasses import dataclass, field
from typing import Optional
import numpy as np
import pandas as pd

@dataclass
class PortfolioPosition:
    symbol: str
    quantity: float
    avg_entry_price: float
    entry_time: int
    unrealized_pnl: float = 0.0
    realized_pnl: float = 0.0

@dataclass
class PortfolioState:
    cash: float
    positions: dict[str, PortfolioPosition] = field(default_factory=dict)
    
    @property
    def total_value(self) -> float:
        return self.cash + sum(
            p.quantity * p.avg_entry_price * (1 + p.unrealized_pnl / (p.quantity * p.avg_entry_price))
            for p in self.positions.values()
        )

    def get_weights(self, current_prices: dict[str, float]) -> dict[str, float]:
        total = self.cash + sum(
            p.quantity * current_prices.get(p.symbol, p.avg_entry_price)
            for p in self.positions.values()
        )
        weights = {}
        for symbol, pos in self.positions.items():
            price = current_prices.get(symbol, pos.avg_entry_price)
            weights[symbol] = (pos.quantity * price) / total
        weights['cash'] = self.cash / total
        return weights

Capital Allocation стратегии

class AllocationEngine:
    def equal_weight(self, signals: list[str], capital: float) -> dict[str, float]:
        """Равное распределение капитала"""
        if not signals:
            return {}
        weight = capital / len(signals)
        return {symbol: weight for symbol in signals}

    def risk_parity(
        self,
        signals: list[str],
        capital: float,
        volatilities: dict[str, float],
    ) -> dict[str, float]:
        """Распределение обратно пропорционально волатильности"""
        if not signals:
            return {}
        
        inv_vols = {s: 1.0 / volatilities.get(s, 0.01) for s in signals}
        total_inv_vol = sum(inv_vols.values())
        
        return {
            symbol: capital * inv_vol / total_inv_vol
            for symbol, inv_vol in inv_vols.items()
        }

    def momentum_weighted(
        self,
        signals: list[str],
        capital: float,
        returns_12m: dict[str, float],
    ) -> dict[str, float]:
        """Больший вес — активам с лучшей 12-месячной доходностью"""
        positive_returns = {s: r for s, r in returns_12m.items() if s in signals and r > 0}
        if not positive_returns:
            return self.equal_weight(signals, capital)
        
        total_return = sum(positive_returns.values())
        return {
            symbol: capital * ret / total_return
            for symbol, ret in positive_returns.items()
        }

Ребалансировка

class RebalancingEngine:
    def __init__(self, commission_pct: float = 0.001, min_trade_usd: float = 10.0):
        self.commission_pct = commission_pct
        self.min_trade_usd = min_trade_usd

    def calculate_rebalancing_trades(
        self,
        current_state: PortfolioState,
        target_weights: dict[str, float],
        current_prices: dict[str, float],
    ) -> list[dict]:
        total_value = sum(
            pos.quantity * current_prices[sym]
            for sym, pos in current_state.positions.items()
        ) + current_state.cash

        trades = []
        for symbol, target_weight in target_weights.items():
            target_value = total_value * target_weight
            current_value = (
                current_state.positions[symbol].quantity * current_prices[symbol]
                if symbol in current_state.positions else 0.0
            )

            delta = target_value - current_value
            
            if abs(delta) < self.min_trade_usd:
                continue  # Слишком маленькая сделка, пропускаем

            trades.append({
                'symbol': symbol,
                'side': 'BUY' if delta > 0 else 'SELL',
                'usd_amount': abs(delta),
                'quantity': abs(delta) / current_prices[symbol],
                'price': current_prices[symbol],
                'commission': abs(delta) * self.commission_pct,
            })

        return trades

Портфельные метрики

def calculate_portfolio_metrics(equity_curve: pd.Series, trades_df: pd.DataFrame) -> dict:
    returns = equity_curve.pct_change().dropna()
    
    # Correlation-adjusted risk
    # Если у нас есть позиции в BTC и ETH одновременно,
    # реальный риск выше чем сумма индивидуальных рисков

    annualized_return = returns.mean() * 252
    annualized_vol = returns.std() * np.sqrt(252)
    sharpe = annualized_return / annualized_vol if annualized_vol > 0 else 0

    # Portfolio-specific: turnover
    total_traded_value = trades_df['usd_amount'].sum()
    avg_portfolio_value = equity_curve.mean()
    annual_turnover = (total_traded_value / avg_portfolio_value) * (252 / len(equity_curve))

    # Transaction cost drag
    total_commissions = trades_df['commission'].sum()
    commission_drag = total_commissions / equity_curve.iloc[0]

    return {
        'annualized_return': annualized_return,
        'annualized_volatility': annualized_vol,
        'sharpe_ratio': sharpe,
        'max_drawdown': calculate_max_drawdown(equity_curve),
        'annual_turnover': annual_turnover,
        'total_commission': total_commissions,
        'commission_drag_pct': commission_drag * 100,
        'avg_positions': trades_df.groupby('timestamp')['symbol'].count().mean(),
    }

Survivorship bias в портфельных тестах

Если тестируете стратегию на портфеле из текущего состава индекса (например, топ-20 крипто по капитализации), вы тестируете на выживших: активы, которые были в топ-20 год назад и вылетели оттуда, не включены. Это систематически завышает результаты.

Правильный подход: использовать point-in-time universe — какие активы были в universe на каждую конкретную дату, не какие в нём сейчас.