Python 3.11.x
prophet==1.1.5
tensorflow==2.15.0
keras==2.15.0
optuna==3.5.0
ccxt==4.2.x
ta==0.11.0
backtesting==0.3.3
pandas==2.1.x
numpy==1.26.x
scikit-learn==1.4.x
plotly==5.18.x

pip3 install ccxt ta prophet tensorflow optuna backtesting plotly scikit-learn

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

def fetch_ohlcv_data(
    symbol: str = 'BTC/USDT',
    timeframe: str = '1d',
    since_str: str = '2020-01-01',
    exchange_id: str = 'binance'
) -> pd.DataFrame:
    """
    ccxtでOHLCVデータを取得してDataFrameに変換する

    Args:
        symbol: 取引ペア（例: 'BTC/USDT'）
        timeframe: 時間軸（'1d', '4h', '1h' など）
        since_str: 取得開始日（'YYYY-MM-DD' 形式）
        exchange_id: 取引所ID

    Returns:
        OHLCV DataFrameに変換したデータ
    """
    exchange = getattr(ccxt, exchange_id)({'enableRateLimit': True})
    since = exchange.parse8601(f'{since_str}T00:00:00Z')

    all_ohlcv = []
    limit = 1000

    while True:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=limit)
        if not ohlcv:
            break
        all_ohlcv.extend(ohlcv)
        since = ohlcv[-1][0] + 1
        if len(ohlcv) < limit:
            break

    df = pd.DataFrame(
        all_ohlcv,
        columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
    )

    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
    df.set_index('timestamp', inplace=True)
    df = df[~df.index.duplicated(keep='first')]
    df.sort_index(inplace=True)
    df.ffill(inplace=True)
    df.dropna(inplace=True)

    print(f"取得完了！レコード数: {len(df)} 件 ({df.index[0]} 〜 {df.index[-1]})")
    return df

import ta
from ta.momentum import RSIIndicator, StochasticOscillator
from ta.trend import MACD
from ta.volatility import BollingerBands

def add_technical_indicators(df: pd.DataFrame) -> pd.DataFrame:
    """
    OHLCVデータにテクニカル指標を追加する

    Args:
        df: OHLCVのDataFrame（open/high/low/close/volumeカラム必須）

    Returns:
        テクニカル指標追加後のDataFrame
    """
    df = df.copy()

    rsi = RSIIndicator(close=df['close'], window=14)
    df['rsi'] = rsi.rsi()

    macd = MACD(close=df['close'], window_slow=26, window_fast=12, window_sign=9)
    df['macd'] = macd.macd()
    df['macd_signal'] = macd.macd_signal()
    df['macd_diff'] = macd.macd_diff()

    bb = BollingerBands(close=df['close'], window=20, window_dev=2)
    df['bb_upper'] = bb.bollinger_hband()
    df['bb_lower'] = bb.bollinger_lband()
    df['bb_width'] = bb.bollinger_wband()
    df['bb_pband'] = bb.bollinger_pband()

    stoch_rsi = StochasticOscillator(
        high=df['high'], low=df['low'], close=df['close'],
        window=14, smooth_window=3
    )
    df['stoch_rsi'] = stoch_rsi.stoch()
    df['stoch_rsi_signal'] = stoch_rsi.stoch_signal()

    df.dropna(inplace=True)
    print(f"テクニカル指標追加完了！特徴量数: {len(df.columns)} カラム")
    return df

from sklearn.preprocessing import MinMaxScaler
from typing import Tuple

def normalize_and_split(
    df: pd.DataFrame,
    train_ratio: float = 0.7,
    val_ratio: float = 0.15,
) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, MinMaxScaler]:
    """
    時間軸に沿ってtrain/validation/testに分割し、正規化する

    MinMaxScaler: 値域が決まっている特徴量（RSI: 0-100など）に適している
    → 今回はOHLCV + テクニカル指標の混在なのでMinMaxを採用
    """
    n = len(df)
    train_end = int(n * train_ratio)
    val_end = int(n * (train_ratio + val_ratio))

    train_df = df.iloc[:train_end].copy()
    val_df   = df.iloc[train_end:val_end].copy()
    test_df  = df.iloc[val_end:].copy()

    feature_cols = [
        'open', 'high', 'low', 'close', 'volume',
        'rsi', 'macd', 'macd_signal', 'macd_diff',
        'bb_width', 'bb_pband', 'stoch_rsi', 'stoch_rsi_signal'
    ]

    scaler = MinMaxScaler(feature_range=(0, 1))
    train_df[feature_cols] = scaler.fit_transform(train_df[feature_cols])
    val_df[feature_cols]   = scaler.transform(val_df[feature_cols])
    test_df[feature_cols]  = scaler.transform(test_df[feature_cols])

    print(f"Train: {len(train_df)} 件 | Val: {len(val_df)} 件 | Test: {len(test_df)} 件")
    return train_df, val_df, test_df, scaler

from prophet import Prophet
import pandas as pd

def build_prophet_model(
    train_df: pd.DataFrame,
    regressor_cols: list = ['rsi', 'macd', 'macd_diff']
) -> Prophet:
    """
    Facebook Prophetモデルを構築・学習する

    Args:
        train_df: 学習用DataFrame（timestampインデックス・closeカラム必須）
        regressor_cols: 外部変数として追加するカラム名のリスト

    Returns:
        学習済みProphetモデル
    """
    prophet_df = train_df.reset_index().rename(
        columns={'timestamp': 'ds', 'close': 'y'}
    )

    model = Prophet(
        changepoint_prior_scale=0.05,
        seasonality_prior_scale=10.0,
        seasonality_mode='multiplicative',
        daily_seasonality=False,
        weekly_seasonality=True,
        yearly_seasonality=True,
    )

    # 仮想通貨特有の4年半減期サイクルを季節性として追加
    model.add_seasonality(
        name='halving_cycle',
        period=365.25 * 4,
        fourier_order=5
    )

    # RSI・MACDなどのテクニカル指標を外部変数（regressors）として追加
    for col in regressor_cols:
        model.add_regressor(col)

    model.fit(prophet_df[['ds', 'y'] + regressor_cols])
    return model

def predict_prophet(
    model: Prophet,
    future_df: pd.DataFrame,
    regressor_cols: list = ['rsi', 'macd', 'macd_diff']
) -> pd.DataFrame:
    """
    Prophetで将来予測を実行する

    Args:
        model: 学習済みProphetモデル
        future_df: 予測対象のDataFrame（dsカラムと各regressor必須）
        regressor_cols: 学習時と同じ外部変数カラム名

    Returns:
        予測結果DataFrame（yhat・yhat_lower・yhat_upper）
    """
    forecast = model.predict(future_df[['ds'] + regressor_cols])
    return forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']]

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import (
    Bidirectional, LSTM, Dense, Dropout, BatchNormalization
)
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau
import numpy as np

def create_sequences(
    df: pd.DataFrame,
    feature_cols: list,
    target_col: str = 'close',
    seq_len: int = 60
) -> tuple:
    """
    時系列DataFrameをLSTM用の3Dシーケンスに変換する

    Args:
        df: 正規化済みDataFrame
        feature_cols: 入力特徴量カラムのリスト
        target_col: 予測対象カラム
        seq_len: 過去何ステップを入力とするか

    Returns:
        X: shape (samples, seq_len, features)
        y: shape (samples,)
    """
    X, y = [], []
    data = df[feature_cols + [target_col]].values

    for i in range(seq_len, len(data)):
        X.append(data[i - seq_len:i, :-1])
        y.append(data[i, -1])

    return np.array(X), np.array(y)

def build_bilstm_model(
    input_shape: tuple,
    units_1: int = 128,
    units_2: int = 64,
    dropout_rate: float = 0.3,
    learning_rate: float = 0.001
) -> tf.keras.Model:
    """
    Bidirectional LSTMモデルを構築する

    Args:
        input_shape: (seq_len, n_features)
        units_1: 第1層のユニット数
        units_2: 第2層のユニット数
        dropout_rate: ドロップアウト率
        learning_rate: Adam最適化の学習率

    Returns:
        コンパイル済みKerasモデル
    """
    model = Sequential([
        Bidirectional(
            LSTM(units_1, return_sequences=True),
            input_shape=input_shape
        ),
        BatchNormalization(),
        Dropout(dropout_rate),

        Bidirectional(LSTM(units_2, return_sequences=False)),
        BatchNormalization(),
        Dropout(dropout_rate),

        Dense(32, activation='relu'),
        Dense(1)
    ])

    model.compile(
        optimizer=tf.keras.optimizers.Adam(learning_rate=learning_rate),
        loss='huber',
        metrics=['mae']
    )

    return model

def train_bilstm(
    model: tf.keras.Model,
    X_train: np.ndarray,
    y_train: np.ndarray,
    X_val: np.ndarray,
    y_val: np.ndarray,
    epochs: int = 100,
    batch_size: int = 32
) -> tf.keras.callbacks.History:
    """Bi-LSTMモデルを訓練する"""
    callbacks = [
        EarlyStopping(
            monitor='val_loss', patience=15, restore_best_weights=True
        ),
        ReduceLROnPlateau(
            monitor='val_loss', factor=0.5, patience=7, min_lr=1e-6
        )
    ]

    history = model.fit(
        X_train, y_train,
        validation_data=(X_val, y_val),
        epochs=epochs,
        batch_size=batch_size,
        callbacks=callbacks,
        verbose=1
    )

    return history

import optuna
optuna.logging.set_verbosity(optuna.logging.WARNING)

def objective(
    trial: optuna.Trial,
    X_train: np.ndarray,
    y_train: np.ndarray,
    X_val: np.ndarray,
    y_val: np.ndarray,
    input_shape: tuple
) -> float:
    """
    Optunaの目的関数：検証損失を最小化するハイパーパラメータを探索する
    """
    units_1 = trial.suggest_int('units_1', 64, 256, step=32)
    units_2 = trial.suggest_int('units_2', 32, 128, step=32)
    dropout_rate = trial.suggest_float('dropout_rate', 0.1, 0.5)
    learning_rate = trial.suggest_float('learning_rate', 1e-4, 1e-2, log=True)
    batch_size = trial.suggest_categorical('batch_size', [16, 32, 64])

    model = build_bilstm_model(
        input_shape=input_shape,
        units_1=units_1,
        units_2=units_2,
        dropout_rate=dropout_rate,
        learning_rate=learning_rate
    )

    history = train_bilstm(
        model, X_train, y_train, X_val, y_val,
        epochs=50, batch_size=batch_size
    )

    return min(history.history['val_loss'])

def run_optuna_optimization(
    X_train: np.ndarray,
    y_train: np.ndarray,
    X_val: np.ndarray,
    y_val: np.ndarray,
    input_shape: tuple,
    n_trials: int = 50
) -> dict:
    """
    Optunaでベイズ最適化を実行し、最適なハイパーパラメータを返す
    """
    study = optuna.create_study(
        direction='minimize',
        sampler=optuna.samplers.TPESampler(seed=42)
    )

    study.optimize(
        lambda trial: objective(
            trial, X_train, y_train, X_val, y_val, input_shape
        ),
        n_trials=n_trials,
        show_progress_bar=True
    )

    print(f"最適パラメータ: {study.best_params}")
    print(f"最良検証損失: {study.best_value:.6f}")

    return study.best_params

import numpy as np
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score

def weighted_ensemble(
    prophet_preds: np.ndarray,
    lstm_preds: np.ndarray,
    prophet_rmse: float,
    lstm_rmse: float
) -> np.ndarray:
    """
    RMSE逆数による加重平均アンサンブル

    RMSEが小さいモデルほど重みを大きく設定する
    """
    w_prophet = 1.0 / prophet_rmse
    w_lstm = 1.0 / lstm_rmse
    total_weight = w_prophet + w_lstm

    ensemble_preds = (
        w_prophet * prophet_preds + w_lstm * lstm_preds
    ) / total_weight

    return ensemble_preds

def evaluate_models(
    y_true: np.ndarray,
    prophet_preds: np.ndarray,
    lstm_preds: np.ndarray,
    ensemble_preds: np.ndarray
) -> dict:
    """
    各モデルのRMSE・MAE・R²を計算して比較する
    """
    results = {}
    for name, preds in [
        ('Prophet', prophet_preds),
        ('Bi-LSTM', lstm_preds),
        ('Ensemble', ensemble_preds)
    ]:
        results[name] = {
            'RMSE': np.sqrt(mean_squared_error(y_true, preds)),
            'MAE':  mean_absolute_error(y_true, preds),
            'R²':   r2_score(y_true, preds)
        }

    for name, metrics in results.items():
        print(f"[{name}] RMSE: {metrics['RMSE']:.4f} | "
              f"MAE: {metrics['MAE']:.4f} | R²: {metrics['R²']:.4f}")

    return results

from sklearn.linear_model import Ridge
import numpy as np

def stacking_ensemble(
    val_prophet_preds: np.ndarray,
    val_lstm_preds: np.ndarray,
    y_val: np.ndarray,
    test_prophet_preds: np.ndarray,
    test_lstm_preds: np.ndarray
) -> np.ndarray:
    """
    Ridgeメタモデルによるスタッキングアンサンブル

    検証データで2モデルの予測を「特徴量」としてメタモデルを学習し、
    テストデータの最終予測を生成する
    """
    X_val_meta  = np.column_stack([val_prophet_preds,  val_lstm_preds])
    X_test_meta = np.column_stack([test_prophet_preds, test_lstm_preds])

    meta_model = Ridge(alpha=1.0)
    meta_model.fit(X_val_meta, y_val)

    final_preds = meta_model.predict(X_test_meta)
    print(f"Stackingの係数: {meta_model.coef_}")
    return final_preds

from backtesting import Backtest, Strategy
import pandas as pd
import numpy as np

class EnsemblePredictionStrategy(Strategy):
    """
    アンサンブル予測に基づくロングオンリー戦略

    エントリー条件：予測価格が現在価格より threshold% 以上高い
    エグジット条件：予測価格が現在価格以下 or ストップロス発動
    """
    threshold     = 0.02   # エントリー閾値（2%）
    stop_loss_pct = 0.05   # ストップロス（5%）

    def init(self):
        self.pred = self.I(lambda: self.data.Predicted, name='Predicted')

    def next(self):
        current_price   = self.data.Close[-1]
        predicted_price = self.pred[-1]
        expected_return = (predicted_price - current_price) / current_price

        if not self.position:
            if expected_return > self.threshold:
                sl_price = current_price * (1 - self.stop_loss_pct)
                self.buy(sl=sl_price)
        else:
            if expected_return <= 0:
                self.position.close()

def run_backtest(
    df: pd.DataFrame,
    ensemble_preds: np.ndarray,
    initial_cash: float = 1_000_000,
    commission: float = 0.001
) -> object:
    """
    backtesting.pyでバックテストを実行する

    Args:
        df: OHLCVのDataFrame（インデックスはdatetime）
        ensemble_preds: アンサンブル予測値の配列
        initial_cash: 初期資金（円）
        commission: 手数料率（0.001 = 0.1%）

    Returns:
        backtesting.pyのstatsオブジェクト
    """
    bt_df = df[['open', 'high', 'low', 'close', 'volume']].copy()
    bt_df.columns = ['Open', 'High', 'Low', 'Close', 'Volume']
    bt_df['Predicted'] = ensemble_preds

    bt = Backtest(
        bt_df,
        EnsemblePredictionStrategy,
        cash=initial_cash,
        commission=commission,
        exclusive_orders=True
    )

    stats = bt.run()
    bt.plot(filename='backtest_result.html')

    print(stats)
    return stats