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

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

在本篇博文中，我们将获取50EFT期权的日行情数据和50ETF的日行情数据，作为环境的状态数据，可以在强化学习环境SopEnv中逐日显示出来。

数据集对象定义

我们定义50ETF日行情数据集类D50etfDataset，在其中统一管理50ETF行情和50ETF期权行情数据以及Greeks信息、VIX恐慌指数等信息。D50etfDataset是PyTorch中Dataset的子类，该类中有两个我们必须提供实现代码的方法，分别为__len__获取数据集中样本数量，__getitem__获取指定索引的样本数据，大家看到这两个方法都是双下划线开头，表明我们通常不使用这两个方法，而是通过后面要介绍的DataLoader来获取数据集中的数据。我们这里先给出一个这个类的实现框架：

import numpy as np
import torch
import torch.utils.data.dataset as Dataset

class D50etfDataset(Dataset.Dataset):
    def __init__(self):
        self.X, self.y = self._load_dataset()

    def __len__(self):
        return self.X.shape[0]

    def __getitem__(self, index):
        return self.X[index], self.y[index]

    def _load_dataset(self):
        X = np.array([
            [1.1, 1.2, 1.3, 1.4, 1.5],
            [2.1, 2.2, 2.3, 2.4, 2.5],
            [3.1, 3.2, 3.3, 3.4, 3.5],
            [4.1, 4.2, 4.3, 4.4, 4.5],
            [5.1, 5.2, 5.3, 5.4, 5.5],
            [6.1, 6.2, 6.3, 6.4, 6.5],
            [7.1, 7.2, 7.3, 7.4, 7.5],
            [8.1, 8.2, 8.3, 8.4, 8.5],
            [9.1, 9.2, 9.3, 9.4, 9.5]
        ])
        y = np.array([1, 1, 1, 0, 0, 0, 1, 1, 1])
        return torch.from_numpy(X), torch.from_numpy(y)

在这里我们在_load_dataset方法中，本来是要从DataSource类中从取出日行情数据来进行初始化，但是这里用写死的简单数据来进行试验。
我们编写一个单元测试用例，来测试一下这个数据集类：

import unittest
import torch.utils.data.dataloader as DataLoader
from apps.sop.d_50etf_dataset import D50etfDataset

class TD50etfDataset(unittest.TestCase):
    @classmethod
    def setUp(cls):
        pass

    @classmethod
    def tearDown(cls):
        pass

    def test_getitem(self):
        ds = D50etfDataset()
        dataloader = DataLoader.DataLoader(ds, batch_size= 2, 
                    shuffle = True, num_workers= 4)
        for idx, (X, y) in enumerate(dataloader):
            print('0: 1 => 2; 3;'.format(idx, X, y, type(y)))
        print('数量：0;'.format(ds.__len__()))
        X, y = ds.__getitem__(3)
        print('样本：0 => 1;'.format(X, y))

运行结果如下所示：

test_getitem (uts.apps.sop.t_d_50etf_dataset.TD50etfDataset) ... 0: tensor([[9.1000, 9.2000, 9.3000, 9.4000, 9.5000],
        [3.1000, 3.2000, 3.3000, 3.4000, 3.5000]], dtype=torch.float64) => tensor([1, 1], dtype=torch.int32); <class 'torch.Tensor'>;
1: tensor([[5.1000, 5.2000, 5.3000, 5.4000, 5.5000],
        [4.1000, 4.2000, 4.3000, 4.4000, 4.5000]], dtype=torch.float64) => tensor([0, 0], dtype=torch.int32); <class 'torch.Tensor'>;
2: tensor([[7.1000, 7.2000, 7.3000, 7.4000, 7.5000],
        [1.1000, 1.2000, 1.3000, 1.4000, 1.5000]], dtype=torch.float64) => tensor([1, 1], dtype=torch.int32); <class 'torch.Tensor'>;
3: tensor([[8.1000, 8.2000, 8.3000, 8.4000, 8.5000],
        [6.1000, 6.2000, 6.3000, 6.4000, 6.5000]], dtype=torch.float64) => tensor([1, 0], dtype=torch.int32); <class 'torch.Tensor'>;
4: tensor([[2.1000, 2.2000, 2.3000, 2.4000, 2.5000]], dtype=torch.float64) => tensor([1], dtype=torch.int32); <class 'torch.Tensor'>;
数量：9;
样本：tensor([4.1000, 4.2000, 4.3000, 4.4000, 4.5000], dtype=torch.float64) => 0;
ok

----------------------------------------------------------------------
Ran 1 test in 1.436s

OK

获取50ETF期权行情数据

我们要交易50ETF期权，就需要获取50ETF日行情数据，我们定义D50etfOptionDataSource类：

import numpy as np
import akshare as ak

class Sh50etfOptionDataSource(object):
    CALL_OPTION_IDX = 0
    PUT_OPTION_IDX = 1
    CALL_OPTION = 101 # 认购期权
    PUT_OPTION = 102 # 认沽期权

    def __init__(self):
        self.refl = ''
        self.symbol = '50ETF'
        self.underlying = '510050'

    def get_data(self):
        print('获取50ETF期权日行情数据')
        option_dict = 
        expire_months = self.get_expire_months()
        option_codes = self.get_option_codes(expire_months[1])
        dates_set = set()
        for option_code in option_codes[Sh50etfOptionDataSource.\\
                        CALL_OPTION_IDX]:
            option_dict[option_code] = self.get_option_daily_quotation(
                option_code, Sh50etfOptionDataSource.CALL_OPTION
            )
        for option_code in option_codes[Sh50etfOptionDataSource.\\
                        PUT_OPTION_IDX]:
            option_dict[option_code] = self.get_option_daily_quotation(
                option_code, Sh50etfOptionDataSource.PUT_OPTION
            )
        return option_dict

    def get_expire_months(self):
        ''' 获取合约到期月份 '''
        return ak.option_sina_sse_list(
                    symbol=self.symbol, exchange="null")

    def get_option_codes(self, trade_date):
        '''
        获取指定月份的期权合约列表
        '''
        return ak.option_sina_sse_codes(trade_date=trade_date,
                     underlying=self.underlying)

    def get_option_daily_quotation(self, option_code, option_type):
        df = ak.option_sina_sse_daily(code=option_code)
        X = []
        dates = df['日期']
        opens = df['开盘']
        highs = df['最高']
        lows = df['最低']
        closes = df['收盘']
        volumes = df['成交']
        for i in range(len(dates)):
            if Sh50etfOptionDataSource.CALL_OPTION == option_type:
                X.append([
                    dates[i], 0.0, 0.0, 0.0,
                    opens[i], highs[i], 
                    lows[i], closes[i], volumes[i]
                ])
            elif Sh50etfOptionDataSource.CALL_OPTION == option_type:
                X.append([
                    dates[i], 1.0, 0.0, 0.0,
                    opens[i], highs[i], 
                    lows[i], closes[i], volumes[i]
                ])
        return np.array(X)

• 第17行：定义一个字典用来存50ETF行情数据，键值为合约编号，值为numpy的数组，每一行代表该天所有合约的行情数据；
• 第18行：获取合约到期月份列表，例如在8月调用时，会返回8月、9月、12月、次年3月；
• 第19行：由于次月到期的期权合约交易最活跃，因此我们只获取次月到期的期权合约编号列表；
• 第20行：我们的主循环是按日期进行循环的，我们将所有日期加入到date_set中，然后进行排序，作为系统日历的日期；
• 第21~25行：循环认购期权合约编号，以其为键，值为每一行的期权合约行情数据，其中第1列为日期，第2到4列为期权类型，目前0,0,0代表认购期权，1,0,0代表认沽期权，剩余列分别为：开盘、最高、最低、收盘、交易量数据；
• 第26~30行：同理处理认沽期权；
  测试用例如下所示：

import unittest
from apps.sop.sh50etf_option_data_source import Sh50etfOptionDataSource

class TSh50etfOptionDataSource(unittest.TestCase):
    @classmethod
    def setUp(cls):
        pass

    @classmethod
    def tearDown(cls):
        pass

    def test_get_expire_months(self):
        ds = Sh50etfOptionDataSource()
        expire_months = ds.get_expire_months()
        print(expire_months)

    def test_get_option_codes(self):
        ds = Sh50etfOptionDataSource()
        trade_date = '202009'
        option_contracts = ds.get_option_codes(trade_date)
        print(option_contracts)

    def test_get_option_daily_quotation(self):
        ds = Sh50etfOptionDataSource()
        option_code = '10002423'
        X = ds.get_option_daily_quotation(option_code)
        print('X: 0;'.format(X.shape))
        print(X)

    def test_get_data(self):
        ds = Sh50etfOptionDataSource()
        option_dict = ds.get_data()
        for key in option_dict.keys():
            ocs = option_dict[key]
            print(ocs)
            break

下面我们将所获取到数据转化为数据集形式，同时从其中找出日期列表，我们来定义Sh50etfDataset类：

class Sh50etfDataset(Dataset.Dataset):
    def __init__(self):
        self.X, self.y, self.r = self._load_dataset()

    def __len__(self):
        return self.X.shape[0]

    def __getitem__(self, index):
        return self.X[index], self.y[index], self.r[index]

    def _load_dataset(self):
        d_50etf = Sh50etfOptionDataSource()
        option_dict = d_50etf.get_data()
        # 获取日期列表
        date_set = set()
        self.key_list = []
        for key in option_dict.keys():
            self.key_list.append(key)
            for oc in option_dict[key]:
                date_set.add(oc[0])
        self.dates = list(date_set)
        list.sort(self.dates, reverse=False)
        list.sort(self.key_list, reverse=False)
        raw_X = []
        for idx in range(len(self.dates)):
            date_row = []
            for key in self.key_list:
                oc = option_dict[key]
                if len(oc) > idx:
                    date_row += [float(oc[idx][1]), float(oc[idx][2]), 
                                float(oc[idx][3]), float(oc[idx][4]), 
                                float(oc[idx][5]), float(oc[idx][5]),
                                float(oc[idx][6]), float(oc[idx][7])]
                else:
                    date_row += [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
            raw_X.append(date_row)
        X = np.array(raw_X, dtype=np.float32)
        y = np.zeros((len(self.dates),))
        r = np.zeros((len(self.dates),))
        return torch.from_numpy(X), torch.from_numpy(y), torch.from_numpy(r)

• 第2~9行：继承自pytorch的Dataset基类，并提供了__len__和__getitem__方法实现；
• 第11行：定义调用Sh50etfOptionDataSource和Sh50etfIndexDataSource来获取数据，形成可以用于强化学习的数据集。我们在这里先介绍调用Sh50etfOptionDataSource类来获取期权数据；
• 第12、13行：调用Sh50etfOptionDataSource来获取如下格式的期权日行情数据：

期权合约编号1: [
	'2020-06-01', 0.0, 0.0, 0.0, 1.1, 2.1, 3.1, 4.1, 888,
	'2020-06-02', 0.0, 0.0,0.0, 1.1. 2.1, 3.1, 4.1, 888,
	...........
],
......

其为一个字典，键为期权合约编号，值为其日行情列表，每一行为一个日行情数据，其中第1列为日期，第2至4列代表是认购还是认沽合约，后面列依次为开盘、最高、最低、收盘、交易量，此处为原始数据，并没有进行归一化；

• 第15~23行：求出日期列表和合约列表，都进行从小到大排序；
• 第25~36行：将当天的所有期权合约行情数据，按照合约编号从小到大依次排列，作为一个数据集的样本；
• 第37行：将样本设计矩阵（Design Matrix）变为numpy数组；
• 第38行：y为监督学习的标签，在强化学习中代表所采取的行动；
• 第39行：对于强化学习，r代表在上一时间点采取行动后，环境返回给Agent的奖励信号；
• 第40行：以Tensor对象形式返回样本集、标签集（行动集）和奖励集；

测试程序如下所示：

class TSh50etfDataset(unittest.TestCase):
    @classmethod
    def setUp(cls):
        pass

    @classmethod
    def tearDown(cls):
        pass

    def test_getitem(self):
        ds = Sh50etfDataset()
        dataloader = DataLoader.DataLoader(ds, batch_size= 2, 
                    shuffle = True, num_workers= 4)
        for idx, (X, y) in enumerate(dataloader):
            print('0: 1 => 2; 3;'.format(idx, X, y, type(y)))
        print('数量：0;'.format(ds.__len__()))
        X, y = ds.__getitem__(3)
        print('样本：0 => 1;'.format(X, y))

    def test__load_dataset(self):
        ds = Sh50etfDataset()
        print('X: 0;'.format(ds.X.shape))
        print('y: 0;'.format(ds.y.shape))
# 运行命令
# python -m unittest uts.apps.sop.t_sh50etf_dataset.TSh50etfDataset.test__load_data -v

运行结果如下所示：

test__load_dataset (uts.apps.sop.t_sh50etf_dataset.TSh50etfDataset) ... 获取50ETF期权日行情数据
X: torch.Size([141, 368]);
y: torch.Size([141]);
ok

这表示我们共有141个日期，每个日期共有368个期权合约的行情数据。

更新环境主循环

我们现在终于有真实的行情数据，我们现在来修改一下主循环的逻辑，依次显示各个交易日，如下所示：

class SopEnv(gym.Env):
    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 = 
            obs, reward, done, info = self.step(action)
            X = obs['X'].cpu().numpy()
            y = obs['y'].cpu().numpy()
            r = obs['r'].cpu().numpy()
            print('    X:0; y:1; r:2'.format(X.shape, 
                        y, r
                        ))
            self.tick += 1

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

运行程序可以得到如下结果：

重置环境到初始状态
2020-01-23:
    X:(368,); y:0.0; r:0.0
2020-02-03:
    X:(368,); y:0.0; r:0.0

我们目前获取了期权合约的行情数据，在实际应用中，我们需要获取期权合约的Black-Scholes估计价格、恐慌指数VIX和希腊字母Greeks信息。由于Black-Scholes和VIX需要额外计算，但是可以直接获取到Greeks信息，因此我们可以把Greeks加入到每个合约的行情数据后面。这个可以作为系统未来的一个可以扩充的功能点，我们在这里就先不实现了。
在下一篇博文中，我们将获取50ETF期权的标的物50ETF指数日行情数据，并且将其加入到合约日行情数据中。