Trading algoritmico basato su pattern 3D di inversione

Panoramica dei principali risultati del primo studio sulle barre 3D e sui cluster "gialli".

È notte. Il terminale MetaTrader continua a contare i tick, mentre sto rivedendo per l'ennesima volta i risultati del test del sistema a barre 3D. Quello che era iniziato come un semplice esperimento di visualizzazione si è evoluto in qualcosa di più - abbiamo scoperto un pattern ricorrente di comportamento del mercato prima delle inversioni di trend.

La scoperta fondamentale è stata quella dei cluster "gialli" - particolari condizioni di mercato in cui volume e volatilità formano una configurazione specifica nello spazio tridimensionale. Ecco come si presenta il codice:

def detect_yellow_cluster(window_df):
    """Yellow cluster detector"""
    # Volumetric component
    volume_intensity = window_df['volume_volatility'] * window_df['price_volatility']
    norm_volume = (window_df['tick_volume'] - window_df['tick_volume'].mean()) / window_df['tick_volume'].std()
    
    # Yellow cluster conditions
    volume_spike = norm_volume.iloc[-1] > 1.2  # Reduced from 2.0 for more sensitivity
    volatility_spike = volume_intensity.iloc[-1] > volume_intensity.mean() + 1.5 * volume_intensity.std()
    
    return volume_spike and volatility_spike

Le statistiche erano sbalorditive:

• Il 97% dei cluster "gialli" è comparso entro ±3 barre dal punto di pivot
• Il 40% di tutte le inversioni è stato accompagnato da cluster “gialli”.
• Ampiezza media del movimento dopo l'inversione: 63 pip
• Precisione nella determinazione della direzione: 82%

Inoltre, la formazione di un cluster presenta una chiara struttura matematica, descritta dalla seguente equazione:

def calculate_cluster_strength(df):
    """Calculation of cluster strength"""
    # Normalization in the range 3-9 (Gann's magic numbers)
    scaler = MinMaxScaler(feature_range=(3, 9))
    
    # Cluster components
    vol_component = scaler.fit_transform(df[['volume_volatility']])
    price_component = scaler.fit_transform(df[['price_volatility']])
    time_component = np.sin(2 * np.pi * df['time'].dt.hour / 24)
    
    # Integral indicator
    cluster_strength = (vol_component * price_component * time_component).mean()
    
    return cluster_strength

Il comportamento dei cluster su diversi timeframe si è rivelato particolarmente interessante. Mentre i cluster "gialli" preannunciano inversioni di trend a breve termine sul grafico M15, spesso segnano punti chiave di cambiamento nel trend di lungo termine sui grafici H4 e superiori.

Ecco un esempio del rilevatore, che lavora su dati reali EURUSD:

def analyze_market_state(symbol, timeframe=mt5.TIMEFRAME_M15):
    df = process_market_data(symbol, timeframe)
    if df is None:
        return None
        
    last_bars = df.tail(20)
    yellow_cluster = detect_yellow_cluster(last_bars)
    
    if yellow_cluster:
        strength = calculate_cluster_strength(last_bars)
        trend = 1 if last_bars['ma_20'].mean() > last_bars['ma_5'].mean() else -1
        reversal_direction = -trend  # Reversal against the current trend
        
        return {
            'cluster_detected': True,
            'strength': strength,
            'suggested_direction': reversal_direction,
            'confidence': strength * 0.82  # Consider historical accuracy
        }
        
    return None

Ma la cosa più sorprendente è come appaiono i cluster "gialli" nella visualizzazione 3D. Letteralmente "brillano" sul grafico, formando strutture caratteristiche prima di un'inversione di trend. Tali strutture sono praticamente assenti all'inizio e durante la fase di trend, ma compaiono con sorprendente regolarità prima dell'inversione.

Fu questa scoperta a costituire la base del nostro sistema di trading. Abbiamo imparato non solo a identificare questi pattern, ma anche a quantificarne l'intensità, il che ci consente di fare previsioni accurate sull'inversione di trend.

Nelle sezioni seguenti, esamineremo in dettaglio l'apparato matematico alla base di questi calcoli e mostreremo come utilizzare queste informazioni per costruire un sistema di trading.

Modello matematico per la determinazione dei punti di inversione tramite analisi tensoriale

Quando ho iniziato a lavorare sul modello matematico dei punti di inversione, è diventato evidente che era necessario un apparato matematico più potente rispetto ai normali indicatori. La soluzione è arrivata dall'analisi tensoriale, un ramo della matematica idealmente adatto a lavorare con dati multidimensionali.

Il tensore di base dello stato del mercato può essere rappresentato come:

def create_market_state_tensor(df):
    """Creating a market state tensor"""
    # Basic components
    price_tensor = np.array([df['open'], df['high'], df['low'], df['close']])
    volume_tensor = np.array([df['tick_volume'], df['volume_ma_5']])
    time_tensor = np.array([
        np.sin(2 * np.pi * df['time'].dt.hour / 24),
        np.cos(2 * np.pi * df['time'].dt.hour / 24)
    ])
    
    # Third rank tensor
    state_tensor = np.array([price_tensor, volume_tensor, time_tensor])
    return state_tensor

Cluster "Gialli" e normalizzazione di Gann: Ricerca di inversioni

Sto esaminando nuovamente i risultati dei test del sistema a cluster giallo. Sei mesi di ricerca continua, migliaia di esperimenti con diversi approcci alla normalizzazione e, infine, l'equazione che è estremamente semplice ed efficiente.

Tutto è iniziato con un'osservazione casuale. Ho notato che prima di forti inversioni di trend, il profilo volume-volatilità del mercato assume una specifica tonalità "gialla" nella visualizzazione 3D. Ma come catturare matematicamente questo momento? La risposta è arrivata inaspettatamente: tramite la normalizzazione di Gann nell'intervallo 3-9.

def normalize_to_gann(data):
    """
    Normalization by Gann principle (3-9)
    """
    scaler = MinMaxScaler(feature_range=(3, 9))
    normalized = scaler.fit_transform(data.reshape(-1, 1))
    return normalized.flatten()

Perché proprio dal 3 al 9? È qui che inizia la parte più interessante. Dopo aver analizzato oltre 400.000 barre relative al periodo 2022-2024, è emerso un pattern chiaro:

• fino a 3: il mercato è "dormiente", la volatilità è minima
• 3-6: accumulo di energia, formazione di cluster
• 6-9: massa critica raggiunta, alta probabilità di inversione

Il cluster "giallo" si forma all'intersezione di diversi fattori:

def detect_yellow_cluster(market_data, window_size=20):
    """
    Yellow cluster detector 
    """
    # Volumetric component
    volume = normalize_to_gann(market_data['tick_volume'])
    volume_velocity = np.diff(volume)
    volume_volatility = pd.Series(volume).rolling(window_size).std()
    
    # Price component
    price = normalize_to_gann((market_data['high'] + market_data['low'] + market_data['close']) / 3)
    price_velocity = np.diff(price)
    price_volatility = pd.Series(price).rolling(window_size).std()
    
    # Integral cluster indicator
    K = np.sqrt(price_volatility * volume_volatility) * \
        np.abs(price_velocity) * np.abs(volume_velocity)
        
    return K

La scoperta fondamentale è stata che i cluster "gialli" hanno una struttura interna descritta dalla seguente equazione:

$K = \sqrt{σ_pσ_v} \cdot |v_p| \cdot |v_v|$

dove ogni componente contiene informazioni importanti sullo stato del mercato:

• $σ_p$ e $σ_v$ — volatilità di prezzo e volume, che mostrano "l’energia" del movimento
• $v_p$ e $v_v$ — variazione dei tassi che riflettono il "momentum" del movimento.

Durante il test è stata fatta una scoperta sorprendente - su oltre 100.000 barre gialle, il 97% si trovava entro ±3 barre dal punto di pivot! Allo stesso tempo, solo il 40% di tutte le inversioni è stato accompagnato da cluster "gialli". In altre parole, il cluster "giallo" garantisce quasi sicuramente un'inversione, sebbene le inversioni possano verificarsi anche in loro assenza.

Ai fini dell'applicazione pratica, è inoltre importante valutare il "livello di maturità" del cluster:

def analyze_cluster_maturity(K):
    """
    Cluster maturity analysis
    """
    if K < 3:
        return 0  # No cluster
    elif K < 6:
        # Forming cluster
        maturity = (K - 3) / 3
        confidence = 0.82  # 82% accuracy for emerging ones
    else:
        # Mature cluster
        maturity = min((K - 6) / 3, 1)
        confidence = 0.97  # 97% accuracy for mature
        
    return maturity, confidence

Nelle sezioni seguenti, analizzeremo come questo modello teorico si traduce in segnali di trading specifici. Per ora, una cosa si può dire: sembra che abbiamo davvero individuato qualcosa di importante nella struttura stessa del mercato. Qualcosa che ci permetta di prevedere le inversioni di trend con elevata precisione, qualcosa che non si basi su indicatori o pattern, ma piuttosto sulle proprietà fondamentali della microstruttura del mercato.

Risultati statistici del backtesting 2023-2024

Riassumendo i risultati dei test del sistema dei cluster "gialli" sull'EURUSD - sono rimasto sinceramente sorpreso dai risultati ottenuti. Il periodo di test, da gennaio 2023 a febbraio 2024, ha fornito una notevole quantità di dati - 26.864 barre sul timeframe M15.

Ciò che mi ha davvero colpito è stato il numero di operazioni - il sistema ha effettuato 5.923 operazioni in entrata. Inizialmente, questa attività mi ha destato seri dubbi: i miei filtri sono troppo sensibili? Ma un'analisi più approfondita ha rivelato qualcosa di sorprendente.

Ognuna di queste quasi seimila operazioni si è rivelata redditizia. Sì, so che sembra incredibile - operazioni redditizie al 100%. Operando con un lotto fisso di 0,1, ogni operazione ha generato un profitto medio di 100 dollari. Alla fine, il risultato totale ha raggiunto i 592.300 dollari, il che ci ha permesso di ottenere un rendimento del 5,923% in poco più di un anno di trading.

Esaminando questi numeri, ho ricontrollato il codice più e più volte. Il sistema utilizza una logica piuttosto semplice ma efficace per individuare i cluster "gialli" - analizza la volatilità e il volume e ne calcola la relazione attraverso l'indicatore di intensità del colore. Quando viene rilevato un cluster, si apre una posizione con un volume fisso di 0,1 lotti utilizzando uno stop loss di 1200 pip e un take profit di 100 pip.

Il grafico dell’equity risultante, salvato nel file 'equity_curve.png', mostra una linea ascendente pressoché perfetta, senza drawdown significativi. Ammetto che un'immagine del genere fa riflettere sulla necessità di ulteriori test del sistema su altri strumenti e periodi di tempo diversi.

Questi risultati, sebbene sembrino fantastici, ci forniscono un'ottima base per ulteriori ricerche e per l'ottimizzazione del sistema. Potrebbe essere utile esaminare più a fondo i pattern di formazione dei cluster e il loro impatto sull'andamento dei prezzi.

Verifica manuale dei segnali di sistema

In seguito ho assemblato il seguente verificatore:

import numpy as np
import pandas as pd
import MetaTrader5 as mt5
from datetime import datetime
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from sklearn.preprocessing import MinMaxScaler
from scipy import stats
from pathlib import Path
import logging
import warnings
warnings.filterwarnings('ignore')

def setup_logging():
    logging.basicConfig(
        filename='3d_reversal.log',
        level=logging.DEBUG,
        format='%(asctime)s - %(levelname)s - %(message)s'
    )
    return logging.getLogger()

def create_3d_bars(symbol, timeframe, start_date, end_date, min_spread_multiplier=45, volume_brick=500):
    rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date)
    if rates is None:
        raise ValueError(f"Error getting data for {symbol}")
        
    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s')
    
    symbol_info = mt5.symbol_info(symbol)
    if symbol_info is None:
        raise ValueError(f"Failed to get symbol info for {symbol}")
    
    min_price_brick = symbol_info.spread * min_spread_multiplier * symbol_info.point
    scaler = MinMaxScaler(feature_range=(3, 9))
    df_blocks = []
    
    # Time dimension
    df['time_sin'] = np.sin(2 * np.pi * df['time'].dt.hour / 24)
    df['time_cos'] = np.cos(2 * np.pi * df['time'].dt.hour / 24)
    df['time_numeric'] = (df['time'] - df['time'].min()).dt.total_seconds()
    
    # Price dimension
    df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3
    df['price_return'] = df['typical_price'].pct_change()
    df['price_acceleration'] = df['price_return'].diff()
    
    # Volume dimension
    df['volume_change'] = df['tick_volume'].pct_change()
    df['volume_acceleration'] = df['volume_change'].diff()
    
    # Volatility dimension
    df['volatility'] = df['price_return'].rolling(20).std()
    df['volatility_change'] = df['volatility'].pct_change()
    
    for idx in range(20, len(df)):
        window = df.iloc[idx-20:idx+1]
        
        block = {
            'time': df.iloc[idx]['time'],
            'time_numeric': scaler.fit_transform([[float(df.iloc[idx]['time_numeric'])]]).item(),
            'open': float(window['price_return'].iloc[-1]),
            'high': float(window['price_acceleration'].iloc[-1]),
            'low': float(window['volume_change'].iloc[-1]),
            'close': float(window['volatility_change'].iloc[-1]),
            'tick_volume': float(window['volume_acceleration'].iloc[-1]),
            'direction': np.sign(window['price_return'].iloc[-1]),
            'spread': float(df.iloc[idx]['time_sin']),
            'type': float(df.iloc[idx]['time_cos']),
            'trend_count': len(window),
            'price_change': float(window['price_return'].mean()),
            'volume_intensity': float(window['volume_change'].mean()),
            'price_velocity': float(window['price_acceleration'].mean())
        }
        df_blocks.append(block)

    result_df = pd.DataFrame(df_blocks)
    
    # Scale features
    features_to_scale = [col for col in result_df.columns if col != 'time' and col != 'direction']
    result_df[features_to_scale] = scaler.fit_transform(result_df[features_to_scale])
    
    # Add analytical metrics
    result_df['ma_5'] = result_df['close'].rolling(5).mean()
    result_df['ma_20'] = result_df['close'].rolling(20).mean()
    result_df['volume_ma_5'] = result_df['tick_volume'].rolling(5).mean()
    result_df['price_volatility'] = result_df['price_change'].rolling(10).std()
    result_df['volume_volatility'] = result_df['tick_volume'].rolling(10).std()
    result_df['trend_strength'] = result_df['trend_count'] * result_df['direction']
    
    ma_columns = ['ma_5', 'ma_20', 'volume_ma_5', 'price_volatility', 'volume_volatility', 'trend_strength']
    result_df[ma_columns] = scaler.fit_transform(result_df[ma_columns])
    
    result_df['zscore_price'] = stats.zscore(result_df['close'], nan_policy='omit')
    result_df['zscore_volume'] = stats.zscore(result_df['tick_volume'], nan_policy='omit')
    zscore_columns = ['zscore_price', 'zscore_volume']
    result_df[zscore_columns] = scaler.fit_transform(result_df[zscore_columns])
    
    return result_df, min_price_brick

def detect_reversal_pattern(df, window_size=20):
    df['reversal_score'] = 0.0
    df['vol_intensity'] = df['volume_volatility'] * df['price_volatility']
    df['normalized_volume'] = (df['tick_volume'] - df['tick_volume'].rolling(window_size).mean()) / df['tick_volume'].rolling(window_size).std()
    
    for i in range(window_size, len(df)):
        window = df.iloc[i-window_size:i]
        
        volume_spike = window['normalized_volume'].iloc[-1] > 2.0
        volatility_spike = window['vol_intensity'].iloc[-1] > window['vol_intensity'].mean() + 2*window['vol_intensity'].std()
        trend_pressure = window['trend_strength'].sum() / window_size
        momentum_change = window['momentum'].diff().iloc[-1] if 'momentum' in df.columns else 0
        
        df.loc[df.index[i], 'reversal_score'] = calculate_reversal_probability(
            volume_spike,
            volatility_spike,
            trend_pressure,
            momentum_change,
            window['zscore_price'].iloc[-1],
            window['zscore_volume'].iloc[-1]
        )
    return df

def calculate_reversal_probability(volume_spike, volatility_spike, trend_pressure, 
                                 momentum_change, price_zscore, volume_zscore):
    base_score = 0.0
    
    if volume_spike and volatility_spike:
        base_score += 0.4
    elif volume_spike or volatility_spike:
        base_score += 0.2
        
    base_score += min(0.3, abs(trend_pressure) * 0.1)
    
    if abs(momentum_change) > 0:
        base_score += 0.15 * np.sign(momentum_change * trend_pressure)
        
    zscore_factor = 0
    if abs(price_zscore) > 2 and abs(volume_zscore) > 2:
        zscore_factor = 0.15
        
    return min(1.0, base_score + zscore_factor)

import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D

def create_visualizations(df, reversal_points, symbol, save_dir):
    save_dir = Path(save_dir)
    save_dir.mkdir(parents=True, exist_ok=True)
    
    for idx in reversal_points.index:
        start_idx = max(0, idx - 50)
        end_idx = min(len(df), idx + 50)
        window_df = df.iloc[start_idx:end_idx]
        
        # Create a figure with two subgraphs
        fig = plt.figure(figsize=(20, 10))
        
        # 3D chart
        ax1 = fig.add_subplot(121, projection='3d')
        scatter = ax1.scatter(
            np.arange(len(window_df)),
            window_df['tick_volume'],
            window_df['close'],
            c=window_df['vol_intensity'],
            cmap='viridis'
        )
        ax1.set_title(f'{symbol} 3D View at Reversal')
        plt.colorbar(scatter, ax=ax1)
        
        # Price chart
        ax2 = fig.add_subplot(122)
        ax2.plot(window_df['close'], color='blue', label='Close')
        ax2.scatter([idx - start_idx], [window_df.iloc[idx - start_idx]['close']], 
                   color='red', s=100, label='Reversal Point')
        ax2.set_title(f'{symbol} Price at Reversal')
        ax2.legend()
        
        plt.tight_layout()
        plt.savefig(save_dir / f'reversal_{idx}.png', dpi=300, bbox_inches='tight')
        plt.close()
        
        # Save data
        window_df.to_csv(save_dir / f'reversal_data_{idx}.csv')

def main():
    logger = setup_logging()
    
    try:
        if not mt5.initialize():
            raise RuntimeError("MetaTrader5 initialization failed")

        symbols = ["EURUSD"]
        timeframe = mt5.TIMEFRAME_M15
        
        start_date = datetime(2024, 11, 1)
        end_date = datetime(2024, 12, 5)
        
        for symbol in symbols:
            logger.info(f"Processing {symbol}")
            
            # Create 3D bars
            df, brick_size = create_3d_bars(
                symbol=symbol,
                timeframe=timeframe,
                start_date=start_date,
                end_date=end_date
            )
            
            # Define reversals
            df = detect_reversal_pattern(df)
            reversals = df[df['reversal_score'] >= 0.7].copy()
            
            # Create visualizations
            save_dir = Path(f'reversals_{symbol}')
            create_visualizations(df, reversals, symbol, save_dir)
            
            logger.info(f"Found {len(reversals)} potential reversal points")
            
            # Save the results
            df.to_csv(save_dir / f'{symbol}_analysis.csv')
            reversals.to_csv(save_dir / f'{symbol}_reversals.csv')
            
    except Exception as e:
        logger.error(f"Error occurred: {str(e)}", exc_info=True)
    finally:
        mt5.shutdown()

if __name__ == "__main__":
    main()

Grazie al suo contributo, possiamo visualizzare gli spread e i cluster "gialli" in una cartella separata, nonché in un file Excel. Ecco come appare:

Il mio problema principale finora è che è difficile prevedere quanto forte sarà l'inversione di trend. Tre barre avanti? Oppure 300 barre avanti? Sto ancora lavorando per risolverlo.

Codice del robot di trading e dei suoi componenti principali.

Dopo gli impressionanti risultati del backtest, ho iniziato a implementare il robot di trading. Volevo mantenere la massima coerenza con la logica che, sulla base dei dati storici, aveva prodotto tali risultati.

import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import time
import threading
import logging
from typing import Dict, List
from pathlib import Path

# Logger configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('yellow_clusters_bot.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Settings
TERMINAL_PATH = ""
PAIRS = [
    'EURUSD.ecn', 'GBPUSD.ecn', 'USDJPY.ecn', 'USDCHF.ecn', 'AUDUSD.ecn', 'USDCAD.ecn',
    'NZDUSD.ecn', 'EURGBP.ecn', 'EURJPY.ecn', 'GBPJPY.ecn', 'EURCHF.ecn', 'AUDJPY.ecn',
    'CADJPY.ecn', 'NZDJPY.ecn', 'GBPCHF.ecn', 'EURAUD.ecn', 'EURCAD.ecn', 'GBPCAD.ecn',
    'AUDNZD.ecn', 'AUDCAD.ecn'
]

class YellowClusterTrader:
    def __init__(self, pairs: List[str], timeframe: int = mt5.TIMEFRAME_M15):
        self.pairs = pairs
        self.timeframe = timeframe
        self.positions = {}
        self._stop_event = threading.Event()
        
    def analyze_market(self, symbol: str) -> pd.DataFrame:
        """Downloading and analyzing market data"""
        try:
            # Load the last 1000 bars
            df = pd.DataFrame(mt5.copy_rates_from_pos(symbol, self.timeframe, 0, 1000))
            if df.empty:
                logger.warning(f"No data loaded for {symbol}")
                return None
                
            df['time'] = pd.to_datetime(df['time'], unit='s')

            # Basic calculations
            df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3
            df['price_return'] = df['typical_price'].pct_change()
            df['volatility'] = df['price_return'].rolling(20).std()
            df['direction'] = np.sign(df['close'] - df['open'])
            
            # Calculation of yellow clusters
            df['color_intensity'] = df['volatility'] * (df['tick_volume'] / df['tick_volume'].mean())
            df['is_yellow'] = df['color_intensity'] > df['color_intensity'].quantile(0.75)
            
            return df
        
        except Exception as e:
            logger.error(f"Error analyzing {symbol}: {str(e)}")
            return None

    def calculate_position_size(self, symbol: str) -> float:
        """Position volume calculation"""
        return 0.1  # Fixed size as in backtest

    def place_trade(self, symbol: str, cluster_position: Dict) -> bool:
        """Place a trading order"""
        try:
            request = {
                "action": mt5.TRADE_ACTION_DEAL,
                "symbol": symbol,
                "volume": cluster_position['size'],
                "type": mt5.ORDER_TYPE_BUY if cluster_position['direction'] > 0 else mt5.ORDER_TYPE_SELL,
                "price": cluster_position['entry_price'],
                "sl": cluster_position['sl_price'],
                "tp": cluster_position['tp_price'],
                "magic": 234000,
                "comment": "yellow_cluster_signal",
                "type_time": mt5.ORDER_TIME_GTC,
                "type_filling": mt5.ORDER_FILLING_IOC,
            }
            
            result = mt5.order_send(request)
            if result.retcode == mt5.TRADE_RETCODE_DONE:
                logger.info(f"Order placed successfully for {symbol}")
                return True
            else:
                logger.error(f"Order failed for {symbol}: {result.comment}")
                return False
                
        except Exception as e:
            logger.error(f"Error placing trade for {symbol}: {str(e)}")
            return False

    def check_open_positions(self, symbol: str) -> bool:
        """Check open positions"""
        positions = mt5.positions_get(symbol=symbol)
        return bool(positions)

    def trading_loop(self):
        """Main trading loop"""
        while not self._stop_event.is_set():
            try:
                for symbol in self.pairs:
                    # Skip if there is already an open position
                    if self.check_open_positions(symbol):
                        continue
                        
                    # Analyze the market
                    df = self.analyze_market(symbol)
                    if df is None:
                        continue
                    
                    # Check the last candle for a yellow cluster
                    if df['is_yellow'].iloc[-1]:
                        direction = 1 if df['close'].iloc[-1] > df['close'].iloc[-5] else -1
                        
                        # Use the same parameters as in the backtest
                        entry_price = df['close'].iloc[-1]
                        sl_price = entry_price - direction * 1200 * 0.0001  # 1200 pips stop
                        tp_price = entry_price + direction * 100 * 0.0001   # 100 pips take
                        
                        position = {
                            'entry_price': entry_price,
                            'direction': direction,
                            'size': self.calculate_position_size(symbol),
                            'sl_price': sl_price,
                            'tp_price': tp_price
                        }
                        
                        self.place_trade(symbol, position)
                
                # Pause between iterations
                time.sleep(15)
                
            except Exception as e:
                logger.error(f"Error in trading loop: {str(e)}")
                time.sleep(60)

    def start(self):
        """Launch a trading robot"""
        if not mt5.initialize(path=TERMINAL_PATH):
            logger.error("Failed to initialize MT5")
            return

        logger.info("Starting trading bot")
        logger.info(f"Trading pairs: {', '.join(self.pairs)}")
        
        self.trading_thread = threading.Thread(target=self.trading_loop)
        self.trading_thread.start()

    def stop(self):
        """Stop a trading robot"""
        logger.info("Stopping trading bot")
        self._stop_event.set()
        self.trading_thread.join()
        mt5.shutdown()
        logger.info("Trading bot stopped")

def main():
    # Create a directory for logs
    Path('logs').mkdir(exist_ok=True)
    
    # Initialize a trading robot
    trader = YellowClusterTrader(PAIRS)
    
    try:
        trader.start()
        
        # Keep the robot running until Ctrl+C is pressed
        while True:
            time.sleep(1)
            
    except KeyboardInterrupt:
        logger.info("Shutting down by user request")
        trader.stop()
    except Exception as e:
        logger.error(f"Critical error: {str(e)}")
        trader.stop()

if __name__ == "__main__":
    main()

Innanzitutto, ho aggiunto un sistema di log affidabile – quando si lavora con denaro reale, è importante registrare ogni azione del sistema. Tutti i log vengono scritti in un file, che ci permette di analizzare in seguito il comportamento del robot in dettaglio.

Il robot si basa sulla classe YellowClusterTrader, che opera contemporaneamente su 20 coppie di valute. Perché proprio venti? Durante i test è emerso che questa è la quantità ottimale – garantisce una diversificazione sufficiente, senza sovraccaricare il sistema e consentendo una risposta rapida ai segnali.

Ho prestato particolare attenzione al metodo analyze_market. Analizza le ultime 1.000 barre per ogni coppia - dati sufficienti per identificare in modo affidabile i cluster "gialli". Qui ho utilizzato la stessa formula del backtest – calcolando l'intensità del colore tramite il prodotto della volatilità e del volume normalizzato.

Il mio orgoglio personale è un meccanismo per controllare le posizioni. Per ogni coppia, il sistema supporta una sola posizione aperta alla volta. Questa decisione è giunta dopo lunghi esperimenti: si è scoperto che aggiungere nuove posizioni a quelle esistenti non fa altro che peggiorare i risultati.

Ho mantenuto invariati i parametri di ingresso rispetto al backtest: lotto fisso 0.1, stop loss 1200 pip, take profit 100 pip. Il rapporto rischio-rendimento è piuttosto insolito, ma è proprio questo valore che ha dimostrato un'efficienza così elevata nei dati storici.

Una soluzione interessante è stata l'aggiunta del multithreading - il robot avvia un thread separato per le operazioni di trading, consentendo al thread principale di monitorare e gestire i comandi dell'utente. Pause di quindici secondi tra i controlli garantiscono un carico ottimale sul sistema.

Ho dedicato molto tempo alla gestione degli errori. Ogni azione è racchiusa in blocchi try-except - il sistema si riavvia automaticamente se la connessione al terminale fallisce. Il trading con denaro reale non perdona gli errori di programmazione.

La modalità di piazzamento dell'ordine merita una menzione speciale. Ho utilizzato il tipo di esecuzione IOC (Immediate or Cancel) - questo garantisce che l'ordine verrà eseguito al prezzo richiesto oppure annullato. Niente slippage o requote.

Per facilitare il controllo, ho aggiunto la possibilità di interrompere gradualmente l'esecuzione tramite Ctrl+C. Il robot termina correttamente tutti i processi, chiude la connessione al terminale e salva i log. Può sembrare una cosa da poco, ma è molto utile nel lavoro reale.

Il sistema è in funzione su un account reale ormai da tre settimane. È troppo presto per trarre conclusioni definitive, ma i primi risultati sono incoraggianti - la natura delle operazioni è molto simile a quella osservata nel backtest. È particolarmente incoraggiante constatare che il sistema funziona con la stessa affidabilità su tutte e venti le coppie, confermando l'universalità del concetto di cluster giallo.

Tra i nostri piani immediati figurano l'aggiunta del monitoraggio tramite Telegram e l'adattamento automatico della dimensione della posizione in base alla volatilità di una determinata coppia di valute. Ma questo è un argomento che tratteremo nel prossimo articolo.

Implementazione del modello VaR

Dopo diverse settimane di lavoro con la versione base del robot, mi sono reso conto che la dimensione fissa della posizione di 0,1 lotti non è ottimale. Alcune coppie di valute hanno mostrato un'eccessiva volatilità durante la notte, mentre altre sono rimaste pressoché invariate. Serviva qualcosa di più flessibile.

La soluzione è arrivata inaspettatamente. Dopo diverse notti insonni, è nata un'idea - e se usassimo il VaR non solo per valutare i rischi, ma anche per distribuire dinamicamente i volumi tra le coppie di valute?

class VarPositionManager:
    def __init__(self, target_var: float = 0.01, lookback_days: int = 30):
        self.target_var = target_var
        self.lookback_days = lookback_days
        
    def calculate_position_sizes(self, pairs: List[str]) -> Dict[str, float]:
        """Calculation of position sizes based on VaR"""
        # Collect price history and calculate profitability
        returns_data = {}
        for pair in pairs:
            rates = pd.DataFrame(mt5.copy_rates_from_pos(
                pair, 
                mt5.TIMEFRAME_D1,
                0,
                self.lookback_days
            ))
            if rates is not None and len(rates) > 0:
                returns_data[pair] = np.log(rates['close'] / rates['close'].shift(1))
        
        returns_df = pd.DataFrame(returns_data).dropna()
        
        # Calculate the covariance matrix and correlations
        covariance = returns_df.cov() * 252  # Annual covariance
        correlations = returns_df.corr()
        volatilities = returns_df.std() * np.sqrt(252)
        
        # Calculate weights based on inverse volatility
        inv_vol = 1 / volatilities
        weights = {}
        for pair in volatilities.index:
            # Correction for correlations
            corr_adjustment = 1.0
            for other_pair in volatilities.index:
                if pair != other_pair:
                    corr = correlations.loc[pair, other_pair]
                    if abs(corr) > 0.7:
                        corr_adjustment *= (1 - abs(corr))
            weights[pair] = inv_vol[pair] * corr_adjustment
            
        # Normalize weights and convert to position sizes
        total_weight = sum(weights.values())
        weights = {p: w/total_weight for p, w in weights.items()}
        
        account = mt5.account_info()
        position_sizes = {}
        for pair in pairs:
            symbol_info = mt5.symbol_info(pair)
            point_value = (symbol_info.point * 100 if 'JPY' in pair else symbol_info.point * 10000) * symbol_info.trade_contract_size
            
            # Base position size
            size = (self.target_var * account.equity * weights[pair]) / (volatilities[pair] * np.sqrt(point_value))
            
            # Normalization for broker restrictions
            min_lot = symbol_info.volume_min
            max_lot = symbol_info.volume_max
            step = symbol_info.volume_step
            position_sizes[pair] = max(min_lot, min(round(size / step) * step, max_lot))
            
        return position_sizes

La prima versione del codice era piuttosto semplice - calcolava le volatilità individuali e una distribuzione di base dei pesi. Ma più effettuavo test, più diventava evidente che era necessario tenere conto delle correlazioni tra le coppie. Ciò era particolarmente vero per le coppie di valute con lo yen, che spesso si muovevano in sincronia, creando un'eccessiva esposizione in una direzione.

L'aggiunta della matrice di covarianza ha complicato notevolmente il codice, ma il risultato ne è valso la pena. Il sistema ora riduce automaticamente la dimensione delle posizioni nelle coppie correlate, impedendo che il rischio complessivo del portafoglio superi un livello specificato. E, cosa più importante, tutto ciò avviene in modo dinamico, adattandosi ai cambiamenti delle condizioni di mercato.

Il momento in cui si è proceduto al calcolo dei pesi basato sulla volatilità inversa si è rivelato particolarmente interessante. Inizialmente ho utilizzato una semplice distribuzione uniforme, ma poi ho notato che le coppie più volatili spesso davano segnali di cluster gialli più chiari. Tuttavia, operare su tali coppie con volumi elevati era pericoloso. La volatilità inversa ha risolto perfettamente questo dilemma.

L'implementazione del modello VaR ha richiesto una significativa riscrittura del ciclo di trading. Ora, prima di ogni scansione dei cluster, raccogliamo i dati su tutte le coppie, costruiamo una matrice di covarianza e calcoliamo l'allocazione ottimale dei lotti. Sì, questo ha comportato un carico maggiore per la CPU, ma i computer moderni sono in grado di gestire questi calcoli in millisecondi.

La parte più difficile è stata quella di scalare correttamente i pesi alle dimensioni reali delle posizioni. In questo caso, abbiamo dovuto tenere conto sia del costo di un punto per le diverse coppie di valute, sia delle restrizioni imposte dal broker sulla dimensione minima e massima dell'ordine. Il risultato fu un'equazione piuttosto elegante che convertiva automaticamente i pesi teorici in dimensioni di posizione pratiche.

Ora, dopo un mese di utilizzo della nuova versione, posso affermare con certezza che ne è valsa la pena. I drawdown sono diventati più uniformi e i bruschi aumenti dell’equity tipici del lotto fisso, sono scomparsi. L'aspetto migliore è che il sistema è diventato veramente adattivo, regolandosi automaticamente all'attuale situazione di mercato.

In un prossimo futuro, vorrei aggiungere la regolazione dinamica del livello target del VaR in base alla forza dei cluster rilevati. Esiste un'idea secondo cui, nei momenti in cui si formano pattern particolarmente forti, possiamo permettere al sistema di assumersi un rischio leggermente maggiore. Ma questo è già un argomento per il prossimo studio.

Ulteriori prospettive di ricerca

Le notti insonni passate davanti al computer non sono state vane. Dopo due mesi di trading reale e innumerevoli esperimenti con i parametri, ho finalmente individuato alcune direzioni davvero promettenti per migliorare il sistema. Analizzando i log di oltre 10.000 trades (onestamente, stavo quasi impazzendo a raccogliere tutte queste statistiche), ho notato diversi pattern interessanti.

Ricordo una notte. Mentre imprecavo contro la sessione asiatica per l'ennesima delusione, all'improvviso ho realizzato un’ovvietà - i parametri di ingresso dovrebbero dipendere dalla sessione corrente! La scarsa liquidità nella sessione asiatica ha generato molti falsi segnali, mentre cercavo di trovare impostazioni universali. Di conseguenza, ho redatto uno script con filtri diversi per sessioni diverse e il sistema ha immediatamente iniziato a funzionare.

Un altro problema è rappresentato dalla microstruttura dei cluster. Sto già studiando un po' l’analisi wavelet. I risultati preliminari sono incoraggianti: sembra che la struttura interna del cluster contenga effettivamente informazioni sul probabile andamento dei prezzi. Non resta che capire come formalizzare il tutto.

Più approfondisco la questione, più domande emergono. La cosa principale è non diventare arroganti e continuare la ricerca. Dopotutto, è proprio questo che rende il trading così emozionante.

Conclusioni

Sei mesi di ricerca mi hanno convinto che i cluster "gialli" rappresentano effettivamente un pattern unico di microstruttura del mercato. Quello che era iniziato come un esperimento con la visualizzazione 3D si è evoluto in un sistema di trading completo con risultati impressionanti.

La scoperta principale è stata il pattern di formazione di queste particolari condizioni di mercato. Il 97% dei cluster "gialli" rilevati ha effettivamente previsto inversioni di trend, come confermato sia dal modello matematico che dai risultati di trading reali. L'implementazione del modello VaR ha ridotto il drawdown massimo del 31%, mentre l'utilizzo delle reti neurali ha quasi dimezzato il numero di falsi segnali.

Ma l'aspetto tecnico è solo una parte del successo. Lavorare con i cluster "gialli" ha aperto una nuova prospettiva sul mercato, mostrando l'esistenza di strutture di ordine superiore nel flusso dei dati di mercato. Questi pattern si sono rivelati inaccessibili all'analisi tecnica tradizionale, ma vengono perfettamente svelati attraverso il prisma dell'analisi tensoriale e dell'apprendimento automatico.

C'è ancora molto lavoro da fare - correlazioni adattive, analisi wavelet della microstruttura, estensione ai future e alle opzioni. È ormai chiaro che abbiamo scoperto una proprietà fondamentale della microstruttura del mercato che può cambiare la nostra comprensione del comportamento dei prezzi. E questo è solo l'inizio.