# requirements.txt
tensorflow==2.16.1
torch==2.3.1
pytorch-forecasting==1.1.1
lightgbm==4.4.0
xgboost==2.1.0
scikit-learn==1.5.1
optuna==3.6.1
shap==0.45.1
mlflow==2.14.1
fastapi==0.111.0
pandas==2.2.2
numpy==1.26.4

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

def calculate_metrics(y_true: np.ndarray, y_pred: np.ndarray) -> dict:
    """
    金融時系列予測のための包括的評価指標を計算する
    
    Returns:
        dict: RMSE, MAE, MAPE, Directional Accuracy を含む辞書
    """
    rmse = np.sqrt(mean_squared_error(y_true, y_pred))
    mae = mean_absolute_error(y_true, y_pred)
    
    # MAPE（ゼロ除算を避けるためのマスク処理）
    mask = y_true != 0
    mape = np.mean(np.abs((y_true[mask] - y_pred[mask]) / y_true[mask])) * 100
    
    # 方向一致率（Directional Accuracy）
    true_direction = np.sign(np.diff(y_true))
    pred_direction = np.sign(np.diff(y_pred))
    directional_accuracy = np.mean(true_direction == pred_direction) * 100
    
    return {
        "RMSE": round(rmse, 4),
        "MAE": round(mae, 4),
        "MAPE(%)": round(mape, 4),
        "Directional_Accuracy(%)": round(directional_accuracy, 2)
    }

from sklearn.model_selection import TimeSeriesSplit
import pandas as pd
from typing import List

def walk_forward_validation(
    model,
    X: pd.DataFrame,
    y: pd.Series,
    n_splits: int = 5,
    test_size: int = 720  # 30日分（1時間足）
) -> List[dict]:
    """
    ウォークフォワード検証の実装
    各フォールで訓練期間を拡大しながらテスト期間を評価する
    """
    tscv = TimeSeriesSplit(n_splits=n_splits, test_size=test_size)
    results = []
    
    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]
        
        # モデルの学習と予測
        model.fit(X_train, y_train)
        y_pred = model.predict(X_test)
        
        # 評価指標の計算
        metrics = calculate_metrics(y_test.values, y_pred)
        metrics["fold"] = fold + 1
        metrics["train_size"] = len(train_idx)
        metrics["test_size"] = len(test_idx)
        results.append(metrics)
        
        print(f"Fold {fold + 1}: RMSE={metrics['RMSE']:.4f}, "
              f"DA={metrics['Directional_Accuracy(%)']:.1f}%")
    
    return results

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

def build_lstm_model(
    lookback: int,
    n_features: int,
    units: list[int] = [128, 64],
    dropout_rate: float = 0.2,
    bidirectional: bool = True,
    learning_rate: float = 0.001
) -> tf.keras.Model:
    """
    双方向LSTMモデルの構築
    
    Args:
        lookback: シーケンス長（ルックバック期間）
        n_features: 特徴量数
        units: 各LSTM層のユニット数リスト
        dropout_rate: ドロップアウト率
        bidirectional: 双方向LSTMを使用するか
        learning_rate: 初期学習率
    
    Returns:
        コンパイル済みKerasモデル
    """
    model = Sequential()
    model.add(Input(shape=(lookback, n_features)))
    
    for i, unit in enumerate(units):
        return_sequences = (i < len(units) - 1)  # 最終層以外はTrue
        
        if bidirectional:
            model.add(Bidirectional(
                LSTM(unit, return_sequences=return_sequences)
            ))
        else:
            model.add(LSTM(unit, return_sequences=return_sequences))
        
        model.add(BatchNormalization())
        model.add(Dropout(dropout_rate))
    
    # 出力層（回帰なので線形活性化）
    model.add(Dense(32, activation='relu'))
    model.add(Dense(1, activation='linear'))
    
    model.compile(
        optimizer=Adam(learning_rate=learning_rate),
        loss='huber',  # MSEより外れ値に対して頑健
        metrics=['mae']
    )
    
    return model

def prepare_sequences(
    data: np.ndarray,
    lookback: int,
    forecast_horizon: int = 1
) -> tuple[np.ndarray, np.ndarray]:
    """
    時系列データをLSTM入力形式に変換する
    
    Returns:
        X: shape (samples, lookback, features)
        y: shape (samples,)
    """
    X, y = [], []
    for i in range(lookback, len(data) - forecast_horizon + 1):
        X.append(data[i - lookback:i])
        y.append(data[i + forecast_horizon - 1, 0])  # 0列目がターゲット（終値）
    return np.array(X), np.array(y)

# 学習の実行
LOOKBACK = 72   # 72時間（3日）ルックバック
BATCH_SIZE = 64
EPOCHS = 200

# データ準備（スケーリング済みデータを想定）
X_train, y_train = prepare_sequences(train_scaled, LOOKBACK)
X_val, y_val = prepare_sequences(val_scaled, LOOKBACK)

model = build_lstm_model(
    lookback=LOOKBACK,
    n_features=X_train.shape[2],
    units=[128, 64],
    bidirectional=True
)

callbacks = [
    EarlyStopping(
        monitor='val_loss',
        patience=20,
        restore_best_weights=True,
        verbose=1
    ),
    ReduceLROnPlateau(
        monitor='val_loss',
        factor=0.5,
        patience=10,
        min_lr=1e-6,
        verbose=1
    )
]

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

import optuna
from optuna.integration import TFKerasPruningCallback

def objective(trial: optuna.Trial) -> float:
    """Optunaの目的関数：検証損失を最小化する"""
    
    # ハイパーパラメータの探索空間定義
    n_layers = trial.suggest_int("n_layers", 1, 3)
    units = [
        trial.suggest_categorical(f"units_l{i}", [32, 64, 128, 256])
        for i in range(n_layers)
    ]
    dropout_rate = trial.suggest_float("dropout_rate", 0.1, 0.5)
    learning_rate = trial.suggest_float("learning_rate", 1e-4, 1e-2, log=True)
    lookback = trial.suggest_categorical("lookback", [24, 48, 72, 120, 168])
    bidirectional = trial.suggest_categorical("bidirectional", [True, False])
    
    # データ再準備（lookbackが変わるため）
    X_tr, y_tr = prepare_sequences(train_scaled, lookback)
    X_v, y_v = prepare_sequences(val_scaled, lookback)
    
    model = build_lstm_model(
        lookback=lookback,
        n_features=X_tr.shape[2],
        units=units,
        dropout_rate=dropout_rate,
        bidirectional=bidirectional,
        learning_rate=learning_rate
    )
    
    pruning_callback = TFKerasPruningCallback(trial, "val_loss")
    
    history = model.fit(
        X_tr, y_tr,
        validation_data=(X_v, y_v),
        epochs=100,
        batch_size=64,
        callbacks=[
            EarlyStopping(patience=15, restore_best_weights=True),
            pruning_callback
        ],
        verbose=0
    )
    
    return min(history.history["val_loss"])

# 最適化の実行（100トライアル）
study = optuna.create_study(
    direction="minimize",
    pruner=optuna.pruners.MedianPruner(n_startup_trials=10)
)
study.optimize(objective, n_trials=100, timeout=3600)

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

import matplotlib.pyplot as plt

def plot_training_history(history: dict, title: str = "LSTM Training History"):
    """学習曲線のプロット（損失・MAEの両軸）"""
    fig, axes = plt.subplots(1, 2, figsize=(14, 5))
    
    # 損失の推移
    axes[0].plot(history.history['loss'], label='Train Loss', color='steelblue')
    axes[0].plot(history.history['val_loss'], label='Val Loss', color='tomato')
    axes[0].set_title('Loss (Huber)')
    axes[0].set_xlabel('Epoch')
    axes[0].legend()
    axes[0].grid(alpha=0.3)
    
    # MAEの推移
    axes[1].plot(history.history['mae'], label='Train MAE', color='steelblue')
    axes[1].plot(history.history['val_mae'], label='Val MAE', color='tomato')
    axes[1].set_title('Mean Absolute Error')
    axes[1].set_xlabel('Epoch')
    axes[1].legend()
    axes[1].grid(alpha=0.3)
    
    plt.suptitle(title, fontsize=14, fontweight='bold')
    plt.tight_layout()
    plt.savefig('lstm_training_history.png', dpi=150, bbox_inches='tight')
    plt.show()

plot_training_history(history)

import pytorch_lightning as pl
from pytorch_forecasting import TemporalFusionTransformer, TimeSeriesDataSet
from pytorch_forecasting.data import GroupNormalizer
from pytorch_forecasting.metrics import QuantileLoss
import pandas as pd

def prepare_tft_dataset(df: pd.DataFrame) -> tuple:
    """
    TFT用データセットの準備
    
    Args:
        df: OHLCV + テクニカル指標を含むDataFrame（'time_idx'列が必要）
    
    Returns:
        訓練・検証DataSetのタプル
    """
    MAX_ENCODER_LENGTH = 168  # 7日間のエンコーダ長
    MAX_PREDICTION_LENGTH = 24  # 24時間先予測
    
    training_cutoff = df["time_idx"].max() - MAX_PREDICTION_LENGTH
    
    training = TimeSeriesDataSet(
        df[df["time_idx"] <= training_cutoff],
        time_idx="time_idx",
        target="close_log_return",  # 対数リターンをターゲットに
        group_ids=["symbol"],  # BTC, ETHなどの銘柄ID
        min_encoder_length=MAX_ENCODER_LENGTH // 2,
        max_encoder_length=MAX_ENCODER_LENGTH,
        min_prediction_length=1,
        max_prediction_length=MAX_PREDICTION_LENGTH,
        # 時変既知共変量（未来がわかる変数：曜日・時間帯など）
        time_varying_known_reals=["hour_of_day", "day_of_week"],
        # 時変未知共変量（予測時点では不明な変数：価格・出来高など）
        time_varying_unknown_reals=[
            "close_log_return", "volume_log",
            "rsi_14", "macd", "bb_width"
        ],
        target_normalizer=GroupNormalizer(
            groups=["symbol"],
            transformation="softplus"
        ),
        add_relative_time_idx=True,
        add_target_scales=True,
        add_encoder_length=True,
    )
    
    validation = TimeSeriesDataSet.from_dataset(
        training,
        df,
        predict=True,
        stop_randomization=True
    )
    
    return training, validation

# データローダー作成
training, validation = prepare_tft_dataset(df)

train_dataloader = training.to_dataloader(train=True, batch_size=64, num_workers=4)
val_dataloader = validation.to_dataloader(train=False, batch_size=64, num_workers=4)

# TFTモデルの定義
tft = TemporalFusionTransformer.from_dataset(
    training,
    learning_rate=3e-3,
    hidden_size=64,
    attention_head_size=4,
    dropout=0.1,
    hidden_continuous_size=32,
    output_size=7,  # 分位点予測（10%, 20%, ..., 90%）
    loss=QuantileLoss(),
    log_interval=10,
    reduce_on_plateau_patience=4,
)

# PyTorch Lightningトレーナー
trainer = pl.Trainer(
    max_epochs=50,
    accelerator="auto",
    enable_model_summary=True,
    gradient_clip_val=0.1,
    callbacks=[
        pl.callbacks.EarlyStopping(
            monitor="val_loss",
            patience=10,
            mode="min"
        )
    ],
)

trainer.fit(tft, train_dataloader, val_dataloader)

from neuralforecast import NeuralForecast
from neuralforecast.models import PatchTST, iTransformer

# neuralforecastを使った最新モデルの利用
nf = NeuralForecast(
    models=[
        PatchTST(
            h=24,             # 予測ホライズン（24時間先）
            input_size=168,   # 入力シーケンス長
            patch_len=16,     # パッチサイズ
            stride=8,
            max_steps=500,
            val_check_steps=50,
        ),
        iTransformer(
            h=24,
            input_size=168,
            n_series=2,  # BTC + ETH
            max_steps=500,
        ),
    ],
    freq="H"
)

nf.fit(df_train)
forecasts = nf.predict()

# PyTorchでのMixed Precision設定
from torch.cuda.amp import GradScaler, autocast

scaler = GradScaler()

for batch in train_dataloader:
    optimizer.zero_grad()
    with autocast():
        loss = model(batch)
    scaler.scale(loss).backward()
    scaler.step(optimizer)
    scaler.update()

import pandas as pd
import numpy as np
import talib  # TA-Libライブラリ（別途インストール要）

def create_lag_features(
    df: pd.DataFrame,
    target_col: str = "close",
    lag_periods: list[int] = [1, 2, 3, 6, 12, 24, 48, 72]
) -> pd.DataFrame:
    """
    ラグ特徴量とローリング統計量の生成
    
    Note: 必ず訓練データで計算した後、テストデータに適用すること（リーケージ防止）
    """
    df = df.copy()
    
    # ラグ特徴量（過去N時間前の値）
    for lag in lag_periods:
        df[f"{target_col}_lag_{lag}h"] = df[target_col].shift(lag)
        df[f"return_lag_{lag}h"] = df[target_col].pct_change(lag)
    
    # ローリング統計量
    for window in [6, 12, 24, 48, 168]:
        df[f"rolling_mean_{window}h"] = df[target_col].rolling(window).mean()
        df[f"rolling_std_{window}h"] = df[target_col].rolling(window).std()
        df[f"rolling_max_{window}h"] = df[target_col].rolling(window).max()
        df[f"rolling_min_{window}h"] = df[target_col].rolling(window).min()
        # ローリング歪度・尖度（分布形状の特徴）
        df[f"rolling_skew_{window}h"] = df[target_col].rolling(window).skew()
        df[f"rolling_kurt_{window}h"] = df[target_col].rolling(window).kurt()
    
    # テクニカル指標
    close = df["close"].values
    high = df["high"].values
    low = df["low"].values
    volume = df["volume"].values
    
    # RSI（複数期間）
    for period in [9, 14, 21]:
        df[f"rsi_{period}"] = talib.RSI(close, timeperiod=period)
    
    # MACD
    macd, macd_signal, macd_hist = talib.MACD(close)
    df["macd"] = macd
    df["macd_signal"] = macd_signal
    df["macd_hist"] = macd_hist
    
    # Bollinger Bands
    upper, middle, lower = talib.BBANDS(close, timeperiod=20)
    df["bb_upper"] = upper
    df["bb_middle"] = middle
    df["bb_lower"] = lower
    df["bb_width"] = (upper - lower) / middle  # バンド幅（ボラティリティ指標）
    df["bb_position"] = (close - lower) / (upper - lower)  # バンド内位置
    
    # ATR（Average True Range：ボラティリティ）
    df["atr_14"] = talib.ATR(high, low, close, timeperiod=14)
    
    # 出来高関連
    df["volume_ma_24h"] = df["volume"].rolling(24).mean()
    df["volume_ratio"] = df["volume"] / df["volume_ma_24h"]  # 出来高比率
    df["obv"] = talib.OBV(close, volume)  # On-Balance Volume
    
    # 時間的特徴量
    df["hour"] = df.index.hour
    df["day_of_week"] = df.index.dayofweek
    df["is_weekend"] = df["day_of_week"].isin([5, 6]).astype(int)
    
    # 三角関数による周期特徴量（周期性を線形モデルで扱えるようにする）
    df["hour_sin"] = np.sin(2 * np.pi * df["hour"] / 24)
    df["hour_cos"] = np.cos(2 * np.pi * df["hour"] / 24)
    df["dow_sin"] = np.sin(2 * np.pi * df["day_of_week"] / 7)
    df["dow_cos"] = np.cos(2 * np.pi * df["day_of_week"] / 7)
    
    return df.dropna()  # NaNを含む行を除去

df_features = create_lag_features(df_ohlcv)

import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit
import shap

def train_lightgbm(
    X_train: pd.DataFrame,
    y_train: pd.Series,
    X_val: pd.DataFrame,
    y_val: pd.Series,
    params: dict = None
) -> lgb.Booster:
    """LightGBMの学習パイプライン"""
    
    if params is None:
        params = {
            "objective": "regression",
            "metric": "rmse",
            "verbosity": -1,
            "boosting_type": "gbdt",
            "num_leaves": 63,
            "learning_rate": 0.05,
            "feature_fraction": 0.8,
            "bagging_fraction": 0.8,
            "bagging_freq": 5,
            "min_child_samples": 20,
            "lambda_l1": 0.1,
            "lambda_l2": 0.1,
            "n_jobs": -1,
            "seed": 42
        }
    
    lgb_train = lgb.Dataset(X_train, label=y_train)
    lgb_val = lgb.Dataset(X_val, label=y_val, reference=lgb_train)
    
    callbacks = [
        lgb.early_stopping(stopping_rounds=50, verbose=True),
        lgb.log_evaluation(period=100)
    ]
    
    model = lgb.train(
        params,
        lgb_train,
        num_boost_round=2000,
        valid_sets=[lgb_train, lgb_val],
        valid_names=["train", "valid"],
        callbacks=callbacks
    )
    
    return model

# ターゲット：24時間後の対数リターン
y = np.log(df_features["close"].shift(-24) / df_features["close"]).dropna()
feature_cols = [c for c in df_features.columns if c not in
                ["open", "high", "low", "close", "volume"]]
X = df_features[feature_cols].iloc[:-24]

# Time Series Split
tscv = TimeSeriesSplit(n_splits=5, test_size=720)
fold_models = []

for fold, (train_idx, val_idx) in enumerate(tscv.split(X)):
    model = train_lightgbm(
        X.iloc[train_idx], y.iloc[train_idx],
        X.iloc[val_idx], y.iloc[val_idx]
    )
    fold_models.append(model)
    print(f"Fold {fold + 1} 完了: Best iteration = {model.best_iteration}")

# SHAP値による特徴量重要度の解釈
explainer = shap.TreeExplainer(fold_models[-1])
shap_values = explainer.shap_values(X.iloc[-1000:])

shap.summary_plot(
    shap_values,
    X.iloc[-1000:],
    plot_type="bar",
    max_display=20
)

import pandas as pd
import numpy as np

def simple_backtest(
    y_pred: np.ndarray,
    actual_prices: np.ndarray,
    threshold: float = 0.005,  # 0.5%以上の予測リターンのみ取引
    fee_rate: float = 0.001    # 取引手数料 0.1%（Binance Maker）
) -> dict:
    """
    シンプルなロング/ショート戦略のバックテスト
    
    Args:
        y_pred: 予測対数リターン
        actual_prices: 実際の価格系列
        threshold: 取引閾値（小さすぎると手数料負け）
        fee_rate: 片道の取引手数料率
    
    Returns:
        総リターン・シャープレシオ・最大ドローダウンなど
    """
    positions = np.where(y_pred > threshold, 1,      # 予測上昇→ロング
                np.where(y_pred < -threshold, -1, 0)) # 予測下落→ショート
    
    # 実際の価格リターン
    actual_returns = np.diff(np.log(actual_prices))
    
    # ポジション調整（前日のポジションに対して今日のリターンが発生）
    strategy_returns = positions[:-1] * actual_returns
    
    # 手数料控除（ポジションが変化した場合のみ）
    position_changes = np.diff(positions) != 0
    transaction_costs = position_changes * fee_rate * 2  # 往復
    net_returns = strategy_returns - transaction_costs[:-1]
    
    # パフォーマンス指標の計算
    total_return = np.exp(np.sum(net_returns)) - 1
    annualized_return = np.exp(np.mean(net_returns) * 8760) - 1  # 年率換算（時間足）
    annualized_vol = np.std(net_returns) * np.sqrt(8760)
    sharpe_ratio = annualized_return / annualized_vol if annualized_vol > 0 else 0
    
    # 最大ドローダウン
    cumulative = np.exp(np.cumsum(net_returns))
    rolling_max = np.maximum.accumulate(cumulative)
    drawdown = (cumulative - rolling_max) / rolling_max
    max_drawdown = np.min(drawdown)
    
    # 勝率
    win_rate = np.mean(net_returns[positions[:-1] != 0] > 0) * 100
    
    return {
        "総リターン(%)": round(total_return * 100, 2),
        "年率リターン(%)": round(annualized_return * 100, 2),
        "シャープレシオ": round(sharpe_ratio, 3),
        "最大ドローダウン(%)": round(max_drawdown * 100, 2),
        "勝率(%)": round(win_rate, 1),
        "取引回数": int(np.sum(position_changes))
    }

# バックテスト実行
backtest_results = simple_backtest(y_pred_lgbm, test_prices)
print(pd.Series(backtest_results))

import numpy as np
import pandas as pd
from sklearn.linear_model import Ridge
from sklearn.preprocessing import StandardScaler

class StackingEnsemble:
    """
    LSTMとLightGBMのスタッキングアンサンブル
    メタモデルにRidge回帰を使用（過学習防止）
    """
    
    def __init__(self, base_models: dict, meta_model=None):
        """
        Args:
            base_models: {"model_name": model_instance} の辞書
            meta_model: メタモデル（デフォルトはRidge）
        """
        self.base_models = base_models
        self.meta_model = meta_model or Ridge(alpha=1.0)
        self.scaler = StandardScaler()
        
    def fit(self, X_train_dict: dict, y_train: np.ndarray,
            X_val_dict: dict, y_val: np.ndarray) -> None:
        """
        ベースモデルを学習し、その予測値でメタモデルを学習する
        
        Args:
            X_train_dict: {"model_name": X_train_array} の辞書
            y_train: 訓練ターゲット
            X_val_dict: {"model_name": X_val_array} の辞書
            y_val: 検証ターゲット（メタモデルの学習に使用）
        """
        # Step 1: ベースモデルの学習
        val_predictions = {}
        for name, model in self.base_models.items():
            print(f"学習中: {name}")
            model.fit(X_train_dict[name], y_train)
            val_pred = model.predict(X_val_dict[name])
            val_predictions[name] = val_pred
        
        # Step 2: 検証データの予測値でメタモデルを学習
        meta_features = np.column_stack(list(val_predictions.values()))
        meta_features_scaled = self.scaler.fit_transform(meta_features)
        self.meta_model.fit(meta_features_scaled, y_val)
        
        print(f"メタモデル係数: {dict(zip(self.base_models.keys(), self.meta_model.coef_))}")
    
    def predict(self, X_test_dict: dict) -> np.ndarray:
        """スタッキングアンサンブルによる予測"""
        test_predictions = {}
        for name, model in self.base_models.items():
            test_predictions[name] = model.predict(X_test_dict[name])
        
        meta_features = np.column_stack(list(test_predictions.values()))
        meta_features_scaled = self.scaler.transform(meta_features)
        return self.meta_model.predict(meta_features_scaled)


# 使用例
ensemble = StackingEnsemble(
    base_models={
        "lstm": lstm_model_wrapper,    # KerasモデルをSklearn互換ラッパーで包む
        "lightgbm": lgb_model,
        "xgboost": xgb_model,
    }
)

ensemble.fit(
    X_train_dict={"lstm": X_train_seq, "lightgbm": X_train_tab, "xgboost": X_train_tab},
    y_train=y_train,
    X_val_dict={"lstm": X_val_seq, "lightgbm": X_val_tab, "xgboost": X_val_tab},
    y_val=y_val
)

y_pred_ensemble = ensemble.predict(
    {"lstm": X_test_seq, "lightgbm": X_test_tab, "xgboost": X_test_tab}
)

from hmmlearn import hmm
import numpy as np

def detect_market_regime(returns: np.ndarray, n_states: int = 3) -> np.ndarray:
    """
    隠れマルコフモデルによる相場レジーム検出
    
    Returns:
        各時点のレジームラベル（0: 低ボラ横ばい, 1: トレンド, 2: 高ボラ不安定）
    """
    # 特徴量：リターンとボラティリティ
    features = np.column_stack([
        returns,
        np.abs(returns),  # 実現ボラティリティの代理
        np.where(returns > 0, returns, 0),   # 上昇幅
        np.where(returns < 0, -returns, 0),  # 下落幅
    ])
    
    model = hmm.GaussianHMM(
        n_components=n_states,
        covariance_type="full",
        n_iter=1000,
        random_state=42
    )
    model.fit(features)
    states = model.predict(features)
    
    return states

# レジームに応じたモデル重みの動的調整
def dynamic_ensemble_predict(
    predictions: dict,  # {"lstm": pred_array, "lgbm": pred_array, ...}
    current_regime: int,
    regime_weights: dict  # {regime_id: {"lstm": w1, "lgbm": w2, ...}}
) -> np.ndarray:
    """レジームに応じた動的加重アンサンブル"""
    weights = regime_weights[current_regime]
    weighted_pred = sum(
        predictions[name] * weight
        for name, weight in weights.items()
    )
    return weighted_pred

# レジームごとの推奨重み（バックテストで最適化）
regime_weights = {
    0: {"lstm": 0.3, "lgbm": 0.5, "tft": 0.2},  # 横ばい：GBM重視
    1: {"lstm": 0.4, "lgbm": 0.3, "tft": 0.3},  # トレンド：LSTM重視
    2: {"lstm": 0.2, "lgbm": 0.3, "tft": 0.5},  # 高ボラ：TFT重視（解釈性）
}

import mlflow
import mlflow.sklearn
import mlflow.keras

# MLflowによる実験管理
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("crypto-price-prediction")

with mlflow.start_run(run_name="lightgbm_v2.1"):
    # パラメータのロギング
    mlflow.log_params(lgb_params)
    
    # 評価指標のロギング
    mlflow.log_metrics(backtest_results)
    
    # モデルの保存
    mlflow.lightgbm.log_model(
        lgb_model=model,
        artifact_path="model",
        registered_model_name="crypto-lgbm-predictor"
    )
    
    # カスタムアーティファクトの保存（特徴量リストなど）
    mlflow.log_artifact("feature_columns.json")
    mlflow.log_artifact("scaler.pkl")

# ONNX形式への変換（推論高速化）
import onnxmltools
from onnxmltools.convert.common.data_types import FloatTensorType

onnx_model = onnxmltools.convert_lightgbm(
    model,
    initial_types=[("input", FloatTensorType([None, len(feature_cols)]))]
)
onnxmltools.utils.save_model(onnx_model, "model.onnx")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np
import onnxruntime as ort
import redis
import json

app = FastAPI(title="Crypto Price Predictor API")

# ONNX Runtimeで推論高速化
ort_session = ort.InferenceSession("model.onnx")
redis_client = redis.Redis(host="localhost", port=6379, db=0)

class PredictionRequest(BaseModel):
    symbol: str  # "BTC/USDT"
    features: list[float]

class PredictionResponse(BaseModel):
    symbol: str
    predicted_return_24h: float
    confidence: float
    timestamp: str

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    """24時間先の価格リターンを予測するエンドポイント"""
    
    # Redisキャッシュ確認（同一特徴量の場合はキャッシュから返却）
    cache_key = f"pred:{request.symbol}:{hash(str(request.features))}"
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # 入力バリデーション
    if len(request.features) != len(feature_cols):
        raise HTTPException(
            status_code=400,
            detail=f"特徴量数が不正です。期待値: {len(feature_cols)}"
        )
    
    # ONNX推論
    input_array = np.array([request.features], dtype=np.float32)
    ort_inputs = {ort_session.get_inputs()[0].name: input_array}
    prediction = ort_session.run(None, ort_inputs)[0][0]
    
    result = {
        "symbol": request.symbol,
        "predicted_return_24h": float(prediction),
        "confidence": calculate_confidence(prediction),
        "timestamp": pd.Timestamp.now().isoformat()
    }
    
    # Redisキャッシュに保存（60秒TTL）
    redis_client.setex(cache_key, 60, json.dumps(result))
    
    return result

def calculate_confidence(prediction: float) -> float:
    """予測値の絶対値をシグモイド変換して信頼度として返す"""
    return float(1 / (1 + np.exp(-abs(prediction) * 100)))

import pandas as pd
from scipy import stats

class ModelDriftDetector:
    """
    予測残差を監視してモデルドリフトを検出するクラス
    PSI（Population Stability Index）とKS検定を使用
    """
    
    def __init__(self, baseline_residuals: np.ndarray, threshold_psi: float = 0.2):
        self.baseline_residuals = baseline_residuals
        self.threshold_psi = threshold_psi
        self.alert_triggered = False
    
    def calculate_psi(self, current_residuals: np.ndarray, n_bins: int = 10) -> float:
        """PSI（分布シフトの定量化）を計算"""
        baseline_counts, bins = np.histogram(self.baseline_residuals, bins=n_bins)
        current_counts, _ = np.histogram(current_residuals, bins=bins)
        
        # ゼロ除算防止
        baseline_pct = (baseline_counts + 0.0001) / len(self.baseline_residuals)
        current_pct = (current_counts + 0.0001) / len(current_residuals)
        
        psi = np.sum((current_pct - baseline_pct) * np.log(current_pct / baseline_pct))
        return psi
    
    def check_drift(self, current_residuals: np.ndarray) -> dict:
        """ドリフトの確認とアラート判定"""
        psi = self.calculate_psi(current_residuals)
        ks_stat, ks_pvalue = stats.ks_2samp(
            self.baseline_residuals, current_residuals
        )
        
        drift_detected = (psi > self.threshold_psi) or (ks_pvalue < 0.05)
        
        result = {
            "psi": round(psi, 4),
            "ks_statistic": round(ks_stat, 4),
            "ks_pvalue": round(ks_pvalue, 4),
            "drift_detected": drift_detected,
            "recommendation": "再学習を推奨します" if drift_detected else "正常範囲内"
        }
        
        if drift_detected and not self.alert_triggered:
            self.trigger_retraining_alert(result)
            self.alert_triggered = True
        
        return result
    
    def trigger_retraining_alert(self, drift_info: dict) -> None:
        """再学習アラートの送信（Slack Webhook等と連携）"""
        print(f"⚠️ モデルドリフト検出: PSI={drift_info['psi']:.4f}")
        # Slack/PagerDuty等への通知実装

from arch import arch_model

def fit_garch_volatility(returns: pd.Series) -> np.ndarray:
    """
    GARCH(1,1)モデルによる条件付きボラティリティ推定
    予測残差の調整やリスク管理に活用する
    """
    model = arch_model(
        returns * 100,  # パーセント表示に変換
        vol="Garch",
        p=1, q=1,
        dist="skewt"  # 歪t分布（仮想通貨の太い裾野に対応）
    )
    result = model.fit(disp="off")
    
    # 条件付きボラティリティ（リスク管理用）
    conditional_vol = result.conditional_volatility / 100
    
    return conditional_vol

# 予測信頼区間の動的調整
def adjust_prediction_with_volatility(
    y_pred: float,
    current_vol: float,
    z_score: float = 1.96  # 95%信頼区間
) -> tuple[float, float]:
    """ボラティリティに基づいた予測の信頼区間を返す"""
    lower = y_pred - z_score * current_vol
    upper = y_pred + z_score * current_vol
    return lower, upper