故事引入:那个亏掉五百万的"天才策略"
2015年,某私募基金经理开发了一个看起来非常漂亮的策略:基于机器学习预测次日涨跌,准确率超过65%。回测显示,三年年化收益可达80%。
基金成立后前三个月,策略确实表现优异,净值涨了30%。然后,股灾来了。策略在连续暴跌中依然坚持"预测会反弹"的判断,满仓持有。结果呢?一个月内净值腰斩。最终基金清盘,投资者血本无归。
事后复盘,这个策略的问题根本不在预测模型——65%的胜率已经是不错的水平。问题出在:它没有止损、没有仓位管理、没有强制出场规则。一个漂亮的脑袋,长在了没有痛感的身体上。
在量化交易中,风控不是策略的"配件",而是策略的"器官"。没有止损的策略,就像没有免疫系统的身体——一个小小的病毒(黑天鹅事件)就能致命。
核心知识:策略的四大核心组件
2.1 入场规则:什么时候开枪?
入场规则回答的问题是:在什么条件下开仓? 它由两部分组成:
- 信号条件:技术指标触发、基本面满足、或者模型预测
- 过滤条件:大盘环境、板块效应、波动率水平等辅助判断
一个常见的错误是过度复杂化入场条件。新手总想把尽可能多的条件都满足才开仓,结果导致信号过于稀少,甚至错过整个趋势。量化交易的本质是在概率优势下频繁下注,而不是追求"必胜"的入场时机。
flowchart TD
A[策略启动] --> B{入场信号?}
B -->|是| C{过滤条件通过?}
C -->|是| D[计算仓位]
C -->|否| E[等待]
B -->|否| E
D --> F{风险检查?}
F -->|通过| G[执行买入]
F -->|不通过| H[降低仓位或放弃]
style G fill:#10b981,color:#fff
style H fill:#e74c3c,color:#fff
2.2 出场规则:什么时候收网?
很多交易者把80%的精力放在研究"怎么买"上,却只花20%的时间思考"怎么卖"。这是本末倒置。出场规则往往比入场规则更重要,因为它直接决定了你能不能把账面利润变成实际利润。
出场规则分为三类:
- 止盈出场:固定比例止盈(如盈利15%卖出)、跟踪止盈(回撤5%时卖出最高点的盈利部分)
- 止损出场:固定止损(亏损8%卖出)、技术止损(跌破支撑位卖出)
- 时间出场:持仓超过N天无论盈亏都卖出(防止资金被无效占用)
一个稳健的出场体系通常会同时包含这三类规则,谁先触发谁执行。这就像一个多保险的安全系统。
2.3 仓位管理:买多少?
仓位管理是量化交易中最被低估的技能。同样一个策略,满仓运行和只用20%资金运行,风险收益特征完全不同。
📐 凯利公式:数学给出的最优仓位
点击展开公式和计算示例
凯利公式:f* = (bp - q) / b
其中:
- f*:最优仓位比例
- b:盈亏比(平均盈利 / 平均亏损)
- p:胜率
- q:败率(q = 1 - p)
举例:某策略胜率55%,盈亏比2:1,则 f* = (2 * 0.55 - 0.45) / 2 = 0.325,即最优仓位为32.5%。
注意:实际交易中很少用"全凯利"仓位,因为凯利公式对胜率估计误差非常敏感。通常使用"半凯利"(16%)甚至更保守的比例。
⚖ 风险平价:每只票承担同样的风险
点击展开原理和实现
风险平价的核心思想是:不是给每只票分配同样的资金,而是让每只票对组合总风险的贡献相同。
实现方法:根据每只股票的波动率(标准差)来反向分配仓位。波动率越大的股票,分配的仓位越小。
公式:仓位_i = (1/σ_i) / Σ(1/σ_j)
其中σ_i是第i只股票的预期波动率。这种方法天然地让高波动股票(如小盘股、科技股)获得较低权重,低波动股票(如银行股、公用事业)获得较高权重。
2.4 止损:策略的免疫系统
止损是交易中最反人性但也最重要的环节。没有人能每次都判断正确,止损的作用就是确保错误的代价可控。
01
固定比例止损
买入价下跌X%即止损。最简单,最常用。适合大多数趋势策略。
02
跟踪止损(移动止损)
记录最高价,从最高点回撤X%即卖出。让利润奔跑的同时保护已有盈利。
03
ATR止损
基于平均真实波幅(ATR)设置止损。适应不同波动率的市场环境。
04
时间止损
持仓超过N天无论盈亏都卖出。防止资金被无效占用,特别适合短线策略。
2.5 完整策略骨架代码示例
下面是一个包含四大核心组件的完整Backtrader策略。这大约是你见过的最"实用主义"的策略模板:
"""
完整策略骨架 - 包含入场、出场、仓位、止损四大组件
基于Backtrader框架
"""
import backtrader as bt
import numpy as np
class CompleteStrategy(bt.Strategy):
params = (
('short_period', 10),
('long_period', 30),
('stop_loss', 0.08), # 固定止损: 8%
('trailing_stop', 0.05), # 跟踪止损: 从最高点回撤5%
('take_profit', 0.15), # 止盈: 15%
('time_stop', 20), # 时间止损: 20天
('risk_per_trade', 0.02), # 每笔交易风险2%资金
('max_position', 0.5), # 单票最大仓位50%
)
def __init__(self):
self.short_ma = bt.ind.SMA(period=self.p.short_period)
self.long_ma = bt.ind.SMA(period=self.p.long_period)
self.crossover = bt.ind.CrossOver(self.short_ma, self.long_ma)
self.atr = bt.ind.ATR(period=14)
self.order = None
self.entry_price = None
self.highest_price = None
self.entry_day = None
def next(self):
if self.order:
return
# ========== 出场检查(优先于入场)==========
if self.position:
current_price = self.data.close[0]
days_held = len(self.data) - self.entry_day
# 1. 固定止损
if current_price < self.entry_price * (1 - self.p.stop_loss):
self.order = self.sell()
print(f'固定止损 @ {current_price:.2f}')
return
# 2. 跟踪止损
self.highest_price = max(self.highest_price, current_price)
if current_price < self.highest_price * (1 - self.p.trailing_stop):
self.order = self.sell()
print(f'跟踪止损 @ {current_price:.2f} (高点:{self.highest_price:.2f})')
return
# 3. 止盈
if current_price > self.entry_price * (1 + self.p.take_profit):
self.order = self.sell()
print(f'止盈 @ {current_price:.2f}')
return
# 4. 时间止损
if days_held >= self.p.time_stop:
self.order = self.sell()
print(f'时间止损 @ {current_price:.2f} (持仓{days_held}天)')
return
# ========== 入场逻辑 ==========
if self.crossover > 0 and not self.position:
# 计算仓位: 风险平价思路
cash = self.broker.getcash()
risk_amount = cash * self.p.risk_per_trade
stop_distance = self.p.stop_loss * self.data.close[0]
if stop_distance > 0:
size = int(risk_amount / stop_distance)
max_size = int(cash * self.p.max_position / self.data.close[0])
size = min(size, max_size)
if size > 0:
self.order = self.buy(size=size)
print(f'买入 {size}股 @ {self.data.close[0]:.2f}')
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.entry_price = order.executed.price
self.highest_price = order.executed.price
self.entry_day = len(self.data)
print(f'买入完成: 价格={order.executed.price:.2f}')
else:
self.entry_price = None
self.highest_price = None
self.entry_day = None
print(f'卖出完成: 价格={order.executed.price:.2f}')
self.order = None
# ========== 使用示例 ==========
if __name__ == "__main__":
cerebro = bt.Cerebro()
cerebro.addstrategy(CompleteStrategy)
# 添加数据 (请替换为实际数据)
# data = bt.feeds.YahooFinanceData(dataname='AAPL', fromdate=datetime(2020,1,1))
# cerebro.adddata(data)
cerebro.broker.setcash(100000.0)
cerebro.broker.setcommission(commission=0.0003)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
print('初始资金: %.2f' % cerebro.broker.getvalue())
results = cerebro.run()
strat = results[0]
print('最终资金: %.2f' % cerebro.broker.getvalue())
sharpe = strat.analyzers.sharpe.get_analysis()
drawdown = strat.analyzers.drawdown.get_analysis()
print(f"夏普比率: {sharpe.get('sharperatio', 'N/A')}")
print(f"最大回撤: {drawdown.max.drawdown:.2f}%")
2.6 仓位计算交互器
下面这个计算器可以帮你快速估算在给定胜率和盈亏比下的最优仓位和风险:
动手实践:给你的策略装上"免疫系统"
下面的代码是一个可以直接运行的"策略骨架模板"。你可以把它当作未来开发所有策略的基础:
"""
策略骨架模板 v1.0
一个包含完整风控的Backtrader策略基类
你可以继承这个类,只重写generate_signal方法即可
"""
import backtrader as bt
class StrategySkeleton(bt.Strategy):
"""
策略骨架基类
子类只需重写 generate_signal() 方法,返回:
1 -> 买入信号
-1 -> 卖出信号
0 -> 无信号
"""
params = (
('stop_loss', 0.08),
('trailing_stop', 0.05),
('take_profit', 0.15),
('time_stop', 20),
('risk_per_trade', 0.02),
('max_position', 0.5),
)
def __init__(self):
self.order = None
self.entry_price = None
self.highest_price = None
self.entry_bar = None
def generate_signal(self):
"""
子类重写此方法
"""
return 0
def calculate_position_size(self, price, stop_distance):
"""根据风险计算仓位"""
cash = self.broker.getcash()
risk_amount = cash * self.p.risk_per_trade
if stop_distance <= 0:
return 0
size = int(risk_amount / stop_distance)
max_size = int(cash * self.p.max_position / price)
return min(size, max_size)
def check_exit(self):
"""检查是否需要出场"""
if not self.position:
return False
price = self.data.close[0]
days = len(self.data) - self.entry_bar
# 固定止损
if price < self.entry_price * (1 - self.p.stop_loss):
self.sell(); return True
# 跟踪止损
self.highest_price = max(self.highest_price, price)
if price < self.highest_price * (1 - self.p.trailing_stop):
self.sell(); return True
# 止盈
if price > self.entry_price * (1 + self.p.take_profit):
self.sell(); return True
# 时间止损
if days >= self.p.time_stop:
self.sell(); return True
return False
def next(self):
if self.order:
return
# 优先检查出场
if self.check_exit():
return
# 检查入场信号
signal = self.generate_signal()
if signal == 1 and not self.position:
price = self.data.close[0]
stop_dist = price * self.p.stop_loss
size = self.calculate_position_size(price, stop_dist)
if size > 0:
self.order = self.buy(size=size)
elif signal == -1 and self.position:
self.order = self.sell()
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.entry_price = order.executed.price
self.highest_price = order.executed.price
self.entry_bar = len(self.data)
else:
self.entry_price = None
self.highest_price = None
self.entry_bar = None
self.order = None
# ========== 使用示例: 均线交叉策略继承骨架 ==========
class MyStrategy(StrategySkeleton):
params = (('short', 10), ('long', 30))
def __init__(self):
super().__init__()
self.short_ma = bt.ind.SMA(period=self.p.short)
self.long_ma = bt.ind.SMA(period=self.p.long)
self.crossover = bt.ind.CrossOver(self.short_ma, self.long_ma)
def generate_signal(self):
if self.crossover > 0:
return 1
elif self.crossover < 0:
return -1
return 0
if __name__ == "__main__":
cerebro = bt.Cerebro()
cerebro.addstrategy(MyStrategy)
cerebro.broker.setcash(100000.0)
cerebro.broker.setcommission(commission=0.0003)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
print('初始资金: %.2f' % cerebro.broker.getvalue())
results = cerebro.run()
strat = results[0]
print('最终资金: %.2f' % cerebro.broker.getvalue())
print(f"夏普: {strat.analyzers.sharpe.get_analysis().get('sharperatio','N/A')}")
print(f"最大回撤: {strat.analyzers.drawdown.get_analysis().max.drawdown:.2f}%")
骨架模板的使用方法
- 继承:创建新策略时,继承
StrategySkeleton类
- 重写:只需重写
generate_signal()方法,其他风控自动生效
- 调参:通过params调整止损比例、止盈比例、时间止损等参数
- 扩展:如需更复杂的仓位计算,重写
calculate_position_size()
知识自测
快速检验你的理解
1. 为什么出场规则往往比入场规则更重要?
出场规则更容易编程实现
出场直接决定利润能否落袋为安
入场规则不需要思考
出场规则决定了胜率
入场只是开始,出场才决定最终结果。再好的入场,如果没有合理的出场规则,可能让盈利变成亏损,或者让小额亏损变成巨额亏损。
2. 凯利公式计算出的仓位通常需要打折使用,为什么?
因为胜率估计可能有误差,公式对误差很敏感
因为凯利公式只适用于赌场
因为打折后的收益更高
因为交易所不允许满仓
凯利公式假设胜率和盈亏比是精确已知的。但在实际交易中,这些参数只能从有限的历史数据中估计,存在统计误差。全凯利仓位在参数估计偏差下可能导致毁灭性后果,因此通常使用半凯利或更低的比例。
3. 跟踪止损(Trailing Stop)的最大优势是什么?
它能在最低点附近止损
它不需要设置任何参数
它让利润奔跑的同时保护已有盈利
它只在牛市中有效
跟踪止损的独特优势在于"动态调整"。随着价格上涨,止损位也会跟着上移,既给趋势继续发展的空间,又在趋势反转时锁定大部分利润。它是"让利润奔跑"理念的最佳实践之一。
4. 时间止损的主要作用是什么?
增加交易频率
防止资金被无效占用过久
提高胜率
减少手续费
时间止损解决的是"机会成本"问题。如果一笔交易在预定时间内没有按预期方向发展,说明初始判断可能错误,或者市场条件已经变化。及时退出可以释放资金去寻找更好的机会。