4.7 DMI ? DMI 指標體系的構建及簡單應用 · Python 量化交易教程

# 4.7 DMI ? DMI 指標體系的構建及簡單應用 > 來源：https://uqer.io/community/share/561727aaf9f06c4ca92fb5a4 DMI指標又叫動向指標或趨向指標，是通過分析股票價格在漲跌過程中買賣雙方力量均衡點的變化情況，即多空雙方的力量的變化受價格波動的影響而發生由均衡到失衡的循環過程，從而提供對趨勢判斷依據的一種技術指標。其由美國技術分析大師威爾斯·威爾德（Wells Wilder）所創造，是一種中長期股市技術分析方法。 DMI指標體系的構建： ``` TR = SUM(MAX(MAX(HIGH - LOW, ABS(HIGH-REF(CLOSE,1))), ABS(LOW - REF(CLOSE, 1))), N) HD = HIGH - REF(HIGH, 1) LD = REF(LOW, 1) - LOW DMP = SUM(IF(HD>0 AND HD>LD, HD, 0), N) DMM = SUM(IF(LD>0 AND LD>HD, LD, 0), N) PDI = DMP*100/TR MDI = DMM*100/TR DX = ABS(MDI - PDI)/(MDI + PDI)*100 ADX = MA(ABS(MDI - PDI)/(MDI + PDI)*100, M) ``` 其中變量與函數定義如下： + `CLOSE`：引用收盤價(在盤中指最新價) + `HIGH`：引用最高價 + `LOW`：引用最低價 + `REF(X, N)`：引用X在N個周期前的值 + `ABS(X)`：求X的絕對值 + `MAX(A, B)`：求A，B中的較大者 + `SUM(X, N)`：得到X在N周期內的總和 + `IF(C, A, B)`：如果C成立返回A，否則返回B 此外，`PDI`簡記為`+DI`，`MDI`簡記為`-DI`；參數：`N=14`(默認)，`M=14` (默認)。實際上從數學上看，`DX`或`ADX`的構建并不一定需要`PDI`與`MDI`，有`DMP`和`DMM`就行了。計算`PDI`與`MDI`是將指標數值控制在0到100之間。 ```py #計算某一天的股票DMP與DMM值 def eq_DMPandDMM(stk_list,current_date,N=14): cal = Calendar('China.SSE') period = '-' + str(N+1) + 'B' begin_date = cal.advanceDate(current_date,period,BizDayConvention.Unadjusted) end_date = cal.advanceDate(current_date,'-1B',BizDayConvention.Unadjusted) eq_hd = {} eq_ld = {} dmp_sum = 0 dmm_sum = 0 eq_dmp = {} eq_dmm = {} eq_Price = DataAPI.MktEqudAdjGet(secID=stk_list,beginDate=begin_date.strftime('%Y%m%d'),endDate=end_date.strftime('%Y%m%d'),field=['secID','highestPrice','lowestPrice'],pandas="1") avaiable_list = eq_Price['secID'].drop_duplicates().tolist() eq_Price.set_index('secID',inplace=True) for stk in avaiable_list: if len(eq_Price.ix[stk]) == (N+1): eq_hd[stk] = np.array(eq_Price.ix[stk]['highestPrice'][1:] - eq_Price.ix[stk]['highestPrice'][:-1]) eq_ld[stk] = np.array(eq_Price.ix[stk]['lowestPrice'][:-1] - eq_Price.ix[stk]['lowestPrice'][1:]) for i in xrange(len(eq_ld[stk])): if eq_hd[stk][i] > 0 and eq_hd[stk][i] > eq_ld[stk][i]: dmp_sum = dmp_sum + eq_hd[stk][i] if eq_ld[stk][i] > 0 and eq_ld[stk][i] > eq_hd[stk][i]: dmm_sum = dmm_sum + eq_ld[stk][i] eq_dmp[stk] = dmp_sum eq_dmm[stk] = dmm_sum dmm_sum = 0 dmp_sum = 0 return eq_dmp,eq_dmm ``` ```py #計算某一天股票的TR值 def eq_TR(stk_list,current_date,N=14): cal = Calendar('China.SSE') period = '-' + str(N+1) + 'B' begin_date = cal.advanceDate(current_date,period,BizDayConvention.Unadjusted) end_date = cal.advanceDate(current_date,'-1B',BizDayConvention.Unadjusted) eq_hl = {} #HIGH - LOW eq_hc = {} #HIGH - CLOSE eq_lc = {} #LOW - CLOSE eq_tr = {} tr_sum = 0 eq_Price = DataAPI.MktEqudAdjGet(secID=stk_list,beginDate=begin_date.strftime('%Y%m%d'),endDate=end_date.strftime('%Y%m%d'),field=['secID','highestPrice','lowestPrice','closePrice'],pandas="1") avaiable_list = eq_Price['secID'].drop_duplicates().tolist() eq_Price.set_index('secID',inplace=True) for stk in avaiable_list: if len(eq_Price.ix[stk]) == (N+1): eq_hl[stk] = np.array(eq_Price.ix[stk]['highestPrice'][1:] - eq_Price.ix[stk]['lowestPrice'][1:]) eq_hc[stk] = np.array(eq_Price.ix[stk]['highestPrice'][1:] - eq_Price.ix[stk]['closePrice'][:-1]) eq_lc[stk] = np.array(eq_Price.ix[stk]['lowestPrice'][:-1] - eq_Price.ix[stk]['closePrice'][1:]) for i in xrange(len(eq_hl[stk])): tr_sum = tr_sum + max(max(eq_hl[stk][i],abs(eq_hc[stk][i])),abs(eq_lc[stk][i])) eq_tr[stk] = tr_sum tr_sum = 0 return eq_tr ``` ```py #計算某一天股票的ADX def eq_ADX(stk_list,current_date,N=14): cal = Calendar('China.SSE') period = '-' + str(N) + 'B' begin_date = cal.advanceDate(current_date,period,BizDayConvention.Unadjusted) end_date = cal.advanceDate(current_date,'-1B',BizDayConvention.Unadjusted) timeSeries = cal.bizDatesList(begin_date,end_date) eq_adx = {} adx_sum = 0 #初始化eq_adx eq_Price = DataAPI.MktEqudAdjGet(secID=stk_list,beginDate=begin_date.strftime('%Y%m%d'),endDate=end_date.strftime('%Y%m%d'),field=['secID','highestPrice','lowestPrice'],pandas="1") avaiable_list = eq_Price['secID'].drop_duplicates().tolist() eq_Price.set_index('secID',inplace=True) for stk in avaiable_list: if len(eq_Price.ix[stk]) == N: eq_adx[stk] = 0 #計算ADX for i in xrange(len(timeSeries)): eq_dmp,eq_dmm = eq_DMPandDMM(stk_list,timeSeries[i],N) for stk in eq_dmp: if eq_dmp[stk] == 0 and eq_dmm[stk] == 0: #當DMP與DMM都為零時，認為無趨勢DX=0 pass else: eq_adx[stk] = eq_adx[stk] + abs(eq_dmp[stk] - eq_dmm[stk])/(eq_dmp[stk] + eq_dmm[stk])*100 for stk in eq_adx: eq_adx[stk] = eq_adx[stk] / len(timeSeries) return eq_adx ``` ## 簡單應用：當`DMP`上穿`DMM`時，意味著，上漲傾向強于下跌傾向，一個買入信號生成。反之則反。而`ADX`用于反映趨向變動的程度，在買入信號時，`ADX`伴隨上升，則預示股價的漲勢可能更強勁。 ```py import numpy as np import pandas as pd from CAL.PyCAL import * start = '2012-08-01' # 回測起始時間 end = '2015-08-01' # 回測結束時間 benchmark = 'HS300' # 策略參考標準 universe = set_universe('HS300') # 證券池，支持股票和基金 capital_base = 1000000 # 起始資金 freq = 'd' # 策略類型，'d'表示日間策略使用日線回測，'m'表示日內策略使用分鐘線回測 refresh_rate = 20 # 調倉頻率，表示執行handle_data的時間間隔，若freq = 'd'時間間隔的單位為交易日，若freq = 'm'時間間隔為分鐘 cal = Calendar('China.SSE') def initialize(account): # 初始化虛擬賬戶狀態 pass def handle_data(account): # 每個交易日的買入賣出指令 eq_dmp_now,eq_dmm_now = eq_DMPandDMM(account.universe,account.current_date,14) eq_adx = eq_ADX(account.universe,account.current_date,14) yestoday = cal.advanceDate(account.current_date,'-1B',BizDayConvention.Unadjusted) eq_dmp_before,eq_dmm_before = eq_DMPandDMM(account.universe,yestoday,14) eq_adx_before = eq_ADX(account.universe,yestoday,14) long_bucket = [] short_bucket = [] for stk in account.universe: try: if eq_dmp_now[stk] > eq_dmm_now[stk] and eq_dmp_before[stk] < eq_dmm_before[stk] and eq_adx[stk] > eq_adx_before[stk]: long_bucket.append(stk) else: short_bucket.append(stk) except: pass #調倉邏輯是調倉時將所有滿足條件的股票等權 stk_num = len(account.valid_secpos) + len(long_bucket) for stk in account.valid_secpos: if stk in short_bucket: order_to(stk,0) stk_num = stk_num - 1 for stk in account.valid_secpos: if stk not in short_bucket: order_to(stk,account.referencePortfolioValue/account.referencePrice[stk]/stk_num) for stk in long_bucket: if stk not in account.avail_secpos: order_to(stk,account.referencePortfolioValue/account.referencePrice[stk]/stk_num) ``` ![](https://box.kancloud.cn/2016-07-30_579cbb033b457.jpg)