Algoritmo de trading evolutivo con aprendizaje por refuerzo y extinción de individuos no rentables (ETARE)

Introducción

¿Sabe qué tienen en común la evolución, las redes neuronales y los tráders? Exacto: todos aprenden de sus errores. Este mismo pensamiento me vino después de otra noche en vela ante el terminal, cuando mi "perfecto" algoritmo comercial volvió a agotar mi depósito por un movimiento inesperado del mercado.

Recuerdo ese día como lo hago ahora: 23 de junio de 2016, referéndum sobre el Brexit. Mi algoritmo, basado en patrones clásicos de análisis técnico, mantenía con confianza una posición larga en la libra esterlina. "Todas las encuestas muestran que Gran Bretaña permanecerá en la UE", pensé entonces. Y así, a las 4 de la madrugada, hora de Moscú, cuando los primeros resultados mostraron la victoria de los partidarios de la salida, la libra se desplomó 1.800 puntos en solo unos minutos. El depósito se evaporó en un 40%.

Así que en marzo de 2023, empecé a desarrollar el ETARE - Algoritmo de trading evolutivo con aprendizaje por refuerzo y extinción (eliminación) de individuos no rentables. ¿Por qué la eliminación? Porque en la naturaleza sobrevive el más fuerte. ¿Y por qué no aplicar este principio a las estrategias comerciales?

¿Está listo para sumergirse en un mundo en el que el análisis técnico clásico se funde con lo último en inteligencia artificial, donde cada estrategia comercial lucha por sobrevivir en la selección natural darwiniana? Pues abróchese el cinturón: será interesante. Lo que está a punto de ver no es un robot de trading más. Es el resultado de 15 años de ensayo y error, miles de horas de programación y, francamente, unos cuantos depósitos quemados. Pero lo principal es un sistema que funciona y que ya aporta beneficios reales a sus usuarios.

Arquitectura del sistema

En el corazón de ETARE se encuentra una arquitectura híbrida que recuerda a una computadora cuántica moderna. ¿Recuerda aquellos tiempos en los que escribíamos sencillos scripts para MetaTrader4 basados en el cruce de dos medias móviles? En aquel momento parecía un gran avance, pero, mirando atrás, me doy cuenta de que éramos como antiguos navegantes intentando cruzar el océano con solo una brújula y las estrellas.

Tras el crack de 2022, se hizo evidente: el mercado resulta demasiado complejo para soluciones sencillas. Fue entonces cuando comenzó mi viaje por el mundo del aprendizaje automático.

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 una colonia de hormigas en la que cada hormiga es una estrategia comercial. Los individuos fuertes sobreviven y transmiten sus genes a su descendencia, mientras que los débiles terminan desapareciendo. En mi sistema, el papel de los genes lo desempeñan los coeficientes de ponderación de la red neuronal.

¿Por qué population_size=50? Porque un número inferior de estrategias no proporciona suficiente diversificación, mientras que un número superior dificulta una rápida adaptación a los cambios del mercado.

En la naturaleza, las hormigas exploran constantemente nuevos territorios, encuentran comida y transmiten la información a sus parientes. En ETARE, cada estrategia investiga el mercado de la misma forma, y los patrones comerciales exitosos se transmiten a la siguiente generación mediante un mecanismo de cruces:

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

En 2024 (diciembre), mientras analizaba los registros comerciales, me di cuenta de que los códigos con más éxito suelen ser "híbridos" de otros enfoques exitosos. Al igual que en la naturaleza los genes fuertes producen una descendencia sana, en el trading algorítmico los patrones exitosos pueden combinarse para crear estrategias aún mejores.

El corazón del sistema era una red LSTM, un tipo especial de red neuronal con "memoria". Tras meses de experimentar con distintas arquitecturas, desde perceptrones multicapa simples hasta complejos transformadores, nos decidimos por esta configuración:

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

Cada 100 transacciones, el sistema realiza una "limpieza", eliminando sin piedad las estrategias no rentables. Este es uno de los mecanismos clave de ETARE, y su creación supone una historia aparte. Recuerdo una noche de diciembre de 2023 en la que estaba analizando los registros de transacciones y observé un patrón sorprendente: la mayoría de las estrategias que mostraban pérdidas en las primeras 100-150 transacciones seguían sin resultar rentables más adelante. Esta observación cambió por completo la arquitectura del 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)

La base de datos de soluciones comerciales actúa como la memoria del sistema. Cada decisión, cada resultado; todo queda registrado para su posterior análisis:

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),)
            )

Este complejo mecanismo funciona como un único organismo, en constante evolución y adaptación a los cambios del mercado. En momentos de alta volatilidad, como cuando el VIX supera 25, el sistema aumenta de manera automática los requisitos de fiabilidad de las estrategias. Y en los periodos de calma, en cambio, se vuelve más agresivo, lo que le permite experimentar con nuevas pautas comerciales.

Mecanismo de aprendizaje por refuerzo

Existe una paradoja en el desarrollo de los robots comerciales: cuanto más complejo sea el algoritmo, peor funcionará en el mercado real. 

Por ello, hemos hecho hincapié en la sencillez y la transparencia del mecanismo de aprendizaje de ETARE. Tras dos años experimentando con distintas arquitecturas, llegamos a un sistema de memoria 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 decisión comercial no consiste simplemente en entrar en el mercado, sino en un complejo equilibrio entre riesgo y beneficio potencial. Observa cómo aprende el sistema de sus decisiones:

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)

He perdido muchas veces porque los modelos no podían readaptarse a unas condiciones de mercado distintas. Simplemente eran barridos. Fue entonces cuando surgió la idea del aprendizaje adaptativo. El sistema analiza ahora cada transacción y ajusta su comportamiento según esta:

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

El punto clave es que el sistema no solo recuerda las transacciones exitosas, sino que aprende a entender por qué dichas transacciones han sido exitosas. Esto es posible gracias a la arquitectura multinivel de retropropagación del error implementada en 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()

Como resultado, obtenemos un sistema que no aprende de pruebas retrospectivas perfectas, sino de la experiencia comercial real. Durante el último año de pruebas en el mercado real, ETARE ha demostrado su capacidad para adaptarse a diversas condiciones de mercado, desde tendencias tranquilas hasta periodos de gran volatilidad.

Pero lo más importante es que el sistema continúa evolucionando. Con cada transacción, con cada ciclo de mercado, se vuelve un poco más inteligente, un poco más eficiente. Como dijo uno de nuestros probadores beta: "Es la primera vez que veo un algoritmo que realmente aprende de sus errores en lugar de limitarse a ajustar los parámetros a los datos históricos".

Mecanismo de extinción de los individuos no rentables

Charles Darwin nunca comerció en los mercados financieros, pero su teoría de la evolución describe de forma asombrosa la dinámica de las estrategias comerciales de éxito. En la naturaleza, no son los individuos más fuertes o más rápidos los que sobreviven, sino los que mejor se adaptan a los cambios del entorno. Lo mismo ocurre en el mercado.

La historia conoce muchos casos en los que un algoritmo comercial "perfecto" se convierte en calabaza tras el primer cisne negro. En 2015, perdí una parte importante de mi depósito cuando el Banco Nacional Suizo desvinculó el franco del euro. En aquel momento, mi algoritmo no estaba preparado para semejante acontecimiento. Esto me hizo pensar: ¿por qué la naturaleza lleva millones de años enfrentándose con éxito a los "cisnes negros" y nuestros algoritmos no?

La respuesta llegó inesperadamente, mientras leía "El origen de las especies". Darwin describió cómo no eran las especies más especializadas las que sobrevivían durante los periodos de cambio climático brusco, sino las que conservaban la capacidad de adaptarse. Este es el principio en el que se basa el mecanismo de extinción de 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)

Al igual que en la naturaleza los periodos de extinción masiva provocan la aparición de nuevas y mejores especies, en nuestro sistema los periodos de alta volatilidad se convierten en un catalizador para la evolución de las estrategias. Eche un vistazo al mecanismo de la selección 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)

Hemos prestado especial atención al mecanismo de evaluación de la adaptabilidad. En la naturaleza, es la capacidad de un individuo para producir descendencia viable; en nuestro caso, será la capacidad de una estrategia para generar beneficios en diferentes condiciones 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

Así es como funcionan las estrategias de mutación de los supervivientes. Este proceso se asemeja a las mutaciones genéticas en la naturaleza, donde los cambios aleatorios en el ADN a veces dan lugar a organismos más viables:

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, en algunas versiones del sistema, durante los periodos de alta volatilidad del mercado, el sistema aumenta la intensidad de las mutaciones de forma automática. Esto resulta similar a la forma en que algunas bacterias aceleran las mutaciones en condiciones de estrés. En nuestro 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
    )

Resulta especialmente importante el mecanismo de diversificación de la población. En la naturaleza, la diversidad genética es la clave para la supervivencia de una especie. En ETARE hemos aplicado un principio similar:

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? Un sistema que no se limita a comerciar, sino que evoluciona con el mercado. Como dijo Darwin, no sobrevive el más apto, sino el más adaptable. En el mundo del trading algorítmico, esto es más relevante que nunca.

Base de datos de soluciones comerciales

La conservación de la experiencia comercial es tan importante como su adquisición. A lo largo de los años de trabajo con sistemas algorítmicos me he convencido muchas veces: sin una base de datos fiable, cualquier sistema comercial "olvidará" tarde o temprano sus mejores estrategias. En ETARE, hemos implantado un almacenamiento por niveles para las soluciones comerciales:

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 transacción, cada decisión, incluso las que parecen insignificantes, se incorpora a la experiencia colectiva del sistema. De esta manera, guardaremos los datos después de cada ciclo comercial:

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)}")

Incluso tras un fallo crítico del servidor, el sistema completo se recuperará en cuestión de minutos, gracias a los registros detallados y las copias de seguridad. Así es como funciona el mecanismo de recuperación:

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)}")

Prestamos especial atención al análisis de los datos históricos. Toda estrategia acertada deja una huella que puede usarse para mejorar futuras decisiones:

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

La base de datos ETARE no supone un mero depósito de información, sino un auténtico "cerebro" del sistema, capaz de analizar el pasado y predecir el futuro. Como solía decir mi viejo mentor: "Un sistema comercial sin memoria es como un tráder sin experiencia: empieza de cero cada día".

Datos y características

A lo largo de años de práctica con el trading algorítmico, he probado cientos de combinaciones de indicadores. En un momento dado, mi sistema comercial ha usado más de 50 indicadores diferentes, desde el clásico RSI hasta indicadores exóticos diseñados por mí. Pero, ¿sabe de qué me di cuenta cierto día, después de otro depósito perdido? No se trata de cantidad, sino de manejar los datos correctamente.

Recuerdo un caso especialmente memorable durante el Brexit: un sistema con docenas de indicadores simplemente se "congeló", incapaz de tomar una decisión debido a señales contradictorias. Fue entonces cuando nació la idea de ETARE, un sistema que usa el conjunto mínimo necesario de indicadores, pero los procesa 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))

El RSI en nuestro sistema no supone solo un indicador de sobrecompra/sobreventa. Lo usamos como parte de un análisis exhaustivo del sentimiento del mercado. Funciona especialmente bien en combinación con el 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']

Las Bandas de Bollinger son nuestro "radar" de volatilidad.  

# 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']

Una historia aparte sería el análisis de la volatilidad y el impulso. 

# 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()

El análisis volumétrico en ETARE es algo más que contabilizar los ticks. Hemos desarrollado un algoritmo especial para detectar volúmenes anormales, que ayuda a predecir movimientos intensos:

# 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()
    )

El toque final será la normalización de los datos. Este es un paso fundamental que mucha gente infravalora.

# 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 atributo de ETARE no supone solo un número, sino que forma parte de un complejo mosaico de análisis de mercado. El sistema se adapta constantemente a los cambios del mercado, ajustando el peso de cada indicador según la situación actual. En los siguientes apartados veremos cómo se convierten estos datos en decisiones comerciales concretas.

Lógica comercial

Le presentamos la descripción de un innovador sistema comercial que incorpora las avanzadas tecnologías del trading algorítmico. El sistema se basa en un enfoque híbrido que combina la optimización genética, el aprendizaje automático y la gestión avanzada de riesgos.

El núcleo del sistema es un ciclo comercial continuo que analiza las condiciones del mercado y se adapta a ellas de forma constante. Al igual que sucede en la evolución natural, el sistema "limpia" periódicamente las estrategias comercial ineficaces, dando paso a enfoques nuevos y más prometedores. Esto ocurre cada 50 transacciones, lo cual garantiza que los algoritmos comercial mejoren constantemente.

Cada instrumento comercial se trata de forma individual, considerando sus características únicas. El sistema analiza los datos históricos de las últimas 100 velas, lo cual le permite formarse una idea precisa del estado actual del mercado. Partiendo de este análisis, se toman decisiones fundamentadas sobre la apertura y el cierre de posiciones.

En ella se presta especial atención a la estrategia de promediado de posiciones (DCA). Al abrir nuevas posiciones, el sistema reduce automáticamente su volumen empezando por 0,1 lote y disminuyendo de forma gradual hasta el valor mínimo de 0,01 lote. Esto permite gestionar eficazmente el riesgo y maximizar los beneficios potenciales.

También hemos estudiado atentamente el proceso de cierre de posiciones. El sistema controla la rentabilidad de cada posición y las cierra cuando se alcanza el nivel de beneficios establecido. Las posiciones de compra y venta se procesan aparte, lo que permite una gestión más flexible del portafolio, mientras que las recompensas o penalizaciones obtenidas de la negociación son la clave para seguir aprendiendo con éxito. 

Toda la información sobre las transacciones comerciales y el estado del sistema se almacena en una base de datos, lo cual permite tanto realizar un análisis detallado como optimizar las estrategias. Y esto ofrece una base sólida para seguir perfeccionando los algoritmos comerciales.

    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, obtenemos un sistema comercial fiable y autodidacta que puede operar eficazmente en diversas condiciones de mercado. La combinación de algoritmos evolutivos, aprendizaje automático y estrategias comerciales probadas lo convierten en una potente herramienta para el trading moderno.

Conclusión

Para concluir, me gustaría subrayar que el ETARE no es un algoritmo comercial más, sino el resultado de muchos años de evolución en el trading algorítmico. El sistema combina las mejores prácticas de diversos campos: algoritmos genéticos para la adaptación a las condiciones del mercado cambiantes, aprendizaje profundo para la toma de decisiones y técnicas clásicas de gestión de riesgos.

La peculiaridad única de ETARE es su capacidad de aprender continuamente de la experiencia. Cada transacción, independientemente del resultado, pasa a formar parte de la memoria colectiva del sistema, lo que ayuda a mejorar las decisiones comerciales futuras. El mecanismo de selección natural de estrategias comerciales, inspirado en la teoría de la evolución de Darwin, garantiza que solo sobrevivan los planteamientos más eficaces.

Durante el desarrollo y las pruebas, el sistema ha demostrado estabilidad en diversas condiciones de mercado, desde movimientos de tendencia tranquilos hasta periodos de gran volatilidad. Cabe destacar la eficacia de la estrategia DCA y el mecanismo de cierre de posiciones separadas, que maximizan el beneficio con un nivel de riesgo controlado.

Ahora sobre la eficiencia. Lo diré simple y llanamente: en mi caso, el propio módulo central de ETARE no comercia, sino que se integra como un módulo en el ecosistema comercial más amplio de Midas.

Actualmente hay 24 módulos en Midas, incluido este. La complejidad crecerá a pasos agigantados, así que hablaremos de ello en futuros artículos. 

El futuro del trading algorítmico reside en estos sistemas adaptativos que pueden evolucionar con el mercado. El ETARE supone un paso en esta dirección, y demuestra cómo puede aplicarse la tecnología moderna para crear soluciones comerciales fiables y rentables.