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

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

Генетический алгоритм (GA) — эволюционный метод оптимизации, вдохновлённый биологической эволюцией. Вместо перебора всех комбинаций (grid search) или случайного поиска, GA эволюционирует популяцию решений через механизмы отбора, скрещивания и мутации. Эффективен для высокоразмерных пространств параметров.

Принцип работы

1. Инициализация — случайная популяция из N наборов параметров (особей)
2. Оценка (Fitness) — каждая особь оценивается через бэктест
3. Отбор — лучшие особи отбираются для размножения (tournament selection, roulette wheel)
4. Скрещивание (Crossover) — параметры двух родителей комбинируются
5. Мутация — случайные изменения параметров для исследования пространства
6. Повтор — итерации до сходимости или достижения максимума поколений

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

from deap import base, creator, tools, algorithms
import random
import numpy as np
from functools import partial

# Определяем задачу максимизации Sharpe ratio
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)

class GeneticOptimizer:
    def __init__(
        self,
        param_bounds: dict[str, tuple],  # {'param': (min, max)}
        backtest_fn: callable,
        population_size: int = 50,
        n_generations: int = 30,
        crossover_prob: float = 0.7,
        mutation_prob: float = 0.2,
        n_jobs: int = 4,
    ):
        self.param_names = list(param_bounds.keys())
        self.param_bounds = list(param_bounds.values())
        self.backtest_fn = backtest_fn
        self.pop_size = population_size
        self.n_gen = n_generations
        self.cx_prob = crossover_prob
        self.mut_prob = mutation_prob
        self.n_jobs = n_jobs

    def decode_individual(self, individual: list) -> dict:
        """Конвертируем список значений [0,1] в реальные параметры"""
        params = {}
        for i, name in enumerate(self.param_names):
            low, high = self.param_bounds[i]
            if isinstance(low, int) and isinstance(high, int):
                # Целочисленный параметр
                params[name] = int(round(low + individual[i] * (high - low)))
            else:
                # Вещественный параметр
                params[name] = low + individual[i] * (high - low)
        return params

    def evaluate(self, individual: list) -> tuple:
        """Функция fitness: запускаем бэктест, возвращаем Sharpe ratio"""
        params = self.decode_individual(individual)
        try:
            metrics = self.backtest_fn(params)
            sharpe = metrics.get('sharpe_ratio', 0)
            # Штраф за слишком мало сделок
            trades = metrics.get('total_trades', 0)
            if trades < 20:
                sharpe *= trades / 20
            return (sharpe,)
        except Exception:
            return (-999.0,)

    def run(self) -> tuple[dict, pd.DataFrame]:
        toolbox = base.Toolbox()

        # Генератор особей: каждый параметр = float в [0, 1]
        toolbox.register("attr_float", random.random)
        toolbox.register(
            "individual",
            tools.initRepeat,
            creator.Individual,
            toolbox.attr_float,
            n=len(self.param_names),
        )
        toolbox.register("population", tools.initRepeat, list, toolbox.individual)
        toolbox.register("evaluate", self.evaluate)
        toolbox.register("mate", tools.cxBlend, alpha=0.3)  # Blend crossover
        toolbox.register("mutate", tools.mutGaussian, mu=0, sigma=0.1, indpb=0.2)
        toolbox.register("select", tools.selTournament, tournsize=3)

        # Ограничиваем значения в [0, 1] после мутации
        def check_bounds(individual):
            for i in range(len(individual)):
                individual[i] = max(0.0, min(1.0, individual[i]))
            return individual,

        toolbox.decorate("mutate", check_bounds)
        toolbox.decorate("mate", check_bounds)

        # Параллельная оценка
        if self.n_jobs > 1:
            from multiprocessing.pool import Pool
            pool = Pool(self.n_jobs)
            toolbox.register("map", pool.map)

        # Запуск эволюции
        population = toolbox.population(n=self.pop_size)
        stats = tools.Statistics(lambda ind: ind.fitness.values[0])
        stats.register("max", np.max)
        stats.register("avg", np.mean)
        hof = tools.HallOfFame(10)  # Топ-10 лучших особей

        population, logbook = algorithms.eaSimple(
            population,
            toolbox,
            cxpb=self.cx_prob,
            mutpb=self.mut_prob,
            ngen=self.n_gen,
            stats=stats,
            halloffame=hof,
            verbose=True,
        )

        if self.n_jobs > 1:
            pool.close()

        # Результаты
        best_params = self.decode_individual(hof[0])
        
        all_results = []
        for ind in hof:
            params = self.decode_individual(ind)
            all_results.append({**params, 'sharpe': ind.fitness.values[0]})

        return best_params, pd.DataFrame(all_results)

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

optimizer = GeneticOptimizer(
    param_bounds={
        'fast_period': (5, 30),
        'slow_period': (15, 100),
        'rsi_period': (7, 21),
        'rsi_oversold': (20, 40),
        'rsi_overbought': (60, 85),
        'stop_loss_pct': (0.01, 0.10),
        'take_profit_pct': (0.02, 0.20),
    },
    backtest_fn=lambda params: run_backtest(params, train_data),
    population_size=60,
    n_generations=40,
    n_jobs=8,
)

best_params, top_results = optimizer.run()
print("Best parameters:", best_params)
print("\nTop 10 results:")
print(top_results)

Convergence мониторинг

def plot_evolution(logbook: tools.Logbook):
    """Визуализируем сходимость эволюции"""
    gen = logbook.select("gen")
    avg_fitness = logbook.select("avg")
    max_fitness = logbook.select("max")

    plt.figure(figsize=(10, 5))
    plt.plot(gen, max_fitness, 'b-', label='Best Sharpe')
    plt.plot(gen, avg_fitness, 'r--', label='Avg Sharpe')
    plt.xlabel('Generation')
    plt.ylabel('Sharpe Ratio')
    plt.title('GA Optimization Convergence')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.savefig('ga_convergence.png')

Сравнение с Grid Search

GA требует меньше вычислений, но его результаты нужно обязательно проверять на out-of-sample данных — эволюционный поиск хорошо находит локальные оптимумы, которые могут быть переобученными.