来源：掘金量化社区 作者：四两 原文链接：/topic/2050
掘金中缺少一些常用的技术指标函数，但在编写策略时，免不了要用到指数指标来帮助决策。因此，小编汇总了几个常用的技术指标，基于掘金的框架编写成函数，大家可以在使用时参考一下。
提示：
行情软件比如通达信的一些指标都是基于股票上市首日计算得到的，如果计算时只用小部分的时间序列，得到的结果和通达信软件是不一致的。为了保持一致，所有指标都按照上市首日计算。
前期准备
# coding=utf-8from __future__ import print_function, absolute_importfrom gm.api import *import pandas as pdimport numpy as npset_token('请输入你的token')
KDJ指标
# KDJ指标计算# 原理：计算N日RSV = （今日收盘 - N日最低）/(N日最高-N日最低) * 100# K = (1-(1/M1))*前一日K值 + 1/M1 * RSV# D = (1-(1/M1))*前一日D值 + 1/M1 * K值# J = 3 * K - 2 * Ddef KDJ(symbol, N, M1, M2,end_time):'''计算KDJ指标公式输入： data <- dataframe，需包含开盘、收盘、最高、最低价N、M1、M2 <- intend_time <- str 结束时间输出： 将K、D、J合并到data后的dataframe'''# 取历史数据，取到上市首日data = history(symbol=symbol, frequency='1d', start_time='-01-01', end_time=end_time, fields='symbol,bob,close,low,high',df=True)# 计算前N日最低和最高，缺失值用前n日（n<N)最小值替代lowList = data['low'].rolling(N).min()lowList.fillna(value=data['low'].expanding().min(), inplace=True)highList = data['high'].rolling(N).max()highList.fillna(value=data['high'].expanding().max(), inplace=True)# 计算rsvrsv = (data['close'] - lowList) / (highList - lowList) * 100# 计算k,d,jdata['kdj_k'] = rsv.ewm(alpha=1/M1, adjust=False).mean()# ewm是指数加权函数data['kdj_d'] = data['kdj_k'].ewm(alpha=1/M2, adjust=False).mean()data['kdj_j'] = 3.0 * data['kdj_k'] - 2.0 * data['kdj_d']return data# 测试一下d = KDJ('DCE.y2101',9,3,3,'-12-31')
MACD指标
# MACD指标计算# 计算12日和26日的EMA数据# 计算12日EMA:EMA(12) = 2/(12+1) * 今日收盘价(12) + 11/(12+1) * 昨日EMA(12)# 计算26日EMA:EMA(26) = 2/(26+1) * 今日收盘价(26) + 25/(26+1) * 昨日EMA(26)# 计算DEA:DEA = 2/(9+1) * 今日DIFF + 8/(9+1) * 昨日DEA# 计算MACD：MACD = 2 * (DIFF-DEA)# 注意：上市首日DIFF、DEA、MACD均为0，次日的EMA均按照上市首日的收盘价计算def MACD(symbol, start_time, end_time):'''计算MACD指标输入参数：symbol <- str标的代码 （以前上市的不可用）start_time <- str 起始时间end_time <- str 结束时间输出数据：macd <- dataframe macd指标，包括DIFF、DEA、MACD'''# 取历史数据，取到上市首日data = history(symbol=symbol, frequency='1d', start_time='-01-01', end_time=end_time, fields='symbol,bob,close',df=True)# 将数据转化为dataframe格式data['bob'] = data['bob'].apply(lambda x: x.strftime('%Y-%m-%d')).tolist()# 计算EMA(12)和EMA(16)data['EMA12'] = data['close'].ewm(alpha=2 / 13, adjust=False).mean()data['EMA26'] = data['close'].ewm(alpha=2 / 27, adjust=False).mean()# 计算DIFF、DEA、MACDdata['DIFF'] = data['EMA12'] - data['EMA26']data['DEA'] = data['DIFF'].ewm(alpha=2 / 10, adjust=False).mean()data['MACD'] = 2 * (data['DIFF'] - data['DEA'])# 上市首日，DIFF、DEA、MACD均为0data['DIFF'].iloc[0] = 0data['DEA'].iloc[0] = 0data['MACD'].iloc[0] = 0# 按照起止时间筛选MACD = data[(data['bob'] >= start_time)]return MACD# 测试一下a = MACD(symbol = 'DCE.y2101',start_time = '-01-01',end_time = '-10-22')
DMA指标
# DMA指标计算# 计算DIF: close的N1日移动平均-close的N2日移动平均# 计算AMA: DIF的M日移动平均def DMA(symbol, start_time, end_time, N1, N2, M):''' 计算DMA输入参数：symbol <- str 标的代码start_time <- str 开始时间end_time <- 结束时间N1 <- 大周期均值N2 <- 小周期均值输出参数：DMA <- dataframe'''# 取历史数据，取到上市首日data = history(symbol=symbol, frequency='1d', start_time='-01-01', end_time=end_time, fields='symbol,bob,close',df=True)data['MA1'] = data['close'].rolling(N1).mean()data['MA2'] = data['close'].rolling(N2).mean()data['DIF'] = data['MA1'] - data['MA2']data['AMA'] = data['DIF'].rolling(M).mean()# 将数据转化为dataframe格式data['bob'] = data['bob'].apply(lambda x: x.strftime('%Y-%m-%d')).tolist()# 按起止时间筛选DMA = data[(data['bob'] >= start_time)]return DMA# 测试一下d1 = DMA(symbol = 'DCE.y2101',start_time = '-01-01',end_time = '-10-22',N1 = 10,N2 = 50,M = 6)
BIAS指标
# BIAS：乖离率指标计算# BIAS1 : (CLOSE-N1日的平均值CLOSE）/N1日的平均值CLOSE*100# BIAS2 : (CLOSE-N2日的平均值CLOSE）/N2日的平均值CLOSE*100# BIAS3 : (CLOSE-N3日的平均值CLOSE）/N3日的平均值CLOSE*100def BIAS(symbol, start_time, end_time, N1, N2, N3):'''计算乖离率指标输入值：symbol <- str 标的代码start_time <- str 开始时间end_time <- str 结束时间N1、N2、N3 <- 移动平均数输出值：BIAS <- dataframe'''# 取历史数据，取到上市首日data = history(symbol=symbol, frequency='1d', start_time='-01-01', end_time=end_time, fields='symbol,bob,close',df=True)# 将数据转化为dataframe格式data['bob'] = data['bob'].apply(lambda x: x.strftime('%Y-%m-%d')).tolist()# 计算指标data['BIAS1'] = (data['close'] - data['close'].rolling(N1).mean())/data['close'].rolling(N1).mean() * 100data['BIAS2'] = (data['close'] - data['close'].rolling(N2).mean())/data['close'].rolling(N2).mean() * 100data['BIAS3'] = (data['close'] - data['close'].rolling(N3).mean())/data['close'].rolling(N3).mean() * 100# 按时间筛选BIAS = data[(data['bob'] >= start_time)]return BIAS# 测试一下d2 = BIAS(symbol='DCE.y2101', start_time='-01-01', end_time='-10-22', N1=6, N2=12, N3=24)
BOLL指标
# BOLL：BOLL带指标计算# 中轨线 = N日收盘价平均值# 上轨线 = 中轨线 + N日收盘价标准差# 下轨线 = 中轨线 - N日收盘价标准差def BOLL(symbol, start_time, end_time, N):''' 计算布林带输入参数：symbol <- str 标的代码start_time <- str 开始日期end_time <- str 结束日期N <- N日移动平均线输出参数：BOLL <- dataframe'''# 取历史数据，取到上市首日data = history(symbol=symbol, frequency='1d', start_time='-01-01', end_time=end_time, fields='symbol,bob,close',adjust=ADJUST_PREV, adjust_end_time=end_time, df=True)# 将数据转化为dataframe格式data['bob'] = data['bob'].apply(lambda x: x.strftime('%Y-%m-%d')).tolist()# 计算指标data['BOLL'] = data['close'].rolling(N).mean()data['UB'] = data['BOLL'] + 2 * data['close'].rolling(N).std()data['LB'] = data['BOLL'] - 2 * data['close'].rolling(N).std()# 按时间筛选BOLL = data[(data['bob'] >= start_time)]return BOLL# 测试一下d3 = BOLL(symbol='DCE.y2101', start_time='-01-01', end_time='-10-22', N=20)
RSI指标
# RSI指标计算# RSI = N日内收盘价涨数和的均值/N日内收盘价涨和跌的均值*100def RSI(symbol, start_time, end_time, N1, N2, N3):''' 计算RSI相对强弱指数输入参数：symbol <- str 标的代码start_time <- str 开始日期end_time <- str 结束日期N <- N日移动平均线输出参数：RSI <- dataframe'''# 取历史数据，取到上市首日data = history(symbol=symbol, frequency='1d', start_time='-01-01', end_time=end_time, fields='symbol,bob,close,pre_close',adjust=ADJUST_PREV, adjust_end_time=end_time, df=True)# 将数据转化为dataframe格式data['bob'] = data['bob'].apply(lambda x: x.strftime('%Y-%m-%d')).tolist()# 计算指标data['change'] = data['close'] - data['pre_close']# 计算涨跌幅data.loc[(data['pre_close'] == 0), 'change'] = 0 # 如果是首日，change记为0data['x'] = data['change'].apply(lambda x: max(x, 0)) # 涨跌幅<0换为0data['RSI1'] = data['x'].ewm(alpha=1 / N1, adjust=False).mean() / (np.abs(data['change']).ewm(alpha=1/N1, adjust=False).mean()) * 100data['RSI2'] = data['x'].ewm(alpha=1 / N2, adjust=False).mean() / (np.abs(data['change']).ewm(alpha=1 / N2, adjust=False).mean()) * 100data['RSI3'] = data['x'].ewm(alpha=1 / N3, adjust=False).mean() / (np.abs(data['change']).ewm(alpha=1 / N3, adjust=False).mean()) * 100# 输出RSI = data[(data['bob'] >= start_time)]return RSI# 测试一下d4 = RSI(symbol='DCE.y2101', start_time='-01-01', end_time='-10-22', N1=6, N2=12, N3=24)
威廉指标WR
# WR威廉指标计算# WR(N) = 100 * [HIGH(N)-C] / [HIGH(N)-LOW(N)]def WR(symbol, start_time, end_time, N1, N2):''' 计算威廉指数输入：symbol <- str 标的代码start_time <- str 开始时间end_time <- 结束时间N <- 周期数输出：WR <- dataframe'''# 取历史数据，取到上市首日data = history(symbol=symbol, frequency='1d', start_time='-01-01', end_time=end_time,fields='symbol,bob,close,high,low',adjust=ADJUST_PREV, adjust_end_time=end_time, df=True)# 计算指标data['WR1'] = 100 * (data['high'].rolling(N1).max() - data['close']) / (data['high'].rolling(N1).max() - data['low'].rolling(N1).min())data['WR2'] = 100 * (data['high'].rolling(N2).max() - data['close']) / (data['high'].rolling(N2).max() - data['low'].rolling(N2).min())# 缺失值填充data['WR1'].fillna(value=100 * (data['high'].expanding().max() - data['close']) / (data['high'].expanding().max() - data['low'].expanding().min()), inplace=True)data['WR2'].fillna(value=100 * (data['high'].expanding().max() - data['close']) / (data['high'].expanding().max() - data['low'].expanding().min()), inplace=True)# 输出WR = data[(data['bob'] >= start_time)]return WR# 测试一下d5 = WR(symbol='DCE.y2101', start_time='-01-01', end_time='-10-22', N1 = 10,N2 = 6)
