生OHLCVデータ取得（CCXT）
  → 欠損値の線形補間
  → IQRクリッピングで外れ値処理
  → RSI / MACD / ATR / ボリンジャーバンド追加
  → RobustScalerで正規化（データリーク防止）
  → Train/Test分割（80:20）

Python       3.12+
ccxt         4.x
pandas-ta    0.3.14b
scikit-learn 1.4+
pandas       2.x
numpy        1.26+

pip3 install ccxt pandas-ta scikit-learn pandas numpy matplotlib seaborn

pip3 install ccxt

import ccxt
import pandas as pd

exchange = ccxt.binance({'enableRateLimit': True})

# 基本的なOHLCV取得（最新1000件）
ohlcv = exchange.fetch_ohlcv('BTC/USDT', timeframe='1h', limit=1000)

df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
df = df.set_index('timestamp')

print(df.tail())

import ccxt
import pandas as pd
import time


def fetch_ohlcv_all(
    symbol: str = 'BTC/USDT',
    timeframe: str = '1h',
    start_date: str = '2022-01-01',
    exchange_name: str = 'binance'
) -> pd.DataFrame:
    """
    CCXTでページネーションを使った全件OHLCV取得

    Args:
        symbol       : 通貨ペア（例: 'BTC/USDT'）
        timeframe    : 時間足（例: '1h', '4h', '1d'）
        start_date   : 取得開始日（'YYYY-MM-DD'形式）
        exchange_name: 取引所名

    Returns:
        OHLCVデータのDataFrame（timestampをインデックス）
    """
    exchange = getattr(ccxt, exchange_name)({
        'rateLimit': 1200,
        'enableRateLimit': True,
    })

    since = exchange.parse8601(f'{start_date}T00:00:00Z')
    all_ohlcv = []
    limit = 1000  # Binanceの1リクエスト上限

    print(f"データ取得開始: {symbol} / {timeframe} / {start_date} 〜")

    while True:
        try:
            ohlcv = exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=limit)

            if not ohlcv:
                break

            all_ohlcv.extend(ohlcv)
            since = ohlcv[-1][0] + 1  # 最後のタイムスタンプの次から取得

            print(f"  取得中: {len(all_ohlcv)}件...")

            # レート制限対策：1秒待機
            exchange.sleep(1000)

            # 現在時刻を超えたら終了
            if since > exchange.milliseconds():
                break

        except ccxt.NetworkError as e:
            print(f"ネットワークエラー: {e} → 3秒後リトライ")
            time.sleep(3)
        except ccxt.ExchangeError as e:
            print(f"取引所エラー: {e}")
            break

    # DataFrameに変換・整形
    df = pd.DataFrame(
        all_ohlcv,
        columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
    )
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
    df = df.set_index('timestamp').sort_index()
    df = df[~df.index.duplicated(keep='first')]  # 重複除去
    df = df.astype(float)

    print(f"取得完了: {len(df)}件 | {df.index[0]} 〜 {df.index[-1]}")
    return df


# 実行例
if __name__ == '__main__':
    df_raw = fetch_ohlcv_all(
        symbol='BTC/USDT',
        timeframe='1h',
        start_date='2022-01-01'
    )
    print(df_raw.shape)
    print(df_raw.head())

# 欠損値の確認
missing = df_raw.isnull().sum()
missing_pct = (missing / len(df_raw)) * 100
print(missing_pct)

import pandas as pd
import numpy as np


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

    - 価格（OHLC）: 線形補間で時系列の連続性を保持
    - 出来高（Volume）: 0補完（取引なしとして扱う）
    """
    df = df.copy()

    # --- 欠損値レポート ---
    missing = df.isnull().sum()
    missing_pct = (missing / len(df)) * 100

    print("=== 欠損値レポート ===")
    for col in df.columns:
        if missing[col] > 0:
            print(f"  {col}: {missing[col]}件 ({missing_pct[col]:.2f}%)")
        else:
            print(f"  {col}: 欠損なし")

    if (missing_pct > 5).any():
        print("WARNING: 欠損値が5%超のカラムがあります。データソースを確認してください。")

    # --- 欠損値補完 ---
    # 価格は線形補間
    for col in ['open', 'high', 'low', 'close']:
        if col in df.columns:
            df[col] = df[col].interpolate(method='linear', limit_direction='both')

    # 出来高は0補完（取引なし）
    if 'volume' in df.columns:
        df['volume'] = df['volume'].fillna(0)

    # 補完後の確認
    remaining_missing = df.isnull().sum().sum()
    print(f"\n補完後の欠損値合計: {remaining_missing}件")

    return df


# 実行例
df_clean = handle_missing_values(df_raw)
print(df_clean.isnull().sum())

下限 = Q1 - multiplier × IQR
上限 = Q3 + multiplier × IQR

import pandas as pd
import numpy as np


def clip_outliers_iqr(
    df: pd.DataFrame,
    columns: list = None,
    multiplier: float = 3.0
) -> pd.DataFrame:
    """
    IQR法による外れ値クリッピング

    Args:
        df        : 対象DataFrame
        columns   : 処理対象カラム（Noneなら数値カラム全て）
        multiplier: IQR倍率（デフォルト3.0）

    Returns:
        外れ値クリッピング済みのDataFrame
    """
    df = df.copy()

    if columns is None:
        columns = df.select_dtypes(include=[np.number]).columns.tolist()

    clip_report = {}

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

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

        n_clipped = ((df[col] < lower) | (df[col] > upper)).sum()
        clip_report[col] = {
            'lower': lower,
            'upper': upper,
            'n_clipped': n_clipped,
        }

        df[col] = df[col].clip(lower=lower, upper=upper)

    print("=== IQRクリッピング レポート ===")
    for col, info in clip_report.items():
        print(
            f"  {col}: {info['n_clipped']}件クリップ"
            f" | 範囲 [{info['lower']:.4f}, {info['upper']:.4f}]"
        )

    return df


# 実行例
df_clipped = clip_outliers_iqr(df_clean, multiplier=3.0)
print(df_clipped.describe())

import pandas as pd
import pandas_ta as ta


def add_technical_indicators(df: pd.DataFrame) -> pd.DataFrame:
    """
    テクニカル指標の追加

    追加指標: RSI(14), MACD(12,26,9), ATR(14), ボリンジャーバンド(20,2σ)

    Args:
        df: OHLCVカラムを持つDataFrame

    Returns:
        テクニカル指標カラムを追加したDataFrame
        ※ウォームアップ期間のNaN行は削除済み
    """
    df = df.copy()

    # RSI（14期間）
    df['RSI_14'] = ta.rsi(df['close'], length=14)

    # MACD（12, 26, 9）
    macd = ta.macd(df['close'], fast=12, slow=26, signal=9)
    df['MACD']        = macd['MACD_12_26_9']
    df['MACD_signal'] = macd['MACDs_12_26_9']
    df['MACD_hist']   = macd['MACDh_12_26_9']

    # ATR（14期間）
    df['ATR_14'] = ta.atr(df['high'], df['low'], df['close'], length=14)

    # ボリンジャーバンド（20期間・2σ）
    bbands = ta.bbands(df['close'], length=20, std=2)
    df['BB_upper']  = bbands['BBU_20_2.0']
    df['BB_middle'] = bbands['BBM_20_2.0']
    df['BB_lower']  = bbands['BBL_20_2.0']
    # バンド幅（ボラティリティの相対的な大きさ）
    df['BB_width'] = (df['BB_upper'] - df['BB_lower']) / df['BB_middle']

    # ウォームアップ期間（NaN行）を削除
    before = len(df)
    df = df.dropna()
    after = len(df)

    print(f"テクニカル指標追加完了")
    print(f"  追加カラム: RSI_14, MACD, MACD_signal, MACD_hist, ATR_14, BB_upper, BB_middle, BB_lower, BB_width")
    print(f"  ウォームアップ期間で削除した行数: {before - after}件")
    print(f"  処理後データ件数: {after}件")

    return df


# 実行例
df_featured = add_technical_indicators(df_clipped)
print(df_featured.columns.tolist())
print(df_featured.tail())

# NG: 全データでfit（データリーク発生！）
scaler.fit(df_all)

# OK: Trainデータのみでfit
scaler.fit(df_train)
df_train_scaled = scaler.transform(df_train)
df_test_scaled  = scaler.transform(df_test)  # transformのみ

import pandas as pd
import numpy as np
from sklearn.preprocessing import RobustScaler


def split_and_scale(
    df: pd.DataFrame,
    test_ratio: float = 0.2
) -> tuple[pd.DataFrame, pd.DataFrame, RobustScaler]:
    """
    時系列Train/Test分割 + RobustScalerによる正規化（データリーク防止）

    Args:
        df        : テクニカル指標追加済みのDataFrame
        test_ratio: テストデータの割合（デフォルト0.2 = 20%）

    Returns:
        (df_train_scaled, df_test_scaled, scaler)
        ※scalerは予測値の逆変換に必要なため返却
    """
    # 時系列分割（シャッフルしない！）
    split_idx = int(len(df) * (1 - test_ratio))
    df_train = df.iloc[:split_idx].copy()
    df_test  = df.iloc[split_idx:].copy()

    print("=== Train/Test 分割結果 ===")
    print(f"  Train: {len(df_train)}件 | {df_train.index[0]} 〜 {df_train.index[-1]}")
    print(f"  Test : {len(df_test)}件  | {df_test.index[0]} 〜 {df_test.index[-1]}")

    feature_cols = df.columns.tolist()

    # スケーラーをTrainデータのみでfit（データリーク防止）
    scaler = RobustScaler()
    scaler.fit(df_train[feature_cols])

    # 変換
    df_train_scaled = df_train.copy()
    df_test_scaled  = df_test.copy()

    df_train_scaled[feature_cols] = scaler.transform(df_train[feature_cols])
    df_test_scaled[feature_cols]  = scaler.transform(df_test[feature_cols])

    print("\n=== 正規化後の統計量（Train） ===")
    print(df_train_scaled.describe().round(3))

    return df_train_scaled, df_test_scaled, scaler


# 実行例
df_train, df_test, scaler = split_and_scale(df_featured, test_ratio=0.2)
print(f"\nTrain shape: {df_train.shape}")
print(f"Test  shape: {df_test.shape}")

# ===== 前処理パイプライン 完成形 =====

# Step 1: OHLCVデータ取得
df_raw = fetch_ohlcv_all(
    symbol='BTC/USDT',
    timeframe='1h',
    start_date='2022-01-01'
)

# Step 2: 欠損値補完
df_clean = handle_missing_values(df_raw)

# Step 3: 外れ値クリッピング
df_clipped = clip_outliers_iqr(df_clean, multiplier=3.0)

# Step 4: テクニカル指標追加
df_featured = add_technical_indicators(df_clipped)

# Step 5: 正規化 + Train/Test分割
df_train, df_test, scaler = split_and_scale(df_featured, test_ratio=0.2)

print("前処理パイプライン完了！！")
print(f"Train: {df_train.shape}, Test: {df_test.shape}")