零基础创建自定义gym环境——以股票市场为例

翻译自medium上的一篇文章Create custom gym environments from scratch — A stock market example，作者是adam king

CSDN上已经有一篇翻译了：链接

github代码

【注】本人认为这篇文章具有较大的参考价值，尤其是其中的代码，文章构建了一个简单的量化交易环境。强化学习算法直接调用了stable-baselines，该库实现了标准的强化学习算法，可以直接使用。略去强化学习常见算法实现的细节，对用初学者很是友好，但是初学者还是有必要理解和分清楚每一种强化学习算法的原理。另外，其构建环境继承自gym.Env也具有很大的学习意义。

OpenAI的gym是一个非常优秀的包，能够让用户自定义配置强化学习的agent。它包含了许多预置的环境比如Cart Pole，MountainCar,以及Atari游戏环境。
这些环境非常适合用来学习，但是最终你都需要设置一个智能体来解决特定的问题。为了做到这一点，你需要自定义环境来解决你的特定问题。下文我们会创建一个用于模仿股票交易的自定义市场环境。所有代码见github.
这些环境非常时候初学者直接调用来进行学习，但是最终我们都需要自定义一个智能体（agent）来解决特定的问题，为此我们需要自己创建一个环境。下文我们将会创建一个基于gym的自定义股票交易环境，所有代码见github。
首先，我们需要明白一个环境到底是什么？一个环境包含一个智能体运行和学习所需要的所有功能性组件。创建一个环境的模板代码如下：

import gym
from gym import spacesclass CustomEnv(gym.Env):"""继承自gym的自定义环境类"""metadata = {'render.modes': ['human']}def __init__(self, arg1, arg2, ...):super(CustomEnv, self).__init__()# 定义动作空间和状态空间# They must be gym.spaces objects# Example when using discrete actions:#动作空间为离散值时的例子self.action_space = spaces.Discrete(N_DISCRETE_ACTIONS)# Example for using image as input:#使用图像作为输入时的例子self.observation_space = spaces.Box(low=0, high=255, shape=(HEIGHT, WIDTH, N_CHANNELS), dtype=np.uint8)def step(self, action):# Execute one time step within the environment#智能体每个时间点执行的动作...def reset(self):# Reset the state of the environment to an initial state#将环境重置为初始化状态...def render(self, mode='human', close=False):# Render the environment to the screen#渲染环境...

在上述代码中，我们首先定义了动作空间（智能体在环境中采取动作的所有可能的集合）的类型和维度。与此同时，我们还定义了状态空间，即智能体在环境中所能观测到的环境状态。当调用reset方法之后，环境将被重置为初始状态，调用render方法可以将当前的环境渲染出来。简单点我们可以直接输出环境数据，稍微复杂点可以在屏幕上渲染出一个基于openGL的3D图像。在本例中，我们只是简单输出环境的数据。

股票交易环境

为了展示这一切是如何进行的，我们将创建一个股票交易环境，然后借此训练智能体，让智能体成为股票交易员，让我们开始吧！
我们首先需要明白的是，人类交易者在做出交易或者不交易的决策之前是依据什么？一个交易者最经常做的事情就是观察股价变化的图表，这些图表中列举计算了大量的技术指标供他们参考。也就是说，他们将图表中的数据和自己的经验相结合来判断股价未来的走势。
因此，我们只需要将这些人类行为转化为智能体感知环境的行为就可以了。observation_space 包含我们希望智能体在做一个交易或者不做交易之前所有能够考虑的变量。
在这个例子中，我们希望我们的智能体能够“看到”过去5天的股票数据（开盘价，最高价，最低价，收盘价和每日成交量），以及一些其他的数据例如资金盈余，当前股票持仓状态，当前盈利额等。直白一点说就是，对于每一次交易，我们希望我们的智能体能够考虑导致当前价格变化的因素，以及他们自己的投资组合的状态，以便为下一步行动做出明智的决定。一旦智能体感知到正确的交易时机，他们就需要立即采取行动。在我们的智能体动作空间action_space中，有三种动作：买，卖或者什么都不做。
但是这还不够，我们还需要确定每次买卖股票得数量，利用gym的Box空间，我们既可以创建离散的动作空间（买，卖，不动），也可以创建连续的动作空间（每次操作股票总量的0到100%）。
你会注意到这个数量对于持仓不动的动作来说不是必要的，但我们每次都会提供。智能体一开始并不能区分这一点，但是随着时间的推移，它会明白的。
最后需要考虑的事情是奖励函数，智能体需要知道随着时间的推移，累计盈利是多少。在每个步骤中，我们将把奖励设置为帐户余额乘以到目前为止的时间步骤数的某个分数。（这样做的目的是考虑到长期的收益，随着时间的推移，累计回报会有一定的折扣，也就是强化学习里常见的累计折扣回报）。

这样做的目的是，在初期对其收益进行折扣，这样能够允许智能体在深入优化单一策略之前进行充分的探索。它还将奖励那些长期维持较高收益的策略，而不是那些通过不可持续的策略迅速赚钱的智能体。

数据说明

在此插入csv数据说明，以便理解下文所述。

可以看到数据格式为 Date,Open,High,Low,Close,Volume 即交易日期，开盘价，最高价，最低价，收盘价和发行股份数量。该样例数据为苹果从1998年1月2日到2018年11月16日的数据。

应用

到现在为止，我们已经定义了状态空间，动作空间，奖励函数。是时候应用我们设定的环境了，首先我们在环境中自定义action_sapce和observation_space，环境希望接收pandas 数据类型的参数。

class StockTradingEnvironment(gym.Env):"""A stock trading environment for OpenAI gym"""metadata = {'render.modes': ['human']}def __init__(self, df):super(StockTradingEnv, self).__init__()self.df = df #接受dataframe数据self.reward_range = (0, MAX_ACCOUNT_BALANCE) #奖励函数的范围，0到账户最大盈余# Actions of the format Buy x%, Sell x%, Hold, etc.self.action_space = spaces.Box( #离散的动作空间low=np.array([0, 0]), high=np.array([3, 1]), dtype=np.float16)# Prices contains the OHCL values for the last five prices#状态空间，包含最近5次的OHCL价格self.observation_space = spaces.Box(low=0, high=1, shape=(6, 6), dtype=np.float16)

接下来，我们将重写reset方法，该方法用来重置环境状态到初始状态。在这里我们将设置每个智能体的初始投入资金和开盘价。

def reset(self):# Reset the state of the environment to an initial state#重置环境状态为初始状态self.balance = INITIAL_ACCOUNT_BALANCE #初始投入资金self.net_worth = INITIAL_ACCOUNT_BALANCE #净资产 = 持仓资产+账户余额self.max_net_worth = INITIAL_ACCOUNT_BALANCE #最大净资产，随着交易的进行逐渐更新，最终为整个交易过程中的最大净资产self.shares_held = 0 #持仓数量self.cost_basis = 0 #单只股票持仓成本self.total_shares_sold = 0 #总共卖出股份self.total_sales_value = 0 #总共卖出金额# Set the current step to a random point within the data frameself.current_step = random.randint(0, len(self.df.loc[:, 'Open'].values) - 6)return self._next_observation()

我们在dataframe中设置一个随机点，因为它能在相同的数据中给我们提供更多独特的经验。（数据之间是相互独立的）。下面的_next_observation方法集合了该随机点之后5天的股票数据和智能体的账户信息并且进行归一化。

def _next_observation(self):# Get the data points for the last 5 days and scale to between 0-1#在dataframe 中获取当前交易日后5天的数据并归一化frame = np.array([self.df.loc[self.current_step: self.current_step +5, 'Open'].values / MAX_SHARE_PRICE,self.df.loc[self.current_step: self.current_step +5, 'High'].values / MAX_SHARE_PRICE,self.df.loc[self.current_step: self.current_step +5, 'Low'].values / MAX_SHARE_PRICE,self.df.loc[self.current_step: self.current_step +5, 'Close'].values / MAX_SHARE_PRICE,self.df.loc[self.current_step: self.current_step +5, 'Volume'].values / MAX_NUM_SHARES,])# Append additional data and scale each value to between 0-1obs = np.append(frame, [[self.balance / MAX_ACCOUNT_BALANCE,self.max_net_worth / MAX_ACCOUNT_BALANCE,self.shares_held / MAX_NUM_SHARES,self.cost_basis / MAX_SHARE_PRICE,self.total_shares_sold / MAX_NUM_SHARES,self.total_sales_value / (MAX_NUM_SHARES * MAX_SHARE_PRICE),]], axis=0)return obs

接下来，我们的环境还需要一个step函数，这个函数是智能体在环境中执行的一个步骤。在每个步骤中智能体执行特定的动作（动作又强化学习算法决定），然后计算立即回报并返回环境的下一个状态。

def step(self, action):# Execute one time step within the environment#根据动作action执行一个步骤self._take_action(action)self.current_step += 1 #当前动作数量+1#已经走到头了，重置为0if self.current_step > len(self.df.loc[:, 'Open'].values) - 6:self.current_step = 0#延迟修改器：用来进行计算累计回报的乘数delay_modifier = (self.current_step / MAX_STEPS)reward = self.balance * delay_modifier #相当于强化学习算法里的折扣因子lambdadone = self.net_worth <= 0 #如果净资产小于0则结束obs = self._next_observation() #下一个状态return obs, reward, done, {}

现在，我们的_take_action方法将根据模型的选择执行动作买，卖或者持仓不动。

def _take_action(self, action):# Set the current price to a random price within the time step#计算在当前位置的均价current_price = random.uniform(self.df.loc[self.current_step, "Open"],self.df.loc[self.current_step, "Close"])#action = [action_type,amount] 即动作类型，该动作操作数量，amount应该是一个0到1的小数action_type = action[0]amount = action[1]if action_type < 1:# Buy amount % of balance in sharestotal_possible = self.balance / current_price #先计算剩下的钱最多能买多少股shares_bought = total_possible * amount #买这么多，amount应该是一个比例prev_cost = self.cost_basis * self.shares_held #之前的单只股票的成本价格*持仓数量，即成本？additional_cost = shares_bought * current_price #买下这些股票需要的钱self.balance -= additional_cost #从账户余额里面扣除self.cost_basis = (prev_cost + additional_cost) / (self.shares_held + shares_bought) #这个是计算单只股票的成本价格么？self.shares_held += shares_bought #持股数量增加elif actionType < 2:# Sell amount % of shares heldshares_sold = self.shares_held * amount . self.balance += shares_sold * current_price #账户余额增加self.shares_held -= shares_sold #持股数量减少self.total_shares_sold += shares_sold #总共卖出的股票数量self.total_sales_value += shares_sold * current_price #总共卖出的金额self.netWorth = self.balance + self.shares_held * current_price #净资产=账户余额+当前持仓股票的价值#净资产增加了if self.net_worth > self.max_net_worth:self.max_net_worth = net_worth#没有持仓，则单只成本价格为0if self.shares_held == 0:self.cost_basis = 0

现在唯一剩下的需要做的事情是渲染环境。简单起见，我们只是展示了到目前为止的利润和其他一些有趣的数据。

def render(self, mode='human', close=False):#what's the meaning of mode?# Render the environment to the screenprofit = self.net_worth - INITIAL_ACCOUNT_BALANCE #利润 = 当前的净资产-初始投入资金print(f'Step: {self.current_step}') #当前在哪一步print(f'Balance: {self.balance}') #账户余额print(f'Shares held: {self.shares_held} #当前持仓(Total sold: {self.total_shares_sold})') #总卖出股份数量print(f'Avg cost for held shares: {self.cost_basis} #持仓成本(Total sales value: {self.total_sales_value})') #总卖出金额print(f'Net worth: {self.net_worth} #当前净资产(Max net worth: {self.max_net_worth})') #当前最大净资产print(f'Profit: {profit}') #利润

我们的环境是复杂的，我们现在可以输入一个dataframe实例化一个StockTradingEnv类，然后从stable-baselines中选取一些算法来进行测试。

import gym
import json
import datetime as dt
from stable_baselines.common.policies import MlpPolicy
from stable_baselines.common.vec_env import DummyVecEnv
from stable_baselines import PPO2
from env.StockTradingEnv import StockTradingEnv
import pandas as pd
df = pd.read_csv('./data/AAPL.csv') #读取数据，apple的股票数据
df = df.sort_values('Date') #根据交易日期进行排序
# The algorithms require a vectorized environment to run
env = DummyVecEnv([lambda: StockTradingEnv(df)]) #stable-baselines中的算法需要一个向量化的环境
model = PPO2(MlpPolicy, env, verbose=1) #选取PPO２算法进行训练，verbose=1表示打印训练过程中的详细信息
model.learn(total_timesteps=20000) #迭代2000次进行训练
obs = env.reset() #重置环境
for i in range(2000):
#利用训练好的模型，来进行实际操作action, _states = model.predict(obs)obs, rewards, done, info = env.step(action)env.render()

当然，我们只是出于乐趣设计了这样一个半成品的用于股票自动化交易的自定义gym环境。如果真的想要在股票市场中利用强化学习算法，那么还需要付出努力继续丰富和优化这个环境。

请持续关注下一篇文章，我们将会创建一个简单的，优雅的可视化环境。即Rendering elegant stock trading agents using Matplotlib and Gym。
这篇文章在Medium上是付费内容，这里是CSDN上的一篇翻译：链接
但是该代码使用的是mpl-finance库，而该库已经被废弃不再维护，应该使用新的库mplfinance

记录运行过程

首先从github上下载源码。
使用anaconda包管理器进行虚拟环境的创建，此处省略anaconda的安装过程。电脑系统是windows10,并且无GPU。

创建虚拟环境

conda create --name DeepLearning37 python=3.7

创建一个名为DeepLearning37,指定python版本为3.7的虚拟环境

进入虚拟环境

conda activate DeepLearning37

对于linux macos系统是source activate env_name
如果需要退出当前虚拟环境conda deactivate，对于linux, macos系统是source deactivate。

安装gym

pip install gym

安装stable-baselines(重点）
这里安装一定要参照官方文档stable basellines documention

#安装支持PPO1算法的版本（建议）
pip install stable-baselines[mpi]#without mpi, 不支持PPO,DDPG,TRPO,GAIL
pip install stable-baselines

安装tensorflow
这里需要注意tensorflow的版本必须匹配，stable-baselines不支持tensorflow2以上的版本。或者在stable-baselines的官方文档中有直接通过git安装的命令，可以解决和tensorflow相互匹配的问题，可自行查阅。
先查看stable-baselines的版本

pip list # 2.10.2

然后到官方文档查看，左下角选择v2.10.0,发现这么一段话,也就是支持的tensorflow版本从1.8.0到1.14.0，如果运行报错是关于tensorflow的可以看看是不是tensorflow版本过于先进。

于是我安装了指定版本的tensorflow，可以先查看可安装的版本有哪些conda search --full --name tensorflow或者直接conda search tensorflow

pip install tensorflow==1.14.0

到此为止如果执行命令

python main.py

会发现报错，是关于MPI的错误（导入mpi4py错误），因此还需要安装MPI
简单了解MPI是什么？

MPI : Message Passing Library
是用于并行计算过程中消息传递的库。

安装MPI
官方文档对此有说明，如果安装的stable-baselines是需要MPI版本的，需要下载并安装MPI for windows。点击链接，下载exe文件，然后双击运行安装即可。
至此，大功告成，可以直接运行main文件。

python main.py