相对活力指数指标（RVI）量化交易

77 阅读 0 评论 51 点赞

我是靠谱客的博主着急犀牛，这篇文章主要介绍相对活力指数指标（RVI）量化交易，现在分享给大家，希望可以做个参考。

背景：

在经过一段时间对backtrader的学习之后，将用于回测的backtrader模型进行修改，修改成为可以进行实盘模拟的代码。

其中下面这篇文章对代码修改起到了很大作用

量化回测框架BackTrader【6】-开发策略

代码还有很多不足和奇怪的地方

希望大神们告诉我谢谢

以下是代码部分：

复制代码

from huobi.client.generic import *
from huobi.client.market import *
from huobi.constant.definition import *
from huobi.client.account import *
from huobi.client.trade import *
from huobi.client.market import *
import pandas as pd
import datetime as dt
import time
import talib as ta
import backtrader as bt
import numpy as np
import math
class RVI(bt.Indicator):
lines = ('std','pos','neg','rvi')
plotlines = dict(
std=dict(_plotskip=True),
pos=dict(_plotskip=True),
neg=dict(_plotskip=True),
rvi=dict(_plotskip=False)
)
params = (
('period', 20),
)
def __init__(self):
self.lines.std = bt.talib.STDDEV(self.datas[0].close, timeperiod=8, nbdev=2.0)
def next(self):
if
self.lines.std[0] > self.lines.std[-1]:
self.lines.pos[0] = self.lines.std[0]
else:
self.lines.pos[0] = 0
if self.lines.std[0] < self.lines.std[-1]:
self.lines.neg[0] = self.lines.std[0]
else:
self.lines.neg[0] = 0
pos_nan = np.nan_to_num(self.lines.pos.get(size=self.params.period))
neg_nan = np.nan_to_num(self.lines.neg.get(size=self.params.period))
Usum = math.fsum(pos_nan)
Dsum = math.fsum(neg_nan)
if (Usum + Dsum) == 0:
self.lines.rvi[0] = 0
return
self.lines.rvi[0] = 100 * Usum / (Usum+Dsum)
class read_csv():
def __init__(self):
# time(时间),currency(币种),balance(余额),type(类型),symbol(交易对),amount(交易量),price(交易价格),order-id(订单ID)
self.list_new = []
#交易日期及时间
now_time = dt.datetime.now().strftime('%F %T')
self.list_new.append(now_time)
def readcsv(self):
self.date = pd.read_csv('交易模拟测试.csv',encoding='utf-8')
try :
# 取最后一行symbol数据
date_old = self.date.iloc[-1,:]
logger.info('取最后一行symbol数据')
symbol_old = date_old['symbol']
if date_old['symbol'] != 'usdt' :
symbol_old = symbol_old.replace('usdt', '')
amount_new = date_old['amount'] * 0.9965
self.list_new.append(symbol_old)
self.list_new.append(amount_new)
except IndexError as err:
spot = 1547906
account_client = AccountClient(api_key=g_api_key, secret_key=g_secret_key)
balance_list = account_client.get_balance(spot)
# 账户中交易币种信息及余额
c_list = ['usdt', 'eth3s', 'eth3l']
c_list_1 = []
for i in balance_list:
currency = i.__dict__
if currency['currency'] in c_list:
c_list_1.append(currency)
date_1 = pd.DataFrame(c_list_1)
date_1['balance'] = date_1['balance'].astype('float')
trade_1 = date_1.loc[(date_1['type'] == 'trade') & (date_1['balance'] > 0.01)]
trade_1.reset_index(drop=True, inplace=True)
self.list_new.append(trade_1['currency'][0])
self.list_new.append(trade_1['balance'][0])
return self.list_new, self.date
class huobi_Strategy(bt.Strategy):
def __init__(self):
self.list_new, self.date = read_csv().readcsv()
self.rvi = RVI()
def amount_result(self,num):
# 取小数点后四位
self.num = num
num_x, num_y = str(self.num).split('.')
self.num = float(num_x + '.' + num_y[0:4])
return self.num
def market_data(self):
# 取得最新价格
market_client = MarketClient(api_key=g_api_key, secret_key=g_secret_key)
market_trade = market_client.get_market_trade('eth3lusdt')
for i in market_trade:
market_trade_1 = i.__dict__
# order_min = round(5.5 / market_trade_1['price'], 4)
# 最小订单数量为 5/价格
self.price_1 = market_trade_1['price']
return self.price_1
def huobi_buy(self):
if self.list_new[1] == 'usdt' :
price_1 = huobi_Strategy().market_data()
amount_1 = huobi_Strategy().amount_result(self.list_new[2] * price_1)
self.list_new.append('buy')
self.list_new.append('eth3l')
self.list_new.append(amount_1)
self.list_new.append(price_1)
self.list_new.append('id')
f = pd.DataFrame(self.list_new).T
f.columns = self.columns
date = pd.concat([self.date,f])
date.to_csv('交易模拟测试.csv',encoding='utf-8',index=False)
def huobi_sell(self):
if self.list_new[1] == 'eth3lusdt' :
price_1 = huobi_Strategy().market_data()
amount_1 = huobi_Strategy().amount_result(self.list_new[2] / price_1)
self.list_new.append('sell')
self.list_new.append('usdt')
self.list_new.append(amount_1)
self.list_new.append(price_1)
self.list_new.append('id')
# csv最后增加一行
f = pd.DataFrame(self.list_new).T
f.columns = self.columns
date = pd.concat([self.date, f])
date.to_csv('交易模拟测试.csv', encoding='utf-8', index=False)
def next(self):
up = 80
down = 20
self.order = 'keep'
if self.rvi.rvi[0] > up:
if self.rvi.rvi[-1] < up and self.rvi.rvi[-2] < up:
self.order = 'buy'
elif self.rvi.rvi[0] < down:
if self.rvi.rvi[-1] > down and self.rvi.rvi[-2] > down:
self.order = 'sell'
return
def stop(self):
if self.order == 'buy':
huobi_Strategy().huobi_buy()
elif self.order == 'sell':
huobi_Strategy().huobi_sell()
def get_data():
# 取当前及30min前的烛台数据
market_client = MarketClient()
data_ethusdt = market_client.get_candlestick('ethusdt', CandlestickInterval.MIN1, 100)
id_1 = []
high_1 = []
low_1 = []
open_1 = []
close_1 = []
vol_1 = []
for i in range(len(data_ethusdt) - 1, -1, -1):
id_1 = data_ethusdt[i].id
high_1.append(data_ethusdt[i].high)
low_1.append(data_ethusdt[i].low)
open_1.append(data_ethusdt[i].open)
close_1.append(data_ethusdt[i].close)
# amount = data_ethusdt[i].amount
# count = data_ethusdt[i].count
vol_1.append(data_ethusdt[i].vol)
datas = pd.DataFrame({'datetime': id_1, 'open': open_1,
'high': high_1, 'low': low_1,
'close': close_1, 'volume': vol_1})
datas['datetime'] = pd.to_datetime(datas['datetime'], unit='s')
datas.set_index('datetime', drop=True, append=False)
return datas
'''
程序结束
'''
# 程序执行：
if __name__ == '__main__':
while True :
time_start = time.perf_counter()
logger.info('-'*5 + 'start' + '-'*5)
# 策略加载
df = get_data()
cerebro = bt.Cerebro()
data = bt.feeds.PandasData(dataname=df,datetime='datetime',
open='open',high='high',low='low',
close='close',volume='volume',openinterest=-1)
cerebro.adddata(data)
cerebro.addstrategy(huobi_Strategy)
result = cerebro.run()
# 统计运行时间
time_used = time.perf_counter() - time_start
time.sleep(60)

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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
from huobi.client.generic import *
from huobi.client.market import *
from huobi.constant.definition import *
from huobi.client.account import *
from huobi.client.trade import *
from huobi.client.market import *
import pandas as pd
import datetime as dt
import time
import talib as ta
import backtrader as bt
import numpy as np
import math
class RVI(bt.Indicator):
lines = ('std','pos','neg','rvi')
plotlines = dict(
std=dict(_plotskip=True),
pos=dict(_plotskip=True),
neg=dict(_plotskip=True),
rvi=dict(_plotskip=False)
)
params = (
('period', 20),
)
def __init__(self):
self.lines.std = bt.talib.STDDEV(self.datas[0].close, timeperiod=8, nbdev=2.0)
def next(self):
if
self.lines.std[0] > self.lines.std[-1]:
self.lines.pos[0] = self.lines.std[0]
else:
self.lines.pos[0] = 0
if self.lines.std[0] < self.lines.std[-1]:
self.lines.neg[0] = self.lines.std[0]
else:
self.lines.neg[0] = 0
pos_nan = np.nan_to_num(self.lines.pos.get(size=self.params.period))
neg_nan = np.nan_to_num(self.lines.neg.get(size=self.params.period))
Usum = math.fsum(pos_nan)
Dsum = math.fsum(neg_nan)
if (Usum + Dsum) == 0:
self.lines.rvi[0] = 0
return
self.lines.rvi[0] = 100 * Usum / (Usum+Dsum)
class read_csv():
def __init__(self):
# time(时间),currency(币种),balance(余额),type(类型),symbol(交易对),amount(交易量),price(交易价格),order-id(订单ID)
self.list_new = []
#交易日期及时间
now_time = dt.datetime.now().strftime('%F %T')
self.list_new.append(now_time)
def readcsv(self):
self.date = pd.read_csv('交易模拟测试.csv',encoding='utf-8')
try :
# 取最后一行symbol数据
date_old = self.date.iloc[-1,:]
logger.info('取最后一行symbol数据')
symbol_old = date_old['symbol']
if date_old['symbol'] != 'usdt' :
symbol_old = symbol_old.replace('usdt', '')
amount_new = date_old['amount'] * 0.9965
self.list_new.append(symbol_old)
self.list_new.append(amount_new)
except IndexError as err:
spot = 1547906
account_client = AccountClient(api_key=g_api_key, secret_key=g_secret_key)
balance_list = account_client.get_balance(spot)
# 账户中交易币种信息及余额
c_list = ['usdt', 'eth3s', 'eth3l']
c_list_1 = []
for i in balance_list:
currency = i.__dict__
if currency['currency'] in c_list:
c_list_1.append(currency)
date_1 = pd.DataFrame(c_list_1)
date_1['balance'] = date_1['balance'].astype('float')
trade_1 = date_1.loc[(date_1['type'] == 'trade') & (date_1['balance'] > 0.01)]
trade_1.reset_index(drop=True, inplace=True)
self.list_new.append(trade_1['currency'][0])
self.list_new.append(trade_1['balance'][0])
return self.list_new, self.date
class huobi_Strategy(bt.Strategy):
def __init__(self):
self.list_new, self.date = read_csv().readcsv()
self.rvi = RVI()
def amount_result(self,num):
# 取小数点后四位
self.num = num
num_x, num_y = str(self.num).split('.')
self.num = float(num_x + '.' + num_y[0:4])
return self.num
def market_data(self):
# 取得最新价格
market_client = MarketClient(api_key=g_api_key, secret_key=g_secret_key)
market_trade = market_client.get_market_trade('eth3lusdt')
for i in market_trade:
market_trade_1 = i.__dict__
# order_min = round(5.5 / market_trade_1['price'], 4)
# 最小订单数量为 5/价格
self.price_1 = market_trade_1['price']
return self.price_1
def huobi_buy(self):
if self.list_new[1] == 'usdt' :
price_1 = huobi_Strategy().market_data()
amount_1 = huobi_Strategy().amount_result(self.list_new[2] * price_1)
self.list_new.append('buy')
self.list_new.append('eth3l')
self.list_new.append(amount_1)
self.list_new.append(price_1)
self.list_new.append('id')
f = pd.DataFrame(self.list_new).T
f.columns = self.columns
date = pd.concat([self.date,f])
date.to_csv('交易模拟测试.csv',encoding='utf-8',index=False)
def huobi_sell(self):
if self.list_new[1] == 'eth3lusdt' :
price_1 = huobi_Strategy().market_data()
amount_1 = huobi_Strategy().amount_result(self.list_new[2] / price_1)
self.list_new.append('sell')
self.list_new.append('usdt')
self.list_new.append(amount_1)
self.list_new.append(price_1)
self.list_new.append('id')
# csv最后增加一行
f = pd.DataFrame(self.list_new).T
f.columns = self.columns
date = pd.concat([self.date, f])
date.to_csv('交易模拟测试.csv', encoding='utf-8', index=False)
def next(self):
up = 80
down = 20
self.order = 'keep'
if self.rvi.rvi[0] > up:
if self.rvi.rvi[-1] < up and self.rvi.rvi[-2] < up:
self.order = 'buy'
elif self.rvi.rvi[0] < down:
if self.rvi.rvi[-1] > down and self.rvi.rvi[-2] > down:
self.order = 'sell'
return
def stop(self):
if self.order == 'buy':
huobi_Strategy().huobi_buy()
elif self.order == 'sell':
huobi_Strategy().huobi_sell()
def get_data():
# 取当前及30min前的烛台数据
market_client = MarketClient()
data_ethusdt = market_client.get_candlestick('ethusdt', CandlestickInterval.MIN1, 100)
id_1 = []
high_1 = []
low_1 = []
open_1 = []
close_1 = []
vol_1 = []
for i in range(len(data_ethusdt) - 1, -1, -1):
id_1 = data_ethusdt[i].id
high_1.append(data_ethusdt[i].high)
low_1.append(data_ethusdt[i].low)
open_1.append(data_ethusdt[i].open)
close_1.append(data_ethusdt[i].close)
# amount = data_ethusdt[i].amount
# count = data_ethusdt[i].count
vol_1.append(data_ethusdt[i].vol)
datas = pd.DataFrame({'datetime': id_1, 'open': open_1,
'high': high_1, 'low': low_1,
'close': close_1, 'volume': vol_1})
datas['datetime'] = pd.to_datetime(datas['datetime'], unit='s')
datas.set_index('datetime', drop=True, append=False)
return datas
'''
程序结束
'''
# 程序执行：
if __name__ == '__main__':
while True :
time_start = time.perf_counter()
logger.info('-'*5 + 'start' + '-'*5)
# 策略加载
df = get_data()
cerebro = bt.Cerebro()
data = bt.feeds.PandasData(dataname=df,datetime='datetime',
open='open',high='high',low='low',
close='close',volume='volume',openinterest=-1)
cerebro.adddata(data)
cerebro.addstrategy(huobi_Strategy)
result = cerebro.run()
# 统计运行时间
time_used = time.perf_counter() - time_start
time.sleep(60)