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

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

Розпізнавання паттернів японських свічок — це CV-завдання, але з важливою застереженням: паттерн сам по собі не є торговим сигналом. Пробій опору на високому обсязі з підтвердженням "молотка" — значимо. "Молоток" посередині флету без обсягу — це шум. Тому модель повинна розпізнавати не ізольований паттерн, а паттерн в контексті.

Підходи до завдання

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

Підхід 2: ML на числових ознаках — витягуємо геометричні ознаки свічок та навчаємо XGBoost/LightGBM. Швидше, більш інтерпретабельно, незалежно від візуалізації.

Підхід 3: Гібридний — числові ознаки + рендеринг графіку → мультимодальна модель. Найкраща точність, висока складність.

Підхід 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 є обов'язковим для фінансових даних.
    Не можна використовувати випадковий поділ (витік даних з майбутнього).
    """
    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,  # важливо для фінансів: уникати перенавчання
        '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   # зсув: -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 + режим ринку.

Часові рамки