OpenAI 澳洲股市Gym 环境

开源  https://github.com/guidebee/asx_gym

from datetime import date
from logging import INFO

import gym
import asx_gym
# from agents.buy_and_keep_agent import BuyAndKeepAgent
from agents.random_agent import RandomAgent
from asx_gym.envs import AsxObservation

gym.logger.set_level(INFO)
start_date = date(2019, 5, 1)
simulate_company_list = [2, 3, 4, 5, 6, 44, 300, 67, 100, 200]
# simulate_company_list = [3]
env = gym.make("AsxGym-v0", start_date=start_date,
               simulate_company_list=simulate_company_list)
stock_agent = RandomAgent(env)
# stock_agent = RandomAgent(env, min_volume=100, max_volume=500)
# stock_agent = BuyAndKeepAgent(env, 3)

observation = env.reset()
for _ in range(200000 * 24):
    env.render()
    company_count = len(env.simulate_company_list)

    observation, reward, done, info = env.step(stock_agent.action())
    if done:
        env.insert_summary_images(30)
        observation = env.reset()
        stock_agent.reset()
    if observation is not None:
        asx_observation = AsxObservation(observation)
        print(asx_observation.to_json_obj())
        print(info)

env.close()

 

Open AI Gym教程(7): Gym 内置二维图形渲染系统

我们使用Open AI Gym 来测试人工智能或是机器学习的算法。尽管Gym以及Gym retro提供了数以千计的环境可供测试。但我们实际的问题总有gym没有提供的情形。因此Open AI Gym 通常有两种问题,一是如果把我们遇到的实际问题转化成Gym 场景,然后利用已有的Gym 机器学习的算法来解决实际问题二是如何选择和设计算法来解决Gym的场景。

我们很都人都玩过魔法,在没有完成三层之前,看到别人完成三层的时候,觉得好牛啊:-),实际情况并非如此。

实际上玩魔方,可能在完成第一层时,还可能需要动点脑筋,后面二层,三层基本上做为玩家来说,基本是套用口诀,判断当前魔方所处的状态,然后套用口诀,并没有任何智能可言。记得魔方所处的状态越多,恢复三层速度越快,这点和我们使用机器学习算法解决问题非常类似。

人工智能和机器学习目前是能够解决不少问题,当在实际应用时,可以除了少许研究人员来说,设计新的算法(类似于发现魔方新的口诀)还需要大量的脑力劳动,对于一般的应用开发。基本是也是套用现成的算法,判断下问题的性质,是supervised learning(监督学习),unsupervised learning(无监督学习) ,还是使用Reinforcement learning(强化学习)。可能在调整算法的超级参数(hyper parameters),还需要一些思考(就想玩魔方的第一层),后面基本也是套路。

机器学习的一个主要特点是无需事先设计好预定的算法,而是通过大量的数据来训练神经网络,而且很多时候数据量的多少可以弥补算法的不足,和我们常规的算法大不相同。

因此在解决实际问题时候,把问题构造成Open AI Gym场景相对显得更为重要,而且需要更多的”智能”。就像当初设计出魔方的人。

Gym的一个重要的方法是render,也就是显示出场景。可以是图像,内存(rgb_array)或是文字。

Gym的经典场景比如(cartpole),以及atari游戏的显示,Open AI Gym提供了一个比较简单的二维图形渲染系统,我们如果自定义场景也可以使用同样的图形系统作为显示:

gym.envs.classic_control.rendering

定义了这个图型系统,简单的说一个是矢量图像(Viewer),一个是光栅图像(SimpleImageViewer)显示,Atari游戏有自己的图像显示,因此直接使用SimpleImageViewer,显示图像imshow(image)。
Gym的经典场景使用矢量图像(Viewer),它提供一些简单的二维图像显示,比如点,线,面,坐标变换等:

下面的代码为CartPole的例子改写而来,旋转平衡杆720度。

from gym.envs.classic_control.rendering import *

length = 0.5

x_threshold = 2.4

screen_width = 600
screen_height = 400

world_width = x_threshold * 2
scale = screen_width / world_width
carty = 100  # TOP OF CART
polewidth = 10.0
polelen = scale * (2 * length)
cartwidth = 50.0
cartheight = 30.0

viewer = Viewer(screen_width, screen_height)

l, r, t, b = -cartwidth / 2, cartwidth / 2, cartheight / 2, -cartheight / 2
axleoffset = cartheight / 4.0
cart = FilledPolygon([(l, b), (l, t), (r, t), (r, b)])
carttrans = Transform()
cart.add_attr(carttrans)
viewer.add_geom(cart)
l, r, t, b = -polewidth / 2, polewidth / 2, polelen - polewidth / 2, -polewidth / 2
pole = FilledPolygon([(l, b), (l, t), (r, t), (r, b)])
pole.set_color(.8, .6, .4)
poletrans = Transform(translation=(0, axleoffset))
pole.add_attr(poletrans)
pole.add_attr(carttrans)
viewer.add_geom(pole)
axle = make_circle(polewidth / 2)
axle.add_attr(poletrans)
axle.add_attr(carttrans)
axle.set_color(.5, .5, .8)
viewer.add_geom(axle)
track = Line((0, carty), (screen_width, carty))
track.set_color(0, 0, 0)
viewer.add_geom(track)

l, r, t, b = -polewidth / 2, polewidth / 2, polelen - polewidth / 2, -polewidth / 2
pole.v = [(l, b), (l, t), (r, t), (r, b)]

cartx = screen_width / 2.0  # MIDDLE OF CART
carttrans.set_translation(cartx, carty)
poletrans.set_rotation(0)
rotate = 720

while rotate > 0:
    poletrans.set_rotation(rotate * math.pi / 180.0)
    rotate -= 1
    viewer.render()

viewer.close()

CartPole由Cart(Polygon),Pole(Polygon),Axis(Circle),Track(Line)构成,

这些简单的几何图像类型都是Geom 的子类,Geom可以定义几个属性,如颜色,线宽,坐标转换的Matrix(比如实现Pole的旋转)。Gym 图像显示是记忆pyglet(OpenGL)。如果你属性OpenGL,这些都是非常容易理解的。

Open AI Gym教程(6):自定义环境的基本步骤

虽然Open AI Gym自带了大量的环境可供测试,但我们终会碰到需要自定义环境的时候,本篇介绍构造自定义环境的基本步骤。

这里借用一个简单的例子,猜数字游戏,这个游戏和我们常见的猜数字类似,稍有不同。游戏随机给出一个浮点数(可以取值的范围事先不知道,但可以给出对于的值域空间Spaces),你可以最多猜200次,然后游戏根据你猜的数字,给出4中不同的测量值(Observation).

  • 0 — 初始,只在系统reset()后可以为0
  • 1 — 猜的数偏低
  • 2 — 猜的数和游戏给定的值相同,由于是浮点数,系统比较两个值偏差在1%内就认为相同。
  • 3 –猜的数偏大。

系统给出的奖励如下:

  • 1 如果猜中数字 ,允许1%偏差
  • 0 没有猜中, 也就是偏差高于1%

根据游戏规则,我们可以定义这个环境的Action 和 Observation 的值域空间,Observation比较简单,使用整数类型就可以

observation_space = spaces.Discrete(4)

而Action 可以用多种定义,最简单的是直接使用猜的数字作为值域空间,比如:

self.bounds = 10000

self.action_space = spaces.Box(low=np.array([-self.bounds]), high=np.array([self.bounds]),
                                       dtype=np.float32)
self.observation_space = spaces.Discrete(4)

此外,还可以使用Tuple类型, (Discrete(1),(Box(low=0, high=np.array([self.bounds]),
dtype=np.float32))
第一个增大或是减小,第二个为增大或是减小的数值。本例采用第一种方法。

根据上述的分析, 我们就可以来设计自定义Gym环境。一个Gym环境的项目结构如下:

.
├── LICENSE
├── README.md
├── guessing_number
│   ├── __init__.py
│   └── envs
│       ├── __init__.py
│       └── guessing_number_env.py
├── setup.py
└── test.py

其中setup.py 一般为:

from setuptools import find_packages, setup

setup(
    name="guessing_number",
    version="0.0.1",
    install_requires=["gym>=0.2.3", "numpy"],
    packages=find_packages(),
)

 

定义项目的名称,版本,以及所依赖的其它软件包。

而 guessing_number/__init__.py 一般如下:

from gym.envs.registration import register

register(id="GuessingNumber-v0", entry_point="guessing_number.envs:GuessingNumberEnv")

 

其中环境个名称的格式为GuessingNumber-v0,一般在环境名称后使用v0,v1来代表不同的版本。
entry_point给出环境人主类的人口点。

参考Open AI Gym教程(3): 环境 Env 我们可以定义GuessingNumberEnv如下:

import numpy as np

import gym
from gym import spaces
from gym.utils import seeding


class GuessingNumberEnv(gym.Env):
    """Number guessing game

    The object of the game is to guess within 1% of the randomly chosen number
    within 200 time steps

    After each step the agent is provided with one of four possible observations
    which indicate where the guess is in relation to the randomly chosen number

    0 - No guess yet submitted (only after reset)
    1 - Guess is lower than the target
    2 - Guess is equal to the target
    3 - Guess is higher than the target

    The rewards are:
    0 if the agent's guess is outside of 1% of the target
    1 if the agent's guess is inside 1% of the target

    The episode terminates after the agent guesses within 1% of the target or
    200 steps have been taken

    The agent will need to use a memory of previously submitted actions and observations
    in order to efficiently explore the available actions

    The purpose is to have agents optimise their exploration parameters (e.g. how far to
    explore from previous actions) based on previous experience. Because the goal changes
    each episode a state-value or action-value function isn't able to provide any additional
    benefit apart from being able to tell whether to increase or decrease the next guess.

    The perfect agent would likely learn the bounds of the action space (without referring
    to them explicitly) and then follow binary tree style exploration towards to goal number
    """
    def __init__(self):
        self.range = 1000  # Randomly selected number is within +/- this value
        self.bounds = 10000

        self.action_space = spaces.Box(low=np.array([-self.bounds]), high=np.array([self.bounds]),
                                       dtype=np.float32)
        self.observation_space = spaces.Discrete(4)

        self.number = 0
        self.guess_count = 0
        self.guess_max = 200
        self.observation = 0

        self.seed()
        self.reset()

    def seed(self, seed=None):
        self.np_random, seed = seeding.np_random(seed)
        return [seed]

    def step(self, action):
        assert self.action_space.contains(action)

        if action < self.number:
            self.observation = 1

        elif action == self.number:
            self.observation = 2

        elif action > self.number:
            self.observation = 3

        reward = 0
        done = False

        if (self.number - self.range * 0.01) < action < (self.number + self.range * 0.01):
            reward = 1
            done = True

        self.guess_count += 1
        if self.guess_count >= self.guess_max:
            done = True

        return self.observation, reward, done, {"number": self.number, "guesses": self.guess_count}

    def reset(self):
        self.number = self.np_random.uniform(-self.range, self.range)
        self.guess_count = 0
        self.observation = 0
        return self.observation

项目完成后可以使用

pip install -e .

来安装这个环境。

最有我们可以设计两个不同的Agent来玩这个猜数字游戏,一个是随机猜:

class RandomAgent(object):
    """The world's simplest agent!"""

    def __init__(self, action_space):
        self.action_space = action_space

    def act(self, observation, reward, done):
        return self.action_space.sample()

 

第二个是改进后的随机猜,根据step返回的observation (1-偏小,3-偏大)适当调整所猜的数字:

class BetterRandomAgent(object):
    """The world's 2nd simplest agent!"""

    def __init__(self, action_space):
        self.action_space = action_space

    def act(self, observation, last_action):
        new_action = last_action
        if observation == 1:
            new_action = last_action + abs(last_action / 2)

        elif observation == 3:
            new_action = last_action - abs(last_action / 2)
        if abs(last_action - new_action) < 1e-1:
            new_action = self.action_space.sample()
        return new_action

 

有了这两个Agent,我们就可以测试这个环境.分别运行100次,看看每个Agent猜中的次数,和平均猜的次数:

import gym

import guessing_number

class RandomAgent(object):
    """The world's simplest agent!"""

    def __init__(self, action_space):
        self.action_space = action_space

    def act(self, observation, reward, done):
        return self.action_space.sample()


class BetterRandomAgent(object):
    """The world's 2nd simplest agent!"""

    def __init__(self, action_space):
        self.action_space = action_space

    def act(self, observation, last_action):
        new_action = last_action
        if observation == 1:
            new_action = last_action + abs(last_action / 2)

        elif observation == 3:
            new_action = last_action - abs(last_action / 2)
        if abs(last_action - new_action) < 1e-1:
            new_action = self.action_space.sample()
        return new_action


if __name__ == '__main__':

    env = gym.make('GuessingNumber-v0')
    env.seed(0)
    agent = BetterRandomAgent(env.action_space)

    episode_count = 100
    reward = 0
    done = False

    total_reward = 0
    total_guesses = 0
    for i in range(episode_count):
        last_action = env.action_space.sample()
        ob = env.reset()
        while True:
            action = agent.act(ob, last_action)
            ob, reward, done, info = env.step(action)
            last_action = action

            # print(f'count={info["guesses"]},number={info["number"]},guess={action},ob={ob},reward={reward}')
            if done:
                total_reward += reward
                total_guesses += int(info["guesses"])
                break

    print(f'Total better random reward {total_reward}, average guess {round(total_guesses / 100, 1)}')

    env.seed(0)
    agent = RandomAgent(env.action_space)
    reward = 0
    done = False

    total_reward = 0
    total_guesses = 0

    for i in range(episode_count):
        ob = env.reset()
        while True:
            action = agent.act(ob, reward, done)
            ob, reward, done, info = env.step(action)

            if done:
                total_reward += reward
                total_guesses += int(info["guesses"])
                break

    # Close the env and write monitor result info to disk
    env.close()
    print(f'Total random reward {total_reward}, average guess {round(total_guesses / 100, 1)}')

 

几个可能的运行结果:

Total better random reward 100, average guess 35.9
Total random reward 15, average guess 180.6
-----
Total better random reward 100, average guess 39.2
Total random reward 20, average guess 175.9
--
Total better random reward 100, average guess 38.2
Total random reward 24, average guess 177.2
--
Total better random reward 100, average guess 38.6
Total random reward 18, average guess 180.4

 

可以看到改进后的Agent,几乎每次都猜中,平均30多次就猜中,而完全随机的Agent,猜中20次左右,每次需要180次左右。

如果我们使用二分法来设计Agent,由于事先不知道所猜数的范围,所以可以先设计算法,得到所猜数的范围,比如从100开始,每次翻倍,直到observation从偏小变成偏大。
我们的例子的取值范围为(-10000,10000),因此只要8次就可以得到范围为[-25600,25600]。然后使用二分法,最多也是8次就可以猜中数字。

本篇源码 https://github.com/guidebee/guessing_number

Open AI Gym教程(5): 各种环境Wrappers

gym.wrappers自带个多种封装后的Wrapper,对Action,Reward,Obersvation 做了处理(其中Observation Wrapper可以看成类似于图像处理时的各种滤镜)

 

其中的Monitor 可以用来观测环境的参数并保持视频到本地。

outdir = '/tmp/random-agent-results'
env = wrappers.Monitor(env, directory=outdir, force=True)

TimeLimit 可以控制最大的episode的步数。
FrameStack 可以多个Observation 组合成一个多帧的Observation,比如你的AI算法需要连续的几个帧的数据做为输入。
RecordEpisodeStatistics 可以用来记录episodes的一些统计信息
PixelObservationWrapper 可以为原先的Observation增加一些图像数据。

此外gym.vector 目录定义了矢量化环境,一般Gym环境是单个环境,如果我们需要同时使用多个环境来设计我们的AI算法,我们使用使用矢量化环境。这时envs, actions, rewards,和 observations都会是个一维数组。

Open AI Gym教程(4): Wrapper封装后的Env

core.py定义一个Wrapper类,它是环境Env的一个子类,便于模块化变换,比如对环境的reward进行换算,Observation的彩色图像进行灰度处理,提供计算效率。

class Wrapper(Env):
    r"""Wraps the environment to allow a modular transformation.

    This class is the base class for all wrappers. The subclass could override
    some methods to change the behavior of the original environment without touching the
    original code.

    .. note::

        Don't forget to call ``super().__init__(env)`` if the subclass overrides :meth:`__init__`.

    """
    def __init__(self, env):
        self.env = env
        self.action_space = self.env.action_space
        self.observation_space = self.env.observation_space
        self.reward_range = self.env.reward_range
        self.metadata = self.env.metadata

    def __getattr__(self, name):
        if name.startswith('_'):
            raise AttributeError("attempted to get missing private attribute '{}'".format(name))
        return getattr(self.env, name)

    @property
    def spec(self):
        return self.env.spec

    @classmethod
    def class_name(cls):
        return cls.__name__

    def step(self, action):
        return self.env.step(action)

    def reset(self, **kwargs):
        return self.env.reset(**kwargs)

    def render(self, mode='human', **kwargs):
        return self.env.render(mode, **kwargs)

    def close(self):
        return self.env.close()

    def seed(self, seed=None):
        return self.env.seed(seed)

    def compute_reward(self, achieved_goal, desired_goal, info):
        return self.env.compute_reward(achieved_goal, desired_goal, info)

    def __str__(self):
        return '<{}{}>'.format(type(self).__name__, self.env)

    def __repr__(self):
        return str(self)

    @property
    def unwrapped(self):
        return self.env.unwrapped

 

在Core.py定义了三个子类ObservationWrapper,RewardWrapper和ActionWrapper。

class ObservationWrapper(Wrapper):
    def reset(self, **kwargs):
        observation = self.env.reset(**kwargs)
        return self.observation(observation)

    def step(self, action):
        observation, reward, done, info = self.env.step(action)
        return self.observation(observation), reward, done, info

    def observation(self, observation):
        raise NotImplementedError


class RewardWrapper(Wrapper):
    def reset(self, **kwargs):
        return self.env.reset(**kwargs)

    def step(self, action):
        observation, reward, done, info = self.env.step(action)
        return observation, self.reward(reward), done, info

    def reward(self, reward):
        raise NotImplementedError


class ActionWrapper(Wrapper):
    def reset(self, **kwargs):
        return self.env.reset(**kwargs)

    def step(self, action):
        return self.env.step(self.action(action))

    def action(self, action):
        raise NotImplementedError

    def reverse_action(self, action):
        raise NotImplementedError

 

从代码中可以看到,我们可以使用action,reward,observation 方法对之前的action,reward,observation进行转换后再使用到step方法中。
这几个类都是抽象类,具体的实现定义在gym.wrappers。我们后面具体介绍。

Open AI Gym教程(3): 环境 Env

gym/core.py 中定义了可能是Open AI Gym最重要的一个类 Env(环境)。它是对现实世界模型的一个简化的抽象。对应到游戏应用中,就是游戏的一个个场景,我们控制游戏人物(Agent)在这个世界中经历种种历险:

Env定义游戏角色(Agent)活动的环境,比如上图的超级玛丽的世界。ENV定义了几个接口函数:step, reset, render, close, seed。以及几个属性:

  • action_space 角色(Agent)可以实施的动作,比如Mario可以上下左右,跳跃等
  • observation_space 环境可能的状态的值域空间,比如Mario世界所有的场景
  • reward_range 奖励值得取值范围。比如Mario游戏的得分是一个可能的奖励值(实际我们实现AI算法时,光使用得分不一定能得到好的结果,可能需要综合其它因素来计算算法的奖励值),它的缺省范围是[-inf,+inf]

Step方法

step(self, action)

step步进游戏一步,比如按下右键(或者什么也不做),环境根据这个动作前进一步,变换游戏的场景,判断游戏是否结束等,这个方法返回四个值:

  • observation (观测量),输入一个动作后下个场景的状态,它的类型是一个对象,不同环境类型不同。
  • reward (奖励),浮点类型,前个动作所带来的奖励。也和环境有关,我们的目前是获取更多的奖励。
  • done ,游戏结束标志,布尔类型。 比如Flappy Bird游戏小鸟撞死了,我们一般需要检测这个标志以便重新开始下一轮尝试。
  • info 游戏中的一些辅助信息,也和场景有关。一般来说我们不能使用使用这些信息来构造我们的算法。

Reset方法

游戏结束后重置游戏,以便于下次尝试。它返回游戏的初始observation (观测量)。

Render方法

render(self, mode='human')

render 用来”显示”游戏的场景,这个方法不是每个环境都必须提供的。 它可以分三种模式

  • human   一般表示屏幕显示,用于人机交互。
  • rgb_array  返回一个numpy数组(x,y,3)代表图像的RGB数值,比较适合生成视频。
  • ansi 文字描述。

Close 方法

用来释放环境可能使用到的资源或其它工作。

Seed方法

seed(self, seed=None)

 

seed 提供环境中用到的随机数值生成器的种子。环境一般无需另外提供该方法,如果环境使用多个随机数值生成器,我们可以提供多个种子以免多个随机数值生成器产生的数值有相关性。