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

65 阅读 0 评论 43 点赞

我是靠谱客的博主着急犀牛，最近开发中收集的这篇文章主要介绍相对活力指数指标（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)