基于强化学习的期权量化交易回测系统4

Posted 2023-02-21 最老程序员闫涛

获取50ETF指数行情数据

50ETF期权的标的物是50ETF指数，我们可以使用akshare来获取该指数的日行情数据，如下所示：

class Sh50etfIndexDataSource(object):
	......
    def get_daily_data(self, start_date, end_date):
        df = ak.stock_zh_index_daily(symbol="sh510050")
        df1 = df.loc[start_date: end_date]
        print(df1)
        open1 = df1['open'][start_date]
        print('open1: 0;'.format(type(open1), open1))
        val1 = df1.loc[start_date]
        print('df1[2020-06-01]: 0; 1;'.format(type(val1), val1))
        print('open: 0; high: 1;  type:2'.format(val1['open'], val1['high'], type(val1['open'])))

运行结果如下所示：

             open   high    low  close       volume
date
2020-06-01  2.830  2.869  2.828  2.864  360043706.0
2020-06-02  2.859  2.888  2.857  2.879  259134405.0
2020-06-03  2.895  2.919  2.883  2.884  396536323.0
2020-06-04  2.897  2.898  2.873  2.878  190374093.0
2020-06-05  2.880  2.895  2.867  2.892  162544519.0
...           ...    ...    ...    ...          ...
2020-06-17  2.876  2.882  2.863  2.881  219333376.0
2020-06-18  2.872  2.896  2.860  2.895  330840064.0
2020-06-19  2.892  2.938  2.890  2.927  343200099.0
2020-06-22  2.927  2.950  2.914  2.924  295088652.0
2020-06-23  2.916  2.932  2.903  2.930  215215341.0

[17 rows x 5 columns]

由上面的运行结果可以看出，每天的行情为一行，以日期为索引，每天的行情数据包括：开盘、最高、最低、收盘、交易量数据。我们可以通过指定开始日期和结束日期来获取指定时间间隔的数据。
我们当前的任务是将50ETF指数行情数据添加到50ETF期权合约行情数据的最后，这部分工作我们在Sh50etfDataset._load_dataset方法中来实现。

向前看n个时间点

我们当前只能看到当天的行情数据，通常我们希望看连续几个交易的行情变化情况，才能做出交易决策。因此在这一节我们将改造我们的数据集，连续向前看几个时间点，形成一个包含历史行情信息的数据集。具体实现代码见apps.sop.ds.Sh50etfDataset._load_dataset方法，如下所示：

class Sh50etfDataset(Dataset.Dataset):
    def _load_dataset(self):
        ......
        # raw_X为每一行为一天的行情数据
        X_n = [] # 向前5天行情组成一行
        for idx in range(SopConfig.lookback_num -1, len(raw_X)):
            tick_data = []
            for j in range(SopConfig.lookback_num-1, -1, -1):
                tick_data += raw_X[idx - j]
            X_n.append(tick_data)
        X = np.array(X_n, dtype=np.float32)
        y = np.zeros((X.shape[0],))
        r = np.zeros((X.shape[0],))
        return torch.from_numpy(X), torch.from_numpy(y), torch.from_numpy(r)

如上所示，raw_X为每行为一天的行情数据，tick_data代表一个时间点的数据，我们在SopConfig中定义向前看的天数lookback_num，缺省值为5。在上面程序中，我们从第lookback-num时刻开始，包括其本身在内，向前再取lookback_num-1个时间点数据，形成新的一行数据，并将其加入到X_n中。这样每个时间点的tick_data数据，就由lookback_num个时间点的数据组成了。

Agent类实现

在我们的体系架构中，环境的状态obs的行情数据，会发给Agent，由Agent调用适当的策略，产生相应的动作，这里对应的就是股市的订单，由风控模块进行审核，通过后通过Broker类来具体执行，在这过程中会考虑到权利金、保证金、手续费、税费等的计算，尽量真实的模拟股市交易过程。
Agent类定义如下所示：

class SopAgent(object):
    IDX_OPTION = 0
    IDX_ACTION = 1
    IDX_PERCENT = 2

    def __init__(self):
        self.refl = 'apps.sop.Agent'
        #self.reset(env)
        self.action = None

    def reset(self, env):
        if self.action is None:
            self.action = [
                np.zeros((len(env.ds.key_list),)),
                np.zeros((3,)),
                np.zeros((10,))
            ]
        self._reset_action()

    def _reset_action(self):
        self.action[SopAgent.IDX_OPTION][self.action\\
                    [SopAgent.IDX_OPTION] > 0] = 0
        self.action[SopAgent.IDX_ACTION][self.action\\
                    [SopAgent.IDX_ACTION] > 0] = 0
        self.action[SopAgent.IDX_PERCENT][self.action\\
                    [SopAgent.IDX_PERCENT] > 0] = 0


    def choose_action(self, obs, reward):
        '''
        根据环境当前状态选择本时间点的行动，将上一时间点行动的奖励信号
        用于策略学习
        '''
        self._reset_action()
        print('看到：0;\\n奖励：1;'.format(obs, reward))
        option_idx = 2
        action_idx = 1
        percent_idx = 2
        self.action[SopAgent.IDX_OPTION][option_idx] = 1
        self.action[SopAgent.IDX_ACTION][action_idx] = 1
        self.action[SopAgent.IDX_PERCENT][percent_idx] = 1
        return self.action

在这里我们定义action为列表类型，第1个数组为代表期权合约编号的one-hot列表，第2个数组代表买入、持有、卖出的one-hot数组，第3个数组代表买卖时操作资金或仓位的百分比，默认为100%。
我们在环境主循环的每一步中，都需要调用Agent来选择当前的行动，如下所示：

class SopEnv(gym.Env):
    def __init__(self):
        self.refl = ''
        self.tick = 0
        # action为3维数组：1维-是期权合约编号；2维-买入持有卖出；
        # 3维-百分比，缺省为100%
        self.agent = SopAgent()

    def startup(self, args=):
        self.ds = Sh50etfDataset()
        self.reset()
        obs, reward, done, info = self._next_observation(), 0, False, 
        for dt in self.ds.dates:
            print('0: '.format(dt))
            action = self.agent.choose_action(obs, reward)
            obs, reward, done, info = self.step(action)
            X = obs['X'].cpu().numpy()
            y = obs['y'].cpu().numpy()
            r = obs['r'].cpu().numpy()
            self.tick += 1

    def reset(self):
        print('重置环境到初始状态')
        self.agent.reset(self)
        self.tick = 0

    def _next_observation(self):
        X, y, r = self.ds.__getitem__(self.tick)
        return 'X': X, 'y': y, 'r': r

我们在环境重置时调用Agent的初始化方法，在系统主循环中，每一步都调用agent的choose_action方法。