強化学習と弱者淘汰を組み合わせた進化型取引アルゴリズム(ETARE)

はじめに

進化、ニューラルネットワーク、そしてトレーダー。この3つに共通するものは何だと思いますか。それは、失敗から学ぶということです。この考えが浮かんだのは、またしても徹夜で端末に向かっていたある夜のことでした。私の「完璧な」取引アルゴリズムが、またもや予想外の市場変動によって資金を失ったときのことです。

その日を今でもはっきり覚えています。2016年6月23日、Brexitの国民投票の日です。私のアルゴリズムは、古典的なテクニカル分析パターンに基づき、GBPのロングポジションを自信満々で保有していました。「どの世論調査を見ても、イギリスはEUに残留する」と私は思っていたのです。しかしモスクワ時間の午前4時、Brexit支持派の勝利が明らかになると、GBPはわずか数分で1800ポイントも急落しました。私の口座残高は一気に40％減少しました。

2023年3月、私はETARE（Evolutionary Trading Algorithm with Reinforcement and Extinction：強化学習と淘汰を組み合わせた進化型取引アルゴリズム）の開発を始めました。なぜ淘汰なのか。それは自然界では強者のみが生き残るからです。ならば、この原則を取引戦略にも応用できるのではないかと考えたのです。

古典的なテクニカル分析と最新の人工知能技術が融合する世界に飛び込む準備はできていますか。すべての取引戦略がダーウィン的な自然選択の中で生存を賭けて競い合う世界です。シートベルトを締めてください。これからお見せするのは、単なる取引ロボットではありません。これは、15年にわたる試行錯誤、何千時間にも及ぶプログラミング、そして正直に言えばいくつかの口座破綻を経て築かれた成果です。そして何よりも重要なのは、実際に利益をもたらしている動作中のシステムだということです。

システムアーキテクチャ

ETAREの中心にあるのは、現代の量子コンピュータを思わせるハイブリッド構造です。かつてMetaTrader 4で、2本の移動平均線の交差に基づく単純なスクリプトを書いていた時代がありました。当時はそれが大きなブレイクスルーに思えたものです。今振り返ると、私たちはまるでコンパスと星だけで大海を渡ろうとする古代の航海士のようでした。

2022年の暴落を経て、市場はあまりにも複雑で単純な手法では通用しないことが明らかになりました。そこから私の機械学習への探求が始まりました。

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

アリの群れを想像してみてください。1匹1匹のアリが取引戦略であり、強い個体が生き残ってその遺伝子を子孫に伝え、弱い個体は淘汰されます。ETAREでは、この遺伝子の役割をニューラルネットワークの重みが担っています。

population_size=50としているのは、少なすぎると多様性が不足し、多すぎると市場変化への適応が遅くなるためです。

自然界のアリが新しい領域を探索し、餌を見つけ、仲間に情報を共有するように、ETAREでも各戦略が市場を探索し、成功した取引パターンを交叉によって次世代に伝えていきます。

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

2024年12月、取引ログを分析していると、最も成功しているコードは、他の優れたアプローチを組み合わせた「ハイブリッド」であることに気づきました。自然界で強い遺伝子が健康な子孫を生み出すように、アルゴリズム取引の世界でも、成功したパターンを組み合わせることで、より高性能な戦略が生まれるのです。

システムの中心にはLSTMネットワークがあります。これは「記憶」を持つ特殊なニューラルネットワークです。私は単純な多層パーセプトロン(MLP)から複雑なTransformerまで、さまざまなアーキテクチャを試しましたが、最終的にこの構成に落ち着きました。

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

100回の取引ごとに、システムは「掃除」を実行し、利益を出せない戦略を容赦なく削除します。これはETAREの重要な仕組みのひとつであり、その誕生には特別な経緯があります。2023年12月、取引ログを分析していたときに、ある驚くべきパターンを発見しました。それは、最初の100〜150回の取引で損失を出した戦略は、その後も不調が続く傾向があるというものでした。この発見がシステムのアーキテクチャを根本的に変えるきっかけとなりました。

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)

取引判断のデータベースは、システムの記憶として機能します。すべての決定、すべての結果が記録され、後の分析に役立てられます。

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

この複雑な仕組み全体が、ひとつの有機体のように連動し、市場の変化に絶えず進化し、適応し続けています。VIXが25を超えるような高ボラティリティ期間には、システムは自動的に戦略の信頼性要件を引き上げます。一方で、市場が落ち着いている期間には、より積極的になり、新しい取引パターンを試す余地をユーザーに与えます。

強化学習メカニズム

取引ロボットの開発には、あるパラドックスがあります。それは、アルゴリズムが複雑になればなるほど、実際の市場でのパフォーマンスが悪くなるということです。 

そのため、ETAREの学習メカニズムにおいてシンプルさと透明性を重視しました。さまざまなアーキテクチャを2年間にわたって試行した結果、私たちは優先度付きメモリシステムにたどり着きました。

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)

すべての取引判断は、単なるエントリーではなく、リスクと潜在的リターンの複雑なバランスであると考えています。次のようにしてシステムは自らの決定から学習します。

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)

私はこれまでに何度も損失を経験してきました。その原因は、モデルが異なる市場環境に適応できなかったことにあります。この経験を通じて、適応型学習のアイデアが生まれました。現在では、システムは各トランザクションを分析し、その結果に応じて行動を調整します。

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

この仕組みの重要な点は、システムが単に成功した取引を記憶するだけでなく、なぜその取引が成功したのかを理解するように学習することです。それを可能にしているのが、PyTorchで実装された多層のバックプロパゲーション構造です。

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

その結果、このシステムは理想的なバックテスト結果からではなく、実際の取引経験から学習するようになりました。過去1年間の実運用テストにおいて、ETAREは穏やかなトレンド相場から極めて高いボラティリティ環境まで、幅広い市場状況に適応できることを実証しました。

しかし、最も重要なのは、このシステムが進化し続けているという点です。すべての取引、すべての市場ループを経るたびに、少しずつ賢く、少しずつ効率的になっていきます。あるベータテスターはこう語りました。「これまで、過去データに合わせてパラメータを調整するのではなく、本当に自分のミスから学習するアルゴリズムを見たのは初めてです。」

弱者淘汰のメカニズム

チャールズ・ダーウィンは金融市場で取引をしたことはありませんでしたが、彼の進化論は、成功する取引戦略のダイナミクスを見事に説明しています。自然界では、最も強い個体や最も速い個体が生き残るわけではなく、環境の変化に最も適応できる個体が生き残るのです。市場でも、まったく同じことが起きています。

歴史を振り返ると、「完璧」と称された取引アルゴリズムが、最初のブラックスワン（予測不能な異常事象）によって壊滅した例は数多く存在します。2015年、スイス国立銀行がCHFとEURのペッグを突然解除したとき、私も資金の大部分を失いました。そのときのアルゴリズムは、このような事態にまったく対応できなかったのです。それ以来、私は考えました。なぜ自然は数百万年もの間ブラックスワンに適応してきたのに、私たちのアルゴリズムはそれができないのか。

答えは意外なところからやってきました。ダーウィンの『種の起源』を読んでいたときのことです。ダーウィンは、急激な気候変化の時期に生き残るのは、最も特化した種ではなく、適応力を保持している種であると述べています。この原則こそが、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)

自然界で大量絶滅が新たな高次の種の誕生を促すように、私たちのシステムでも、高ボラティリティの時期が戦略進化の触媒となります。以下に示すのは、ETAREにおける自然選択の仕組みです。

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)

私たちは、適応度評価のメカニズムに特に注意を払いました。自然界においては、適応度とは生存可能な子孫を残す能力を意味します。一方、私たちのシステムにおいては、さまざまな市場環境下で利益を生み出す能力を指します。

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

このようにして、生き残った戦略の突然変異が発生します。このプロセスは、自然界における遺伝的突然変異に似ています。DNAの偶発的な変化が、より生存力の高い個体を生み出すことがあるのです。

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)

興味深いことに、システムの一部のバージョンでは、市場のボラティリティが高い時期に自動的に突然変異の強度が高まります。これは、ストレス環境下で突然変異速度を上げる一部の細菌の挙動にも似ています。私たちの場合：

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
    )

特に重要なのは、集団の多様化メカニズムです。自然界では、遺伝的多様性が種の生存にとって不可欠な要素です。ETAREでも、この原則を同様に実装しています。

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
                )

そして結果として、このシステムは単に取引をおこなうだけでなく、市場とともに進化するようになりました。ダーウィンが言ったように、生き残るのは最も強いものではなく、最も適応したものなのです。アルゴリズム取引の世界では、この言葉がこれまでになく現実的な意味を持っています。

取引判断データベース

取引経験を蓄積し続けることは、それを得ることと同じくらい重要です。アルゴリズム取引システムに長年携わってきた中で、私は何度も痛感してきました。信頼できるデータベースがなければ、どんな取引システムでも、いずれ自分の最良の戦略を「忘れてしまう」ということを。ETAREでは、この問題を解決するために、取引判断を多層的に記録し、保存する仕組みを実装しています。

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

すべての取引、すべての判断は、たとえ一見取るに足らないものであっても、すべて、システム全体の集合的経験の一部となります。以下は、各取引ループの後にデータを保存する方法です。

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

たとえサーバーが致命的な障害を起こしたとしても、詳細なログとバックアップのおかげで、システム全体は数分で復旧します。次のコードは、その復元メカニズムの仕組みを示しています。

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

過去データの分析にも特別な注意を払っています。すべての成功した戦略は、その後の意思決定を改善するために活用できる痕跡を残します。

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

ETAREのデータベースは、単なる情報の保管庫ではありません。それは過去を分析し、未来を予測することのできる、まさにシステムの「脳」そのものです。私の古い師匠がよく言っていた言葉があります。「記憶のない取引システムは、経験のないトレーダーと同じだ。毎日ゼロからやり直すことになる。」

データと特徴量

アルゴリズム取引に取り組んできた年月の中で、私は何百通りものインジケーターの組み合わせを試してきました。ある時点では、私の取引システムはRSIなどの古典的な指標から、自作のエキゾチックなインジケーターまで、50種類以上を同時に使用していました。しかし、またしても資金を失った後に気づいたのです。重要なのは数ではなく、データの扱い方であるということです。

Brexit時の出来事をよく覚えています。当時のシステムは数十ものインジケーターを搭載していましたが、相反するシグナルが同時に出た結果、判断不能になり「フリーズ」してしまったのです。この経験こそが、ETAREの誕生につながりました。つまり、必要最小限のインジケーターだけを用い、それを知的に扱うシステムです。

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

ETAREにおけるRSIは、単なる買われ過ぎと売られ過ぎの検出ツールではありません。市場心理の総合的な分析の一部として機能しており、特に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 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']

次に重要なのは、ボラティリティとモメンタムの分析です。 

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

ETAREにおける出来高分析は、単なるティック数のカウントではありません。異常なボリュームを検知する専用アルゴリズムを開発し、強い値動きの前兆を察知できるようにしています。

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

そして最後の仕上げが、データの正規化です。これは、多くのトレーダーが見落としがちなステップです。

# 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

ETAREにおける各インジケーターは、単なる数値ではありません。それぞれが市場分析という複雑なモザイクの一片を構成しているのです。システムは市場の変化に絶えず適応し、現在の状況に応じて各インジケーターの重みを動的に調整します。次の章では、こうして準備されたデータが、どのようにして実際の取引判断へと変換されていくのかを見ていきましょう。

取引ロジック

ここで紹介するのは、最先端のアルゴリズム取引技術を集約した革新的な取引システムの概要です。このシステムは、遺伝的最適化、機械学習、そして高度なリスク管理を組み合わせたハイブリッド型のアプローチを採用しています。

システムの中核は、常時稼働する取引ループです。このループはリアルタイムで市場状況を解析し、それに適応し続けます。まるで自然界の進化のように、システムは一定周期ごとに非効率な取引戦略を自動的に「淘汰」し、新しくより有望な戦略へと置き換えます。これは50取引ごとにおこなわれ、アルゴリズムの継続的な進化を保証しています。

各取引銘柄は、その固有の特性を考慮して個別に処理されます。システムは過去100本のローソク足データを分析し、現在の市場状態を正確に把握します。その分析結果に基づき、ポジションの新規エントリーやクローズに関する意思決定がおこなわれます。

特に注目すべきは、ポジション平均化戦略(DCA: Dollar Cost Averaging)です。新規ポジションを開く際、システムは自動的にロットサイズを調整します。初期値の0.1ロットから開始し、ポジション数が増えるごとに最小0.01ロットまで段階的に縮小します。これにより、リスクを効率的に抑えつつ、潜在的利益を最大化します。

ポジション決済プロセスも、綿密に設計されています。システムは各ポジションの収益性をリアルタイムで監視し、指定された利益水準に達した時点で自動的にクローズします。このとき、買いと売りのポジションは別々に管理されるため、より柔軟で精密なポートフォリオ運用が可能になります。取引の結果得られる報酬や損失は、次の学習サイクルにおける重要な学習データとして活用されます。 

また、すべての取引操作およびシステムの状態はデータベースに保存されます。これにより、後から詳細なパフォーマンス分析をおこなったり、戦略の最適化を実施したりすることができます。こうした構造が、アルゴリズムの継続的な改良を支える強固な基盤となっています。

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

この結果として生まれたのが、信頼性が高く、自己学習型の取引システムです。多様な市場環境下でも効率的に稼働できるよう設計されており、進化的アルゴリズム、機械学習、そして実証済み取引戦略の組み合わせが、現代の取引における強力な武器となっています。

結論

最後に強調しておきたいのは、ETAREは単に普通の取引アルゴリズムではなく、アルゴリズム取引における長年の進化の成果であるということです。このシステムは、さまざまな分野のベストプラクティスを統合しています。市場環境の変化に適応するための遺伝的アルゴリズム、意思決定を支える深層学習、そして安定的な運用を可能にする古典的なリスク管理手法を組み合わせています。

ETAREの独自性は、自身の経験から絶えず学習し続ける能力にあります。取引の結果が良くても悪くても、すべての取引がシステム全体の集合的記憶として蓄積され、将来の取引判断をより精度の高いものに改善する助けとなります。ダーウィンの進化論に着想を得た取引戦略の自然淘汰メカニズムによって、最も効果的なアプローチだけが生き残るようになっています。

開発およびテストの過程において、ETAREは静かなトレンド相場から高ボラティリティ相場まで、さまざまな市場環境での耐性を証明しました。特に注目すべきは、DCA戦略とポジションを個別に決済するメカニズムの有効性です。これらにより、リスクを適切にコントロールしつつ、利益を最大化することが可能です。

効率性について率直に申し上げますと、ETAREのメインモジュール自体は、私の代わりに取引をおこなうものではありません。ETAREは、より大きなMidas取引エコシステムの一部として統合されています。

現在、Midasにはこのモジュールを含めて24個のモジュールが存在しています。複雑性は今後さらに増していく予定であり、詳細については今後の記事で順次解説する予定です。 

アルゴリズム取引の未来は、まさにこのような市場とともに進化する適応型システムにあります。ETAREはその方向性を示す一歩であり、現代のテクノロジーを活用して信頼性が高く、収益性のある取引ソリューションを構築できることを示しています。