[データ取得] → [欠損値処理] → [外れ値除去] → [正規化・スケーリング]
     ↓
[テクニカル指標生成（SMA/EMA/RSI/MACD/BB/ATR/OBV）← 特徴量エンジニアリング]
     ↓
[データリーク防止チェック] → [時系列分割（ウォークフォワード法）] → [学習データ完成！！🎉]

pip install ccxt pandas pandas-ta scikit-learn keras tensorflow matplotlib

import sys
import ccxt
import pandas as pd

# Python実行環境の確認
print(f"Python バージョン: {sys.version}")

# Binance取引所に接続
exchange = ccxt.binance({
    'enableRateLimit': True,
    'options': {
        'defaultType': 'spot',
        'adjustForTimeDifference': True,
    },
    'headers': {
        'User-Agent': 'Mozilla/5.0',
    },
})

def fetch_ohlcv(symbol: str, timeframe: str = '1h', limit: int = 1000) -> pd.DataFrame:
    """
    Binance APIからOHLCVデータを取得する

    Args:
        symbol: 通貨ペア（例: 'BTC/USDT'）
        timeframe: 時間足（'15m', '1h', '4h', '1d'）
        limit: 取得件数（最大1000）

    Returns:
        OHLCVデータのDataFrame
    """
    try:
        exchange.load_markets()
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
    except ccxt.BadSymbol as e:
        raise ValueError(f"無効なシンボルです: {symbol}") from e
    except ccxt.NetworkError as e:
        raise ConnectionError(f"ネットワークエラー: {e}") from e
    except ccxt.ExchangeError as e:
        raise RuntimeError(f"取引所APIエラー: {e}") from e

    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

# BTC/USDTの1時間足データを1000本取得
df = fetch_ohlcv('BTC/USDT', timeframe='1h', limit=1000)

print(df.head())
print(f"\nデータ件数: {len(df)}")
print(f"期間: {df.index[0]} 〜 {df.index[-1]}")

import numpy as np
import pandas as pd

def handle_missing_values(df: pd.DataFrame) -> pd.DataFrame:
    """
    欠損値の検出と処理

    Args:
        df: OHLCVデータのDataFrame

    Returns:
        欠損値処理済みのDataFrame
    """
    missing_count = df.isnull().sum()
    print("=== 欠損値の確認 ===")
    print(missing_count)

    missing_ratio = df.isnull().mean()

    for col in df.columns:
        if missing_ratio[col] == 0:
            continue
        elif missing_ratio[col] < 0.05:
            # 欠損率5%未満 → forward-fill（pandas 2.x対応）
            df[col] = df[col].ffill().bfill()
        else:
            # 欠損率が高い場合 → 線形補間
            df[col] = df[col].interpolate(method='linear')

    print(f"\n処理後の欠損値: {df.isnull().sum().sum()} 件")
    return df

df = handle_missing_values(df)

def remove_outliers_iqr(df: pd.DataFrame, columns: list, threshold: float = 3.0) -> pd.DataFrame:
    """
    IQR法による外れ値の除去

    Args:
        df: 対象のDataFrame
        columns: 外れ値チェック対象の列名リスト
        threshold: IQRの何倍を外れ値とするか（デフォルト3.0）

    Returns:
        外れ値処理済みのDataFrame
    """
    df_clean = df.copy()

    for col in columns:
        Q1 = df_clean[col].quantile(0.25)
        Q3 = df_clean[col].quantile(0.75)
        IQR = Q3 - Q1

        lower = Q1 - threshold * IQR
        upper = Q3 + threshold * IQR

        outlier_count = ((df_clean[col] < lower) | (df_clean[col] > upper)).sum()
        print(f"{col}: 外れ値 {outlier_count} 件を検出")

        # 外れ値はNaNに置換してからforward-fill
        df_clean.loc[(df_clean[col] < lower) | (df_clean[col] > upper), col] = np.nan
        df_clean[col] = df_clean[col].ffill().bfill()

    return df_clean

df = remove_outliers_iqr(df, columns=['close', 'volume'])

from sklearn.preprocessing import MinMaxScaler, RobustScaler
import numpy as np

def scale_features(df: pd.DataFrame, method: str = 'minmax') -> tuple:
    """
    特徴量のスケーリング

    Args:
        df: 特徴量のDataFrame
        method: 'minmax' / 'robust' / 'log_return'

    Returns:
        (スケーリング済みDataFrame, scalerオブジェクト)
    """
    df_scaled = df.copy()

    if method == 'minmax':
        scaler = MinMaxScaler(feature_range=(0, 1))
        price_cols = ['open', 'high', 'low', 'close']
        df_scaled[price_cols] = scaler.fit_transform(df[price_cols])
        # 出来高は対数変換してからMin-Maxスケーリング
        vol_scaler = MinMaxScaler(feature_range=(0, 1))
        df_scaled['volume'] = vol_scaler.fit_transform(
            np.log1p(df[['volume']])
        )
        return df_scaled, scaler

    elif method == 'robust':
        scaler = RobustScaler()
        feature_cols = ['open', 'high', 'low', 'close', 'volume']
        df_scaled[feature_cols] = scaler.fit_transform(df[feature_cols])
        return df_scaled, scaler

    elif method == 'log_return':
        df_scaled['log_return'] = np.log(df['close'] / df['close'].shift(1))
        df_scaled = df_scaled.dropna()
        return df_scaled, None

    else:
        raise ValueError(f"未対応のメソッドです: {method}")

# LSTMを使う場合はminmax推奨
df_scaled, scaler = scale_features(df, method='minmax')
print(f"スケーリング完了！！ shape: {df_scaled.shape}")

import pandas_ta as ta

def add_technical_indicators(df: pd.DataFrame) -> pd.DataFrame:
    """
    テクニカル指標の一括生成（特徴量エンジニアリング）

    Args:
        df: OHLCVデータのDataFrame

    Returns:
        テクニカル指標を追加したDataFrame
    """
    df_ind = df.copy()

    # --- トレンド系：SMA（単純移動平均）---
    df_ind['sma_7']  = ta.sma(df['close'], length=7)
    df_ind['sma_25'] = ta.sma(df['close'], length=25)
    df_ind['sma_99'] = ta.sma(df['close'], length=99)

    # --- トレンド系：EMA（指数移動平均）---
    df_ind['ema_12'] = ta.ema(df['close'], length=12)
    df_ind['ema_26'] = ta.ema(df['close'], length=26)

    # --- モメンタム系：RSI（相対力指数）---
    # 70超 → 買われすぎ、30未満 → 売られすぎ
    df_ind['rsi_14'] = ta.rsi(df['close'], length=14)

    # --- モメンタム系：MACD ---
    macd = ta.macd(df['close'], fast=12, slow=26, signal=9)
    df_ind['macd']        = macd['MACD_12_26_9']
    df_ind['macd_signal'] = macd['MACDs_12_26_9']
    df_ind['macd_hist']   = macd['MACDh_12_26_9']

    # --- ボラティリティ系：ボリンジャーバンド ---
    bb = ta.bbands(df['close'], length=20, std=2)
    df_ind['bb_upper']  = bb['BBU_20_2.0']
    df_ind['bb_middle'] = bb['BBM_20_2.0']
    df_ind['bb_lower']  = bb['BBL_20_2.0']
    # バンド幅（ボラティリティの大きさを数値化）
    df_ind['bb_width'] = (
        (df_ind['bb_upper'] - df_ind['bb_lower']) / df_ind['bb_middle']
    )

    # --- ボラティリティ系：ATR（平均真の値動き）---
    df_ind['atr_14'] = ta.atr(df['high'], df['low'], df['close'], length=14)

    # --- 出来高系：OBV（オンバランスボリューム）---
    df_ind['obv'] = ta.obv(df['close'], df['volume'])

    # 最長ウィンドウ（SMA99）分のNaN行を削除
    df_ind.dropna(inplace=True)

    print(f"特徴量数: {len(df_ind.columns)} 列")
    print(f"有効データ件数: {len(df_ind)}")
    print(df_ind.tail())

    return df_ind

df_with_features = add_technical_indicators(df)

import matplotlib.pyplot as plt

def plot_indicators(df: pd.DataFrame, n: int = 200) -> None:
    """
    主要テクニカル指標を可視化する

    Args:
        df: テクニカル指標付きのDataFrame
        n: 直近何本分を表示するか
    """
    df_plot = df.tail(n)

    fig, axes = plt.subplots(4, 1, figsize=(14, 12), sharex=True)

    # 価格 + ボリンジャーバンド
    axes[0].plot(df_plot['close'], label='Close', color='black', linewidth=1.0)
    axes[0].plot(df_plot['bb_upper'], label='BB Upper', color='blue', linestyle='--', alpha=0.6)
    axes[0].plot(df_plot['bb_lower'], label='BB Lower', color='blue', linestyle='--', alpha=0.6)
    axes[0].fill_between(df_plot.index, df_plot['bb_lower'], df_plot['bb_upper'], alpha=0.1)
    axes[0].set_title('Close Price + Bollinger Bands')
    axes[0].legend()

    # RSI
    axes[1].plot(df_plot['rsi_14'], label='RSI(14)', color='purple')
    axes[1].axhline(70, color='red', linestyle='--', alpha=0.7)
    axes[1].axhline(30, color='green', linestyle='--', alpha=0.7)
    axes[1].set_title('RSI (14)')
    axes[1].legend()

    # MACD
    axes[2].plot(df_plot['macd'], label='MACD', color='blue')
    axes[2].plot(df_plot['macd_signal'], label='Signal', color='orange')
    axes[2].bar(df_plot.index, df_plot['macd_hist'], label='Histogram', color='gray', alpha=0.5)
    axes[2].set_title('MACD')
    axes[2].legend()

    # 出来高 + OBV
    axes[3].bar(df_plot.index, df_plot['volume'], label='Volume', color='teal', alpha=0.5)
    ax3b = axes[3].twinx()
    ax3b.plot(df_plot['obv'], label='OBV', color='red', linewidth=0.8)
    axes[3].set_title('Volume + OBV')
    axes[3].legend(loc='upper left')
    ax3b.legend(loc='upper right')

    plt.tight_layout()
    plt.savefig('indicators_check.png', dpi=150)
    plt.show()

plot_indicators(df_with_features)

from sklearn.model_selection import TimeSeriesSplit
from sklearn.preprocessing import MinMaxScaler
import numpy as np

def walk_forward_split(
    df: pd.DataFrame,
    n_splits: int = 5,
    target_col: str = 'close'
) -> list:
    """
    ウォークフォワード法による時系列分割（データリーク・過学習防止済み）

    Args:
        df: 特徴量DataFrame
        n_splits: 分割数（デフォルト5）
        target_col: 目的変数の列名

    Returns:
        (X_train, X_test, y_train, y_test, scaler) のリスト
    """
    tscv = TimeSeriesSplit(n_splits=n_splits)

    feature_cols = [col for col in df.columns if col != target_col]
    X = df[feature_cols].values
    y = df[target_col].values

    splits = []

    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]

        # ポイント：訓練データのみでfit → テストはtransformのみ
        scaler = MinMaxScaler(feature_range=(0, 1))
        X_train_scaled = scaler.fit_transform(X_train)
        X_test_scaled  = scaler.transform(X_test)

        splits.append((X_train_scaled, X_test_scaled, y_train, y_test, scaler))

        print(f"Fold {fold + 1}: 訓練={len(train_idx)}件 / テスト={len(test_idx)}件")

    return splits

splits = walk_forward_split(df_with_features, n_splits=5)
print(f"\n時系列分割完了！！ {len(splits)} folds 準備OKデス✨")