Robot comercial multimodular en Python y MQL5 (Parte I): Creamos la arquitectura básica y los primeros módulos

Introducción

Un día se me ocurrió una idea: los robots comerciales son demasiado simples para el mercado actual, necesitamos algo más flexible e inteligente.

El mercado cambia constantemente: una estrategia funciona hoy y mañana resultará inútil. Llevo mucho tiempo observando esto y me he dado cuenta de que hace falta un enfoque completamente nuevo. La decisión surgió de la nada: ¿y si el sistema fuera modular? Imagine un equipo de profesionales: uno sigue las tendencias, el segundo analiza los volúmenes comerciales, y el tercero controla los riesgos. Así es exactamente como debería funcionar un robot comercial moderno.

Así que la elección de la tecnología era obvia. Python era perfecto para el análisis de datos: se pueden hacer maravillas con sus bibliotecas. MQL5 se encargó de la ejecución de las transacciones. Ambos hacían un gran tándem. Empezamos poco a poco: primero creamos una base sólida, una arquitectura que puede crecer y desarrollarse, y luego añadimos la interacción entre Python y MQL5. El sistema de gestión de datos resultó sorprendentemente sencillo y eficaz.

¡La asincronía fue un verdadero avance! Ahora el robot puede controlar varios instrumentos a la vez, y la productividad se ha disparado.

¿Y sabe qué es lo mejor? Este sistema realmente funciona en el mercado. No supone un ejemplo de algún manual: es una herramienta de combate. Obviamente, empezaremos con la versión básica, pero incluso esa resulta impresionante. Tenemos un gran viaje por delante: hoy crearemos un sistema capaz de aprender y adaptarse, y lo mejoraremos paso a paso. Entre tanto, vamos a empezar por lo más importante, por la construcción de unos cimientos sólidos.

Arquitectura básica del sistema. En busca del equilibrio perfecto

Llevo tres años esforzándome por crear robots comerciales. ¿Y sabes de qué me di cuenta? Lo importante no son los algoritmos en sí, sino cómo estos funcionan juntos. Y este descubrimiento lo cambió todo.

Imagínese una orquesta. Todos los músicos son geniales, pero sin un director de orquesta, la música no se podrá tocar. En mi sistema, MarketMaker se ha convertido en un director de este tipo. Dirige cuatro módulos, y cada uno conoce su tarea:

• El primer módulo controla los volúmenes comerciales, es decir, cuándo y a qué precios se realizan las transacciones.
• El segundo módulo busca oportunidades de arbitraje.
• El tercer módulo analiza la economía. 
• El cuarto módulo evita que el sistema se despiste y controla los riesgos.

El mercado no espera a nadie: cambia a la velocidad del rayo, por lo que todos los módulos trabajan simultáneamente, comunicándose entre sí de manera constante. El módulo de arbitraje ve una oportunidad e informa al centro. Este contrasta la información con los demás módulos y toma una decisión.

Al principio pensaba establecer normas estrictas para entrar en el mercado, pero la vida no tardó en demostrar que ese no era el camino. A veces, una señal fuerte es más importante que varias débiles. ¡Y cuánto tiempo lleva organizar los datos! Cada módulo tiene su propia información: cotizaciones, macroindicadores, historia de transacciones. Todo debe almacenarse, actualizarse y compartirse con los demás módulos. Así que tuve que crear un sistema especial de sincronización.

Lo curioso es que cuanto más independientes eran los módulos, mejor funcionaba el sistema. El fallo de un componente no detenía al resto. Y los fallos suceden: la conexión puede interrumpirse o las cotizaciones puede congelarse. La principal ventaja de esta arquitectura es que se puede ampliar. ¿Quiere añadir análisis de noticias? No hay problema. Cree un módulo, conéctelo a MarketMaker y todo funcionará.

El sistema seguirá vivo y coleando. No es perfecto, pero su estructura basada en la modularidad, el paralelismo y la flexibilidad le permite mirar al futuro con confianza. Pronto hablaré más de cada componente.

Clase principal del sistema

Después de experimentar mucho con diferentes enfoques de la arquitectura de los robots comerciales, me he dado cuenta de que el éxito de un sistema depende en gran medida de lo bien organizado que esté su núcleo. MarketMaker ha sido el resultado de esta comprensión, encarnando todas las mejores prácticas que he acumulado a lo largo de los años de desarrollo de sistemas algorítmicos.

Empezaremos por la estructura básica de la clase. Este será el aspecto de su inicialización:

def __init__(self, pairs: List[str], terminal_path: str, 
             volume: float = 1.0, levels: int = 5, spacing: float = 3.0):
    # Main parameters
    self.pairs = pairs
    self.base_volume = volume
    self.levels = levels
    self.spacing = spacing
    self.magic = 12345
    
    # Trading parameters
    self.portfolio_iterations = 10
    self.leverage = 50
    self.min_profit_pips = 1.0
    self.max_spread_multiplier = 2.0
    
    # Data warehouses
    self.symbols_info = {}
    self.trading_parameters = {}
    self.optimal_horizons = {}

A primera vista, parece bastante sencillo, pero detrás de cada parámetro hay una historia. Por ejemplo, portfolio_iterations: este parámetro surgió cuando noté que abrir posiciones de forma demasiado agresiva podía provocar problemas de liquidez. El sistema desglosa ahora el volumen disponible en partes, lo cual hace que la negociación sea más equilibrada.

He prestado especial atención a la inicialización de los datos históricos. Así es como funciona:

def _initialize_history(self, pair: str):
    """Initializing historical data for a pair"""
    try:
        rates = mt5.copy_rates_from(pair, mt5.TIMEFRAME_M1, 
                                  datetime.now()-timedelta(days=1), 1440)
        if rates is not None:
            df = pd.DataFrame(rates)
            df['time'] = pd.to_datetime(df['time'], unit='s')
            df.set_index('time', inplace=True)
            returns = np.log(df['close'] / df['close'].shift(1)).dropna()
            self.returns_history[pair] = pd.Series(returns.values, 
                                                 index=df.index[1:])
    except Exception as e:
        logger.error(f"Error initializing history for {pair}: {e}")

Aquí el punto interesante es el uso de rendimientos logarítmicos en lugar de simples cambios porcentuales. No se trata de una elección al azar. En la práctica, he comprobado que los rendimientos logarítmicos ofrecen resultados más estables a la hora de calcular medidas estadísticas, especialmente en lo que se refiere a la volatilidad.

Uno de los mayores retos ha sido la realización de las previsiones de volumen. Tras muchos experimentos, creé este código:

async def update_volume_predictions(self):
    """Updating volume predictions for each pair"""
    for pair in self.pairs:
        try:
            df = volume_model.get_volume_data(
                symbol=pair,
                timeframe=mt5.TIMEFRAME_H1,
                n_bars=100
            )
            
            if pair in self.volume_models:
                feature_columns = [
                    'volume_sma_5', 'volume_sma_20', 'relative_volume', 
                    'volume_change', 'volume_volatility', 'price_sma_5', 
                    'price_sma_20', 'price_change', 'price_volatility',
                    'rsi', 'macd', 'macd_signal', 'bb_upper', 'bb_lower'
                ]
                
                X = df[feature_columns].iloc[-1:].copy()
                prediction = self.volume_models[pair].predict(X)[0]
                current_price = df['close'].iloc[-1]
                predicted_change = (prediction - current_price) / current_price
                
                self.volume_predictions[pair] = predicted_change
                
        except Exception as e:
            logger.error(f"Error updating prediction for {pair}: {e}")

Preste atención al conjunto de características: no se trata solo de un conjunto aleatorio de indicadores. Cada uno de ellos se añadió de forma gradual, tras una cuidadosa comprobación. Por ejemplo, el volumen relativo (relative_volume) ha demostrado ser especialmente útil para identificar la actividad anormal en el mercado.

Y así es como se ve el corazón del sistema, el ciclo comercial:

async def trade_cycle(self):
    """Main trading loop"""
    try:
        await self.update_volume_predictions()
        await self.economic_module.update_forecasts()
        
        all_positions = mt5.positions_get() or []
        open_positions = [pos for pos in all_positions if pos.magic == self.magic]
        
        if open_positions:
            await self.manage_positions()
            return
            
        valid_signals = []
        available_volume = self.calculate_available_volume() * len(self.pairs)
        
        for pair in self.pairs:
            signal = await self.get_combined_signal(pair)
            if signal and self._validate_signal(signal):
                valid_signals.append(signal)
        
        if valid_signals:
            volume_per_trade = available_volume / len(valid_signals)
            for signal in valid_signals:
                signal['adjusted_volume'] = volume_per_trade
                await self.place_order(signal)
                
    except Exception as e:
        logger.error(f"Error in trade cycle: {e}")

Este código es el resultado de largas reflexiones sobre la correcta organización del proceso comercial. La naturaleza asíncrona del ciclo permite procesar eficientemente muchos pares al mismo tiempo, mientras que la clara secuencia de acciones (actualización de previsiones → comprobación de posiciones → búsqueda de señales → ejecución) garantiza un comportamiento predecible del sistema.

El mecanismo de validación de las señales merece especial atención:

def _validate_signal(self, signal: Dict) -> bool:
    """Trading signal check"""
    spread = signal['spread']
    diff_pips = signal['diff_pips']
    
    # Basic checks
    if spread > self.max_spread_multiplier * diff_pips:
        return False
        
    if diff_pips < self.min_profit_pips:
        return False
        
    # Check economic factors
    if signal['economic_volatility'] > self.volatility_threshold:
        return False
        
    # Check the volume prediction
    if abs(signal['volume_prediction']) < self.min_volume_change:
        return False
        
    return True

Todas las comprobaciones realizadas aquí son el resultado de una experiencia comercial real. Por ejemplo, la comprobación de la volatilidad económica se ha añadido después de notar que la negociación durante las noticias importantes a menudo provoca mayores pérdidas debido a los movimientos bruscos de los precios.

Para terminar, me gustaría señalar que MarketMaker supone un sistema vivo que sigue evolucionando. Cada día comercial aporta nuevas ideas y mejoras. La arquitectura modular facilita la aplicación de estas mejoras sin alterar los componentes básicos.

Trabajando con los datos

El trabajo con datos siempre ha sido uno de los aspectos más difíciles del trading algorítmico. Recuerdo cómo, al inicio del desarrollo, me enfrenté a una pregunta aparentemente simple: ¿cómo organizar el almacenamiento y el procesamiento de la información de mercado de la forma adecuada? Pronto quedó claro que una base de datos normal o unos arrays sencillos no serían suficientes.

Todo empezó con la creación de una estructura básica para recuperar los datos. Tras varias iteraciones, surgió este método:

def _initialize_history(self, pair: str):
    try:
        rates = mt5.copy_rates_from(pair, mt5.TIMEFRAME_M1, 
                                  datetime.now()-timedelta(days=1), 1440)
        if rates is None:
            logger.error(f"Failed to get history data for {pair}")
            return
            
        df = pd.DataFrame(rates)
        df['time'] = pd.to_datetime(df['time'], unit='s')
        df.set_index('time', inplace=True)
        
        # Calculate logarithmic returns
        returns = np.log(df['close'] / df['close'].shift(1)).dropna()
        
        # Add new metrics
        df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3
        df['price_velocity'] = df['close'].diff() / df['time'].diff().dt.total_seconds()
        df['volume_intensity'] = df['tick_volume'] / df['time'].diff().dt.total_seconds()
        
        self.returns_history[pair] = pd.Series(returns.values, index=df.index[1:])
        self.price_data[pair] = df
        
    except Exception as e:
        logger.error(f"Error initializing history for {pair}: {e}")

Lo interesante aquí es el cálculo de la "velocidad" del cambio de precios (price_velocity) y la intensidad del volumen (volume_intensity). Estas métricas no aparecieron de golpe. Al principio solo trabajaba con datos de precios ordinarios, pero enseguida noté que el mercado no supone solo una secuencia de precios, sino un complejo sistema dinámico en el que no solo es importante la magnitud de los cambios, sino también su velocidad.

Tuve que prestar especial atención al procesamiento de los datos ausentes. Este es el aspecto del sistema de validación y limpieza:

def _validate_and_clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
    """Validation and data cleaning"""
    if df.empty:
        raise ValueError("Empty dataset received")
        
    # Check gaps
    missing_count = df.isnull().sum()
    if missing_count.any():
        logger.warning(f"Found missing values: {missing_count}")
        
        # Use 'forward fill' for prices
        price_cols = ['open', 'high', 'low', 'close']
        df[price_cols] = df[price_cols].ffill()
        
        # Use interpolation for volumes
        df['tick_volume'] = df['tick_volume'].interpolate(method='linear')
    
    # Check outliers
    for col in ['high', 'low', 'close']:
        zscore = stats.zscore(df[col])
        outliers = abs(zscore) > 3
        if outliers.any():
            logger.warning(f"Found {outliers.sum()} outliers in {col}")
            
            # Replace extreme outliers
            df.loc[outliers, col] = df[col].mean() + 3 * df[col].std() * np.sign(zscore[outliers])
    
    return df

Recuerdo un caso en el que la omisión de un solo tick provocó un cálculo incorrecto de los indicadores y, como consecuencia, una señal comercial incorrecta. Desde entonces, el sistema de limpieza se ha perfeccionado sustancialmente.

Y así es como procesamos el volumen, una de las características más importantes del mercado:

def analyze_volume_profile(self, pair: str, window: int = 100) -> Dict:
    """Volume profile analysis"""
    try:
        df = self.price_data[pair].copy().last(window)
        
        # Normalize volumes
        volume_mean = df['tick_volume'].rolling(20).mean()
        volume_std = df['tick_volume'].rolling(20).std()
        df['normalized_volume'] = (df['tick_volume'] - volume_mean) / volume_std
        
        # Calculate volume clusters
        price_levels = pd.qcut(df['close'], q=10)
        volume_clusters = df.groupby(price_levels)['tick_volume'].sum()
        
        # Find support/resistance levels by volume
        significant_levels = volume_clusters[volume_clusters > volume_clusters.mean() + volume_clusters.std()]
        
        # Analyze imbalances
        buy_volume = df[df['close'] > df['open']]['tick_volume'].sum()
        sell_volume = df[df['close'] <= df['open']]['tick_volume'].sum()
        volume_imbalance = (buy_volume - sell_volume) / (buy_volume + sell_volume)
        
        return {
            'normalized_profile': volume_clusters.to_dict(),
            'significant_levels': significant_levels.index.to_list(),
            'volume_imbalance': volume_imbalance,
            'current_percentile': stats.percentileofscore(df['tick_volume'], df['tick_volume'].iloc[-1])
        }
        
    except Exception as e:
        logger.error(f"Error analyzing volume profile: {e}")
        return None

Este código es el resultado de un largo estudio de la microestructura del mercado. El cálculo del desequilibrio del volumen entre compras y ventas resulta especialmente interesante. Originalmente estudié esto en el mercado de criptomonedas, no sé si la administración de MQL5 dará el visto bueno para publicar el código con la integración de cripto bolsas, MetaTrader 5, y Python....

Bueno, estoy divagando. A primera vista, puede parecer que una simple comparación de los volúmenes en las barras alcistas y bajistas no aportará información útil. Pero la práctica ha demostrado que este sencillo indicador suele advertir sobre una reversión próxima de tendencia.

Trabajar con datos económicos es ya harina de otro costal. Aquí tuve que crear todo un sistema de sincronización:

async def synchronize_market_data(self):
    """Market data synchronization"""
    while True:
        try:
            # Update basic data
            for pair in self.pairs:
                latest_data = await self._get_latest_ticks(pair)
                if latest_data is not None:
                    self._update_price_data(pair, latest_data)
                    
            # Update derived metrics
            await self._update_derivatives()
            
            # Check data integrity
            self._verify_data_integrity()
            
            await asyncio.sleep(1)  # Dynamic delay
            
        except Exception as e:
            logger.error(f"Error in data synchronization: {e}")
            await asyncio.sleep(5)  # Increased delay on error

El punto clave es la naturaleza asíncrona de las actualizaciones de datos. En versiones anteriores del sistema usaba consultas síncronas, pero esto provocaba retrasos al procesar un gran número de pares. El cambio a un modelo asíncrono ha mejorado notablemente el rendimiento.

En conclusión, organizar bien los datos no supone solo una cuestión técnica. Esta es la base sobre la que se construye toda la estrategia comercial. Los datos limpios y bien estructurados permiten ver patrones de mercado que permanecen ocultos en análisis superficiales.

Módulo uno: análisis de volumen

La historia de la creación del módulo de análisis de volumen comenzó con una simple observación: los indicadores clásicos suelen fallar porque solo trabajan con los precios. Pero el mercado no es solo cuestión de precios, sino también de volúmenes comerciales, que con frecuencia predicen el movimiento de las cotizaciones. Por eso, el primer módulo de nuestro sistema fue un analizador de volumen.

Empezaremos por la función básica de recuperación de datos:

def get_volume_data(symbol, timeframe=mt5.TIMEFRAME_H1, n_bars=2000):
    """Getting volume and price data from MT5"""
    try:
        bars = mt5.copy_rates_from_pos(symbol, timeframe, 0, n_bars)
        if bars is None:
            logger.error(f"Failed to get data for {symbol}")
            return None
        
        df = pd.DataFrame(bars)
        df['time'] = pd.to_datetime(df['time'], unit='s')
        return df
        
    except Exception as e:
        logger.error(f"Error getting data for {symbol}: {e}")
        return None

A primera vista, la función parece sencilla, pero esta simplicidad esconde una decisión importante: tomamos exactamente 2000 barras de la historia. ¿Y por qué? Experimentalmente he descubierto que esto basta para construir un modelo de calidad, pero sin crear una carga excesiva en la memoria del servidor en caso de entrenar modelos muy grandes con conjuntos de datos voluminosos y una entrada de características como secuencias de lotes.

La parte más interesante del módulo es la creación de las características para el análisis. Así es como funciona:

def create_features(df, forecast_periods=None):
    """Create features for the forecasting model"""
    try:
        # Basic volume indicators
        df['volume_sma_5'] = df['tick_volume'].rolling(window=5).mean()
        df['volume_sma_20'] = df['tick_volume'].rolling(window=20).mean()
        df['relative_volume'] = df['tick_volume'] / df['volume_sma_20']
        
        # Volume dynamics
        df['volume_change'] = df['tick_volume'].pct_change()
        df['volume_acceleration'] = df['volume_change'].diff()
        
        # Volume volatility
        df['volume_volatility'] = df['tick_volume'].rolling(window=20).std()
        df['volume_volatility_5'] = df['tick_volume'].rolling(window=5).std()
        df['volume_volatility_ratio'] = df['volume_volatility_5'] / df['volume_volatility']

Aquí deberemos prestar especial atención a volume_volatility_ratio. Este indicador apareció después de observar un patrón interesante: antes de los movimientos fuertes, la volatilidad del volumen a corto plazo suele crecer más rápido que la volatilidad a largo plazo. Este indicador se ha convertido en uno de los indicadores clave a la hora de identificar posibles puntos de entrada.

El cálculo del perfil de volumen tiene una historia interesante:

# Volume profile
        df['volume_percentile'] = df['tick_volume'].rolling(window=100).apply(
            lambda x: pd.Series(x).rank(pct=True).iloc[-1]
        )
        df['volume_density'] = df['tick_volume'] / (df['high'] - df['low'])
        df['volume_density_ma'] = df['volume_density'].rolling(window=20).mean()
        df['cumulative_volume'] = df['tick_volume'].rolling(window=20).sum()
        df['volume_ratio'] = df['tick_volume'] / df['cumulative_volume']

El indicador volume_density no ha aparecido por casualidad. Me he dado cuenta de que el volumen por sí mismo puede resultar engañoso: debemos tener en cuenta en qué rango de precios se ha ido ganando. Un volumen elevado en un rango de precios estrecho suele indicar la formación de un nivel de apoyo o resistencia importante.

Hemos desarrollado una función especial para predecir la dirección del movimiento de los precios:

def predict_direction(model, X):
    """Price movement direction forecast"""
    try:
        prediction = model.predict(X)[0]
        current_price = X['close'].iloc[-1] if 'close' in X else None
        if current_price is None:
            return 0
            
        # Return 1 for rise, -1 for fall, 0 for neutral
        price_change = (prediction - current_price) / current_price
        if abs(price_change) < 0.0001:  # Minimum change threshold
            return 0
        return 1 if price_change > 0 else -1
        
    except Exception as e:
        logger.error(f"Error predicting direction: {e}")
        return 0

Observe el umbral de cambio de 0.0001. No se trata de un número aleatorio: este se elige partiendo del análisis del movimiento mínimo que puede calcularse considerando el spread y las comisiones de diversa índole. Para el mercado bursátil, el indicador deberá seleccionarse por separado.

La última etapa es el entrenamiento del modelo:

def train_model(X_train, X_test, y_train, y_test, model_params=None):
    try:
        if model_params is None:
            model_params = {'n_estimators': 400, 'random_state': 42}
            
        model = RandomForestRegressor(**model_params)
        model.fit(X_train, y_train)
        
        # Model evaluation
        train_predictions = model.predict(X_train)
        test_predictions = model.predict(X_test)
        
        train_rmse = np.sqrt(mean_squared_error(y_train, train_predictions))
        test_rmse = np.sqrt(mean_squared_error(y_test, test_predictions))
        test_r2 = r2_score(y_test, test_predictions)

¿Sabe por qué elegí RandomForest con 400 árboles? Después de probar un montón de cosas, desde una simple regresión hasta arquitecturas neuronales de enorme complejidad, he llegado a la conclusión de que este método es el más fiable. Puede que no sea el más preciso, pero es estable: el mercado está agitado y nervioso, pero RandomForest resiste bien.

Obviamente, esto es solo el principio. Lo siguiente resulta más interesante: ¿cómo se unen todas estas señales, cómo se ajusta el sistema para aprender sobre la marcha? Pero dejaremos eso para la próxima vez.

Gestión del riesgo: el arte de preservar nuestro capital

Ahora viene lo más importante: los riesgos. Resulta divertido escuchar a todo el mundo hablar de estrategias y redes neuronales geniales. En diez años en el mercado he aprendido lo principal: sin control del riesgo, ninguna de estas estrategias sirve de nada. Puede tener un algoritmo comercial estupendo, pero sin una gestión adecuada del riesgo seguirá en negativo.

Así que en nuestro sistema, la protección del capital es algo sagrado. ¿Y sabe qué? Es precisamente este enfoque conservador el que nos permite ganar dinero sistemáticamente mientras otros pierden dinero con estrategias "perfectas".

def calculate_available_volume(self) -> float:
    try:
        account = mt5.account_info()
        if not account:
            return 0.01
            
        # Use balance and free margin
        balance = account.balance
        free_margin = account.margin_free
        
        # Take the minimum value for safety
        available_margin = min(balance, free_margin)
        
        # Calculate the maximum volume taking into account the margin
        margin_ratio = 0.1  # Use only 10% of the available margin
        base_volume = (available_margin * margin_ratio) / 1000
        
        # Limit to maximum volume
        max_volume = min(base_volume, 1.0)  # max 1 lot

Preste atención a que margin_ratio = 0,1 . No es un número aleatorio. Tras meses de pruebas, he llegado a la conclusión de que usar más de un 10% de margen disponible aumenta significativamente el riesgo de que se produzcan llamadas de margen durante los movimientos de mercado intensos. Esto resulta especialmente importante cuando se negocian varios pares al mismo tiempo.

El siguiente punto importante es el cálculo del stop-loss y el take-profit:

async def calculate_position_limits(self, signal: Dict) -> Tuple[float, float]:
    try:
        pair = signal['pair']
        direction = signal['direction']
        
        # Get volatility
        volatility = signal['price_volatility']
        economic_volatility = signal['economic_volatility']
        
        # Base values in pips
        base_sl = 20
        base_tp = 40
        
        # Adjust for volatility
        volatility_factor = 1 + (volatility * 2)
        sl_points = base_sl * volatility_factor
        tp_points = base_tp * volatility_factor
        
        # Take economic volatility into account
        if economic_volatility > 0.5:
            sl_points *= 1.5
            tp_points *= 1.2
            
        # Check minimum distances
        info = self.symbols_info[pair]
        min_stop_level = info.trade_stops_level if hasattr(info, 'trade_stops_level') else 0
        
        return max(sl_points, min_stop_level), max(tp_points, min_stop_level)
        
    except Exception as e:
        logger.error(f"Error calculating position limits: {e}")
        return 20, 40  # return base values in case of an error

La historia de volatility_factor es especialmente interesante. Al principio usaba stops fijos, pero rápidamente me di cuenta de que durante los periodos de alta volatilidad a menudo se activaban demasiado pronto, cayendo mucho. Los ajustes dinámicos de los stops según la volatilidad actual han mejorado considerablemente los resultados de las transacciones.

Y este es el aspecto del sistema de gestión de posiciones:

async def manage_positions(self):
    """Managing open positions"""
    try:
        positions = mt5.positions_get() or []
        for position in positions:
            if position.magic == self.magic:
                # Check the time in the position
                time_in_trade = datetime.now() - pd.to_datetime(position.time, unit='s')
                
                # Get current market data
                signal = await self.get_combined_signal(position.symbol)
                
                # Check the need to modify the position
                if self._should_modify_position(position, signal, time_in_trade):
                    await self._modify_position(position, signal)
                    
                # Check the closing conditions
                if self._should_close_position(position, signal, time_in_trade):
                    await self.close_position(position)
                    
    except Exception as e:
        logger.error(f"Error managing positions: {e}")

Aquí prestamos especial atención al tiempo en la posición. La práctica ha demostrado que cuanto más tiempo esté abierta una posición, mayores deberán ser los requisitos para su mantenimiento. Esto se logra mediante el endurecimiento dinámico de las condiciones de mantenimiento de la posición a lo largo del tiempo.

Un punto interesante tiene que ver con el cierre parcial de posiciones:

def calculate_partial_close(self, position, profit_threshold: float = 0.5) -> float:
    """Volume calculation for partial closure"""
    try:
        # Check the current profit
        if position.profit <= 0:
            return 0.0
            
        profit_ratio = position.profit / (position.volume * 1000)  # approximate ROI estimate
        
        if profit_ratio >= profit_threshold:
            # Close half of the position when the profit threshold is reached
            return position.volume * 0.5
        return 0.0
        
    except Exception as e:
        logger.error(f"Error calculating partial close: {e}")
        return 0.0

Esta característica surgió después de analizar las transacciones. He observado que el cierre parcial de posiciones cuando se alcanza un determinado nivel de beneficios mejora sustancialmente las estadísticas comerciales generales. Esto le permite fijar algunos beneficios, dejando al mismo tiempo la posibilidad de nuevas subidas.

En conclusión, el sistema de gestión de riesgos es un organismo vivo en constante evolución. Cada transacción fallida, cada movimiento inesperado del mercado supone una nueva experiencia que utilizamos para mejorar nuestros algoritmos de protección del capital. En las próximas versiones del sistema tenemos previsto añadir el aprendizaje automático para optimizar dinámicamente los parámetros de gestión del riesgo, así como un híbrido del sistema VaR y la teoría de portafolios de Markowitz, pero esto es otra historia......

Módulo económico: cuando el análisis fundamental se encuentra con al aprendizaje automático

Mientras trabajaba en el sistema comercial, noté un patrón interesante: incluso las señales técnicas más fuertes pueden fallar si contradicen los factores fundamentales. Fue esta observación la que me llevó a crear el módulo económico, un componente que analiza los indicadores macroeconómicos y su repercusión en el movimiento de los pares de divisas.

Vamos a empezar por la estructura básica del módulo. Esta será la inicialización de los principales indicadores económicos:

def __init__(self):
    self.indicators = {
        'NY.GDP.MKTP.KD.ZG': 'GDP growth',
        'FP.CPI.TOTL.ZG': 'Inflation',
        'FR.INR.RINR': 'Real interest rate',
        'NE.EXP.GNFS.ZS': 'Exports',
        'NE.IMP.GNFS.ZS': 'Imports',
        'BN.CAB.XOKA.GD.ZS': 'Current account balance',
        'GC.DOD.TOTL.GD.ZS': 'Government debt',
        'SL.UEM.TOTL.ZS': 'Unemployment rate',
        'NY.GNP.PCAP.CD': 'GNI per capita',
        'NY.GDP.PCAP.KD.ZG': 'GDP per capita growth'
    }

La elección de estos indicadores no ha sido casual. Tras analizar miles de transacciones, he notado que estos son los indicadores que más influyen en las tendencias a largo plazo de los pares de divisas. La relación entre el tipo de interés real y los movimientos de las divisas resulta especialmente interesante: a menudo, un cambio en este indicador precede a un cambio de tendencia.

He desarrollado un método especial para obtener los datos económicos:

def fetch_economic_data(self):
    data_frames = []
    for indicator, name in self.indicators.items():
        try:
            data_frame = wbdata.get_dataframe({indicator: name}, country='all')
            data_frames.append(data_frame)
        except Exception as e:
            logger.error(f"Error fetching data for indicator '{indicator}': {e}")

    if data_frames:
        self.economic_data = pd.concat(data_frames, axis=1)
        return self.economic_data

Un punto interesante aquí es el uso de la biblioteca wbdata para recuperar los datos del Banco Mundial. Elegí esta fuente tras experimentar con varias API, ya que ofrece los datos más completos y verificados.

Presté especial atención a la preparación de los datos para el análisis:

def prepare_data(self, symbol_data):
    data = symbol_data.copy()
    data['close_diff'] = data['close'].diff()
    data['close_corr'] = data['close'].rolling(window=30).corr(data['close'].shift(1))

    for indicator in self.indicators.keys():
        if indicator in self.economic_data.columns:
            data[indicator] = self.economic_data[indicator].ffill()

    data.dropna(inplace=True)
    return data

Obsérvese el uso del forward fill para los indicadores económicos. Esta solución no surgió de inmediato: al principio probé con la interpolación, pero resultó que para los datos económicos es más correcto usar el último valor conocido.

El corazón del módulo es el sistema de predicción:

def forecast(self, symbol, symbol_data):
    if len(symbol_data) < 50:
        return None, None

    X = symbol_data.drop(columns=['close'])
    y = symbol_data['close']

    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)

    model = CatBoostRegressor(iterations=1000, learning_rate=0.1, depth=8, loss_function='RMSE')
    model.fit(X_train, y_train, verbose=False)

La elección de CatBoost como algoritmo de aprendizaje automático tampoco resulta algo casual. Tras experimentar con varios modelos (desde la regresión lineal simple hasta complejas redes neuronales), resultó que CatBoost es el que mejor maneja la naturaleza irregular de los datos económicos.

La última etapa es la interpretación de los resultados:

def interpret_results(self, symbol):
    forecast = self.forecasts.get(symbol)
    importance_df = self.feature_importances.get(symbol)

    if forecast is None or importance_df is None:
        return f"Insufficient data for interpretation of {symbol}"

    trend = "upward" if forecast[-1] > forecast[0] else "downward"
    volatility = "high" if forecast.std() / forecast.mean() > 0.1 else "low"
    top_feature = importance_df.iloc[0]['feature']

El cálculo de la volatilidad resulta especialmente interesante. Tras analizar los datos históricos, elegimos un umbral de 0,1 para definir la alta volatilidad, pues este valor separa bien los periodos de calma del mercado y los de aumento de las turbulencias.

Mientras trabajaba en el módulo, observé algo interesante: los factores económicos suelen actuar con retraso, pero su influencia es más persistente que la de los factores técnicos. Esto ha dado lugar a un sistema de ponderación en el que la importancia de las señales económicas aumenta en marcos temporales más largos.

Obviamente, el módulo económico no es una varita mágica y no puede predecir todos los movimientos del mercado. Pero cuando se combina con el análisis técnico y de volumen, ofrece una dimensión adicional a la comprensión de los procesos del mercado. En las próximas versiones del sistema hemos previsto añadir el análisis de los flujos de noticias y su impacto en los indicadores económicos, pero este es un tema aparte.

Módulo de arbitraje: en busca de la verdad en el precio

La idea de crear un módulo de arbitraje se me ocurrió tras estudiar largamente el mercado de divisas. Observé un patrón interesante: a menudo los precios reales de los pares de divisas se desvían de su valor teórico calculado mediante tipos cruzados. Estas desviaciones crean oportunidades de arbitraje, pero lo más importante es que pueden servir como indicador de la evolución futura de los precios.

Empezaremos por la estructura básica del módulo:

class ArbitrageModule:
    def __init__(self, terminal_path: str = "C:/Program Files/RannForex MetaTrader 5/terminal64.exe", max_trades: int = 10):
        self.terminal_path = terminal_path
        self.MAX_OPEN_TRADES = max_trades
        self.symbols = [
            "AUDUSD.ecn", "AUDJPY.ecn", "CADJPY.ecn", "AUDCHF.ecn", "AUDNZD.ecn", 
            "USDCAD.ecn", "USDCHF.ecn", "USDJPY.ecn", "NZDUSD.ecn", "GBPUSD.ecn", 
            "EURUSD.ecn", "CADCHF.ecn", "CHFJPY.ecn", "NZDCAD.ecn", "NZDCHF.ecn", 
            "NZDJPY.ecn", "GBPCAD.ecn", "GBPCHF.ecn", "GBPJPY.ecn", "GBPNZD.ecn", 
            "EURCAD.ecn", "EURCHF.ecn", "EURGBP.ecn", "EURJPY.ecn", "EURNZD.ecn"
        ]

Debemos prestar especial atención a la estructura de los pares básicos para calcular los tipos cruzados:

     self.usd_pairs = {
            "EUR": "EURUSD.ecn",
            "GBP": "GBPUSD.ecn", 
            "AUD": "AUDUSD.ecn",
            "NZD": "NZDUSD.ecn",
            "USD": None,
            "CAD": ("USDCAD.ecn", True),
            "CHF": ("USDCHF.ecn", True),
            "JPY": ("USDJPY.ecn", True)
        }

Aquí hay un punto interesante: algunos pares están etiquetados como inversos (True). No es casualidad: para algunas divisas como CAD, CHF y JPY, la cotización básica será USD/XXX en lugar de XXX/USD. Este es un matiz importante que con frecuencia se pasa por alto al calcular los tipos cruzados.

El corazón del módulo es la función de cálculo sintético de precios:

def calculate_synthetic_prices(self, data: Dict[str, pd.DataFrame]) -> pd.DataFrame:
    """Calculation of synthetic prices through cross rates"""
    synthetic_prices = {}
    
    try:
        for symbol in self.symbols:
            base = symbol[:3]
            quote = symbol[3:6]
            
            # Calculate the synthetic price using cross rates
            fair_price = self.calculate_cross_rate(base, quote, data)
            synthetic_prices[f'{symbol}_fair'] = pd.Series([fair_price])

Recuerdo cuánto luché para optimizar este código. Al principio, traté de calcular todas las rutas de conversión posibles y elegir la mejor. Pero resulta que un cálculo simple a través de USD ofrece resultados más estables, especialmente en entornos de alta volatilidad.

También es interesante la función de cálculo del tipo de cambio a USD:

def get_usd_rate(self, currency: str, data: dict) -> float:
    """Get exchange rate to USD"""
    if currency == "USD":
        return 1.0
        
    pair_info = self.usd_pairs[currency]
    if isinstance(pair_info, tuple):
        pair, inverse = pair_info
        rate = data[pair]['close'].iloc[-1]
        return 1 / rate if inverse else rate
    else:
        pair = pair_info
        return data[pair]['close'].iloc[-1]

Esta función surgió tras muchos experimentos con distintas formas de calcular los tipos cruzados. La clave aquí es el procesamiento adecuado de los pares inversos. Un cálculo incorrecto (incluso en un solo par) podría provocar una cascada de errores en los precios sintéticos.

Así, he desarrollado una función especial para procesar datos reales:

def get_mt5_data(self, symbol: str, count: int = 1000) -> Optional[pd.DataFrame]:
    try:
        timezone = pytz.timezone("Etc/UTC")
        utc_from = datetime.now(timezone) - timedelta(days=1)

        ticks = mt5.copy_ticks_from(symbol, utc_from, count, mt5.COPY_TICKS_ALL)
        if ticks is None:
            logger.error(f"Failed to fetch data for {symbol}")
            return None

        ticks_frame = pd.DataFrame(ticks)
        ticks_frame['time'] = pd.to_datetime(ticks_frame['time'], unit='s')
        return ticks_frame

La elección del número de ticks (1000) supone un compromiso entre la precisión del cálculo y la velocidad de procesamiento de los datos. En la práctica, esto ha demostrado bastar para determinar con fiabilidad un precio justo.

Mientras trabajaba en el módulo, observé algo interesante: las discrepancias entre los precios reales y los sintéticos suelen aparecer antes de que se produzcan movimientos significativos en el mercado. Es como si el dinero inteligente empezara a mover algunos pares, creando tensión en el sistema de tipos cruzados, que luego se desactiva con un movimiento intenso.

Claro que el módulo de arbitraje no es una varita mágica, pero cuando se combina con el análisis de volumen y los indicadores económicos, aporta una dimensión adicional a la comprensión del mercado. En próximas versiones, tenemos la intención de añadir el análisis de las correlaciones entre las desviaciones en diferentes pares, pero eso es otra historia.

Conclusión

Cuando empecé este proyecto, no tenía ni idea de en qué se convertiría. Pensé que simplemente combinaría Python con MQL5 y se acabaría todo. ¡Y obtuve toda una plataforma comercial! Cada trocito es como una pieza en un reloj suizo, y este artículo es solo la primera de muchas.

He aprendido muchas cosas durante el desarrollo. Por ejemplo, que no hay caminos fáciles en el trading algorítmico. Tomemos por ejemplo el cálculo del volumen de las posiciones: uno diría que no tiene nada de complicado. Pero si consideramos todos los riesgos y el comportamiento del mercado, la cabeza le da a uno vueltas.

¡Y qué bien funciona la arquitectura modular! Si falla un módulo, los demás siguen operativos. Podemos mejorar tranquilamente cada parte sin miedo a romper todo el sistema.

Lo divertido es ver cómo funcionan las distintas partes del sistema. Un módulo busca el arbitraje, otro supervisa los volúmenes, el tercero analiza la economía y el cuarto controla el riesgo. Juntos, ven el mercado de una forma que ningún análisis individual puede.

Obviamente, aún hay margen para crecer. Querría añadir el análisis de noticias, mejorar el aprendizaje automático y desarrollar nuevos modelos de evaluación de riesgos. Resulta especialmente interesante trabajar en la visualización de mercados en 3D, presentando el precio, el volumen y el tiempo en el mismo espacio.

La principal lección de este proyecto es que el sistema comercial debe estar vivo. El mercado no se detiene y el sistema debe cambiar con él. Así que hay que aprender de los errores, encontrar nuevas pautas y descartar planteamientos anticuados.

Espero que mi experiencia sea útil a quienes también crean algoritmos comerciales. Y recuerde, en este negocio no hay línea de meta. ¡Solo existe el camino!