AI FX予測の精度を検証する｜研究データと自分で確かめる方法

方向予測精度 = 正しく予測した件数 / 全予測件数 × 100%

import numpy as np
import pandas as pd
from sklearn.metrics import accuracy_score, classification_report

def calculate_baseline_accuracy(prices: pd.Series, horizon_bars: int = 5) -> dict:
    """
    ベースライン精度を計算する
    ランダム予測と「常に上昇」予測の精度を確認
    """
    
    # 実際の方向（N本後の変化）
    future_return = prices.shift(-horizon_bars) / prices - 1
    actual_direction = (future_return > 0).astype(int)
    
    # NaNを除外
    valid_mask = actual_direction.notna()
    actual = actual_direction[valid_mask].values
    
    # ランダム予測（理論値は50%）
    np.random.seed(42)
    random_pred = np.random.randint(0, 2, size=len(actual))
    random_acc = accuracy_score(actual, random_pred)
    
    # 常に上昇を予測
    always_up_pred = np.ones(len(actual), dtype=int)
    always_up_acc = accuracy_score(actual, always_up_pred)
    
    # 常に前バーの方向を予測（モメンタム）
    prev_direction = (prices.diff() > 0).astype(int).shift(1)
    momentum_pred = prev_direction[valid_mask].values
    momentum_acc = accuracy_score(actual[:len(momentum_pred)], momentum_pred[:len(actual)])
    
    up_rate = actual.mean()
    
    print(f"=== ベースライン精度 ({horizon_bars}本後方向予測) ===")
    print(f"実際の上昇率: {up_rate:.3f} ({up_rate*100:.1f}%)")
    print(f"ランダム予測精度: {random_acc:.3f} ({random_acc*100:.1f}%)")
    print(f"常に上昇予測精度: {always_up_acc:.3f} ({always_up_acc*100:.1f}%)")
    print(f"前バー方向モメンタム精度: {momentum_acc:.3f} ({momentum_acc*100:.1f}%)")
    
    return {
        "up_rate": up_rate,
        "random_accuracy": random_acc,
        "always_up_accuracy": always_up_acc,
        "momentum_accuracy": momentum_acc
    }

from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import TimeSeriesSplit
import lightgbm as lgb

def compare_models(X: pd.DataFrame, y: pd.Series) -> pd.DataFrame:
    """
    複数モデルをウォークフォワードで比較評価する
    """
    
    models = {
        "LogisticRegression": LogisticRegression(max_iter=1000, C=0.1),
        "RandomForest": RandomForestClassifier(n_estimators=100, max_depth=3, random_state=42),
        "LightGBM": lgb.LGBMClassifier(n_estimators=100, max_depth=3,
                                         learning_rate=0.05, random_state=42)
    }
    
    tscv = TimeSeriesSplit(n_splits=5)
    results = {name: [] for name in models}
    
    for fold, (train_idx, test_idx) in enumerate(tscv.split(X)):
        X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
        y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
        
        for name, model in models.items():
            try:
                model.fit(X_train, y_train)
                y_pred = model.predict(X_test)
                acc = accuracy_score(y_test, y_pred)
                results[name].append(acc)
                print(f"  Fold {fold+1} | {name}: {acc:.3f}")
            except Exception as e:
                print(f"  Fold {fold+1} | {name}: エラー ({e})")
                results[name].append(0.5)  # エラー時は50%として記録
    
    summary = pd.DataFrame({
        name: {
            "mean_accuracy": np.mean(scores),
            "std_accuracy": np.std(scores),
            "min_accuracy": np.min(scores),
            "max_accuracy": np.max(scores)
        }
        for name, scores in results.items()
    }).T
    
    summary["beat_random"] = summary["mean_accuracy"] > 0.51  # 1%以上の優位性があるか
    
    return summary

def comprehensive_signal_evaluation(
    predictions: np.ndarray,
    actuals: np.ndarray,
    price_changes: np.ndarray,  # 実際の価格変化（pips）
    spread_pips: float = 1.5
) -> dict:
    """
    シグナルの総合評価（方向精度 + 収益性の両面）
    
    Args:
        predictions: 予測方向（1=上昇, 0=下落）
        actuals: 実際の方向（1=上昇, 0=下落）
        price_changes: 実際の価格変化（pips）
        spread_pips: スプレッド（往復）
    """
    
    n = len(predictions)
    
    # 1. 方向精度
    direction_accuracy = accuracy_score(actuals, predictions)
    
    # 2. 予測が「上昇」のときの正解率（精度）
    long_signals = predictions == 1
    if long_signals.sum() > 0:
        long_precision = actuals[long_signals].mean()
    else:
        long_precision = 0
    
    # 3. 予測が「下落」のときの正解率
    short_signals = predictions == 0
    if short_signals.sum() > 0:
        short_precision = (1 - actuals[short_signals]).mean()
    else:
        short_precision = 0
    
    # 4. 理論的なPnL（スプレッド込み）
    pnl_per_trade = []
    for i in range(n):
        if predictions[i] == 1:  # ロング
            pnl = price_changes[i] - spread_pips
        else:  # ショート
            pnl = -price_changes[i] - spread_pips
        pnl_per_trade.append(pnl)
    
    pnl_array = np.array(pnl_per_trade)
    
    winning_trades = pnl_array > 0
    losing_trades = pnl_array <= 0
    
    win_rate = winning_trades.mean()
    avg_win = pnl_array[winning_trades].mean() if winning_trades.sum() > 0 else 0
    avg_loss = abs(pnl_array[losing_trades].mean()) if losing_trades.sum() > 0 else 0
    profit_factor = (avg_win * winning_trades.sum()) / (avg_loss * losing_trades.sum()) \
                    if avg_loss * losing_trades.sum() > 0 else 0
    
    total_pnl = pnl_array.sum()
    sharpe_approx = pnl_array.mean() / pnl_array.std() if pnl_array.std() > 0 else 0
    
    return {
        "direction_accuracy": direction_accuracy,
        "long_precision": long_precision,
        "short_precision": short_precision,
        "signal_count": n,
        "long_signal_count": long_signals.sum(),
        "short_signal_count": short_signals.sum(),
        "win_rate_with_spread": win_rate,
        "avg_win_pips": avg_win,
        "avg_loss_pips": avg_loss,
        "profit_factor": profit_factor,
        "total_pnl_pips": total_pnl,
        "sharpe_approx": sharpe_approx,
        "note": "スリッページ未考慮。実運用ではさらに悪化する可能性あり"
    }