从零到一：我的第一个量化交易策略开发实战

Python►量化交易 评论

“代码不会说谎，但市场会教你谦逊。” —— 一个量化新手的感悟

🌟 前言：量化之路的起点

作为一名编程爱好者，我一直对量化交易充满好奇。看着那些复杂的数学模型和算法在金融市场中驰骋，总觉得自己离这个领域很遥远。直到今天，我完成了自己的第一个完整的量化交易策略——基于Backtrader框架的农业银行(601288)交易策略。

这篇文章记录了我从环境搭建到策略开发的全过程，包括遇到的坑、解决思路，以及完整的代码实现。希望能帮助更多想要入门量化的朋友少走弯路。

🛠️ 环境搭建：选择正确的工具

为什么选择Backtrader？

在开始之前，我调研了几个量化框架：

  • Zipline：功能强大但文档较少，社区相对冷清
  • vn.py：国内优秀框架，但学习曲线较陡
  • Backtrader：文档完善、社区活跃、Python原生API友好

最终选择了Backtrader，因为它对新手最友好，而且有大量中文教程。

安装步骤

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 创建项目目录
mkdir QuantGWB && cd QuantGWB/backtrader

# 使用Python创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Mac/Linux
# .venv\Scripts\activate   # Windows

# 安装依赖包
pip install backtrader akshare pandas matplotlib yfinance

# 创建项目配置文件
cat > pyproject.toml << EOF
[project]
name = "agricultural-bank-quant-strategy"
version = "1.0.0"
dependencies = [
    "backtrader>=1.9.78",
    "akshare>=1.11.0",
    "pandas>=2.0.0",
    "matplotlib>=3.7.0",
    "yfinance>=0.2.0"
]
EOF

📊 策略思路：技术分析的回归

策略原理

我选择了一个相对简单的技术分析策略，基于以下几个核心思想：

  1. 趋势判断：使用移动平均线判断市场趋势
  2. 买入信号：短期趋势向上 + 突破长期趋势
  3. 风险控制：多层次止盈止损机制

技术指标选择

  • MA3, MA5, MA10, MA60：多周期移动平均线
  • MA5 > MA10：短期趋势向上
  • MA3上穿MA60：短期突破长期

🔧 策略开发：从V1到V3的演进

V1版本：基础框架

第一个版本很简单，只有基本的买卖逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
class ShanxiFenjiuStrategy(bt.Strategy):
    params = (('ma5_period', 5), ('ma10_period', 10))

    def __init__(self):
        self.ma5 = bt.indicators.SMA(self.data.close, period=self.params.ma5_period)
        self.ma10 = bt.indicators.SMA(self.data.close, period=self.params.ma10_period)

    def next(self):
        if not self.position and self.ma5[0] > self.ma10[0]:
            self.buy()
        elif self.position and self.ma5[0] < self.ma10[0]:
            self.sell()

V2版本：参数化改进

很快发现硬编码参数不方便调优，于是实现了参数化：

1
2
3
4
5
6
7
8
params = (
    ('ma5_period', 5),
    ('ma10_period', 10),
    ('first_profit_target', 0.10),
    ('protection_profit_target', 0.08),
    ('mobile_profit_start', 0.15),
    ('mobile_retracement', 0.05),
)

V3版本：逻辑完善

最重要的改进是修复了止盈止损的逻辑问题：

🎯 V3完整策略代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
农业银行量化交易策略 V3版本
作者：GWB
创建时间：2024-11-11
"""

from __future__ import print_function
import backtrader as bt
import pandas as pd
import numpy as np
from datetime import datetime
import argparse

class AgriculturalBankStrategyV3(bt.Strategy):
    """
    农业银行量化交易策略 V3版本
    核心特性：完整的止盈止损体系 + 移动止盈机制
    """

    params = (
        # 均线参数
        ('ma3_period', 2),
        ('ma5_period', 5),
        ('ma10_period', 10),
        ('ma60_period', 60),

        # 止盈止损参数
        ('first_profit_target', 0.10),    # 第一层止盈10%
        ('protection_profit_target', 0.08),  # 保护线8%
        ('mobile_profit_start', 0.15),      # 移动止盈启动15%
        ('mobile_retracement', 0.05),       # 移动止盈回撤5%
        ('stop_loss_pct', -0.02),            # 硬止损-2%

        ('printlog', True),
        ('initial_size', 0.95),
    )

    def __init__(self):
        # 计算技术指标
        self.ma3 = bt.indicators.SMA(self.data.close, period=self.params.ma3_period)
        self.ma5 = bt.indicators.SMA(self.data.close, period=self.params.ma5_period)
        self.ma10 = bt.indicators.SMA(self.data.close, period=self.params.ma10_period)
        self.ma60 = bt.indicators.SMA(self.data.close, period=self.params.ma60_period)

        # 仓位信息跟踪
        self.position_info = {
            'entry_price': 0,
            'half_sold': False,
            'highest_price': 0,
            'mobile_stop_price': 0,
            'mobile_stop_active': False,
        }

        # 交易统计
        self.trades = 0
        self.wins = 0

    def log(self, txt, dt=None):
        if self.params.printlog:
            dt = dt or self.datas[0].datetime.date(0)
            price = float(self.data.close[0])
            print(f'{dt.isoformat()} 价格:{price:.2f} {txt}')

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return

        if order.status in [order.Completed]:
            if order.isbuy():
                self.position_info['entry_price'] = order.executed.price
                self.position_info['half_sold'] = False
                self.position_info['highest_price'] = order.executed.price
                self.position_info['mobile_stop_active'] = False
                self.trades += 1
                profit_pct = (self.data.close[0] - order.executed.price) / order.executed.price
                self.log(f'🚀 买入 - 价格:{order.executed.price:.2f} 数量:{order.executed.size}')

            elif order.issell():
                if self.position_info['entry_price'] > 0:
                    profit_pct = (order.executed.price - self.position_info['entry_price']) / self.position_info['entry_price']
                    if profit_pct > 0:
                        self.wins += 1
                    self.log(f'💰 卖出 - 价格:{order.executed.price:.2f} 数量:{order.executed.size} 总盈亏:{profit_pct:+.2%}')

                if not self.position:
                    self.position_info = {
                        'entry_price': 0, 'half_sold': False,
                        'highest_price': 0, 'mobile_stop_price': 0,
                        'mobile_stop_active': False,
                    }

    def check_buy_signal(self):
        if len(self) < self.params.ma60_period:
            return False

        ma5 = float(self.ma5[0])
        ma10 = float(self.ma10[0])
        ma3_current = float(self.ma3[0])
        ma3_prev = float(self.ma3[-2]) if len(self) > 1 else ma3_current
        ma60_current = float(self.ma60[0])
        ma60_prev = float(self.ma60[-2]) if len(self) > 1 else ma60_current

        # 买入条件：短期趋势向上 + 突破长期趋势
        condition1 = ma5 > ma10
        condition2 = (ma3_prev < ma60_prev) and (ma3_current > ma60_current)

        return condition1 and condition2

    def check_sell_signal(self):
        if len(self) < self.params.ma10_period:
            return False

        ma5 = float(self.ma5[0])
        ma10 = float(self.ma10[0])
        ma_diff_pct = (ma5 - ma10) / ma10

        return ma_diff_pct < -0.02  # MA5-MA10差值小于-2%

    def process_mobile_stop_loss(self, current_price, profit_pct):
        # 移动止盈逻辑
        if profit_pct >= self.params.mobile_profit_start:
            if current_price > self.position_info['highest_price']:
                self.position_info['highest_price'] = current_price

            mobile_stop_price = self.position_info['highest_price'] * (1 - self.params.mobile_retracement)
            self.position_info['mobile_stop_price'] = mobile_stop_price
            self.position_info['mobile_stop_active'] = True

            self.log(f'📊 移动止盈监控中: 最高价{self.position_info["highest_price"]:.2f} 止盈线{mobile_stop_price:.2f}')

        if (self.position_info['mobile_stop_active'] and
            current_price < self.position_info['mobile_stop_price']):
            retracement_amt = (self.position_info['highest_price'] - current_price) / self.position_info['highest_price']
            return 'mobile_stop', {
                'reason': f'移动止盈触发：从最高价回撤{retracement_amt:.1%}超过{self.params.mobile_retracement:.0%}',
                'profit_pct': profit_pct
            }

        return None

    def check_all_exit_conditions(self):
        if not self.position or self.position_info['entry_price'] == 0:
            return None, {}

        current_price = float(self.data.close[0])
        entry_price = self.position_info['entry_price']
        profit_pct = (current_price - entry_price) / entry_price

        # 1. 硬止损（最高优先级）
        if profit_pct <= self.params.stop_loss_pct:
            return 'stop_loss', {
                'reason': f'🛑 硬止损：亏损{abs(profit_pct):.1%}超过{abs(self.params.stop_loss_pct):.0%}',
                'profit_pct': profit_pct
            }

        # 2. 第一层止盈
        if profit_pct >= self.params.first_profit_target and not self.position_info['half_sold']:
            return 'take_profit_half', {
                'reason': f'💰 第一层止盈：上涨{profit_pct:.1%}达到{self.params.first_profit_target:.0%}',
                'profit_pct': profit_pct
            }

        # 3. 如果已卖出一半，检查后续止盈条件
        if self.position_info['half_sold']:
            # 3a. 保护止盈
            if profit_pct < self.params.protection_profit_target:
                return 'protection_stop', {
                    'reason': f'🛡️ 保护止盈：涨幅回撤至{profit_pct:.1%}低于{self.params.protection_profit_target:.0%}',
                    'profit_pct': profit_pct
                }

            # 3b. 移动止盈
            mobile_result = self.process_mobile_stop_loss(current_price, profit_pct)
            if mobile_result:
                return mobile_result

        # 4. 技术卖出信号
        if self.check_sell_signal():
            ma5_val = float(self.ma5[0])
            ma10_val = float(self.ma10[0])
            ma_diff_pct = (ma5_val - ma10_val) / ma10_val * 100
            return 'technical_sell', {
                'reason': f'📉 技术卖出：MA5({ma5_val:.2f})与MA10({ma10_val:.2f})差值{ma_diff_pct:.1f}% < -2%',
                'profit_pct': profit_pct
            }

        return None, {}

    def next(self):
        current_price = float(self.data.close[0])

        # 检查退出条件
        if self.position:
            exit_condition, exit_info = self.check_all_exit_conditions()

            if exit_condition:
                if exit_condition == 'stop_loss':
                    self.close()
                    self.log(exit_info['reason'])
                elif exit_condition == 'take_profit_half':
                    sell_size = int(self.position.size / 2)
                    if sell_size > 0:
                        self.sell(size=sell_size)
                        self.position_info['half_sold'] = True
                        self.log(exit_info['reason'])
                elif exit_condition == 'protection_stop':
                    self.close()
                    self.log(exit_info['reason'])
                elif exit_condition == 'mobile_stop':
                    self.close()
                    self.log(exit_info['reason'])
                elif exit_condition == 'technical_sell':
                    self.close()
                    self.log(exit_info['reason'])
                return

        # 检查买入信号
        if not self.position and self.check_buy_signal():
            cash = self.broker.cash
            size = int(cash * self.params.initial_size / current_price)
            if size > 0:
                self.buy(size=size)
                self.log(f'🚀 买入信号：MA5>MA10且MA3上穿MA60，数量:{size}')

def get_data_from_akshare(ticker='601288', start_date='2022-01-01', end_date='2025-11-10'):
    try:
        import akshare as ak
        print(f"从AKShare获取 {ticker} 数据...")

        start_date_fmt = start_date.replace('-', '')
        end_date_fmt = end_date.replace('-', '')

        data = ak.stock_zh_a_hist(symbol=ticker, period="daily",
                                   start_date=start_date_fmt, end_date=end_date_fmt)

        if not data.empty:
            bt_data = data[['日期', '开盘', '最高', '最低', '收盘', '成交量']].copy()
            bt_data.columns = ['date', 'open', 'high', 'low', 'close', 'volume']
            bt_data['date'] = pd.to_datetime(bt_data['date'])
            bt_data.set_index('date', inplace=True)

            print(f"✅ 成功获取AKShare数据: {len(bt_data)} 条记录")
            print(f"📅 数据范围: {bt_data.index[0].date()} 到 {bt_data.index[-1].date()}")
            print(f"💰 价格范围: ¥{bt_data['close'].min():.2f} - ¥{bt_data['close'].max():.2f}")
            return bt_data
    except Exception as e:
        print(f"❌ AKShare获取失败: {e}")
        return None

def run_backtest():
    print("=" * 80)
    print("农业银行量化交易策略回测（V3版本）")
    print("=" * 80)

    # 获取数据
    data = get_data_from_akshare('601288', '2022-01-01', '2025-11-10')

    if data is None:
        print("❌ 数据获取失败")
        return

    # 创建回测引擎
    cerebro = bt.Cerebro()
    cerebro.addstrategy(AgriculturalBankStrategyV3)

    # 添加数据
    bt_data = bt.feeds.PandasData(dataname=data)
    cerebro.adddata(bt_data)

    # 设置初始资金和手续费
    cerebro.broker.setcash(100000)
    cerebro.broker.setcommission(commission=0.0003)

    # 添加分析器
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')

    # 运行回测
    start_value = cerebro.broker.getvalue()
    print(f"开始回测，初始资金: ¥{start_value:,.2f}")
    print("-" * 80)

    results = cerebro.run()
    strategy = results[0]

    # 计算结果
    final_value = cerebro.broker.getvalue()
    total_return = (final_value - start_value) / start_value * 100

    print("-" * 80)
    print("📈 回测结果:")
    print(f"初始资金: ¥{start_value:,.2f}")
    print(f"最终资金: ¥{final_value:,.2f}")
    print(f"收益率: {total_return:+.2f}%")
    print(f"交易次数: {strategy.trades}")
    print(f"盈利次数: {strategy.wins}")

    if strategy.trades > 0:
        win_rate = strategy.wins / strategy.trades * 100
        print(f"胜率: {win_rate:.1f}%")

    try:
        dd = strategy.analyzers.drawdown.get_analysis()
        if 'max' in dd:
            print(f"最大回撤: {dd['max']['drawdown']:.2f}%")
    except:
        pass

    print("=" * 80)

    # 生成图表
    try:
        cerebro.plot(style='candlestick', barup='red', bardown='green',
                   title='农业银行V3策略回测结果', figsize=(15, 8))
    except Exception as e:
        print(f"⚠️ 图表生成失败: {e}")

if __name__ == '__main__':
    run_backtest()

🏃 运行策略

完整运行命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 进入项目目录
cd /Users/gggwb/Documents/QuantGWB/backtrader

# 激活虚拟环境
source .venv/bin/activate

# 运行V3策略（农业银行 601288）
.venv/bin/python examples/shanxi_fenjiu_real_data_V3.py \
  --first-profit 0.1 \
  --protection-profit 0.08 \
  --mobile-start 0.15 \
  --mobile-retracement 0.05 \
  --ma3 2 \
  --ma5 5 \
  --ma10 10 \
  --ma60 60 \
  --data-source akshare \
  --start 2022-01-01 \
  --end 2025-11-10 \
  --ticker 601288

参数说明

参数 说明 默认值
--ma3 3日均线周期 3
--ma5 5日均线周期 5
--ma10 10日均线周期 10
--ma60 60日均线周期 60
--first-profit 第一层止盈比例 0.10 (10%)
--protection-profit 保护止盈线 0.08 (8%)
--mobile-start 移动止盈启动点 0.15 (15%)
--mobile-retracement 移动止盈回撤比例 0.05 (5%)
--ticker 股票代码 600809

💡 遇到的坑与解决思路

1. 数据获取问题

问题：Yahoo Finance API被限制
解决：改用AKShare，免费且稳定

1
2
3
4
5
6
7
# 最初的Yahoo Finance方案
import yfinance as yf
data = yf.download('600809.SS', start=start_date, end=end_date)

# 改为AKShare方案
import akshare as ak
data = ak.stock_zh_a_hist(symbol=ticker, period="daily", ...)

2. 移动止盈逻辑错误

问题：移动止盈只有在创新高时才检查触发条件
症状：价格跌破移动止盈线时没有触发止损

1
2
3
4
5
6
7
8
9
10
11
# 错误逻辑
if current_price > highest_price:  # ❌ 只有创新高才进入
    update_stop_loss()
    if current_price < stop_price:
        return True  # ❌ 永远执行不到

# 正确逻辑
if mobile_stop_active and current_price < stop_price:  # ✅ 优先检查触发
    return True
if current_price > highest_price:  # ✅ 再更新
    update_stop_loss()

3. 止盈止损优先级混乱

问题：技术卖出信号优先级高于移动止盈
症状：应该触发移动止盈时却触发了技术卖出

1
2
3
4
5
6
7
8
9
10
11
# 修正前的优先级（错误）
1. 硬止损
2. 技术卖出  # ❌ 优先级过高
3. 移动止盈  # ❌ 优先级过低

# 修正后的优先级（正确）
1. 硬止损
2. 第一层止盈
3. 移动止盈  # ✅ 保护已有收益
4. 保护止盈
5. 技术卖出  # ✅ 优先级最低

4. 买卖信号优化

问题：MA5=MA10的卖出信号过于敏感
症状：频繁交易，手续费影响收益
解决：改为MA5-MA10差值小于-2%才卖出

1
2
3
4
5
6
7
8
# 优化前
if self.ma5[0] < self.ma10[0]:
    self.sell()

# 优化后
ma_diff_pct = (ma5 - ma10) / ma10
if ma_diff_pct < -0.02:  # 差值小于-2%
    self.sell()

🎯 策略逻辑详解

止盈止损体系设计

我的策略设计了四层风险控制机制：

  1. 硬止损：亏损超过2%立即止损，保护本金
  2. 第一层止盈：上涨10%卖出一半，锁定部分收益
  3. 保护止盈：剩余仓位如果涨幅跌破8%全部卖出
  4. 移动止盈：涨幅超过15%后启动，回撤5%触发卖出

每日检查逻辑

1
2
3
4
5
6
7
8
9
def next(self):
    # 每日按优先级检查：
    # 1. 硬止损（亏损2%）
    # 2. 第一层止盈（上涨10%卖出一半）
    # 3. 如果已卖出一半：
    #    3a. 保护止盈（涨幅低于8%清仓）
    #    3b. 移动止盈（涨幅≥15%时启动，回撤5%触发）
    # 4. 技术卖出信号（MA5-MA10差值<-2%）
    # 5. 买入信号检查

买入信号设计

1
买入条件 = (MA5 > MA10) AND (MA3上穿MA60)

这个组合的含义：

  • MA5 > MA10：确保短期趋势向上
  • MA3上穿MA60：捕捉短期突破长期趋势的机会

📊 性能评估

关键指标

  • 胜率：盈利交易占总交易的比例
  • 盈亏比：平均盈利与平均亏损的比值
  • 最大回撤：策略面临的最大损失
  • 夏普比率：风险调整后的收益

回测结果分析

运行完成后，策略会输出详细的回测报告，包括：

  • 总收益率
  • 交易次数和胜率
  • 最大回撤
  • 交易明细

🚀 改进方向

短期优化

  1. 参数调优：使用网格搜索找到最优参数组合
  2. 多因子模型：加入成交量、波动率等因子
  3. 资金管理：动态调整仓位大小

长期规划

  1. 机器学习：使用ML模型预测价格走势
  2. 组合策略：多品种多策略分散风险
  3. 实盘交易：连接券商API进行实盘交易

具体优化建议

  1. 参数网格搜索

    1
    2
    3
    4
    5
    6
    7
    8
    # 测试不同参数组合
    ma5_options = [3, 5, 8, 10]
    ma10_options = [8, 10, 15, 20]
    profit_targets = [0.08, 0.10, 0.12, 0.15]

    for ma5 in ma5_options:
        for ma10 in ma10_options:
            # 运行回测并记录结果

  2. 多品种验证

    1
    2
    3
    4
    # 测试其他银行股
    stocks = ['601288', '601398', '601939', '601988']
    for stock in stocks:
        run_backtest(ticker=stock)

📝 总结与感悟

通过开发这个策略，我深刻体会到了几个要点：

  1. 简单有效：复杂的策略不一定比简单的策略好
  2. 风险控制：止盈止损比盈利预测更重要
  3. 持续学习：量化交易是一个不断学习和优化的过程
  4. 回测验证：任何策略都必须经过充分的历史数据验证
  5. 逻辑严谨：代码中的每个if-else都要仔细考虑优先级

开发过程中的关键收获

  1. 框架选择很重要：Backtrader对新手友好，文档完善
  2. 数据源是基础：AKShare比Yahoo Finance更适合A股数据
  3. 逻辑比算法重要：正确的事务逻辑比复杂的算法更有价值
  4. 调试很关键：每个bug都要彻底理解原因
  5. 参数化很必要：硬编码参数不利于调优

对量化交易的新认识

  • 量化不是圣杯：只是用数据和规则来辅助决策
  • 回测≠实盘：历史表现不代表未来
  • 风险永远第一：保本比盈利更重要
  • 持续改进：市场在变，策略也要随之调整

🔗 参考资料

免责声明：本文仅供学习交流使用，不构成投资建议。量化交易存在风险，入市需谨慎。

🎉 后记

完成第一个量化策略让我受益匪浅。虽然策略可能还不完美，但整个开发过程让我学到了：

  1. 系统化思维：从数据获取到策略执行的全流程
  2. 风险管理：多层次的风险控制机制
  3. 工程实践：代码组织、参数化、调试技巧
  4. 金融理解：技术分析在实际中的应用

这只是一个开始，量化交易的道路还很长。但相信有了这个基础，后续的学习会事半功倍。

如果这篇文章对您有帮助，欢迎点赞收藏！有任何问题欢迎在评论区讨论。

最后更新：2024-11-11

浪浪山小郎君Developer

Dream big, work hard, stay focused.