Разработка AI-модели анализа паттернов свечных графиков

AI-модель анализа свечных паттернов на графиках

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

Подходы к задаче

Подход 1: CV на скриншотах графика — обучаем детектор на PNG/JPEG изображениях. Просто, но теряет числовые данные OHLCV. Точность ограничена разрешением и стилем графика.

Подход 2: ML на числовых признаках — извлекаем геометрические признаки свечей и обучаем XGBoost/LightGBM. Быстрее, интерпретируемее, не зависит от визуализации.

Подход 3: Гибридный — числовые признаки + рендеринг графика → multimodal модель. Лучшая точность, высокая сложность.

Подход 2: числовые признаки (рекомендуемый)

import numpy as np
import pandas as pd
from typing import Optional

class CandlestickFeatureExtractor:
    """
    Извлекаем геометрические и относительные признаки свечей.
    Все признаки нормализованы к ATR (Average True Range) —
    это делает их масштабо-инвариантными.
    """

    def compute_candle_features(
        self,
        df: pd.DataFrame,   # OHLCV DataFrame
        lookback: int = 5   # количество предыдущих свечей
    ) -> pd.DataFrame:
        """
        Признаки одной свечи:
        - body_ratio: (close-open) / ATR — размер тела
        - upper_shadow_ratio: верхняя тень / ATR
        - lower_shadow_ratio: нижняя тень / ATR
        - body_position: позиция тела в диапазоне high-low
        - gap: разрыв от предыдущего close / ATR
        - volume_ratio: объём / MA(volume, 20)
        """
        atr = self._calculate_atr(df, period=14)

        features = pd.DataFrame(index=df.index)

        for i in range(lookback):
            shift = i + 1
            c = df.shift(shift) if i > 0 else df

            body = c['close'] - c['open']
            total_range = c['high'] - c['low'] + 1e-8

            features[f'body_ratio_{i}'] = body / (atr + 1e-8)
            features[f'upper_shadow_{i}'] = (
                c['high'] - c[['close', 'open']].max(axis=1)
            ) / (atr + 1e-8)
            features[f'lower_shadow_{i}'] = (
                c[['close', 'open']].min(axis=1) - c['low']
            ) / (atr + 1e-8)
            features[f'body_pos_{i}'] = (
                (c[['close', 'open']].min(axis=1) - c['low']) / total_range
            )
            if i == 0:
                features[f'gap_{i}'] = (
                    (c['open'] - df['close'].shift(1)) / (atr + 1e-8)
                )
            features[f'vol_ratio_{i}'] = c['volume'] / (
                c['volume'].rolling(20).mean() + 1e-8
            )

        # Контекстные признаки
        features['trend_5'] = (
            df['close'] - df['close'].shift(5)
        ) / (atr + 1e-8)
        features['trend_20'] = (
            df['close'] - df['close'].shift(20)
        ) / (atr + 1e-8)
        features['volatility_norm'] = atr / df['close']

        return features.fillna(0)

    def _calculate_atr(self, df: pd.DataFrame, period: int = 14) -> pd.Series:
        high_low   = df['high'] - df['low']
        high_close = (df['high'] - df['close'].shift()).abs()
        low_close  = (df['low']  - df['close'].shift()).abs()
        true_range = pd.concat(
            [high_low, high_close, low_close], axis=1
        ).max(axis=1)
        return true_range.ewm(span=period, adjust=False).mean()

Разметка паттернов и обучение

import talib   # TA-Lib для классических паттернов
import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import f1_score

def label_patterns(df: pd.DataFrame) -> pd.DataFrame:
    """
    Авторазметка паттернов через TA-Lib.
    Значения: 0 = нет паттерна, 100 = бычий, -100 = медвежий.
    """
    patterns = {
        'hammer':        talib.CDLHAMMER,
        'doji':          talib.CDLDOJI,
        'engulfing':     talib.CDLENGULFING,
        'morning_star':  talib.CDLMORNINGSTAR,
        'evening_star':  talib.CDLEVENINGSTAR,
        'shooting_star': talib.CDLSHOOTINGSTAR,
        'harami':        talib.CDLHARAMI,
        'three_white':   talib.CDL3WHITESOLDIERS,
    }

    for name, func in patterns.items():
        df[f'pattern_{name}'] = func(
            df['open'].values, df['high'].values,
            df['low'].values,  df['close'].values
        )

    # Целевая переменная: значимое движение вперёд на 3 свечи
    df['target'] = np.where(
        df['close'].shift(-3) > df['close'] * 1.005, 1,   # +0.5% = бычий
        np.where(
            df['close'].shift(-3) < df['close'] * 0.995, -1,  # -0.5% = медвежий
            0  # флет
        )
    )
    return df

def train_pattern_classifier(
    features: pd.DataFrame,
    labels: pd.Series
) -> lgb.Booster:
    """
    TimeSeriesSplit — обязателен для финансовых данных.
    Нельзя использовать random split (future leakage).
    """
    tscv = TimeSeriesSplit(n_splits=5)
    models = []

    params = {
        'objective': 'multiclass',
        'num_class': 3,           # -1, 0, 1
        'learning_rate': 0.05,
        'n_estimators': 500,
        'max_depth': 6,
        'min_child_samples': 50,  # важно для финансов: избегаем overfit
        'subsample': 0.8,
        'colsample_bytree': 0.8,
        'reg_lambda': 1.0,
        'metric': 'multi_logloss',
        'verbose': -1
    }

    for fold, (train_idx, val_idx) in enumerate(tscv.split(features)):
        X_train = features.iloc[train_idx]
        y_train = labels.iloc[train_idx] + 1   # shift: -1,0,1 → 0,1,2
        X_val   = features.iloc[val_idx]
        y_val   = labels.iloc[val_idx] + 1

        train_data = lgb.Dataset(X_train, label=y_train)
        val_data   = lgb.Dataset(X_val,   label=y_val)

        model = lgb.train(
            params,
            train_data,
            valid_sets=[val_data],
            callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)]
        )

        preds = model.predict(X_val).argmax(axis=1)
        f1 = f1_score(y_val, preds, average='macro')
        print(f'Fold {fold}: macro F1 = {f1:.4f}')
        models.append(model)

    return models

Важное предупреждение

Паттерн сам по себе предсказывает движение с точностью лишь немного выше 50%. В тестах на 10 лет данных SPY: точность модели ~58% при macro F1 ~0.41. Это не торговая система — это один из сигналов. Реальный прирост даёт ансамбль: паттерн + объёмный анализ + RSI/MACD контекст + режим рынка.

Сроки