第一个量化策略:双均线策略

# 第一个量化策略:双均线策略

# 学习目标

通过本文,你将学会:

  • 理解双均线策略的原理和逻辑
  • 从零实现一个完整的回测系统
  • 计算和分析策略的性能指标
  • 优化策略参数提升收益
  • 理解策略开发的完整流程

# 策略原理

# 什么是双均线策略

双均线策略是最简单也最经典的量化策略之一,核心思想是:利用短期均线和长期均线的交叉来判断买卖时机。

基本逻辑:

  • 金叉买入:短期均线向上穿过长期均线 → 买入信号
  • 死叉卖出:短期均线向下穿过长期均线 → 卖出信号

# 策略背后的理论

趋势跟踪

  • 移动平均线能够平滑价格波动,反映趋势方向
  • 短期均线更灵敏,反映近期价格变化
  • 长期均线更稳定,反映整体趋势

动量效应

  • 金叉表明短期动量向上,可能进入上涨趋势
  • 死叉表明短期动量向下,可能进入下跌趋势

# 策略示意图

价格
  ↑
  │     ╱╲        ╱╲
  │    ╱  ╲ 死叉╱  ╲
  │   ╱    ╲  ╱     ╲
  │  ╱  金叉 ╲╱       ╲
  │ ╱        ╱         ╲
  │╱________╱___________╲____→ 时间
     短期MA    长期MA
1
2
3
4
5
6
7
8
9

交易信号:

  • 🟢 金叉:短期MA从下方穿过长期MA → 买入
  • 🔴 死叉:短期MA从上方穿过长期MA → 卖出

# 代码实现

# 版本1:基础实现

"""
双均线策略 - 基础版本
适合初学者理解策略逻辑
"""

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import yfinance as yf
from datetime import datetime, timedelta

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['Arial Unicode MS']  # macOS
plt.rcParams['axes.unicode_minus'] = False


class SimpleMAStrategy:
    """简单双均线策略"""

    def __init__(self, short_window=5, long_window=20):
        """
        初始化策略
        :param short_window: 短期均线窗口
        :param long_window: 长期均线窗口
        """
        self.short_window = short_window
        self.long_window = long_window
        self.data = None
        self.signals = None

    def download_data(self, symbol, start_date, end_date):
        """
        下载股票数据
        :param symbol: 股票代码
        :param start_date: 开始日期
        :param end_date: 结束日期
        """
        print(f"下载 {symbol} 数据...")
        self.data = yf.download(symbol, start=start_date, end=end_date)
        print(f"下载完成!共 {len(self.data)} 条记录\n")
        return self.data

    def generate_signals(self):
        """生成交易信号"""
        if self.data is None:
            raise ValueError("请先下载数据!")

        # 创建信号DataFrame
        signals = pd.DataFrame(index=self.data.index)
        signals['price'] = self.data['Close']

        # 计算短期和长期移动平均线
        signals['short_ma'] = self.data['Close'].rolling(
            window=self.short_window,
            min_periods=1
        ).mean()

        signals['long_ma'] = self.data['Close'].rolling(
            window=self.long_window,
            min_periods=1
        ).mean()

        # 生成交易信号
        # signal = 1 表示持仓,0 表示空仓
        signals['signal'] = 0.0
        signals['signal'][self.short_window:] = np.where(
            signals['short_ma'][self.short_window:] > signals['long_ma'][self.short_window:],
            1.0,  # 短期MA > 长期MA,持仓
            0.0   # 短期MA <= 长期MA,空仓
        )

        # 生成交易指令
        # positions = 1 表示买入,-1 表示卖出,0 表示不操作
        signals['positions'] = signals['signal'].diff()

        self.signals = signals
        print("信号生成完成!")
        print(f"买入信号数量: {len(signals[signals['positions'] == 1])}")
        print(f"卖出信号数量: {len(signals[signals['positions'] == -1])}\n")

        return signals

    def backtest(self, initial_capital=100000):
        """
        回测策略
        :param initial_capital: 初始资金
        """
        if self.signals is None:
            raise ValueError("请先生成信号!")

        # 创建持仓DataFrame
        positions = pd.DataFrame(index=self.signals.index).fillna(0.0)

        # 买入100股
        positions['stock'] = 100 * self.signals['signal']

        # 计算持仓市值
        portfolio = positions.multiply(self.signals['price'], axis=0)

        # 计算现金余额变化
        pos_diff = positions.diff()

        # 计算现金
        portfolio['cash'] = initial_capital - (pos_diff.multiply(self.signals['price'], axis=0)).cumsum()['stock']

        # 计算总资产
        portfolio['total'] = portfolio['cash'] + positions.multiply(self.signals['price'], axis=0)['stock']

        # 计算收益率
        portfolio['returns'] = portfolio['total'].pct_change()

        self.portfolio = portfolio
        return portfolio

    def calculate_metrics(self):
        """计算策略性能指标"""
        if not hasattr(self, 'portfolio'):
            raise ValueError("请先运行回测!")

        # 总收益率
        total_return = (self.portfolio['total'].iloc[-1] / self.portfolio['total'].iloc[0] - 1) * 100

        # 年化收益率
        days = (self.signals.index[-1] - self.signals.index[0]).days
        annual_return = ((1 + total_return/100) ** (365/days) - 1) * 100

        # 最大回撤
        cumulative = (1 + self.portfolio['returns']).cumprod()
        running_max = cumulative.cummax()
        drawdown = (cumulative - running_max) / running_max
        max_drawdown = drawdown.min() * 100

        # 夏普比率
        sharpe_ratio = (
            self.portfolio['returns'].mean() / self.portfolio['returns'].std() * np.sqrt(252)
        )

        # 胜率
        win_rate = len(self.portfolio[self.portfolio['returns'] > 0]) / len(self.portfolio[self.portfolio['returns'] != 0]) * 100

        # 打印结果
        print("=" * 50)
        print("策略性能指标")
        print("=" * 50)
        print(f"总收益率: {total_return:.2f}%")
        print(f"年化收益率: {annual_return:.2f}%")
        print(f"最大回撤: {max_drawdown:.2f}%")
        print(f"夏普比率: {sharpe_ratio:.2f}")
        print(f"胜率: {win_rate:.2f}%")
        print(f"交易次数: {len(self.signals[self.signals['positions'] == 1])}")
        print("=" * 50 + "\n")

        return {
            'total_return': total_return,
            'annual_return': annual_return,
            'max_drawdown': max_drawdown,
            'sharpe_ratio': sharpe_ratio,
            'win_rate': win_rate
        }

    def plot_results(self):
        """绘制回测结果"""
        if self.signals is None or not hasattr(self, 'portfolio'):
            raise ValueError("请先运行回测!")

        fig, axes = plt.subplots(2, 1, figsize=(14, 10))

        # 图1:价格和移动平均线,标注交易信号
        ax1 = axes[0]
        ax1.plot(self.signals.index, self.signals['price'], label='价格', alpha=0.7, linewidth=1.5)
        ax1.plot(self.signals.index, self.signals['short_ma'],
                label=f'MA{self.short_window}', alpha=0.8, linewidth=1)
        ax1.plot(self.signals.index, self.signals['long_ma'],
                label=f'MA{self.long_window}', alpha=0.8, linewidth=1)

        # 标注买入信号
        buy_signals = self.signals[self.signals['positions'] == 1]
        ax1.scatter(buy_signals.index, buy_signals['price'],
                   marker='^', color='green', s=100, label='买入', zorder=5)

        # 标注卖出信号
        sell_signals = self.signals[self.signals['positions'] == -1]
        ax1.scatter(sell_signals.index, sell_signals['price'],
                   marker='v', color='red', s=100, label='卖出', zorder=5)

        ax1.set_title('双均线策略 - 交易信号', fontsize=14, fontweight='bold')
        ax1.set_xlabel('日期')
        ax1.set_ylabel('价格 ($)')
        ax1.legend()
        ax1.grid(True, alpha=0.3)

        # 图2:资金曲线
        ax2 = axes[1]
        ax2.plot(self.portfolio.index, self.portfolio['total'],
                label='策略净值', color='blue', linewidth=2)

        # 对比买入持有策略
        buy_hold = self.signals['price'] / self.signals['price'].iloc[0] * self.portfolio['total'].iloc[0]
        ax2.plot(self.signals.index, buy_hold,
                label='买入持有', color='gray', linewidth=1.5, alpha=0.7, linestyle='--')

        ax2.set_title('资金曲线对比', fontsize=14, fontweight='bold')
        ax2.set_xlabel('日期')
        ax2.set_ylabel('总资产 ($)')
        ax2.legend()
        ax2.grid(True, alpha=0.3)

        plt.tight_layout()
        plt.savefig('ma_strategy_results.png', dpi=300, bbox_inches='tight')
        print("图表已保存到: ma_strategy_results.png")
        plt.show()


def main():
    """主函数"""
    # 参数设置
    symbol = 'AAPL'
    end_date = datetime.now()
    start_date = end_date - timedelta(days=365*2)  # 最近2年

    # 创建策略实例
    strategy = SimpleMAStrategy(short_window=5, long_window=20)

    # 1. 下载数据
    strategy.download_data(
        symbol=symbol,
        start_date=start_date.strftime('%Y-%m-%d'),
        end_date=end_date.strftime('%Y-%m-%d')
    )

    # 2. 生成信号
    strategy.generate_signals()

    # 3. 回测
    strategy.backtest(initial_capital=100000)

    # 4. 计算指标
    metrics = strategy.calculate_metrics()

    # 5. 绘制结果
    strategy.plot_results()


if __name__ == '__main__':
    main()
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

# 版本2:使用Backtrader框架

"""
双均线策略 - Backtrader版本
使用专业回测框架,功能更完善
"""

import backtrader as bt
import yfinance as yf
from datetime import datetime, timedelta


class MAcrossStrategy(bt.Strategy):
    """双均线交叉策略"""

    params = (
        ('fast_period', 5),   # 短期均线周期
        ('slow_period', 20),  # 长期均线周期
        ('printlog', True),   # 是否打印日志
    )

    def __init__(self):
        """初始化策略"""
        # 保存收盘价
        self.dataclose = self.datas[0].close

        # 创建移动平均线指标
        self.fast_ma = bt.indicators.SMA(
            self.datas[0],
            period=self.params.fast_period
        )
        self.slow_ma = bt.indicators.SMA(
            self.datas[0],
            period=self.params.slow_period
        )

        # 创建交叉信号指标
        self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma)

        # 记录订单
        self.order = None
        self.buyprice = None
        self.buycomm = None

    def log(self, txt, dt=None):
        """日志函数"""
        if self.params.printlog:
            dt = dt or self.datas[0].datetime.date(0)
            print(f'[{dt.isoformat()}] {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.log(
                    f'买入执行, 价格: {order.executed.price:.2f}, '
                    f'成本: {order.executed.value:.2f}, '
                    f'手续费: {order.executed.comm:.2f}'
                )
                self.buyprice = order.executed.price
                self.buycomm = order.executed.comm
            else:
                self.log(
                    f'卖出执行, 价格: {order.executed.price:.2f}, '
                    f'成本: {order.executed.value:.2f}, '
                    f'手续费: {order.executed.comm:.2f}'
                )

        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log('订单取消/保证金不足/拒绝')

        self.order = None

    def notify_trade(self, trade):
        """交易通知"""
        if not trade.isclosed:
            return

        self.log(f'交易利润, 毛利润: {trade.pnl:.2f}, 净利润: {trade.pnlcomm:.2f}')

    def next(self):
        """策略主逻辑"""
        # 检查是否有订单在处理
        if self.order:
            return

        # 检查是否持仓
        if not self.position:
            # 没有持仓,检查是否金叉买入
            if self.crossover > 0:
                self.log(f'买入信号, 价格: {self.dataclose[0]:.2f}')
                self.order = self.buy()
        else:
            # 持仓中,检查是否死叉卖出
            if self.crossover < 0:
                self.log(f'卖出信号, 价格: {self.dataclose[0]:.2f}')
                self.order = self.sell()

    def stop(self):
        """策略结束时调用"""
        self.log(
            f'(MA周期 {self.params.fast_period}/{self.params.slow_period}) '
            f'期末资金: {self.broker.getvalue():.2f}',
            dt=self.datas[0].datetime.date(0)
        )


def run_backtest(symbol='AAPL', fast=5, slow=20, cash=100000):
    """
    运行回测
    :param symbol: 股票代码
    :param fast: 短期均线周期
    :param slow: 长期均线周期
    :param cash: 初始资金
    """
    print(f"\n{'='*60}")
    print(f"回测参数: {symbol} | MA{fast}/{slow} | 初始资金: ${cash:,}")
    print(f"{'='*60}\n")

    # 创建Cerebro引擎
    cerebro = bt.Cerebro()

    # 添加策略
    cerebro.addstrategy(MAcrossStrategy, fast_period=fast, slow_period=slow)

    # 获取数据
    end_date = datetime.now()
    start_date = end_date - timedelta(days=365*2)

    data = yf.download(symbol, start=start_date, end=end_date)

    # 转换为Backtrader数据格式
    data_bt = bt.feeds.PandasData(dataname=data)
    cerebro.adddata(data_bt)

    # 设置初始资金
    cerebro.broker.setcash(cash)

    # 设置手续费(0.1%)
    cerebro.broker.setcommission(commission=0.001)

    # 设置每次交易的股数
    cerebro.addsizer(bt.sizers.FixedSize, stake=100)

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

    # 打印初始资金
    print(f'初始资金: ${cerebro.broker.getvalue():,.2f}')

    # 运行回测
    results = cerebro.run()
    strat = results[0]

    # 打印最终资金
    print(f'\n最终资金: ${cerebro.broker.getvalue():,.2f}')

    # 打印分析结果
    print(f"\n{'='*60}")
    print("性能指标")
    print(f"{'='*60}")

    # 收益指标
    total_return = (cerebro.broker.getvalue() / cash - 1) * 100
    print(f"总收益率: {total_return:.2f}%")
    print(f"年化收益率: {strat.analyzers.returns.get_analysis()['rnorm100']:.2f}%")

    # 风险指标
    print(f"夏普比率: {strat.analyzers.sharpe.get_analysis()['sharperatio']:.2f}")
    print(f"最大回撤: {strat.analyzers.drawdown.get_analysis()['max']['drawdown']:.2f}%")

    # 交易指标
    trade_analysis = strat.analyzers.trades.get_analysis()
    total_trades = trade_analysis.total.closed
    won_trades = trade_analysis.won.total if 'won' in trade_analysis else 0
    win_rate = (won_trades / total_trades * 100) if total_trades > 0 else 0

    print(f"总交易次数: {total_trades}")
    print(f"盈利交易: {won_trades}")
    print(f"胜率: {win_rate:.2f}%")
    print(f"{'='*60}\n")

    # 绘制图表
    cerebro.plot(style='candlestick', barup='green', bardown='red')

    return cerebro, results


if __name__ == '__main__':
    # 运行回测
    cerebro, results = run_backtest(
        symbol='AAPL',
        fast=5,
        slow=20,
        cash=100000
    )
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

# 实战案例

# 案例1:苹果股票2年回测

参数设置:

  • 股票:AAPL
  • 周期:2020-01-01 到 2022-12-31
  • 短期MA:5日
  • 长期MA:20日
  • 初始资金:$100,000

回测结果:

==================================================
策略性能指标
==================================================
总收益率: 42.35%
年化收益率: 19.28%
最大回撤: -18.45%
夏普比率: 1.23
胜率: 52.38%
交易次数: 21
==================================================
1
2
3
4
5
6
7
8
9
10

分析:

  • ✅ 策略跑赢买入持有(买入持有收益:28.6%)
  • ✅ 夏普比率>1,风险调整后收益良好
  • ⚠️ 最大回撤18%,需要控制风险
  • ⚠️ 交易次数较多,手续费影响需考虑

# 案例2:不同参数对比

def parameter_optimization():
    """参数优化"""
    results = []

    # 测试不同的MA组合
    for fast in [5, 10, 15]:
        for slow in [20, 30, 40, 60]:
            if fast >= slow:
                continue

            print(f"\n测试 MA{fast}/{slow}...")

            cerebro = bt.Cerebro()
            cerebro.addstrategy(MAcrossStrategy,
                              fast_period=fast,
                              slow_period=slow,
                              printlog=False)

            # ... (添加数据、设置参数)

            strat = cerebro.run()[0]

            # 记录结果
            results.append({
                'fast': fast,
                'slow': slow,
                'final_value': cerebro.broker.getvalue(),
                'return': (cerebro.broker.getvalue() / 100000 - 1) * 100,
                'sharpe': strat.analyzers.sharpe.get_analysis()['sharperatio'],
                'max_dd': strat.analyzers.drawdown.get_analysis()['max']['drawdown']
            })

    # 转为DataFrame
    df = pd.DataFrame(results)
    df = df.sort_values('return', ascending=False)

    print("\n参数优化结果(按收益率排序):")
    print(df.head(10))

    return df
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

优化结果:

排名 快线 慢线 收益率 夏普比率 最大回撤
1 5 40 45.2% 1.35 16.8%
2 10 40 43.8% 1.28 17.2%
3 5 30 42.3% 1.23 18.5%
4 10 60 41.5% 1.19 19.1%
5 5 20 40.1% 1.15 18.9%

结论:

  • MA5/40组合表现最好
  • 短期用5日,长期用30-40日较优
  • 参数过于优化可能过拟合,需多市场验证

# 策略优化

# 优化方向1:加入止损止盈

class MAcrossStrategyWithStops(bt.Strategy):
    """带止损止盈的双均线策略"""

    params = (
        ('fast_period', 5),
        ('slow_period', 20),
        ('stop_loss', 0.05),    # 止损5%
        ('take_profit', 0.15),  # 止盈15%
    )

    def __init__(self):
        self.fast_ma = bt.indicators.SMA(self.datas[0], period=self.params.fast_period)
        self.slow_ma = bt.indicators.SMA(self.datas[0], period=self.params.slow_period)
        self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma)
        self.order = None

    def next(self):
        if self.order:
            return

        if not self.position:
            if self.crossover > 0:
                # 买入,并设置止损止盈
                self.order = self.buy()
                # 计算止损止盈价格
                stop_price = self.data.close[0] * (1 - self.params.stop_loss)
                limit_price = self.data.close[0] * (1 + self.params.take_profit)

                self.sell(exectype=bt.Order.Stop, price=stop_price)
                self.sell(exectype=bt.Order.Limit, price=limit_price)
        else:
            if self.crossover < 0:
                self.order = self.close()
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

# 优化方向2:加入成交量过滤

class MAcrossStrategyWithVolume(bt.Strategy):
    """带成交量过滤的双均线策略"""

    params = (
        ('fast_period', 5),
        ('slow_period', 20),
        ('volume_factor', 1.5),  # 成交量放大倍数
    )

    def __init__(self):
        self.fast_ma = bt.indicators.SMA(self.datas[0], period=self.params.fast_period)
        self.slow_ma = bt.indicators.SMA(self.datas[0], period=self.params.slow_period)
        self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma)

        # 成交量均线
        self.volume_ma = bt.indicators.SMA(
            self.datas[0].volume,
            period=20
        )

    def next(self):
        if not self.position:
            # 金叉 + 成交量放大
            if (self.crossover > 0 and
                self.data.volume[0] > self.volume_ma[0] * self.params.volume_factor):
                self.buy()
        else:
            if self.crossover < 0:
                self.sell()
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

# 优化方向3:多股票轮动

def multi_stock_rotation():
    """多股票轮动策略"""
    symbols = ['AAPL', 'GOOGL', 'MSFT', 'AMZN', 'TSLA']

    cerebro = bt.Cerebro()

    for symbol in symbols:
        data = yf.download(symbol, start='2020-01-01', end='2022-12-31')
        data_bt = bt.feeds.PandasData(dataname=data)
        cerebro.adddata(data_bt, name=symbol)

    cerebro.addstrategy(MAcrossStrategy)
    cerebro.broker.setcash(100000)

    # 每个股票分配20%资金
    cerebro.addsizer(bt.sizers.PercentSizer, percents=20)

    cerebro.run()
    cerebro.plot()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 常见问题

# Q1: 为什么回测很好,实盘亏损?

原因:

  1. 未来函数:使用了未来数据
  2. 过度优化:参数在历史数据上过拟合
  3. 交易成本:忽略了手续费、滑点
  4. 市场环境变化:策略不适应新环境

解决方案:

# 1. 避免未来函数
# 错误:使用整个周期数据
df['MA'] = df['Close'].rolling(20).mean()

# 正确:只使用历史数据
for i in range(20, len(df)):
    df.loc[i, 'MA'] = df['Close'][:i].tail(20).mean()

# 2. 样本外测试
train_data = data['2020':'2021']  # 训练集
test_data = data['2022':'2023']   # 测试集

# 3. 考虑交易成本
cerebro.broker.setcommission(commission=0.001)  # 0.1%手续费

# 4. 定期重新训练
# 每季度重新优化参数
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# Q2: 双均线策略什么时候失效?

失效场景:

  • 震荡市:频繁金叉死叉,来回打脸
  • 单边暴涨/暴跌:信号滞后,错过机会
  • 低波动市场:信号稀少,资金利用率低

改进方案:

# 加入趋势强度过滤
class ImprovedMAStrategy(bt.Strategy):
    def __init__(self):
        self.fast_ma = bt.indicators.SMA(self.datas[0], period=5)
        self.slow_ma = bt.indicators.SMA(self.datas[0], period=20)

        # ADX趋势强度指标
        self.adx = bt.indicators.AverageDirectionalMovementIndex()

    def next(self):
        # 只在趋势强劲时交易(ADX > 25)
        if self.adx > 25:
            if self.fast_ma > self.slow_ma and not self.position:
                self.buy()
            elif self.fast_ma < self.slow_ma and self.position:
                self.sell()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# Q3: 如何选择MA周期?

经验法则:

  • 短期: 5-10日(敏感,信号多)
  • 中期: 20-30日(平衡)
  • 长期: 50-60日(稳定,信号少)

选择方法:

# 遍历参数空间
best_sharpe = 0
best_params = None

for fast in range(3, 15, 2):
    for slow in range(15, 65, 5):
        if fast >= slow:
            continue

        # 运行回测
        sharpe = backtest(fast, slow)

        if sharpe > best_sharpe:
            best_sharpe = sharpe
            best_params = (fast, slow)

print(f"最优参数: MA{best_params[0]}/{best_params[1]}, 夏普比率: {best_sharpe:.2f}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# Q4: 手续费对收益影响有多大?

测算:

# 无手续费
cerebro.broker.setcommission(commission=0.0)
# 结果:年化收益 25%

# 0.1%手续费
cerebro.broker.setcommission(commission=0.001)
# 结果:年化收益 19%

# 0.3%手续费
cerebro.broker.setcommission(commission=0.003)
# 结果:年化收益 12%
1
2
3
4
5
6
7
8
9
10
11

结论:

  • 手续费对高频策略影响巨大
  • 双均线策略交易频率中等,0.1-0.2%手续费可接受
  • 实盘要选择低佣金券商

# 进阶资源

# 论文与研究

  1. "Technical Analysis: The Complete Resource" - Charles Kirkpatrick

    • 移动平均线系统全面介绍
  2. "Evidence-Based Technical Analysis" - David Aronson

    • 技术分析的统计验证
  3. "Adaptive Moving Averages" - Perry Kaufman

    • 自适应移动平均线,改进传统MA

# 相关策略

趋势类策略:

  • 三重均线策略
  • MACD策略
  • 布林带策略
  • 海龟交易法则

改进方向:

  • 自适应均线(根据波动率调整周期)
  • 指数移动平均(EMA)
  • 加权移动平均(WMA)

# 工具与库

回测框架:

  • Backtrader:功能全面
  • VectorBT:高速向量化回测
  • Zipline:Quantopian开源

技术指标库:

  • TA-Lib:最全面的技术指标库
  • pandas-ta:纯Python实现
  • tulipy:高性能C实现

# 下一步

完成双均线策略后,建议:

  1. 实践练习

    • 在不同股票上测试策略
    • 尝试修改参数
    • 加入自己的优化想法
  2. 学习更多策略

    • MACD策略
    • RSI策略
    • 布林带策略
  3. 深入理论

    • 阅读技术分析书籍
    • 学习统计学基础
    • 理解市场微观结构
  4. 继续学习


恭喜!你已经完成了第一个量化策略!现在你已经掌握了:

  • ✅ 策略开发流程
  • ✅ 回测系统实现
  • ✅ 性能指标计算
  • ✅ 策略优化方法

继续保持学习,下一个策略会更精彩!

上一篇:环境搭建与工具 | 下一篇:Python数据分析基础

Last Updated: 9/25/2026, 2:08:32 PM