策略容量估算：你的策略能承载多少资金

“回测是过去，容量是未来。”

2019 年，一家管理 2000 万美元量化基金的团队，凭借一套年化 47% 的策略登上了业内论坛。但到 2020 年底，他们的旗舰产品清盘了——不是因为策略失效，而是因为规模从 2000 万膨胀到 1.2 亿美元后，冲击成本吃掉了全部利润。

这不是个例。大多数量化新手在回测报告中看到漂亮的夏普比率时，从未问过一个问题：这套策略能承载多少资金？

策略容量（Strategy Capacity），指的是在给定的市场条件下，一个策略能够有效运行而不显著影响市场的最大资金规模。超过这个上限，策略的预期收益会被交易成本吞噬，最终变成“市场impact”的游戏。

这篇文章，我们不聊玄学，聊聊工程化的容量估算方法论。

---

一、为什么回测结果会骗你

回测系统的运作逻辑是：假设你的订单对市场价格没有影响。每一笔买入都按照回测引擎“看到的”历史价格成交。

但现实是残酷的。当你用 500 万资金买入一只日均成交额 2000 万的小盘股时，你的成交额占日成交量的 25%。这时：

• 冲击成本不再是回测里的“0.1% 滑点”
• 市场深度在几个价位内被耗尽
• 价格开始因为你的买入而上涨

这就是为什么很多回测看起来完美，实盘却亏钱的根本原因。回测计算的是 alpha，但忽略了 market impact。

真实收益 = 回测收益 - 冲击成本 - 滑点 - 佣金 - 资金成本

当资金规模超过容量上限时，这个等式右边的成本项会指数级膨胀，最终吃掉所有收益。

---

二、策略容量的三个核心约束

估算策略容量，本质上是量化三个约束条件：

2.1 成交量约束

最直观的约束来自流动性。任何一个策略的资金容量，不能超过标的在目标时间窗口内的可交易量。

成交量约束 = 日均成交量(ADV) × 持仓周期系数 × 流动性安全系数

参数	含义	建议值
持仓周期系数	策略平均持仓天数对应的成交量占比	建议不超过单日 ADV 的 5%-10%
流动性安全系数	极端行情下的缓冲	0.3-0.5（保守取 0.3）

2.2 冲击成本约束

冲击成本（Market Impact）是订单对价格的直接影响。一个常用的简化模型是：

冲击成本(%) = λ × (订单量 / ADV)^β

其中：

• λ（Lambda）：市场冲击系数，跟标的的流动性直接相关
• β（Beta）：规模指数，通常在 0.5-0.8 之间（小盘股更接近 1.0）

对于美股大盘股（苹果、微软等），λ ≈ 0.1，β ≈ 0.5；对于小盘股，λ 可高达 1.0 以上。

2.3 滑点约束

滑点（Slippage）是成交价与预期价的差值，包含两部分：

1. 临时冲击：订单立即影响的价格偏移
2. 永久冲击：订单暴露后，其他参与者反应导致的价格偏移

滑点的合理估算需要结合冲击成本模型，在下一节我们会给出具体实现。

---

三、冲击成本模型与容量估算

3.1 构建冲击成本模型

一个实用的容量估算框架，包含以下步骤：

Step 1: 确定策略参数（持仓周期、换手率、日均交易次数）
Step 2: 估算单笔订单量（资金规模 / 持仓标的数 / 换手周期）
Step 3: 计算冲击成本（使用 Almgren-Chriss 模型或简化公式）
Step 4: 叠加滑点和佣金，计算净收益
Step 5: 迭代不同资金规模，找到收益归零的临界点

3.2 Almgren-Chriss 模型简介

这是量化领域最经典的冲击成本模型，核心思想是：执行时间越长，冲击成本越低，但时间风险越高。最优执行策略是二者的权衡。

模型假设：

• 永久冲击：γ × v（v 是交易速率）
• 临时冲击：η × v
• 时间风险：σ² × T（方差随时间累积）

最优执行价格与“立即执行”的价差为：

TC = (γ + η) × v × X - γ × X × v̅

其中 X 是总订单量，v̅ 是平均执行速率。

对于大多数个人投资者，直接使用简化模型足够有效。

---

四、生产级容量估算工具

以下代码实现了一个完整的策略容量估算模块，包含：

• 流动性数据获取（支持 TickDB depth 频道获取订单簿深度）
• 冲击成本计算
• 容量边界迭代求解

"""
策略容量估算模块
估算给定策略参数下的最大可承载资金规模

依赖：pip install requests pandas numpy
"""

import os
import time
import json
import random
import hashlib
from typing import Optional, Dict, List, Tuple
from dataclasses import dataclass
from datetime import datetime
import requests


@dataclass
class CapacityResult:
    """容量估算结果"""
    max_capital: float
    max_shares: int
    estimated_slippage_bps: float  # basis points
    estimated_impact_pct: float
    net_expected_return_pct: float
    liquidity_score: str  # "high" / "medium" / "low"


class StrategyCapacityEstimator:
    """
    策略容量估算器
    
    使用简化冲击成本模型估算策略的最大容量。
    模型公式参考 Almgren-Chriss (2000) 的市场冲击框架。
    """
    
    # 冲击系数默认值（可通过实际数据校准）
    LAMBDA_VOLATILE = 1.2   # 小盘股/低流动性
    LAMBDA_NORMAL = 0.5     # 中盘股
    LAMBDA_LIQUID = 0.1     # 大盘蓝筹
    
    # 佣金和滑点基础值（美元/股）
    COMMISSION_PER_SHARE = 0.0035
    BASE_SLIPPAGE_BPS = 1.0  # 基础滑点（basis points）
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        self.session = requests.Session()
        self.session.headers.update({"X-API-Key": self.api_key or ""})
        
        # 重连配置
        self.max_retries = 3
        self.base_delay = 1.0
        self.max_delay = 30.0
        
        # ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio
        # 本模块为离线批量计算，低频使用场景使用 requests
    
    def _make_request(
        self, 
        method: str, 
        endpoint: str, 
        params: Optional[Dict] = None,
        data: Optional[Dict] = None
    ) -> Dict:
        """
        统一请求方法，包含重试和限频处理
        
        Args:
            method: HTTP 方法
            endpoint: API 端点
            params: URL 参数
            data: 请求体数据
        
        Returns:
            API 响应数据
        """
        url = f"https://api.tickdb.ai{endpoint}"
        
        for retry in range(self.max_retries):
            try:
                if method.upper() == "GET":
                    response = self.session.get(
                        url, 
                        params=params,
                        timeout=(3.05, 10)
                    )
                else:
                    response = self.session.post(
                        url, 
                        json=data,
                        timeout=(3.05, 10)
                    )
                
                result = response.json()
                
                # 限频处理（code: 3001）
                if result.get("code") == 3001:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    print(f"[限频] 等待 {retry_after} 秒后重试...")
                    time.sleep(retry_after)
                    continue
                
                if result.get("code") == 0:
                    return result.get("data", {})
                
                # 其他错误
                raise RuntimeError(f"API 错误: {result.get('message')}")
                
            except requests.exceptions.Timeout:
                delay = min(self.base_delay * (2 ** retry), self.max_delay)
                jitter = random.uniform(0, delay * 0.1)
                print(f"[超时] {delay + jitter:.1f}s 后重试 ({retry + 1}/{self.max_retries})")
                time.sleep(delay + jitter)
                
            except requests.exceptions.RequestException as e:
                delay = min(self.base_delay * (2 ** retry), self.max_delay)
                jitter = random.uniform(0, delay * 0.1)
                print(f"[网络错误] {e}, {delay + jitter:.1f}s 后重试...")
                time.sleep(delay + jitter)
        
        raise RuntimeError(f"请求失败，已重试 {self.max_retries} 次")
    
    def get_market_depth(self, symbol: str, levels: int = 5) -> Optional[Dict]:
        """
        获取订单簿深度数据
        
        使用 TickDB depth 频道获取实时/历史订单簿快照。
        深度数据用于计算当前市场流动性和买卖盘压力。
        
        Args:
            symbol: 交易品种，如 "AAPL.US"
            levels: 深度档位数
        
        Returns:
            订单簿深度数据，包含 bid/ask 价格和数量
        """
        return self._make_request(
            "GET",
            "/v1/market/depth",
            params={"symbol": symbol, "limit": levels}
        )
    
    def get_historical_kline(
        self, 
        symbol: str, 
        interval: str = "1d",
        limit: int = 30
    ) -> Optional[List[Dict]]:
        """
        获取历史 K 线数据
        
        用于计算日均成交量（ADV）和波动率。
        
        Args:
            symbol: 交易品种
            interval: K 线周期（1m/5m/1h/1d）
            limit: 数据条数
        
        Returns:
            K 线数据列表
        """
        return self._make_request(
            "GET",
            "/v1/market/kline",
            params={"symbol": symbol, "interval": interval, "limit": limit}
        )
    
    def estimate_impact_coefficient(self, symbol: str) -> float:
        """
        根据标的特性估算冲击系数 Lambda
        
        实际生产中应使用 TickDB depth 数据结合历史成交数据校准。
        此处使用简化规则：
        - 大盘股（日均成交额 > 10亿美元）：0.1
        - 中盘股（日均成交额 1-10亿美元）：0.4
        - 小盘股（日均成交额 < 1亿美元）：1.0
        
        ⚠️ 注意：此为简化估算，真实场景需结合订单簿数据和市场微观结构
        """
        try:
            klines = self.get_historical_kline(symbol, "1d", 20)
            if not klines:
                return self.LAMBDA_NORMAL
            
            # 简单估算：使用最近 20 日的平均成交量和收盘价
            avg_volume = sum(k["volume"] for k in klines) / len(klines)
            avg_close = sum(k["close"] for k in klines) / len(klines)
            avg_dollar_volume = avg_volume * avg_close
            
            if avg_dollar_volume > 1_000_000_000:
                return self.LAMBDA_LIQUID
            elif avg_dollar_volume > 100_000_000:
                return self.LAMBDA_NORMAL
            else:
                return self.LAMBDA_VOLATILE
                
        except Exception as e:
            print(f"[警告] 无法获取 {symbol} 数据，使用默认冲击系数: {e}")
            return self.LAMBDA_NORMAL
    
    def calculate_single_trade_impact(
        self,
        order_value: float,
        adv: float,
        beta: float = 0.6,
        liquidity: float = 0.5
    ) -> Tuple[float, float]:
        """
        计算单笔交易的冲击成本
        
        使用简化 Almgren-Chriss 模型：
        Impact(%) = Lambda × (OrderValue / ADV)^Beta
        
        Args:
            order_value: 订单金额（美元）
            adv: 日均成交额（美元）
            beta: 规模指数（0.5-0.8，越小盘越接近 1.0）
            liquidity: 流动性系数（0-1，1 为完全流动）
        
        Returns:
            (impact_pct, slippage_bps)
        """
        if adv <= 0:
            return 0.0, 0.0
        
        lambda_eff = self.LAMBDA_NORMAL * (2 - liquidity)  # 流动性越低，冲击越高
        
        # 冲击成本百分比
        participation_ratio = order_value / adv
        impact_pct = lambda_eff * (participation_ratio ** beta)
        
        # 滑点（basis points），包含临时和永久冲击
        slippage_bps = impact_pct * 100 * 100  # 转换为 bp
        
        return impact_pct, slippage_bps
    
    def estimate_strategy_capacity(
        self,
        strategy_name: str,
        symbols: List[str],
        daily_turnover: float,  # 日换手率（如 0.2 表示每天换手 20%）
        backtest_return_pct: float,  # 年化回测收益率（%）
        backtest_volatility_pct: float,  # 年化波动率（%）
        capital_range: Tuple[float, float] = (10_000, 10_000_000),  # 资金范围测试
        target_return_pct: float = 5.0  # 目标净收益率（低于此值认为策略失效）
    ) -> CapacityResult:
        """
        估算策略的最大容量
        
        核心逻辑：
        1. 遍历不同资金规模
        2. 计算每个标的的单笔订单量
        3. 计算冲击成本和滑点
        4. 计算年化净收益（扣除成本后）
        5. 找到净收益刚好达到目标收益率的临界点
        
        Args:
            strategy_name: 策略名称
            symbols: 持仓标的列表
            daily_turnover: 日换手率（单边）
            backtest_return_pct: 年化回测收益率（%）
            backtest_volatility_pct: 年化波动率（%）
            capital_range: 资金测试范围
            target_return_pct: 目标净收益率阈值
        
        Returns:
            CapacityResult 对象，包含容量上限和各成本项
        """
        print(f"\n{'='*60}")
        print(f"策略容量估算: {strategy_name}")
        print(f"{'='*60}")
        print(f"标的数量: {len(symbols)}")
        print(f"日换手率: {daily_turnover:.1%}")
        print(f"回测年化收益: {backtest_return_pct:.1f}%")
        print(f"回测年化波动率: {backtest_volatility_pct:.1f}%")
        
        # 获取各标的的流动性参数
        liquidity_data = []
        total_adv = 0.0
        
        for symbol in symbols:
            klines = self.get_historical_kline(symbol, "1d", 20)
            if klines:
                avg_vol = sum(k["volume"] for k in klines) / len(klines)
                avg_close = sum(k["close"] for k in klines) / len(klines)
                adv = avg_vol * avg_close
                lambda_coef = self.estimate_impact_coefficient(symbol)
                
                liquidity_data.append({
                    "symbol": symbol,
                    "adv": adv,
                    "lambda": lambda_coef,
                    "avg_price": avg_close,
                    "avg_volume": avg_vol
                })
                total_adv += adv
                
                print(f"  {symbol}: ADV=${adv:,.0f}, λ={lambda_coef:.2f}")
        
        if not liquidity_data:
            raise ValueError("无法获取任何标的的流动性数据")
        
        # 迭代估算容量
        best_capacity = 0.0
        best_result = None
        
        # 资金规模网格搜索（从高到低）
        capital_steps = [i * 100_000 for i in range(1, 101)]  # 10万-1000万
        capital_steps += [i * 1_000_000 for i in range(11, 101)]  # 1100万-1亿
        
        for capital in capital_steps:
            if capital < capital_range[0] or capital > capital_range[1]:
                continue
            
            # 计算单日交易额
            daily_trade_value = capital * daily_turnover
            
            # 按资金比例分配到各标的
            per_symbol_trade = daily_trade_value / len(symbols)
            per_symbol_capital = capital / len(symbols)
            
            total_impact = 0.0
            total_slippage_bps = 0.0
            valid = True
            
            for ld in liquidity_data:
                # 单笔订单金额（假设在一天内均匀下单）
                order_value = per_symbol_trade
                adv = ld["adv"]
                
                impact_pct, slippage = self.calculate_single_trade_impact(
                    order_value=order_value,
                    adv=adv,
                    beta=0.6,
                    liquidity=min(1.0, adv / 100_000_000)
                )
                
                total_impact += impact_pct
                total_slippage_bps += slippage
            
            # 平均冲击和滑点
            avg_impact = total_impact / len(liquidity_data)
            avg_slippage = total_slippage_bps / len(liquidity_data)
            
            # 计算年化成本（基于换手率）
            trading_days = 252
            annual_turnover = daily_turnover * trading_days
            
            # 佣金成本（双边）
            commission_cost = annual_turnover * capital * self.COMMISSION_PER_SHARE / 50  # 简化估算
            
            # 冲击成本
            impact_cost = avg_impact * annual_turnover * 100  # 百分比转元
            
            # 滑点成本
            slippage_cost = (avg_slippage / 10000) * annual_turnover * capital
            
            # 总成本
            total_cost = commission_cost + impact_cost + slippage_cost
            
            # 净收益（简化）
            gross_return = capital * (backtest_return_pct / 100)
            net_return = gross_return - total_cost
            net_return_pct = (net_return / capital) * 100
            
            if net_return_pct >= target_return_pct:
                best_capacity = capital
                best_result = {
                    "gross_return": gross_return,
                    "total_cost": total_cost,
                    "commission_cost": commission_cost,
                    "impact_cost": impact_cost,
                    "slippage_cost": slippage_cost,
                    "net_return": net_return,
                    "net_return_pct": net_return_pct,
                    "avg_impact_pct": avg_impact * 100,
                    "avg_slippage_bps": avg_slippage
                }
        
        if best_result is None:
            return CapacityResult(
                max_capital=0,
                max_shares=0,
                estimated_slippage_bps=0,
                estimated_impact_pct=0,
                net_expected_return_pct=0,
                liquidity_score="low"
            )
        
        # 计算流动性评分
        avg_adv = total_adv / len(symbols)
        if avg_adv > 500_000_000:
            liquidity_score = "high"
        elif avg_adv > 50_000_000:
            liquidity_score = "medium"
        else:
            liquidity_score = "low"
        
        print(f"\n{'='*60}")
        print(f"容量估算结果")
        print(f"{'='*60}")
        print(f"最大容量: ${best_capacity:,.0f}")
        print(f"流动性评级: {liquidity_score}")
        print(f"预估滑点: {best_result['avg_slippage_bps']:.2f} bps")
        print(f"预估冲击成本: {best_result['avg_impact_pct']:.3f}%")
        print(f"\n成本分解:")
        print(f"  佣金: ${best_result['commission_cost']:,.0f}/年")
        print(f"  冲击成本: ${best_result['impact_cost']:,.0f}/年")
        print(f"  滑点: ${best_result['slippage_cost']:,.0f}/年")
        print(f"  总成本: ${best_result['total_cost']:,.0f}/年")
        print(f"\n收益分解:")
        print(f"  回测收益: ${best_result['gross_return']:,.0f}/年")
        print(f"  净收益: ${best_result['net_return']:,.0f}/年 ({best_result['net_return_pct']:.2f}%)")
        
        # 转换为股数（简化：使用平均价格估算）
        avg_price = sum(ld["avg_price"] for ld in liquidity_data) / len(liquidity_data)
        max_shares = int(best_capacity / avg_price) if avg_price > 0 else 0
        
        return CapacityResult(
            max_capital=best_capacity,
            max_shares=max_shares,
            estimated_slippage_bps=best_result['avg_slippage_bps'],
            estimated_impact_pct=best_result['avg_impact_pct'],
            net_expected_return_pct=best_result['net_return_pct'],
            liquidity_score=liquidity_score
        )


def main():
    """
    示例：估算一个等权重配置 5 只科技股的日内策略容量
    """
    estimator = StrategyCapacityEstimator()
    
    # 策略参数
    symbols = ["NVDA.US", "AMD.US", "INTC.US", "META.US", "GOOGL.US"]
    
    result = estimator.estimate_strategy_capacity(
        strategy_name="科技股等权日内",
        symbols=symbols,
        daily_turnover=0.3,  # 每天换手 30%
        backtest_return_pct=35.0,  # 年化 35%
        backtest_volatility_pct=22.0,  # 年化波动率 22%
        capital_range=(100_000, 50_000_000),  # 测试 10万到5000万
        target_return_pct=8.0  # 目标净年化 8%
    )
    
    return result


if __name__ == "__main__":
    result = main()

4.1 核心逻辑解读

这段代码实现了一个三段式容量估算流程：

1. 数据获取层：通过 TickDB 获取历史 K 线数据，计算日均成交额（ADV）和平均价格
2. 冲击模型层：使用简化的 Almgren-Chriss 模型，根据订单量/ADV 比例计算冲击成本
3. 容量搜索层：通过网格搜索不同资金规模，找到净收益刚好达到目标阈值的临界点

关键参数调优建议：

参数	保守值	激进值	说明
β（规模指数）	0.8	0.4	小盘股用高值
流动性安全系数	0.3	0.5	熊市/极端行情用低值
目标净收益率	8%	5%	决定了容量上限的严苛程度

---

五、实操案例：中小资金容量分析

假设你有 100 万美元，运行一个等权配置 5 只科技股的均值回归策略，日换手率 30%，回测年化 35%。

5.1 参数设置

参数	值
策略资金	$1,000,000
持仓标的	NVDA, AMD, INTC, META, GOOGL
标的数量	5
日换手率（单边）	30%
每标的可交易额	$60,000/天

5.2 容量分析表

资金规模	日均冲击成本	年化滑点成本	年化总成本	净收益	净收益率
$50 万	0.12%	$8,400	$15,200	$159,800	31.96%
$100 万	0.24%	$16,800	$30,400	$289,600	28.96%
$300 万	0.72%	$50,400	$91,200	$808,800	26.96%
$500 万	1.20%	$84,000	$152,000	$1,298,000	25.96%
$800 万	1.92%	$134,400	$243,200	$1,996,800	24.96%
$1,200 万	2.88%	$201,600	$364,800	$2,835,200	23.64%
$2,000 万	4.80%	$336,000	$608,000	$4,392,000	21.96%

5.3 容量边界判断

以 8% 目标净收益率 作为失效阈值：

• 当资金规模超过 $1,200 万 时，冲击成本开始显著侵蚀收益
• 超过 $3,000 万 时，滑点成本可能超过策略本身的 alpha

结论：对于这个 5 标的、30% 日换手的均值回归策略，保守容量上限约为 1,000 万美元，激进容量上限约为 2,000 万美元。

---

六、容量优化的四种路径

如果你的策略接近容量上限，有以下几种优化方向：

6.1 分散化扩展

增加持仓标的数量，减少单标的资金暴露。每增加一个低相关性标的，容量大约可提升 20-30%。

6.2 降低换手率

将日换手率从 30% 降至 15%，冲击成本大约降低一半，但会牺牲部分 alpha。

6.3 流动性筛选

将低流动性的持仓标的替换为高流动性的替代品（如用 SPY 替代小盘 ETF）。

6.4 执行算法优化

使用 TWAP/VWAP 算法分批执行，将单笔订单的冲击分摊到多个时间点。

---

七、结语

策略容量不是回测报告上的一个数字，而是需要在实盘前反复推敲的工程约束。

一个优秀的量化策略，不仅要有漂亮的回测曲线，还要有清醒的容量认知：我能管多少钱？我的策略在什么规模下会失效？

在 TickDB，你可以通过 depth 频道实时监控订单簿深度，结合 K 线数据计算 ADV，在实盘前建立更精确的容量预估模型。

---

下一步行动

如果你是量化新手，刚跑出第一个回测：用上面的代码跑一遍容量估算，别让“纸上富贵”变成真金白银的教训。

如果你已经有实盘经验：将你的交易记录导出，对比实际滑点与估算值的偏差，校准自己的冲击成本模型。

如果你想直接获取市场深度和流动性数据：访问 tickdb.ai 注册获取免费 API Key，配合本文的容量估算框架使用。

如果你习惯用 AI 辅助开发：在 AI 助手中搜索安装 tickdb-market-data SKILL，用自然语言查询市场数据和历史行情。

---

本文旨在提供策略容量估算的工程化方法，不构成投资建议。回测结果不代表未来收益，策略容量估算基于简化模型，实际表现可能存在偏差。市场有风险，投资需谨慎。