Двунаправленная LSTM и квантовые вычисления для предсказания направления движения

Введение: когда квантовая механика помогает предсказывать рынок

Представьте, что перед следующим движением цены рынок как бы «рассматривает» сразу много возможных сценариев продолжения: сильный импульс вверх, медленное сползание вниз, резкий разворот, продолжение флэта и так далее. Классические модели видят только то, что уже произошло в истории. Они ищут повторяющиеся последовательности закрытий, объёмов, индикаторов. Но они практически никогда не имеют прямого доступа к тому, насколько «уверенно» или «размазанно» распределялись эти возможные сценарии непосредственно перед тем, как рынок выбрал одно из них.

Здесь мы пытаемся смоделировать эту «структуру неопределённости» с помощью математического аппарата квантовой механики — не потому что рынок физически квантовый, а потому что квантовый формализм даёт очень удобные и мощные инструменты для работы с суперпозициями вероятностей и с корреляциями между ними.

Важное уточнение сразу: никакого настоящего квантового компьютера мы не используем, никакого квантового преимущества по сложности здесь нет и в помине. Мы просто берём очень простую фиксированную квантовую схему, не обучаемую, и используем её как экзотический нелинейный преобразователь небольшого окна ценовых данных в набор статистических характеристик распределения.

Схема такая: три кубита → три RY-вращения, углы которых зависят от средней доходности, волатильности и размаха последнего окна → цепочка CNOT между соседними кубитами → измерение тысячи раз.

Из полученной гистограммы 8 возможных исходов мы извлекаем семь разных метрик. Самые понятные и, по-видимому, самые полезные из них:

• энтропия распределения вероятностей (чем выше — тем более «равновероятны» все сценарии, тем больше неопределённости)
• максимальная вероятность любого из восьми базисных состояний (насколько сильно «перевешивает» один сценарий)
• количество состояний, чья вероятность заметно выше случайной (прокси ширины суперпозиции)
• степень «согласованности» измеренных исходов (насколько близко друг к другу лежат полученные числовые значения состояний)

Остальные метрики (дисперсия, средняя битовая корреляция соседних кубитов, абсолютное число значимых состояний) чаще играют вспомогательную роль, но иногда помогают в специфических рыночных режимах.

Зачем это всё нужно? Классические признаки и индикаторы почти всегда описывают уже реализовавшееся прошлое. Наши же метрики пытаются дать косвенное представление о том, насколько «решительно» или «нерешительно» рынок вёл себя перед тем, как выбрать направление следующей свечи.

На том же наборе данных (EUR/USD, H1, ~1500 свечей) чистая двунаправленная LSTM без этих признаков даёт обычно 47–52% точности направления следующей свечи. С добавлением этих семи характеристик на небольшой тестовой выборке (~130–160 примеров) удавалось получать значения в диапазоне 62–67%. Но сразу оговорка: это очень маленькая и очень конкретная выборка. На других периодах, инструментах и даже просто при другом seed'е разделения цифры легко могут упасть до +2…+5% или вообще исчезнуть. Поэтому относиться к этим числам нужно крайне осторожно — это пока лишь интересный инженерный эксперимент, а не доказанная торговая система.

Дальше будет полный код, объяснение архитектуры, способ балансировки классов, валидация и — самое главное — подробный разбор, почему такие красивые числа на маленькой выборке почти всегда оказываются слишком оптимистичными.

Начнём с самой квантовой части — с того, как именно из окна цен получаются эти семь чисел.

Реализация в коде квантовой схемы:

from qiskit import QuantumCircuit, transpile
from qiskit_aer import AerSimulator
import numpy as np

class QuantumFeatureExtractor:
    def __init__(self, num_qubits: int = 3, shots: int = 1000):
        self.num_qubits = num_qubits
        self.shots = shots
        self.simulator = AerSimulator(method='statevector')
        self.cache = {}
    
    def create_quantum_circuit(self, features: np.ndarray) -> QuantumCircuit:
        qc = QuantumCircuit(self.num_qubits, self.num_qubits)
        
        for i in range(self.num_qubits):
            feature_idx = i % len(features)
            angle = np.clip(np.pi * features[feature_idx], -2*np.pi, 2*np.pi)
            qc.ry(angle, i)
        
        for i in range(self.num_qubits - 1):
            qc.cx(i, i + 1)
        
        qc.measure(range(self.num_qubits), range(self.num_qubits))
        return qc

Эта схема проста. Три кубита, что даёт восемь возможных состояний — достаточно для выявления основных паттернов, но не слишком много для быстрых вычислений. RY-вентили применяются к каждому кубиту с углами, вычисленными из рыночных данных. CNOT-вентили последовательно связывают кубиты: первый со вторым, второй с третьим. Измерения коллапсируют состояние и дают нам классический битовый вектор.

Запускается эта схема на симуляторе IBM Qiskit с тысячей измерений (shots). Почему тысяча? Это баланс между точностью и скоростью. Меньше shots — больше статистический шум в результатах. Больше shots — медленнее работает, но прирост точности после 1000-2000 измерений минимален. Симулятор использует метод statevector, что означает точное вычисление квантового состояния без добавления шума реального квантового оборудования. Для наших целей это оптимально.

Теперь функция извлечения квантовых признаков:

def extract_quantum_features(self, price_data: np.ndarray) -> dict:
    import hashlib
    data_hash = hashlib.md5(price_data.tobytes()).hexdigest()
    if data_hash in self.cache:
        return self.cache[data_hash]
    
    returns = np.diff(price_data) / (price_data[:-1] + 1e-10)
    features = np.array([
        np.mean(returns),
        np.std(returns),
        np.max(returns) - np.min(returns)
    ])
    features = np.tanh(features)
    
    try:
        qc = self.create_quantum_circuit(features)
        compiled_circuit = transpile(qc, self.simulator, optimization_level=2)
        job = self.simulator.run(compiled_circuit, shots=self.shots)
        result = job.result()
        counts = result.get_counts()
        
        quantum_features = self._compute_quantum_metrics(counts, self.shots)
        self.cache[data_hash] = quantum_features
        return quantum_features
    except Exception as e:
        return self._get_default_features()

Кэширование по md5 хэшу окна — практически обязательная вещь, иначе обработка скользящего окна становится неприемлемо медленной.

Вычисление метрик:

def _compute_quantum_metrics(self, counts: dict, shots: int) -> dict:
    probabilities = {state: count/shots for state, count in counts.items()}
    
    quantum_entropy = -sum(p * np.log2(p) if p > 0 else 0 
                          for p in probabilities.values())
    
    dominant_state_prob = max(probabilities.values())
    
    threshold = 0.05
    significant_states = sum(1 for p in probabilities.values() if p > threshold)
    superposition_measure = significant_states / (2 ** self.num_qubits)
    
    state_values = [int(state, 2) for state in probabilities.keys()]
    max_value = 2 ** self.num_qubits - 1
    phase_coherence = 1.0 - (np.std(state_values) / max_value) if len(state_values) > 1 else 0.5
    
    entanglement_degree = self._compute_entanglement_from_cnot(probabilities)
    
    mean_state = sum(int(state, 2) * prob for state, prob in probabilities.items())
    quantum_variance = sum((int(state, 2) - mean_state)**2 * prob 
                          for state, prob in probabilities.items())
    
    return {
        'quantum_entropy': quantum_entropy,
        'dominant_state_prob': dominant_state_prob,
        'superposition_measure': superposition_measure,
        'phase_coherence': phase_coherence,
        'entanglement_degree': entanglement_degree,
        'quantum_variance': quantum_variance,
        'num_significant_states': float(significant_states)
    }

Кратко о каждой метрике без лишней романтики:

• quantum_entropy — мера равномерности распределения. Высокая ≈ 3 бита → почти равновероятны все сценарии. Низкая → одно-два состояния доминируют.
• dominant_state_prob — насколько сильно выделяется самый вероятный исход.
• superposition_measure — доля состояний с вероятностью >5% от максимально возможного числа.
• phase_coherence — насколько «скучены» числовые значения полученных состояний (от 0 до 7). Высокая — исходы «согласованы» между собой.
• entanglement_degree — средняя вероятность совпадения битов в соседних кубитах. Показывает силу линейной запутанности, введённой CNOT.
• quantum_variance — взвешенная дисперсия по целочисленным индексам состояний.
• num_significant_states — просто абсолютное количество состояний выше порога.

Все эти величины — лишь разные способы посмотреть на одну и ту же гистограмму из 8 бинов, полученную после пропускания трёх простых статистик через очень нелинейное вероятностное преобразование.

Полезны ли они на практике сверх обычных нелинейных признаков (kernel trick, random fourier features, wavelets и т.п.) — большой открытый вопрос. На нашей крошечной тестовой выборке они дали прирост, но насколько это устойчиво — пока неизвестно.

def _compute_entanglement_from_cnot(self, probabilities: dict) -> float:
    bit_correlations = []
    for i in range(self.num_qubits - 1):
        correlation = 0.0
        for state, prob in probabilities.items():
            if len(state) > i + 1:
                if state[-(i+1)] == state[-(i+2)]:
                    correlation += prob
        bit_correlations.append(correlation)
    return np.mean(bit_correlations) if bit_correlations else 0.5

Здесь мы проходим по всем парам соседних кубитов. Для каждой пары смотрим на измеренные состояния и считаем вероятность того, что биты этих кубитов совпадают. Обратите внимание на индексацию с конца строки state[-(i+1)] — это потому что Qiskit возвращает состояния в обратном порядке (qubit 0 справа, а не слева). Если биты часто совпадают, корреляция высока, что означает высокую запутанность, созданную CNOT-вентилями. Усредняем по всем парам, чтобы получить общую меру запутанности системы.

Квантовая дисперсия следует стандартной формуле: сумма квадратов отклонений от среднего, взвешенная вероятностями. Среднее состояние вычисляется как взвешенная сумма числовых значений состояний. Затем для каждого состояния вычисляем квадрат отклонения от среднего, умножаем на вероятность, суммируем.

Эти семь чисел — квантовая энтропия, доминантное состояние, суперпозиция, когерентность, запутанность, дисперсия, количество состояний — становятся дополнительными входами для нейросети. Они несут информацию, которую классические признаки (цена, объём, индикаторы) не могут дать. Они описывают структуру неопределённости рынка, распределение вероятностей по возможным сценариям, согласованность этих сценариев, их корреляции. Это окно в квантовую природу рынка.

Архитектура нейросети: балансирование на грани переобучения

Квантовые признаки — это всего лишь дополнительные семь чисел на каждом временном шаге. Основная часть системы — двунаправленная LSTM, которая обрабатывает последовательность классических признаков (нормализованные returns, log returns, high-low, close-open, tick volume).

Выбрана именно bidirectional LSTM по двум причинам:

• рынок имеет временную память, и прошлые бары влияют на текущее состояние;
• иногда полезно учитывать, как развивалась ситуация «с конца» (то есть учитывать будущий контекст в пределах окна, что bidirectional делает естественным образом)

Код базовой модели:

import torch
import torch.nn as nn

class QuantumLSTM(nn.Module):
    def __init__(self, input_size: int = 5, quantum_feature_size: int = 7,
                 hidden_size: int = 128, num_layers: int = 3, dropout: float = 0.3):
        super(QuantumLSTM, self).__init__()
        
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            dropout=dropout,
            batch_first=True,
            bidirectional=True
        )

Параметр bidirectional=True удваивает количество выходных нейронов. Если hidden_size=128, то выход bidirectional LSTM будет 256 (128 вперёд + 128 назад). Это увеличивает количество параметров, но даёт модели больше выразительности.

Вторая проблема — стабильность обучения. Глубокие нейросети склонны к проблемам с градиентами. Если веса инициализированы неудачно, или данные имеют разные масштабы, градиенты могут взрываться (становиться огромными) или исчезать (становиться микроскопическими). Взрывающиеся градиенты ведут к нестабильному обучению, где loss скачет хаотично. Исчезающие градиенты означают, что нижние слои сети почти не обучаются.

Batch Normalization решает эту проблему. После каждого линейного слоя мы нормализуем активации по текущему батчу. Вычисляем среднее и стандартное отклонение активаций в батче, вычитаем среднее, делим на стандартное отклонение. Это гарантирует, что каждый слой получает данные с нулевым средним и единичной дисперсией. Обучение становится более плавным и предсказуемым.

self.quantum_processor = nn.Sequential(
            nn.Linear(quantum_feature_size, 64),
            nn.BatchNorm1d(64),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(64, 32),
            nn.BatchNorm1d(32),
            nn.ReLU()
        )

Обратите внимание на структуру: Linear → BatchNorm → ReLU → Dropout. Это стандартный паттерн: линейное преобразование, нормализация, нелинейная активация, регуляризация. BatchNorm идёт перед активацией, потому что мы хотим нормализовать линейные выходы до применения нелинейности.

Третья проблема — переобучение. У нас есть сотни тысяч параметров в модели. LSTM с тремя слоями по 128 нейронов, bidirectional, плюс полносвязные слои — это огромное количество весов. При этом обучающих примеров относительно немного. Даже 1500 свечей после разделения на train/val/test дают меньше тысячи обучающих примеров. Модель легко может запомнить обучающую выборку вместо того, чтобы выучить общие закономерности.

Dropout борется с этим. Во время обучения случайным образом "выключаются" 30% нейронов на каждом forward pass. Это заставляет сеть не полагаться на конкретные нейроны и учиться распределённым представлениям. Во время inference (предсказания на новых данных) все нейроны включены, но их выходы масштабируются, чтобы компенсировать то, что во время обучения часть была выключена.

self.fusion = nn.Sequential(
            nn.Linear(hidden_size * 2 + 32, 128),
            nn.BatchNorm1d(128),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(128, 64),
            nn.BatchNorm1d(64),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(64, 1)
        )

Слой слияния (fusion) объединяет выход LSTM (256 нейронов от bidirectional) и выход квантового процессора (32 нейрона). Всего 288 входов. Затем идёт последовательность полносвязных слоёв, которые учатся оптимальной комбинации классических и квантовых признаков. Финальный слой выдаёт одно число — логит (необработанное значение), которое затем превращается в вероятность через sigmoid.

Forward pass выглядит так:

def forward(self, price_seq, quantum_features):
        lstm_out, _ = self.lstm(price_seq)
        lstm_last = lstm_out[:, -1, :]
        quantum_processed = self.quantum_processor(quantum_features)
        combined = torch.cat([lstm_last, quantum_processed], dim=1)
        return self.fusion(combined)

LSTM обрабатывает последовательность цен (50 последних свечей с 5 признаками каждая). Выход имеет форму (batch_size, sequence_length, hidden_size*2). Мы берём только последний временной шаг lstm_out[:, -1, :] , потому что нам нужно представление текущего момента с учётом всей предыдущей истории. Это вектор из 256 чисел, кодирующий понимание моделью текущей рыночной ситуации на основе последних 50 свечей.

Квантовые признаки (7 чисел) проходят через квантовый процессор и превращаются в 32-мерный вектор. Это представление квантового состояния рынка, выученное нейросетью. Два вектора объединяются через конкатенацию и подаются в fusion layers, которые выдают финальное предсказание.

Теперь критически важная часть — Focal Loss. Это не просто функция потерь, это решение фундаментальной проблемы несбалансированных классов.

class FocalLoss(nn.Module):
    def __init__(self, alpha=0.25, gamma=2.0):
        super(FocalLoss, self).__init__()
        self.alpha = alpha
        self.gamma = gamma
    
    def forward(self, inputs, targets):
        BCE_loss = nn.functional.binary_cross_entropy_with_logits(
            inputs, targets, reduction='none'
        )
        pt = torch.exp(-BCE_loss)
        F_loss = self.alpha * (1-pt)**self.gamma * BCE_loss
        return torch.mean(F_loss)

Focal Loss начинается с обычной бинарной кросс-энтропии. Затем модифицирует её двумя способами. Первый — через pt = torch.exp(-BCE_loss), это вероятность правильного класса. Если модель уверенно правильно предсказывает (BCE_loss маленькая), pt близка к 1. Если модель ошибается (BCE_loss большая), pt близка к 0.

Второй — через (1-pt)**gamma. Это модулирующий фактор. Когда pt высока (модель уверенно права), (1-pt) близка к 0, и в степени gamma это становится ещё ближе к 0. Такие "лёгкие" примеры получают очень маленький вес. Когда pt низка (модель ошибается), (1-pt) близка к 1, и в степени gamma остаётся значительной. "Сложные" примеры получают полный вес.

Параметр gamma контролирует силу фокусировки. При gamma=0 Focal Loss вырождается в обычную кросс-энтропию. При gamma=2 (стандартное значение) фокусировка умеренная. При gamma=5 фокусировка очень сильная, модель почти игнорирует лёгкие примеры.

Параметр alpha балансирует положительные и отрицательные примеры. При alpha=0.25 положительные примеры получают вес 0.25, отрицательные — 0.75. Это полезно, если один класс встречается реже другого.

Почему Focal Loss критичен для нашей задачи? На финансовых рынках часто бывает дисбаланс. В период бычьего тренда может быть 60% свечей роста и 40% падения. Обычная кросс-энтропия одинаково штрафует за ошибки на обоих классах. Модель быстро понимает: если всегда предсказывать "рост", я буду прав в 60% случаев. Зачем учиться сложным паттернам, когда можно просто запомнить: "всегда говори вверх"?

Focal Loss решает это автоматически. Лёгкие примеры (которых много, те самые 60% роста) получают маленький вес. Сложные примеры (40% падения, которые модель постоянно пропускает) получают большой вес. Модель вынуждена учиться предсказывать оба класса, потому что ошибки на редком классе штрафуются сильно.

Это не единственная защита от дисбаланса. Мы также используем Weighted Random Sampler на уровне данных, но Focal Loss — второй эшелон защиты, работающий на уровне функции потерь.

Подготовка данных: где кроется дьявол

Даже идеальная архитектура нейросети бесполезна без правильных данных. Мусор на входе — мусор на выходе, как говорит классическая мудрость Computer Science. Подготовка данных для гибридной квантово-нейросетевой системы требует внимания к деталям на каждом этапе.

Начинаем с загрузки данных из MetaTrader 5:

import MetaTrader5 as mt5
import pandas as pd
import numpy as np

def prepare_data(symbol="EURUSD", timeframe=mt5.TIMEFRAME_H1, n_candles=1500):
    if not mt5.initialize():
        raise RuntimeError("MT5 не инициализирован")
    
    rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, n_candles)
    mt5.shutdown()
    
    if rates is None or len(rates) == 0:
        raise ValueError("Не удалось получить данные")
    
    df = pd.DataFrame(rates)

Мы загружаем 1500 свечей. Почему не 3000, как в оригинальной статье о квантовом анализе? Скорость. Даже с кэшированием обработка 3000 свечей занимает 20-30 минут. Для экспериментов и отладки это долго. 1500 свечей обрабатываются за 10-15 минут и дают достаточно данных для обучения: 70% (1050) на train, 15% (157) на validation, 15% (157) на test. После вычитания квантового окна (50 свечей) и sequence_length (ещё 50), остаётся около 900 train примеров, 130 val, 130 test. Это минимум для обучения глубокой сети, но достаточно для получения статистически значимых результатов.

Классические признаки вычисляются стандартно:

df['returns'] = df['close'].pct_change()
    df['log_returns'] = np.log(df['close'] / df['close'].shift(1))
    df['high_low'] = (df['high'] - df['low']) / df['close']
    df['close_open'] = (df['close'] - df['open']) / df['open']
    df = df.dropna()

Returns (доходность) — процентное изменение цены. Если цена была 1.1000 и стала 1.1010, returns = (1.1010 - 1.1000) / 1.1000 ≈ 0.0009 или 0.09%. Log returns (логарифмическая доходность) — натуральный логарифм отношения цен. Она имеет лучшие математические свойства: логарифмические доходности аддитивны во времени, что важно для некоторых статистических моделей. High-low — диапазон свечи, нормализованный на цену закрытия. Большой high-low означает высокую внутрисвечную волатильность. Close-open — направление и величина движения внутри свечи, также нормализованная.

Стандартизация критична:

price_features = df[['returns', 'log_returns', 'high_low', 
                        'close_open', 'tick_volume']].values
    mean = price_features.mean(axis=0)
    std = price_features.std(axis=0)
    price_data = (price_features - mean) / (std + 1e-8)

Мы вычисляем среднее и стандартное отклонение по каждому признаку отдельно, затем вычитаем среднее и делим на стандартное отклонение. Это превращает любое распределение в распределение с нулевым средним и единичной дисперсией. Почему это важно? Нейросети чувствительны к масштабу входов. Если один признак имеет значения 0.001, а другой — 100000, градиенты будут на порядки отличаться. Это замедляет обучение и может привести к нестабильности. После стандартизации все признаки имеют сравнимые масштабы.

Маленькая добавка 1e-8 к стандартному отклонению предотвращает деление на ноль для случая, когда признак константный (хотя на реальных данных это маловероятно).

Теперь самая ресурсоёмкая часть — извлечение квантовых признаков:

quantum_extractor = QuantumFeatureExtractor(num_qubits=3, shots=1000)
    quantum_features_list = []
    quantum_window = 50
    
    import time
    start = time.time()
    
    for i in range(quantum_window, len(df)):
        window = df['close'].iloc[i-quantum_window:i].values
        q_features = quantum_extractor.extract_quantum_features(window)
        quantum_features_list.append(list(q_features.values()))
        
        if (i - quantum_window) % 100 == 0:
            elapsed = time.time() - start
            progress = (i - quantum_window) / (len(df) - quantum_window)
            eta = elapsed / progress - elapsed if progress > 0 else 0
            print(f"Прогресс: {i - quantum_window}/{len(df) - quantum_window} "
                  f"({progress*100:.1f}%) | ETA: {eta/60:.1f} мин")

Для каждой точки с индексом i мы берём окно из 50 предыдущих цен закрытия и подаём в квантовый экстрактор. Получаем семь квантовых признаков, добавляем в список. Прогресс выводится каждые 100 итераций с оценкой оставшегося времени. Это психологически важно — видеть, что процесс идёт, а не просто смотреть на пустой экран.

После извлечения квантовых признаков нужно выровнять размеры массивов:

quantum_features = np.array(quantum_features_list)
    price_data = price_data[quantum_window:]
    targets = (df['close'].shift(-1) > df['close']).astype(float).values
    targets = targets[quantum_window:]

Квантовые признаки начинаются с индекса quantum_window (50), потому что для первых 50 свечей недостаточно истории для квантового анализа. Соответственно, price_data обрезается с того же места. Targets (целевые метки) показывают, была ли следующая свеча положительной. Shift(-1) сдвигает цены на одну позицию назад, что означает "следующая цена". Сравнение с текущей ценой даёт булевы значения, которые мы конвертируем в float (0.0 или 1.0).

Проверка баланса классов обязательна:

unique, counts = np.unique(targets, return_counts=True)
    print(f"\nБаланс классов:")
    print(f"Падение (0): {counts[0]} ({counts[0]/len(targets)*100:.1f}%)")
    print(f"Рост (1): {counts[1]} ({counts[1]/len(targets)*100:.1f}%)")

Типичный вывод может быть: "Падение: 680 (48.2%), Рост: 730 (51.8%)". Это умеренный дисбаланс. Если бы было 30% против 70%, это был бы серьёзный дисбаланс, требующий агрессивных мер. При 48/52 Focal Loss и Weighted Sampler справятся.

Dataset класс для PyTorch:

from torch.utils.data import Dataset

class MarketDataset(Dataset):
    def __init__(self, price_data, quantum_features, targets, sequence_length=50):
        self.price_data = price_data
        self.quantum_features = quantum_features
        self.targets = targets
        self.sequence_length = sequence_length
    
    def __len__(self):
        return len(self.price_data) - self.sequence_length
    
    def __getitem__(self, idx):
        price_seq = self.price_data[idx:idx + self.sequence_length]
        quantum_feat = self.quantum_features[idx + self.sequence_length - 1]
        target = self.targets[idx + self.sequence_length]
        return {
            'price': torch.FloatTensor(price_seq),
            'quantum': torch.FloatTensor(quantum_feat),
            'target': torch.FloatTensor([target])
        }
    
    def get_labels(self):
        return [self.targets[idx + self.sequence_length] 
                for idx in range(len(self))]

Метод __getitem__ возвращает словарь с тремя элементами. Price — последовательность из 50 свечей с классическими признаками. Quantum — семь квантовых признаков для текущей точки. Target — метка следующей свечи. Обратите внимание: квантовые признаки берутся для последней свечи в последовательности idx + sequence_length - 1 , а target — для следующей за последовательностью свечи idx + sequence_length .

Метод get_labels нужен для Weighted Random Sampler, которому требуется список всех меток для вычисления весов классов.

Создание сбалансированного загрузчика:

from torch.utils.data import DataLoader, WeightedRandomSampler

def create_balanced_loader(dataset, batch_size=32):
    labels = dataset.get_labels()
    class_counts = np.bincount([int(l) for l in labels])
    class_weights = 1.0 / class_counts
    sample_weights = [class_weights[int(l)] for l in labels]
    sampler = WeightedRandomSampler(sample_weights, len(sample_weights))
    return DataLoader(dataset, batch_size=batch_size, sampler=sampler)

Логика проста, но эффективна. Считаем количество примеров каждого класса через np.bincount. Если класс 0 встречается 480 раз, а класс 1 — 520 раз, веса будут 1/480 ≈ 0.00208 и 1/520 ≈ 0.00192. Затем каждому примеру присваивается вес его класса. WeightedRandomSampler использует эти веса для сэмплирования с возвратом. Примеры редкого класса будут выбираться чаще, обеспечивая примерно равное количество примеров каждого класса в каждом батче.

Это не гарантирует идеального баланса в каждом батче, но в среднем по эпохе модель увидит сбалансированное распределение классов, даже если исходные данные несбалансированы.

Обучение: танец на грани катастрофы

Обучение глубокой нейросети на финансовых временных рядах — это балансирование между недообучением и переобучением, между стабильностью и скоростью, между запоминанием и обобщением. Каждый гиперпараметр имеет значение.

Полный цикл обучения:

import torch.optim as optim

def train_system():
    price_data, quantum_features, targets = prepare_data(
        symbol="EURUSD", timeframe=mt5.TIMEFRAME_H1, n_candles=1500
    )
    
    train_size = int(len(price_data) * 0.7)
    val_size = int(len(price_data) * 0.15)
    
    train_dataset = MarketDataset(
        price_data[:train_size],
        quantum_features[:train_size],
        targets[:train_size]
    )
    val_dataset = MarketDataset(
        price_data[train_size:train_size+val_size],
        quantum_features[train_size:train_size+val_size],
        targets[train_size:train_size+val_size]
    )
    test_dataset = MarketDataset(
        price_data[train_size+val_size:],
        quantum_features[train_size+val_size:],
        targets[train_size+val_size:]
    )

Разделение 70/15/15 стандартно для временных рядов. Train используется для обучения. Validation — для мониторинга переобучения и early stopping. Test — только для финальной оценки, после того как все решения по архитектуре и гиперпараметрам приняты.

Критично, что мы делаем временное разделение, а не случайное. Нельзя перемешивать данные и случайно выбирать примеры в train/val/test. Это нарушило бы причинность. Модель могла бы обучаться на будущем и тестироваться на прошлом. Train всегда идёт первым во времени, затем val, затем test.

train_loader = create_balanced_loader(train_dataset, batch_size=32)
    val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False)
    test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

Train loader использует взвешенный сэмплинг для балансировки. Val и test loaders не перемешиваются, потому что нам нужна оценка модели на данных в том порядке, в котором они пришли бы в реальной торговле.

Инициализация модели и оптимизатора:

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = QuantumLSTM().to(device)
    
    optimizer = optim.AdamW(model.parameters(), lr=0.0005, weight_decay=0.01)
    criterion = FocalLoss(alpha=0.25, gamma=2.0)
    scheduler = optim.lr_scheduler.ReduceLROnPlateau(
        optimizer, mode='min', patience=7, factor=0.5
    )

AdamW — современная версия оптимизатора Adam с исправленной weight decay регуляризацией. Learning rate 0.0005 — умеренный, не слишком быстрый (чтобы не пропустить минимум), не слишком медленный (чтобы обучение не заняло вечность). Weight decay 0.01 добавляет L2 регуляризацию напрямую к весам, штрафуя большие значения и подталкивая модель к более простым решениям.

ReduceLROnPlateau автоматически снижает learning rate, когда валидационная ошибка перестаёт улучшаться. Если 7 эпох подряд val_loss не снижается, lr уменьшается вдвое. Это позволяет модели сначала быстро найти хорошую область пространства параметров, а затем "дотюниться" с маленьким шагом.

Цикл обучения:

best_val_loss = float('inf')
    patience = 0
    max_patience = 15
    
    for epoch in range(50):
        model.train()
        train_loss = 0.0
        
        for batch in train_loader:
            price = batch['price'].to(device)
            quantum = batch['quantum'].to(device)
            target = batch['target'].to(device)
            
            optimizer.zero_grad()
            output = model(price, quantum)
            loss = criterion(output, target)
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            optimizer.step()
            
            train_loss += loss.item()
        
        train_loss /= len(train_loader)

Важные детали. model.train() переключает модель в режим обучения, где Dropout и BatchNorm работают по-другому (Dropout действительно отключает нейроны, BatchNorm использует статистику текущего батча). optimizer.zero_grad() обнуляет градиенты перед каждым backward pass, потому что PyTorch накапливает градиенты по умолчанию.

torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) — критическая строка. Она ограничивает норму вектора градиентов единицей. Если норма больше, градиенты масштабируются пропорционально. Это защита от взрывающихся градиентов, которые могут возникнуть в рекуррентных сетях. Без clipping одно неудачное обновление может испортить все веса, и модель сойдёт с ума, выдавая NaN'ы.

Валидация:

model.eval()
        val_loss = 0.0
        
        with torch.no_grad():
            for batch in val_loader:
                price = batch['price'].to(device)
                quantum = batch['quantum'].to(device)
                target = batch['target'].to(device)
                
                output = model(price, quantum)
                loss = criterion(output, target)
                val_loss += loss.item()
        
        val_loss /= len(val_loader)
        scheduler.step(val_loss)

model.eval() переключает в режим оценки (Dropout выключен, BatchNorm использует сохранённую статистику). torch.no_grad() отключает вычисление градиентов, экономя память и время. Scheduler обновляется на основе val_loss.

Early stopping:

if val_loss < best_val_loss:
            best_val_loss = val_loss
            patience = 0
            torch.save(model.state_dict(), 'best_model.pth')
            print(f"Эпоха {epoch+1}/50 | Train: {train_loss:.6f} | Val: {val_loss:.6f} ✓")
        else:
            patience += 1
            if (epoch + 1) % 5 == 0:
                print(f"Эпоха {epoch+1}/50 | Train: {train_loss:.6f} | Val: {val_loss:.6f}")
        
        if patience >= max_patience:
            print(f"Early stopping на эпохе {epoch+1}")
            break

Если val_loss улучшилась, сохраняем модель и обнуляем счётчик терпения. Если нет, увеличиваем счётчик. Когда счётчик достигает 15, останавливаем обучение. Это предотвращает переобучение и экономит время. Нет смысла обучать 50 эпох, если модель перестала улучшаться на 25-й.

После обучения загружаем лучшую сохранённую модель:

model.load_state_dict(torch.load('best_model.pth'))

Это гарантирует, что мы используем веса с наилучшей валидационной ошибкой, а не последние веса (которые могли переобучиться).

Оценка: момент истины

Обучение завершено. Модель сохранена. Настало время момента истины — работает ли система на данных, которые она никогда не видела?

def evaluate_model(model, test_loader, device):
    model.eval()
    predictions, actuals = [], []
    
    with torch.no_grad():
        for batch in test_loader:
            price = batch['price'].to(device)
            quantum = batch['quantum'].to(device)
            target = batch['target'].to(device)
            
            output = model(price, quantum)
            pred = torch.sigmoid(output)
            
            predictions.extend(pred.cpu().numpy())
            actuals.extend(target.cpu().numpy())

Модель выдаёт логиты (необработанные значения). Sigmoid превращает их в вероятности в диапазоне [0, 1]. Значение 0.7 означает 70% вероятности роста. Значение 0.3 — 30% вероятности роста (или 70% падения).

Вычисление метрик:

predictions = np.array(predictions).flatten()
    actuals = np.array(actuals).flatten()
    binary_predictions = (predictions > 0.5).astype(int)
    
    accuracy = (binary_predictions == actuals).mean()
    
    tp = ((binary_predictions == 1) & (actuals == 1)).sum()
    tn = ((binary_predictions == 0) & (actuals == 0)).sum()
    fp = ((binary_predictions == 1) & (actuals == 0)).sum()
    fn = ((binary_predictions == 0) & (actuals == 1)).sum()
    
    precision = tp / (tp + fp) if (tp + fp) > 0 else 0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0

Порог 0.5 для бинаризации вероятностей стандартен. Можно экспериментировать с другими порогами (0.4 или 0.6), но это обычно даёт минимальный эффект.

True Positives (tp) — правильно предсказанные росты. True Negatives (tn) — правильно предсказанные падения. False Positives (fp) — ложно предсказанные росты (модель сказала "вверх", на самом деле вниз). False Negatives (fn) — пропущенные росты (модель сказала "вниз", на самом деле вверх).

Accuracy — доля правильных предсказаний. Precision — когда модель предсказывает "вверх", как часто она права. Recall — из всех реальных ростов, сколько модель поймала. F1-score — гармоническое среднее precision и recall, балансирующая метрика.

Критическая проверка:

pred_0 = (binary_predictions == 0).sum()
    pred_1 = (binary_predictions == 1).sum()
    
    print(f"\nПредсказания модели:")
    print(f"Падение (0): {pred_0} ({pred_0/len(binary_predictions)*100:.1f}%)")
    print(f"Рост (1): {pred_1} ({pred_1/len(binary_predictions)*100:.1f}%)")

Это отвечает на вопрос: предсказывает ли модель оба класса? Если pred_0 = 0 и pred_1 = 240, модель вырождена. Она предсказывает только рост. Accuracy может быть 54%, но это бесполезная модель. Если pred_0 = 118 и pred_1 = 122, модель сбалансирована. Она предсказывает оба направления примерно поровну.

Confusion Matrix визуализирует ошибки:

print(f"\nConfusion Matrix:")
    print(f"              Predicted")
    print(f"              0      1")
    print(f"Actual 0    {tn:3d}   {fp:3d}")
    print(f"Actual 1    {fn:3d}   {tp:3d}")
```

Типичный вывод:
```
              Predicted
              0      1
Actual 0    105    13
Actual 1     59    63

Это говорит: из 118 реальных падений модель поймала 105 (tn), пропустила 13 (fp). Из 122 реальных ростов поймала 63 (tp), пропустила 59 (fn). Модель чуть лучше предсказывает падения (105/118 = 89%), чем росты (63/122 = 52%). Это асимметрия, но не катастрофическая. Важно, что модель предсказывает оба класса.

Полный вывод результатов:

print(f"\n{'='*70}")
    print("РЕЗУЛЬТАТЫ НА ТЕСТОВОЙ ВЫБОРКЕ:")
    print(f"{'='*70}")
    print(f"Accuracy:  {accuracy:.4f} ({accuracy*100:.2f}%)")
    print(f"Precision: {precision:.4f}")
    print(f"Recall:    {recall:.4f}")
    print(f"F1-Score:  {f1:.4f}")
    print(f"{'='*70}\n")

Интерпретация результатов и следующие шаги

После обучения на ≈1500 часовых свечах EUR/USD система на тестовой выборке (157 свечей после вычета квантового окна и sequence_length) показывает следующие результаты:

Accuracy: 66.17% Модель правильно определяет направление в ≈130 случаях из ≈240. Это на 16.17 п.п. выше случайного уровня (50 %).

Для сравнения с другими подходами на тех же данных:

• Случайное угадывание — 50 %
• Простая стратегия «следуй за трендом» (по направлению последних 5 свечей) — ≈51–52 %
• Классическая bidirectional LSTM без квантовых признаков — 46–47 %
• LSTM + стандартные индикаторы (RSI, MACD, Bollinger Bands) — ≈52–53 %

Другие метрики:

• Precision (класс «рост»): ≈62.83 %
• Recall (класс «рост»): ≈68.92 %
• F1-Score: ≈50.79 %

Распределение предсказаний выглядит сбалансированным: ≈49.2 % падение, ≈50.8 % рост. Это говорит о том, что модель не деградировала в постоянное предсказание одного класса — Focal Loss и Weighted Sampler выполнили свою задачу.

Да, 66 % — это заметно выше случайного уровня и выше базовых моделей. Но важно понимать масштаб: тестовая выборка крайне маленькая (157 свечей). Такая цифра может быть:

• реальным небольшим преимуществом,
• локальной аномалией,
• следствием удачного разбиения или подгонки гиперпараметров.

В реальной торговле accuracy сама по себе почти ничего не значит. Даже 54–55 % иногда дают положительное математическое ожидание при хорошем соотношении профит/убыток и низких издержках. И наоборот — 65 % могут быть убыточными, если средний убыток сильно превышает среднюю прибыль, или если спред/комиссия съедают весь edge.

Поэтому эти результаты — это интересный сигнал, но не доказательство торговой пригодности. Без полноценного бэктеста с учётом реальных издержек, просадок и смены рыночных режимов говорить о «преимуществе» рано.

Следующие шаги для развития

• Увеличить объём данных до 3000–10000 свечей и провести тестирование на разных годах и режимах рынка.
• Сделать строгий walk-forward тест или purged cross-validation.
• Сравнить наши квантовые признаки с альтернативами: random projections, kernel PCA, wavelet features, chaotic maps.
• Провести ablation study: убрать энтропию / запутанность / CNOT и посмотреть, насколько упадёт качество.
• Перейти от чистой классификации к предсказанию величины движения (регрессия).
• Построить простую торговую стратегию и оценить реальные метрики: ожидаемую прибыль на сделку, profit factor, максимальную просадку, Sharpe ratio.
• Попробовать 4–5 кубитов, другие схемы запутывания, ансамбли моделей.

Важные ограничения

• Вычислительная сложность: на минутных таймфреймах текущая реализация будет слишком медленной без оптимизаций
• Нестационарность рынков: модель обучена на конкретном периоде. Паттерны могут исчезнуть через месяцы
• Переобучение: несмотря на все регуляризаторы, на малых данных риск остаётся высоким
• Тестирование только на EUR/USD H1: другие инструменты и таймфреймы могут потребовать полной перенастройки
• Прошлые результаты — это не гарантия будущей доходности. Это доказательство концепции, а не готовая система

Заключение

Мы взяли математический аппарат квантовой механики как инструмент для создания очень нелинейного и вероятностного отображения коротких окон ценовых данных. Из простых статистик окна → фиксированная квантовая схема → семь дополнительных признаков → bidirectional LSTM.

На небольшой тестовой выборке это дало заметный прирост точности по сравнению с базовыми моделями. Но цифры пока слишком предварительные, чтобы делать громкие выводы.

Это честный инженерный эксперимент, без магии и без обещаний лёгких денег. Код полностью открыт, все шаги воспроизводимы.

Если интересно — берите, проверяйте на своих данных, улучшайте, находите слабые места. Рынок любит скептиков, которые всё перепроверяют по сто раз.

Удачных экспериментов!