Настройка автоматического переобучения торговой модели на новых данных

Настройка автоматического переобучения торговой модели на новых данных

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

Когда переобучать

По расписанию:

• Еженедельное переобучение: стандарт для большинства mean-reversion и momentum стратегий
• Ежедневное: для внутридневных стратегий с высоким alpha decay
• Ежемесячное: для долгосрочных стратегий с фундаментальными факторами

По триггеру:

• IC (Information Coefficient) упал ниже порогового значения
• PSI входных признаков > 0.2
• Обнаружен структурный break в данных
• Sharpe ratio за последние N дней < 0.5

Pipeline переобучения

from prefect import flow, task
import mlflow

@task(retries=2)
def collect_training_data(lookback_days: int) -> pd.DataFrame:
    """Сбор данных строго без look-ahead"""
    end_date = pd.Timestamp.now().normalize()  # Только закрытые данные
    start_date = end_date - pd.Timedelta(days=lookback_days)

    market_data = data_store.get_ohlcv(start_date, end_date)
    features = feature_pipeline.compute(market_data)

    # Проверка на данные из будущего
    assert features.index.max() < pd.Timestamp.now(), "Look-ahead bias detected!"
    return features

@task
def validate_data_quality(features: pd.DataFrame) -> bool:
    """Качество данных перед обучением"""
    # Пропуски
    if features.isnull().mean().max() > 0.05:
        raise ValueError("Too many missing values")

    # Достаточность данных
    if len(features) < 500:
        raise ValueError("Insufficient training data")

    # Выбросы
    z_scores = np.abs((features - features.mean()) / features.std())
    if (z_scores > 5).any().any():
        logging.warning("Extreme outliers detected, clipping")

    return True

@task
def train_model(features: pd.DataFrame, params: dict) -> str:
    with mlflow.start_run() as run:
        X_train, X_val, y_train, y_val = time_series_split(features)

        model = LGBMClassifier(**params)
        model.fit(X_train, y_train,
                  eval_set=[(X_val, y_val)],
                  callbacks=[mlflow.lightgbm.autolog()])

        # Метрики
        val_ic = compute_ic(model.predict(X_val), y_val)
        mlflow.log_metric('information_coefficient', val_ic)

        # Сохранение
        model_path = f"models/trading_model_{run.info.run_id}.pkl"
        joblib.dump(model, model_path)
        mlflow.log_artifact(model_path)

        return run.info.run_id

@task
def validate_new_model(run_id: str, production_model_id: str) -> bool:
    new_model = load_model_from_mlflow(run_id)
    prod_model = load_model_from_mlflow(production_model_id)

    # Walk-forward evaluation на hold-out периоде
    wf_results_new = walk_forward_evaluate(new_model, hold_out_data)
    wf_results_prod = walk_forward_evaluate(prod_model, hold_out_data)

    checks = {
        'sharpe_improvement': wf_results_new['sharpe'] > wf_results_prod['sharpe'] * 0.95,
        'no_drawdown_increase': wf_results_new['max_dd'] < wf_results_prod['max_dd'] * 1.2,
        'min_ic': wf_results_new['ic'] > 0.03,  # IC > 3%
        'min_trades': wf_results_new['trade_count'] > 50,  # Достаточно сделок
    }

    if not all(checks.values()):
        failed = [k for k, v in checks.items() if not v]
        mlflow.set_tag('validation_status', f'FAILED: {failed}')
        return False

    return True

@flow(name="trading-model-retraining")
def retraining_pipeline(trigger_reason: str):
    features = collect_training_data(lookback_days=252)
    validate_data_quality(features)
    run_id = train_model(features, params=TRAINING_PARAMS)

    production_id = get_current_production_model_id()
    if validate_new_model(run_id, production_id):
        # Деплой только в нерабочие часы
        schedule_deployment(run_id, deploy_window="02:00-09:00 UTC")
    else:
        alert_team(f"Retraining failed validation. Trigger: {trigger_reason}")

Безопасный деплой в нерабочие часы

Переключение торговой модели должно происходить когда рынки закрыты (или в период минимальной ликвидности для крипто). Это снижает риск переключения в момент активной торговли. Архивирование предыдущей версии обязательно — откат должен занимать секунды.

Типичный результат внедрения: переход от ручного переобучения каждые 2-3 месяца к автоматическому еженедельному циклу с воспроизводимыми результатами и audit trail каждого деплоя.