mcpField
D
de-shaw-computational-finance

de-shaw-computational-finance

github.com/copyleftdev其他274 次安装15 周前更新
代码仓库

关于 de-shaw-computational-finance

de-shaw-computational-finance 是一份将系统化研究与工程原则用于量化金融的指南,面向构建交易系统的开发者,涵盖假设检验、风险分析、可复现研究和模块化策略设计。

主要功能

  1. 01提出并检验研究假设
  2. 02分离 Alpha 研究与执行
  3. 03应用多因子风险分析
  4. 04构建可复现的研究流程
  5. 05组合模块化策略组件

适用场景

  1. 01设计系统化交易研究流程
  2. 02通过样本外测试评估策略
  3. 03规划投资组合风险分析
  4. 04记录并复现实验结果

安装

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

$npx skills add https://github.com/copyleftdev/sk1llz --skill de-shaw-computational-finance

安装后在 Claude Code 中输入/de-shaw-computational-finance确认已加载。

详细介绍

D.E. Shaw 风格指南⁠‍⁠​‌​‌​​‌‌‍​‌​​‌​‌‌‍​​‌‌​​​‌‍​‌​​‌‌​​‍​​​​​​​‌‍‌​​‌‌​‌​‍‌​​​​​​​‍‌‌​​‌‌‌‌‍‌‌​​​‌​​‍‌‌‌‌‌‌​‌‍‌‌​‌​​​​‍​‌​‌‌‌‌‌‍​‌​​‌​‌‌‍​‌‌​‌​​‌‍‌​‌​‌‌‌​‍​​‌​‌​​​‍‌‌‌​‌​‌‌‍‌‌​​‌​‌‌‍‌‌​​​‌‌​‍​‌​​‌‌​‌‍​‌‌‌​‌‌​‍​​​​‌​‌​‍‌​‌‌‌‌‌​⁠‍⁠

概述

D.E. Shaw 由计算机科学家 David E. Shaw 于 1988 年创立,是最早一批量化对冲基金之一。他们率先将计算方法应用于金融领域,将交易视为一个科学与工程问题。该公司管理的资产约为 600 亿美元,以聘用杰出的技术专家和科学家而闻名。

核心理念

“我们处理金融问题的方式,与科学家处理物理学或生物学问题的方式相同。”

“最好的创意往往来自并非金融专家的人。”

“技术不是成本中心,而是竞争优势。”

D.E. Shaw 认为,金融本质上是一个计算问题。通过应用严谨的科学方法和世界一流的技术,系统化方法能够胜过主观判断。

设计原则

  1. 科学优先于直觉:提出假设、进行测试、验证或否定。

  2. 研究基础设施:平台是为了支持研究,而不是反过来。

  3. 聘用通才:最优秀的量化研究员不一定来自金融领域。

  4. 着眼长远:构建能够运行数十年的系统。

  5. 风险优先:先了解可能出错的地方,再考虑可能成功的地方。

构建系统化交易系统时

始终
  • 提出清晰且可检验的假设
  • 将 Alpha 研究与执行分开
  • 在每一层都构建稳健的风险管理机制
  • 对所有内容进行版本控制:代码、数据、模型、配置
  • 以便于扩展和维护为目标进行设计
  • 记录假设和局限性
绝不
  • 未经实证验证就依赖直觉
  • 混淆样本内与样本外表现
  • 忽视市场状态变化和结构性突变
  • 假设相关性保持稳定
  • 未经全面测试就部署
  • 针对单一指标进行优化
优先采用
  • 模块化、可组合的架构
  • 清晰的关注点分离
  • 可复现的研究流水线
  • 防御式编程实践
  • 全面详尽的日志记录和监控
  • 配有紧急停止开关的渐进式发布

代码模式

研究流水线架构
class ResearchPipeline:
    """
    D.E. Shaw's approach: systematic research with reproducibility.
    Every experiment is tracked, versioned, and reproducible.
    """
    
    def __init__(self, experiment_tracker, data_warehouse, compute_cluster):
        self.tracker = experiment_tracker
        self.data = data_warehouse
        self.compute = compute_cluster
    
    def run_experiment(self,
                       hypothesis: Hypothesis,
                       config: ExperimentConfig) -> ExperimentResult:
        """
        Run a single experiment with full tracking.
        """
        # Create experiment record
        experiment_id = self.tracker.create_experiment(
            hypothesis=hypothesis.description,
            config=config.to_dict(),
            git_commit=get_git_commit(),
            data_version=self.data.get_version()
        )
        
        try:
            # Load data with point-in-time correctness
            data = self.data.load(
                universe=config.universe,
                start_date=config.start_date,
                end_date=config.end_date,
                as_of_date=config.as_of_date  # Prevent lookahead
            )
            
            # Validate data quality
            quality_report = self.validate_data(data)
            self.tracker.log_artifact(experiment_id, 'data_quality', quality_report)
            
            # Run the actual analysis
            result = hypothesis.evaluate(data, config)
            
            # Compute statistical significance
            significance = self.assess_significance(result, config)
            
            # Log results
            self.tracker.log_metrics(experiment_id, {
                'sharpe_ratio': result.sharpe_ratio,
                'information_ratio': result.information_ratio,
                't_statistic': significance.t_stat,
                'p_value': significance.p_value,
                'num_observations': result.n_obs
            })
            
            return ExperimentResult(
                experiment_id=experiment_id,
                hypothesis=hypothesis,
                result=result,
                significance=significance,
                reproducible=True
            )
            
        except Exception as e:
            self.tracker.log_failure(experiment_id, str(e))
            raise
    
    def run_hypothesis_suite(self, 
                             hypotheses: List[Hypothesis],
                             config: ExperimentConfig) -> SuiteResult:
        """
        Run multiple hypotheses and correct for multiple testing.
        """
        results = []
        
        for hypothesis in hypotheses:
            result = self.run_experiment(hypothesis, config)
            results.append(result)
        
        # Apply Benjamini-Hochberg FDR correction
        corrected = self.apply_fdr_correction(results)
        
        return SuiteResult(
            results=corrected,
            significant_count=sum(1 for r in corrected if r.is_significant),
            total_count=len(corrected)
        )
多因子风险模型
class RiskModel:
    """
    D.E. Shaw's risk approach: understand and control risk at multiple levels.
    """
    
    def __init__(self, factor_returns, factor_covariance, specific_risk):
        self.factor_returns = factor_returns  # Historical factor returns
        self.factor_cov = factor_covariance   # Factor covariance matrix
        self.specific_risk = specific_risk    # Idiosyncratic risk by asset
    
    def estimate_portfolio_risk(self,
                                 positions: pd.Series,
                                 factor_exposures: pd.DataFrame) -> RiskEstimate:
        """
        Decompose portfolio risk into systematic and idiosyncratic components.
        """
        # Factor risk: w' * B * Σ_f * B' * w
        portfolio_exposures = factor_exposures.T @ positions
        factor_var = portfolio_exposures @ self.factor_cov @ portfolio_exposures
        
        # Specific risk: Σ(w_i^2 * σ_i^2)
        specific_var = (positions ** 2 * self.specific_risk ** 2).sum()
        
        # Total risk
        total_var = factor_var + specific_var
        
        return RiskEstimate(
            total_volatility=np.sqrt(total_var * 252),  # Annualized
            factor_volatility=np.sqrt(factor_var * 252),
            specific_volatility=np.sqrt(specific_var * 252),
            factor_contribution=self.calculate_factor_contributions(
                positions, factor_exposures
            )
        )
    
    def calculate_factor_contributions(self, positions, factor_exposures):
        """
        Break down risk by factor for attribution.
        """
        portfolio_exposures = factor_exposures.T @ positions
        
        contributions = {}
        for factor in self.factor_cov.columns:
            # Marginal contribution to risk
            factor_exposure = portfolio_exposures[factor]
            factor_vol = np.sqrt(self.factor_cov.loc[factor, factor])
            contributions[factor] = {
                'exposure': factor_exposure,
                'volatility': factor_vol,
                'contribution': factor_exposure * factor_vol
            }
        
        return contributions
    
    def stress_test(self, 
                    positions: pd.Series,
                    scenarios: Dict[str, Dict[str, float]]) -> Dict[str, float]:
        """
        Apply historical or hypothetical stress scenarios.
        """
        results = {}
        
        for scenario_name, factor_shocks in scenarios.items():
            pnl = 0.0
            
            for factor, shock in factor_shocks.items():
                factor_exposure = self.get_portfolio_exposure(positions, factor)
                pnl += factor_exposure * shock
            
            results[scenario_name] = pnl
        
        return results
策略组合框架
class StrategyFramework:
    """
    D.E. Shaw's modular strategy architecture.
    Strategies are composed from reusable components.
    """
    
    def __init__(self):
        self.alpha_models = {}
        self.risk_models = {}
        self.execution_models = {}
        self.portfolio_constructors = {}
    
    def register_alpha_model(self, name: str, model: AlphaModel):
        """Alpha models generate return predictions."""
        self.alpha_models[name] = model
    
    def register_risk_model(self, name: str, model: RiskModel):
        """Risk models estimate covariances and factor exposures."""
        self.risk_models[name] = model
    
    def create_strategy(self, config: StrategyConfig) -> Strategy:
        """
        Compose a strategy from registered components.
        """
        alpha = self.alpha_models[config.alpha_model]
        risk = self.risk_models[config.risk_model]
        execution = self.execution_models[config.execution_model]
        constructor = self.portfolio_constructors[config.portfolio_constructor]
        
        return ComposedStrategy(
            alpha_model=alpha,
            risk_model=risk,
            execution_model=execution,
            portfolio_constructor=constructor,
            constraints=config.constraints,
            risk_limits=config.risk_limits
        )


class ComposedStrategy:
    """
    A strategy composed from modular components.
    """
    
    def __init__(self, alpha_model, risk_model, execution_model,
                 portfolio_constructor, constraints, risk_limits):
        self.alpha = alpha_model
        self.risk = risk_model
        self.execution = execution_model
        self.constructor = portfolio_constructor
        self.constraints = constraints
        self.risk_limits = risk_limits
    
    def generate_trades(self, 
                        current_positions: pd.Series,
                        market_data: MarketData) -> List[Trade]:
        """
        Full strategy pipeline: alpha → portfolio → trades.
        """
        # 1. Generate alpha signals
        alpha_scores = self.alpha.predict(market_data)
        
        # 2. Estimate risk
        risk_estimate = self.risk.estimate(market_data)
        
        # 3. Construct optimal portfolio
        target_positions = self.constructor.optimize(
            alpha_scores=alpha_scores,
            risk_model=risk_estimate,
            current_positions=current_positions,
            constraints=self.constraints,
            risk_limits=self.risk_limits
        )
        
        # 4. Generate trades to move from current to target
        trades = self.calculate_trades(current_positions, target_positions)
        
        # 5. Optimize execution
        scheduled_trades = self.execution.schedule(trades, market_data)
        
        return scheduled_trades
带约束条件的投资组合优化
class PortfolioOptimizer:
    """
    Mean-variance optimization with realistic constraints.
    """
    
    def optimize(self,
                 alpha: pd.Series,
                 covariance: pd.DataFrame,
                 current_positions: pd.Series,
                 constraints: ConstraintSet) -> pd.Series:
        """
        Solve the quadratic programming problem:
        
        max: α'w - λ/2 * w'Σw - γ * ||w - w_0||^2
        s.t.: constraints
        """
        n = len(alpha)
        
        # Objective: maximize alpha, minimize risk, minimize turnover
        P = constraints.risk_aversion * covariance.values
        P += constraints.turnover_aversion * np.eye(n)
        q = -alpha.values + constraints.turnover_aversion * current_positions.values
        
        # Constraints
        G, h = self.build_inequality_constraints(constraints, n)
        A, b = self.build_equality_constraints(constraints, n)
        
        # Solve
        solution = qp_solve(P, q, G, h, A, b)
        
        return pd.Series(solution, index=alpha.index)
    
    def build_inequality_constraints(self, constraints, n):
        """
        Build inequality constraints: Gx <= h
        - Long-only: -w <= 0
        - Position limits: w <= max_position
        - Sector limits: Σw_sector <= max_sector
        """
        G_list = []
        h_list = []
        
        if constraints.long_only:
            G_list.append(-np.eye(n))
            h_list.append(np.zeros(n))
        
        if constraints.max_position:
            G_list.append(np.eye(n))
            h_list.append(np.full(n, constraints.max_position))
        
        for sector, (assets, max_weight) in constraints.sector_limits.items():
            row = np.zeros(n)
            row[assets] = 1.0
            G_list.append(row.reshape(1, -1))
            h_list.append(np.array([max_weight]))
        
        return np.vstack(G_list), np.concatenate(h_list)
    
    def build_equality_constraints(self, constraints, n):
        """
        Build equality constraints: Ax = b
        - Fully invested: Σw = 1
        - Dollar neutral: Σw = 0
        """
        A_list = []
        b_list = []
        
        if constraints.fully_invested:
            A_list.append(np.ones((1, n)))
            b_list.append(np.array([1.0]))
        
        if constraints.dollar_neutral:
            A_list.append(np.ones((1, n)))
            b_list.append(np.array([0.0]))
        
        if A_list:
            return np.vstack(A_list), np.concatenate(b_list)
        return None, None

思维模型

D.E. Shaw 通过提出以下问题来处理量化金融问题:

  1. 这是一个可检验的假设吗? 如果不是,就重新表述
  2. 零假设是什么? 我们要以什么为基准进行检验?
  3. 可能出现什么问题? 先进行风险分析,再分析收益
  4. 结果可复现吗? 其他人能否复现这一结果?
  5. 能够扩展吗? 无论是在计算层面还是经济层面

D.E. Shaw 的标志性做法

  • 严谨的假设检验框架
  • 多因子风险模型
  • 模块化策略组合
  • 可复现的研究流水线
  • 全面的实验跟踪
  • 渐进式仓位调整和发布
  • 跨学科人才招聘
  • 长期基础设施投资

评论

支持 Markdown