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

三个资金量，同一个策略，截然不同的命运

2023 年夏天，一位量化交易者带着他的均值回归策略进入实盘。

回测 5 年，年化收益 82%，夏普比率 3.2，最大回撤 6.3%。一切看起来近乎完美。

第一阶段，他放入 20 万。收益曲线与回测几乎重合，月胜率稳定在 68%。

第二阶段，他追加到 50 万。前两个月正常，第 91 天开始，滑点从预期的 0.02% 悄然爬升至 0.18%，月胜率跌破 50%。他以为是市场风格切换。

第三阶段，他押入全部身家 150 万。第 30 天，策略宣告失效——不是亏损，而是冲击成本彻底吃掉了所有 alpha，扣除手续费后年化收益变成 -2.7%。

这个故事每天都在量化社区上演。回测是理论的乌托邦，资金量是策略的毒药。

这不是策略本身的问题，而是大多数交易者在实盘前从未认真估算过：自己的策略能承载多少资金？

---

一、为什么回测有效的策略，实盘会失效

回测失效的原因通常被归结为过拟合、前视偏差、数据清洗不干净。但还有一个被严重低估的因素：你回测时假设的交易，在现实中不存在。

回测引擎撮合订单时，默认你是价格的接受者（Price Taker）——下单即成交，价格不受影响。但实盘中，你的订单本身就是市场的一部分。当资金量足够大时，你会从"价格的接受者"变成"价格的移动者"。

这背后有三条核心机制在起作用：

1.1 成交量约束：你的订单能否被市场消化

任何市场在任意时刻的流动性都是有限的。以 A 股主板为例，流动性呈现明显的结构性特征：

标的类型	日均成交额（亿元）	盘口深度（万元/档）	冲击系数
蓝筹权重股（茅台、工行）	50-200	200-800	0.0001
中盘成长股	10-50	50-200	0.0005
小盘题材股	1-10	5-50	0.002
冷门股/ST股	<1	<5	0.01+

冲击系数定义为：每增加 100 万元买单，对股价造成的瞬时偏移（%）。

假设你的策略单笔建仓规模为 20 万元，买入小盘题材股。冲击系数 0.002 意味着这笔交易会瞬时推高股价 0.04%。看起来不大，但如果策略每天交易 50 笔，累计冲击成本就是 2%——这已经超过了许多 alpha 策略的日均收益。

1.2 冲击成本模型：订单如何影响价格

学术上，冲击成本分为两部分：

临时冲击（Temporary Impact）：你的订单改变了短期供需平衡，价格暂时偏移，在订单完成后会部分回归。

永久冲击（Permanent Impact）：你的交易向市场传递了信息（"有人在买"），聪明的做市商和其他参与者会调整报价，这部分不会回归。

Almgren-Chriss 模型（2000）给出了冲击成本的经典表达：

$$c(v) = \eta \cdot \frac{v}{V} + \gamma \cdot \left(\frac{v}{V}\right)^{1/2}$$

其中：

• $v$ = 单笔订单规模
• $V$ = 日均成交量（ADV, Average Daily Volume）
• $\eta$ = 临时冲击系数
• $\gamma$ = 永久冲击系数

策略容量上限通常定义为：冲击成本等于策略预期收益的那一刻。超过这个点，交易越多，亏损越多。

1.3 信号衰减：规模是 alpha 的稀释剂

即便冲击成本为零，规模本身也会稀释收益。当你的资金量占市场日均成交量的比例过高时：

1. 持仓周期被迫延长：你无法在理想点位一次性建仓，需要分批买入
2. 滑点预期被打破：原本"收盘前价格回归"的假设失效
3. 信息优势窗口关闭：大资金的操作会被市场提前识别

一个在 50 万资金量下有效的信号，在 500 万规模下可能完全不成立。不是策略错了，而是市场已经对你的存在做出了定价。

---

二、容量估算的三层框架

估算策略容量不是一道数学题，而是一个需要分层求解的系统工程。

2.1 第一层：流动性边界

第一个问题是纯粹的物理约束：市场每天能消化多少你的交易量？

流动性边界的经验公式：

$$L_{max} = ADV \times \lambda \times \sqrt{N}$$

其中：

• $ADV$ = 日均成交量
• $\lambda$ = 最大持仓占比（通常取 5%-20%，取决于策略风格）
• $N$ = 策略覆盖的标的数量

为什么用 $\sqrt{N}$ 而不是 $N$？

因为分散化不是线性的。10 只标的的总容量不是 1 只标的的 10 倍，而是大约 3.16 倍（$\sqrt{10}$）。这是因为资金会在不同标的之间轮动，整体上受到市场整体流动性的约束。

2.2 第二层：冲击成本边界

第二个问题是经济约束：冲击成本会吃掉多少收益？

基于 Almgren-Chriss 模型的反推公式。设策略预期年化收益为 $r$，日均交易额为 $T$，则最大可承受的单边冲击成本为：

$$c_{max} = \frac{r}{252} / T$$

举例：策略预期年化 30%，日均交易额 100 万（单边），则最大可承受冲击成本约为：

$$c_{max} = \frac{0.30}{252} / 1,000,000 = 1.19 \times 10^{-9}$$

换算为百分比：0.00012%，即万分之一。

这意味着：如果单笔交易规模超过市场能安全吸收的范围，哪怕万分之一的额外冲击，也会让策略从盈利变为亏损。

2.3 第三层：信号有效性边界

第三个问题是策略层面的约束：策略本身的逻辑是否支撑规模扩张？

有些策略天然具备容量优势：

策略类型	容量弹性	原因
指数增强/ETF套利	高	盘口深，对手盘充足
跨市场套利	中	受汇率、结算周期限制
事件驱动（财报）	低	窗口期短，流动性集中释放
日内趋势跟踪	极低	高频信号密集，容易自相关
小盘股均值回归	极低	流动性差，冲击成本高

---

三、生产级容量估算代码实现

接下来用 Python 实现一个完整的容量估算模块。代码包含：

• 基于市场深度的冲击成本估算
• 三层容量边界的综合计算
• 蒙特卡洛压力测试

import os
import time
import random
import asyncio
import aiohttp
import numpy as np
import pandas as pd
from typing import Optional
from dataclasses import dataclass
from collections import deque


# ============================================================
# 核心数据类：策略配置与市场参数
# ============================================================
@dataclass
class StrategyConfig:
    """策略配置参数"""
    expected_return: float      # 预期年化收益率（例：0.30 表示 30%）
    daily_trade_value: float    # 日均交易额（单边，单位：元）
    holding_period_days: int    # 平均持仓周期（天）
    max_position_ratio: float  # 单标的最大持仓占比（例：0.05 表示 5%）
    trading_days: int = 252     # 年化交易日
    slippage_assumption: float = 0.0002  # 回测假设滑点（万分之二）
    
    def __post_init__(self):
        if not 0 < self.expected_return < 5:
            raise ValueError("预期收益率应在 0-500% 之间")
        if not 0 < self.daily_trade_value:
            raise ValueError("日均交易额必须为正数")


@dataclass
class CapacityResult:
    """容量估算结果"""
    liquidity_bound: float      # 流动性边界（元）
    impact_cost_bound: float    # 冲击成本边界（元）
    signal_bound: float         # 信号有效性边界（元）
    capacity_ceiling: float     # 综合容量上限（元）
    margin_of_safety: float     # 安全边际（元）
    risk_level: str             # 风险等级：LOW/MEDIUM/HIGH/CRITICAL
    
    def summary(self) -> str:
        return (
            f"容量估算结果：\n"
            f"  ├─ 流动性边界：{self.liquidity_bound:,.0f} 元\n"
            f"  ├─ 冲击成本边界：{self.impact_cost_bound:,.0f} 元\n"
            f"  ├─ 信号有效性边界：{self.signal_bound:,.0f} 元\n"
            f"  ├─ 综合容量上限：{self.capacity_ceiling:,.0f} 元\n"
            f"  └─ 安全边际：{self.margin_of_safety:,.0f} 元（{self.margin_of_safety/self.capacity_ceiling*100:.1f}%）\n"
            f"  风险等级：{self.risk_level}"
        )


# ============================================================
# 深度数据客户端（含重连、限频、超时保护）
# ⚠️ 生产环境建议使用 asyncio 架构处理高频订阅
# ============================================================
class DepthDataClient:
    """
    市场深度数据客户端
    
    功能：从 TickDB 获取 orderbook depth 数据，用于估算市场冲击成本
    ⚠️ 当前为同步实现，高频场景请切换为 aiohttp/asyncio
    """
    
    BASE_URL = "https://api.tickdb.ai/v1"
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("API Key 未设置，请设置环境变量 TICKDB_API_KEY")
        
        self.session = None
        self._retry_count = 0
        self._max_retries = 5
        self._base_delay = 1.0  # 基础重连延迟（秒）
    
    def _get_session(self) -> aiohttp.ClientSession:
        """延迟初始化 session，支持重连"""
        if self.session is None or self.session.closed:
            timeout = aiohttp.ClientTimeout(total=10, connect=5)
            connector = aiohttp.TCPConnector(limit=10)
            self.session = aiohttp.ClientSession(
                timeout=timeout,
                connector=connector,
                headers={"X-API-Key": self.api_key}
            )
        return self.session
    
    async def get_market_depth(self, symbol: str, depth: int = 10) -> Optional[dict]:
        """
        获取市场深度数据
        
        Args:
            symbol: 交易品种，如 "AAPL.US"
            depth: 深度档位（美股支持 1 档，港股/数字货币支持 10 档）
        
        Returns:
            包含 bids/asks 的深度数据字典
        
        ⚠️ 注意：美股 depth 仅支持 1 档（买卖一价），港股/数字货币支持 10 档
        """
        session = self._get_session()
        
        for retry in range(self._max_retries):
            try:
                async with session.get(
                    f"{self.BASE_URL}/market/depth",
                    params={"symbol": symbol, "depth": depth},
                    timeout=aiohttp.ClientTimeout(total=5)
                ) as response:
                    data = await response.json()
                    
                    # 限频处理（code: 3001）
                    if response.status == 429 or data.get("code") == 3001:
                        retry_after = int(response.headers.get("Retry-After", 5))
                        wait_time = retry_after + random.uniform(0, 1)  # 加抖动
                        print(f"[限频] 等待 {wait_time:.1f} 秒后重试...")
                        await asyncio.sleep(wait_time)
                        continue
                    
                    if data.get("code") == 0:
                        self._retry_count = 0  # 重置重试计数
                        return data.get("data")
                    
                    # 其他 API 错误
                    self._handle_api_error(data, symbol)
                    
            except aiohttp.ClientError as e:
                delay = min(self._base_delay * (2 ** retry), 30)
                jitter = random.uniform(0, delay * 0.1)
                wait_time = delay + jitter
                
                print(f"[连接错误] {symbol}: {e}, {retry+1}/{self._max_retries}, "
                      f"等待 {wait_time:.1f}s...")
                
                await asyncio.sleep(wait_time)
                self._retry_count += 1
        
        raise RuntimeError(f"获取 {symbol} 深度数据失败，已重试 {self._max_retries} 次")
    
    async def get_available_symbols(self, market: str = "US") -> list:
        """获取指定市场的可用交易品种"""
        session = self._get_session()
        
        try:
            async with session.get(
                f"{self.BASE_URL}/symbols/available",
                params={"market": market}
            ) as response:
                data = await response.json()
                if data.get("code") == 0:
                    return data.get("data", [])
                return []
        except Exception:
            return []
    
    def _handle_api_error(self, response: dict, symbol: str = None):
        """统一错误处理"""
        code = response.get("code", 0)
        message = response.get("message", "未知错误")
        
        error_map = {
            1001: "API Key 无效，请检查环境变量 TICKDB_API_KEY",
            1002: "API Key 缺失，请先获取 API Key",
            2002: f"交易品种 {symbol} 不存在，请确认代码格式（如 AAPL.US）",
        }
        
        if code in error_map:
            raise ValueError(f"[错误 {code}] {error_map[code]}")
        
        raise RuntimeError(f"[错误 {code}] {message}")
    
    async def close(self):
        """关闭 session"""
        if self.session and not self.session.closed:
            await self.session.close()


# ============================================================
# 冲击成本估算器
# ============================================================
class ImpactCostEstimator:
    """
    基于市场深度的冲击成本估算
    
    模型说明：
    - 使用 TickDB depth 数据估算市场吸收能力
    - 冲击成本 = 订单规模 × 深度加权平均价差
    - 采用 Almgren-Chriss 的平方根模型
    
    ⚠️ 注意：这是一个简化模型，实际冲击成本受更多因素影响
    """
    
    def __init__(self, depth_client: DepthDataClient):
        self.client = depth_client
    
    async def estimate_impact_cost(
        self,
        symbol: str,
        order_value: float,
        is_buy: bool = True,
        adv: float = None  # 日均成交量
    ) -> dict:
        """
        估算单笔交易的冲击成本
        
        Args:
            symbol: 交易品种
            order_value: 订单金额（元）
            is_buy: 是否为买入（True=买，False=卖）
            adv: 日均成交量（如果已知）
        
        Returns:
            {
                'impact_cost_pct': 冲击成本百分比,
                'impact_cost_abs': 冲击成本绝对值（元）,
                'price_shift': 价格偏移,
                'market_impact_ratio': 相对于日均成交的比例
            }
        """
        depth_data = await self.client.get_market_depth(symbol)
        
        if not depth_data:
            # 降级处理：使用保守估计
            return self._conservative_estimate(order_value, adv)
        
        bids = depth_data.get("bids", [])  # 买盘 [[price, volume], ...]
        asks = depth_data.get("asks", [])  # 卖盘
        
        if not bids or not asks:
            return self._conservative_estimate(order_value, adv)
        
        # 计算 VWAP 深度
        levels = asks if is_buy else bids  # 买入看卖盘，卖出看买盘
        
        remaining_value = order_value
        filled_value = 0.0
        avg_fill_price = 0.0
        
        for price, volume in levels:
            level_value = price * volume
            
            if remaining_value <= level_value:
                # 订单在此档位被完全吸收
                avg_fill_price += (remaining_value / volume) * price / order_value
                filled_value += remaining_value
                remaining_value = 0
                break
            else:
                # 需要跨越多个档位
                avg_fill_price += price  # 简化：取档位价格均值
                filled_value += level_value
                remaining_value -= level_value
        
        if remaining_value > 0:
            # 流动性不足：使用最差档位价格
            worst_price = levels[-1][0] if levels else (bids[0][0] + asks[0][0]) / 2
            avg_fill_price = worst_price
            filled_value += remaining_value
        
        # 基准价格：盘口中间价
        mid_price = (bids[0][0] + asks[0][0]) / 2
        
        # 冲击成本 = 实际成交均价 - 基准价格（相对于基准价格）
        if filled_value > 0:
            # 简化：avg_fill_price 实际上是档位数的均值，需要转换
            impact_cost_pct = abs(avg_fill_price - mid_price) / mid_price
        else:
            impact_cost_pct = 0.0
        
        # Almgren-Chriss 冲击模型调整
        # c(v) = η * (v/ADV) + γ * sqrt(v/ADV)
        if adv and adv > 0:
            temp_impact_coeff = 0.1   # η: 临时冲击系数（需校准）
            perm_impact_coeff = 0.05  # γ: 永久冲击系数（需校准）
            
            vol_ratio = order_value / adv
            ac_impact = (temp_impact_coeff * vol_ratio + 
                        perm_impact_coeff * np.sqrt(vol_ratio))
            impact_cost_pct = max(impact_cost_pct, ac_impact)
        
        return {
            "impact_cost_pct": impact_cost_pct,
            "impact_cost_abs": order_value * impact_cost_pct,
            "mid_price": mid_price,
            "filled_levels": len(levels),
            "market_impact_ratio": order_value / adv if adv else None
        }
    
    def _conservative_estimate(self, order_value: float, adv: float = None) -> dict:
        """
        降级估算：当无法获取深度数据时的保守估计
        
        假设：千分之一到万分之五的保守冲击成本范围
        """
        if adv and adv > 0:
            impact_ratio = order_value / adv
            # 保守估计：冲击成本与交易量占比呈非线性关系
            impact_cost_pct = 0.001 * np.sqrt(impact_ratio)
        else:
            # 完全未知：使用万分之五作为保守估计
            impact_cost_pct = 0.0005
        
        return {
            "impact_cost_pct": impact_cost_pct,
            "impact_cost_abs": order_value * impact_cost_pct,
            "mid_price": None,
            "filled_levels": 0,
            "market_impact_ratio": None,
            "is_estimate": True  # 标记为估算值
        }


# ============================================================
# 策略容量估算器
# ============================================================
class CapacityEstimator:
    """
    综合容量估算器
    
    结合流动性边界、冲击成本边界、信号有效性边界，
    计算策略的综合容量上限和风险等级。
    """
    
    def __init__(
        self,
        config: StrategyConfig,
        depth_client: DepthDataClient = None
    ):
        self.config = config
        self.depth_client = depth_client
        self.impact_estimator = ImpactCostEstimator(depth_client) if depth_client else None
    
    async def estimate_capacity(
        self,
        symbols: list,
        adv_data: dict = None  # {symbol: adv_value}
    ) -> CapacityResult:
        """
        执行完整的容量估算
        
        Args:
            symbols: 策略覆盖的交易品种列表
            adv_data: 日均成交量数据（可选）
        
        Returns:
            CapacityResult 对象
        """
        n_symbols = len(symbols) if symbols else 1
        
        # 第一层：流动性边界
        liquidity_bound = self._calc_liquidity_bound(n_symbols, adv_data)
        
        # 第二层：冲击成本边界
        impact_bound = await self._calc_impact_bound(symbols)
        
        # 第三层：信号有效性边界（需要根据策略类型自定义）
        signal_bound = self._calc_signal_bound()
        
        # 综合容量上限：取三层的最小值
        capacity_ceiling = min(liquidity_bound, impact_bound, signal_bound)
        
        # 安全边际：容量上限的 20% 作为安全边际
        margin_of_safety = capacity_ceiling * 0.2
        
        # 风险等级判定
        risk_level = self._assess_risk(
            self.config.daily_trade_value,
            capacity_ceiling
        )
        
        return CapacityResult(
            liquidity_bound=liquidity_bound,
            impact_cost_bound=impact_bound,
            signal_bound=signal_bound,
            capacity_ceiling=capacity_ceiling,
            margin_of_safety=margin_of_safety,
            risk_level=risk_level
        )
    
    def _calc_liquidity_bound(
        self,
        n_symbols: int,
        adv_data: dict = None
    ) -> float:
        """
        流动性边界计算
        
        公式：L_max = sum(ADV_i * λ) * sqrt(N) / N * N
        简化：总容量 ≈ avg(ADV) * λ * sqrt(N) * N_adj
        """
        if adv_data:
            total_adv = sum(adv_data.values())
        else:
            # 假设单标的日均成交 1000 万
            total_adv = n_symbols * 10_000_000
        
        # 分散化调整因子
        diversification_factor = np.sqrt(n_symbols)
        
        liquidity_bound = (
            total_adv 
            * self.config.max_position_ratio 
            * diversification_factor
        )
        
        return liquidity_bound
    
    async def _calc_impact_bound(self, symbols: list) -> float:
        """
        冲击成本边界计算
        
        策略可承受的最大资金规模 = 日均收益 / 单笔冲击成本
        """
        if not self.impact_estimator or not symbols:
            # 降级：使用保守假设（万分之一冲击成本）
            max_acceptable_impact = 0.0001
        else:
            # 基于实时数据估算
            sample_symbol = symbols[0]
            impact_result = await self.impact_estimator.estimate_impact_cost(
                sample_symbol,
                self.config.daily_trade_value,
                is_buy=True,
                adv=adv_data.get(sample_symbol) if adv_data else None
            )
            max_acceptable_impact = impact_result["impact_cost_pct"]
        
        # 反推：如果每笔交易承受 max_impact 的成本，
        # 日均交易额最多为 (日收益) / (单笔冲击成本)
        daily_return = self.config.expected_return / self.config.trading_days
        impact_bound = self.config.daily_trade_value * (1 - max_acceptable_impact)
        
        # 容量上限的反推：日均交易额 / 换手率
        turnover_rate = 1 / self.config.holding_period_days
        impact_bound = (daily_return / max_acceptable_impact) if max_acceptable_impact > 0 else float('inf')
        
        # 上限 = 冲击成本边界 * 换手率倒数（持仓周期）
        impact_bound = impact_bound / turnover_rate * (1 + max_acceptable_impact)
        
        return impact_bound
    
    def _calc_signal_bound(self) -> float:
        """
        信号有效性边界（简化估算）
        
        需要根据具体策略回测数据校准
        此处使用策略参数和持仓周期估算
        """
        # 经验法则：日内策略容量 << 隔夜策略
        if self.config.holding_period_days <= 1:
            signal_factor = 0.5  # 日内策略容量折扣
        elif self.config.holding_period_days <= 5:
            signal_factor = 0.7
        else:
            signal_factor = 0.9
        
        # 基于持仓周期估算信号边界
        signal_bound = (
            self.config.daily_trade_value 
            * self.config.holding_period_days 
            * 100  # 基础倍数
            * signal_factor
        )
        
        return signal_bound
    
    def _assess_risk(self, current_capital: float, capacity_ceiling: float) -> str:
        """
        评估策略当前资金量的风险等级
        """
        utilization = current_capital / capacity_ceiling if capacity_ceiling > 0 else 1.0
        
        if utilization < 0.3:
            return "LOW"
        elif utilization < 0.5:
            return "MEDIUM"
        elif utilization < 0.8:
            return "HIGH"
        else:
            return "CRITICAL"
    
    def monte_carlo_stress_test(
        self,
        n_simulations: int = 1000,
        capacity_ceiling: float = None
    ) -> pd.DataFrame:
        """
        蒙特卡洛压力测试
        
        模拟不同资金规模下的策略表现分布
        
        ⚠️ 这是一个简化模型，真实压力测试需要考虑更多因素
        """
        if capacity_ceiling is None:
            # 如果没有提供容量上限，使用默认值
            capacity_ceiling = self.config.daily_trade_value * 100
        
        # 模拟资金规模范围
        capital_range = np.linspace(
            capacity_ceiling * 0.1,
            capacity_ceiling * 1.2,
            n_simulations
        )
        
        results = []
        for capital in capital_range:
            # 模拟收益分布
            # 假设：收益服从正态分布，标准差随规模增加
            scale_factor = capital / capacity_ceiling
            sigma = 0.15 * np.sqrt(scale_factor)  # 波动率随规模放大
            
            expected_return = self.config.expected_return * (1 - scale_factor * 0.5)
            realized_return = np.random.normal(expected_return, sigma)
            
            # 冲击成本计算
            impact_cost = self._estimate_impact_for_capital(capital)
            net_return = realized_return - impact_cost
            
            results.append({
                "capital": capital,
                "expected_return": expected_return,
                "realized_return": realized_return,
                "impact_cost": impact_cost,
                "net_return": net_return,
                "profitability": "PROFIT" if net_return > 0 else "LOSS"
            })
        
        return pd.DataFrame(results)
    
    def _estimate_impact_for_capital(self, capital: float) -> float:
        """估算资金规模对应的冲击成本"""
        # 简化：冲击成本与规模的平方根成正比
        base_capital = self.config.daily_trade_value * 100
        scale_factor = capital / base_capital
        impact_cost = 0.001 * np.sqrt(scale_factor)
        return impact_cost


# ============================================================
# 使用示例
# ============================================================
async def main():
    """
    使用示例：估算一个中小资金均值回归策略的容量
    """
    # 配置参数
    config = StrategyConfig(
        expected_return=0.30,        # 预期年化 30%
        daily_trade_value=200_000,    # 日均单边交易额 20 万
        holding_period_days=3,       # 平均持仓 3 天
        max_position_ratio=0.05,     # 单标的最大持仓 5%
    )
    
    print("=" * 60)
    print("策略容量估算器")
    print("=" * 60)
    print(f"预期年化收益：{config.expected_return * 100}%")
    print(f"日均交易额：{config.daily_trade_value:,.0f} 元")
    print(f"持仓周期：{config.holding_period_days} 天")
    print(f"单标的最大持仓：{config.max_position_ratio * 100}%")
    print("-" * 60)
    
    # 初始化深度数据客户端
    client = DepthDataClient()
    
    try:
        estimator = CapacityEstimator(config, client)
        
        # 示例：估算 AAPL.US 的容量
        symbols = ["AAPL.US", "MSFT.US", "GOOGL.US"]
        
        # 如果有日均成交量数据
        adv_data = {
            "AAPL.US": 80_000_000,   # AAPL 日均成交 8000 万
            "MSFT.US": 60_000_000,
            "GOOGL.US": 50_000_000,
        }
        
        print(f"覆盖标的：{', '.join(symbols)}")
        print("-" * 60)
        
        # 执行容量估算
        result = await estimator.estimate_capacity(symbols, adv_data)
        
        print(result.summary())
        print("-" * 60)
        
        # 蒙特卡洛压力测试
        print("\n运行蒙特卡洛压力测试（1000 次模拟）...")
        stress_results = estimator.monte_carlo_stress_test(
            n_simulations=1000,
            capacity_ceiling=result.capacity_ceiling
        )
        
        # 分析结果
        profitable_ratio = (
            stress_results["profitability"] == "PROFIT"
        ).mean()
        
        print(f"\n压力测试结果（容量上限 {result.capacity_ceiling:,.0f} 元）：")
        print(f"  ├─ 盈利概率（0.1x-1.2x 容量范围）：{profitable_ratio * 100:.1f}%")
        print(f"  ├─ 平均净收益：{stress_results['net_return'].mean() * 100:.2f}%")
        print(f"  └─ 收益标准差：{stress_results['realized_return'].std() * 100:.2f}%")
        
        # 找到盈亏平衡点
        break_even_idx = stress_results[stress_results['net_return'] > 0]['capital'].min()
        print(f"\n盈亏平衡点：约 {break_even_idx:,.0f} 元")
        
        if break_even_idx < result.capacity_ceiling:
            safety_ratio = break_even_idx / result.capacity_ceiling
            print(f"安全边际：{(1 - safety_ratio) * 100:.1f}%")
    
    finally:
        await client.close()


if __name__ == "__main__":
    # ⚠️ 需要设置环境变量：export TICKDB_API_KEY="your-api-key"
    # 生产环境建议使用 asyncio.run(main())
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n估算中断")
    except Exception as e:
        print(f"错误：{e}")

代码核心逻辑说明

上述代码实现了完整的容量估算框架：

第一层：流动性边界——基于日均成交量（ADV）和持仓占比估算。这个数字回答的是"市场能承接多少"。

第二层：冲击成本边界——调用 TickDB 的 depth 频道获取实时盘口数据，计算实际成交均价与中间价的偏离。这是"你的交易会移动多少价格"。

第三层：信号有效性边界——基于持仓周期和策略类型估算。日内策略的容量天然小于隔夜策略。

三层取最小值，就是策略的容量上限。

---

四、容量验证与压力测试

估算完成后，下一步是验证。

4.1 逆向检验：回测收益剥离冲击成本

将回测结果重新跑一遍，但这次在每笔成交价上叠加估算的冲击成本。如果调整后的收益仍然为正，说明容量估算合理。

def backtest_with_impact(
    original_returns: np.ndarray,
    capital: float,
    capacity_ceiling: float,
    trading_cost_bps: float = 2.0  # 佣金+滑点（基点）
) -> dict:
    """
    叠加冲击成本后的回测检验
    
    Args:
        original_returns: 原始回测收益序列
        capital: 策略资金规模
        capacity_ceiling: 估算的容量上限
        trading_cost_bps: 基础交易成本（基点）
    
    Returns:
        调整后的收益统计
    """
    utilization = capital / capacity_ceiling
    
    # 规模调整因子：资金利用率越高，冲击成本越高
    # 使用平方根模型
    impact_multiplier = np.sqrt(utilization) if utilization > 0 else 0
    
    # 额外冲击成本（年化）
    additional_impact = 0.005 * impact_multiplier  # 基准 0.5%，随规模放大
    
    # 调整后收益
    adjusted_returns = original_returns - additional_impact - (trading_cost_bps / 10000)
    
    return {
        "original_annual_return": original_returns.mean() * 252,
        "adjusted_annual_return": adjusted_returns.mean() * 252,
        "return_erosion": (original_returns.mean() - adjusted_returns.mean()) * 252,
        "original_sharpe": original_returns.mean() / original_returns.std() * np.sqrt(252),
        "adjusted_sharpe": adjusted_returns.mean() / adjusted_returns.std() * np.sqrt(252) if adjusted_returns.std() > 0 else 0,
        "still_profitable": adjusted_returns.mean() > 0
    }

4.2 蒙特卡洛压力测试解读

回到前面的代码示例，压力测试的核心输出是盈亏平衡点。

这个数字告诉你：当资金规模达到多少时，策略扣除冲击成本后会开始亏损。

如果你的当前资金量已经超过盈亏平衡点，那么容量估算就是你的预警信号——不是让你停止交易，而是让你意识到：需要降低交易频率、减少单笔规模，或者寻找流动性更好的标的。

---

五、不同资金规模的容量应对策略

资金规模	容量等级	核心问题	应对策略
<50 万	小资金	冲击成本可控，但样本不足	聚焦高流动性大盘股，避免小票
50-200 万	中小资金	冲击成本开始显现	分批建仓，延长建仓周期
200-500 万	中等资金	需要主动管理流动性	引入 TWAP/VWAP 执行算法
>500 万	大资金	信号有效性存疑	考虑多策略分散，或降低收益预期

对于大多数个人量化交易者，50-200 万是最尴尬的区域：这个规模足以触发冲击成本，但又不至于支撑专业级的执行算法。

在这个区间，建议的核心原则是：

1. 降低换手率：持仓周期从 1 天延长到 3-5 天，减少冲击成本的累积
2. 聚焦流动性：只做日均成交额超过 5 亿的标的
3. 分批执行：单笔建仓不超过日均成交量的 1%
4. 定期复盘：每季度重新估算容量，观察侵蚀是否加剧

---

六、结语

容量估算不是一次性的工作，而是策略生命周期管理的一部分。

随着你的资金量增长、市场环境变化、竞争对手涌入，容量上限会持续收缩。一年前有效的策略，今天可能已经触及边界。

这不是失败，而是市场的自我修正机制。

真正成熟的量化交易者，不是找到了一只永不失效的圣杯，而是建立了一套持续监控容量、及时调整策略的系统。

回测是起点，实盘是考场，容量估算是贯穿全程的生存指南。

---

下一步行动

如果你刚起步，策略资金量在 50 万以内：
优先选择高流动性大盘股，建立容量友好的交易习惯比短期收益更重要。

如果你已经进入中等资金规模：
运行本文代码，基于 TickDB 的实时深度数据，测算出你策略的真实容量上限。如果当前资金量已超过安全边际，考虑调整策略参数或执行方式。

如果你想获取本文的历史 K 线数据做进一步回测验证：
访问 tickdb.ai 注册获取 API Key，10 年级别的美股清洗数据可以支撑跨周期策略验证。

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

---

风险提示：本文不构成任何投资建议。策略容量受市场环境、流动性状况、竞争格局等多重因素影响，历史估算结果不能预测未来表现。市场有风险，投资需谨慎。