你的第一个完整量化策略——从想法到回测的全流程实战
恭喜你，到这里你已经完成了数据篇的学习。从这一章开始，我们正式进入策略构建篇。你将要亲手打造你的第一个可运行的量化策略。它可能不是世界上最赚钱的策略，但它会是你在量化之路上最重要的里程碑——从"消费者"变成"创造者"的转折点。
1983年，著名的"海龟交易实验"中，理查德·丹尼斯把一群毫无交易经验的人关进一间屋子，教给他们简单的趋势跟踪规则。其中核心之一就是：当价格上穿20日高点时买入，跌破10日低点时卖出。几年后，这群"海龟"中的不少人成为了百万富翁。
而在太平洋另一端，一个中国的散户老张，2007年牛市时听朋友推荐买入了中信证券。他没有研究财报，也不懂K线形态，唯一的操作依据是：股价站上60日均线就持有，跌破就卖出。就这么一个简单的规则，让他在牛市中赚了3倍，又在2008年股灾前成功离场——只因为他的规则在跌破均线的那一天发出了卖出信号。
老张后来回忆："我当时根本不知道什么是量化交易。但现在回头看，我做的就是最原始的趋势跟踪策略。"
移动平均线（Moving Average, MA）是最古老、最常用的技术指标之一。它的计算非常简单：
N日移动平均线 = 过去N天收盘价的平均值
比如，5日均线就是把最近5天的收盘价加起来除以5。每天收盘后，剔除最老的一天，加入最新的一天，再算一次平均值。这个过程就像一辆公交车：到站时下一个人，上一个人，车里始终保持N个人。
均线的本质作用是平滑价格波动。短期均线（如5日、10日）紧跟价格，反应灵敏但噪声大；长期均线（如60日、120日）反应迟钝，但能过滤掉短期噪音，反映大趋势。
双均线交叉策略的核心逻辑就两句话：
背后的逻辑也很直觉：短期均线代表市场短期共识，长期均线代表长期共识。当短期共识超过长期共识时，说明 momentum 正在向上聚集，趋势可能开始；反之亦然。
双均线策略有两个核心参数：短期周期和长期周期。常见的组合有：
没有"最好"的参数组合，只有"最适合当前市场"的组合。这也是量化交易的精髓：用回测找出在历史数据上表现最好的参数，但心里清楚它不一定在未来继续有效。
均线交叉策略在趋势明确的市场中表现优异，但在震荡市中会反复被打脸。想象价格在区间内来回波动，短期均线会频繁上下穿越长期均线，导致连续多次假信号——买入后立刻死叉卖出，卖出后又金叉买入，来回割肉。
正方：回测数据显示，在A股和美股的大级别趋势行情中，均线交叉策略能显著跑赢买入持有。它的价值不在于每次都赚钱，而在于避开大跌。2008年和2015年股灾中，均线策略的卖出信号让持有者避开了60%以上的跌幅。
反方：在震荡市中，均线策略会产生大量假信号，交易成本高企。据统计，约70%的交易日市场处于无明显趋势状态，这意味着均线策略大部分时间都在"亏钱交易"。
量化交易者的务实态度：把均线交叉作为多策略组合的一部分，而不是唯一策略。同时加入过滤条件（如只在ADX>25的趋势市场中使用），可以大幅减少假信号。
在引入Backtrader之前，我们先手写一个最简版的均线交叉策略。这能帮你理解策略的每一个底层逻辑。
"""
双均线交叉策略 - 从零实现版
不使用任何回测框架，纯Python手写逻辑
"""
import pandas as pd
import numpy as np
import akshare as ak
def get_data(symbol="000001", start="20220101", end="20240101"):
"""获取股票历史数据"""
df = ak.stock_zh_a_hist(symbol=symbol, period="daily",
start_date=start, end_date=end, adjust="qfq")
df['日期'] = pd.to_datetime(df['日期'])
df = df.sort_values('日期').reset_index(drop=True)
return df
def moving_average(prices, window):
"""计算简单移动平均线"""
return prices.rolling(window=window).mean()
def backtest_ma_cross(df, short_window=10, long_window=30,
initial_capital=100000):
"""
双均线交叉回测
参数:
short_window: 短期均线周期
long_window: 长期均线周期
initial_capital: 初始资金
"""
df = df.copy()
df['short_ma'] = moving_average(df['收盘'], short_window)
df['long_ma'] = moving_average(df['收盘'], long_window)
# 生成信号: 1=多头, -1=空头, 0=空仓
df['position'] = 0
df['signal'] = 0
for i in range(1, len(df)):
if i < long_window:
continue
prev_short = df['short_ma'].iloc[i-1]
prev_long = df['long_ma'].iloc[i-1]
curr_short = df['short_ma'].iloc[i]
curr_long = df['long_ma'].iloc[i]
# 金叉: 短期从下向上穿越长期
if prev_short <= prev_long and curr_short > curr_long:
df.loc[i, 'signal'] = 1
df.loc[i, 'position'] = 1
# 死叉: 短期从上向下穿越长期
elif prev_short >= prev_long and curr_short < curr_long:
df.loc[i, 'signal'] = -1
df.loc[i, 'position'] = 0
else:
# 保持上一期仓位
df.loc[i, 'position'] = df['position'].iloc[i-1]
# 计算每日收益率
df['daily_return'] = df['收盘'].pct_change()
df['strategy_return'] = df['position'].shift(1) * df['daily_return']
# 计算累计收益
df['cum_market'] = (1 + df['daily_return']).cumprod()
df['cum_strategy'] = (1 + df['strategy_return']).cumprod()
# 统计指标
total_return = df['cum_strategy'].iloc[-1] - 1
market_return = df['cum_market'].iloc[-1] - 1
trades = df[df['signal'] != 0]
print(f"=== 回测结果 ({short_window}日/{long_window}日) ===")
print(f"策略总收益: {total_return*100:.2f}%")
print(f"市场总收益: {market_return*100:.2f}%")
print(f"交易次数: {len(trades)}")
print(f"胜率: {(trades['signal']==1).sum()}/{len(trades)} (买入/总交易)")
return df
# 运行回测
if __name__ == "__main__":
df = get_data(symbol="000001", start="20220101", end="20240101")
result = backtest_ma_cross(df, short_window=10, long_window=30)
# 查看最近几次交易信号
signals = result[result['signal'] != 0][['日期','收盘','short_ma','long_ma','signal']].tail(10)
print("\n=== 最近交易信号 ===")
print(signals.to_string(index=False))上面的手写版虽然直观，但缺少很多实战要素：手续费、滑点、仓位管理、详细的结果分析。下面我们用Backtrader来实现一个生产级的均线交叉策略：
"""
双均线交叉策略 - Backtrader完整版
安装依赖: pip install backtrader akshare pandas
"""
import backtrader as bt
import akshare as ak
import pandas as pd
from datetime import datetime
class SmaCrossStrategy(bt.Strategy):
"""
双均线交叉策略
"""
params = (
('short_period', 10), # 短期均线周期
('long_period', 30), # 长期均线周期
('printlog', False), # 是否打印日志
)
def __init__(self):
# 计算两条移动平均线
self.short_ma = bt.indicators.SimpleMovingAverage(
self.data.close, period=self.p.short_period
)
self.long_ma = bt.indicators.SimpleMovingAverage(
self.data.close, period=self.p.long_period
)
# 交叉信号指标
self.crossover = bt.indicators.CrossOver(self.short_ma, self.long_ma)
# 记录订单状态
self.order = None
def next(self):
"""每个交易日调用一次"""
# 如果有待执行订单，跳过
if self.order:
return
# 金叉且空仓 -> 买入
if self.crossover > 0 and not self.position:
self.order = self.buy()
if self.p.printlog:
print(f'BUY @ {self.data.close[0]:.2f}')
# 死叉且持仓 -> 卖出
elif self.crossover < 0 and self.position:
self.order = self.sell()
if self.p.printlog:
print(f'SELL @ {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():
print(f'买入执行: 价格={order.executed.price:.2f}, '
f'数量={order.executed.size}')
else:
print(f'卖出执行: 价格={order.executed.price:.2f}, '
f'数量={order.executed.size}')
self.order = None
def run_backtest():
"""运行回测"""
# 创建Cerebro引擎
cerebro = bt.Cerebro()
# 添加策略
cerebro.addstrategy(SmaCrossStrategy,
short_period=10,
long_period=30,
printlog=True)
# 获取数据
df = ak.stock_zh_a_hist(symbol="000001", period="daily",
start_date="20220101", end_date="20240101",
adjust="qfq")
df['日期'] = pd.to_datetime(df['日期'])
df = df.sort_values('日期')
df.set_index('日期', inplace=True)
# 转换为Backtrader数据格式
data = bt.feeds.PandasData(
dataname=df,
datetime=None, # 使用索引作为日期
open='开盘',
high='最高',
low='最低',
close='收盘',
volume='成交量',
openinterest=-1
)
cerebro.adddata(data)
# 设置初始资金
cerebro.broker.setcash(100000.0)
# 设置手续费 (A股: 万分之3，最低5元)
cerebro.broker.setcommission(commission=0.0003, mincommission=5)
# 添加分析器
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
# 运行回测
print(f'初始资金: {cerebro.broker.getvalue():.2f}')
results = cerebro.run()
strat = results[0]
print(f'最终资金: {cerebro.broker.getvalue():.2f}')
print(f'总收益率: {(cerebro.broker.getvalue()/100000 - 1)*100:.2f}%')
# 输出分析结果
sharpe = strat.analyzers.sharpe.get_analysis()
drawdown = strat.analyzers.drawdown.get_analysis()
returns = strat.analyzers.returns.get_analysis()
print(f"\n=== 绩效报告 ===")
print(f"夏普比率: {sharpe.get('sharperatio', 'N/A')}")
print(f"最大回撤: {drawdown.max.drawdown:.2f}%")
print(f"年化收益: {returns.get('rnorm100', 'N/A')}")
# 绘制结果 (需要在本地Jupyter或Python环境中运行)
# cerebro.plot(style='candlestick')
if __name__ == "__main__":
run_backtest()下面这个模拟器可以让你直观地看到不同参数组合下，均线交叉信号是如何产生的。调整滑块，观察信号频率的变化：
Prompt模板：
请用Backtrader写一个双均线交叉策略，要求：
1. 短期均线5日，长期均线20日
2. 金叉时全仓买入，死叉时全仓卖出
3. 初始资金10万元，手续费万分之3
4. 使用AKShare获取贵州茅台(600519)从2020年到2024年的日线数据
5. 输出夏普比率、最大回撤、总收益率
6. 添加注释解释每一行关键代码Prompt模板：
请逐行解释以下Backtrader策略代码的每个部分：
[粘贴上面代码中的策略类]
要求：
1. 解释__init__、next、notify_order三个方法的作用
2. 解释self.position的含义
3. 解释为什么用self.order来判断是否有未执行订单
4. 解释CrossOver指标的原理Prompt模板：
请在刚才的双均线策略基础上，添加以下风控规则：
1. 单笔亏损超过5%时自动止损卖出
2. 单笔盈利超过15%时止盈卖出50%仓位
3. 最大回撤超过20%时停止开新仓
请给出修改后的完整代码，并用注释标出新增的部分。同一个策略，不同的参数，结果可能天差地别。下面这段代码可以帮你快速对比多组参数的表现：
"""
双均线参数对比测试
一次性测试多组参数，找出历史表现最好的组合
"""
import backtrader as bt
import akshare as ak
import pandas as pd
class SmaCrossStrategy(bt.Strategy):
params = (
('short_period', 10),
('long_period', 30),
)
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.order = None
def next(self):
if self.order: return
if self.crossover > 0 and not self.position:
self.order = self.buy()
elif self.crossover < 0 and self.position:
self.order = self.sell()
def notify_order(self, order):
if order.status in [order.Completed]:
self.order = None
def stop(self):
# 回测结束时记录最终资金
self.final_value = self.broker.getvalue()
def run_param_scan():
# 参数组合列表
param_grid = [
(5, 20), (10, 20), (10, 30), (10, 60),
(20, 60), (20, 120), (50, 200)
]
# 获取数据
df = ak.stock_zh_a_hist(symbol="000001", period="daily",
start_date="20200101", end_date="20240101",
adjust="qfq")
df['日期'] = pd.to_datetime(df['日期'])
df = df.sort_values('日期')
df.set_index('日期', inplace=True)
results = []
for short, long in param_grid:
cerebro = bt.Cerebro()
cerebro.addstrategy(SmaCrossStrategy, short_period=short, long_period=long)
data = bt.feeds.PandasData(
dataname=df, datetime=None,
open='开盘', high='最高', low='最低',
close='收盘', volume='成交量', openinterest=-1
)
cerebro.adddata(data)
cerebro.broker.setcash(100000.0)
cerebro.broker.setcommission(commission=0.0003)
strat = cerebro.run()[0]
final = strat.final_value
ret = (final / 100000 - 1) * 100
results.append({
'短期': short,
'长期': long,
'最终资金': final,
'收益率%': round(ret, 2)
})
print(f"{short}/{long}: 收益率 {ret:.2f}%")
# 输出最优参数
best = max(results, key=lambda x: x['收益率%'])
print(f"\n最优参数: {best['短期']}日/{best['长期']}日, 收益率: {best['收益率%']}%")
return pd.DataFrame(results)
if __name__ == "__main__":
run_param_scan()恭喜你！你已经拥有了自己的第一个完整量化策略。但一个成熟的策略远不只"金叉买、死叉卖"这么简单。真正的策略是一个精密的系统，包含入场规则、出场规则、仓位管理和止损止盈四大核心组件。
下一章《策略的骨架——入场、出场、仓位、止损》，我们将为你的策略装上"骨架"和"神经系统"。你会学到：如何设计多种出场方案？如何计算最优仓位？移动止损怎么实现？准备好让你的策略从"毛坯房"升级成"精装房"了吗？
如果这本书对你有所启发，欢迎支持我继续创作
