def create_features_no_lookahead(df: pd.DataFrame) -> pd.DataFrame:
    """先読みバイアスを排除した特徴量生成"""
    df = df.copy()
    # 確定済みバーの終値のみを使用
    close_shifted = df['close'].shift(1)

    df['ma_20'] = close_shifted.rolling(20).mean()
    df['ma_50'] = close_shifted.rolling(50).mean()

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

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

    # ターゲットのみ未来方向にshift
    df['target'] = df['close'].pct_change(1).shift(-1)

    return df.dropna()

Python 3.11+
pandas / NumPy          ── データ処理
TensorFlow / PyTorch    ── モデル訓練
CCXT                    ── 取引所API統合
VectorBT                ── 高速バックテスト
Freqtrade               ── 自動売買フレームワーク
Docker + Kubernetes     ── 本番デプロイ
Prometheus + Grafana    ── 監視
Telegram / Slack        ── アラート通知

from sklearn.model_selection import TimeSeriesSplit
import numpy as np

def walk_forward_validation(model, X, y, n_splits=5):
    """時系列データのウォークフォワード検証"""
    tscv = TimeSeriesSplit(n_splits=n_splits)
    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)
        oos_score = model.score(X_test, y_test)
        results.append(oos_score)
        print(f'Fold {fold+1}: OOS Score = {oos_score:.4f}')

    return {'oos_mean': np.mean(results), 'oos_std': np.std(results)}

from freqtrade.freqai.base_models.BaseRegressionModel import BaseRegressionModel
import tensorflow as tf
import numpy as np

class LSTMPricePredictor(BaseRegressionModel):
    """FreqAI対応LSTMモデル"""

    def fit(self, data_dictionary, dk):
        X_train = data_dictionary['train_features'].values
        y_train = data_dictionary['train_labels'].values

        # LSTM用に3次元へ変換 (samples, timesteps, features)
        X_train = X_train.reshape((X_train.shape[0], 1, X_train.shape[1]))

        model = tf.keras.Sequential([
            tf.keras.layers.LSTM(64, return_sequences=True,
                                 input_shape=(1, X_train.shape[2])),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.LSTM(32),
            tf.keras.layers.Dense(1)
        ])
        model.compile(optimizer='adam', loss='mse')
        model.fit(X_train, y_train, epochs=50, batch_size=32, verbose=0)

        self.model = model
        return self

    def predict(self, unfiltered_df, dk):
        X = unfiltered_df[dk.training_features_list].values
        X = X.reshape((X.shape[0], 1, X.shape[1]))
        return self.model.predict(X).flatten(), unfiltered_df.index

import shap
import matplotlib.pyplot as plt

def analyze_model_with_shap(model, X_test, feature_names):
    """SHAP値で予測根拠を可視化"""
    explainer = shap.TreeExplainer(model)
    shap_values = explainer.shap_values(X_test)

    shap.summary_plot(shap_values, X_test,
                      feature_names=feature_names,
                      show=False)
    plt.tight_layout()
    plt.savefig('shap_summary.png', dpi=150)
    return shap_values

① 取引所API（CCXT）→ リアルタイムOHLCVデータ取得
② テクニカル指標算出（RSI・MACD・Bollinger Bands）
③ MLモデルによる価格予測シグナル生成
④ リスク管理チェック（ポジションサイズ・ドローダウン上限）
⑤ 注文発注
⑥ ログ記録・Telegram/Slack通知

import ccxt
import pandas as pd

class ExchangeConnector:
    """CCXT統合クラス"""

    def __init__(self, exchange_id: str, api_key: str, secret: str):
        exchange_class = getattr(ccxt, exchange_id)
        self.exchange = exchange_class({
            'apiKey': api_key,
            'secret': secret,
            'enableRateLimit': True,
        })

    def fetch_ohlcv(self, symbol: str, timeframe: str = '1h',
                    limit: int = 500) -> pd.DataFrame:
        """OHLCVデータ取得（レート制限対応）"""
        ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        df = pd.DataFrame(ohlcv,
                          columns=['timestamp','open','high','low','close','volume'])
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        return df.set_index('timestamp')

    def place_order(self, symbol: str, side: str,
                    amount: float, order_type: str = 'market'):
        """注文発注（エラーハンドリング付き）"""
        try:
            return self.exchange.create_order(symbol, order_type, side, amount)
        except ccxt.InsufficientFunds as e:
            raise RuntimeError(f"残高不足: {e}")
        except ccxt.NetworkError as e:
            raise RuntimeError(f"ネットワークエラー: {e}")

class RiskManager:
    """リスク管理モジュール"""

    def __init__(self, max_drawdown: float = 0.15,
                 max_daily_loss: float = 0.05):
        self.max_drawdown = max_drawdown
        self.max_daily_loss = max_daily_loss
        self.peak_equity = None
        self.daily_loss = 0.0
        self.is_active = True

    def kelly_position_size(self, win_rate: float, avg_win: float,
                            avg_loss: float, capital: float,
                            kelly_fraction: float = 0.5) -> float:
        """ハーフケリー基準によるポジションサイジング"""
        if avg_loss == 0:
            return 0.0
        kelly = (win_rate - (1 - win_rate) / (avg_win / avg_loss))
        # ハーフケリーで保守的に適用
        return max(0.0, kelly * kelly_fraction) * capital

    def check_circuit_breaker(self, current_equity: float) -> bool:
        """サーキットブレーカー：ドローダウン超過で自動停止"""
        if self.peak_equity is None:
            self.peak_equity = current_equity

        self.peak_equity = max(self.peak_equity, current_equity)
        drawdown = (self.peak_equity - current_equity) / self.peak_equity

        if drawdown >= self.max_drawdown:
            self.is_active = False
            print(f"[CIRCUIT BREAKER] ドローダウン {drawdown:.1%} が上限を超過。取引停止。")

        return self.is_active

# docker-compose.yml
version: '3.8'
services:
  trading-bot:
    build: .
    environment:
      - EXCHANGE_API_KEY=${EXCHANGE_API_KEY}
      - EXCHANGE_SECRET=${EXCHANGE_SECRET}
      - TELEGRAM_TOKEN=${TELEGRAM_TOKEN}
    volumes:
      - ./models:/app/models
      - ./logs:/app/logs
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 2G

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    restart: unless-stopped

import anthropic

class LLMSentimentAnalyzer:
    """Claude APIを使ったセンチメント分析"""

    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)

    def analyze_sentiment(self, news_text: str) -> dict:
        message = self.client.messages.create(
            model="claude-opus-4-5",
            max_tokens=256,
            messages=[{
                "role": "user",
                "content": (
                    f"以下の仮想通貨関連ニュースの市場センチメントを分析してください。\n\n"
                    f"ニュース: {news_text}\n\n"
                    f"JSON形式で返答: "
                    f'{{"sentiment": "bullish/bearish/neutral", '
                    f'"confidence": 0.0-1.0, "summary": "要約"}}'
                )
            }]
        )
        import json
        return json.loads(message.content[0].text)

[バックテスト設計]
  └─ 3大バイアス排除（ルックアヘッド・過学習・データリーク）
  └─ ウォークフォワード検証
  └─ 評価指標による多角的評価

[システム統合]
  └─ CCXT による取引所接続
  └─ リスク管理（ケリー基準・サーキットブレーカー）
  └─ Docker による本番デプロイ
  └─ Prometheus + Grafana による監視

[高度化]
  └─ LLMセンチメント分析との統合
  └─ マルチエージェントアーキテクチャ
  └─ 定期再学習・ウォークフォワード更新の自動化