本文说明设计回测系统时，如何实现行情数据处理部分，包括接口基类、历史和实时tick 示例。
from __future__ import print_function
from abc import ABCMeta, abstractmethod
import datetime
import threading
import os, os.path
import numpy as np
import pandas as pd
import akshare as ak
#from event import MarketEvent
1 数据抽象基类
# 定于数据处理的抽象基类
class DataHandler(object):
# ABCMeta 是 Python 的一个内置元类（metaclass），用于实现抽象基类（Abstract Base Classes, ABCs）。
#抽象基类是一种特殊的类，它不能被直接实例化，而是用来定义其他类应该具有的共同接口。
__metaclass__ = ABCMeta
@abstractmethod
def get_latest_bar(self, symbol):
"""
Returns the last bar updated.
"""
raise NotImplementedError("Should implement get_latest_bar()")
@abstractmethod
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars updated.
"""
raise NotImplementedError("Should implement get_latest_bars()")
@abstractmethod
def get_latest_bar_datetime(self, symbol):
"""
Returns a Python datetime object for the last bar. """
raise NotImplementedError("Should implement get_latest_bar_datetime()")
@abstractmethod
def get_latest_bar_value(self, symbol, val_type):
"""
Returns one of the Open, High, Low, Close, Volume or OI from the last bar.
"""
raise NotImplementedError("Should implement get_latest_bar_value()")
@abstractmethod
def get_latest_bars_values(self, symbol, val_type, N=1):
"""
Returns the last N bar values from the
latest_symbol list, or N-k if less available. """
raise NotImplementedError("Should implement get_latest_bars_values()")
@abstractmethod
def update_bars(self):
"""
Pushes the latest bars to the bars_queue for each symbol
in a tuple OHLCV format: (datetime, open, high, low,
close, volume).
"""
raise NotImplementedError("Should implement update_bars()")
2 从csv文件读取行情
# 继承抽象基类，从英为下载历史行情csv文件，加载到对象
# 比如平安银行：https://cn.investing.com/equities/ping-an-bank-a-historical-data
#
class HistoricCSVDataHandler(DataHandler):
"""
HistoricCSVDataHandler is designed to read CSV files
"""
def __init__(self, events, csv_dir, symbol_list):
self.events = events
self.csv_dir = csv_dir
self.symbol_list = symbol_list
self.symbol_data = {}
self.latest_symbol_data = {}
self.continue_backtest = True
self._open_convert_csv_files()
def _open_convert_csv_files(self):
"""
Opens the CSV files from the data directory, converting
them into pandas DataFrames within a symbol dictionary.
"""
comb_index = None
for s in self.symbol_list:
# Load the CSV file with no header information, indexed on date
quote_file = os.path.join(self.csv_dir, '%s.csv' %s)
print("quote_file:", quote_file)
# filepath：要读取的 CSV 文件的路径或类似文件的对象。
# sep 或 delimiter：用于分隔字段的字符，默认为 ','。
# header：用作列名的行号，默认为 0（第一行）。如果设置为 None，则不会使用任何行作为列名，列名将被命名为 0, 1, 2, ...。
# index_col：用作行索引的列编号或列名。
# usecols：要读取的列列表。
# dtype：指定列的数据类型。
# parse_dates：要解析为日期的列名列表。
# self.symbol_data[s] = pd.read_csv(quote_file,
# header=0, index_col=0, parse_dates=True, skiprows = 1,
# names=[
# 'datetime', 'close', 'open',
# 'high', 'low', 'volume'])
self.symbol_data[s] = pd.read_csv(quote_file,header = 0,index_col=0, skiprows = 1,
names=['close', 'open','high', 'low', 'volume','rise'])
#print(">>>>>>")
# Combine the index to pad forward values
if comb_index is None:
comb_index = self.symbol_data[s].index
else:
comb_index.union(self.symbol_data[s].index)
# Set the latest symbol_data to None
#print(self.symbol_data[s])
self.latest_symbol_data[s] = pd.DataFrame()
# Reindex the dataframes
for s in self.symbol_list:
# reindex是一个Pandas方法，用于调整DataFrame或Series的索引。
# index=comb_index表示将DataFrame或Series的索引重新设置为comb_index。
# method='pad'是一个填充方法，表示使用前一个有效值来填充缺失值。
# iterrows 迭代处理调整后的每一行数据
# [::-1] 反转索引
self.symbol_data[s] = self.symbol_data[s].reindex(index=comb_index[::-1])
#print(self.symbol_data[s])
def _get_new_bar(self, symbol):
"""
Returns the latest bar from the data feed.
"""
if len(self.symbol_data[symbol]) > 0:
return self.symbol_data[symbol]
return None
def get_latest_bar(self, symbol):
"""
Returns the last bar from the latest_symbol list.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
#print(type(bars_list))
#print(bars_list)
return bars_list[-1:]
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars from the latest_symbol list,
or N-k if less available.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list[-N:]
def get_latest_bar_datetime(self, symbol):
"""
Returns a Python datetime object for the last bar.
"""
try:
bars_list = self.latest_symbol_data[symbol].index
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list[-1]
def get_latest_bar_value(self, symbol, val_type):
"""
Returns one of the Open, High, Low, Close, Volume or OI
values from the pandas Bar series object.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return (bars_list.iloc[-1][val_type])
def get_latest_bars_values(self, symbol, val_type, N=1):
"""
Returns the last N bar values from the latest_symbol list, or N-k if less available.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list.iloc[-N:][val_type].to_numpy()
def update_bars(self):
"""
Pushes the latest bar to the latest_symbol_data structure for all symbols in the symbol list.
"""
for s in self.symbol_list:
try:
bar = self._get_new_bar(s)
except StopIteration:
self.continue_backtest = False
else:
if bar is not None:
# 添加新增行情到行情数据尾部
self.latest_symbol_data[s] = pd.concat([self.latest_symbol_data[s], bar], axis=0)
# 发送行情通知事件
#self.events.put(MarketEvent())
hisData = HistoricCSVDataHandler(None, "", ["000001","0700"])
hisData.update_bars()
print("获取最新的一条数据--->")
print(hisData.get_latest_bar("0700"))
print("获取最新的3条数据--->")
print(hisData.get_latest_bars("0700", 3))
print("获取最新日期--->")
print(hisData.get_latest_bar_datetime("0700")) # out : 2024-2-23
print("获取最新的收盘价--->")
print(hisData.get_latest_bar_value("0700", "close")) # out : 290.8
print("获取最新的收盘价集合--->")
close_np = hisData.get_latest_bars_values("0700", "close" , 3) # out : [288.8 291.4 290.8]
print(type(close_np))
print(close_np)
#print(hisData.get_latest_bar("0700"))
quote_file: ./000001.csv
quote_file: ./0700.csv
获取最新的一条数据--->
close open high low volume rise
2024-2-23 290.8 290.0 296.4 287.0 13.23M -0.21%
获取最新的3条数据--->
close open high low volume rise
2024-2-21 288.8 280.8 294.6 280.8 23.95M 1.62%
2024-2-22 291.4 288.2 291.4 286.8 12.44M 0.90%
2024-2-23 290.8 290.0 296.4 287.0 13.23M -0.21%
获取最新日期--->
2024-2-23
获取最新的收盘价--->
290.8
获取最新的收盘价集合--->
<class 'numpy.ndarray'>
[288.8 291.4 290.8]
3 从akshare 获取历史数据
# 继承抽象基类
#
class HistoricCSVDataHandler(DataHandler):
"""
HistoricCSVDataHandler is designed to read CSV files
"""
def __init__(self, events, market, symbol_list):
self.events = events
self.market = market
self.symbol_list = symbol_list
self.symbol_data = {}
self.latest_symbol_data = {}
self.continue_backtest = True
self._get_akshare_data()
def _get_akshare_data(self):
"""
Get data from akshare, converting
them into pandas DataFrames within a symbol dictionary.
"""
comb_index = None
for s in self.symbol_list:
self.symbol_data[s] = ak.futures_zh_daily_sina(s)
#print("----")
#print(self.symbol_data[s])
self.latest_symbol_data[s] = pd.DataFrame()
# Reindex the dataframes
for s in self.symbol_list:
comb_index = self.symbol_data[s]['date']
self.symbol_data[s] = self.symbol_data[s].set_index(comb_index)
#print(self.symbol_data[s])
def _get_new_bar(self, symbol):
"""
Returns the latest bar from the data feed.
"""
if len(self.symbol_data[symbol]) > 0:
return self.symbol_data[symbol]
return None
def get_latest_bar(self, symbol):
"""
Returns the last bar from the latest_symbol list.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
#print(type(bars_list))
#print(bars_list)
return bars_list[-1:]
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars from the latest_symbol list,
or N-k if less available.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list[-N:]
def get_latest_bar_datetime(self, symbol):
"""
Returns a Python datetime object for the last bar.
"""
try:
bars_list = self.latest_symbol_data[symbol].index
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list[-1]
def get_latest_bar_value(self, symbol, val_type):
"""
Returns one of the Open, High, Low, Close, Volume or OI
values from the pandas Bar series object.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return (bars_list.iloc[-1][val_type])
def get_latest_bars_values(self, symbol, val_type, N=1):
"""
Returns the last N bar values from the latest_symbol list, or N-k if less available.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list.iloc[-N:][val_type].to_numpy()
def update_bars(self):
"""
Pushes the latest bar to the latest_symbol_data structure for all symbols in the symbol list.
"""
for s in self.symbol_list:
try:
bar = self._get_new_bar(s)
except StopIteration:
self.continue_backtest = False
else:
if bar is not None:
# 添加新增行情到行情数据尾部
self.latest_symbol_data[s] = pd.concat([self.latest_symbol_data[s], bar], axis=0)
# 发送行情通知事件
#self.events.put(MarketEvent())
hisData = HistoricCSVDataHandler(None, "", ["V2403", "P2403", "B2403", "M2403"])
hisData.update_bars()
print("获取最新的一条数据--->")
print(hisData.get_latest_bar("V2403"))
print("获取最新的3条数据--->")
print(hisData.get_latest_bars("V2403", 3))
print("获取最新日期--->")
print(hisData.get_latest_bar_datetime("V2403")) # out : 2024-2-23
print("获取最新的收盘价--->")
print(hisData.get_latest_bar_value("V2403", "close")) # out : 290.8
print("获取最新的收盘价集合--->")
close_np = hisData.get_latest_bars_values("V2403", "close" , 3) # out : [288.8 291.4 290.8]
print(type(close_np))
print(close_np)
#print(hisData.get_latest_bar("0700"))
获取最新的一条数据--->
date open high low close volume hold settle
date
2024-02-29 2024-02-29 5569.0 5606.0 5541.0 5600.0 5094 42171 5580.0
获取最新的3条数据--->
date open high low close volume hold settle
date
2024-02-27 2024-02-27 5537.0 5576.0 5528.0 5573.0 5525 42932 5557.0
2024-02-28 2024-02-28 5578.0 5610.0 5544.0 5560.0 5622 42949 5565.0
2024-02-29 2024-02-29 5569.0 5606.0 5541.0 5600.0 5094 42171 5580.0
获取最新日期--->
2024-02-29
获取最新的收盘价--->
5600.0
获取最新的收盘价集合--->
<class 'numpy.ndarray'>
[5573. 5560. 5600.]
4 模拟从akshare 实时获取TICK
# 继承抽象基类
#
class TickAKShareDataHandler(DataHandler):
"""
HistoricCSVDataHandler is designed to read CSV files
"""
def __init__(self, events, market, symbol_list):
self.events = events
self.market = market
self.symbol_list = symbol_list
self.symbol_data = {}
self.latest_symbol_data = {}
self.continue_backtest = True
self._get_akshare_data()
# 事件处理线程
self._task_thread = threading.Thread(target = self._run)
# 模拟tick 3秒请求一次
self._timerSleep= 3
self._stop = True
def _run(self):
while not self._stop :
time.sleep(self.timerSleep_)
_get_akshare_data(self)
update_bars(self)
def _get_akshare_data(self):
"""
Get data from akshare, converting
them into pandas DataFrames within a symbol dictionary.
"""
comb_index = None
for s in self.symbol_list:
self.symbol_data[s] = ak.futures_zh_spot(s)
#print("----")
#print(self.symbol_data[s])
self.latest_symbol_data[s] = pd.DataFrame()
# Reindex the dataframes
for s in self.symbol_list:
comb_index = self.symbol_data[s]['time']
self.symbol_data[s] = self.symbol_data[s].set_index(comb_index)
#print(self.symbol_data[s])
def _get_new_bar(self, symbol):
"""
Returns the latest bar from the data feed.
"""
if len(self.symbol_data[symbol]) > 0:
return self.symbol_data[symbol]
return None
def start(self):
self._stop = False
self._task_thread.start()
def get_latest_bar(self, symbol):
"""
Returns the last bar from the latest_symbol list.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
#print(type(bars_list))
#print(bars_list)
return bars_list[-1:]
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars from the latest_symbol list,
or N-k if less available.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list[-N:]
def get_latest_bar_datetime(self, symbol):
"""
Returns a Python datetime object for the last bar.
"""
try:
bars_list = self.latest_symbol_data[symbol].index
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list[-1]
def get_latest_bar_value(self, symbol, val_type):
"""
Returns one of the Open, High, Low, Close, Volume or OI
values from the pandas Bar series object.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return (bars_list.iloc[-1][val_type])
def get_latest_bars_values(self, symbol, val_type, N=1):
"""
Returns the last N bar values from the latest_symbol list, or N-k if less available.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list.iloc[-N:][val_type].to_numpy()
def update_bars(self):
"""
Pushes the latest bar to the latest_symbol_data structure for all symbols in the symbol list.
"""
for s in self.symbol_list:
try:
bar = self._get_new_bar(s)
except StopIteration:
self.continue_backtest = False
else:
if bar is not None:
# 添加新增行情到行情数据尾部
self.latest_symbol_data[s] = pd.concat([self.latest_symbol_data[s], bar], axis=0)
# 发送行情通知事件
#self.events.put(MarketEvent())
hisData = TickAKShareDataHandler(None, "", ["V2403", "P2403", "B2403", "M2403"])
hisData.update_bars()
print("获取最新的一条数据--->")
print(hisData.get_latest_bar("V2403"))
print("获取最新的3条数据--->")
print(hisData.get_latest_bars("V2403", 3))
print("获取最新日期--->")
print(hisData.get_latest_bar_datetime("V2403")) # out : 2024-2-23
print("获取最新的收盘价--->")
print(hisData.get_latest_bar_value("V2403", "last_close")) # out : 290.8
print("获取最新的收盘价集合--->")
close_np = hisData.get_latest_bars_values("V2403", "last_close" , 3) # out : [288.8 291.4 290.8]
print(type(close_np))
print(close_np)
#print(hisData.get_latest_bar("0700"))
获取最新的一条数据--->
symbol time open high low current_price bid_price \
time
150230 PVC2403 150230 5569.0 5606.0 5541.0 5600.0 5582.0
ask_price buy_vol sell_vol hold volume avg_price last_close \
time
150230 5606.0 10 2 42171.0 5094 5580.0 5600.0
last_settle_price
time
150230 5565.0
获取最新的3条数据--->
symbol time open high low current_price bid_price \
time
150230 PVC2403 150230 5569.0 5606.0 5541.0 5600.0 5582.0
ask_price buy_vol sell_vol hold volume avg_price last_close \
time
150230 5606.0 10 2 42171.0 5094 5580.0 5600.0
last_settle_price
time
150230 5565.0
获取最新日期--->
150230
获取最新的收盘价--->
5600.0
获取最新的收盘价集合--->
<class 'numpy.ndarray'>
[5600.]
