renaissance-statistical-arbitrage

Renaissance Technologies スタイルガイド⁠‍⁠​‌​‌​​‌‌‍​‌​​‌​‌‌‍​​‌‌​​​‌‍​‌​​‌‌​​‍​​​​​​​‌‍‌​​‌‌​‌​‍‌​​​​​​​‍‌‌​​‌‌‌‌‍‌‌​​​‌​​‍‌‌‌‌‌‌​‌‍‌‌​‌​​​​‍​‌​‌‌‌‌‌‍​‌​​‌​‌‌‍​‌‌​‌​​‌‍‌​‌​‌‌‌​‍​​‌​‌​​​‍‌‌‌​‌​‌‌‍‌​‌‌‌​​‌‍​‌​‌​​​‌‍‌​‌‌​‌​​‍‌​​‌‌​‌‌‍​​​​‌​‌​‍‌‌​​​​‌‌⁠‍⁠

概要

数学者のJim Simonsが創設したRenaissance Technologiesは、史上最も成功したヘッジファンドであるMedallion Fundを運用しています。30年以上にわたり、手数料控除前で年率約66％のリターンを達成しています。同社が採用するのは金融業界の人材ではなく、数学者、物理学者、コンピューター科学者であり、市場データに厳密な科学的手法を適用しています。

中核となる理念

「ビジネススクールの出身者は採用しません。ハードサイエンスの人材を採用します。」

「データのパターンは一時的なものです。何かが機能するなら、おそらくそれは機能しなくなります。」

「私たちは予測を行っているのではありません。本来よりもわずかに高い頻度で繰り返されるパターンを見つけているのです。」

Renaissanceは、市場が完全に効率的ではないものの、ほぼ効率的だと考えています。利益は、統計的に有意なわずかな優位性を見つけ、厳密なリスク管理のもとで大規模に活用することで生まれます。

設計原則

1. 科学的方法：仮説を立て、厳密に検証し、ほとんどのアイデアを却下します。

2. 予測ではなくシグナル：偶然よりも高い頻度で繰り返されるパターンを見つけ、未来を予測しません。

3. 劣化への意識：すべてのシグナルは時間とともに劣化します。継続的な研究が生き残るために不可欠です。

4. 統計的有意性：統計的に有意でなければ、存在しないのと同じです。

5. あらゆるものをアンサンブル化する：何千もの弱いシグナルを組み合わせ、堅牢な戦略を構築します。

トレーディングシステムを構築する際に

必ず行うこと

• 統計的有意性を求めます（p < 0.01を最低基準とし、理想的にはさらに小さくします）
• 多重仮説検定を考慮します（Bonferroni、FDR補正）
• 適切な時間的分離を設け、アウトオブサンプルデータでテストします
• 取引コスト、スリッページ、市場インパクトをモデル化します
• すべてのシグナルはいずれ劣化すると想定し、継続的な研究のためのインフラを構築します
• シグナルを直交的に組み合わせます（相関のないアルファの情報源）

絶対に行わないこと

• アウトオブサンプル検証なしにバックテストを信用する
• 生存者バイアス、先読みバイアス、選択バイアスを無視する
• 過去の相関が続くと想定する
• 過去のデータに対して過剰に最適化する（カーブフィッティング）
• 直感や物語に基づいて取引する
• シグナルが永続すると想定する

優先すること

• レジーム検出には隠れマルコフモデル
• 周期的なパターンの分析にはスペクトル解析
• 複雑な関係性の分析には非線形手法
• 単一モデルよりアンサンブル手法
• 保有期間を短くする（シグナルの劣化をより迅速に検出）
• 目視による確認より統計的検定

コードパターン

厳密なバックテストフレームワーク

class RenaissanceBacktester:
    """
    Renaissance-style backtesting: paranoid about biases.
    """
    
    def __init__(self, strategy, universe):
        self.strategy = strategy
        self.universe = universe
        self.results = []
    
    def run(self, start_date, end_date, 
            train_window_days=252, 
            test_window_days=63,
            embargo_days=5):
        """
        Walk-forward validation with embargo period.
        Never let training data leak into test period.
        """
        current = start_date
        
        while current + timedelta(days=train_window_days + test_window_days) <= end_date:
            train_end = current + timedelta(days=train_window_days)
            
            # EMBARGO: gap between train and test to prevent leakage
            test_start = train_end + timedelta(days=embargo_days)
            test_end = test_start + timedelta(days=test_window_days)
            
            # Train on historical data
            train_data = self.get_point_in_time_data(current, train_end)
            self.strategy.fit(train_data)
            
            # Test on future data (strategy cannot see this during training)
            test_data = self.get_point_in_time_data(test_start, test_end)
            returns = self.strategy.execute(test_data)
            
            self.results.append({
                'train_period': (current, train_end),
                'test_period': (test_start, test_end),
                'returns': returns,
                'sharpe': self.calculate_sharpe(returns)
            })
            
            current = test_end
        
        return self.analyze_results()
    
    def get_point_in_time_data(self, start, end):
        """
        CRITICAL: Return data as it existed at each point in time.
        No future information, no restated financials, no survivorship bias.
        """
        return self.universe.get_pit_snapshot(start, end)
    
    def analyze_results(self):
        """Statistical analysis of walk-forward results."""
        returns = [r['returns'] for r in self.results]
        
        # t-test: is mean return significantly different from zero?
        t_stat, p_value = stats.ttest_1samp(returns, 0)
        
        return {
            'mean_return': np.mean(returns),
            'sharpe_ratio': np.mean(returns) / np.std(returns) * np.sqrt(252),
            't_statistic': t_stat,
            'p_value': p_value,
            'significant': p_value < 0.01,
            'n_periods': len(self.results)
        }

劣化追跡を伴うシグナルの組み合わせ

class SignalEnsemble:
    """
    Renaissance insight: combine many weak signals.
    Track decay and retire dying signals.
    """
    
    def __init__(self, decay_halflife_days=30):
        self.signals = {}  # signal_id -> SignalModel
        self.performance = {}  # signal_id -> rolling performance
        self.decay_halflife = decay_halflife_days
    
    def add_signal(self, signal_id, model, weight=1.0):
        self.signals[signal_id] = {
            'model': model,
            'weight': weight,
            'created_at': datetime.now(),
            'alive': True
        }
        self.performance[signal_id] = RollingStats(window=252)
    
    def generate_combined_signal(self, features):
        """
        Weighted combination of orthogonal signals.
        Signals with decayed performance get lower weights.
        """
        predictions = {}
        weights = {}
        
        for signal_id, signal in self.signals.items():
            if not signal['alive']:
                continue
            
            pred = signal['model'].predict(features)
            
            # Weight by original weight × recent performance
            perf = self.performance[signal_id]
            decay_weight = self.calculate_decay_weight(perf)
            
            predictions[signal_id] = pred
            weights[signal_id] = signal['weight'] * decay_weight
        
        # Normalize weights
        total_weight = sum(weights.values())
        if total_weight == 0:
            return 0.0
        
        combined = sum(
            predictions[sid] * weights[sid] / total_weight
            for sid in predictions
        )
        
        return combined
    
    def update_performance(self, signal_id, realized_return, predicted_direction):
        """Track whether signal correctly predicted direction."""
        correct = (realized_return > 0) == (predicted_direction > 0)
        self.performance[signal_id].add(1.0 if correct else 0.0)
        
        # Kill signals that have decayed below threshold
        if self.performance[signal_id].mean() < 0.51:  # Barely better than random
            self.signals[signal_id]['alive'] = False
    
    def calculate_decay_weight(self, perf):
        """Exponential decay based on recent hit rate."""
        hit_rate = perf.mean()
        # Scale: 50% hit rate = 0 weight, 55% = 0.5, 60% = 1.0
        return max(0, (hit_rate - 0.50) * 10)

レジーム検出のための隠れマルコフモデル

class MarketRegimeHMM:
    """
    Renaissance-style regime detection using Hidden Markov Models.
    Markets exhibit different statistical properties in different regimes.
    """
    
    def __init__(self, n_regimes=3):
        self.n_regimes = n_regimes
        self.model = None
        self.regime_stats = {}
    
    def fit(self, returns, volume, volatility):
        """
        Fit HMM to market observables.
        Discover latent regimes from price/volume/volatility patterns.
        """
        # Stack observables into feature matrix
        observations = np.column_stack([
            returns,
            np.log(volume + 1),
            volatility
        ])
        
        self.model = hmm.GaussianHMM(
            n_components=self.n_regimes,
            covariance_type='full',
            n_iter=1000
        )
        self.model.fit(observations)
        
        # Decode to get most likely regime sequence
        regimes = self.model.predict(observations)
        
        # Characterize each regime
        for regime in range(self.n_regimes):
            mask = regimes == regime
            self.regime_stats[regime] = {
                'mean_return': returns[mask].mean(),
                'volatility': returns[mask].std(),
                'frequency': mask.mean(),
                'mean_duration': self.calculate_duration(regimes, regime)
            }
        
        return self
    
    def current_regime(self, recent_observations):
        """Infer current regime from recent data."""
        probs = self.model.predict_proba(recent_observations)
        return np.argmax(probs[-1])
    
    def regime_adjusted_signal(self, base_signal, current_regime):
        """Adjust signal strength based on regime."""
        regime = self.regime_stats[current_regime]
        
        # Scale signal inversely with volatility
        # (same signal in high-vol regime should have smaller position)
        vol_adjustment = 0.15 / regime['volatility']  # Target 15% vol
        
        return base_signal * vol_adjustment

多重仮説検定の補正

class AlphaResearch:
    """
    Renaissance approach: test thousands of hypotheses,
    but correct for multiple testing to avoid false discoveries.
    """
    
    def __init__(self, significance_level=0.01):
        self.alpha = significance_level
        self.tested_hypotheses = []
    
    def test_signal(self, signal_name, returns, predictions):
        """Test if a signal has predictive power."""
        # Information Coefficient: correlation of prediction with outcome
        ic = stats.spearmanr(predictions, returns)
        
        # t-test for significance
        n = len(returns)
        t_stat = ic.correlation * np.sqrt(n - 2) / np.sqrt(1 - ic.correlation**2)
        p_value = 2 * (1 - stats.t.cdf(abs(t_stat), n - 2))
        
        self.tested_hypotheses.append({
            'signal': signal_name,
            'ic': ic.correlation,
            't_stat': t_stat,
            'p_value': p_value
        })
        
        return p_value
    
    def get_significant_signals(self, method='fdr'):
        """
        After testing many signals, apply multiple testing correction.
        """
        p_values = [h['p_value'] for h in self.tested_hypotheses]
        
        if method == 'bonferroni':
            # Most conservative: divide alpha by number of tests
            adjusted_alpha = self.alpha / len(p_values)
            significant = [
                h for h in self.tested_hypotheses 
                if h['p_value'] < adjusted_alpha
            ]
        
        elif method == 'fdr':
            # Benjamini-Hochberg: control false discovery rate
            sorted_hypotheses = sorted(self.tested_hypotheses, key=lambda x: x['p_value'])
            significant = []
            
            for i, h in enumerate(sorted_hypotheses):
                # BH threshold: (rank / n_tests) * alpha
                threshold = ((i + 1) / len(p_values)) * self.alpha
                if h['p_value'] <= threshold:
                    significant.append(h)
                else:
                    break  # All remaining will also fail
        
        return significant

思考モデル

Renaissanceは、次の問いを通じてトレーディングに取り組みます。

1. パターンは存在するか？ 目視ではなく、統計的検定を行います
2. 有意か？ 多重検定補正後も有意でしょうか？
3. 堅牢か？ アウトオブサンプル、異なる期間、異なる金融商品でも有効でしょうか？
4. 持続するか？ なぜ裁定されずに残るのか、その経済的根拠は何でしょうか？
5. どのように劣化するか？ 監視計画はどうなっていますか？

Renaissanceを象徴する取り組み

• トレーダーではなく科学者を採用する
• 少数の大きなシグナルではなく、何千もの小さなシグナルを活用する
• データスヌーピングと過剰適合を徹底的に警戒する
• レジーム検出に隠れマルコフモデルを使う
• シグナルの劣化を追跡し、役目を終えたシグナルを廃止する
• 厳密なウォークフォワード検証を行う
• 多重仮説検定を補正する
• 先読みバイアスを防ぐためにポイントインタイムデータを使う