python自动交易源码_【硬核福利】量化交易神器talib中28个技术指标的Python实现(附全部源码)…

内容首发

乐学偶得(http://lexueoude.com) 公众号: 乐学Fintech

用代码理解分析解决金融问题

之前跟大家分享过用Python调用talib实现技术指标分析,但是许多小伙伴有更高的需求:比如需要指标自定义,或者想明白技术分析背后的原理。所以,这一期我们跟大家分享一下通过纯Python+Pandas+Numpy+Math实现talib中的常见指标,是学习底层算法与自定义交易指标,提升内功的非常好的材料。

我们需要的库非常简单,只需要这三个:

import numpy

from pandas import *

from math import *

关于pandas版本的问题,我们视频中跟大家分享过,注意以下用法需要pandas的版本为0.21,若报ewma无法调用的错误,可以指定安装此版本的pandas,或者通过

pandas.DataFrame(ts).ewm(span=12).mean()

这样调用的方法解决。

1.移动平均

def MA(df, n):

MA = Series(rolling_mean(df['Close'], n), name = 'MA_' + str(n))

df = df.join(MA)

return df

2.指数移动平均

def EMA(df, n):

EMA = Series(ewma(df['Close'], span = n, min_periods = n – 1), name = 'EMA_' + str(n))

df = df.join(EMA)

return df

3.动量

def MOM(df, n):

M = Series(df['Close'].diff(n), name = 'Momentum_' + str(n))

df = df.join(M)

return df

4.变化率

def ROC(df, n):

M = df['Close'].diff(n – 1)

N = df['Close'].shift(n – 1)

ROC = Series(M / N, name = 'ROC_' + str(n))

df = df.join(ROC)

return df

5.均幅指标

def ATR(df, n):

i = 0

TR_l = [0]

while i < df.index[-1]:

TR = max(df.get_value(i + 1, 'High'), df.get_value(i, 'Close')) – min(df.get_value(i + 1, 'Low'), df.get_value(i, 'Close'))

TR_l.append(TR)

i = i + 1

TR_s = Series(TR_l)

ATR = Series(ewma(TR_s, span = n, min_periods = n), name = 'ATR_' + str(n))

df = df.join(ATR)

return df

6.布林线

def BBANDS(df, n):

MA = Series(rolling_mean(df['Close'], n))

MSD = Series(rolling_std(df['Close'], n))

b1 = 4 * MSD / MA

B1 = Series(b1, name = 'BollingerB_' + str(n))

df = df.join(B1)

b2 = (df['Close'] – MA + 2 * MSD) / (4 * MSD)

B2 = Series(b2, name = 'Bollinger%b_' + str(n))

df = df.join(B2)

return df

7.转折、支撑、阻力点

def PPSR(df):

PP = Series((df['High'] + df['Low'] + df['Close']) / 3)

R1 = Series(2 * PP – df['Low'])

S1 = Series(2 * PP – df['High'])

R2 = Series(PP + df['High'] – df['Low'])

S2 = Series(PP – df['High'] + df['Low'])

R3 = Series(df['High'] + 2 * (PP – df['Low']))

S3 = Series(df['Low'] – 2 * (df['High'] – PP))

psr = {'PP':PP, 'R1':R1, 'S1':S1, 'R2':R2, 'S2':S2, 'R3':R3, 'S3':S3}

PSR = DataFrame(psr)

df = df.join(PSR)

return df

8.随机振荡器(%K线)

def STOK(df):

SOk = Series((df['Close'] – df['Low']) / (df['High'] – df['Low']), name = 'SO%k')

df = df.join(SOk)

return df

9.随机振荡器(%D线)

def STO(df, n):

SOk = Series((df['Close'] – df['Low']) / (df['High'] – df['Low']), name = 'SO%k')

SOd = Series(ewma(SOk, span = n, min_periods = n – 1), name = 'SO%d_' + str(n))

df = df.join(SOd)

return df

10.三重指数平滑平均线

def TRIX(df, n):

EX1 = ewma(df['Close'], span = n, min_periods = n – 1)

EX2 = ewma(EX1, span = n, min_periods = n – 1)

EX3 = ewma(EX2, span = n, min_periods = n – 1)

i = 0

ROC_l = [0]

while i + 1 <= df.index[-1]:

ROC = (EX3[i + 1] – EX3[i]) / EX3[i]

ROC_l.append(ROC)

i = i + 1

Trix = Series(ROC_l, name = 'Trix_' + str(n))

df = df.join(Trix)

return df

11.平均定向运动指数

def ADX(df, n, n_ADX):

i = 0

UpI = []

DoI = []

while i + 1 <= df.index[-1]:

UpMove = df.get_value(i + 1, 'High') – df.get_value(i, 'High')

DoMove = df.get_value(i, 'Low') – df.get_value(i + 1, 'Low')

if UpMove > DoMove and UpMove > 0:

UpD = UpMove

else: UpD = 0

UpI.append(UpD)

if DoMove > UpMove and DoMove > 0:

DoD = DoMove

else: DoD = 0

DoI.append(DoD)

i = i + 1

i = 0

TR_l = [0]

while i < df.index[-1]:

TR = max(df.get_value(i + 1, 'High'), df.get_value(i, 'Close')) – min(df.get_value(i + 1, 'Low'), df.get_value(i, 'Close'))

TR_l.append(TR)

i = i + 1

TR_s = Series(TR_l)

ATR = Series(ewma(TR_s, span = n, min_periods = n))

UpI = Series(UpI)

DoI = Series(DoI)

PosDI = Series(ewma(UpI, span = n, min_periods = n – 1) / ATR)

NegDI = Series(ewma(DoI, span = n, min_periods = n – 1) / ATR)

ADX = Series(ewma(abs(PosDI – NegDI) / (PosDI + NegDI), span = n_ADX, min_periods = n_ADX – 1), name = 'ADX_' + str(n) + '_' + str(n_ADX))

df = df.join(ADX)

return df

12.MACD

def MACD(df, n_fast, n_slow):

EMAfast = Series(ewma(df['Close'], span = n_fast, min_periods = n_slow – 1))

EMAslow = Series(ewma(df['Close'], span = n_slow, min_periods = n_slow – 1))

MACD = Series(EMAfast – EMAslow, name = 'MACD_' + str(n_fast) + '_' + str(n_slow))

MACDsign = Series(ewma(MACD, span = 9, min_periods = 8), name = 'MACDsign_' + str(n_fast) + '_' + str(n_slow))

MACDdiff = Series(MACD – MACDsign, name = 'MACDdiff_' + str(n_fast) + '_' + str(n_slow))

df = df.join(MACD)

df = df.join(MACDsign)

df = df.join(MACDdiff)

return df

13.梅斯线(高低价趋势反转)

def MassI(df):

Range = df['High'] – df['Low']

EX1 = ewma(Range, span = 9, min_periods = 8)

EX2 = ewma(EX1, span = 9, min_periods = 8)

Mass = EX1 / EX2

MassI = Series(rolling_sum(Mass, 25), name = 'Mass Index')

df = df.join(MassI)

return df

14.涡旋指标

def Vortex(df, n):

i = 0

TR = [0]

while i < df.index[-1]:

Range = max(df.get_value(i + 1, 'High'), df.get_value(i, 'Close')) – min(df.get_value(i + 1, 'Low'), df.get_value(i, 'Close'))

TR.append(Range)

i = i + 1

i = 0

VM = [0]

while i < df.index[-1]:

Range = abs(df.get_value(i + 1, 'High') – df.get_value(i, 'Low')) – abs(df.get_value(i + 1, 'Low') – df.get_value(i, 'High'))

VM.append(Range)

i = i + 1

VI = Series(rolling_sum(Series(VM), n) / rolling_sum(Series(TR), n), name = 'Vortex_' + str(n))

df = df.join(VI)

return df

15.KST振荡器

def KST(df, r1, r2, r3, r4, n1, n2, n3, n4):

M = df['Close'].diff(r1 – 1)

N = df['Close'].shift(r1 – 1)

ROC1 = M / N

M = df['Close'].diff(r2 – 1)

N = df['Close'].shift(r2 – 1)

ROC2 = M / N

M = df['Close'].diff(r3 – 1)

N = df['Close'].shift(r3 – 1)

ROC3 = M / N

M = df['Close'].diff(r4 – 1)

N = df['Close'].shift(r4 – 1)

ROC4 = M / N

KST = Series(rolling_sum(ROC1, n1) + rolling_sum(ROC2, n2) * 2 + rolling_sum(ROC3, n3) * 3 + rolling_sum(ROC4, n4) * 4, name = 'KST_' + str(r1) + '_' + str(r2) + '_' + str(r3) + '_' + str(r4) + '_' + str(n1) + '_' + str(n2) + '_' + str(n3) + '_' + str(n4))

df = df.join(KST)

return df

16.相对强度指标

def RSI(df, n):

i = 0

UpI = [0]

DoI = [0]

while i + 1 <= df.index[-1]:

UpMove = df.get_value(i + 1, 'High') – df.get_value(i, 'High')

DoMove = df.get_value(i, 'Low') – df.get_value(i + 1, 'Low')

if UpMove > DoMove and UpMove > 0:

UpD = UpMove

else: UpD = 0

UpI.append(UpD)

if DoMove > UpMove and DoMove > 0:

DoD = DoMove

else: DoD = 0

DoI.append(DoD)

i = i + 1

UpI = Series(UpI)

DoI = Series(DoI)

PosDI = Series(ewma(UpI, span = n, min_periods = n – 1))

NegDI = Series(ewma(DoI, span = n, min_periods = n – 1))

RSI = Series(PosDI / (PosDI + NegDI), name = 'RSI_' + str(n))

df = df.join(RSI)

return df

17.真实强度指标

def TSI(df, r, s):

M = Series(df['Close'].diff(1))

aM = abs(M)

EMA1 = Series(ewma(M, span = r, min_periods = r – 1))

aEMA1 = Series(ewma(aM, span = r, min_periods = r – 1))

EMA2 = Series(ewma(EMA1, span = s, min_periods = s – 1))

aEMA2 = Series(ewma(aEMA1, span = s, min_periods = s – 1))

TSI = Series(EMA2 / aEMA2, name = 'TSI_' + str(r) + '_' + str(s))

df = df.join(TSI)

return df

18.吸筹/派发指标

def ACCDIST(df, n):

ad = (2 * df['Close'] – df['High'] – df['Low']) / (df['High'] – df['Low']) * df['Volume']

M = ad.diff(n – 1)

N = ad.shift(n – 1)

ROC = M / N

AD = Series(ROC, name = 'Acc/Dist_ROC_' + str(n))

df = df.join(AD)

return df

19.佳庆指标CHAIKIN振荡器

def Chaikin(df):

ad = (2 * df['Close'] – df['High'] – df['Low']) / (df['High'] – df['Low']) * df['Volume']

Chaikin = Series(ewma(ad, span = 3, min_periods = 2) – ewma(ad, span = 10, min_periods = 9), name = 'Chaikin')

df = df.join(Chaikin)

return df

20.资金流量与比率指标

def MFI(df, n):

PP = (df['High'] + df['Low'] + df['Close']) / 3

i = 0

PosMF = [0]

while i < df.index[-1]:

if PP[i + 1] > PP[i]:

PosMF.append(PP[i + 1] * df.get_value(i + 1, 'Volume'))

else:

PosMF.append(0)

i = i + 1

PosMF = Series(PosMF)

TotMF = PP * df['Volume']

MFR = Series(PosMF / TotMF)

MFI = Series(rolling_mean(MFR, n), name = 'MFI_' + str(n))

df = df.join(MFI)

return df

21.能量潮指标

def OBV(df, n):

i = 0

OBV = [0]

while i < df.index[-1]:

if df.get_value(i + 1, 'Close') – df.get_value(i, 'Close') > 0:

OBV.append(df.get_value(i + 1, 'Volume'))

if df.get_value(i + 1, 'Close') – df.get_value(i, 'Close') == 0:

OBV.append(0)

if df.get_value(i + 1, 'Close') – df.get_value(i, 'Close') < 0:

OBV.append(-df.get_value(i + 1, 'Volume'))

i = i + 1

OBV = Series(OBV)

OBV_ma = Series(rolling_mean(OBV, n), name = 'OBV_' + str(n))

df = df.join(OBV_ma)

return df

22.强力指数指标

def FORCE(df, n):

F = Series(df['Close'].diff(n) * df['Volume'].diff(n), name = 'Force_' + str(n))

df = df.join(F)

return df

23.简易波动指标

def EOM(df, n):

EoM = (df['High'].diff(1) + df['Low'].diff(1)) * (df['High'] – df['Low']) / (2 * df['Volume'])

Eom_ma = Series(rolling_mean(EoM, n), name = 'EoM_' + str(n))

df = df.join(Eom_ma)

return df

24.顺势指标

def CCI(df, n):

PP = (df['High'] + df['Low'] + df['Close']) / 3

CCI = Series((PP – rolling_mean(PP, n)) / rolling_std(PP, n), name = 'CCI_' + str(n))

df = df.join(CCI)

return df

25.估波指标

def COPP(df, n):

M = df['Close'].diff(int(n * 11 / 10) – 1)

N = df['Close'].shift(int(n * 11 / 10) – 1)

ROC1 = M / N

M = df['Close'].diff(int(n * 14 / 10) – 1)

N = df['Close'].shift(int(n * 14 / 10) – 1)

ROC2 = M / N

Copp = Series(ewma(ROC1 + ROC2, span = n, min_periods = n), name = 'Copp_' + str(n))

df = df.join(Copp)

return df

26.肯特纳通道

def KELCH(df, n):

KelChM = Series(rolling_mean((df['High'] + df['Low'] + df['Close']) / 3, n), name = 'KelChM_' + str(n))

KelChU = Series(rolling_mean((4 * df['High'] – 2 * df['Low'] + df['Close']) / 3, n), name = 'KelChU_' + str(n))

KelChD = Series(rolling_mean((-2 * df['High'] + 4 * df['Low'] + df['Close']) / 3, n), name = 'KelChD_' + str(n))

df = df.join(KelChM)

df = df.join(KelChU)

df = df.join(KelChD)

return df

27.终极指标(终极振荡器)

def ULTOSC(df):

i = 0

TR_l = [0]

BP_l = [0]

while i < df.index[-1]:

TR = max(df.get_value(i + 1, 'High'), df.get_value(i, 'Close')) – min(df.get_value(i + 1, 'Low'), df.get_value(i, 'Close'))

TR_l.append(TR)

BP = df.get_value(i + 1, 'Close') – min(df.get_value(i + 1, 'Low'), df.get_value(i, 'Close'))

BP_l.append(BP)

i = i + 1

UltO = Series((4 * rolling_sum(Series(BP_l), 7) / rolling_sum(Series(TR_l), 7)) + (2 * rolling_sum(Series(BP_l), 14) / rolling_sum(Series(TR_l), 14)) + (rolling_sum(Series(BP_l), 28) / rolling_sum(Series(TR_l), 28)), name = 'Ultimate_Osc')

df = df.join(UltO)

return df

28.唐奇安通道指标

def DONCH(df, n):

i = 0

DC_l = []

while i < n – 1:

DC_l.append(0)

i = i + 1

i = 0

while i + n – 1 < df.index[-1]:

DC = max(df['High'].ix[i:i + n – 1]) – min(df['Low'].ix[i:i + n – 1])

DC_l.append(DC)

i = i + 1

DonCh = Series(DC_l, name = 'Donchian_' + str(n))

DonCh = DonCh.shift(n – 1)

df = df.join(DonCh)

return df

参考资料:

乐学偶得《Python零基础入门到6大方向热门应用》、《零基础Python玩Fintech金融量化》、《Python股票量化投资策略与交易》等笔记

开源项目:ultrafinance

内容首发

公众号: 乐学Fintech

用代码理解分析解决金融问题

Published by

风君子

独自遨游何稽首 揭天掀地慰生平

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注