我们在这里向大家介绍如何从零开始，实现一个适合于A股市场的回测系统。在这里我们以A股日K线数据为例，实际上可以比较方便的扩展为分级的数据源。
Tick数据类
我们首先定义一个Tick数据的基类，这个类维护所有金融市场标的的Tick数据具有共性的内容：
class TickData(object):
def __init__(self, symbol, timestamp):
'''
symbol 股票代码
timestamp 时间点
'''
self.symbol = symbol
self.timestamp = timestamp
接下来我们定义A股市场日K线TickData类：
class AsdkTickData(TickData):
def __init__(self, symbol, timestamp,
open_price=0.0, high_price=0.0, low_price=0.0,
close_price=0.0, total_volume=0.0):
'''
symbol 股票代码
timestamp 时间点
open_price 开盘价
hight_price 最高价
low_price 最低价
close_price 收盘价
total_volume 成交量
'''
super(AsdkTickData, self).__init__(symbol, timestamp)
self.open_price = open_price
self.high_price = high_price
self.low_price = low_price
self.close_price = close_price
self.total_volume = total_volume
其继承了TickData基类。
市场行情类
下面我们来定义市场行情类，我们在其中保存市场行情信息。我们希望我们的系统可以处理各种市场行精，所以我们先讲一下通用的方法。但是我们要开发的是一个简单的回测系统，所以在初期版本中，我们尽量将问题简化，等核心功能全部实现之后，我们再添加这些可扩展功能比较合理。
我们先来定义一个金融分析系统的配置类：
class FasConfig(object):
TDT_ASDK = 'asdk' # A股日K线数据
def __init__(self):
self.name = 'fas.bktr.FasConfig'
def new_tick_data(self, tick_data_type, symbol, timestamp, **args):
if tick_data_type == FasConfig.TDT_ASDK:
return AsdkTickData(symbol, timestamp,
open_price = args['open_price'],
high_price = args['high_price'],
low_price = args['low_price'],
close_price = args['close_price'],
total_volume = args['total_volume']
)
return None
fas_config = FasConfig()
我们在其中定义了FasConfig类和一个实例，在类中我们定义了TickData类型的常量，目前我们只研究A股K线行情，因此只有一个常量定义，后面随着系统越来越完善，会定义更多的常量。
我们在new_tick_data方法中，生成一个新的TickData子类实例。tick_data_type指定子类类型，symbol是金融标的代码，timestamp是时间点，**args是以字典形式传过来的可变数量参数。由上面的代码可以看到，这个方法最终会生成一个AsdkTickData类实例。
下面我们来看市场数据类：
class MarketData(object):
def __init__(self):
self.name = 'fas.bktr.MarketData'
self.__tick_datas = {}
def add_close_price(self, symbol, timestamp, close_price):
if symbol not in self.__tick_datas:
tick_data = fas_config.new_tick_data(
fas_config.TDT_ASDK, symbol, timestamp,
open_price=1.0, high_price=2.0, low_price=3.0,
close_price=4.0, total_volume=5.0
)
print('tick_data: {0};'.format(tick_data.close_price))
在这个类中，我们定义私有属性__tick_datas来保存行情数据，其是一个字典，键值为标的代码。
在add_close_price方法中，如果标的代码不在行情字典中，则生成一个新的对应的TickData子类实例。请大家注意，前面3个参数分别指定的TickData的子类，标的代码和时间点，后面的参数会形成**args参数表示的字典。这样就可以生成任意的行情数据类了。
但是我们为了简化当前问题，先不考虑系统的可扩展性，我们只处理A股日K线数据，所以MarketData类定义如下所示：
class MarketData(object):
def __init__(self):
self.name = 'fas.bktr.MarketData'
self.__tick_datas = {}
def get_tick_data(self, symbol, timestamp):
if symbol not in self.__tick_datas:
return None
else:
return self.__tick_datas[symbol]
def set_tick_data(self, symbol, tick_data):
self.__tick_datas[symbol] = tick_data
定义市场数据源
下面我们来看从外部数据源获取行情数据，我们使用akshare库来获取行情数据：
# pip install akshare
class MarketDataSource(object):
def __init__(self):
self.name = 'fas.bktr.MarketDataSource'
self.event_tick = None
self.symbol = 'sh600582'
self.market_data = MarketData()
def start_market_simulation(self):
datas = ak.stock_zh_a_daily(symbol=self.symbol, adjust='hfq')
for time, row in datas.iterrows():
tick_data = AsdkTickData(self.symbol,
time, open=row['open'], high=row['high'],
low=row['low'], close=row['close'],
outstanding_share=row['outstanding_share'],
turn_over=row['turn_over']
)
self.market_data.set_tick_data(self.symbol, tick_data)
if self.event_tick is not None:
self.event_tick(self.market_data)
第5行：self.event_tick是一个行情tick数据的响应函数，每个tick均会被调用；
第7行：用self.market_data来保存当前tick的数据；
第10行：通过akshare获取指定A股市场股票日K线数据；
第11~20行：遍历所有日K线数据，对于一个时间点，生成AsdkTickData类实例，赋给self.market_data属性，然后调用事件响应函数self.event_tick；
通过上面的程序，我们可以看到，当调用start_market_simulation之后，就可以依次处理日K线数据，从而完成整个回测过程了。只是这里我们并没有定义事件响应函数self.event_tick，关于这个函数，我们将在下面章节进行介绍。
