量化交易学习（六十二）MACD金叉死叉上涨下跌概率回测

发表于 分类于

这篇文章的统计结果存在幸存者偏差，MACD金叉死叉的成功率在50%左右

一直以来各种文章教程都在讲MACD金叉是买入信号，死叉是卖出信号，今天我就用量化
的方法来判断一下MACD金叉死叉到底有没有用。

首先来说一下我的回测方法，我用各行业的ETF来代表整个股市，这样能避开个股暴雷的情况。统计所有这些ETF金叉死叉之后一天、二天、三天的涨幅情况，并计算其平均数作为结果。

策略代码部分如下：

首先在初始化函数中定义好MACD指标，定义一些需要用到的变量，并打开一个以股票代码为名的csv文件，用于保存数据：

1
2
3
4
5
6
7
8
9
10
def __init__(self):
    self.macd=bt.indicators.MACD()
    self.cross=bt.indicators.CrossOver(self.macd.macd,self.macd.signal)
    self.csv_file=open(self.data._name+'.csv','w',encoding='utf-8')
    
    self.csv_writer=csv.writer(self.csv_file)
    self.last_signal=''
    self.last_date=None
    self.cnt=10

在next函数中把出现金叉死叉信号后三天的收盘价记录下来：

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
def next(self):
        if self.cross==1:
        
        crossInfo['金叉']=self.data.datetime.date(0)
        if self.macd.macd[0]>=0:
            self.cnt=0
            self.last_signal='零上金叉'
            self.csv_writer.writerow([self.data.datetime.date(0).isoformat(),'零上金叉',self.data.close[0]])
        else:
            self.last_signal='零下金叉'
            self.cnt=0
            self.csv_writer.writerow([self.data.datetime.date(0).isoformat(),'零下金叉',self.data.close[0]])

    elif self.cross==-1:
        if self.macd.macd[0]>=0:
            self.last_signal='零上死叉'
            self.cnt=0
            self.csv_writer.writerow([self.data.datetime.date(0).isoformat(),'零上死叉',self.data.close[0]])
        else:
            self.last_signal='零下死叉'
            self.cnt=0
            self.csv_writer.writerow([self.data.datetime.date(0).isoformat(),'零下死叉',self.data.close[0]])

    if self.cnt>0 and self.cnt <=3:
        self.csv_writer.writerow([self.data.datetime.date(0).isoformat(),'-',self.data.close[0]])
    
    self.cnt+=1

在main.py中使用多进程并发执行回测：

导入库：

1
2
3
4
5
6
7
8
9
10
import multiprocessing
import time
import backtrader as bt
import datetime

import mysqlDataFeed
import strategy
import argparse
# pip install python-dateutil
from dateutil.relativedelta import relativedelta

定义行业ETF列表：

1
2
3
4
5
6
7
symbols=['SHSE.513030','SHSE.512550','SZSE.159752','SHSE.513080','SHSE.513520',
         'SHSE.510050','SHSE.510300','SZSE.159915','SHSE.588000','SHSE.510500','SHSE.512100','SZSE.159920','SHSE.513180','SZSE.159941','SHSE.513500',
          'SHSE.513330','SHSE.513050','SHSE.512480','SZSE.159997','SHSE.515880','SZSE.159819','SHSE.512720','SHSE.515400','SHSE.515230','SZSE.159869',
          'SHSE.512980','SHSE.513360','SHSE.562500','SHSE.515250','SHSE.515030','SHSE.515790','SHSE.512660','SHSE.513060','SHSE.512170','SHSE.512010',
          'SZSE.159992','SHSE.560080','SZSE.159928','SHSE.512690','SZSE.159865','SZSE.159996','SZSE.159766','SHSE.512800','SHSE.512880','SHSE.512070',
          'SHSE.512200','SHSE.516970','SZSE.159870','SZSE.159611','SHSE.515220','SHSE.515210','SHSE.512400','SHSE.516150','SHSE.518880','SHSE.510880',
          ]

定义监测函数，它就是每个进程并发运行时执行的函数，通过fromdate参数可以控制回测的范围：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def monitor(symbol,q):
    cerebro = bt.Cerebro()
    cerebro.addstrategy(strategy.MACDStrategy)
    today=datetime.datetime.today()
    data=mysqlDataFeed.MySQLData(
                            dataname=symbol,
                            timeframe=bt.TimeFrame.Days,
                            fromdate=today-relativedelta(months=60),
                            todate=today,
                            adj_base_date=today
                            )
    cerebro.adddata(data)
    cerebro.broker.setcash(10000)
    cerebro.run()
	q.put(1)

在程序执行时，建一个容量为10的并发池，并发对所有的标的进行回测，这段程序其实是在我在每天MACD金叉死叉统计的代码的基础上改的，在这里队列q没有什么具体的作用，只是起一个确保所有进程都正常结束的等待作用：

1
2
3
4
5
6
7
8
9
10
11
12
13
if __name__=='__main__':
    print(datetime.datetime.now())

    pool=multiprocessing.Pool(processes=10)
    m=multiprocessing.Manager()
    q=m.Queue()

    workers=[pool.apply_async(monitor,(symbol,q)) for symbol in symbols]

    while q.qsize()< len(symbols):
        time.sleep(1)

    print(datetime.datetime.now())

回测执行完后，会得到一系列csv文件：
9a8e4279b5c6c06cf12e693354559533.png

文件内容如下：
cfa21de6f86e0872937b44ce25c09503.png

然后我又写了一个统计程序，用来统计结果：

读取csv结果并存到列表中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import csv

def read_csv_file(file_name,data):
    """
    读取CSV文件，并返回包含数据的列表。
    :param file_name: 文件名
    :return: 包含数据的列表
    """
    # 打开一个文件对象，文件名为"input.csv"，模式为读取（'r'），并使用'utf-8'编码
    with open(file_name, mode="r", encoding="utf-8") as file:
        # 创建一个csv读取器对象
        csv_reader = csv.reader(file)

        # 获取CSV文件的表头
        header = next(csv_reader)


        # 遍历CSV文件中的每一行数据
        for row in csv_reader:
            # 将行数据转换为字典，并添加到列表中

            data.append(dict(zip(header, row)))

统计数据：

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
# 主程序
if __name__ == "__main__":
    symbols=['SHSE.513030','SHSE.512550','SZSE.159752','SHSE.513080','SHSE.513520',
         'SHSE.510050','SHSE.510300','SZSE.159915','SHSE.588000','SHSE.510500','SHSE.512100','SZSE.159920','SHSE.513180','SZSE.159941','SHSE.513500',
          'SHSE.513330','SHSE.513050','SHSE.512480','SZSE.159997','SHSE.515880','SZSE.159819','SHSE.512720','SHSE.515400','SHSE.515230','SZSE.159869',
          'SHSE.512980','SHSE.513360','SHSE.562500','SHSE.515250','SHSE.515030','SHSE.515790','SHSE.512660','SHSE.513060','SHSE.512170','SHSE.512010',
          'SZSE.159992','SHSE.560080','SZSE.159928','SHSE.512690','SZSE.159865','SZSE.159996','SZSE.159766','SHSE.512800','SHSE.512880','SHSE.512070',
          'SHSE.512200','SHSE.516970','SZSE.159870','SZSE.159611','SHSE.515220','SHSE.515210','SHSE.512400','SHSE.516150','SHSE.518880','SHSE.510880',
          ]
    # 读取CSV文件
    # 初始化一个空列表，用于存储数据
    data=[]
    for symbol in symbols: 
      read_csv_file(symbol+'.csv',data)
    statistic = {
        '零上金叉':{
            '一天':[],
            '二天':[],
            '三天':[],
            '一天涨幅':0,
            '二天涨幅':0,
            '三天涨幅':0,
        },
        '零下金叉':{
            '一天':[],
            '二天':[],
            '三天':[],
            '一天涨幅':0,
            '二天涨幅':0,
            '三天涨幅':0,
        },
        '零上死叉':{
            '一天':[],
            '二天':[],
            '三天':[],
            '一天涨幅':0,
            '二天涨幅':0,
            '三天涨幅':0,
        },
        '零下死叉':{
            '一天':[],
            '二天':[],
            '三天':[],
            '一天涨幅':0,
            '二天涨幅':0,
            '三天涨幅':0,
        }
    }
    cnt=10
    tp=''
    base=0
    for x in data:
      if x['信号类型'] != '-':
        cnt=0
        base=float(x['收盘价'])
        tp=x['信号类型']
      else:
        if cnt==1:
          statistic[tp]['一天'].append((float(x['收盘价'])-base)/base)
        if cnt==2:
          statistic[tp]['二天'].append((float(x['收盘价'])-base)/base)
        if cnt==3:
          statistic[tp]['三天'].append((float(x['收盘价'])-base)/base)

      cnt+=1

    tps=['零上金叉','零下金叉','零上死叉','零下死叉']
    dys=['一天','二天','三天']
    dyups=['一天涨幅','二天涨幅','三天涨幅']

    for tp in tps:
       for dy in dys:
          statistic[tp][dy+'涨幅']=sum(statistic[tp][dy])/len(statistic[tp][dy]) if len(statistic[tp][dy])>0 else 0

    for tp in tps:
       for dyup in dyups:
          print(tp,dyup,f'{statistic[tp][dyup]*100:.2f}%')

我分别对近半年、近1年、近2年、近3年、近5年的数据进行了统计，结果如下：

近半年：

e6cf164f989c0cbf50e00e4180b3380f.png

近1年：

01b9261c056b9b93b0e0e73c7375000e.png

近2年：

0fef53c42c0143eaf29b408958ea096a.png

近3年：

8fcf0e211cd18a97accabfde53b5e5fa.png

近5年：

e62fbb88657699269e8c3b198c8d1437.png

从这些数据中可以看出，对于行业ETF，MACD金叉死叉的确有用，金叉后确实上涨的概率大，死叉后下跌的概率大，零上金叉涨幅最大，零下死叉跌幅最大。

希望这篇文章能帮助到大家。如果你有任何问题或建议，欢迎留言讨论，私信。感谢你的阅读，觉得不错，点个赞哦！还没有关注我的朋友可以关注 江达小记

江达小记