Evolutionärer Handelsalgorithmus mit Verstärkungslernen und Auslöschung von schwachen Individuen (ETARE)

Einführung

Wissen Sie, was die Evolution, neuronale Netze und Händler gemeinsam haben? Sie alle lernen aus ihren Fehlern. Genau dieser Gedanke kam mir nach einer weiteren schlaflosen Nacht am Terminal, als mein „perfekter“ Handelsalgorithmus aufgrund einer unerwarteten Marktbewegung wieder einmal sein Depot verlor.

Ich erinnere mich an diesen Tag, als wäre es gestern gewesen: 23. Juni 2016, das Brexit-Referendum. Mein Algorithmus, der auf klassischen technischen Analysemustern basiert, hielt zuversichtlich eine Kaufposition in GBP. „Alle Umfragen zeigen, dass Großbritannien in der EU bleiben wird“, dachte ich damals. Um 4 Uhr morgens. Als in Moskau die ersten Ergebnisse einen Sieg der Brexit-Befürworter anzeigten, brach das GBP innerhalb weniger Minuten um 1800 Punkte ein. Meine Einlage verlor 40%.

Im März 2023 begann ich mit der Entwicklung von ETARE – Evolutionary Trading Algorithm with Reinforcement and Extinction (Elimination). Warum Eliminierung? Denn in der Natur überleben die Stärksten. Warum also nicht dieses Prinzip auf Handelsstrategien anwenden?

Sind Sie bereit, in die Welt einzutauchen, in der die klassische technische Analyse auf die neuesten Fortschritte der künstlichen Intelligenz trifft? Wo jede Handelsstrategie in der darwinistischen natürlichen Selektion ums Überleben kämpft? Dann schnallen Sie sich an – es wird interessant werden. Denn was Sie gleich sehen werden, ist nicht nur ein weiterer Handelsroboter. Es ist das Ergebnis von 15 Jahren Versuch und Irrtum, Tausenden von Programmierstunden und, offen gesagt, ein paar zerstörten Einlagen. Aber das Wichtigste ist, dass es sich um ein funktionierendes System handelt, das seinen Nutzern bereits echten Gewinn bringt.

Systemarchitektur

Das Herzstück von ETARE ist eine hybride Architektur, die an einen modernen Quantencomputer erinnert. Erinnern Sie sich noch an die Zeit, als wir einfache Skripte für MetaTrader 4 geschrieben haben, die auf dem Schnittpunkt zweier gleitender Durchschnitte basierten? Damals schien dies ein Durchbruch zu sein. Wenn ich jetzt zurückblicke, wird mir klar, dass wir wie alte Seefahrer waren, die nur mit Hilfe eines Kompasses und der Sterne versuchten, den Ozean zu überqueren.

Nach dem Crash von 2022 wurde klar, dass der Markt zu komplex für einfache Lösungen ist. Damals begann meine Reise in die Welt des maschinellen Lernens.

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

Stellen Sie sich eine Ameisenkolonie vor, in der jede Ameise eine Handelsstrategie darstellt. Starke Individuen überleben und geben ihre Gene an ihre Nachkommen weiter, während schwache Individuen verschwinden. In meinem System wird die Rolle der Gene von den Gewichtsverhältnissen des neuronalen Netzes übernommen.

Warum population_size=50? Denn weniger Strategien bieten keine ausreichende Diversifizierung, während mehr Strategien eine schnelle Anpassung an Marktveränderungen erschweren.

In der Natur erkunden Ameisen ständig neue Gebiete, finden Nahrung und geben Informationen an ihre Verwandten weiter. In ETARE erforscht jede Strategie auch den Markt, und erfolgreiche Handelsmuster werden durch einen Kreuzungsmechanismus an künftige Generationen weitergegeben:

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

Im Dezember 2024 fiel mir bei der Analyse von Handelsprotokollen auf, dass die erfolgreichsten Codes oft „Mischformen“ aus anderen erfolgreichen Ansätzen sind. So wie in der Natur starke Gene gesunde Nachkommen hervorbringen, können auch im algorithmischen Handel erfolgreiche Muster kombiniert werden, um noch effizientere Strategien zu entwickeln.

Das Herzstück des Systems war das LSTM-Netz, eine spezielle Art von neuronalem Netz mit „Gedächtnis“. Nach monatelangem Experimentieren mit verschiedenen Architekturen, von einfachen mehrschichtigen Perceptrons bis hin zu komplexen Transformatoren, haben wir uns für diese Konfiguration entschieden:

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

Alle 100 Handelsgeschäfte führt das System eine „Reinigung“ durch, bei der unrentable Strategien rücksichtslos entfernt werden. Dies ist einer der Schlüsselmechanismen von ETARE, und seine Entstehung ist eine eigene Geschichte. Ich erinnere mich an eine Nacht im Dezember 2023, als ich meine Handelsprotokolle analysierte und ein überraschendes Muster feststellte: Die meisten Strategien, die in den ersten 100-150 Handelsgeschäfte Verluste aufwiesen, waren auch danach unrentabel. Diese Beobachtung hat die Systemarchitektur völlig verändert:

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)

Die Datenbank für Handelsentscheidungen dient als Systemspeicher. Jede Entscheidung, jedes Ergebnis – alles wird zur späteren Analyse aufgezeichnet:

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

Dieser gesamte komplexe Mechanismus funktioniert wie ein einziger Organismus, der sich ständig weiterentwickelt und an die Veränderungen des Marktes anpasst. In Zeiten hoher Volatilität, z. B. wenn der VIX über 25 liegt, erhöht das System automatisch die Zuverlässigkeitsanforderungen an die Strategien. Und in ruhigen Zeiten wird er aggressiver, sodass die Nutzer mit neuen Handelsmustern experimentieren können.

Mechanismus des Verstärkungslernens

Bei der Entwicklung von Handelsrobotern gibt es ein Paradoxon: Je komplexer der Algorithmus ist, desto schlechter schneidet er auf dem realen Markt ab. 

Deshalb haben wir uns bei dem ETARE-Lernmechanismus auf Einfachheit und Transparenz konzentriert. Nachdem wir zwei Jahre lang mit verschiedenen Architekturen experimentiert hatten, kamen wir zu einem nach Prioritäten geordneten Speichersystem:

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)

Jede Handelsentscheidung ist mehr als nur ein Markteintritt, sondern ein komplexes Gleichgewicht zwischen Risiko und potenziellem Gewinn. Sehen Sie, wie das System aus seinen Entscheidungen lernt:

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)

Ich habe viele Male verloren, weil sich die Modelle nicht an eine andere Marktsituation anpassen konnten. Damals wurde die Idee des adaptiven Lernens geboren. Jetzt analysiert das System jede Transaktion und passt sein Verhalten an:

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

Der entscheidende Punkt ist, dass sich das System nicht nur an erfolgreiche Handelsgeschäfte erinnert, sondern auch lernt zu verstehen, warum sie erfolgreich waren. Ermöglicht wird dies durch die in PyTorch implementierte mehrschichtige Architektur des Rückwärtsdurchlaufs:

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

Das Ergebnis ist ein System, das nicht aus idealen Backtests, sondern aus echter Handelserfahrung lernt. Im Laufe des letzten Jahres, in dem wir den Markt live getestet haben, hat ETARE seine Fähigkeit unter Beweis gestellt, sich an eine Vielzahl von Marktbedingungen anzupassen, von ruhigen Trends bis hin zu sehr volatilen Phasen.

Das Wichtigste ist jedoch, dass sich das System weiterentwickelt. Mit jedem Handel, mit jeder Marktschleife wird er ein bisschen schlauer, ein bisschen effizienter. Einer unserer Beta-Tester sagte: „Das ist das erste Mal, dass ich einen Algorithmus gesehen habe, der tatsächlich aus seinen Fehlern lernt, anstatt nur die Parameter an die historischen Daten anzupassen.“

Der Mechanismus des Aussterbens von schwachen Individuen

Charles Darwin hat nie an den Finanzmärkten gehandelt, aber seine Evolutionstheorie liefert eine bemerkenswerte Beschreibung der Dynamik erfolgreicher Handelsstrategien. In der Natur überleben nicht die stärksten oder schnellsten Individuen, sondern diejenigen, die sich am besten an Umweltveränderungen anpassen. Das Gleiche geschieht auf dem Markt.

Die Geschichte kennt viele Fälle, in denen ein „perfekter“ Handelsalgorithmus nach dem ersten schwarzen Schwan ausradiert wurde. Im Jahr 2015 verlor ich einen erheblichen Teil meiner Einlagen, als die Schweizerische Nationalbank die Bindung des CHF an den EUR aufhob. Mein damaliger Algorithmus erwies sich als völlig unvorbereitet auf ein solches Ereignis. Das brachte mich zum Nachdenken: Warum ist die Natur seit Millionen von Jahren in der Lage, mit schwarzen Schwänen umzugehen, während unsere Algorithmen das nicht können?

Die Antwort kam unerwartet, als ich das Buch „On the Origin of Species“ las. Darwin beschrieb, dass in Zeiten abrupter Klimaveränderungen nicht die am stärksten spezialisierten Arten überlebten, sondern diejenigen, die sich die Fähigkeit zur Anpassung bewahrt hatten. Dieses Prinzip bildet die Grundlage für den Auslöschungsmechanismus in 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)

So wie in der Natur Zeiten des Massenaussterbens zur Entstehung neuer, fortschrittlicherer Arten führen, werden in unserem System Zeiten hoher Volatilität zu einem Katalysator für die Entwicklung von Strategien. Schauen Sie sich den Mechanismus der natürlichen Selektion an:

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)

Besonderes Augenmerk haben wir auf den Mechanismus der Fitnessbewertung gelegt. In der Natur ist dies die Fähigkeit eines Individuums, lebensfähige Nachkommen zu erzeugen; in unserem Fall ist es die Fähigkeit einer Strategie, unter verschiedenen Marktbedingungen Gewinne zu erzielen:

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

So kommt es zur Mutation von Überlebensstrategien. Dieser Prozess erinnert an genetische Mutationen in der Natur, wo zufällige Veränderungen in der DNA manchmal zur Entstehung lebensfähigerer Organismen führen:

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)

Interessanterweise erhöht das System in einigen Versionen in Zeiten hoher Marktvolatilität automatisch die Intensität der Mutationen. Dies erinnert daran, wie einige Bakterien unter Stressbedingungen Mutationen beschleunigen. In unserem Fall:

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
    )

Der Mechanismus der Diversifizierung der Population ist besonders wichtig. In der Natur ist die genetische Vielfalt der Schlüssel zum Überleben der Arten. In ETARE haben wir ein ähnliches Prinzip eingeführt:

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
                )

Ergebnis? Das System, das nicht nur handelt, sondern sich mit dem Markt entwickelt. Wie Darwin sagte, überlebt nicht der Stärkste, sondern der Anpassungsfähigste. In der Welt des algorithmischen Handels ist dies aktueller denn je.

Datenbank für Handelsentscheidungen

Die Aufrechterhaltung der Handelserfahrung ist ebenso wichtig wie das Sammeln von Erfahrungen. Im Laufe der Jahre, in denen ich mit algorithmischen Systemen gearbeitet habe, bin ich immer wieder zu der Überzeugung gelangt, dass jedes Handelssystem ohne eine zuverlässige Datenbank früher oder später seine besten Strategien „vergisst“. In ETARE haben wir eine mehrstufige Speicherung für Handelsentscheidungen implementiert:

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

Jeder Handel, jede Entscheidung, auch die scheinbar unbedeutenden, werden Teil der kollektiven Erfahrung des Systems. Hier sehen Sie, wie wir die Daten nach jeder Handelsschleife speichern:

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

Selbst nach einem kritischen Serverausfall kann das gesamte System dank detaillierter Protokolle und Backups innerhalb weniger Minuten wiederhergestellt werden. Der Wiederherstellungsmechanismus funktioniert folgendermaßen:

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

Wir werden der Analyse historischer Daten besondere Aufmerksamkeit widmen. Jede erfolgreiche Strategie hinterlässt eine Spur, die zur Verbesserung künftiger Entscheidungen genutzt werden kann:

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

Die ETARE-Datenbank ist nicht nur ein Speicher für Informationen, sondern das eigentliche „Gehirn“ des Systems, das in der Lage ist, die Vergangenheit zu analysieren und die Zukunft vorherzusagen. Wie mein alter Mentor zu sagen pflegte: „Ein Handelssystem ohne Gedächtnis ist wie ein Händler ohne Erfahrung: er fängt jeden Tag bei Null an“.

Daten und Merkmale

Im Laufe der Jahre, in denen ich mich mit dem algorithmischen Handel beschäftigt habe, habe ich Hunderte von Indikatorenkombinationen ausprobiert. Zu einem bestimmten Zeitpunkt verwendete mein Handelssystem mehr als 50 verschiedene Indikatoren, vom klassischen RSI bis zu exotischen, von mir selbst entwickelten Indikatoren. Aber wissen Sie, was mir nach einer weiteren verlorenen Einzahlung klar wurde? Es geht nicht um die Menge, sondern um den richtigen Umgang mit den Daten.

Ich erinnere mich an einen Vorfall während des Brexit: Ein System mit Dutzenden von Indikatoren ist einfach „eingefroren“ und konnte aufgrund widersprüchlicher Signale keine Entscheidung treffen. So entstand die Idee für ETARE – ein System, das ein Minimum an Indikatoren verwendet, diese aber auf intelligente Weise verarbeitet.

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

Der RSI ist in unserem System nicht nur ein überkaufter/überverkaufter Indikator. Wir verwenden sie als Teil einer umfassenden Analyse der Marktstimmung. Er arbeitet besonders effektiv in Kombination mit dem 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']

Bollinger-Bänder sind unser Volatilitätsradar“.  

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

Eine andere Geschichte ist die Analyse von Volatilität und Momentum. 

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

Die Volumenanalyse in ETARE ist mehr als nur das Zählen von Ticks. Wir haben einen speziellen Algorithmus zur Erkennung abnormaler Volumina entwickelt, der hilft, starke Bewegungen vorherzusagen:

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

Der letzte Schliff ist die Normalisierung der Daten. Dies ist ein entscheidender Schritt, den viele Menschen unterschätzen.

# 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

Jeder Indikator in ETARE ist nicht nur eine Zahl, sondern Teil eines komplexen Mosaiks von Marktanalysen. Das System passt sich ständig den Marktveränderungen an, indem es die Gewichtung der einzelnen Indikatoren je nach aktueller Situation anpasst. In den folgenden Abschnitten werden wir sehen, wie diese Daten in konkrete Handelsentscheidungen umgesetzt werden.

Handelslogik

Ich präsentiere Ihnen eine Beschreibung eines innovativen Handelssystems, das modernste algorithmische Handelstechnologien verkörpert. Das System basiert auf einem hybriden Ansatz, der genetische Optimierung, maschinelles Lernen und fortschrittliches Risikomanagement kombiniert.

Das Herzstück des Systems ist eine kontinuierlich arbeitende Handelsschleife, die ständig die Marktbedingungen analysiert und sich an diese anpasst. Wie die natürliche Evolution „säubert“ das System in regelmäßigen Abständen unwirksame Handelsstrategien und macht Platz für neue, vielversprechendere Ansätze. Dies geschieht alle 50 Handelsgeschäfte, wodurch eine kontinuierliche Verbesserung der Handelsalgorithmen gewährleistet wird.

Jedes Handelsinstrument wird individuell behandelt, wobei seine einzigartigen Merkmale berücksichtigt werden. Das System analysiert die historischen Daten der letzten 100 Kerzen und kann sich so ein genaues Bild von der aktuellen Marktlage machen. Auf der Grundlage dieser Analyse werden fundierte Entscheidungen über die Eröffnung und Schließung von Positionen getroffen.

Besonderes Augenmerk wird auf die Strategie der Positionsmittelung (DCA) gelegt. Bei der Eröffnung neuer Positionen reduziert das System automatisch deren Volumen, beginnend mit 0,1 Lot und allmählich abnehmend bis zum Mindestwert von 0,01 Lot. Dies ermöglicht ein effizientes Risikomanagement und eine Maximierung der potenziellen Gewinne.

Auch der Prozess der Positionsschließung ist sorgfältig durchdacht. Das System überwacht die Rentabilität der einzelnen Positionen und schließt sie, wenn ein bestimmtes Gewinnniveau erreicht ist. In diesem Fall werden Kauf- und Verkaufspositionen getrennt behandelt, was eine flexiblere Portfolioverwaltung ermöglicht. Die Belohnungen oder Bestrafungen, die sich aus dem Handel ergeben, sind der Schlüssel zu weiterem erfolgreichem Lernen. 

Alle Informationen über die Handelsvorgänge und den Systemstatus werden in der Datenbank gespeichert, sodass detaillierte Analysen durchgeführt und Strategien optimiert werden können. Dies schafft eine solide Grundlage für die weitere Verbesserung von Handelsalgorithmen.

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

Das Ergebnis ist ein zuverlässiges, selbstlernendes Handelssystem, das in der Lage ist, unter einer Vielzahl von Marktbedingungen effizient zu arbeiten. Die Kombination aus evolutionären Algorithmen, maschinellem Lernen und bewährten Handelsstrategien macht es zu einem leistungsstarken Werkzeug für den modernen Handel.

Schlussfolgerung

Abschließend möchte ich betonen, dass ETARE nicht einfach nur ein weiterer Handelsalgorithmus ist, sondern das Ergebnis einer langjährigen Entwicklung im algorithmischen Handel. Das System kombiniert bewährte Verfahren aus verschiedenen Bereichen: genetische Algorithmen zur Anpassung an sich ändernde Marktbedingungen, Deep Learning zur Entscheidungsfindung und klassische Risikomanagementmethoden.

Die Einzigartigkeit von ETARE liegt in seiner Fähigkeit, kontinuierlich aus seinen eigenen Erfahrungen zu lernen. Jeder Handel, unabhängig vom Ergebnis, wird Teil des kollektiven Gedächtnisses des Systems und trägt dazu bei, zukünftige Handelsentscheidungen zu verbessern. Der Mechanismus der natürlichen Selektion von Handelsstrategien, der sich an Darwins Evolutionstheorie orientiert, sorgt dafür, dass nur die effektivsten Ansätze überleben.

Während der Entwicklung und Erprobung hat das System seine Widerstandsfähigkeit in einer Vielzahl von Marktbedingungen bewiesen, von ruhigen Trendbewegungen bis hin zu hochvolatilen Perioden. Besonders hervorzuheben sind die Effizienz der DCA-Strategie und der Mechanismus der separaten Positionsschließung, die es uns ermöglichen, die Gewinne zu maximieren und gleichzeitig das Risikoniveau zu kontrollieren.

Nun zur Effizienz. Ich will es gleich sagen: Das ETARE-Hauptmodul selbst ist für mich nicht handelbar. Es ist als Modul in das breitere Midas-Handelsökosystem integriert.

Derzeit gibt es 24 Module in Midas, darunter auch dieses. Die Komplexität wird immer weiter zunehmen, und ich werde vieles davon in künftigen Artikeln beschreiben. 

Die Zukunft des algorithmischen Handels liegt genau in solchen adaptiven Systemen, die sich mit dem Markt weiterentwickeln können. ETARE ist ein Schritt in diese Richtung und zeigt, wie moderne Technologien eingesetzt werden können, um zuverlässige und profitable Handelslösungen zu schaffen.