第 17 課：オンライン学習と戦略進化

目標: 静的モデルが必然的に減衰する理由を理解し、継続的な戦略進化の核心メカニズムを習得する。

典型的なシナリオ(例示)

注:以下は一般的な現象を説明するための合成例です。数字は例示的なものであり、特定の個人やアカウントには対応していません。

2018年、あるQuantトレーダーがRandom Forestベースの株式選択モデルをA株用に開発しました。バックテストは年率35%リターン、1.8 Sharpeを示し、すべてのメトリクスで優れていました。

最初の6ヶ月間、モデルは安定してバックテストと一致しました。その後、パフォーマンスがゆっくりと低下し始めました。突然の失敗ではなく、ゆっくり熱せられる水の中のカエルのように - 毎月少しずつ収益が減り、Drawdownが少しずつ増えました。

12ヶ月後、年率リターンは35%から8%に低下し、インデックスをかろうじて上回る程度でした。

彼はコードをチェックしましたが、バグはありませんでした。データをチェックしましたが、エラーはありませんでした。モデルは設計通りに正確に動作していましたが、もはや利益を生まなくなっていました。

何が間違っていたのか? 市場が変わったのに、モデルは変わらなかったのです。2018年初頭は個人投資家が支配的で、Momentum Factorが機能していました。2018年後半には機関投資家の資金が入り、Momentum Factorは裁定取引されました。モデルが学習したパターンは特定の期間のパターンであり、永遠の真理ではありませんでした。

これがConcept Drift - モデルが学習したデータ分布が、取引しているデータ分布と一致しなくなることです。モデルは間違っていません、世界が変わったのです。

17.1 なぜ静的モデルは必然的に減衰するのか

市場の非定常性

金融市場には基本的な特徴があります:非定常性。これは以下を意味します:

1. 統計的特性が変化: 平均、分散、相関は定数ではない
2. パターンが期限切れ: 有効なAlphaが発見され、裁定取引され、消失する
3. 参加者が適応: あなたの戦略が利益を上げる -> 他の人がコピー -> Alphaが減衰

紙上演習: Alpha減衰を計算

シナリオ: あなたのFactor IC = 0.05、市場効率向上によりICが月5%減衰すると仮定。

計算:

月    IC        期待年率 (IC x sqrt(252) x sigma、sigma=20%と仮定)
0      0.050      15.8%
6      0.050 x 0.95^6 = 0.037      11.7%
12     0.050 x 0.95^12 = 0.027      8.5%
18     0.050 x 0.95^18 = 0.020      6.3%
24     0.050 x 0.95^24 = 0.015      4.7%

結論: 月5%の減衰だけでも、2年後には戦略のリターンは元の30%にしかなりません。

自己チェック: 現在の戦略のICを計算し、3%、5%、10%の月次減衰を仮定して、許容できないレベルに低下するまでの期間を計算してください。

なぜ「定期的な再トレーニング」では不十分なのか

多くの人の解決策は「毎月モデルを再トレーニングする」ことです。これは静的モデルより良いですが、3つの問題があります:

正しいアプローチ: 「再学習」ではなく「継続的に適応」 - 過去の知識を保持しながら新しい環境に徐々に調整。

17.2 Online Learningの核心メカニズム

Batch vs Online Learning

Online Learningの3つのモード

紙上演習: 忘却係数の選択

シナリオ: あなたの戦略を2つの市場環境でテストします。

式: 有効ルックバック日数 ~ ln(0.1) / ln(lambda)

自己チェック: あなたの戦略の市場環境に基づいて、適切なlambda値を選択してください。間違った選択をするとどうなりますか?

• lambdaが大きすぎる (忘れるのが遅すぎる): 適応が遅すぎる、新しい環境で損失
• lambdaが小さすぎる (忘れるのが速すぎる): 最近のノイズに過学習、安定性が悪い

class ExponentialMovingModel:
    """指数忘却を持つOnline learningモデル"""

    def __init__(self, decay_factor: float = 0.95):
        self.lambda_ = decay_factor
        self.weights = None
        self.cumulative_weight = 0

    def update(self, X: np.ndarray, y: float, learning_rate: float = 0.01):
        """増分更新"""
        if self.weights is None:
            self.weights = np.zeros(X.shape[0])

        # 予測
        pred = np.dot(self.weights, X)
        error = y - pred

        # ウェイトを更新 (忘却付き勾配降下)
        self.weights = self.lambda_ * self.weights + learning_rate * error * X
        self.cumulative_weight = self.lambda_ * self.cumulative_weight + 1

        return pred, error

    def get_effective_lookback(self, threshold: float = 0.1) -> int:
        """有効ルックバック日数を計算"""
        return int(np.log(threshold) / np.log(self.lambda_))

17.3 Strategy進化の4つのレベル

Online learningはStrategy進化の一部に過ぎません。完全な進化システムには4つのレベルがあります:

レベル1: Signal適応 (最速)

Signal適応は最速の進化レベルで、モデルの再トレーニングは必要ありません。

例: 動的閾値調整

静的閾値: Signal > 0.5の場合に買う

動的閾値:
- 過去N日のSignal分布を計算
- 閾値 = Signal平均 + k x Signal標準偏差
- 高市場ボラティリティ、閾値が自動増加、取引を減らす
- 低市場ボラティリティ、閾値が自動減少、取引を増やす

紙上演習:

レベル2: モデル進化 (中速)

モデル進化にはパラメータ更新と特徴エンジニアリングが含まれます。

重要なメカニズム: 特徴重要度監視

+-------------------------------------------------+
|           特徴重要度変化監視                      |
+----------------+-----------+-----------+---------+
| Feature        | 6ヶ月前   | 現在      | Trend  |
+----------------+-----------+-----------+---------+
| Momentum_20D   | 0.25      | 0.08      | 減衰中 |
| Volatility_30D | 0.15      | 0.22      | 強化中 |
| Volume_Ratio   | 0.18      | 0.05      | 失敗   |
| Earnings_Surprise | 0.12   | 0.18      | 有効   |
+----------------+-----------+-----------+---------+

-> 決定: Momentumウェイトを下げる、Volume ratioを削除、Volatilityウェイトを増やす

進化トリガー条件:

レベル3: Strategy進化 (より遅い)

Strategy進化にはStrategy Portfolioの再構成が含まれます。

Strategyライフサイクル管理:

                    Incubation      Validation      Maturity        Decay
Return/Risk    ................ ............ ............... .........
               Paper trading    Small live      Main strategy    Phase out
               3-6ヶ月          3-6ヶ月         無期限           無期限

入場基準:     Backtest合格    Sharpe>1.0      Sharpe>1.5       Sharpe<0.8
             Quality Gate   Live DD<15%   6ヶ月プラス      3ヶ月損失

資本配分:      0%           5-10%           20-40%          ゼロへ段階的

Strategy Portfolioの動的ウェイト:

紙上演習: Strategyウェイト配分

計算プロセス:

• Inverse volatility: 1/5 = 0.20、1/8 = 0.125、1/8 = 0.125、合計 = 0.45
  • A: 0.20/0.45 = 0.44、B: 0.125/0.45 = 0.28、C: 0.28
• Sharpe proportional: 合計 = 1.6 + 0.75 + 1.25 = 3.6
  • A: 1.6/3.6 = 0.44、B: 0.75/3.6 = 0.21、C: 1.25/3.6 = 0.35

レベル4: アーキテクチャ進化 (最も遅い)

アーキテクチャ進化は最も基本的な変更で、通常は人間の介入が必要です。

典型的なアーキテクチャ進化パス:

ステージ1: 単一Strategy
+-- 1つのシンプルなトレンドフォロー戦略

ステージ2: Strategy Portfolio
+-- Trend + Mean Reversion + Momentum、固定ウェイト

ステージ3: 動的Portfolio
+-- 市場状態に基づいてStrategyウェイトを動的に調整

ステージ4: Multi-Agent
+-- Expert Agents + Meta Agent + Risk Agent

ステージ5: 自己進化システム
+-- 新しい戦略を自動的に発見、テスト、デプロイできるシステム

アーキテクチャ進化トリガー条件:

17.4 Online Learning実装の重要技術

1. Concept Drift検出

モデルを更新する前に、実際にDriftが発生したかを確認します。

一般的な検出方法:

紙上演習: シンプルなDrift検出

監視メトリック: モデル予測スライディング精度

時間    精度      5日平均      警告? (平均<50%)
D1      55%       55%           No
D2      52%       53.5%         No
D3      48%       51.7%         No
D4      45%       50.0%         No
D5      42%       48.4%         Yes <- Drift検出をトリガー
D6      40%       45.4%         Yes

Driftが検出されたときにモデル更新をトリガー、盲目的な定期更新ではありません。

2. 壊滅的な忘却の防止

Online learningのリスクの1つは「壊滅的な忘却」 - 新しいことを学ぶときに古い知識を忘れることです。

解決策:

実用的な推奨:

1. 過去の代表的なサンプルの10-20%を保持
2. 各更新で、新しいデータ + 過去のサンプルを混ぜる
3. 過去のテストセットでパフォーマンスを監視して忘却がないことを確認

3. 更新タイミングと頻度

更新が頻繁すぎる:

• 最近のノイズに過学習
• 計算リソースの無駄
• モデルが不安定

更新が不十分:

• 適応ラグ
• 市場変化を見逃す
• 蓄積されたエラー

推奨戦略:

class AdaptiveUpdateScheduler:
    """適応更新スケジューラー"""

    def __init__(self, min_interval: int = 5, max_interval: int = 20):
        self.min_interval = min_interval  # 最小更新間隔 (日数)
        self.max_interval = max_interval  # 最大更新間隔 (日数)
        self.last_update = 0
        self.drift_detector = DDM()  # Drift検出器

    def should_update(self, day: int, recent_errors: list) -> tuple[bool, str]:
        """モデルを更新すべきか決定"""
        days_since_update = day - self.last_update

        # Driftをチェック
        drift_detected = self.drift_detector.detect(recent_errors)

        # 決定ロジック
        if days_since_update >= self.max_interval:
            return True, "最大間隔に到達、強制更新"

        if drift_detected and days_since_update >= self.min_interval:
            return True, "Drift検出、更新をトリガー"

        return False, "更新不要"

    def record_update(self, day: int):
        self.last_update = day
        self.drift_detector.reset()

17.5 Online Learningのエンジニアリング実装

1. スライディングウィンドウ再トレーニング実装

from datetime import datetime, timedelta
from typing import Optional, Tuple
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier

class SlidingWindowRetrainer:
    """スライディングウィンドウモデル再トレーナー"""

    def __init__(self,
                 window_size_days: int = 252,
                 min_samples: int = 200,
                 retrain_interval_days: int = 20):
        """
        window_size_days: トレーニングウィンドウサイズ (日数)
        min_samples: 最小トレーニングサンプル
        retrain_interval_days: 再トレーニング間隔 (日数)
        """
        self.window_size = window_size_days
        self.min_samples = min_samples
        self.retrain_interval = retrain_interval_days

        self.model = None
        self.last_train_date = None
        self.training_history = []

    def should_retrain(self, current_date: datetime) -> Tuple[bool, str]:
        """再トレーニングが必要か決定"""
        if self.model is None:
            return True, "初期トレーニング"

        days_since_train = (current_date - self.last_train_date).days

        if days_since_train >= self.retrain_interval:
            return True, f"最後のトレーニングから{days_since_train}日"

        return False, "再トレーニング不要"

    def train(self, data: pd.DataFrame,
              feature_cols: list,
              target_col: str,
              current_date: datetime) -> dict:
        """
        スライディングウィンドウを使用してモデルをトレーニング

        data: 'date'列とFeature列を含む必要があります
        """
        # ウィンドウ境界を計算
        window_end = current_date
        window_start = current_date - timedelta(days=self.window_size)

        # ウィンドウ内のデータをフィルター
        mask = (data['date'] >= window_start) & (data['date'] < window_end)
        train_data = data[mask]

        if len(train_data) < self.min_samples:
            return {
                'success': False,
                'reason': f'サンプル不足: {len(train_data)} < {self.min_samples}'
            }

        X = train_data[feature_cols].values
        y = train_data[target_col].values

        # モデルをトレーニング
        self.model = RandomForestClassifier(
            n_estimators=100,
            max_depth=5,
            min_samples_leaf=20,
            random_state=42
        )
        self.model.fit(X, y)

        # トレーニングを記録
        self.last_train_date = current_date

        train_info = {
            'success': True,
            'date': current_date,
            'window': (window_start, window_end),
            'n_samples': len(train_data),
            'train_accuracy': self.model.score(X, y)
        }
        self.training_history.append(train_info)

        return train_info

    def predict(self, X: np.ndarray) -> np.ndarray:
        """予測"""
        if self.model is None:
            raise ValueError("モデルがまだトレーニングされていません")
        return self.model.predict_proba(X)[:, 1]

2. Drift検出メトリクス実装

Population Stability Index (PSI)

PSIはFeature分布Driftを検出するための標準メトリックです。

def calculate_psi(expected: np.ndarray,
                  actual: np.ndarray,
                  n_bins: int = 10) -> float:
    """
    Population Stability Indexを計算

    PSI < 0.1: 有意なDriftなし
    0.1 <= PSI < 0.25: 中程度のDrift、注意が必要
    PSI >= 0.25: 有意なDrift、対応が必要
    """
    # ビンを作成
    breakpoints = np.percentile(expected,
                                np.linspace(0, 100, n_bins + 1))
    breakpoints[0] = -np.inf
    breakpoints[-1] = np.inf

    # ビン比率を計算
    expected_counts = np.histogram(expected, bins=breakpoints)[0]
    actual_counts = np.histogram(actual, bins=breakpoints)[0]

    # 比率に変換 (ゼロ除算を避ける)
    expected_pct = (expected_counts + 1) / (len(expected) + n_bins)
    actual_pct = (actual_counts + 1) / (len(actual) + n_bins)

    # PSIを計算
    psi = np.sum((actual_pct - expected_pct) *
                 np.log(actual_pct / expected_pct))

    return psi


class FeatureDriftMonitor:
    """Feature Drift監視"""

    def __init__(self, psi_threshold: float = 0.2):
        self.baseline_distributions = {}
        self.psi_threshold = psi_threshold
        self.drift_history = []

    def set_baseline(self, feature_name: str, values: np.ndarray):
        """ベースライン分布を設定"""
        self.baseline_distributions[feature_name] = values.copy()

    def check_drift(self, feature_name: str,
                    current_values: np.ndarray,
                    date: datetime) -> dict:
        """単一FeatureのDriftをチェック"""
        if feature_name not in self.baseline_distributions:
            raise ValueError(f"ベースライン未設定: {feature_name}")

        baseline = self.baseline_distributions[feature_name]
        psi = calculate_psi(baseline, current_values)

        result = {
            'feature': feature_name,
            'date': date,
            'psi': psi,
            'drift_detected': psi >= self.psi_threshold,
            'severity': 'high' if psi >= 0.25 else
                       ('medium' if psi >= 0.1 else 'low')
        }

        self.drift_history.append(result)
        return result

    def check_all_features(self, current_data: pd.DataFrame,
                           date: datetime) -> dict:
        """すべてのFeature Driftをチェック"""
        results = {}
        drifted_features = []

        for feature in self.baseline_distributions:
            if feature in current_data.columns:
                result = self.check_drift(
                    feature,
                    current_data[feature].values,
                    date
                )
                results[feature] = result
                if result['drift_detected']:
                    drifted_features.append(feature)

        return {
            'date': date,
            'features_checked': len(results),
            'features_drifted': len(drifted_features),
            'drifted_features': drifted_features,
            'details': results,
            'action_required': len(drifted_features) &gt; 0
        }

Kolmogorov-Smirnov 検定

from scipy import stats

def ks_drift_test(baseline: np.ndarray,
                  current: np.ndarray,
                  significance: float = 0.05) -> dict:
    """
    KS検定を用いた分布Driftの検出
    """
    statistic, p_value = stats.ks_2samp(baseline, current)

    return {
        'ks_statistic': statistic,
        'p_value': p_value,
        'drift_detected': p_value < significance,
        'interpretation':
            '分布に有意な差異あり' if p_value < significance
            else '分布に有意な差異なし'
    }

3. 再トレーニング vs 一時停止決定フレームワーク

class RetrainDecisionEngine:
    """再トレーニング決定エンジン"""

    def __init__(self,
                 psi_threshold: float = 0.2,
                 performance_drop_threshold: float = 0.3,
                 min_confidence_for_retrain: float = 0.6):
        self.psi_threshold = psi_threshold
        self.perf_threshold = performance_drop_threshold
        self.min_confidence = min_confidence_for_retrain

    def decide(self,
               drift_report: dict,
               performance_report: dict,
               data_quality_report: dict) -> dict:
        """
        次のアクションを決定

        Returns: 'retrain', 'pause', 'continue', 'investigate'
        """
        # シグナルを収集
        drift_detected = drift_report.get('action_required', False)
        perf_drop = performance_report.get('ic_change', 0)
        data_quality_ok = data_quality_report.get('quality_score', 1) > 0.8
        sample_size_ok = data_quality_report.get('n_samples', 0) > 200

        # 決定マトリックス
        #
        #                    パフォーマンス低下?
        #                    Yes        No
        #   Drift?   Yes    [Retrain]  [Observe]
        #            No     [Pause]    [Continue]
        #

        if drift_detected:
            if perf_drop > self.perf_threshold:
                # Drift + パフォーマンス低下: 再トレーニングが必要
                if data_quality_ok and sample_size_ok:
                    return {
                        'action': 'retrain',
                        'confidence': 0.9,
                        'reason': 'Drift検出とパフォーマンス低下、データ品質十分',
                        'urgency': 'high'
                    }
                else:
                    return {
                        'action': 'investigate',
                        'confidence': 0.5,
                        'reason': '再トレーニングが必要だがデータ品質不十分、人間の介入が必要',
                        'urgency': 'high'
                    }
            else:
                # Driftだがパフォーマンスは正常: 観察を続ける
                return {
                    'action': 'continue',
                    'confidence': 0.6,
                    'reason': 'Drift検出だがパフォーマンスはまだ影響なし、監視を続ける',
                    'urgency': 'low',
                    'next_check_days': 3
                }
        else:
            if perf_drop > self.perf_threshold:
                # Driftなしだがパフォーマンス低下: 過学習または市場変化の可能性
                return {
                    'action': 'pause',
                    'confidence': 0.7,
                    'reason': 'パフォーマンス低下だが明らかなDriftなし、一時停止して調査を推奨',
                    'urgency': 'medium'
                }
            else:
                # Driftなしでパフォーマンス正常: 実行を続ける
                return {
                    'action': 'continue',
                    'confidence': 0.95,
                    'reason': 'すべて正常',
                    'urgency': 'none'
                }

4. 紙上演習: 再トレーニング vs 一時停止の判断

重要な原則:

1. Drift + パフォーマンス低下 = 再トレーニング (市場が変わった、モデルは適応が必要)
2. Driftなし + パフォーマンス低下 = 一時停止 (他の問題の可能性、盲目的な再トレーニングは避ける)
3. Drift + パフォーマンス正常 = 観察 (一時的な変動の可能性)
4. データ品質不良 = データを先に修正 (そうでなければ再トレーニングは無駄)

17.6 Multi-Agent視点からの進化

Multi-agentアーキテクチャでは、進化は個々のモデルの更新だけでなく、Agents間の協調です。

進化アーキテクチャ

Evolution Agentの責任

Agent-レベルA/Bテスト

協調更新の課題

合格基準

このレッスンを完了した後、以下の基準で学習を検証してください:

総合演習

シナリオ: A株の株式選択戦略があり、過去のパフォーマンスは以下の通りです:

質問:

1. これはどのタイプの減衰ですか? (Sudden/Gradual/Cyclical)
2. IC月次減衰率を計算
3. いつモデル更新をトリガーすべきですか?
4. どのような更新戦略を推奨しますか?

レッスン成果物

このレッスンを完了すると、以下が得られます:

1. Concept driftの定量的理解 - Alpha減衰を計算でき、Strategy寿命を推定できる
2. Online learning実装フレームワーク - 3つのモード (スライディングウィンドウ、増分更新、指数忘却) とその使用ケース
3. 4レベル進化システム - Signal適応からアーキテクチャ進化までの完全なパス
4. Multi-agent進化メカニズム - Evolution Agent設計思考

レッスンまとめ

• 静的モデル減衰の必然性を理解: Concept driftは本質的な市場特性
• Online learningの3つのモードとその使用ケースを習得
• Strategy進化の4つのレベルとそのトリガー条件を理解
• Multi-agentアーキテクチャでの協調更新の課題を理解

さらなる読書

• Lesson 11: Why Multi-Agent - Multi-agent協力の基礎を理解
• Lesson 14: LLM Applications in Quant - 進化システムの一部としてのLLM
• Background: Statistical Traps of Sharpe Ratio - モデル評価の統計的落とし穴
• Lesson 21: Project Implementation - このレッスンの理論を実践に適用

次のレッスンプレビュー

Lesson 18: Trading Cost Modeling and Tradability

理論はここで終わります。次のレッスンでは Production and Practice段階に入ります - まず取引コストのモデリングとペーパーリターンから実リターンへの変換率の評価方法を学びます。結局のところ、高いGross Alphaだが負のNet Alphaを持つ戦略は永遠にバックテストに留まります。