mahmoud20138 Strategy Genetic Optimizer
Overview
Uses evolutionary algorithms (GA) to search the parameter space of trading strategies. Breeds top performers, mutates for exploration, and applies selection pressure via risk-adjusted fitness functions. Prevents overfitting through walk-forward validation and population diversity enforcement.
1. Gene Encoding & Strategy Genome
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Callable, Optional
import random, copy
@dataclass
class Gene:
"""Single parameter with its valid range."""
name: str
min_val: float
max_val: float
step: float = 1.0
gene_type: str = "float" # float, int, bool, choice
choices: list = field(default_factory=list)
def random_value(self):
if self.gene_type == "bool": return random.choice([True, False])
if self.gene_type == "choice": return random.choice(self.choices)
if self.gene_type == "int": return random.randint(int(self.min_val), int(self.max_val))
val = random.uniform(self.min_val, self.max_val)
return round(val / self.step) * self.step
def mutate(self, value, mutation_strength: float = 0.2):
if self.gene_type == "bool": return not value
if self.gene_type == "choice": return random.choice(self.choices)
range_size = self.max_val - self.min_val
delta = random.gauss(0, range_size * mutation_strength)
new_val = np.clip(value + delta, self.min_val, self.max_val)
if self.gene_type == "int": return int(round(new_val))
return round(new_val / self.step) * self.step
@dataclass
class Genome:
"""Complete strategy parameter set."""
genes: dict # {gene_name: value}
fitness: float = 0.0
generation: int = 0
id: str = ""
# Example: MA crossover strategy genome definition
MA_CROSSOVER_GENES = [
Gene("fast_period", 5, 50, 1, "int"),
Gene("slow_period", 20, 200, 1, "int"),
Gene("rsi_filter", 0, 100, 1, "int"),
Gene("atr_stop_mult", 1.0, 5.0, 0.1, "float"),
Gene("atr_tp_mult", 1.0, 8.0, 0.1, "float"),
Gene("use_volume_filter", 0, 1, 1, "bool"),
Gene("entry_type", 0, 0, 0, "choice", choices=["market", "limit_pullback", "stop_entry"])]
2. Fitness Functions (Risk-Adjusted)
def fitness_sharpe_dd(returns: pd.Series, max_dd_threshold: float = -0.20) -> float:
"""Sharpe ratio penalized by drawdown. Primary fitness function."""
if len(returns) < 30 or returns.std() == 0: return -999
sharpe = (returns.mean() / returns.std()) * np.sqrt(252)
equity = (1 + returns).cumprod()
dd = (equity / equity.cummax() - 1).min()
if dd < max_dd_threshold: return sharpe + (dd - max_dd_threshold) * 10 # Heavy penalty
return sharpe
def fitness_expectancy(trades: pd.DataFrame) -> float:
"""Expectancy * frequency. Rewards consistent edges."""
if trades.empty: return -999
wins = trades[trades["pnl"] > 0]
losses = trades[trades["pnl"] <= 0]
wr = len(wins) / len(trades)
avg_w = wins["pnl"].mean() if len(wins) > 0 else 0
avg_l = abs(losses["pnl"].mean()) if len(losses) > 0 else 1
expectancy = wr * avg_w - (1 - wr) * avg_l
frequency = len(trades) / 252 # trades per year
return expectancy * np.sqrt(frequency)
def fitness_sortino_calmar(returns: pd.Series) -> float:
"""Combined Sortino + Calmar for downside-focused optimization."""
if len(returns) < 30: return -999
downside = returns[returns < 0].std() * np.sqrt(252)
sortino = returns.mean() * 252 / max(downside, 1e-10)
equity = (1 + returns).cumprod()
max_dd = abs((equity / equity.cummax() - 1).min())
calmar = returns.mean() * 252 / max(max_dd, 1e-10)
return (sortino + calmar) / 2
3. Genetic Algorithm Engine
class GeneticOptimizer:
"""Core evolutionary optimization engine."""
def __init__(self, gene_defs: list[Gene], fitness_fn: Callable,
population_size: int = 50, elite_pct: float = 0.1,
mutation_rate: float = 0.15, crossover_rate: float = 0.7):
self.gene_defs = {g.name: g for g in gene_defs}
self.fitness_fn = fitness_fn
self.pop_size = population_size
self.elite_pct = elite_pct
self.mutation_rate = mutation_rate
self.crossover_rate = crossover_rate
self.population = []
self.history = []
def initialize_population(self) -> list[Genome]:
self.population = []
for i in range(self.pop_size):
genes = {name: gene.random_value() for name, gene in self.gene_defs.items()}
self.population.append(Genome(genes=genes, id=f"gen0_{i}"))
return self.population
def evaluate(self, strategy_runner: Callable, data: pd.DataFrame):
"""Evaluate all genomes. strategy_runner(data, params) -> returns Series."""
for genome in self.population:
try:
returns = strategy_runner(data, genome.genes)
genome.fitness = self.fitness_fn(returns)
except Exception:
genome.fitness = -999
self.population.sort(key=lambda g: g.fitness, reverse=True)
def select_parents(self) -> tuple[Genome, Genome]:
"""Tournament selection."""
def tournament(k=3):
contestants = random.sample(self.population, min(k, len(self.population)))
return max(contestants, key=lambda g: g.fitness)
return tournament(), tournament()
def crossover(self, parent_a: Genome, parent_b: Genome) -> Genome:
"""Uniform crossover — each gene randomly from either parent."""
child_genes = {}
for name in self.gene_defs:
child_genes[name] = parent_a.genes[name] if random.random() < 0.5 else parent_b.genes[name]
return Genome(genes=child_genes)
def mutate(self, genome: Genome, strength: float = 0.2) -> Genome:
mutated = copy.deepcopy(genome)
for name, gene_def in self.gene_defs.items():
if random.random() < self.mutation_rate:
mutated.genes[name] = gene_def.mutate(mutated.genes[name], strength)
return mutated
def evolve_generation(self, strategy_runner: Callable, data: pd.DataFrame, gen_num: int) -> dict:
"""One full generation: evaluate → select → breed → mutate."""
self.evaluate(strategy_runner, data)
n_elite = max(int(self.pop_size * self.elite_pct), 1)
elites = [copy.deepcopy(g) for g in self.population[:n_elite]]
new_pop = list(elites)
while len(new_pop) < self.pop_size:
p1, p2 = self.select_parents()
child = self.crossover(p1, p2) if random.random() < self.crossover_rate else copy.deepcopy(p1)
child = self.mutate(child)
child.generation = gen_num
child.id = f"gen{gen_num}_{len(new_pop)}"
new_pop.append(child)
self.population = new_pop
best = self.population[0]
gen_stats = {
"generation": gen_num, "best_fitness": round(best.fitness, 4),
"best_params": best.genes, "avg_fitness": round(np.mean([g.fitness for g in self.population]), 4),
"diversity": self._population_diversity(),
}
self.history.append(gen_stats)
return gen_stats
def run(self, strategy_runner: Callable, data: pd.DataFrame, n_generations: int = 30) -> dict:
"""Full optimization run."""
self.initialize_population()
for gen in range(n_generations):
stats = self.evolve_generation(strategy_runner, data, gen)
print(f"Gen {gen}: best={stats['best_fitness']:.4f} avg={stats['avg_fitness']:.4f} div={stats['diversity']:.3f}")
if stats["diversity"] < 0.05:
print("WARNING: Population converged — injecting random individuals")
for i in range(self.pop_size // 4):
genes = {name: gene.random_value() for name, gene in self.gene_defs.items()}
self.population[-(i+1)] = Genome(genes=genes, generation=gen, id=f"random_{gen}_{i}")
self.evaluate(strategy_runner, data)
return {
"best_genome": self.population[0],
"top_5": [(g.genes, round(g.fitness, 4)) for g in self.population[:5]],
"history": self.history,
"WARNING": "Validate with walk-forward OOS before live deployment. GA results overfit easily.",
}
def _population_diversity(self) -> float:
"""Measure population diversity (0=identical, 1=maximum spread)."""
if len(self.population) < 2: return 0
diversities = []
for name, gene_def in self.gene_defs.items():
vals = [g.genes[name] for g in self.population if isinstance(g.genes[name], (int, float))]
if vals and (gene_def.max_val - gene_def.min_val) > 0:
diversities.append(np.std(vals) / (gene_def.max_val - gene_def.min_val))
return np.mean(diversities) if diversities else 0
Anti-Overfitting Safeguards
- Always split data: optimize on 60%, validate on 40% OOS
- Diversity enforcement: inject randoms when population converges
- Penalize complexity: fewer parameters = better (Occam's razor)
- Walk-forward validation: use walk-forward-optimizer skill on best genomes
- Multiple fitness functions: rank by Sharpe AND Sortino AND Calmar — not just one