MACD、RSI、Boll以及分型指标的实现与回测 |
您所在的位置：网站首页 › 回测指标 › MACD、RSI、Boll以及分型指标的实现与回测 |
MACD、RSI、Boll以及分型指标的实现与回测
|
对指标的实现
分为两部分： 信号的计算实现信号算法 检测历史信号 保存到数据库 信号使用提供查询接口 我们将信号的计算与回测分离开，将计算后的信号结果保存到数据库中，供回测时调用，模式图如下：
走势展示：
具体的计算方法： \(EMA_i\) \[EMA_i = \frac{1}{N+1}(CLOSE_i - EMA_{i-1})+EMA_{i-1} \]\[其中EMA_0 = CLOSE_0 \]短时EMA \[EMA_1 = EMA(CLOSE,short) \]长时EMA \[EMA_2= EMA(CLOSE,long \]DIFF \[DIFF=EMA_1 - EMA_2 \]DEA \[DEA = EMA(DIFF,m) \]通常将短时EMA中参数short设为12,长时EMA中参数long设为26，m=9 早期国内股市一周是6个交易日，short=12相当于两周，一个月就26个交易日， 金叉与死叉死叉 DIFF下穿DEA \(DIFF_{i-1}>=DEA_{i-1}\) && \(DIFF_{i}= 0)&(df_daily['delta']0: update_result = DB_CONN['macd'].bulk_write(update_requests,ordered=False) print('Save MACD ,股票代码：%s,插入：%4d，更新:%4d' %(code,update_result.upserted_count,update_result.modified_count),flush=True) except: print('错误发生： %s' % code, flush=True) traceback.print_exc() if __name__ == "__main__": compute_macd('2019-01-01','2019-12-31')提供外部接口，供回测 def is_macd_gold(code,date): """ 判断某只股票在某个交易日是否出现MACD金交信号 :param code:股票代码 :param date:日期 :return :True -有金叉信号，False-无金叉信号 """ count = DB_CONN['macd'].count({'code':code,'date':date,'signal':'gold'}) return count == 1 def is_macd_dead(code, date): """ 判断某只股票在某个交易日是否出现MACD死叉信号 :param code: 股票代码 :param date: 日期 :return: True - 有死叉信号，False - 无死叉信号 """ count = DB_CONN['macd'].count({'code': code, 'date': date, 'signal': 'dead'}) return count == 1 回测结果展示
指的是一定时间窗口内，上涨幅度之和占整体涨跌幅度绝对值之和的比例 表现的是买入意愿相对于总体成交的强弱。当买方意愿达到超买区，意味着有可能会跌了，发出卖出预警，当卖方意愿达到超卖区，意味着有可能涨了，发出买入信号。这都蕴含着盛极而衰、否极泰来的意思。 超买区、超卖区超卖区：RSI < 20 超买区：RSI > 80 强弱信号超买：RSI上穿80 超卖：RSI下穿20 计算公式 \[change = close - prev_close \]\[up_change = max(change,0) \]\[RSI = mean(up_change,N) * 100 / mean(abs(change),N) \]RSI超买超卖直观展示
信号计算 from stock_util import get_all_codes,DB_CONN import pandas as pd import matplotlib.pyplot as plt from pymongo import UpdateOne,ASCENDING import traceback def compute_rsi(begin_date,end_date): """ 计算指定时间段内的RSI信号，并保存到数据库中 :param begin_date:开始日期 :param end_date:结束日期 """ #获取所有股票代码 codes = get_all_codes() #codes = ['002468'] #计算RSI N = 12 #计算所有股票的RSI信号 for code in codes: try: #获取后复权价格，使用后复权的价格计算RSI daily_cursor = DB_CONN['daily_hfq'].find( {'code':code,'date':{'$gte':begin_date,'$lte':end_date},'index':False}, sort=[('date',ASCENDING)], projection={'date':True,'close':True,'_id':False}) df_daily = pd.DataFrame([daily for daily in daily_cursor]) #如果查询出来的行情数据不足以计算N天的平均值，则不参与计算 # if len(df_daily) < N: if df_daily.index.size < N: print('数据量不足:%s' %code,flush= True) continue #将日期设置为索引 df_daily.set_index(['date'],drop=True,inplace=True) #将close移一位作为当日的pre_close df_daily['pre_close'] = df_daily['close'].shift(1) #计算当日的涨跌幅：（close- pre_close) * 100/pre_close df_daily['change_pct'] = (df_daily['close'] - df_daily['pre_close'])*100 / df_daily['pre_close'] #只保留上涨的日期的涨幅 df_daily['up_pct'] = pd.DataFrame({'up_pct':df_daily['change_pct'],'zero':0}).max(1) #计算RSI df_daily['RSI'] = df_daily['up_pct'].rolling(N).mean() / abs(df_daily['change_pct']).rolling(N).mean()*100 #移位，记作prev_rsi df_daily['PREV_RSI'] = df_daily['RSI'].shift(1) print(df_daily,flush=True) #超买，RSI下穿80，作为卖出信号 df_daily_over_bough = df_daily[(df_daily['RSI']=80)] #超卖，RSI上穿20，作为买入信号 df_daily_over_sold = df_daily[(df_daily['RSI'] >20) & (df_daily['PREV_RSI'] 0: update_result = DB_CONN['rsi'].bulk_write(update_requests,ordered=False) print('Save RSI, 股票代码：%s, 插入：%4d, 更新：%4d' % (code, update_result.upserted_count, update_result.modified_count), flush=True) except: print('错误发生：%s'%code,flush=True) traceback.print_exc() if __name__ == "__main__": compute_rsi('2019-01-01','2019-12-31')提供外部接口，供回测 def is_rsi_over_sold(code, date): """ 判断某只股票在某个交易日是出现了超卖信号 :param code: 股票代码 :param date: 日期 :return: True - 出现了超卖信号，False - 没有出现超卖信号 """ count = DB_CONN['rsi'].count({'code': code, 'date': date, 'signal': 'over_sold'}) return count == 1 def is_rsi_over_bought(code,date): """ 判断某只股票在某个交易日是出现了超买信号 :param code: 股票代码 :param date: 日期 :return: True - 出现了超买信号，False - 没有出现超买信号 """ count = count = DB_CONN['rsi'].count({'code':code,'date':date,'signal':'over_bought'}) return count ==1 回测结果展示
注：时间窗口长度、超买、超卖区间的阈值可根据实际情况自行调整 Boll（突破上轨与突破下轨） 定义Boll是基于统计学的标准差原理 包含三条轨道： 上轨：压力线 中轨：价格平均线 下轨：支撑线 计算公式(N = 20,k =2 )价格均值 \[MA = \frac{1}{N}\sum_{i=0}^{N}CLOSE_i \]标准差 \[STD = \sqrt{\frac{1}{N}\sum_{i=0}^{N}(CLOSE_i - MA)^2} \]中轨(盘中实时计算，当日收盘价不稳定，所以计算的是前一天的均价) \[MB = \frac{1}{N-1}\sum_{i=0}^{N-1}CLOSE_i \]上轨 \[UP = MB + k * STD \]下轨 \[DOWN = MB - k * STD \]Boll直观展示
信号计算 from stock_util import get_all_codes from database import DB_CONN import pandas as pd from pymongo import ASCENDING,UpdateOne import traceback def compute_boll(begin_date,end_date): """ 计算指定日期内的boll突破上轨和突破下轨信号，并保存到数据库中，方便查询调用 :param begin_date:开始日期 :param end_date:结束日期 """ #获取所有股票列表 codes = get_all_codes() #计算每只股票的Boll值 for code in codes: try: #获取后复权的价格，使用后复权的价格计算BOLL daily_cursor = DB_CONN['daily_hfq'].find( {'code':code,'date':{'$gte':begin_date,'$lte':end_date},'index':False}, sort=[('date',ASCENDING)], projection={'date':True,'close':True,'_id':False}) df_daily = pd.DataFrame([daily for daily in daily_cursor]) #计算中轨；计算MB，盘后计算，这里用当日的close df_daily['MB'] = df_daily['close'].rolling(20).mean() #计算标准差，计算STD20，即20日的标准差 df_daily['std'] = df_daily['close'].rolling(20).std() print(df_daily,flush=True) #计算上轨up df_daily['UP'] = df_daily['MB'] + 2 * df_daily['std'] #计算下轨down df_daily['DOWN'] = df_daily['MB'] - 2 * df_daily['std'] print(df_daily,flush=True) #将日期作为索引 df_daily.set_index(['date'],inplace=True) #将close移动一位，变成当前索引位置的前收 last_close = df_daily['close'].shift(1) #将上轨移动一位，实现前一日的上轨和前一日的收盘价都在当日了 shifted_up = df_daily['UP'].shift(1) #计算突破上轨，是向上突破，条件是前一日的收盘价小于前一日上轨，当日收盘价大于当日上轨 df_daily['up_mask'] = (last_close shifted_up) #将下轨移动一位，前一日的下轨和前一日的收盘就就都在当日了 shifted_down = df_daily['DOWN'].shift(1) #突破下轨，向下突破，条件是前一日收盘价大于前一日下轨，当日收盘价小于当日下轨 df_daily['down_mask'] = (last_close >= shifted_down) & (df_daily['close'] < shifted_down) #对结果进行过滤，只保留向上突破或者向下突破的数据 df_daily = df_daily[df_daily['up_mask'] | df_daily['down_mask']] #从DataFrame中扔掉不用的数据 df_daily.drop(['close','std','MB','UP','DOWN'],1,inplace=True) #将信号保存到数据库中 update_requsts = [] #DataFrame的索引为日期 for date in df_daily.index: #保存数据包括股票代码、日期和信号类型，结合数据集的名字，就表示某只股票在某日 doc = { 'code':code, 'date':date, #方向，向上突破，UP，向下突破，DOWN 'direction':'up' if df_daily.loc[date]['up_mask'] else 'down' } update_requsts.append(UpdateOne(doc,{'$set':doc},upsert=True)) #如有信号数据，将保存到数据库中 if len(update_requsts)>0: update_result = DB_CONN['boll'].bulk_write(update_requsts,ordered=False) print('%s ,upserted:%4d,modified:%4d'%(code,update_result.upserted_count,update_result.modified_count),flush=True) except : traceback.print_exc() if __name__ == "__main__": compute_boll('2019-01-01','2019-12-31')提供外部接口，供回测 def is_boll_break_down(code,date): """ 查询某只股票是否在某日出现了突破下轨信号 :param code: 股票代码 :param date: 日期 :return: True - 出现了突破下轨信号，False - 没有出现突破下轨信号 """ count = DB_CONN['boll'].count({'code':code,'date':date,'direction':'down'}) return count==1 def is_boll_break_up(code,date): """ 查询某只股票是否在某日出现了突破上轨信号 :param code: 股票代码 :param date: 日期 :return: True - 出现了突破上轨信号，False - 没有出现突破上轨信号 """ count = DB_CONN['boll'].count({'code':code,'date':date,'direction':'up'}) return count == 1 回测结果展示
获取数据的周期，可以是日、周、月甚至是分钟；上述代码中有N和k两个可变参数 分型（顶分型与底分型） 分型直观展示
信号计算 import pandas as pd import traceback from pymongo import ASCENDING,UpdateOne from stock_util import get_all_codes from database import DB_CONN def compute_fractal(begin_date,end_date): """ 计算给定周期内的分型信号，把结果保存在数据库中 :param begin_date:开始日期 :param end_date:结束日期 """ #获取所有股票代码 codes = get_all_codes() #计算每只股票的信号 for code in codes: try: #获取后复权的价格，使用后复权价格计算分型信号 daily_cursor = DB_CONN['daily_hfq'].find( {'code':code,'date':{'$gte':begin_date,'$lte':end_date},'index':False}, sort=[('date',ASCENDING)], projection={'date':True,'high':True,'low':True,'_id':False}) df_daily = pd.DataFrame([daily for daily in daily_cursor]) #将日期设置为索引 df_daily.set_index(['date'],1,inplace=True) #通过移位将前两天和后两天对齐到中间一天 df_daily_shift_1 = df_daily.shift(1) df_daily_shift_2 = df_daily.shift(2) df_daily_shift_3 = df_daily.shift(3) df_daily_shift_4 = df_daily.shift(4) #顶分型，中间日的最高价既大于前两天的价格，又大于后两天的价格 df_daily['up'] = (df_daily_shift_3['high'] > df_daily_shift_1['high']) & \ (df_daily_shift_3['high'] > df_daily_shift_2['high']) & \ (df_daily_shift_3['high'] > df_daily_shift_4['high']) & \ (df_daily_shift_3['high'] > df_daily['high']) #底分型，中间日的最低价既低于前两天的最低价，又低于后两天的最低价 df_daily['down'] = (df_daily_shift_3['low'] < df_daily_shift_1['low']) &\ (df_daily_shift_3['low'] < df_daily_shift_2['low']) &\ (df_daily_shift_3['low'] < df_daily_shift_4['low']) &\ (df_daily_shift_3['low'] < df_daily['low']) #只保留出现顶分型和低分型的日期 df_daily = df_daily[(df_daily['up'] | df_daily['down'])] #抛掉不用的数据 df_daily.drop(['high','low'],1,inplace = True) print(df_daily) #将信号保存在数据库中 update_requests = [] for date in df_daily.index: doc = { 'code':code, 'date':date, #up：顶分型，down：低分型 'direction':'up' if df_daily.loc[date]['up'] else 'down' } #保存时以code、date和direction做条件，那么就需要在这三个字段上建立索引 #db.fractal_signal.createIndex({'code':1,'date':1,'direction':1}) update_requests.append( UpdateOne(doc,{'$set':doc},upsert=True)) if len(update_requests)>0: update_result = DB_CONN['fractal_signal'].bulk_write(update_requests,ordered=False) print('%s ,upserted:%4d,modified:%4d' %(code,update_result.upserted_count,update_result.modified_count),flush=True) except : print('错误发生：%s'%code,flush=True) traceback.print_exc() if __name__ == "__main__": compute_fractal('2019-01-01','2019-12-31')提供外部接口，供回测 #提供外部接口 def is_fractal_up(code,date): count = DB_CONN['fractal_signal'].count({'code':code,'date':date,'direction':'up'}) return count == 1 """ 查询某只股票在某个日期是否出现顶分型信号 :param code: 股票代码 :param date: 日期 :return: True - 出现顶分型信号，False - 没有出现顶分型信号 """ def is_fractal_down(code,date): """ 查询某只股票在某个日期是否出现底分型信号 :param code: 股票代码 :param date: 日期 :return: True - 出现底分型信号，False - 没有出现底分型信号 """ count = DB_CONN['fractal-signal'].count({'code':code,'date':date,'direction':'down'}) return count == 1 回测结果展示
检测宽度：N \(\in (3,5,7,9···)\)；符合对称性 就目前的数据利用这四种信号得到回测结果都不尽人意，不排除本地获取、处理数据的问题，但对股票的选股、信号的计算、简单的回测、简单的持仓管理都有了简单的了解。 |
【本文地址】
今日新闻 |
推荐新闻 |