import pandas as pd
import numpy as np

def create_features_no_lookahead(df: pd.DataFrame) -> pd.DataFrame:
    """
    先読みバイアスを排除した特徴量生成。
    すべての特徴量は予測時点では既知の値のみを使用する。
    """
    df = df.copy()

    # shift(1)でバーが確定した後の値のみ使用
    df['ma_20'] = df['close'].shift(1).rolling(20).mean()
    df['ma_50'] = df['close'].shift(1).rolling(50).mean()

    close_shifted = df['close'].shift(1)
    delta = close_shifted.diff()
    gain = delta.clip(lower=0)
    loss = -delta.clip(upper=0)
    avg_gain = gain.rolling(14).mean()
    avg_loss = loss.rolling(14).mean()
    rs = avg_gain / avg_loss.replace(0, 1e-10)
    df['rsi_14'] = 100 - (100 / (1 + rs))

    rolling_std = close_shifted.rolling(20).std()
    df['bb_upper'] = df['ma_20'] + 2 * rolling_std
    df['bb_lower'] = df['ma_20'] - 2 * rolling_std
    bb_range = (df['bb_upper'] - df['bb_lower']).replace(0, 1e-10)
    df['bb_position'] = (close_shifted - df['bb_lower']) / bb_range

    df['return_1']  = close_shifted.pct_change(1)
    df['return_5']  = close_shifted.pct_change(5)
    df['return_24'] = close_shifted.pct_change(24)

    df['target'] = df['close'].pct_change(1).shift(-1)

    return df.dropna()

import pandas as pd
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import mean_squared_error
import numpy as np

def walk_forward_validation(model, X: pd.DataFrame, y: pd.Series,
                             n_splits: int = 5) -> dict:
    """
    時系列データの正しいクロス検証。
    TimeSeriesSplitで未来データの混入を防ぐ。
    """
    tscv = TimeSeriesSplit(n_splits=n_splits)
    is_scores  = []
    oos_scores = []

    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)

        is_pred  = model.predict(X_train)
        is_rmse  = np.sqrt(mean_squared_error(y_train, is_pred))
        is_scores.append(is_rmse)

        oos_pred = model.predict(X_test)
        oos_rmse = np.sqrt(mean_squared_error(y_test, oos_pred))
        oos_scores.append(oos_rmse)

        print(f'Fold {fold+1}: IS RMSE={is_rmse:.4f}, OOS RMSE={oos_rmse:.4f}')

    return {
        'is_mean':       np.mean(is_scores),
        'oos_mean':      np.mean(oos_scores),
        'overfit_ratio': np.mean(oos_scores) / np.mean(is_scores),
        'is_scores':     is_scores,
        'oos_scores':    oos_scores
    }

時間軸 ─────────────────────────────────→

ウィンドウ1: [────IS期間(学習)────][OOS]
ウィンドウ2:      [────IS期間(学習)────][OOS]
ウィンドウ3:           [────IS期間(学習)────][OOS]
ウィンドウ4:                [────IS期間(学習)────][OOS]

IS = インサンプル（最適化に使う）
OOS = アウトオブサンプル（検証のみ、パラメータ調整に使わない）

WFE = OOS成績 ÷ IS成績

import pandas as pd
import numpy as np
import xgboost as xgb
from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class WFOConfig:
    lookback_period:    int = 252
    retrain_frequency:  int = 30
    prediction_horizon: int = 1

@dataclass
class WFOResult:
    is_return:  float
    oos_return: float
    is_sharpe:  float
    oos_sharpe: float
    is_max_dd:  float
    oos_max_dd: float
    wfe:        float
    n_trades:   int

def compute_sharpe(returns: np.ndarray, freq: int = 252) -> float:
    if len(returns) < 2:
        return 0.0
    mean = np.mean(returns) * freq
    std  = np.std(returns, ddof=1) * np.sqrt(freq)
    return mean / std if std != 0 else 0.0

def compute_max_drawdown(returns: np.ndarray) -> float:
    cumulative = np.cumprod(1 + returns)
    peak = np.maximum.accumulate(cumulative)
    drawdown = (cumulative - peak) / peak
    return float(np.min(drawdown))

def run_walk_forward_optimization(
    df: pd.DataFrame,
    feature_cols: List[str],
    target_col: str,
    config: WFOConfig
) -> Tuple[List[WFOResult], pd.Series]:
    """
    ウォークフォワード最適化のメインループ。
    非定常な時系列変化に追従するため、窓をスライドさせながら再学習する。
    """
    results: List[WFOResult] = []
    all_oos_returns = []
    total_len = len(df)
    start_idx = config.lookback_period

    while start_idx + config.retrain_frequency <= total_len:
        is_end   = start_idx
        is_start = max(0, is_end - config.lookback_period)
        is_data  = df.iloc[is_start:is_end]

        oos_start = is_end
        oos_end   = min(total_len, oos_start + config.retrain_frequency)
        oos_data  = df.iloc[oos_start:oos_end]

        if len(is_data) < 50 or len(oos_data) < 5:
            start_idx += config.retrain_frequency
            continue

        X_is  = is_data[feature_cols]
        y_is  = is_data[target_col]
        X_oos = oos_data[feature_cols]
        y_oos = oos_data[target_col]

        model = xgb.XGBRegressor(
            n_estimators=100, max_depth=4,
            learning_rate=0.05, subsample=0.8,
            random_state=42, verbosity=0
        )
        model.fit(X_is, y_is)

        is_preds   = model.predict(X_is)
        is_signals = np.sign(is_preds)
        is_rets    = is_signals * y_is.values
        is_sharpe  = compute_sharpe(is_rets)
        is_mdd     = compute_max_drawdown(is_rets)
        is_ret     = float(np.sum(is_rets))

        oos_preds   = model.predict(X_oos)
        oos_signals = np.sign(oos_preds)
        oos_rets    = oos_signals * y_oos.values
        oos_sharpe  = compute_sharpe(oos_rets)
        oos_mdd     = compute_max_drawdown(oos_rets)
        oos_ret     = float(np.sum(oos_rets))

        wfe = (oos_sharpe / is_sharpe) if is_sharpe != 0 else 0.0

        results.append(WFOResult(
            is_return=is_ret, oos_return=oos_ret,
            is_sharpe=is_sharpe, oos_sharpe=oos_sharpe,
            is_max_dd=is_mdd, oos_max_dd=oos_mdd,
            wfe=wfe, n_trades=len(oos_data)
        ))
        all_oos_returns.extend(oos_rets.tolist())
        start_idx += config.retrain_frequency

    return results, pd.Series(all_oos_returns)

┌─────────────────────────────────────────────┐
│                  FreqAI                      │
│                                              │
│  ┌──────────┐    ┌──────────┐    ┌────────┐ │
│  │ Feature   │ → │  Train   │ → │ Model  │ │
│  │Engineering│    │(XGBoost)│    │ Store  │ │
│  └──────────┘    └──────────┘    └────┬───┘ │
│                                        │     │
│  ┌──────────┐    ┌──────────┐    ┌────▼───┐ │
│  │Live OHLCV│ → │ Predict  │ ← │Retrain │ │
│  │  Data    │    │ (signal) │    │Trigger │ │
│  └──────────┘    └────┬─────┘    └────────┘ │
│                        │                     │
└────────────────────────┼─────────────────────┘
                         ↓
              Freqtrade Bot (売買執行)

{
  "freqai": {
    "enabled": true,
    "purge_old_models": true,
    "train_period_days": 60,
    "backtest_period_days": 7,
    "live_retrain_hours": 24,
    "identifier": "btc_xgb_v1",
    "feature_parameters": {
      "include_timeframes": ["1h", "4h", "1d"],
      "include_corr_pairlist": ["BTC/USDT", "ETH/USDT"],
      "label_period_candles": 24,
      "include_shifted_candles": 3,
      "DI_threshold": 0.9,
      "weight_factor": 0.9,
      "indicator_periods_candles": [10, 20, 50]
    },
    "data_split_parameters": {
      "test_size": 0.15,
      "random_state": 42,
      "shuffle": false
    },
    "model_training_parameters": {
      "n_estimators": 200,
      "max_depth": 6,
      "learning_rate": 0.05,
      "subsample": 0.8,
      "colsample_bytree": 0.8
    }
  }
}

from freqtrade.strategy import IStrategy
import pandas as pd
import talib.abstract as ta

class FreqAIXGBStrategy(IStrategy):
    """
    FreqAI + XGBoostによる仮想通貨自動売買戦略。
    24時間ごとに自動再学習し、モデル劣化に対応する。
    """
    minimal_roi      = {"0": 0.05}
    stoploss         = -0.03
    trailing_stop    = True
    trailing_stop_positive = 0.01
    timeframe        = '1h'
    can_short        = True

    def feature_engineering_expand_all(self, dataframe, period, **kwargs):
        dataframe["%-rsi"] = ta.RSI(dataframe, timeperiod=period)
        dataframe["%-mfi"] = ta.MFI(dataframe, timeperiod=period)
        dataframe["%-adx"] = ta.ADX(dataframe, timeperiod=period)
        dataframe["%-cci"] = ta.CCI(dataframe, timeperiod=period)
        return dataframe

    def feature_engineering_standard(self, dataframe, **kwargs):
        dataframe["%-day_of_week"]       = dataframe["date"].dt.dayofweek
        dataframe["%-hour_of_day"]       = dataframe["date"].dt.hour
        dataframe["%-volume_mean_ratio"] = (
            dataframe["volume"] / dataframe["volume"].rolling(24).mean()
        )
        return dataframe

    def set_freqai_targets(self, dataframe, **kwargs):
        dataframe["&-return_24h"] = (
            dataframe["close"].shift(-24) / dataframe["close"] - 1
        )
        return dataframe

    def populate_entry_trend(self, df, metadata):
        df.loc[
            (df["freqai.target_mean"] > 0.01) &
            (df["freqai.DI_values"] < df["freqai.DI_cutoff"]),
            "enter_long"
        ] = 1
        df.loc[
            (df["freqai.target_mean"] < -0.01) &
            (df["freqai.DI_values"] < df["freqai.DI_cutoff"]),
            "enter_short"
        ] = 1
        return df

    def populate_exit_trend(self, df, metadata):
        df.loc[df["freqai.target_mean"] < 0, "exit_long"]  = 1
        df.loc[df["freqai.target_mean"] > 0, "exit_short"] = 1
        return df

docker compose run --rm freqtrade backtesting \
  --strategy FreqAIXGBStrategy \
  --config config.json \
  --timerange 20240101-20241231 \
  --freqaimodel XGBoostRegressor

docker compose run -d --name freqtrade_live freqtrade trade \
  --strategy FreqAIXGBStrategy \
  --config config.json \
  --freqaimodel XGBoostRegressor

docker logs -f freqtrade_live

import numpy as np
from collections import deque
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class PerformanceMonitor:
    """
    リアルタイムでモデルのパフォーマンスを監視し、
    劣化を検出したら再学習トリガーを発火させる。
    """
    window_size:       int   = 100
    sharpe_threshold:  float = 0.5
    drawdown_limit:    float = -0.15
    recent_returns:    deque = field(default_factory=lambda: deque(maxlen=100))
    baseline_sharpe:   Optional[float] = None
    is_trading_halted: bool  = False

    def update(self, trade_return: float) -> dict:
        self.recent_returns.append(trade_return)
        if len(self.recent_returns) < 30:
            return {"status": "warming_up", "action": "none"}

        rets           = np.array(self.recent_returns)
        current_sharpe = self._compute_sharpe(rets)
        current_mdd    = self._compute_max_drawdown(rets)

        if self.baseline_sharpe is None and len(self.recent_returns) >= 100:
            self.baseline_sharpe = current_sharpe

        action = "continue"
        status = "healthy"

        if current_mdd < self.drawdown_limit:
            action = "halt_trading"
            status = "critical"
            self.is_trading_halted = True
        elif (self.baseline_sharpe is not None and
              current_sharpe < self.baseline_sharpe * 0.5):
            action = "trigger_retrain"
            status = "degraded"
        elif current_sharpe < self.sharpe_threshold:
            action = "increase_monitoring"
            status = "warning"

        return {
            "status":         status,
            "action":         action,
            "current_sharpe": round(current_sharpe, 3),
            "current_mdd":    round(current_mdd, 3),
            "n_trades":       len(self.recent_returns)
        }

    def _compute_sharpe(self, returns: np.ndarray) -> float:
        if np.std(returns) == 0:
            return 0.0
        return (np.mean(returns) / np.std(returns)) * np.sqrt(252)

    def _compute_max_drawdown(self, returns: np.ndarray) -> float:
        cumulative = np.cumprod(1 + returns)
        peak       = np.maximum.accumulate(cumulative)
        return float(np.min((cumulative - peak) / peak))

def kelly_position_size(
    win_rate:           float,
    avg_win:            float,
    avg_loss:           float,
    max_kelly_fraction: float = 0.25
) -> float:
    """
    ケリー基準によるポジションサイズ計算。
    フルケリーは破産リスクが高いため、最大25%にキャップする。
    """
    if avg_loss == 0:
        return 0.0
    b      = avg_win / abs(avg_loss)
    q      = 1 - win_rate
    kelly  = (b * win_rate - q) / b
    return min(max(kelly * 0.5, 0.0), max_kelly_fraction)