mcpField
A
aqr-factor-investing

aqr-factor-investing

github.com/copyleftdev数据与数据库2119 次安装15 周前更新
代码仓库

关于 aqr-factor-investing

面向量化投资研究与策略开发者的因子投资技能,提供以学术证据、经济逻辑和现实交易成本为核心的策略设计原则与代码模式。

主要功能

  1. 01基于学术研究筛选因子
  2. 02构建复合因子与多因子组合
  3. 03纳入交易成本与流动性约束
  4. 04进行因子归因和透明报告
  5. 05强调跨时期与地域测试

适用场景

  1. 01设计有经济依据的因子策略
  2. 02评估策略扣除交易成本后的表现
  3. 03构建分散化多因子组合

安装

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

$npx skills add https://github.com/copyleftdev/sk1llz --skill aqr-factor-investing

安装后在 Claude Code 中输入/aqr-factor-investing确认已加载。

详细介绍

AQR Capital Management 风格指南⁠‍⁠​‌​‌​​‌‌‍​‌​​‌​‌‌‍​​‌‌​​​‌‍​‌​​‌‌​​‍​​​​​​​‌‍‌​​‌‌​‌​‍‌​​​​​​​‍‌‌​​‌‌‌‌‍‌‌​​​‌​​‍‌‌‌‌‌‌​‌‍‌‌​‌​​​​‍​‌​‌‌‌‌‌‍​‌​​‌​‌‌‍​‌‌​‌​​‌‍‌​‌​‌‌‌​‍​​‌​‌​​​‍‌‌‌​‌​‌‌‍‌​‌‌‌‌​‌‍​​​‌‌‌​​‍‌​‌​​​​​‍‌‌​‌​‌​‌‍​​​​‌​‌​‍‌​‌‌‌​​​⁠‍⁠

概述

AQR(Applied Quantitative Research)由 Cliff Asness 和其他来自 Goldman Sachs 的学者创立,是一家管理约 1,000 亿美元资产的量化投资公司。AQR 以将学术因子研究应用于实际投资而闻名,强调透明度、严谨的方法论,以及量化技术的普及。

核心理念

“金融领域最好的创意源自严谨的学术研究,而非华尔街直觉。”

“因子之所以有效,是因为风险、行为或结构——投资前要弄清楚是哪一种。”

“如果你无法简单地解释它,就说明你还没有充分理解它。”

AQR 认为,系统性因子(价值、动量、质量等)代表着持久的收益来源,可以通过有纪律的实施加以获取。AQR 强调要理解策略为何有效,而不只是知道它们有效。

设计原则

  1. 学术基础:从同行评审的研究入手。

  2. 因子纪律:坚持采用具有经济学依据的因子。

  3. 透明度:公布方法论,承认错误。

  4. 多元化:跨因子、地域和资产类别进行分散投资。

  5. 实施至关重要:交易成本可能会侵蚀纸面收益。

构建因子策略时

始终
  • 以学术研究为策略奠定基础
  • 理解经济学依据(风险、行为或结构)
  • 在多个时期和地域进行测试
  • 纳入切合实际的交易成本
  • 结合多个因子以实现多元化
  • 构建达到投资级标准的因子(流动性、容量)
绝不
  • 追逐通过数据挖掘发现的因子
  • 忽视实施差距(纸面收益与实际收益)
  • 假设因子溢价随时间保持稳定
  • 集中投资于单一因子或市场
  • 忘记因子拥挤问题
  • 进行不必要的交易
优先选择
  • 复合因子,而非单一指标
  • 多空策略,而非纯多头策略,以获得纯粹的因子敞口
  • 等风险加权,而非等金额加权
  • 渐进式再平衡,而非离散交易
  • 考虑交易成本的优化
  • 对因子择时保持审慎

代码模式

因子构建
class FactorBuilder:
    """
    AQR-style factor construction: robust, diversified, investment-grade.
    """
    
    def __init__(self, data_provider):
        self.data = data_provider
    
    def build_value_factor(self, 
                           universe: List[str],
                           date: date) -> pd.Series:
        """
        Value factor: composite of multiple value metrics.
        AQR uses book/price, earnings/price, forecast earnings/price, etc.
        """
        metrics = {}
        
        # Book to Price (classic Fama-French)
        metrics['book_to_price'] = self.data.get_fundamentals(
            universe, 'book_value', date
        ) / self.data.get_prices(universe, date)
        
        # Earnings to Price
        metrics['earnings_to_price'] = self.data.get_fundamentals(
            universe, 'trailing_earnings', date
        ) / self.data.get_prices(universe, date)
        
        # Forward Earnings to Price (analyst estimates)
        metrics['forward_ep'] = self.data.get_fundamentals(
            universe, 'forward_earnings', date
        ) / self.data.get_prices(universe, date)
        
        # Cash Flow to Price
        metrics['cf_to_price'] = self.data.get_fundamentals(
            universe, 'operating_cf', date
        ) / self.data.get_prices(universe, date)
        
        # Composite: z-score and average
        composite = pd.DataFrame(metrics)
        z_scores = composite.apply(lambda x: self.winsorize_and_zscore(x), axis=0)
        
        return z_scores.mean(axis=1)
    
    def build_momentum_factor(self,
                               universe: List[str],
                               date: date) -> pd.Series:
        """
        Momentum: 12-month return, skipping most recent month.
        Classic Jegadeesh-Titman with AQR refinements.
        """
        # 12-1 momentum (skip last month to avoid reversal)
        prices = self.data.get_price_history(universe, date, lookback_months=13)
        
        # Return from t-12 to t-1
        momentum_12_1 = prices.iloc[-22] / prices.iloc[0] - 1  # Skip last month
        
        # AQR enhancement: also consider intermediate momentum
        momentum_6_1 = prices.iloc[-22] / prices.iloc[-132] - 1
        
        # Industry-adjusted (avoid sector bets)
        industries = self.data.get_industries(universe)
        mom_adj = momentum_12_1.groupby(industries).transform(
            lambda x: x - x.mean()
        )
        
        return self.winsorize_and_zscore(mom_adj)
    
    def build_quality_factor(self,
                              universe: List[str],
                              date: date) -> pd.Series:
        """
        Quality: profitability, stability, and financial health.
        Based on AQR's "Quality Minus Junk" research.
        """
        profitability = self.calculate_profitability(universe, date)
        growth = self.calculate_growth_stability(universe, date)
        safety = self.calculate_safety(universe, date)
        payout = self.calculate_payout(universe, date)
        
        # Composite quality score
        quality = pd.DataFrame({
            'profitability': self.winsorize_and_zscore(profitability),
            'growth': self.winsorize_and_zscore(growth),
            'safety': self.winsorize_and_zscore(safety),
            'payout': self.winsorize_and_zscore(payout)
        })
        
        return quality.mean(axis=1)
    
    def calculate_profitability(self, universe, date):
        """Gross profits / assets, ROE, ROA, etc."""
        gp = self.data.get_fundamentals(universe, 'gross_profit', date)
        assets = self.data.get_fundamentals(universe, 'total_assets', date)
        return gp / assets
    
    def calculate_safety(self, universe, date):
        """Low leverage, low volatility, low beta."""
        leverage = self.data.get_fundamentals(universe, 'debt_to_equity', date)
        volatility = self.data.get_volatility(universe, date, lookback_days=252)
        
        # Invert so higher is better
        return -(leverage.rank() + volatility.rank()) / 2
    
    def winsorize_and_zscore(self, series: pd.Series, clip_std: float = 3.0):
        """Winsorize outliers and standardize."""
        z = (series - series.mean()) / series.std()
        z = z.clip(-clip_std, clip_std)
        return (z - z.mean()) / z.std()
多因子投资组合构建
class FactorPortfolio:
    """
    AQR's portfolio construction: factor exposure with risk management.
    """
    
    def __init__(self, factors: Dict[str, FactorBuilder], 
                 risk_model: RiskModel,
                 transaction_cost_model: TCostModel):
        self.factors = factors
        self.risk = risk_model
        self.tcost = transaction_cost_model
    
    def construct_portfolio(self,
                            universe: List[str],
                            date: date,
                            factor_weights: Dict[str, float],
                            risk_target: float = 0.10) -> pd.Series:
        """
        Build a portfolio with target factor exposures.
        """
        # Calculate factor scores
        factor_scores = {}
        for name, builder in self.factors.items():
            factor_scores[name] = builder.build(universe, date)
        
        # Combine factors with weights
        combined_score = sum(
            factor_scores[name] * weight 
            for name, weight in factor_weights.items()
        )
        
        # Convert scores to weights (long-short)
        raw_weights = self.scores_to_weights(combined_score)
        
        # Scale to target risk
        portfolio_vol = self.risk.estimate_volatility(raw_weights)
        scaled_weights = raw_weights * (risk_target / portfolio_vol)
        
        return scaled_weights
    
    def scores_to_weights(self, scores: pd.Series) -> pd.Series:
        """
        Convert z-scores to portfolio weights.
        AQR approach: proportional to score, with constraints.
        """
        # Long top tercile, short bottom tercile
        n = len(scores)
        tercile = n // 3
        
        sorted_idx = scores.sort_values().index
        
        weights = pd.Series(0.0, index=scores.index)
        weights[sorted_idx[:tercile]] = -1.0 / tercile  # Short bottom
        weights[sorted_idx[-tercile:]] = 1.0 / tercile   # Long top
        
        return weights
    
    def calculate_turnover_cost(self,
                                 current: pd.Series,
                                 target: pd.Series,
                                 date: date) -> float:
        """
        Estimate transaction costs from rebalancing.
        """
        trades = (target - current).abs()
        costs = self.tcost.estimate(trades, date)
        return costs.sum()
    
    def optimize_with_turnover(self,
                                current: pd.Series,
                                target: pd.Series,
                                max_turnover_cost: float) -> pd.Series:
        """
        Trade toward target, but respect turnover budget.
        """
        trades = target - current
        
        # If unconstrained cost is acceptable, trade fully
        full_cost = self.calculate_turnover_cost(current, target, date)
        if full_cost <= max_turnover_cost:
            return target
        
        # Otherwise, trade partially (proportionally)
        trade_fraction = max_turnover_cost / full_cost
        return current + trades * trade_fraction
因子归因与报告
class FactorAttribution:
    """
    AQR-style transparent performance attribution.
    Understand exactly where returns came from.
    """
    
    def __init__(self, factor_returns: pd.DataFrame):
        self.factor_returns = factor_returns
    
    def attribute_returns(self,
                          portfolio_returns: pd.Series,
                          factor_exposures: pd.DataFrame) -> AttributionResult:
        """
        Decompose portfolio returns into factor contributions.
        
        R_p = Σ(β_i * F_i) + α + ε
        """
        # Align data
        common_dates = portfolio_returns.index.intersection(
            self.factor_returns.index
        )
        
        port_ret = portfolio_returns.loc[common_dates]
        fact_ret = self.factor_returns.loc[common_dates]
        exposures = factor_exposures.loc[common_dates]
        
        # Calculate factor contributions
        contributions = {}
        total_factor_return = 0
        
        for factor in fact_ret.columns:
            factor_contribution = (exposures[factor] * fact_ret[factor]).sum()
            contributions[factor] = {
                'avg_exposure': exposures[factor].mean(),
                'factor_return': fact_ret[factor].sum(),
                'contribution': factor_contribution,
                'contribution_pct': factor_contribution / port_ret.sum() * 100
            }
            total_factor_return += factor_contribution
        
        # Alpha is unexplained return
        alpha = port_ret.sum() - total_factor_return
        
        return AttributionResult(
            total_return=port_ret.sum(),
            factor_contributions=contributions,
            alpha=alpha,
            r_squared=self.calculate_r_squared(port_ret, fact_ret, exposures)
        )
    
    def factor_performance_report(self,
                                   start_date: date,
                                   end_date: date) -> pd.DataFrame:
        """
        Generate factor performance summary.
        AQR publishes these regularly for transparency.
        """
        returns = self.factor_returns.loc[start_date:end_date]
        
        report = pd.DataFrame({
            'Total Return': returns.sum(),
            'Annualized Return': returns.mean() * 252,
            'Volatility': returns.std() * np.sqrt(252),
            'Sharpe Ratio': returns.mean() / returns.std() * np.sqrt(252),
            'Max Drawdown': self.calculate_max_drawdown(returns),
            'Hit Rate': (returns > 0).mean()
        })
        
        return report
纳入实际摩擦因素的回测
class RealisticBacktest:
    """
    AQR emphasizes the gap between paper and real returns.
    Model all frictions realistically.
    """
    
    def __init__(self, 
                 tcost_model: TransactionCostModel,
                 borrow_cost_model: BorrowCostModel,
                 market_impact_model: MarketImpactModel):
        self.tcost = tcost_model
        self.borrow = borrow_cost_model
        self.impact = market_impact_model
    
    def run_backtest(self,
                     strategy: Strategy,
                     start_date: date,
                     end_date: date,
                     initial_capital: float = 1e8) -> BacktestResult:
        """
        Backtest with realistic transaction costs and frictions.
        """
        capital = initial_capital
        positions = pd.Series(dtype=float)
        
        results = []
        
        for date in trading_days(start_date, end_date):
            # Generate target portfolio
            target = strategy.generate_positions(date, capital)
            
            # Calculate trading costs
            trades = target - positions
            
            trading_cost = self.tcost.estimate(trades, date)
            market_impact = self.impact.estimate(trades, date)
            
            # Borrow costs for short positions
            short_positions = positions[positions < 0]
            borrow_cost = self.borrow.estimate(short_positions, date)
            
            # Execute trades (adjust for costs)
            capital -= trading_cost + market_impact
            positions = target
            
            # Calculate return
            price_returns = self.get_returns(positions.index, date)
            gross_pnl = (positions * price_returns).sum()
            
            net_pnl = gross_pnl - trading_cost - market_impact - borrow_cost
            capital += net_pnl
            
            results.append({
                'date': date,
                'gross_pnl': gross_pnl,
                'trading_cost': trading_cost,
                'market_impact': market_impact,
                'borrow_cost': borrow_cost,
                'net_pnl': net_pnl,
                'capital': capital,
                'turnover': trades.abs().sum() / capital
            })
        
        return self.analyze_results(pd.DataFrame(results))
    
    def analyze_results(self, results: pd.DataFrame) -> BacktestResult:
        """Compute performance metrics with cost breakdown."""
        gross_returns = results['gross_pnl'] / results['capital'].shift(1)
        net_returns = results['net_pnl'] / results['capital'].shift(1)
        
        return BacktestResult(
            gross_sharpe=gross_returns.mean() / gross_returns.std() * np.sqrt(252),
            net_sharpe=net_returns.mean() / net_returns.std() * np.sqrt(252),
            implementation_drag=(gross_returns.sum() - net_returns.sum()) / len(results) * 252,
            avg_turnover=results['turnover'].mean(),
            total_trading_costs=results['trading_cost'].sum(),
            total_impact_costs=results['market_impact'].sum(),
            total_borrow_costs=results['borrow_cost'].sum()
        )

思维模型

AQR 通过以下问题来进行因子投资:

  1. 是否有学术证据? 同行评审的研究,而非营销宣传
  2. 经济学逻辑是什么? 风险溢价、行为偏差,还是结构性因素?
  3. 扣除交易成本后是否仍然有效? 纸面收益 ≠ 实际收益
  4. 是否存在拥挤? 因子的受欢迎程度会侵蚀收益
  5. 能否大规模实施? 流动性和容量限制

AQR 的标志性做法

  • 采用复合因子,而非单一指标
  • 遵循高学术标准的研究流程
  • 采用透明的方法论
  • 对交易成本进行切合实际的建模
  • 跨多种资产类别进行多元化投资
  • 对因子择时保持审慎
  • 通过多空策略获得纯粹的因子敞口
  • 公布因子收益,供基准比较

评论

支持 Markdown