Разработка системы оптимизации параметров стратегии (Bayesian optimization)

Разработка системы оптимизации параметров стратегии (Bayesian optimization)

Bayesian optimization — умный поиск оптимума: вместо случайного перебора строит суррогатную модель зависимости метрики от параметров и выбирает следующую точку для оценки там, где ожидается наибольшее улучшение. Находит хорошие параметры за 50–200 итераций там, где grid search требует тысяч.

Принцип Bayesian Optimization

Суррогатная модель (Gaussian Process) — аппроксимирует функцию bэктеста. После каждой оценки обновляет своё представление о том, как параметры влияют на результат.

Acquisition Function — определяет где искать дальше. Expected Improvement (EI) — стандартный выбор: ищем точки, где ожидаемое улучшение максимально.

Exploration vs Exploitation — баланс между исследованием неизведанных областей и уточнением вокруг известных хороших точек.

Реализация с Optuna

import optuna
from optuna.samplers import TPESampler
import pandas as pd

optuna.logging.set_verbosity(optuna.logging.WARNING)

class BayesianOptimizer:
    def __init__(
        self,
        backtest_fn: callable,
        n_trials: int = 100,
        n_startup_trials: int = 10,  # случайные начальные пробы
        n_jobs: int = 1,
        metric: str = 'sharpe_ratio',
        direction: str = 'maximize',
    ):
        self.backtest_fn = backtest_fn
        self.n_trials = n_trials
        self.n_startup = n_startup_trials
        self.n_jobs = n_jobs
        self.metric = metric
        self.direction = direction
        self.results_log = []

    def create_objective(self, param_space: dict):
        def objective(trial: optuna.Trial) -> float:
            params = {}
            for name, spec in param_space.items():
                if spec['type'] == 'int':
                    params[name] = trial.suggest_int(name, spec['low'], spec['high'])
                elif spec['type'] == 'float':
                    params[name] = trial.suggest_float(name, spec['low'], spec['high'])
                elif spec['type'] == 'categorical':
                    params[name] = trial.suggest_categorical(name, spec['choices'])
                elif spec['type'] == 'log':
                    params[name] = trial.suggest_float(name, spec['low'], spec['high'], log=True)

            try:
                metrics = self.backtest_fn(params)
                value = metrics.get(self.metric, float('-inf'))

                # Штраф за слишком мало сделок
                n_trades = metrics.get('total_trades', 0)
                if n_trades < 15:
                    value = value * n_trades / 15

                # Штраф за экстремальный drawdown
                max_dd = abs(metrics.get('max_drawdown_pct', 0))
                if max_dd > 40:
                    value = value * (40 / max_dd) ** 2

                self.results_log.append({**params, self.metric: value, 'total_trades': n_trades})
                return value
            except Exception as e:
                return float('-inf')

        return objective

    def run(self, param_space: dict) -> tuple[dict, pd.DataFrame]:
        sampler = TPESampler(
            n_startup_trials=self.n_startup,
            seed=42,
        )

        study = optuna.create_study(
            direction=self.direction,
            sampler=sampler,
        )

        study.optimize(
            self.create_objective(param_space),
            n_trials=self.n_trials,
            n_jobs=self.n_jobs,
            show_progress_bar=True,
        )

        best_params = study.best_params
        results_df = pd.DataFrame(self.results_log).sort_values(self.metric, ascending=False)

        return best_params, results_df, study

Использование

optimizer = BayesianOptimizer(
    backtest_fn=lambda params: run_backtest(params, train_data),
    n_trials=150,
    n_startup_trials=15,
    n_jobs=4,
)

param_space = {
    'fast_period': {'type': 'int', 'low': 5, 'high': 30},
    'slow_period': {'type': 'int', 'low': 15, 'high': 100},
    'rsi_period': {'type': 'int', 'low': 7, 'high': 21},
    'rsi_oversold': {'type': 'int', 'low': 20, 'high': 40},
    'stop_loss_pct': {'type': 'float', 'low': 0.01, 'high': 0.10},
    'take_profit_pct': {'type': 'float', 'low': 0.02, 'high': 0.25},
    'commission': {'type': 'categorical', 'choices': ['market', 'limit']},
}

best_params, results, study = optimizer.run(param_space)
print("Best parameters:", best_params)
print(f"Best {optimizer.metric}: {study.best_value:.3f}")

Анализ результатов

import optuna.visualization as vis

# Важность параметров
fig = vis.plot_param_importances(study)
fig.show()

# Contour plot двух параметров
fig = vis.plot_contour(study, params=['fast_period', 'slow_period'])
fig.show()

# История оптимизации
fig = vis.plot_optimization_history(study)
fig.show()

Защита от overfitting: Cross-Validation

def time_series_cv_objective(params: dict, data: pd.DataFrame, n_splits: int = 5) -> dict:
    """K-fold кросс-валидация для временных рядов"""
    fold_size = len(data) // (n_splits + 1)
    sharpe_scores = []

    for fold in range(n_splits):
        train_start = 0
        train_end = (fold + 1) * fold_size
        test_start = train_end
        test_end = train_end + fold_size

        train_data = data.iloc[train_start:train_end]
        test_data = data.iloc[test_start:test_end]

        # Оптимизируем на train, оцениваем на test
        metrics = run_backtest(params, test_data)
        sharpe_scores.append(metrics.get('sharpe_ratio', 0))

    return {
        'sharpe_ratio': np.mean(sharpe_scores),
        'sharpe_std': np.std(sharpe_scores),
        'min_sharpe': min(sharpe_scores),  # худший период
    }

Сравнение методов оптимизации

Для большинства задач оптимизации торговых стратегий Bayesian optimization через Optuna — оптимальный выбор: быстро находит хорошие параметры, поддерживает parallelism, имеет отличную визуализацию.