易经量化交易系统之回测系统1

Posted 2022-12-05 最老程序员闫涛

我们在这里向大家介绍如何从零开始，实现一个适合于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，关于这个函数，我们将在下面章节进行介绍。