Algoritmo evolutivo de trading com aprendizado por reforço e extinção de estratégias não lucrativas (ETARE)

Introdução

Sabe o que a evolução, as redes neurais e os traders têm em comum? Isso mesmo, todos aprendem com os próprios erros. Essa foi a ideia que me veio à mente depois de mais uma madrugada em claro diante do terminal, quando meu "algoritmo de trading perfeito" mais uma vez evaporou o saldo em um movimento inesperado do mercado.

Lembro daquele dia como se fosse hoje: 23 de junho de 2016, o referendo do Brexit. Meu algoritmo, baseado nos padrões clássicos da análise técnica, mantinha com confiança uma posição comprada em libra. "Todas as pesquisas mostram que o Reino Unido vai permanecer na UE", pensava eu na época. E então, às 4 da manhã, horário de Moscou, quando os primeiros resultados indicaram a vitória dos que queriam sair, a libra despencou 1800 pontos em questão de minutos. O saldo evaporou em 40%.

Em março de 2023 comecei a desenvolver o ETARE, o algoritmo evolutivo de trading com reforço e extinção (eliminação). Por que extinção? Porque na natureza, só os mais fortes sobrevivem. Então, por que não aplicar esse princípio às estratégias de trading?

Pronto para mergulhar em um mundo onde a análise técnica clássica encontra os avanços mais recentes da inteligência artificial? Onde cada estratégia de negociação luta pela sobrevivência em um darwinismo puro? Então aperte o cinto, porque isso vai ser interessante. Porque o que você está prestes a ver não é apenas mais um robô de trading. É o resultado de 15 anos de tentativas e erros, milhares de horas programando e, para ser honesto, algumas contas queimadas. Mas o mais importante: é um sistema funcional, que já gera lucro real para seus usuários.

Arquitetura do sistema

No coração do ETARE está uma arquitetura híbrida, semelhante à de um computador quântico moderno. Lembra de quando escrevíamos scripts simples para o MetaTrader4, baseados no cruzamento de duas médias móveis? Na época, aquilo parecia um avanço. Hoje, olhando para trás, percebo: éramos como navegadores antigos tentando atravessar o oceano apenas com uma bússola e as estrelas.

Depois do colapso de 2022 ficou claro: o mercado é complexo demais para soluções simples. Foi então que começou minha jornada pelo universo do aprendizado de máquina.

class HybridTrader:
    def __init__(self, symbols, population_size=50):
        self.population = []  # Population of strategies
        self.extinction_rate = 0.3  # Extinction rate
        self.elite_size = 5  # Elite individuals
        self.inefficient_extinction_interval = 5  # Cleaning interval

Imagine uma colônia de formigas, onde cada formiga representa uma estratégia de negociação. As estratégias fortes sobrevivem e passam seus genes para as próximas gerações, as fracas, desaparecem. No meu sistema, o papel dos genes é desempenhado pelos pesos da rede neural.

Por que population_size=50? Porque uma quantidade menor de estratégias não proporciona diversificação suficiente, e uma quantidade maior dificulta a adaptação rápida às mudanças do mercado.

Na natureza, as formigas estão constantemente explorando novos territórios, encontrando alimento e transmitindo informações aos seus pares. No ETARE, cada estratégia também explora o mercado, e os padrões de negociação bem-sucedidos são passados para as próximas gerações por meio de um mecanismo de cruzamento:

def _crossover(self, parent1, parent2):
    child = TradingIndividual(self.input_size)
    # Cross scales through a mask
    for attr in ['input_weights', 'hidden_weights', 'output_weights']:
        parent1_weights = getattr(parent1.weights, attr)
        parent2_weights = getattr(parent2.weights, attr)
        mask = np.random.random(parent1_weights.shape) < 0.5
        child_weights = np.where(mask, parent1_weights, parent2_weights)
        setattr(child.weights, attr, child_weights)
    return child

Em 2024 (em dezembro), ao analisar os logs de negociação, notei que os códigos mais bem-sucedidos frequentemente eram "híbridos" de outras abordagens também bem-sucedidas. Assim como na natureza genes fortes geram descendentes saudáveis, na negociação algorítmica os padrões vencedores podem se combinar e dar origem a estratégias ainda mais eficazes.

O coração do sistema passou a ser uma rede LSTM, que é um tipo especial de rede neural com "memória". Após vários meses de experimentação com diferentes arquiteturas, desde perceptrons multicamadas simples até transformadores mais complexos, chegamos a essa configuração:

class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(LSTMModel, self).__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, batch_first=True)
        self.dropout = nn.Dropout(0.4)  # Protection from overfitting
        self.fc = nn.Linear(hidden_size, output_size)
    
    def forward(self, x):
        out, _ = self.lstm(x)
        out = self.dropout(out[:, -1, :])  # Use the last LSTM output
        out = self.fc(out)
        return out

A cada 100 operações a sistema realiza uma "limpeza", eliminando sem piedade as estratégias com prejuízo. Esse é um dos mecanismos centrais do ETARE, e sua criação é uma história à parte. Lembro de uma noite em dezembro de 2023, quando analisava os logs de operações e notei uma correlação surpreendente: a maioria das estratégias que apresentou prejuízo nas primeiras 100-150 operações continuava sendo deficitária dali em diante. Essa observação mudou completamente a arquitetura do sistema:

def _inefficient_extinction_event(self):
    """Periodic extinction of inefficient individuals"""
    initial_size = len(self.population)
    
    # Analyze efficiency of each strategy
    performance_metrics = []
    for individual in self.population:
        metrics = {
            'profit_factor': individual.total_profit / abs(individual.max_drawdown) if individual.max_drawdown != 0 else 0,
            'win_rate': len([t for t in individual.trade_history if t.profit > 0]) / len(individual.trade_history) if individual.trade_history else 0,
            'risk_adjusted_return': individual.total_profit / individual.volatility if individual.volatility != 0 else 0
        }
        performance_metrics.append(metrics)
    
    # Remove unprofitable strategies taking into account a comprehensive assessment
    self.population = [ind for ind, metrics in zip(self.population, performance_metrics)
                      if metrics['profit_factor'] > 1.5 or metrics['win_rate'] > 0.6]
    
    # Create new individuals with improved initialization
    while len(self.population) < initial_size:
        new_individual = TradingIndividual(self.input_size)
        new_individual.mutate()  # Random mutations
        
        # Inherit successful patterns
        if len(self.population) > 0:
            parent = random.choice(self.population)
            new_individual.inherit_patterns(parent)
            
        self.population.append(new_individual)

O banco de dados de decisões de trade funciona como a memória do sistema. Cada decisão, cada resultado, tudo é registrado para posterior análise:

def _save_to_db(self):
    with self.conn:
        self.conn.execute('DELETE FROM population')
        for individual in self.population:
            data = {
                'weights': individual.weights.to_dict(),
                'fitness': individual.fitness,
                'profit': individual.total_profit
            }
            self.conn.execute(
                'INSERT INTO population (data) VALUES (?)',
                (json.dumps(data),)
            )

Todo esse mecanismo complexo funciona como um organismo único, evoluindo constantemente e se adaptando às mudanças do mercado. Em momentos de alta volatilidade, por exemplo, quando o VIX ultrapassa 25, o sistema automaticamente eleva os requisitos de confiabilidade das estratégias. Já em períodos mais tranquilos, ele se torna mais agressivo, permitindo testar novos padrões de negociação.

Mecanismo de aprendizado por reforço

No desenvolvimento de EAs existe um paradoxo: quanto mais complexo o algoritmo, pior ele se sai no mercado real. 

Por isso, no mecanismo de aprendizado do ETARE optamos pela simplicidade e transparência. Após dois anos de testes com diferentes arquiteturas, chegamos a um sistema de memória priorizada:

class RLMemory:
    def __init__(self, capacity=10000):
        self.memory = deque(maxlen=capacity)
        self.priorities = deque(maxlen=capacity)
        
    def add(self, state, action, reward, next_state):
        priority = max(self.priorities) if self.priorities else 1.0
        self.memory.append((state, action, reward, next_state))
        self.priorities.append(priority)

Cada decisão de trade não é apenas uma entrada no mercado, mas sim um equilíbrio complexo entre risco e lucro potencial. Veja como o sistema aprende com suas próprias decisões:

def update(self, state, action, reward, next_state):
    self.memory.add(state, action, reward, next_state)
    self.total_profit += reward

    if len(self.memory.memory) >= 32:
        batch = self.memory.sample(32)
        self._train_on_batch(batch)

Perdi muitas vezes porque os modelos não conseguiam se ajustar a um novo estado do mercado. Eles eram simplesmente varridos. Foi aí que surgiu a ideia de aprendizado adaptativo. Agora o sistema analisa cada operação e ajusta seu comportamento:

def _calculate_confidence(self, prediction, patterns):
    # Baseline confidence from ML model
    base_confidence = abs(prediction - 0.5) * 2
    
    # Consider historical experience
    pattern_confidence = self._get_pattern_confidence(patterns)
    
    # Dynamic adaptation to the market
    market_volatility = self._get_current_volatility()
    return (base_confidence * 0.7 + pattern_confidence * 0.3) / market_volatility

O ponto chave é que o sistema não apenas memoriza operações bem-sucedidas, ele aprende a entender por que elas foram bem-sucedidas. Isso só foi possível graças à arquitetura multinível de propagação reversa do erro, implementada em PyTorch:

def _train_on_batch(self, batch):
    states = torch.FloatTensor(np.array([x[0] for x in batch]))
    actions = torch.LongTensor(np.array([x[1].value for x in batch]))
    rewards = torch.FloatTensor(np.array([x[2] for x in batch]))
    next_states = torch.FloatTensor(np.array([x[3] for x in batch]))
    
    current_q = self.forward(states).gather(1, actions.unsqueeze(1))
    next_q = self.forward(next_states).max(1)[0].detach()
    target = rewards + self.gamma * next_q
    
    loss = self.criterion(current_q.squeeze(), target)
    self.optimizer.zero_grad()
    loss.backward()
    self.optimizer.step()

O resultado foi um sistema que aprende não com backtests ideais, mas com experiência real de mercado. No último ano de testes em conta real, o ETARE demonstrou capacidade de adaptação a diversos cenários: desde tendências suaves até períodos de alta volatilidade.

Mas o mais importante é que o sistema continua evoluindo. A cada operação, a cada ciclo de mercado, ele se torna um pouco mais inteligente, um pouco mais eficiente. Como disse um de nossos beta testers: "É a primeira vez que vejo um algoritmo que realmente aprende com os próprios erros, e não apenas ajusta os parâmetros para se encaixar nos dados históricos".

Mecanismo de extinção das estratégias com prejuízo

Charles Darwin nunca operou nos mercados financeiros, mas sua teoria da evolução descreve de forma impressionante a dinâmica das estratégias de negociação bem-sucedidas. Na natureza, não sobrevivem os indivíduos mais fortes ou mais rápidos, mas sim os que melhor se adaptam às mudanças do ambiente. O mercado funciona da mesma forma.

A história tem muitos exemplos de algoritmos "perfeitos" que viraram abóboras no primeiro cisne negro. Em 2015 perdi uma parte significativa do saldo quando o Banco Nacional Suíço desatrelou o franco do euro. Meu algoritmo da época não estava nem remotamente preparado para algo assim. Isso me fez pensar: por que a natureza sobrevive há milhões de anos a eventos "imprevisíveis", e nossos algoritmos não?

A resposta veio de forma inesperada, enquanto eu lia o livro "A Origem das Espécies". Darwin descrevia como, em períodos de mudanças climáticas bruscas, não eram as espécies mais especializadas que sobreviviam, mas sim aquelas que mantinham a capacidade de adaptação. Esse foi exatamente o princípio que deu origem ao mecanismo de extinção no ETARE:

def _inefficient_extinction_event(self):
    """Periodic extinction of inefficient individuals"""
    initial_population = len(self.population)
    market_conditions = self._analyze_market_state()
    
    # Assessing the adaptability of each strategy
    adaptability_scores = []
    for individual in self.population:
        score = self._calculate_adaptability(
            individual, 
            market_conditions
        )
        adaptability_scores.append(score)
    
    # Dynamic survival threshold
    survival_threshold = np.percentile(
        adaptability_scores, 
        30  # The bottom 30% of the population is dying out
    )
    
    # Merciless extinction
    survivors = []
    for ind, score in zip(self.population, adaptability_scores):
        if score > survival_threshold:
            survivors.append(ind)
    
    self.population = survivors
    
    # Restore population through mutations and crossbreeding
    while len(self.population) < initial_population:
        if len(self.population) >= 2:
            # Crossbreeding of survivors
            parent1 = self._tournament_selection()
            parent2 = self._tournament_selection()
            child = self._crossover(parent1, parent2)
        else:
            # Create a new individual 
            child = TradingIndividual(self.input_size)
        
        # Mutations for adaptation
        child.mutate(market_conditions.volatility)
        self.population.append(child)

Assim como na natureza os períodos de extinção em massa levam ao surgimento de espécies mais avançadas, em nosso sistema os momentos de alta volatilidade funcionam como catalisadores da evolução das estratégias. Veja como funciona esse mecanismo de seleção natural:

def _extinction_event(self):
    # Analyze market conditions
    market_phase = self._identify_market_phase()
    volatility = self._calculate_market_volatility()
    trend_strength = self._measure_trend_strength()
    
    # Adaptive sorting by survival
    def fitness_score(individual):
        return (
            individual.profit_factor * 0.4 +
            individual.sharp_ratio * 0.3 +
            individual.adaptability_score * 0.3
        ) * (1 + individual.correlation_with_market)
    
    self.population.sort(
        key=fitness_score, 
        reverse=True
    )
    
    # Preserve elite with diversity in mind
    elite_size = max(
        5, 
        int(len(self.population) * 0.1)
    )
    survivors = self.population[:elite_size]
    
    # Create a new generation
    while len(survivors) < self.population_size:
        if random.random() < 0.8:  # 80% crossover
            # Tournament selection of parents
            parent1 = self._tournament_selection()
            parent2 = self._tournament_selection()
            
            # Crossbreeding considering account market conditions
            child = self._adaptive_crossover(
                parent1, 
                parent2, 
                market_phase
            )
        else:  # 20% elite mutation
            # Clone with mutations
            template = random.choice(survivors[:3])
            child = self._clone_with_mutations(
                template,
                volatility,
                trend_strength
            )
        survivors.append(child)

Dedicamos atenção especial ao mecanismo de avaliação da adaptabilidade. Na natureza, isso significa a capacidade do indivíduo de gerar descendentes viáveis; no nosso caso, a capacidade da estratégia de gerar lucro em diferentes condições de mercado:

def evaluate_fitness(self, individual):
    # Basic metrics
    profit_factor = individual.total_profit / max(
        abs(individual.total_loss), 
        1e-6
    )
    
    # Resistance to drawdowns
    max_dd = max(individual.drawdown_history) if individual.drawdown_history else 0
    drawdown_resistance = 1 / (1 + max_dd)
    
    # Profit sequence analysis
    profit_sequence = [t.profit for t in individual.trade_history[-50:]]
    consistency = self._analyze_profit_sequence(profit_sequence)
    
    # Correlation with the market
    market_correlation = self._calculate_market_correlation(
        individual.trade_history
    )
    
    # Adaptability to changes
    adaptability = self._measure_adaptability(
        individual.performance_history
    )
    
    # Comprehensive assessment
    fitness = (
        profit_factor * 0.3 +
        drawdown_resistance * 0.2 +
        consistency * 0.2 +
        (1 - abs(market_correlation)) * 0.1 +
        adaptability * 0.2
    )
    
    return fitness

Veja como ocorre a mutação das estratégias sobreviventes. Esse processo lembra as mutações genéticas na natureza, onde alterações aleatórias no DNA às vezes geram organismos mais viáveis:

def mutate(self, market_conditions):
    """Adaptive mutation considering market conditions"""
    # Dynamic adjustment of mutation strength
    self.mutation_strength = self._calculate_mutation_strength(
        market_conditions.volatility,
        market_conditions.trend_strength
    )
    
    if np.random.random() < self.mutation_rate:
        # Mutation of neural network weights
        for weight_matrix in [
            self.weights.input_weights,
            self.weights.hidden_weights,
            self.weights.output_weights
        ]:
            # Mutation mask with adaptive threshold
            mutation_threshold = 0.1 * (
                1 + market_conditions.uncertainty
            )
            mask = np.random.random(weight_matrix.shape) < mutation_threshold
            
            # Volatility-aware mutation generation
            mutations = np.random.normal(
                0,
                self.mutation_strength * market_conditions.volatility,
                size=mask.sum()
            )
            
            # Apply mutations
            weight_matrix[mask] += mutations
            
        # Mutation of hyperparameters
        if random.random() < 0.3:  # 30% chance
            self._mutate_hyperparameters(market_conditions)

Curiosamente, em algumas versões do sistema — durante períodos de alta volatilidade do mercado — o sistema automaticamente aumenta a intensidade das mutações. Isso se assemelha ao que ocorre com algumas bactérias que aceleram suas mutações em situações de estresse. No nosso caso:

def _calculate_mutation_strength(self, volatility, trend_strength):
    """Calculate mutation strength based on market conditions"""
    base_strength = self.base_mutation_strength
    
    # Mutation enhancement under high volatility
    volatility_factor = 1 + (volatility / self.average_volatility - 1)
    
    # Weaken mutations in a strong trend
    trend_factor = 1 / (1 + trend_strength)
    
    # Mutation total strength
    mutation_strength = (
        base_strength * 
        volatility_factor * 
        trend_factor
    )
    
    return np.clip(
        mutation_strength,
        self.min_mutation_strength,
        self.max_mutation_strength
    )

É especialmente importante o mecanismo de diversificação da população. Na natureza, a diversidade genética é a chave para a sobrevivência da espécie. No ETARE implementamos um princípio semelhante:

def _maintain_population_diversity(self):
    """ Maintain diversity in the population"""
    # Calculate the strategy similarity matrix
    similarity_matrix = np.zeros(
        (len(self.population), len(self.population))
    )
    
    for i, ind1 in enumerate(self.population):
        for j, ind2 in enumerate(self.population[i+1:], i+1):
            similarity = self._calculate_strategy_similarity(ind1, ind2)
            similarity_matrix[i,j] = similarity_matrix[j,i] = similarity
    
    # Identify clusters of similar strategies
    clusters = self._identify_strategy_clusters(similarity_matrix)
    
    # Forced diversification when necessary
    for cluster in clusters:
        if len(cluster) > self.max_cluster_size:
            # We leave only the best strategies in the cluster
            survivors = sorted(
                cluster,
                key=lambda x: x.fitness,
                reverse=True
            )[:self.max_cluster_size]
            
            # Replace the rest with new strategies
            for idx in cluster[self.max_cluster_size:]:
                self.population[idx] = TradingIndividual(
                    self.input_size,
                    mutation_rate=self.high_mutation_rate
                )

Resultado? Um sistema que não apenas opera, mas evolui junto com o mercado. Como disse Darwin, sobrevive não o mais forte, mas o mais adaptável. No mundo do trading algorítmico, isso nunca foi tão verdadeiro.

Banco de dados de decisões de trade

Guardar a experiência de trading é tão importante quanto adquiri-la. Ao longo dos anos trabalhando com sistemas algorítmicos, percebi muitas vezes: sem um banco de dados confiável, qualquer sistema de trading acabará por "esquecer" suas melhores estratégias. No ETARE, implementamos um armazenamento multinível das decisões de negociação:

def _create_tables(self):
    """ Create a database structure"""
    with self.conn:
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS population (
                id INTEGER PRIMARY KEY,
                individual TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_update TIMESTAMP
            )
        ''')
        
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS history (
                id INTEGER PRIMARY KEY,
                generation INTEGER,
                individual_id INTEGER,
                trade_history TEXT,
                market_conditions TEXT,
                FOREIGN KEY(individual_id) REFERENCES population(id)
            )
        ''')

Cada operação, cada decisão — mesmo aquelas que parecem insignificantes — se tornam parte da experiência coletiva do sistema. Veja como os dados são armazenados após cada ciclo de trading:

def _save_to_db(self):
    try:
        with self.conn:
            self.conn.execute('DELETE FROM population')
            for individual in self.population:
                individual_data = {
                    'weights': {
                        'input_weights': individual.weights.input_weights.tolist(),
                        'hidden_weights': individual.weights.hidden_weights.tolist(),
                        'output_weights': individual.weights.output_weights.tolist(),
                        'hidden_bias': individual.weights.hidden_bias.tolist(),
                        'output_bias': individual.weights.output_bias.tolist()
                    },
                    'fitness': individual.fitness,
                    'total_profit': individual.total_profit,
                    'trade_history': list(individual.trade_history),
                    'market_metadata': self._get_market_conditions()
                }
                self.conn.execute(
                    'INSERT INTO population (individual) VALUES (?)', 
                    (json.dumps(individual_data),)
                )
    except Exception as e:
        logging.error(f"Error saving population: {str(e)}")

Mesmo após uma falha crítica no servidor, todo o sistema pode ser restaurado em questão de minutos, graças aos logs detalhados e backups de segurança. É assim que funciona o mecanismo de recuperação:

def _load_from_db(self):
    """Load population from database"""
    try:
        cursor = self.conn.execute('SELECT individual FROM population')
        rows = cursor.fetchall()
        for row in rows:
            individual_data = json.loads(row[0])
            individual = TradingIndividual(self.input_size)
            individual.weights = GeneticWeights(**individual_data['weights'])
            individual.fitness = individual_data['fitness']
            individual.total_profit = individual_data['total_profit']
            individual.trade_history = deque(
                individual_data['trade_history'], 
                maxlen=1000
            )
            self.population.append(individual)
    except Exception as e:
        logging.error(f"Error loading population: {str(e)}")

Dedicamos atenção especial à análise dos dados históricos. Cada estratégia bem-sucedida deixa um rastro que pode ser usado para melhorar decisões futuras:

def analyze_historical_performance(self):
    """ Historical performance analysis"""
    query = '''
        SELECT h.*, p.individual 
        FROM history h 
        JOIN population p ON h.individual_id = p.id 
        WHERE h.generation > ? 
        ORDER BY h.generation DESC
    '''
    
    cursor = self.conn.execute(query, (self.generation - 100,))
    performance_data = cursor.fetchall()
    
    # Analyze patterns of successful strategies
    success_patterns = defaultdict(list)
    for record in performance_data:
        trade_data = json.loads(record[3])
        if trade_data['profit'] > 0:
            market_conditions = json.loads(record[4])
            key_pattern = self._extract_key_pattern(market_conditions)
            success_patterns[key_pattern].append(trade_data)
    
    return success_patterns

O banco de dados do ETARE não é apenas um repositório de informações, é o verdadeiro "cérebro" do sistema, capaz de analisar o passado e prever o futuro. Como dizia meu antigo mentor: "Um sistema de trading sem memória é como um trader sem experiência: todo dia começa do zero".

Dados e características

Depois de anos trabalhando com negociação algorítmica, testei centenas de combinações de indicadores. Em certo momento, meu sistema usava mais de 50 indicadores diferentes, desde o RSI clássico até indicadores exóticos desenvolvidos por mim mesmo. Mas sabe o que percebi depois de mais um depósito evaporado? A questão não está na quantidade, mas no processamento correto dos dados.

Lembro de um episódio durante o Brexit: o sistema, sobrecarregado por dezenas de indicadores, simplesmente "congelou", incapaz de tomar uma decisão diante de sinais contraditórios. Foi justamente aí que nasceu a ideia do ETARE, que é um sistema que usa o conjunto mínimo necessário de indicadores, mas os processa de forma inteligente.

def prepare_features(data: pd.DataFrame) -> pd.DataFrame:
    """Prepare features for analysis"""
    df = data.copy()

    # RSI - as an overbought/oversold detector
    delta = df['close'].diff()
    gain = delta.where(delta > 0, 0).rolling(14).mean()
    loss = -delta.where(delta < 0, 0).rolling(14).mean()
    rs = gain / loss
    df['rsi'] = 100 - (100 / (1 + rs))

O RSI no nosso sistema não é apenas um indicador de sobrecompra/sobrevenda. Usamos ele como parte de uma análise complexa do sentimento de mercado. Funciona especialmente bem em combinação com o MACD:

# MACD - to determine the trend
    exp1 = df['close'].ewm(span=12, adjust=False).mean()
    exp2 = df['close'].ewm(span=26, adjust=False).mean()
    df['macd'] = exp1 - exp2
    df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
    df['macd_hist'] = df['macd'] - df['macd_signal']

As Bandas de Bollinger são nosso "radar" de volatilidade.  

# Bollinger Bands with adaptive period
    volatility = df['close'].rolling(50).std()
    adaptive_period = int(20 * (1 + volatility.mean()))
    
    df['bb_middle'] = df['close'].rolling(adaptive_period).mean()
    df['bb_std'] = df['close'].rolling(adaptive_period).std()
    df['bb_upper'] = df['bb_middle'] + 2 * df['bb_std']
    df['bb_lower'] = df['bb_middle'] - 2 * df['bb_std']

A análise de volatilidade e impulso é uma história à parte. 

# Momentum - market "temperature"
    df['momentum'] = df['close'] / df['close'].shift(10)
    df['momentum_ma'] = df['momentum'].rolling(20).mean()
    df['momentum_std'] = df['momentum'].rolling(20).std()
    
    # Volatility is our "seismograph"
    df['atr'] = df['high'].rolling(14).max() - df['low'].rolling(14).min()
    df['price_change'] = df['close'].pct_change()
    df['price_change_abs'] = df['price_change'].abs()
    
    # Volume volatility
    df['volume_volatility'] = df['tick_volume'].rolling(20).std() / df['tick_volume'].rolling(20).mean()

A análise de volume no ETARE não é apenas a contagem de ticks. Desenvolvemos um algoritmo específico para detectar volumes anômalos, que ajuda a prever movimentos fortes:

# Volume analysis - market "pulse"
    df['volume_ma'] = df['tick_volume'].rolling(20).mean()
    df['volume_std'] = df['tick_volume'].rolling(20).std()
    df['volume_ratio'] = df['tick_volume'] / df['volume_ma']
    
    # Detection of abnormal volumes
    df['volume_spike'] = (
        df['tick_volume'] > df['volume_ma'] + 2 * df['volume_std']
    ).astype(int)
    
    # Cluster analysis of volumes
    df['volume_cluster'] = (
        df['tick_volume'].rolling(3).sum() / 
        df['tick_volume'].rolling(20).sum()
    )

O toque final é a normalização dos dados. Esta é uma etapa criticamente importante, e que muitos subestimam.

# Normalization considering market phases
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    for col in numeric_cols:
        # Adaptive normalization
        rolling_mean = df[col].rolling(100).mean()
        rolling_std = df[col].rolling(100).std()
        df[col] = (df[col] - rolling_mean) / (rolling_std + 1e-8)
    
    # Removing outliers
    df = df.clip(-4, 4)  # Limit values to the range [-4, 4]
    
    return df

Cada característica no ETARE não é apenas um número, mas parte de um mosaico complexo de análise de mercado. O sistema se adapta continuamente às mudanças do mercado, ajustando o peso de cada indicador conforme a situação atual. Nas próximas seções, veremos como esses dados se transformam em decisões de trading concretas.

Lógica de trading

Apresento a você a descrição de um sistema de negociação inovador que incorpora o que há de mais avançado em tecnologia de trading algorítmico. A base do sistema é uma abordagem híbrida, que une otimização genética, aprendizado de máquina e gerenciamento de risco avançado.

O coração do sistema é um ciclo de trading contínuo, que analisa constantemente as condições do mercado e se adapta a elas. Assim como na evolução natural, o sistema realiza periodicamente uma “limpeza” das estratégias ineficazes, abrindo espaço para novas abordagens mais promissoras. Isso ocorre a cada 50 operações, garantindo o aprimoramento constante dos algoritmos de trading.

Cada instrumento financeiro é tratado de forma individual, levando em consideração suas características únicas. O sistema analisa os dados históricos das últimas 100 velas, o que permite formar uma visão precisa do estado atual do mercado. Com base nessa análise, são tomadas decisões ponderadas de abertura e fechamento de posições.

Damos atenção especial à estratégia de média de preço (DCA). Ao abrir novas posições, o sistema reduz automaticamente o volume de cada ordem, começando com 0.1 lote e diminuindo gradualmente até o valor mínimo de 0.01 lote. Isso permite um gerenciamento de risco eficiente e maximiza o potencial de lucro.

O processo de fechamento de posições também foi cuidadosamente planejado. O sistema acompanha a lucratividade de cada posição e as encerra ao atingir um nível predefinido de lucro. As posições Buy e Sell são tratadas separadamente, o que permite uma gestão mais flexível do portfólio. E as recompensas ou penalidades geradas pelas operações são a chave para o aprendizado contínuo e bem-sucedido do sistema. 

Todas as informações sobre as operações de trading e o estado do sistema são armazenadas no banco de dados, garantindo a possibilidade de análise detalhada e otimização das estratégias. Isso cria uma base sólida para o aperfeiçoamento contínuo dos algoritmos de negociação.

    def _process_individual(self, symbol: str, individual: TradingIndividual, current_state: np.ndarray):
        """Handle trading logic for an individual using DCA and split closing by profit"""
        try:
            positions = individual.open_positions.get(symbol, [])

            if not positions:  # Open a new position
                action, _ = individual.predict(current_state)
                if action in [Action.OPEN_BUY, Action.OPEN_SELL]:
                    self._open_position(symbol, individual, action)
            else:  # Manage existing positions
                current_price = mt5.symbol_info_tick(symbol).bid

                # Close positions by profit
                self._close_positions_by_profit(symbol, individual, current_price)

                # Check for the need to open a new position by DCA
                if len(positions) < self.max_positions_per_pair:
                    action, _ = individual.predict(current_state)
                    if action in [Action.OPEN_BUY, Action.OPEN_SELL]:
                        self._open_dca_position(symbol, individual, action, len(positions))

        except Exception as e:
            logging.error(f"Error processing individual: {str(e)}")

    def _open_position(self, symbol: str, individual: TradingIndividual, action: Action):
        """Open a position"""
        try:
            volume = 0.1
            price = mt5.symbol_info_tick(symbol).ask if action == Action.OPEN_BUY else mt5.symbol_info_tick(symbol).bid

            request = {
                "action": mt5.TRADE_ACTION_DEAL,
                "symbol": symbol,
                "volume": volume,
                "type": mt5.ORDER_TYPE_BUY if action == Action.OPEN_BUY else mt5.ORDER_TYPE_SELL,
                "price": price,
                "deviation": 20,
                "magic": 123456,
                "comment": f"Gen{self.generation}",
                "type_time": mt5.ORDER_TIME_GTC,
                "type_filling": mt5.ORDER_FILLING_FOK,
            }

            result = mt5.order_send(request)
            if result and result.retcode == mt5.TRADE_RETCODE_DONE:
                trade = Trade(symbol=symbol, action=action, volume=volume,
                              entry_price=result.price, entry_time=time.time())
                if symbol not in individual.open_positions:
                    individual.open_positions[symbol] = []
                individual.open_positions[symbol].append(trade)

        except Exception as e:
            logging.error(f"Error opening position: {str(e)}")

    def _open_dca_position(self, symbol: str, individual: TradingIndividual, action: Action, position_count: int):
        """Open a position using the DCA strategy"""
        try:
            # Basic volume
            base_volume = 0.1  # Initial volume in lots
            # Reduce the volume by 0.01 lot for each subsequent position
            volume = max(0.01, base_volume - (position_count * 0.01))  # Minimum volume of 0.01 lots
            price = mt5.symbol_info_tick(symbol).ask if action == Action.OPEN_BUY else mt5.symbol_info_tick(symbol).bid

            request = {
                "action": mt5.TRADE_ACTION_DEAL,
                "symbol": symbol,
                "volume": volume,
                "type": mt5.ORDER_TYPE_BUY if action == Action.OPEN_BUY else mt5.ORDER_TYPE_SELL,
                "price": price,
                "deviation": 20,
                "magic": 123456,
                "comment": f"Gen{self.generation} DCA",
                "type_time": mt5.ORDER_TIME_GTC,
                "type_filling": mt5.ORDER_FILLING_FOK,
            }

            result = mt5.order_send(request)
            if result and result.retcode == mt5.TRADE_RETCODE_DONE:
                trade = Trade(symbol=symbol, action=action, volume=volume,
                              entry_price=result.price, entry_time=time.time())
                if symbol not in individual.open_positions:
                    individual.open_positions[symbol] = []
                individual.open_positions[symbol].append(trade)

        except Exception as e:
            logging.error(f"Error opening DCA position: {str(e)}")

    def _close_positions_by_profit(self, symbol: str, individual: TradingIndividual, current_price: float):
        """Close positions by profit separately for Buy and Sell"""
        try:
            positions = individual.open_positions.get(symbol, [])
            buy_positions = [pos for pos in positions if pos.action == Action.OPEN_BUY]
            sell_positions = [pos for pos in positions if pos.action == Action.OPEN_SELL]

            # Close Buy positions
            for position in buy_positions:
                profit = calculate_profit(position, current_price)
                if profit >= self.min_profit_pips:
                    self._close_position(symbol, individual, position)

            # Close Sell positions
            for position in sell_positions:
                profit = calculate_profit(position, current_price)
                if profit >= self.min_profit_pips:
                    self._close_position(symbol, individual, position)

        except Exception as e:
            logging.error(f"Error closing positions by profit: {str(e)}")

    def _close_position(self, symbol: str, individual: TradingIndividual, position: Trade):
        """Close a position with a model update"""
        try:
            close_type = mt5.ORDER_TYPE_SELL if position.action == Action.OPEN_BUY else mt5.ORDER_TYPE_BUY
            price = mt5.symbol_info_tick(symbol).bid if close_type == mt5.ORDER_TYPE_SELL else mt5.symbol_info_tick(symbol).ask

            request = {
                "action": mt5.TRADE_ACTION_DEAL,
                "symbol": symbol,
                "volume": position.volume,
                "type": close_type,
                "price": price,
                "deviation": 20,
                "magic": 123456,
                "comment": "Close",
                "type_time": mt5.ORDER_TIME_GTC,
                "type_filling": mt5.ORDER_FILLING_FOK,
            }

            result = mt5.order_send(request)
            if result and result.retcode == mt5.TRADE_RETCODE_DONE:
                position.is_open = False
                position.exit_price = result.price
                position.exit_time = time.time()
                position.profit = calculate_profit(position, result.price)
                
                # Generate data for training
                trade_data = {
                    'symbol': symbol,
                    'action': position.action,
                    'entry_price': position.entry_price,
                    'exit_price': position.exit_price,
                    'volume': position.volume,
                    'profit': position.profit,
                    'holding_time': position.exit_time - position.entry_time
                }
                
                # Update the model with new data
                individual.model.update(trade_data)
                
                # Save history and update open positions
                individual.trade_history.append(position)
                individual.open_positions[symbol].remove(position)
                
                # Log training results
                logging.info(f"Model updated with trade data: {trade_data}")

        except Exception as e:
            logging.error(f"Error closing position: {str(e)}")

def main():
    symbols = ['EURUSD.ecn', 'GBPUSD.ecn', 'USDJPY.ecn', 'AUDUSD.ecn']
    trader = HybridTrader(symbols)
    trader.run_trading_cycle()

if __name__ == "__main__":
    main()

Como resultado, temos um sistema de trading confiável e autoaprendente, capaz de operar de forma eficiente em diversas condições de mercado. A combinação de algoritmos evolutivos, aprendizado de máquina e estratégias de negociação comprovadas o torna uma ferramenta poderosa para o trading moderno.

Considerações finais

Para concluir, quero destacar que o ETARE não é apenas mais um algoritmo de negociação, mas o resultado de anos de evolução na área de trading algorítmico. O sistema reúne as melhores práticas de diferentes áreas: algoritmos genéticos para adaptação às mudanças do mercado, aprendizado profundo para tomada de decisões, e métodos clássicos de gerenciamento de risco.

A singularidade do ETARE está na sua capacidade de aprender continuamente com a própria experiência. Cada operação, independentemente do resultado, torna-se parte da memória coletiva do sistema, contribuindo para o aperfeiçoamento das decisões futuras. O mecanismo de seleção natural de estratégias de negociação, inspirado na teoria da evolução de Darwin, garante que apenas as abordagens mais eficazes sobrevivam.

Durante seu desenvolvimento e testes, o sistema demonstrou resiliência em diferentes cenários de mercado — de movimentos de tendência suaves a períodos de alta volatilidade. É especialmente importante destacar a eficácia da estratégia DCA e do mecanismo de fechamento separado de posições, que permitem maximizar os lucros mantendo o risco sob controle.

Agora, sobre a questão da efetividade. Vou ser direto: o módulo principal do ETARE, por si só, não realiza operações de trading. Ele está embutido como um módulo dentro de um ecossistema de negociação mais amplo chamado "Midas".

Atualmente, o Midas conta com 24 módulos, incluindo este. A complexidade vai aumentar gradualmente, e muitos detalhes ainda serão descritos nos artigos futuros. 

O futuro do trading algorítmico está justamente em sistemas adaptativos como este, capazes de evoluir junto com o mercado. O ETARE é um passo nessa direção, mostrando como a tecnologia moderna pode ser aplicada na criação de soluções de negociação confiáveis e lucrativas.