機械学習による仮想通貨価格予測第1部：時系列データの前処理とテクニカル指標

はじめに

仮想通貨市場では、AIや機械学習を活用した価格予測への関心が急速に高まっています。本シリーズでは、実際に収益を上げているトレーダーが使用する予測モデルの構築方法を、3部構成で体系的に解説します。

第1部では、予測精度を左右する最も重要な要素である「データの前処理」について、基本から実践的なテクニックまでを詳しく紹介します。

なぜデータ前処理が重要なのか

機械学習モデルの予測精度は、入力データの質に大きく依存します。実際のトレーディングで使える予測モデルを構築するには：

適切な時間足の選択：スキャルピング（1分〜5分）、デイトレード（15分〜1時間）、スイングトレード（4時間〜日足）で異なる
特徴量エンジニアリング：単純な価格データから予測に有効な指標を生成
データの正規化：異なるスケールのデータを統一的に扱う

これらの要素を正しく実装することで、予測精度が50%から80%以上に向上することも珍しくありません。

基本的なデータ収集の流れ

まずは、CCXTライブラリを使用した基本的なデータ収集から始めましょう：

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class CryptoDataCollector:
    def __init__(self, exchange_name='binance'):
        self.exchange = getattr(ccxt, exchange_name)()
        
    def fetch_ohlcv_data(self, symbol='BTC/USDT', timeframe='1h', limit=1000):
        """
        指定されたシンボルのOHLCVデータを取得
        """
        try:
            # データ取得
            ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
            
            # DataFrameに変換
            df = pd.DataFrame(ohlcv, 
                columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            
            # タイムスタンプを日時に変換
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            df.set_index('timestamp', inplace=True)
            
            return df
            
        except Exception as e:
            print(f"データ取得エラー: {e}")
            return None
    
    def fetch_multiple_timeframes(self, symbol='BTC/USDT', timeframes=['1h', '4h', '1d']):
        """
        複数の時間足データを取得して結合
        """
        data_dict = {}
        
        for tf in timeframes:
            df = self.fetch_ohlcv_data(symbol, tf)
            if df is not None:
                # カラム名に時間足を追加
                df.columns = [f"{col}_{tf}" for col in df.columns]
                data_dict[tf] = df
        
        # データの結合（時間軸で整合性を保つ）
        if data_dict:
            # 最も細かい時間足を基準に結合
            base_df = data_dict[timeframes[0]]
            for tf in timeframes[1:]:
                if tf in data_dict:
                    base_df = base_df.join(data_dict[tf], how='left')
                    # 欠損値を前方補完
                    base_df.fillna(method='ffill', inplace=True)
            
            return base_df
        
        return None

このような基本的なコードでデータは取得できますが、実際の予測モデルで使用するには、さらに高度な処理が必要です。

テクニカル指標の実装と重要性

機械学習モデルが価格の動きを予測するには、単純な価格データだけでは不十分です。プロのトレーダーが使用する以下のような指標を特徴量として追加する必要があります：

トレンド系指標の実装

import talib

def add_trend_indicators(df):
    """
    トレンド系指標を追加
    """
    # 移動平均線（短期・中期・長期）
    df['SMA_10'] = talib.SMA(df['close'].values, timeperiod=10)
    df['SMA_20'] = talib.SMA(df['close'].values, timeperiod=20)
    df['SMA_50'] = talib.SMA(df['close'].values, timeperiod=50)
    df['SMA_200'] = talib.SMA(df['close'].values, timeperiod=200)
    
    # 指数移動平均線
    df['EMA_12'] = talib.EMA(df['close'].values, timeperiod=12)
    df['EMA_26'] = talib.EMA(df['close'].values, timeperiod=26)
    
    # MACD（移動平均収束拡散法）
    df['MACD'], df['MACD_signal'], df['MACD_hist'] = talib.MACD(
        df['close'].values, 
        fastperiod=12, 
        slowperiod=26, 
        signalperiod=9
    )
    
    # ADX（平均方向性指数）
    df['ADX'] = talib.ADX(df['high'].values, df['low'].values, df['close'].values, timeperiod=14)
    
    # パラボリックSAR
    df['SAR'] = talib.SAR(df['high'].values, df['low'].values, acceleration=0.02, maximum=0.2)
    
    return df

オシレーター系指標の実装

def add_oscillator_indicators(df):
    """
    オシレーター系指標を追加
    """
    # RSI（相対力指数）
    df['RSI_14'] = talib.RSI(df['close'].values, timeperiod=14)
    df['RSI_7'] = talib.RSI(df['close'].values, timeperiod=7)
    
    # ストキャスティクス
    df['STOCH_K'], df['STOCH_D'] = talib.STOCH(
        df['high'].values, 
        df['low'].values, 
        df['close'].values,
        fastk_period=14,
        slowk_period=3,
        slowd_period=3
    )
    
    # CCI（商品チャンネル指数）
    df['CCI'] = talib.CCI(df['high'].values, df['low'].values, df['close'].values, timeperiod=20)
    
    # Williams %R
    df['WILLR'] = talib.WILLR(df['high'].values, df['low'].values, df['close'].values, timeperiod=14)
    
    # MFI（マネーフローインデックス）
    df['MFI'] = talib.MFI(df['high'].values, df['low'].values, df['close'].values, df['volume'].values, timeperiod=14)
    
    return df

ボラティリティ指標と出来高指標

def add_volatility_volume_indicators(df):
    """
    ボラティリティと出来高指標を追加
    """
    # ボリンジャーバンド
    df['BB_upper'], df['BB_middle'], df['BB_lower'] = talib.BBANDS(
        df['close'].values,
        timeperiod=20,
        nbdevup=2,
        nbdevdn=2,
        matype=0
    )
    
    # ATR（平均真の範囲）
    df['ATR'] = talib.ATR(df['high'].values, df['low'].values, df['close'].values, timeperiod=14)
    
    # ボラティリティ比率
    df['volatility_ratio'] = df['ATR'] / df['close']
    
    # OBV（オンバランスボリューム）
    df['OBV'] = talib.OBV(df['close'].values, df['volume'].values)
    
    # VWAP（出来高加重平均価格）- 手動計算
    df['VWAP'] = (df['volume'] * (df['high'] + df['low'] + df['close']) / 3).cumsum() / df['volume'].cumsum()
    
    # 出来高移動平均
    df['volume_SMA'] = df['volume'].rolling(window=20).mean()
    df['volume_ratio'] = df['volume'] / df['volume_SMA']
    
    return df

実践的なデータ前処理のテクニック

1. 欠損値の高度な処理

def handle_missing_values(df):
    """
    欠損値の高度な処理
    """
    # 欠損値の割合を確認
    missing_ratio = df.isnull().sum() / len(df)
    
    # 線形補間を使用（価格データに適している）
    price_columns = ['open', 'high', 'low', 'close']
    for col in price_columns:
        if col in df.columns:
            df[col] = df[col].interpolate(method='linear')
    
    # 出来高は0で補完（取引がなかったと仮定）
    if 'volume' in df.columns:
        df['volume'].fillna(0, inplace=True)
    
    # テクニカル指標は前方補完
    indicator_columns = [col for col in df.columns if col not in price_columns + ['volume']]
    for col in indicator_columns:
        df[col].fillna(method='ffill', inplace=True)
    
    return df

2. 外れ値の検出と処理

def detect_and_handle_outliers(df, columns=['close'], n_std=3):
    """
    統計的手法による外れ値の検出と処理
    """
    for col in columns:
        if col in df.columns:
            # ローリング平均と標準偏差を計算
            rolling_mean = df[col].rolling(window=20).mean()
            rolling_std = df[col].rolling(window=20).std()
            
            # 外れ値の閾値を設定
            upper_bound = rolling_mean + (n_std * rolling_std)
            lower_bound = rolling_mean - (n_std * rolling_std)
            
            # 外れ値をクリップ
            df[f'{col}_cleaned'] = df[col].clip(lower=lower_bound, upper=upper_bound)
            
            # 外れ値フラグを作成
            df[f'{col}_outlier'] = ((df[col] > upper_bound) | (df[col] < lower_bound)).astype(int)
    
    return df

3. 特徴量の正規化とスケーリング

from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler

def normalize_features(df, method='standard'):
    """
    特徴量の正規化
    """
    # 価格関連の特徴量を抽出
    feature_columns = [col for col in df.columns if col not in ['timestamp']]
    
    if method == 'standard':
        scaler = StandardScaler()
    elif method == 'minmax':
        scaler = MinMaxScaler()
    elif method == 'robust':
        scaler = RobustScaler()  # 外れ値に強い
    
    # スケーリングを実行
    df_scaled = pd.DataFrame(
        scaler.fit_transform(df[feature_columns]),
        columns=[f'{col}_scaled' for col in feature_columns],
        index=df.index
    )
    
    # 元のデータフレームと結合
    df = pd.concat([df, df_scaled], axis=1)
    
    return df, scaler

高度な特徴量エンジニアリング

予測精度をさらに向上させるための高度な特徴量を生成します：

def create_advanced_features(df):
    """
    高度な特徴量の生成
    """
    # 価格変化率（リターン）
    df['returns'] = df['close'].pct_change()
    df['log_returns'] = np.log(df['close'] / df['close'].shift(1))
    
    # 価格の移動平均からの乖離率
    df['price_sma20_ratio'] = df['close'] / df['SMA_20']
    df['price_sma50_ratio'] = df['close'] / df['SMA_50']
    
    # ボリンジャーバンドの位置
    df['bb_position'] = (df['close'] - df['BB_lower']) / (df['BB_upper'] - df['BB_lower'])
    
    # 高値・安値からの位置
    df['high_low_ratio'] = (df['close'] - df['low']) / (df['high'] - df['low'])
    
    # ローリング統計量
    for window in [5, 10, 20]:
        df[f'return_mean_{window}'] = df['returns'].rolling(window=window).mean()
        df[f'return_std_{window}'] = df['returns'].rolling(window=window).std()
        df[f'volume_mean_{window}'] = df['volume'].rolling(window=window).mean()
    
    # 時間的特徴量
    df['hour'] = df.index.hour
    df['day_of_week'] = df.index.dayofweek
    df['month'] = df.index.month
    
    # 価格パターン検出
    df['higher_high'] = ((df['high'] > df['high'].shift(1)) & 
                        (df['high'].shift(1) > df['high'].shift(2))).astype(int)
    df['lower_low'] = ((df['low'] < df['low'].shift(1)) & 
                       (df['low'].shift(1) < df['low'].shift(2))).astype(int)
    
    return df

📘 完全版をnoteで公開中

本記事では基本的な概念と実装例を紹介しましたが、実際に収益を生み出す予測モデルの構築には、より詳細な実装が必要です。

note有料記事では以下の内容を完全収録：

✅ 50種類以上のテクニカル指標の具体的な実装コード ✅ 機関投資家が使用する高度な特徴量（Order Flow Imbalance、Microstructure等） ✅ データクリーニングの完全自動化スクリプト（APIエラー対応含む） ✅ バックテストで検証済みの最適なパラメータ設定（通貨ペア別） ✅ 実際の取引で使用している前処理パイプラインの完全コード ✅ 特徴量重要度分析と自動選択アルゴリズム

さらに、購入者限定特典として：

📊 過去3年分の処理済みデータセット（BTC/ETH/BNB/SOL）
💬 専用Discordでの質問サポート（24時間以内回答）
🔄 コードの定期アップデート（月1回）
📈 リアルタイムデータ処理のサンプルコード

🔗 note有料記事はこちら

まとめ

機械学習による価格予測では、データの前処理が成功の鍵を握ります。適切な特徴量エンジニアリングにより、予測精度を大幅に向上させることができます。

本記事で紹介した手法を実装することで、基本的な予測モデルの構築が可能になりますが、実際の取引で利益を生み出すには、さらに高度な最適化が必要です。

次回の第2部では、Prophet とLSTMを使った予測モデルの構築と、ハイパーパラメータの最適化について詳しく解説します。