renaissance-statistical-arbitrage

Name: renaissance-statistical-arbitrage
Author: copyleftdev

github.com/copyleftdev数据与数据库2115 次安装15 周前更新

代码仓库

关于 renaissance-statistical-arbitrage

面向量化交易研究的技能，用于设计统计套利系统，强调假设检验、偏差控制、集成信号与衰减监测，适合构建和评估数据驱动策略的研究者。

主要功能

01统计假设检验
02样本外验证
03多重检验校正
04信号集成与衰减跟踪
05隐马尔可夫状态识别

适用场景

01评估候选市场信号
02减少回测偏差与过拟合
03监测并淘汰衰减信号

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/copyleftdev/sk1llz --skill renaissance-statistical-arbitrage

安装后在 Claude Code 中输入/renaissance-statistical-arbitrage确认已加载。

详细介绍

Renaissance Technologies 风格指南⁠‍⁠‌‌‌‌‍‌‌‌‌‍‌‌‌‍‌‌‌‍‌‍‌‌‌‌‍‌‍‌‌‌‌‌‌‍‌‌‌‍‌‌‌‌‌‌‌‍‌‌‌‍‌‌‌‌‌‌‍‌‌‌‌‍‌‌‌‌‍‌‌‌‌‌‍‌‌‍‌‌‌‌‌‌‍‌‌‌‌‌‍‌‌‌‍‌‌‌‌‍‌‌‌‌‌‍‌‌‍‌‌‌‌⁠‍⁠

概述

由数学家 Jim Simons 创立的 Renaissance Technologies 运营着 Medallion Fund——这是史上最成功的对冲基金，在 30 多年间，扣除费用前的年回报率约为 66%。该公司聘用数学家、物理学家和计算机科学家（而非金融从业者），并将严谨的科学方法应用于市场数据。

核心理念

“我们不从商学院招人。我们招的是自然科学领域的人才。”

“数据中的模式转瞬即逝。如果某个方法有效，它很可能很快就会失效。”

“我们的业务不是预测，而是寻找出现频率略高于预期的重复模式。”

Renaissance 认为，市场并非完全有效，但已近乎有效。利润来自发现微小且具有统计显著性的优势，并通过严谨的风险管理大规模利用这些优势。

设计原则

科学方法：提出假设、严格检验，并否定大多数想法。
信号，而非预测：寻找出现频率高于随机概率的重复模式；不要预测未来。
关注衰减：每个信号都会随时间退化。持续研究是生存之道。
统计显著性：如果不具备统计显著性，就等于不存在。
全面采用集成方法：将数千个较弱的信号组合成稳健的策略。

构建交易系统时

始终

要求具备统计显著性（p < 0.01 为最低标准，理想情况下应远低于此）
考虑多重假设检验（Bonferroni、FDR 校正）
使用样本外数据进行测试，并确保适当的时间区隔
对交易成本、滑点和市场冲击建模
假设每个信号都会衰减——构建支持持续研究的基础设施
以正交方式组合信号（不相关的 alpha 来源）

绝不

不经样本外验证就相信回测
忽视幸存者偏差、前视偏差或选择偏差
假设过去的相关性会持续存在
在历史数据上过度优化（曲线拟合）
凭直觉或叙事进行交易
假设某个信号会永远有效

优先采用

使用隐马尔可夫模型进行状态识别
使用谱分析寻找周期性模式
使用非线性方法处理复杂关系
优先采用集成方法，而非单一模型
采用较短的持有期（更快检测信号衰减）
优先采用统计检验，而非目视检查

代码模式

严谨的回测框架

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 通过以下问题来审视交易：

**是否存在某种模式？**进行统计检验，而不是凭肉眼判断。
**是否显著？**经过多重检验校正后是否仍然显著？
**是否稳健？**在样本外、不同时间段和不同金融工具上是否都成立？
**能否持续？**有什么经济学依据，能说明这种优势为何不会被套利消除？
**将如何衰减？**监控计划是什么？

Renaissance 的标志性做法

聘用科学家，而非交易员
使用数千个小信号，而非少数几个大信号
对数据窥探和过拟合保持高度警惕
使用隐马尔可夫模型进行状态识别
追踪信号衰减，并及时淘汰信号
严谨地进行滚动前向验证
进行多重假设检验校正
使用时点数据，防止前视偏差

支持 Markdown

登录后评论

基本信息

作者: copyleftdev
仓库: copyleftdev/sk1llz
分类: 数据与数据库
安装次数: 115
GitHub Stars: 2

报告问题