有些量化教程让人摸不着头脑。他们一上来就教人怎么写策略,但你连回测都做不到。即使你有一个很棒的策略,你敢让他直接跑吗?因此,本期主要解决vnpy量化回测怎么做的问题。我们的目的很简单,无视策略,让回测跑起来就算成功。
一、CCXT获取历史数据
def fetch_ohlcv(self, symbol, timeframe='1m', since=None, limit=None, params={}):使用这种方法可以获得交易所的历史k线,symbol指定交易对,timeframe指定周期,since指定开始时间,limit指定(如果我没记错的话,得到根数) 其实在这里爬一分钟k线就够了,其他周期的k线数据都可以合成。拿到k线数据后,我们先叠加到一个。list里面,存入DataFrame()在变量中,注意我们要去重(同时)和空行(交易所维护),然后把数据写进去。csv在文件中。以下是我自己的代码。 写作比较杂乱,如果想要复制粘贴试一试,首先要确认一下。pandas和ccxt是否安装,第二个交易所需要科学上网,第三个交易main内部路径改变到您需要存储的位置(确保文件夹中没有其他杂乱的东西csv文件),我这是爬行爬行。binanceBTC/USDT一分钟k线数据,如需修改,可通过修改。symbol和timeframe,第一次爬行还是比较费时的,可以扔到服务器里让他慢慢爬,到时候再去拿就行了。
import ccxt
import pandas as pd
import time
import os
pd.set_option('display.max_rows',1000)
pd.set_option("expand_frame_repr",False)
class kline_Crawler(object):
def __init__(self,symbol,data_dir):
self.symbol = symbol
self.data_dir = data_dir
self.exist_df = pd.DataFrame()
self.new_data_df = pd.DataFrame()
self.delete_file_list = []
def get_symbol(self):
'''
将BTC/USDT 切割为 btc_usdt 这样
:return:
'''
symbol_split = self.symbol.lower().split("/")
symbol_name = symbol_split[0] + "_" + symbol_split[1]
return symbol_name
def get_last_crawler_date_time(self):
'''
本方法判断最初文件夹下有无合适的csv文件,若无csv文件那么就指定一个开始时间开始爬取
:return:
'''
symbol_name = self.get_symbol()
flag = 0
for path, dir, files in os.walk(self.data_dir):
for file in files:
if file.endswith(".csv"):
if file[:8] == symbol_name:
flag = 1
exist_file_dir = self.data_dir + file
self.delete_file_list.append(exist_file_dir)
self.exist_df = pd.read_csv(exist_file_dir,parse_dates=["candle_begin_time"])
last_crawl_time = self.exist_df["candle_begin_time"].iloc[-1] - pd.Timedelta(hours=8)
last_ios_date = str(last_crawl_time).replace(" ","T")
last_ios_date += "Z"
return last_ios_date
#如果没有的话怎么办 此处指定无文件时候爬取k线的开始时间
if flag == 0:
start_time = pd.to_datetime('2017/7/1 8:00', format='%Y/%m/%d %H:%M:%S') - pd.Timedelta(hours=8)
start_time = str(start_time).replace(" ", "T")
start_time += "Z"
return start_time
def crawling(self, exchange, since, end):
'''
爬取方法
主要通过fetch_ohlcv来进行
:param exchange:
:param since:
:param end:
:return:
'''
all_kilne_data = []
while since < end:
count = 0
if count < 3:
try:
last_time = pd.to_datetime(since,unit="ms") + pd.Timedelta(hours=8)
print(f"正在读取{self.symbol}从{last_time}开始的数据")
kline_data = exchange.fetch_ohlcv(self.symbol,since=since,timeframe="1m")
time.sleep(0.5)
except:
count+=1
print(f"*******从交易所获取数据失败,2秒后重试第{count}次*******")
time.sleep(2)
continue
else:
if len(kline_data):
since = kline_data[len(kline_data) - 1][0]
all_kilne_data += kline_data
else:
break
elif count >= 3:
print(f"获取{self.symbol}数据失败次数过多,请手工重试")
exit()
all_kilne_data_df = pd.DataFrame(all_kilne_data,dtype="float")
all_kilne_data_df.columns = ['candle_begin_time', 'open', 'high', 'low', 'close', 'volume']
all_kilne_data_df['candle_begin_time'] = pd.to_datetime(all_kilne_data_df['candle_begin_time'],unit='ms')+pd.Timedelta(hours=8)
all_kilne_data_df.drop_duplicates(subset=['candle_begin_time'], inplace=True)
return all_kilne_data_df
def delete_old_files(self):
'''
删去旧文件
:return:
'''
if len(self.delete_file_list) >= 1:
for file in self.delete_file_list:
try:
os.remove(file)
print(f"成功删除旧{file}\n")
except:
print(f"删除{file}失败\n")
def run(self):
'''
修改这一行可以指定交易所爬取 例如爬取bitmex的为 exchange = ccxt.bitmex()
但是要注意每个交易所的symbol 或者 最早k线历史数据不一样
'''
exchange = ccxt.binance()
exchange.load_markets()
last_ios_date = self.get_last_crawler_date_time()
print(last_ios_date)
start_date = pd.to_datetime(exchange.parse8601(last_ios_date),unit="ms") + pd.Timedelta(hours=8)
print(start_date)
print(f"{self.symbol}最近一次爬取的数据截止如期是{start_date}")
since = exchange.parse8601(last_ios_date)
end = exchange.milliseconds() - 60 * 1000
self.new_data_df = self.crawling(exchange, since, end)
final_df = self.exist_df.append(self.new_data_df,sort=False)
final_df.drop_duplicates(subset=['candle_begin_time'],inplace=True)
final_df.reset_index(drop=True,inplace=True)
latest_crawl_date = self.new_data_df["candle_begin_time"].iloc[-1].strftime("%Y%m%d%H%M")
symbol_for_filename = self.get_symbol()
final_df.to_csv(
f"{self.data_dir}{symbol_for_filename}_1m_{latest_crawl_date}.csv",index=False
)
print(f"{self.symbol}数据更新完成,已存入CSV文件")
self.delete_old_files()
if __name__ == "__main__":
#初始化对象要传2个参数,一个是要爬取的symbol,一个是你的指定放csv文件的路径
Crawler = kline_Crawler('BTC/USDT',"C:\\python\\123\\")
Crawler.run()
以上便是小编对vnpy量化回测的相关介绍啦!对于投资者来说好的投资工具无疑是交易的垫脚石,那么更多股票实盘交易工具信息也可以通过https://gitee.com/metatradeapi
查询,有不明白的地方也可通过下方名片找小编交流!