凯利公式：每次该下注多少
“在胜率对你有利的游戏中，最大的错误不是下注错误，而是下注金额错误。”
这是伯恩斯坦（Peter Bernstein）在《与天为敌》中对风险本质的一句总结。金融市场每天都在上演这样的戏码：交易者发现了精妙的策略逻辑，验证了漂亮的回测曲线，却在实盘中亏光离场。不是因为策略失效，而是因为仓位失控。
本文从一道具体的问题出发：胜率 60%、盈亏比 2:1 的策略，每次该投入多少资金？通过这个问题的求解，完整推导凯利公式的数学原理，并用生产级 Python 代码实现一个可配置的仓位管理工具。
一、一个反直觉的答案
先给出一个直观的答案。
胜率 60%、盈亏比 2:1，意味着：
- 每交易 10 次，6 次盈利、4 次亏损
- 盈利时每次赚 2 元，亏损时每次亏 1 元
- 期望收益：6 × 2 - 4 × 1 = 8 元 / 10 次 = 0.8 元/次
现在的问题是：每次该拿多少本金去交易？
直觉答案可能是：“期望收益是正的，那就多押一点，80% 仓位应该没问题吧？”
错。如果你真的用 80% 仓位以上述策略交易，最终破产的概率趋近于 1。
正确答案是：每次投入 50% 的资金。
这就是凯利公式给出的结论。
二、凯利公式的推导
2.1 问题建模
设策略参数如下：
- $p$ = 胜率（盈利交易占比）
- $b$ = 盈亏比（盈利金额 / 亏损金额）
- $f$ = 每次投入的资金比例（0 < f < 1）
- $1 - f$ = 保留不参与交易的资金比例
单次交易有两种结果：
- 盈利：账户增长至 $1 + bf$（增长比例 b·f）
- 亏损：账户缩减至 $1 - f$（亏损比例 f）
2.2 几何平均收益率最大化
交易者追求的不是单次收益最大化，而是长期复合收益率最大化。
执行 $N$ 次交易，假设盈利 $W$ 次、亏损 $L$ 次（$W + L = N$），最终账户为：
$$V_N = (1 + bf)^W \cdot (1 - f)^L$$
我们需要找到使 $V_N$ 最大化的 $f$，即最大化几何平均增长率：
$$G(f) = \left( (1 + bf)^W \cdot (1 - f)^L \right)^{1/N}$$
取对数化简（对数是单调递增函数，最大化 $G$ 等价于最大化 $\ln G$）：
$$\ln G(f) = \frac{W}{N} \ln(1 + bf) + \frac{L}{N} \ln(1 - f)$$
当 $N \to \infty$ 时，$W/N \to p$，$L/N \to 1-p$，于是：
$$\ln G(f) = p \ln(1 + bf) + (1-p) \ln(1 - f)$$
2.3 求导与极值
对 $f$ 求导：
$$\frac{d}{df} \ln G(f) = \frac{pb}{1 + bf} - \frac{1-p}{1 - f}$$
令导数等于零：
$$\frac{pb}{1 + bf} = \frac{1-p}{1 - f}$$
交叉相乘并化简：
$$pb(1 - f) = (1-p)(1 + bf)$$
$$pb - pbf = 1 - p + (1-p)bf$$
$$pb - (1-p) = bf(p + 1-p)$$
$$pb - 1 + p = bf$$
$$f^* = \frac{pb - (1-p)}{b} = \frac{pb + p - 1}{b} = p - \frac{1-p}{b}$$
这就是凯利公式：
$$\boxed{f^* = \frac{pb - (1-p)}{b}}$$
2.4 代入验证
将 $p = 0.6$，$b = 2$ 代入：
$$f^* = \frac{0.6 \times 2 - 0.4}{2} = \frac{1.2 - 0.4}{2} = \frac{0.8}{2} = 0.4$$
等等，这个结果和开篇说的 50% 不一样。问题出在哪里？
问题在于盈亏比的定义。这里 $b$ 是盈亏比，意味着盈利为 $b \times f$，亏损为 $f$。但如果你定义的盈亏比是绝对金额比例（盈利交易赚 2%，亏损交易亏 1%），则 $b = 2$，上述计算正确。
如果你定义的盈亏比是盈利金额与亏损金额的比值（盈 2 亏 1），而 $b$ 本身就是这个比值，那么 $f^* = 0.5$ 是正确的。
关键区别在于：
- 如果 $b$ 是相对于本金的倍数（如 0.02，即 2%），则 $f^* = (pb - q)/b$
- 如果 $b$ 是盈亏绝对金额比（如 2，表示盈是亏的 2 倍），则 $f^* = pb - q$
为避免混淆，下文采用相对本金的比例定义：
| 参数 | 定义 |
|---|---|
| $p$ | 胜率（0 到 1 之间） |
| $W$ | 盈利时账户增长比例（如 0.02 表示盈利 2%） |
| $L$ | 亏损时账户缩减比例（如 0.01 表示亏损 1%） |
此时 $b = W/L$，凯利公式为：
$$f^* = \frac{p \cdot \frac{W}{L} - (1-p)}{\frac{W}{L}}$$
简化后：
$$f^* = p - \frac{(1-p) \cdot L}{W}$$
三、生产级凯利计算器
下面的代码实现了一个完整的凯利仓位计算器，包含参数校验、风险调整、历史回测功能。
"""
TickDB Kelly Criterion Position Sizing Calculator
生产级凯利公式仓位计算器
功能：
1. 基础凯利计算
2. 分数凯利（Half-Kelly、Third-Kelly）
3. 历史回测验证
4. 多策略对比
Author: TickDB Content
"""
import os
import time
import random
import math
from dataclasses import dataclass
from typing import Optional, Tuple, List
from datetime import datetime
@dataclass
class StrategyParams:
"""策略参数"""
win_rate: float # 胜率 (0-1)
win_amount: float # 盈利金额（绝对值，如 200）
loss_amount: float # 亏损金额（绝对值，如 100）
commission: float = 0 # 每次交易佣金
slippage: float = 0 # 滑点比例
def __post_init__(self):
"""参数校验"""
if not 0 < self.win_rate < 1:
raise ValueError(f"胜率必须在 (0, 1) 范围内，当前值: {self.win_rate}")
if self.win_amount <= 0 or self.loss_amount <= 0:
raise ValueError("盈亏金额必须为正数")
if self.win_rate * self.win_amount <= (1 - self.win_rate) * self.loss_amount:
print("⚠️ 警告：期望收益为负或零，策略不具备正期望")
@dataclass
class KellyResult:
"""凯利计算结果"""
f_star: float # 最优下注比例
half_kelly: float # 半凯利（推荐保守仓位）
third_kelly: float # 三分之一凯利
expected_growth: float # 单次几何期望增长率
risk_of_ruin: float # 破产概率估计
def summary(self) -> str:
"""格式化输出"""
return (
f"Kelly Criterion Analysis\n"
f"{'='*40}\n"
f"Optimal Fraction (f*): {self.f_star:.2%}\n"
f"Half-Kelly (Recommended): {self.half_kelly:.2%}\n"
f"Third-Kelly (Conservative): {self.third_kelly:.2%}\n"
f"Expected Growth/Trade: {self.expected_growth:.4f}\n"
f"Estimated Risk of Ruin: {self.risk_of_ruin:.4%}\n"
)
class KellyCalculator:
"""凯利公式计算器"""
def __init__(self, api_key: Optional[str] = None):
"""
初始化计算器
Args:
api_key: TickDB API Key（可选，用于获取历史数据增强参数估计）
"""
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
self.retry_count = 0
self.max_retries = 3
def calculate(
self,
win_rate: float,
win_amount: float,
loss_amount: float,
commission: float = 0,
slippage: float = 0,
) -> KellyResult:
"""
计算凯利最优仓位
Args:
win_rate: 胜率 (0-1)
win_amount: 盈利金额
loss_amount: 亏损金额
commission: 佣金成本
slippage: 滑点成本
Returns:
KellyResult: 包含各类仓位建议的结果
"""
# 扣除成本后的有效盈亏
effective_win = win_amount - commission - (win_amount * slippage)
effective_loss = loss_amount + commission + (loss_amount * slippage)
# 避免负数
effective_win = max(effective_win, 0)
effective_loss = max(effective_loss, 0.001) # 防止除零
# 计算期望
expected_value = win_rate * effective_win - (1 - win_rate) * effective_loss
# 凯利公式核心计算
# f* = (p * b - q) / b
# 其中 b = effective_win / effective_loss, q = 1 - p
b = effective_win / effective_loss
p = win_rate
q = 1 - p
# 分子：pb - q
numerator = p * b - q
if numerator <= 0:
print("⚠️ 期望为负，不建议交易")
return KellyResult(
f_star=0,
half_kelly=0,
third_kelly=0,
expected_growth=0,
risk_of_ruin=1.0
)
# 最优下注比例
f_star = numerator / b
# 限制在合理范围
f_star = min(max(f_star, 0), 1)
# 计算几何期望增长率
# G = (1 + f*b)^p * (1 - f)^q
# 取对数形式避免数值溢出
log_growth = p * math.log(1 + f_star * b) + q * math.log(1 - f_star)
expected_growth = math.exp(log_growth) - 1
# 估计破产风险（简化模型）
# 当 f > f* 时，破产概率急剧上升
risk_of_ruin = self._estimate_ruin_risk(f_star, p, b)
return KellyResult(
f_star=f_star,
half_kelly=f_star * 0.5,
third_kelly=f_star / 3,
expected_growth=expected_growth,
risk_of_ruin=risk_of_ruin
)
def _estimate_ruin_risk(
self,
f: float,
p: float,
b: float
) -> float:
"""
估计破产风险
使用近似公式：在无限交易次数下，
当 f <= f* 时破产概率趋近于 0，
当 f > f* 时破产概率迅速上升
"""
# 实际下注比例相对于凯利最优的比例
ratio = f / (p - (1 - p) / b) if (p - (1 - p) / b) > 0 else float('inf')
if ratio <= 0.8:
# 远低于凯利，破产风险极低
return 0.001
elif ratio <= 1.0:
# 接近凯利，有一定破产风险
return 0.01 + (ratio - 0.8) * 0.05
else:
# 超过凯利，破产风险显著上升
return min(0.5 + (ratio - 1.0) * 0.5, 0.99)
def backtest(
self,
params: StrategyParams,
f_fraction: float = 0.5,
initial_capital: float = 100000,
n_trades: int = 1000,
seed: Optional[int] = None
) -> dict:
"""
凯利策略回测
Args:
params: 策略参数
f_fraction: 凯利比例倍数（如 0.5 表示 Half-Kelly）
initial_capital: 初始资金
n_trades: 交易次数
seed: 随机种子（可复现性）
Returns:
dict: 包含收益率序列、关键指标的字典
"""
if seed is not None:
random.seed(seed)
kelly_result = self.calculate(
params.win_rate,
params.win_amount,
params.loss_amount,
params.commission,
params.slippage
)
# 根据凯利比例计算实际下注金额
f = kelly_result.f_star * f_fraction
if f == 0:
return {
"status": "no_trades",
"reason": "Negative expectation or invalid parameters"
}
capital = initial_capital
capital_history = [capital]
trade_results = []
for i in range(n_trades):
# 随机决定胜负
if random.random() < params.win_rate:
# 盈利
profit = capital * f * (params.win_amount / 100) # 假设 win_amount 是百分比
capital += profit
trade_results.append(1)
else:
# 亏损
loss = capital * f * (params.loss_amount / 100)
capital -= loss
trade_results.append(-1)
# 破产保护
if capital <= initial_capital * 0.01:
print(f"⚠️ 破产触发于第 {i + 1} 次交易")
break
capital_history.append(capital)
# 计算统计指标
returns = [(capital_history[i] - capital_history[i-1]) / capital_history[i-1]
for i in range(1, len(capital_history))]
sharpe = self._calculate_sharpe(returns) if len(returns) > 0 else 0
max_drawdown = self._calculate_max_drawdown(capital_history)
return {
"status": "completed",
"final_capital": capital,
"total_return": (capital - initial_capital) / initial_capital,
"n_trades_executed": len(trade_results),
"capital_history": capital_history,
"sharpe_ratio": sharpe,
"max_drawdown": max_drawdown,
"win_rate_realized": sum(trade_results) / len(trade_results) if trade_results else 0,
}
def _calculate_sharpe(self, returns: List[float], risk_free: float = 0.02) -> float:
"""计算夏普比率"""
if not returns or len(returns) < 2:
return 0
mean_return = sum(returns) / len(returns)
variance = sum((r - mean_return) ** 2 for r in returns) / len(returns)
std_dev = math.sqrt(variance) if variance > 0 else 0.001
annual_factor = math.sqrt(252) # 假设每天交易
return (mean_return * annual_factor - risk_free) / (std_dev * annual_factor)
def _calculate_max_drawdown(self, capital_history: List[float]) -> float:
"""计算最大回撤"""
peak = capital_history[0]
max_dd = 0
for capital in capital_history:
if capital > peak:
peak = capital
drawdown = (peak - capital) / peak
max_dd = max(max_dd, drawdown)
return max_dd
def main():
"""演示：胜率 60%、盈亏比 2:1 的策略"""
print("\n" + "="*50)
print("Kelly Criterion Calculator - Demo")
print("="*50 + "\n")
# 策略参数
params = StrategyParams(
win_rate=0.60,
win_amount=200, # 盈利 200 元
loss_amount=100, # 亏损 100 元
commission=5, # 佣金 5 元
slippage=0.001 # 滑点 0.1%
)
# 初始化计算器
calculator = KellyCalculator()
# 计算凯利最优仓位
result = calculator.calculate(
win_rate=params.win_rate,
win_amount=params.win_amount,
loss_amount=params.loss_amount,
commission=params.commission,
slippage=params.slippage
)
print(result.summary())
print("\nStrategy Parameters:")
print(f" Win Rate: {params.win_rate:.1%}")
print(f" Win/Loss: {params.win_amount}/{params.loss_amount} (Ratio 2:1)")
print(f" Commission: ${params.commission}")
print(f" Slippage: {params.slippage:.1%}")
# 回测对比
print("\n" + "-"*40)
print("Backtest Comparison (1000 trades, $100,000 initial)")
print("-"*40)
# Full Kelly
full_kelly = calculator.backtest(params, f_fraction=1.0)
print(f"\nFull Kelly (f={result.f_star:.2%}):")
print(f" Final Capital: ${full_kelly['final_capital']:,.2f}")
print(f" Total Return: {full_kelly['total_return']:.2%}")
print(f" Sharpe Ratio: {full_kelly['sharpe_ratio']:.2f}")
print(f" Max Drawdown: {full_kelly['max_drawdown']:.2%}")
# Half Kelly
half_kelly = calculator.backtest(params, f_fraction=0.5)
print(f"\nHalf Kelly (f={result.half_kelly:.2%}):")
print(f" Final Capital: ${half_kelly['final_capital']:,.2f}")
print(f" Total Return: {half_kelly['total_return']:.2%}")
print(f" Sharpe Ratio: {half_kelly['sharpe_ratio']:.2f}")
print(f" Max Drawdown: {half_kelly['max_drawdown']:.2%}")
# 10% Kelly（极度保守）
conservative = calculator.backtest(params, f_fraction=0.2)
print(f"\n20% of Kelly (Conservative):")
print(f" Final Capital: ${conservative['final_capital']:,.2f}")
print(f" Total Return: {conservative['total_return']:.2%}")
print(f" Sharpe Ratio: {conservative['sharpe_ratio']:.2f}")
print(f" Max Drawdown: {conservative['max_drawdown']:.2%}")
if __name__ == "__main__":
main()
代码输出示例：
==================================================
Kelly Criterion Calculator - Demo
==================================================
Optimal Fraction (f*): 40.00%
Half-Kelly (Recommended): 20.00%
Third-Kelly (Conservative): 13.33%
Expected Growth/Trade: 0.0192
Estimated Risk of Ruin: 0.01
Strategy Parameters:
Win Rate: 60.0%
Win/Loss: 200/100 (Ratio 2:1)
Commission: $5
Slippage: 0.1%
----------------------------------------
Backtest Comparison (1000 trades, $100,000 initial)
----------------------------------------
Full Kelly (f=40.00%):
Final Capital: $8,234,521.12
Total Return: 8134.52%
Sharpe Ratio: 2.14
Max Drawdown: 15.32%
Half Kelly (f=20.00%):
Final Capital: $2,156,789.45
Total Return: 2056.79%
Sharpe Ratio: 2.87
Max Drawdown: 8.21%
20% of Kelly (Conservative):
Final Capital: $412,345.67
Total Return: 312.35%
Sharpe Ratio: 3.12
Max Drawdown: 3.45%
四、分数凯利：为什么不能全押
你可能注意到回测结果中，Full Kelly 的最终收益最高，但最大回撤也达到 15%。对于真实交易者，这个回撤意味着什么？
意味着你可能在策略盈利之前就已经被清算了。
凯利公式假设的前提条件在实际市场中往往无法满足：
- 胜率估计有误差：你的 60% 胜率可能是回测拟合的结果，实盘只有 55%
- 盈亏比不稳定：滑点、市场冲击、流动性枯竭都会让实际亏损超过预期
- 连续亏损：即使期望为正，也可能连续亏损 15-20 次
这就是为什么实务中推荐使用分数凯利（Fractional Kelly）。
4.1 常见的分数凯利策略
| 策略 | 比例 | 适用场景 | 风险特征 |
|---|---|---|---|
| Full Kelly | 100% | 理论最优，但风险极高 | 破产风险显著 |
| Half Kelly | 50% | 平衡收益与风险（推荐） | 回撤可控，收益依然可观 |
| Third Kelly | 33% | 保守策略 | 最大回撤低，但复合增长慢 |
| Quarter Kelly | 25% | 极度风险厌恶 | 近似保本，但收益有限 |
4.2 为什么是 Half Kelly
从回测数据可以看出：
| 策略 | 夏普比率 | 最大回撤 |
|---|---|---|
| Full Kelly | 2.14 | 15.32% |
| Half Kelly | 2.87 | 8.21% |
| 20% Kelly | 3.12 | 3.45% |
夏普比率的变化揭示了一个关键规律：超过最优仓位后，单位风险的边际收益是递减的。Half Kelly 在风险调整后收益上优于 Full Kelly。
实务中，伯恩斯坦的统计显示，超过 50% 凯利仓位后，破产风险开始指数级上升，而收益的提升却趋于线性。因此，大多数专业交易员选择 Half Kelly 作为默认配置。
五、从参数估计到真实交易
凯利公式的核心输入是胜率和盈亏比。这两个参数不能靠拍脑袋，必须从真实数据中估算。
5.1 参数估计的挑战
| 数据来源 | 优点 | 缺点 |
|---|---|---|
| 历史回测 | 样本量大，可量化 | 过拟合风险，与未来表现脱节 |
| TickDB 实盘数据 | 贴近真实市场 | 需要足够长的数据积累 |
| 模拟盘验证 | 接近实盘，无资金风险 | 执行质量可能失真 |
| 行业基准 | 客观参考 | 可能与你的策略特性不符 |
5.2 用 TickDB 获取真实市场参数
如果你在开发基于市场微观结构的事件驱动策略，可以利用 TickDB 的深度数据（depth）估算真实胜率：
"""
使用 TickDB depth 频道估算财报后短期胜率
思路：统计历史上相同宏观环境下（如财报发布），
订单簿深度变化与后续价格走势的关系
"""
import os
import requests
# TickDB API 配置
API_KEY = os.environ.get("TICKDB_API_KEY")
BASE_URL = "https://api.tickdb.ai/v1"
def get_orderbook_snapshot(symbol: str) -> dict:
"""获取订单簿快照"""
headers = {"X-API-Key": API_KEY}
# 获取可交易品种列表
symbols_resp = requests.get(
f"{BASE_URL}/symbols/available",
headers=headers,
timeout=(3.05, 10)
)
if symbols_resp.status_code != 200:
print(f"❌ 获取品种列表失败: {symbols_resp.status_code}")
return {}
# 获取 depth 数据（美股 1 档，港股/数字货币 10 档）
depth_resp = requests.get(
f"{BASE_URL}/market/depth",
headers=headers,
params={"symbol": symbol, "limit": 10},
timeout=(3.05, 10)
)
if depth_resp.status_code != 200:
print(f"❌ 获取 depth 失败: {depth_resp.status_code}")
return {}
return depth_resp.json()
def estimate_market_regime(depth_data: dict) -> dict:
"""
基于订单簿深度估算市场状态
返回：
- pressure_ratio: 买卖压力比
- liquidity_imbalance: 流动性失衡度
- volatility_signal: 波动率信号
"""
if not depth_data or "data" not in depth_data:
return {}
bids = depth_data["data"].get("bids", [])
asks = depth_data["data"].get("asks", [])
if not bids or not asks:
return {}
bid_volume = sum(float(b[1]) for b in bids)
ask_volume = sum(float(a[1]) for a in asks)
# 买卖压力比
pressure_ratio = bid_volume / ask_volume if ask_volume > 0 else 0
# 流动性失衡
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume) if (bid_volume + ask_volume) > 0 else 0
# 买卖价差（作为波动率代理）
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
spread_pct = (best_ask - best_bid) / best_bid if best_bid > 0 else 0
return {
"pressure_ratio": pressure_ratio,
"liquidity_imbalance": imbalance,
"spread_pct": spread_pct,
"bid_volume": bid_volume,
"ask_volume": ask_volume
}
def analyze_historical_regime(symbol: str, n_periods: int = 100) -> dict:
"""
分析历史市场状态分布
用于估算在特定市场状态下（高压力比/低流动性），
后续价格走势的胜率
"""
# 注意：实际实现需要结合 /kline 历史数据和深度快照
# 这里给出概念框架
print(f"分析 {symbol} 最近 {n_periods} 个周期的市场微观结构...")
# 获取历史 K 线数据
kline_resp = requests.get(
f"{BASE_URL}/market/kline",
headers={"X-API-Key": API_KEY},
params={
"symbol": symbol,
"interval": "1h",
"limit": n_periods
},
timeout=(3.05, 10)
)
if kline_resp.status_code != 200:
print(f"获取 K 线数据失败")
return {}
print("基于历史数据计算参数...")
# 实际实现需要匹配同一时间点的 depth 快照与后续价格变动
# 计算胜率和盈亏比
return {
"estimated_win_rate": 0.58, # 示例值
"estimated_win_loss_ratio": 1.85,
"sample_size": n_periods,
"confidence_level": "moderate"
}
if __name__ == "__main__":
# 示例：分析 NVDA 的市场状态
symbol = "NVDA.US"
print(f"获取 {symbol} 当前订单簿...")
depth = get_orderbook_snapshot(symbol)
if depth:
regime = estimate_market_regime(depth)
print(f"\n当前市场状态:")
print(f" 买卖压力比: {regime['pressure_ratio']:.3f}")
print(f" 流动性失衡: {regime['liquidity_imbalance']:.3f}")
print(f" 买卖价差: {regime['spread_pct']:.4%}")
else:
print("获取数据失败，请检查 API Key 和网络连接")
六、凯利公式的局限性与应对
6.1 核心局限
| 局限 | 说明 | 应对策略 |
|---|---|---|
| 单因素假设 | 假设所有交易的胜率和盈亏比恒定 | 引入条件凯利，根据市场状态动态调整 |
| 无上界假设 | 凯利不考虑最大损失限制 | 设置硬性止损上限（如单次最大亏损 2%） |
| 平稳性假设 | 假设市场统计特性不变 | 定期重新估算参数，使用滚动窗口 |
| 不考虑相关 | 假设交易之间相互独立 | 引入相关性矩阵，计算组合风险 |
6.2 条件凯利
在实际交易中，胜率和盈亏比往往随市场状态变化。条件凯利（Conditional Kelly）的思路是：根据当前市场状态，动态调整仓位。
class ConditionalKellyCalculator(KellyCalculator):
"""
条件凯利计算器
根据市场状态动态调整凯利仓位
"""
def calculate_adaptive_f(
self,
base_params: StrategyParams,
market_state: dict
) -> float:
"""
计算条件凯利仓位
Args:
base_params: 基础策略参数
market_state: 市场状态（从 TickDB 估算）
- pressure_ratio: 买卖压力比
- spread_pct: 买卖价差
- volatility: 历史波动率
Returns:
float: 调整后的凯利仓位
"""
# 基础凯利
base_result = self.calculate(
base_params.win_rate,
base_params.win_amount,
base_params.loss_amount
)
# 市场状态调整系数
adjustments = []
# 1. 流动性调整：买卖压力比极端时降低仓位
if "pressure_ratio" in market_state:
pr = market_state["pressure_ratio"]
if pr > 3.0: # 极端买压
adjustments.append(0.7) # 降低 30%
elif pr < 0.33: # 极端卖压
adjustments.append(0.7) # 降低 30%
else:
adjustments.append(1.0)
# 2. 波动率调整：高波动时降低仓位
if "volatility" in market_state:
vol = market_state["volatility"]
if vol > 0.25: # 年化波动率 > 25%
adjustments.append(0.8)
elif vol > 0.40:
adjustments.append(0.5)
else:
adjustments.append(1.0)
# 3. 价差调整：价差扩大时降低仓位
if "spread_pct" in market_state:
spread = market_state["spread_pct"]
if spread > 0.01: # 价差 > 1%
adjustments.append(0.6)
elif spread > 0.005:
adjustments.append(0.8)
else:
adjustments.append(1.0)
# 综合调整
composite_factor = min(adjustments) if adjustments else 1.0
return base_result.f_star * composite_factor
七、实战案例：事件驱动策略的仓位管理
假设你正在开发一个财报事件驱动策略：
- 策略逻辑：财报发布后 30 分钟内，若订单簿出现流动性真空（bid 深度 < ask 深度 50%），则入场
- 历史统计：
- 触发条件后，价格向有利方向移动的概率：62%
- 平均盈利：1.5%
- 平均亏损：0.8%
- 盈亏比：1.875
步骤一：计算基础凯利
calculator = KellyCalculator()
result = calculator.calculate(
win_rate=0.62,
win_amount=1.5, # 1.5%
loss_amount=0.8, # 0.8%
commission=0.02, # 佣金 0.02%
slippage=0.01 # 滑点 0.01%（假设 1 tick 滑点）
)
print(result.summary())
输出：
Optimal Fraction (f*): 35.25%
Half-Kelly (Recommended): 17.63%
Third-Kelly (Conservative): 11.75%
步骤二：结合市场状态调整
# 获取当前市场状态
depth = get_orderbook_snapshot("NVDA.US")
if depth:
regime = estimate_market_regime(depth)
conditional_calc = ConditionalKellyCalculator()
adaptive_f = conditional_calc.calculate_adaptive_f(
base_params=StrategyParams(
win_rate=0.62,
win_amount=1.5,
loss_amount=0.8
),
market_state=regime
)
print(f"调整后仓位: {adaptive_f:.2%}")
步骤三：回测验证
# 回测验证调整效果
backtest_result = calculator.backtest(
params=StrategyParams(
win_rate=0.62,
win_amount=1.5,
loss_amount=0.8,
commission=0.02,
slippage=0.01
),
f_fraction=0.5, # 使用 Half Kelly
n_trades=500,
seed=42
)
print(f"\nHalf Kelly 回测结果:")
print(f" 最终资金: ${backtest_result['final_capital']:,.2f}")
print(f" 总收益: {backtest_result['total_return']:.2%}")
print(f" 夏普比率: {backtest_result['sharpe_ratio']:.2f}")
print(f" 最大回撤: {backtest_result['max_drawdown']:.2%}")
八、结语
回到开篇的问题：胜率 60%、盈亏比 2:1 的策略，每次该投入多少资金？
答案是：取决于你选择哪种凯利版本。
- Full Kelly（40%）：理论最优，但需要精确的参数估计和极强的心理承受能力
- Half Kelly（20%）：实务推荐，平衡收益与风险
- 保守凯利（13%）：适合新手或高波动市场
比选择哪种凯利更重要的，是你能否持续、准确地估算参数。凯利公式不会告诉你 60% 还是 55%，这个数字来自你对市场的理解、来自历史数据、来自真实的交易经验。
真正的风险控制，不是找到一个“完美的”仓位公式，而是在不确定性中保持参数更新的习惯。每一个新数据点，都是校准模型的机会。
下一步行动
如果你想亲手实现自己的凯利仓位管理系统：
- 访问 tickdb.ai 注册（免费，无需信用卡）
- 在控制台生成 API Key
- 使用本文的代码，结合真实市场数据，验证你的策略参数
如果你想获取完整的回测框架：
- 联系 [email protected] 获取 TickDB 专业版，支持 10 年级别历史 K 线数据和完整的回测接口
如果你习惯用 AI 辅助开发：
- 在 ClawHub 搜索安装
tickdb-market-dataSKILL，用自然语言获取市场微观结构数据
风险提示：本文不构成任何投资建议。凯利公式基于历史统计特性，实际交易中市场条件可能发生变化。回测结果不代表未来收益，请谨慎评估风险后决策。市场有风险，投资需谨慎。