Архитектура машинного обучения для MetaTrader 5 (Часть 9): Интеграция байесовской оптимизации гиперпараметров в производственный пайплайн

Оглавление

1. Что изменилось и почему
2. Архитектура: два пути обучения, один пайплайн
3. Определение первичной и вторичной модели
4. Обученный препроцессор
5. Переработанный метод train_model
6. _train_model_optuna подробно
7. Режим refit и бэггинг
8. Отчет визуализации Optuna
9. Живой мониторинг с optuna-dashboard (панель мониторинга)
10. Расчет весов выборки в пути Optuna
11. Интеграция кэширования
12. Long-Short пайплайн Bid/Ask
13. LearnedStrategy: связующее звено между двумя этапами
14. Конфигурация model_params
15. Практические соображения
16. Заключение
17. Прикрепленные файлы

Введение

В Части 8 система Optuna HPO была разработана изолированно: целевая функция, отсекатель, оркестратор и набор средств визуализации. Эта статья интегрирует эту систему в производственный пайплайн из Части 7. В результате появляется единый класс ModelDevelopmentPipeline, который может запускать либо исходный бэкенд clf_hyper_fit бэкенд (GridSearchCV / RandomizedSearchCV), либо новый бэкенд Optuna, управляемый одним флагом в model_params.

Помимо основной интеграции HPO, рассматриваются еще семь тем:

1. Теперь пайплайн определяет, обучает ли он первичную направленную модель или вторичную модель мета-разметки, проверяя, присутствует ли side в DataFrame событий. Этот флаг управляет вычислением скользящих мета-признаков и именованием артефактов.
2. Обученный препроцессор: DropConstantFeatures и DropDuplicateFeatures теперь сохраняются и добавляются перед best_model, делая инференс самодостаточным.
3. Последовательный бутстрэппинг: флаг bagging_sequential направляет этап ансамблирования после HPO в SequentiallyBootstrappedBaggingClassifier в обоих путях обучения.
4. Отчет визуализации Optuna: четыре интерактивных графика Plotly и PNG со сравнением с базовой линией автоматически сохраняются после каждого запуска Optuna.
5. Живой мониторинг через optuna-dashboard.
6. Сопутствующий long-short пайплайн bid/ask.
7. LearnedStrategy, подкласс BaseStrategy, который оборачивает обученную первичную модель, чтобы она могла генерировать прогнозы направления для этапа разметки вторичного пайплайна.

Важное предварительное условие — сначала нужно исследовать важность признаков. Описанный в этой статье пайплайн предполагает, что набор признаков, передаваемый в feature_config, уже проверен и отобран. Запуск HPO на плохо построенном наборе признаков не решает исходную проблему — он оптимизирует гиперпараметры шумной модели или модели с утечкой данных, получая результаты, которые хорошо выглядят внутри выборки и проваливаются вне выборки. Анализ важности признаков — серьезная задача. Он требует отделять сигнальные признаки от признаков с look-ahead-смещением и избыточных признаков. Необходимые техники (MDI, MDA, SFI, кластеризованная важность) и их интеграция с purged cross-validation из этой серии станут темой отдельных будущих статей. Не запускайте этот пайплайн, пока исследование важности ваших кандидатных признаков не завершено.

Что изменилось и почему

Исходный метод ModelDevelopmentPipeline.train_model вызывает clf_hyper_fit, который оборачивает GridSearchCV или RandomizedSearchCV. Веса выборки передаются как заранее вычисленный массив, оценка использует те же веса, а лучший пайплайн возвращается напрямую. Это работает, но имеет три ограничения, которые устраняет интеграция Optuna.

Ограничение 1: схема весов фиксируется до HPO. В пути sklearn get_optimal_sample_weight выбирает лучшую схему взвешивания до начала HPO. Поэтому схема весов и гиперпараметры модели оптимизируются последовательно, а не совместно. Конфигурация, которая хорошо работает только с весами return-attribution, может быть исключена лишь потому, что сначала была выбрана uniqueness.

Ограничение 2: нет раннего отсечения. GridSearchCV и RandomizedSearchCV оценивают каждый фолд для каждого испытания. При дорогих PurgedKFold оценках на нескольких годах тиковых данных это расходует вычисления на конфигурации, которые явно хуже уже после фолда 1.

Ограничение 3: нет постоянного хранилища исследования. При сбое запуска все завершенные испытания теряются.

Путь Optuna решает все три проблемы: схема весов, decay и linearity сэмплируются совместно с гиперпараметрами модели внутри _WeightedEstimator; HyperbandPruner отсекает бесперспективные испытания после первого фолда, а хранилище SQLite означает, что повторный запуск продолжается с последнего завершенного испытания.

Интеграция сопровождалась рядом структурных изменений. _WeightedEstimator был вынесен из model_development.py в weighted_estimator.py, устранив циклический импорт с optuna_hyper_fit.py. Утилиты управления файлами вынесены в file_manager.py. Теперь пайплайн автоматически различает первичную и вторичную модель. Обученный препроцессор удаления столбцов стал постоянной частью best_model. Флаг bagging_sequential направляет bagging после HPO в SequentiallyBootstrappedBaggingClassifier. А LearnedStrategy связывает два этапа обучения.

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

Переработанный пайплайн содержит единую диспетчеризацию в train_model:

if self.model_params.get('use_optuna', False):
    self._train_model_optuna()
else:
    self._train_model_sklearn()

# Prepend the fitted preprocessor so best_model.predict(raw_features) is self-contained.
self.best_model = Pipeline([('preprocessor', self.preprocessor), *self.best_model.steps])
self.best_model = set_pipeline_params(self.best_model, n_jobs=-1)
self.completed_steps['model_training'] = True

Рис. 1 — архитектура ModelDevelopmentPipeline v4.0. Шаги 1–6 выполняются одинаково в обоих путях. Ветка is_primary на шаге 3 управляет вычислением мета-признаков. Диспетчеризация HPO на шаге 7 направляет выполнение в один из бэкендов; блок после диспетчеризации добавляет обученный препроцессор перед best_model в обоих случаях.

Все до и после train_model одинаково в обоих путях. Оба пути должны соблюдать один и тот же контракт выходных данных. self.best_model должен быть обученным sklearn Pipeline, первым шагом которого является препроцессор удаления столбцов. self.cv_results должен содержать как минимум best_params, best_score, и cv_results (в формате scikit-learn cv_results_).

Определение первичной и вторичной модели

Пайплайн определяет, какой тип модели обучается, проверяя, присутствует ли side в DataFrame событий — это ровно тот же сигнал, который внутренне использует код разметки. Чтение исходного кода разметки делает это очевидным.

В triple_barrier.py, get_events() явно удаляет столбец side когда side_prediction не передан:

# Primary model path: side_prediction is None
if side_prediction is None:
    side = pd.Series(1.0, index=target.index)
    events = events.drop('side', axis=1)   # ← side absent
else:
    side = side_prediction.reindex(target.index)
    # side retained: secondary / meta-labeling model

А в get_bins(), наличие side определяет пространство меток и режим мета-разметки:

if 'side' in events:
    out_df['ret'] *= events['side']              # meta-labeling: bin ∈ {0, 1}
    out_df.loc[out_df['ret'].values <= 0, 'bin'] = 0
    out_df['side'] = events['side'].astype('int8')
else:
    # primary: bin ∈ {-1, 0, 1}, direction from price action

trend_scanning_labels() никогда не создает столбец side — он всегда генерирует первичные метки с bin ∈ {-1, 0, 1}. Полное правило выглядит так:

32

	Вызов разметки	side в events?	пространство меток bin	Тип модели
1.	Тройной барьер, side_prediction=None	Нет	{-1, 0, 1}	Первичная
2.	Тройной барьер, side_prediction передан	Да	{0, 1}	Вторичная
3.	Сканирование тренда	Нет	{-1, 0, 1}	Первичная

Рис. 2 — поток определения первичной/вторичной модели. Единственный различитель — присутствует ли side в DataFrame событий после этапа разметки — тот же сигнал, который внутренне использует код разметки. Этот флаг управляет вычислением мета-признаков и именованием артефактов в оставшейся части запуска.

Пайплайн устанавливает self.is_primary сразу после generate_labels(). Он также записывает model_role в словарь конфигурации, чтобы это попало в хэш каталога артефактов и в сводку обучения:

def generate_labels(self):
    self.events = generate_events_triple_barrier(
        self.bar_data, self.strategy, self.target_config, **self.label_config
    )
    self.is_primary = 'side' not in self.events.columns
    self.config['model_role'] = 'primary' if self.is_primary else 'secondary'
    logger.info(
        f"Model role: {self.config['model_role']} | "
        f"Label space: {sorted(self.events['bin'].unique().astype('int8'))}"
    )
    logger.info(f"Average uniqueness: ({self.events['tW'].mean():.4f})")
    self.completed_steps['label_generation'] = True

Скользящие мета-признаки пропускаются для первичных моделей, потому что нет предыдущей модели, производительность которой можно отслеживать: эта концепция здесь становится самореферентной и имеет смысл только при оценке того, заслуживают ли сигналы первичной модели исполнения:

def add_meta_features(self):
    if self.is_primary:
        self.meta_features = pd.DataFrame(index=self.events.index)
        logger.info('Primary model — rolling meta-features skipped.')
    else:
        self.meta_features = calculate_rolling_metrics(self.events, self.sample_weight)
    self.completed_steps['meta_features'] = True

И preprocess_features() пропускает внутреннее соединение, когда meta_features пусты — соединение с пустым DataFrame дало бы ноль строк:

if self.meta_features.empty:
    combined = self.features.dropna()
else:
    combined = self.features.join(self.meta_features, how='inner').dropna()

Обученный препроцессор

Предыдущий пайплайн обучал DropConstantFeatures и DropDuplicateFeatures в preprocess_features() и отбрасывал обученные объекты. Сохранялся только preprocessed_features; best_model не помнил, какие столбцы были удалены во время обучения. Это создает скрытый путь ошибки во время инференса. RandomForestClassifier, обученный на 47 признаках, не выдает ошибки, когда ему дают 52 признака: он обращается к столбцам по целочисленной позиции и без предупреждения строит прогнозы по неправильным переменным. Исправление — сохранить обученный препроцессор и добавить его перед best_model после обучения. preprocess_features() теперь сохраняет self.preprocessor:

def preprocess_features(self):
    if self.meta_features.empty:
        combined = self.features.dropna()
    else:
        combined = self.features.join(self.meta_features, how='inner').dropna()

    # Store the fitted preprocessor. It is prepended to best_model in train_model
    # so that inference applies exactly the same column selection as training.
    self.preprocessor = Pipeline([
        ('dcf', DropConstantFeatures()),
        ('ddf', DropDuplicateFeatures()),
    ])
    self.preprocessed_features = self.preprocessor.fit_transform(combined)
    self.events = self.events.loc[self.preprocessed_features.index]

Рис. 3 — до v4.0 best_model содержал только классификатор. Новые данные с другим набором столбцов вызывали скрытые ошибочные прогнозы, потому что sklearn использует доступ по целочисленной позиции. После v4.0 обученный препроцессор является шагом 0 в best_model и принудительно применяет набор столбцов времени обучения при каждом последующем прогнозе.

После диспетчеризации HPO train_model() добавляет его в начало:

self.best_model = Pipeline([
    ('preprocessor', self.preprocessor),
    *self.best_model.steps,
])

Теперь сохраненная модель имеет структуру preprocessor → clf (или preprocessor → bag). Вызов best_model.predict(raw_features) во время инференса применяет тот же выбор столбцов, что и при обучении. Это требуется для LearnedStrategy.generate_signals()(Раздел 13) и для экспорта ONNX.

_get_feature_names() в результате упрощается — интроспекция пайплайна не нужна, поскольку preprocessed_features уже является итоговым набором столбцов после препроцессинга:

def _get_feature_names(self):
    if self.preprocessed_features is None:
        return []
    return self.preprocessed_features.columns.tolist()

Переработанный метод train_model

Три решения перед диспетчеризацией стоит отметить отдельно.

Во-первых, когда bagging_n_estimators равен нулю, а базовый классификатор — RandomForestClassifier, max_samples добавляется в param_grid как scipy.stats.uniform распределение, ограниченное средней уникальностью меток. Это делает долю бутстрэп-выборки настраиваемым гиперпараметром, а не фиксированным значением. DecisionTreeClassifier — это одиночное дерево без шага bootstrap, и у него нет параметра max_samples — включение его в условие в ранних версиях было ошибкой, которая приводила к скрытому no-op.

Во-вторых, bagging_max_samples преобразуется в среднюю уникальность меток, когда вызывающий код оставляет его как None. Это делается в train_model() потому что self.events['tW'] доступен в этой точке, а преобразование должно одинаково применяться к обоим путям обучения.

В-третьих, блок после диспетчеризации добавляет обученный препроцессор в начало и восстанавливает n_jobs=-1.

def train_model(self):
    self.model_params['pipe_clf'] = make_custom_pipeline(self.model_params['pipe_clf'])
    pipe = clone(self.model_params['pipe_clf'])

    bagging_n_estimators = self.model_params.get('bagging_n_estimators', 0)

    if bagging_n_estimators > 0:
        if self.model_params.get('bagging_max_samples') is None:
            av_uniqueness = self.events['tW'].mean().round(2)
            self.model_params['bagging_max_samples'] = av_uniqueness
            logger.info(f"bagging_max_samples set to average uniqueness ({av_uniqueness:.4f})")
    elif isinstance(pipe.steps[-1][-1], RandomForestClassifier):
        # Add max_samples as a searchable hyperparameter bounded by uniqueness.
        # DecisionTreeClassifier has no bootstrap step and no max_samples parameter.
        av_uniqueness = self.events['tW'].mean().round(2)
        self.model_params['param_grid']['max_samples'] = uniform(av_uniqueness, 1 - av_uniqueness)

    self.model_params['pipe_clf'] = pipe

    if self.model_params.get('use_optuna', False):
        self._train_model_optuna()
    else:
        self._train_model_sklearn()

    self.best_model = Pipeline([
        ('preprocessor', self.preprocessor),
        *self.best_model.steps,
    ])
    self.best_model = set_pipeline_params(self.best_model, n_jobs=-1)
    self.completed_steps['model_training'] = True

Обе цели диспетчеризации — _train_model_sklearn и _train_model_optuna — не принимают аргументов. Каждая читает пайплайн и конфигурацию напрямую из self.model_params. _train_model_sklearn использует inspect.signature для фильтрации model_params до ключей, принимаемых clf_hyper_fit_cached, предотвращая неожиданные именованные аргументы. Когда bagging_sequential=True, он удаляет ключи bagging, чтобы clf_hyper_fit вернул обычный настроенный пайплайн, а затем вызывает _apply_sequential_bagging после HPO с self.sample_weight:

def _train_model_sklearn(self):
    bagging_sequential = self.model_params.get('bagging_sequential', False)
    bagging_n = self.model_params.get('bagging_n_estimators', 0)
    sample_weight_train = self.sample_weight.loc[self.events.index]
    sample_weight_score = self.events['w'].loc[sample_weight_train.index]

    # Filter to keys accepted by clf_hyper_fit_cached via signature introspection.
    included = inspect.signature(clf_hyper_fit_cached).parameters.keys()
    params = {k: v for k, v in self.model_params.items() if k in included}

    if bagging_sequential and bagging_n > 0:
        params['bagging_n_estimators'] = 0
        tuned_pipeline, self.cv_results = clf_hyper_fit_cached(
            features=self.preprocessed_features, labels=self.events['bin'],
            t1=self.events['t1'], **params,
            sample_weight_train=sample_weight_train,
            sample_weight_score=sample_weight_score,
        )
        self.best_model = self._apply_sequential_bagging(
            self.preprocessed_features, self.events['bin'],
            tuned_pipeline, sample_weight=sample_weight_train,
        )
    else:
        self.best_model, self.cv_results = clf_hyper_fit_cached(
            features=self.preprocessed_features, labels=self.events['bin'],
            t1=self.events['t1'], **params,
            sample_weight_train=sample_weight_train,
            sample_weight_score=sample_weight_score,
        )

 _train_model_optuna подробно

Метод не принимает аргументов — он читает пайплайн и пространство поиска из self.model_params. Он преобразует ключи model_params, автоматически выводит study_name и db_path, вычисляет метрику оценки из пространства меток, запускает исследование с refit=True, а затем направляет выполнение в один из трех путей после исследования на основе bagging_sequential и bagging_n_estimators.

Фильтрация параметров использует inspect.signature для извлечения ключей, принимаемых optimize_trading_model. Это означает, что любой новый параметр, добавленный в сигнатуру функции в будущей версии, пройдет автоматически без изменения этого метода:

def _train_model_optuna(self):
    X, y = self.preprocessed_features, self.events['bin']
    base_clf = self.model_params['pipe_clf'].steps[-1][1]
    metric = 'f1' if set(y.unique()) == {0, 1} else 'neg_log_loss'

    # Filter model_params to keys accepted by optimize_trading_model.
    included = inspect.signature(optimize_trading_model).parameters.keys()
    opt_params = {'metric': metric}
    for k, v in self.model_params.items():
        if k == 'param_grid':
            opt_params['param_distributions'] = v
        elif k in included:
            opt_params[k] = v

    config_hash = self.file_paths['base_dir'].name
    study_config_hash = self._get_study_config_hash()

    opt_params['study_name'] = (
        f"{self.strategy.get_strategy_name()}"
        f"_{self.symbol}"
        f"_{self.data_config.get('bar_type', 'unk')}"
        f"_{self.data_config.get('bar_size', 'unk')}"
        f"_{config_hash}"
        f"_s{study_config_hash}"
    )

    db_path: Path = self.file_paths['db_path']
    db_path.parent.mkdir(parents=True, exist_ok=True)
    opt_params['db_path'] = f"sqlite:///{db_path.resolve()}"
    opt_params['reports_path'] = self.file_paths['reports'] / "trials"
    callbacks = [check_for_overfitting, print_best_trial]

    # Attempt auto-launch of optuna-dashboard in a background thread.
    try:
        from .dashboard import launch_optuna_dashboard
        launch_optuna_dashboard(storage=opt_params['db_path'], timeout=60)
    except Exception as e:
        logger.error(e)

    # ── Run the study (refit=True: handled internally) ───────────────────
    self.study, cv_results_df = optimize_trading_model(
        classifier=base_clf, X=X, y=y, events=self.events,
        data_index=self.bar_data.index,
        refit=True,
        callbacks=callbacks,
        **opt_params,
    )

    logger.info(
        f"Optuna complete. Best score: {self.study.best_value:.4f} | "
        f"Best params: {self.study.best_params}"
    )
    best_estimator = make_custom_pipeline(self.study.best_estimator_.base_estimator)

    bagging_sequential   = self.model_params.get('bagging_sequential', False)
    bagging_n_estimators = self.model_params.get('bagging_n_estimators', 0)
    bagging_max_samples  = self.model_params.get('bagging_max_samples', 1.0)
    bagging_max_features = self.model_params.get('bagging_max_features', 1.0)
    n_jobs               = self.model_params.get('n_jobs', -1)
    random_state         = self.model_params.get('random_state', None)

    if bagging_sequential and bagging_n_estimators > 0:
        self.best_model = self._apply_sequential_bagging(
            X, y, best_estimator,
            sample_weight=self.study.best_estimator_.sample_weight_,
        )
    elif bagging_n_estimators > 0:
        base_est = set_pipeline_params(best_estimator, n_jobs=1)
        bag = BaggingClassifier(
            estimator=MyPipeline(base_est.steps),
            n_estimators=int(bagging_n_estimators),
            max_samples=bagging_max_samples, max_features=bagging_max_features,
            n_jobs=n_jobs, random_state=random_state,
        )
        bag.fit(X, y, sample_weight=self.study.best_estimator_.sample_weight_)
        self.best_model = Pipeline([('bag', bag)])
    else:
        self.best_model = best_estimator

    pruner_type = self.model_params.get('pruner_type', 'hyperband')
    self.cv_results = {
        'best_params':        self.study.best_params,
        'best_score':         self.study.best_value,
        'cv_results':         cv_results_df,
        'scoring':            metric,
        'search_method':      'optuna',
        'pruner_type':        pruner_type,
        'n_trials_completed': len([t for t in self.study.trials
                                     if t.state.name == 'COMPLETE']),
        'n_trials_pruned':    len([t for t in self.study.trials
                                     if t.state.name == 'PRUNED']),
    }

Название исследования кодирует каждое измерение эксперимента: стратегию, символ, тип баров, размер баров и полный хэш конфигурации. Финальный сегмент _s{study_config_hash} отражает конфигурацию бэггинга, тип базовой модели и структуру пространства поиска через короткий MD5-хэш. Этот хэш вычисляется с помощью _get_study_config_hash(), который хэширует словарь параметров бэггинга, имя класса классификатора и фиксированные параметры, а также отсортированные ключи param_grid. Любое структурное изменение — добавление параметра в пространство поиска, переход от стандартного к последовательному бэггингу или смена базового классификатора — создает новое название исследования, не позволяя Optuna возобновить исследование, пространство параметров которого несовместимо с текущим запуском. Изменения границ распределений намеренно исключены из хэша, потому что Optuna корректно обрабатывает их в рамках одного исследования.

Обратите внимание, что n_splits не передается как явный именованный аргумент. Он проходит через opt_params через цикл inspect.signature, потому что optimize_trading_model принимает его как именованный параметр. Атрибут пайплайна — self.n_splits, установленный из model_params['n_splits'] в __init__.

Режим refit и бэггинг

refit=True

optimize_trading_model вызывается с refit=True. После завершения исследования, библиотека внутри себя вызывает FinancialModelSuggester.apply_from_params со словарем параметров победившего испытания, создает корректный _WeightedEstimator и обучает его на полном наборе данных. Результат сохраняется в study.best_estimator_ вместе с sample_weight_ — финальными вычисленными весами с примененной оптимальной схемой и decay. apply_from_params является детерминированной парой для suggest_and_apply: все, что стохастически сэмплировалось во время поиска, точно реконструируется для финального fit.

Стандартный бэггинг

Для стандартного бэггинга нужно извлечь базовый оцениватель из _WeightedEstimator. BaggingClassifier извлекает бутстрэп-выборки через целочисленное индексирование NumPy, которое удаляет индекс pandas. _WeightedEstimator.fit выравнивает веса через .loc[] и поэтому требует индекс; иначе выравнивание тихо ломается. Решение: извлечь уже вычисленные оптимальные веса из study.best_estimator_.sample_weight_ и явно передать их в BaggingClassifier.fit, используя стандартный MyPipeline([("clf", RF)]) как базовый оцениватель. Одного вызова bag.fit достаточно.

Последовательный бэггинг

Когда bagging_sequential=True, _apply_sequential_bagging вызывается вместо этого. Последовательный бутстрэппинг выбирает наблюдения пропорционально средней уникальности меток, а не равномерно. Поскольку метки triple-barrier перекрываются во времени, стандартный бутстрэп многократно пересэмплирует одни и те же конкурентные метки вместе, создавая коррелированные ансамбли. Последовательный бутстрэппинг уменьшает эту избыточность, давая менее коррелированный ансамбль при том же числе оцениватели — прямое применение принципа взвешивания по уникальности из Части 5.

После обучения метод преобразует SequentiallyBootstrappedBaggingClassifier в стандартный BaggingClassifier для совместимости с ONNX. skl2onnx не имеет конвертера для SequentiallyBootstrappedBaggingClassifier поскольку он не является частью scikit-learn. Преобразование копирует все обученное состояние — estimators_, estimators_features_, classes_, and n_features_in_ — из последовательного bagger в стандартную оболочку BaggingClassifier. Последовательный бутстрэппинг влияет только на то, как выбираются обучающие выборки; прогнозирование у двух классов идентично. Обученные оцениватели — те же объекты, а не копии, поэтому информация не теряется:

def _apply_sequential_bagging(self, X, y, tuned_pipeline, sample_weight=None):
    bagging_n       = self.model_params.get('bagging_n_estimators', 0)
    bagging_samples = self.model_params.get('bagging_max_samples', 1.0)
    bagging_feats   = self.model_params.get('bagging_max_features', 1.0)
    random_state    = self.model_params.get('random_state', 1)

    base_est = set_pipeline_params(tuned_pipeline, n_jobs=1)

    seq_bag = SequentiallyBootstrappedBaggingClassifier(
        estimator=MyPipeline(base_est.steps),
        n_estimators=int(bagging_n),
        max_samples=bagging_samples, max_features=bagging_feats,
        samples_info_sets=self.events['t1'],
        price_bars_index=self.bar_data.index,
        random_state=random_state,
    )

    if sample_weight is not None:
        seq_bag.fit(X, y, sample_weight=sample_weight)
    else:
        seq_bag.fit(X, y)

    # ── Convert to standard BaggingClassifier for ONNX compatibility ─────
    standard_bag = BaggingClassifier(
        estimator=MyPipeline(base_est.steps),
        n_estimators=len(seq_bag.estimators_),
        max_samples=1.0,
        max_features=seq_bag.max_features,
        bootstrap=seq_bag.bootstrap,
        bootstrap_features=seq_bag.bootstrap_features,
        random_state=random_state,
        n_jobs=seq_bag.n_jobs,
    )
    standard_bag.estimators_ = seq_bag.estimators_
    standard_bag.estimators_features_ = seq_bag.estimators_features_
    standard_bag.classes_ = getattr(seq_bag, 'classes_', np.array(sorted(y.unique())))
    standard_bag.n_classes_ = len(standard_bag.classes_)
    standard_bag.n_features_in_ = X.shape[1]

    return Pipeline([('seq_bag', standard_bag)])

Рис. 4 — дерево решений ансамбля после HPO, общее для обоих путей обучения. HPO всегда настраивает базовый классификатор без какой-либо ансамблевой обертки. Тип ансамбля определяется после завершения исследования. В обеих ветках бэггинга _WeightedEstimator разворачивается, а веса лучшего испытания передаются явно.

В пути sklearn, когда bagging_sequential=True, из вызова clf_hyper_fit удаляются параметры бэггинга, чтобы он вернул обычный настроенный пайплайн. _apply_sequential_bagging затем вызывается с self.sample_weight. HPO всегда настраивает базовый классификатор без ансамблевой обертки, независимо от запрошенного варианта бэггинга.

Отчет визуализации Optuna

Когда используется путь Optuna, _generate_analysis_reports() вызывает _generate_optuna_report() после стандартного markdown-отчета по гиперпараметрам и сводки обучения. Метод создает самодостаточный HTML-файл по адресу file_paths['reports'] / 'optuna_study_report.html' с четырьмя интерактивными графиками Plotly и сохраняет пятый статический график как сопутствующий PNG.

32

	График	На какой вопрос отвечает
1.	plot_optimization_history	Сошелся ли TPE-сэмплер? Ровное плато к испытанию 20–30 говорит, что бюджета было достаточно; кривая, продолжающая улучшаться на границе бюджета, означает, что нужны дополнительные испытания.
2.	plot_intermediate_values	Отсекатель делает решения видимыми. Каждая линия — одно испытание; линии, заканчивающиеся до оценки всех фолдов, были отсечены. То, что фолд 1 является основной точкой отсечения, нормально — агрессивная ветка HyperbandPruner работает.
3.	plot_param_importances	Какие гиперпараметры влияли на дисперсию оценки? fANOVA ранжирует каждый параметр. Параметры в нижней части можно зафиксировать на лучшем значении и убрать из последующих поисков, чтобы сократить эффективное пространство поиска.
4.	plot_parallel_coordinate	Показывает взаимодействия параметров. Одна линия на завершенное испытание, цвет — по оценке. Сходящиеся линии выявляют область высоких оценок; параллельные несходящиеся линии указывают на нечувствительный параметр.
5.	plot_model_vs_baseline (PNG)	Оценка лучшей итерации по фолдам относительно наивной базовой линии на основе энтропии. Заштрихованная область показывает, где модель демонстрирует экономическое преимущество. Для расчета базовой линии используются веса return-attribution, согласованные с критерием оценки, применявшимся во время исследования.

Все четыре HTML-графика используют темную тему пайплайна и plotly.io.to_html с include_plotlyjs='cdn' — единый переносимый файл без локальных зависимостей ресурсов. Каждый график обернут в try/except, чтобы неудачный импорт или недостаточное количество испытаний не прерывали пайплайн:

def _generate_analysis_reports(self):
    try:
        if self.cv_results and 'cv_results' in self.cv_results:
            cv_df = pd.DataFrame(self.cv_results['cv_results'])
            generate_complete_hyperparameter_report(
                cv_results=cv_df, strategy_config=self.config,
                output_dir=self.file_paths['reports']
            )
        self._generate_training_summary_html()
        if self.study is not None:
            self._generate_optuna_report()
    except Exception as e:
        logger.warning(f"Report generation failed: {e}")

Живой мониторинг с optuna-dashboard (панель мониторинга)

Поскольку db_path всегда заполняется из реестра путей файлов пайплайна, исследование всегда сохраняется в SQLite. optuna-dashboard поэтому можно направить на базу данных до запуска исследования, и он будет обновляться в реальном времени по мере завершения испытаний — изменения кода пайплайна не требуются. Панель мониторинга читает напрямую из файла SQLite и не требует подключения к запущенному Python-процессу.

Пайплайн пытается автоматически запустить optuna-dashboard в фоновом потоке непосредственно перед началом исследования. Это выполняет вспомогательная функция launch_optuna_dashboard, которая импортируется из dashboard.py. Если импорт или запуск не удается — например, потому что optuna-dashboard не установлен или порт уже занят — ошибка логируется, а исследование продолжается в штатном режиме. Панель мониторинга всегда можно запустить вручную в отдельном терминале:

pip install optuna-dashboard

# In a separate terminal while the pipeline runs
optuna-dashboard sqlite:////absolute/path/to/Models/BollingerBand/optuna_studies.db

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

База данных находится на уровне стратегии в дереве артефактов, поэтому все эксперименты по стратегии отображаются в одном представлении панели мониторинга.

Расчет весов выборки в пути Optuna

compute_sample_weights (Шаг 4) выполняется в обоих путях. Три вещи зависят от self.sample_weight независимо от HPO: мета-признаки (Шаг 5, только вторичная модель) используют его для скользящих взвешенных метрик производительности; сводка обучения показывает лучшую схему взвешивания; веса сохраняются на диск, чтобы повторный запуск мог полностью пропустить их вычисление.

Веса, вычисленные методом get_optimal_sample_weight, никогда не передаются в систему Optuna HPO. _WeightedEstimator внутренне вычисляет собственные веса из events и data_index на каждом испытании, потому что схема весов — это гиперпараметр поиска, который меняется между испытаниями. Эти два ряда весов независимы, и их нельзя путать.

Интеграция кэширования

Две системы персистентности работают на разных слоях и должны оставаться разделенными. Шаги 1–5 декорированы @cacheable() или @cv_cacheable. Повторный запуск после изменения кода в optimize_trading_model перезагружает препроцессированные признаки, events и веса из кэша joblib и повторно выполняет только этап обучения. Исследование Optuna в SQLite хранит завершенные испытания; повторный запуск с теми же автоматически полученными именами продолжается с последнего завершенного испытания. В совокупности это означает, что после любого прерывания вычисления не повторяются ни на одном уровне.

Не оборачивайте optimize_trading_model в @cacheable декоратор. Это заморозило бы исследование в одном снимке и нарушило бы гарантию восстановления после сбоя при каждом последующем запуске.

model_params = {
    'pipe_clf':             RandomForestClassifier(n_jobs=1, random_state=42),
    'param_grid':           FinancialModelSuggester.get_search_space('random_forest'),
    'n_splits':             5,
    'use_optuna':           True,
    'n_trials':             150,
    'timeout':              7200,
    'pruner_type':          'hyperband',
    'bagging_n_estimators': 100,
    'bagging_sequential':   True,   # SequentiallyBootstrappedBaggingClassifier post-HPO
    'bagging_max_samples':  None,   # auto-resolved to events['tW'].mean() in train_model
    'bagging_max_features': 1.0,
    'random_state':         42,
    # study_name and db_path are auto-derived.
}

Long-Short пайплайн Bid/Ask

Зачем нужны отдельные модели для каждой стороны

Стандартная разметка triple-barrier рассматривает long- и short-входы симметрично. В исполнении это не так: long-вход исполняется по ask, short — по bid. Спред — это транзакционная стоимость, оплачиваемая при входе и выходе. Сигнал, выглядящий прибыльным на mid-price, может терять деньги на одной стороне при использовании цен исполнения. Обучение на реалистичных ценах исполнения позволяет учесть эту стоимость уже на этапе разметки, а значит и для модели.

BidAskLongShortPipeline

BidAskLongShortPipeline оборачивает два экземпляра ModelDevelopmentPipeline, настроенные с price='ask' и price='bid' соответственно. Он один раз загружает бары с price='bid_ask', разделяет их на ask-бары для long-пайплайна и bid-бары для short-пайплайна, генерирует events отдельно для каждой стороны и фильтрует по соответствующему направлению, затем независимо запускает стандартные Шаги 4–7 для каждого подпайплайна.

from afml.production.dual_model_development import BidAskLongShortPipeline

pipeline = BidAskLongShortPipeline(
    strategy=strategy, data_config=data_config,
    feature_config=feature_config, target_config=target_config,
    label_config=label_config, model_params=model_params,
    base_dir='Models/BidAsk',
)
results = pipeline.run()

spread = results['spread_stats']
print(f"Mean spread: {spread['spread_mean']:.5f} ({spread['spread_bps']:.2f} bps)")
print(f"Long  CV: {results['combined_metrics']['long_cv_score']:.4f}")
print(f"Short CV: {results['combined_metrics']['short_cv_score']:.4f}")

Если CV оценка short-модели существенно ниже, чем у long-модели, bid-ask спред съедает преимущество short-стороны, и это направление не следует торговать при текущем уровне спредов.

LearnedStrategy: связующее звено между двумя этапами

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

Когда экземпляр LearnedStrategy передается как аргумент strategy во вторичный ModelDevelopmentPipeline, этап разметки вызывает generate_signals() для получения прогнозов стороны. Они передаются в triple_barrier_labels как side_prediction, из-за чего get_events() сохраняет side в DataFrame событий, а get_bins() переходит в режим мета-разметки. Поэтому флаг is_primary вторичного пайплайна равен False, вычисляются скользящие мета-признаки, а каталог артефактов содержит model_role='secondary'.

from afml.strategies.learned_strategy import LearnedStrategy

# ── Stage 1: primary model ────────────────────────────────────────────────
primary_pipeline = ModelDevelopmentPipeline(
    strategy=BollingerBandStrategy(window=20, std=1.5),
    data_config=data_config, feature_config=feature_config,
    target_config=target_config, label_config=label_config,
    model_params=primary_model_params,
)
primary_pipeline.run()

# ── Wrap the trained model as a strategy ─────────────────────────────────
learned = LearnedStrategy.from_pipeline(primary_pipeline)

# ── Stage 2: secondary model ──────────────────────────────────────────────
secondary_pipeline = ModelDevelopmentPipeline(
    strategy=learned,              # generate_signals() provides side predictions
    data_config=data_config,
    feature_config=feature_config,
    target_config=target_config,
    label_config=secondary_label_config,
    model_params=secondary_model_params,
)
secondary_pipeline.run()

Рис. 5 — двухэтапный рабочий процесс. LearnedStrategy.from_pipeline() напрямую преобразует обученный первичный пайплайн в BaseStrategy. generate_labels() вторичного пайплайна вызывает generate_signals(), который вызывает best_model.predict() — препроцессор внутри best_model обеспечивает выравнивание столбцов без внешнего шага препроцессинга.

generate_signals() применяет feature_config['func'] к данным баров и вызывает fitted_pipeline.predict(). Поскольку best_model первичного пайплайна теперь включает обученный препроцессор как шаг 0, выравнивание столбцов выполняется внутри — при инференсе применяется тот же набор столбцов, который был выбран во время обучения. Без препроцессора внутри best_model, generate_signals() строил бы прогнозы по неправильным столбцам всякий раз, когда постоянный или дублирующийся признак появлялся в новых данных, но отсутствовал в обучающем наборе.

Есть одно ограничение: LearnedStrategy оборачивает только первичные модели. best_model вторичного пайплайна был обучен на признаках, включающих скользящие мета-признаки — столбцы, полученные из производительности предыдущей модели, — которые невозможно воспроизвести во время инференса без этой предыдущей модели в области видимости. from_pipeline() вызывает ValueError, если pipeline.is_primary равен False.

Метка 0 (достигнут вертикальный барьер во время обучения первичной модели) отображается в 1 в generate_signals(). Роль первичной модели — только предоставить направление; вторичная модель решает, действовать ли по каждому сигналу. Значение side, равное 0, в этом контексте не имеет смысла.

Чтобы поддержать реконструкцию без загрузки полной модели, объект стратегии сохраняется как самостоятельный артефакт cloudpickle рядом с model joblib. _save_all_artifacts вторичного пайплайна вызывает self.file_manager.save_object(self.strategy, "strategy") чтобы LearnedStrategy можно было перезагрузить независимо от пайплайна, который его создал.

Конфигурация model_params

Полный набор ключей, распознаваемых путем Optuna, приведен ниже. study_name и db_path выводятся автоматически и больше не настраиваются пользователем.

32

	Ключ	Тип	По умолчанию	Описание
1.	use_optuna	bool	False	Переключение на Optuna бэкенд HPO.
2.	pipe_clf	estimator или Pipeline	—	Базовый классификатор или пайплайн. Когда use_optuna=True, estimator последнего шага извлекается и передается в FinancialModelSuggester.
3.	param_grid	dict	{}	Пространство поиска. Принимает списки, scipy.stats распределения или объекты range. Внутренне переименовывается в param_distributions. Используйте FinancialModelSuggester.get_search_space() для подобранных значений по умолчанию. Когда базовый классификатор — RandomForestClassifier и бэггинг не активен, max_samples автоматически добавляется как uniform распределение, ограниченное средней уникальностью меток.
4.	n_splits	int	5	Число PurgedKFold разбиений. Также задает HyperbandPruner для max_resource. Передается в optimize_trading_model через интроспекцию сигнатуры.
5.	n_trials	int	100	Бюджет испытаний. Выполнение останавливается при достижении лимита или когда истекает timeout.
6.	timeout	int	3600	Ограничение по реальному времени в секундах. Второй критерий остановки.
7.	pruner_type	str	'hyperband'	hyperband (рекомендуется) или median (TradingModelPruner с энтропийной базовой линией). Механику brackets см. в Части 8.
8.	bagging_n_estimators	int	0	Число бэггинг. 0 отключает бэггинг. Ансамбль строится после HPO из настроенного базового классификатора.
9.	bagging_sequential	bool	False	Когда True и bagging_n_estimators > 0, используется SequentiallyBootstrappedBaggingClassifier после HPO вместо стандартного BaggingClassifier. Обученный ансамбль преобразуется в стандартный BaggingClassifier для совместимости с ONNX. Рекомендуется, когда метки triple-barrier имеют высокую concurrency (низкую среднюю уникальность). HPO всегда настраивает базовый классификатор без ансамблевой обертки.
10.	bagging_max_samples	float или None	None	Доля выборок на бэг. Если оставить как None и бэггинг активен, train_model заменяет его на events['tW'].mean() во время выполнения. Задайте явный float, чтобы переопределить.
11.	bagging_max_features	float	1.0	Доля признаков, выбираемых на один bag.
12.	random_state	int	None	Задает начальное значение генератора для TPESampler, для random_state базового классификатора и для ансамбля бэггинг ради воспроизводимости.

Практические соображения

Доступ к графикам исследования после запуска. HTML-отчет из четырех графиков автоматически сохраняется в file_paths['reports'] / 'optuna_study_report.html'. self.study также содержит завершенный объект исследования для доступа внутри процесса. Чтобы перезагрузить предыдущее исследование из базы данных:

import optuna

db_uri = f"sqlite:///{pipeline.file_paths['db_path'].resolve()}"
study  = optuna.load_study(study_name=pipeline.study.study_name, storage=db_uri)
# study_name format: StrategyName_Symbol_BarType_BarSize_ConfigHash_sStudyHash
# e.g. BollingerBand_EURUSD_tick_M1_a3f7c912_s4b2e1f

Разворачивание важности признаков.best_model теперь имеет вид preprocessor → clf/bag. Последний шаг — классификатор. Правильная последовательность разворачивания обрабатывает все формы модели:

from afml.production.weighted_estimator import _WeightedEstimator

clf = self.best_model.steps[-1][1]

if isinstance(clf, SequentiallyBootstrappedBaggingClassifier):
    importances = np.mean([
        est.steps[-1][1].feature_importances_ for est in clf.estimators_
    ], axis=0)
elif isinstance(clf, BaggingClassifier):
    importances = np.mean([
        est.steps[-1][1].feature_importances_ for est in clf.estimators_
    ], axis=0)
elif isinstance(clf, _WeightedEstimator):
    importances = clf.base_estimator.feature_importances_
else:
    importances = clf.feature_importances_

Обратите внимание, что isinstance(clf, _WeightedEstimator) используется вместо hasattr(clf, 'base_estimator'). BaggingClassifier также предоставляет атрибут base_estimator (свой необученный шаблон), поэтому проверка hasattr дает ложноположительный результат. Импорт из weighted_estimator обязателен — это тот же импорт, который внутренне использует метод пайплайна analyze_features().

Экспорт ONNX и MyPipeline.skl2onnx распознает только sklearn.pipeline.Pipeline. MyPipeline — подкласс, который добавляет передачу sample_weight; это относится ко времени обучения и не имеет представления в ONNX. Перед конвертацией в ONNX _convert_mypipeline_for_onnx() рекурсивно заменяет каждый экземпляр MyPipeline внутри обученного пайплайна на стандартный Pipeline, сохраняя все обученное состояние. Это обрабатывает три схемы вложенности: шаг пайплайна, который напрямую является MyPipeline, подкласс BaggingClassifier, шаблон estimator которого — MyPipeline, и обученные estimators_ внутри BaggingClassifier, которые являются экземплярами MyPipeline. Шаг препроцессора также удаляется перед конвертацией, потому что DropConstantFeatures и DropDuplicateFeatures не имеют сопоставления с операторами ONNX. Применяйте self.preprocessor.transform() как самостоятельный шаг перед передачей данных в развернутую ONNX-модель.

Когда use_optuna=False по-прежнему правильно. TPE-сэмплер Optuna требует минимум 20–30 завершенных испытаний, прежде чем его модель поверхности целевой функции начнет превосходить random search. Для пространств поиска с менее чем тремя параметрами и быстрым обучением use_optuna=False с rnd_search_iter=0 дает детерминированные результаты без накладных расходов SQLite.

Параллельные процессы.RDBStorage настроен с timeout=30 и pool_pre_ping=True, поэтому несколько процессов могут одновременно указывать на одну базу данных. Установите n_jobs=1 внутри классификатора и полагайтесь на параллелизм на уровне процессов, чтобы избежать oversubscription потоков.

Заключение

Переработанный ModelDevelopmentPipeline объединяет оба бэкенда HPO за единым интерфейсом. Обученный препроцессор теперь является постоянной частью best_model, делая модель самодостаточной для инференса и обеспечивая безопасную межэтапную генерацию сигналов через LearnedStrategy.

Определение первичной/вторичной модели автоматическое. Если side присутствует в DataFrame событий, запуск считается вторичным; иначе — первичным. Этот выбор определяет вычисление мета-признаков, ожидаемое пространство меток и записанный model_role в каталоге артефактов. LearnedStrategy напрямую преобразует выход первичного пайплайна в BaseStrategy, связывая два этапа обучения без ручной конфигурации. Ключевое условие — best_model теперь включает препроцессор: generate_signals() вторичного пайплайна вызывает fitted_pipeline.predict() и выбор столбцов обрабатывается внутри, независимо от того, что функция признаков создает на новых данных.

Последовательный бутстрэппинг доступен через bagging_sequential=True. HPO всегда настраивает базовый классификатор без ансамблевой обертки; ансамбль применяется после HPO в одном общем методе, используемом обоими путями обучения. Обученный SequentiallyBootstrappedBaggingClassifier преобразуется в стандартный BaggingClassifier для совместимости с ONNX — последовательный бутстрэппинг влияет только на способ выбора обучающих выборок, поэтому поведение прогнозирования идентично.

После каждого запуска Optuna четыре интерактивных графика Plotly сохраняются в самодостаточный HTML-отчет рядом с существующими markdown отчётами и сводными отчетами. Панель мониторинга URI логируется при старте исследования для живого мониторинга в отдельном терминале, дополняя статический отчет после запуска. Пайплайн также пытается автоматически запустить optuna-dashboard через фоновый поток.

Установите export_onnx=True для использования пайплайна экспорта ONNX при развертывании обученных моделей в MetaTrader 5. Перед конвертацией _convert_mypipeline_for_onnx() заменяет все экземпляры MyPipeline стандартными объектами sklearn Pipeline, а шаг препроцессора удаляется. Применяйте препроцессор как отдельное преобразование при подаче данных в развернутую ONNX-модель.

Прикрепленные файлы

В таблице ниже описан каждый файл, прикрепленный к этой статье.

32

	Файл	Модуль	Роль в этой статье	Ключевые зависимости
1.	model_development.py	afml.production	Центральный пайплайн. Добавляет диспетчеризацию use_optuna, определение primary/secondary через is_primary, управление мета-признаками, сохранение обученного препроцессора и добавление его перед best_model, bagging_sequential, маршрутизацию _apply_sequential_bagging с преобразованием после fit в стандартный BaggingClassifier для ONNX, _convert_mypipeline_for_onnx для экспорта ONNX, хэширование конфигурации исследования через _get_study_config_hash, и _generate_optuna_report() с пятью результатами визуализации. Версия пайплайна 4.0.	optuna_hyper_fit.py, weighted_estimator.py, file_manager.py, unified_cache_system.py, plotly
2.	learned_strategy.py	afml.strategies	Новый модуль. LearnedStrategy(BaseStrategy) оборачивает обученный первичный пайплайн, чтобы его прогнозы могли служить side_prediction на этапе разметки вторичного пайплайна. from_pipeline() проверяет, что источник является первичной моделью. generate_signals() полагается на то, что препроцессор находится внутри best_model для инференса с корректным согласованием столбцов. Отображает метку 0 в 1 для совместимости с triple-barrier.	model_development.py, trading_strategies.py
3.	weighted_estimator.py	afml.production	Самостоятельный модуль. _WeightedEstimator вынесен из model_development.py для устранения циклического импорта с optuna_hyper_fit.py. Добавляет проверку во время выполнения, что базовый оцениватель поддерживает sample_weight.	optimized_attribution.py, scikit-learn, numpy, pandas
4.	optuna_hyper_fit.py	afml.cross_validation	Обновлен по сравнению с Частью 8. Импортирует _WeightedEstimator из weighted_estimator.py. optimize_trading_model теперь принимает callbacks и reports_path. create_study получает объект RDBStorage, чтобы применялись настройки timeout и pool. Логирует команду optuna-dashboard при старте исследования.	weighted_estimator.py, cross_validation.py, optuna ≥ 3.0, scikit-learn ≥ 1.3
5.	file_manager.py	afml.production	Самостоятельный модуль. ConfigPathGenerator и ModelFileManager вынесены из utils.py. get_standard_file_paths добавляет запись db_path на уровне стратегии (Models/StrategyName/optuna_studies.db).	model_export.py, pandas, cloudpickle
6.	dual_model_development.py	afml.production	Сопутствующий пайплайн, обучающий отдельные модели на ask-ценах для long-входов и bid-ценах для short-входов. Вычисляет статистику спреда на уровне баров, чтобы количественно оценить асимметрию цен исполнения между двумя моделями.	model_development.py, pandas, loguru
7.	cross_validation.py	afml.cross_validation	Без изменений из Части 5. Предоставляет PurgedKFold, используемый обоими путями обучения и целевой функцией.	scikit-learn, pandas, numpy
8.	unified_cache_system.py	afml.cache	Без изменений из Части 6. Кэширует Шаги 1–5, чтобы при повторном запуске после прерванного исследования Optuna предварительная обработка загружалась из кэша.	joblib, loguru, scikit-learn, pandas, scipy