機械学習 × FX バックテストをPythonで正しく実装する：データリーク・過学習を防ぐ設計

import pandas as pd
import numpy as np

# データ読み込み（時系列順）
df = pd.read_csv('usdjpy_daily.csv', parse_dates=['date'], index_col='date')
df = df.sort_index()  # 時間順にソート

# 正しい分割（時間順を保持）
total_len = len(df)
train_end = int(total_len * 0.70)
val_end = int(total_len * 0.85)

train_df = df.iloc[:train_end]      # 訓練: 全体の70%
val_df = df.iloc[train_end:val_end] # 検証: 全体の15%
test_df = df.iloc[val_end:]         # テスト: 全体の15%（未来データ）

# NG: ランダム分割（時系列ではこれをやってはいけない）
# from sklearn.model_selection import train_test_split
# X_train, X_test = train_test_split(X, test_size=0.3, shuffle=True)  # 絶対禁止

from sklearn.model_selection import TimeSeriesSplit

# 時系列専用のクロスバリデーション
tscv = TimeSeriesSplit(n_splits=5, gap=20)  # gap: 訓練と検証の間に設けるバッファ

for fold, (train_idx, test_idx) in enumerate(tscv.split(X)):
    X_train, X_test = X[train_idx], X[test_idx]
    y_train, y_test = y[train_idx], y[test_idx]
    
    # モデル訓練
    model.fit(X_train, y_train)
    
    # 各フォールドでの評価
    score = model.score(X_test, y_test)
    print(f"Fold {fold+1}: 方向精度 = {score:.4f}")

def create_features(df, lookback=20):
    """過去データのみを使った特徴量生成"""
    features = pd.DataFrame(index=df.index)
    
    # 価格変化率（過去のみ参照）
    features['return_1d'] = df['close'].pct_change(1)
    features['return_5d'] = df['close'].pct_change(5)
    features['return_20d'] = df['close'].pct_change(20)
    
    # ボラティリティ（過去20日）
    features['volatility_20d'] = df['close'].pct_change().rolling(20).std()
    
    # RSI（過去データのみ使用）
    def rsi(prices, period=14):
        delta = prices.diff()
        gain = delta.where(delta > 0, 0).rolling(period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(period).mean()
        rs = gain / loss
        return 100 - (100 / (1 + rs))
    
    features['rsi_14'] = rsi(df['close'])
    
    # 移動平均乖離率
    ma20 = df['close'].rolling(20).mean()
    features['ma20_deviation'] = (df['close'] - ma20) / ma20
    
    # ラベル生成（注意: 未来の価格を使うが、訓練時の扱いに注意）
    features['label'] = (df['close'].shift(-1) > df['close']).astype(int)
    
    return features.dropna()

features_df = create_features(df)

pip install backtesting scikit-learn xgboost

from backtesting import Backtest, Strategy
from backtesting.lib import crossover
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
import numpy as np

class MLStrategy(Strategy):
    # クラス変数として機械学習モデルを渡す
    model = None
    lookback = 20
    
    def init(self):
        # 特徴量を事前に計算しておく
        pass
    
    def next(self):
        # next()内では「現在以前のデータのみ」参照できる
        if len(self.data.Close) < self.lookback + 5:
            return
        
        # 直近データで特徴量を計算
        closes = pd.Series(self.data.Close[-self.lookback:])
        returns = closes.pct_change().dropna()
        
        # 特徴量ベクトル（単純化した例）
        features = np.array([
            returns.mean(),       # 平均リターン
            returns.std(),        # ボラティリティ
            returns.iloc[-1],     # 直近リターン
        ]).reshape(1, -1)
        
        # モデルで予測
        if self.model is not None:
            prediction = self.model.predict(features)[0]
            
            if prediction == 1 and not self.position:
                self.buy()
            elif prediction == 0 and self.position:
                self.position.close()

# モデルの訓練（テストデータは渡さない）
# ... （特徴量生成・訓練コードを挿入）

# バックテスト実行
MLStrategy.model = trained_model
bt = Backtest(test_df, MLStrategy, 
              cash=1_000_000,
              commission=0.0002,  # スプレッドを考慮
              trade_on_close=False)
result = bt.run()
print(result)