Python OpenAI Gym - カスタムGym環境の作成

OpenAI Gymで準備されている環境ではなく、自作の環境（カスタムGym環境）を作成してみます。

カスタムGym環境の作成

右への移動を学ぶ環境GoRightを実装します。

エージェントが左右に移動する５マスの環境になります。

[コード]

go_right.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
import numpy as np
import gym

# 右への移動を学ぶ環境
class GoRight(gym.Env):
    # 定数定義
    GRID_SIZE = 5
    LEFT = 0
    RIGHT = 1

    # 初期化
    def __init__(self):
        super(GoRight, self).__init__()
        # グリッドのサイズ
        self.grid_size = self.GRID_SIZE
        # 初期位置の指定
        self.agent_pos = self.GRID_SIZE - 1
        # 行動空間と状態空間の定義
        self.action_space = gym.spaces.Discrete(2)
        self.observation_space = gym.spaces.Box(low=0, high=self.GRID_SIZE - 1, shape=(1,), dtype=np.float32)

    # 環境のリセット
    def reset(self):
        # 初期位置の指定
        self.agent_pos = 0
        # 初期位置をfloat32のnumpy配列に変換
        return np.array(self.agent_pos).astype(np.float32)

    # 環境の１ステップ実行
    def step(self, action):
        # 移動
        if action == self.LEFT:
            self.agent_pos -= 1
        elif action == self.RIGHT:
            self.agent_pos += 1
        self.agent_pos = np.clip(self.agent_pos, 0, self.GRID_SIZE)
        # エピソード完了の計算
        done = self.agent_pos == self.GRID_SIZE - 1
        # 報酬の計算
        reward = 1 if done else - 0.1
        return np.array(self.agent_pos).astype(np.float32), reward, done, {}

    # 環境の描画
    def render(self, mode='console', close=False):
        # エージェントはA、他は.で表現する
        print('.' * self.agent_pos, end='')
        print('A', end='')
        print('.' * (self.GRID_SIZE - 1 - self.agent_pos))

カスタムGym環境の動作確認

自作した環境をランダム行動で実行してみます。

gym.make関数を使わずに、GoRightクラスを直接生成しています。

[コード]

test_go_right.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import gym
from go_right import GoRight

# 環境の生成
env = GoRight()

# １エピソードのループ
state = env.reset()

while True:
    # ランダム行動の取得
    action = env.action_space.sample()
    # １ステップの実行
    state, reward, done, info = env.step(action)
    # 環境の描画
    env.render()
    print('reward:', reward)
    # エピソード完了
    if done:
        print('done')
        break

上記コードを実行するとコンソールに次のような表示がされます。
（ランダム行動のため実行結果は毎回異なります。）

[実行結果]

reward: -0.1
.A...
reward: -0.1
A....
reward: -0.1
.A...
reward: -0.1
..A..
reward: -0.1
...A.
reward: 1
....A
done

エージェントAが左右に移動し、最終的に一番右に移動し終了していることが分かります。

Python OpenAI Gym - CartPole（棒たてゲーム）を試す② 強化学習編

Stable Baselinesという強化学習アルゴリズムを使ってCartPoleを実行します。

インストール

下記のコマンドを実行しStable Baselinesを準備します。

1
2
3
4
pip install stable-baselines[mpi]
pip install tensorflow==1.14.0
pip install pyqt5
pip install imageio

強化学習アルゴリズムを使ってCartPole実行

強化学習のモデルを作成し、100,000回学習を行ってからCartPoleを実行してみます。

[コード]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import gym
from stable_baselines.common.vec_env import DummyVecEnv
from stable_baselines import PPO2

# 環境の作成
env = gym.make('CartPole-v0')
env = DummyVecEnv([lambda: env])

# モデルの作成
model = PPO2('MlpPolicy', env, verbose=1)

# モデルの学習
model.learn(total_timesteps=100000)

# モデルのテスト
state = env.reset()
for i in range(200):
    # 環境の描画
    env.render()
    
    # モデルの推論
    action, _ = model.predict(state, deterministic=True)
    
    # 1ステップ実行
    state, rewards, done, info = env.step(action)
    
    # エピソード完了判定
    if done:
        break

# 環境のクローズ
env.close()

実行してみると、棒が倒れることなくうまくバランスをとっていることが確認できます。

実行結果

Python OpenAI Gym - CartPole（棒たてゲーム）を試す①

OpenAI Gymに含まれるCartPole（棒たてゲーム）を試してみます。

インストール

下記のコマンドを実行しOpenAI Gym環境を準備します。

1
pip install gym

CartPole（棒たてゲーム）を実行する

ランダムにカートを動かすコードを準備します。

[コード]

1
2
3
4
5
6
7
8
9
10
import gym

# 環境の作成
env = gym.make('CartPole-v0')

# ランダム行動による動作確認
env.reset()
while True:
    env.render()
    env.step(env.action_space.sample())

実行すると次のような画面が表示されますが、ランダム行動のためすぐに棒が倒れてしまいます。

実行結果

Python AnyTrading - 投資の強化学習環境を試す② 学習編

FXのトレードを学習させて、利益総額が増えるかどうかを確認します。

インストール

Windowsで実行する場合、Microsoft MPIをインストールする必要があります。

Microsoft MPI - https://www.microsoft.com/en-us/download/details.aspx?id=100593

さらに下記のコマンドを実行し学習のための環境を準備します。

1
2
3
4
pip install stable-baselines[mpi]
pip install tensorflow==1.14.0
pip install pyqt5
pip install imageio

学習投資

FXトレードを学習してから投資を行う処理を実装します。

[コード]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import gym
import gym_anytrading
import matplotlib.pyplot as plt
import os
from gym_anytrading.envs import TradingEnv, ForexEnv, StocksEnv, Actions, Positions
from gym_anytrading.datasets import FOREX_EURUSD_1H_ASK, STOCKS_GOOGL
from stable_baselines.common import set_global_seeds

from stable_baselines.common.vec_env import DummyVecEnv
from stable_baselines import PPO2
from stable_baselines.bench import Monitor

# ログフォルダの作成
log_dir = './logs/'
os.makedirs(log_dir, exist_ok=True)

# 環境を生成（frame_boundはデータセット内の訓練範囲を開始行数と終了行数で指定）
env = gym.make('forex-v0', frame_bound=(50, 100), window_size=10)
env = Monitor(env, log_dir, allow_early_resets=True)

# シードの指定
env.seed(0)
set_global_seeds(0)

# ベクトル化環境の生成
env = DummyVecEnv([lambda: env])

# モデルの生成
model = PPO2('MlpPolicy', env, verbose=1)

# モデルの読み込み(学習済みデータがある場合)
# model = PPO2.load('trading_model')

# モデルの学習
model.learn(total_timesteps=128000)

# モデルの保存
model.save('trading_model')

# モデルのテスト
env = gym.make('forex-v0', frame_bound=(50, 100), window_size=10)
env.seed(0)
state = env.reset()
while True:
    # 行動の取得
    action, _ = model.predict(state)
    # 1ステップ実行
    state, reward, done, info = env.step(action)
    # エピソード完了
    if done:
        print('info:', info)
        break

# グラフのプロット
plt.cla()
env.render_all()
plt.show()

実行結果は下記のとおりです。

[実行結果]

info: {'total_reward': 76.70000000000952, 'total_profit': 0.993213633946179, 'position': 1}

累計報酬（total_reward）、純利益（total_profit）、ポジション（position：0がショート、1がロング）が表示されます。

ランダムの時と比べてあまり結果がよくなりませんでした・・・というよりむしろ成績が落ちていて、もう少し理解を深める必要がありそうです。

グラフに表示される結果は以下の通りです。

結果

赤●が「0:Sell」で落ちる予想、緑●が「1:Buy」で上がる予想を表します。

Python AnyTrading - 投資の強化学習環境を試す①

AnyTradingは、FXや株式のトレーディングアルゴリズムを試すための強化学習環境です

インストール

下記のコマンドを実行しAnyTradingの実行環境をインストールします。

1
2
3
4
pip install gym-anytrading
pip install stable-baselines
pip install tensorflow==1.14.0

ランダム投資

ランダムに投資を行う処理を実装します。

[コード]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import gym
import gym_anytrading
import matplotlib.pyplot as plt
from gym_anytrading.envs import TradingEnv, ForexEnv, StocksEnv, Actions, Positions
from gym_anytrading.datasets import FOREX_EURUSD_1H_ASK, STOCKS_GOOGL
from stable_baselines.common import set_global_seeds

# 環境を生成（frame_boundはデータセット内の訓練範囲を開始行数と終了行数で指定）
env = gym.make('forex-v0', frame_bound=(50, 100), window_size=10)

# シードの指定
env.seed(0)
set_global_seeds(0)

# ランダム行動による動作確認
state = env.reset()
while True:
    action = env.action_space.sample()

    # 1ステップ実行
    state, reward, done, info = env.step(action)

    # エピソード完了
    if done:
        print('info:', info)
        break

# グラフのプロット
plt.cla()
env.render_all()
plt.show()

実行結果は下記のとおりです。

[実行結果]

info: {'total_reward': 85.50000000000279, 'total_profit': 0.9914207128440394, 'position': 0}

累計報酬（total_reward）、純利益（total_profit）、ポジション（position：0がショート、1がロング）が表示されます。
ランダムの割には利益が上がっています。

また下記のようなグラフが表示されます。

結果

赤●が「0:Sell」で落ちる予想、緑●が「1:Buy」で上がる予想を表します。

Python - 環境変数の取得

osモジュールのenviron変数には様々な環境変数が辞書型で格納されています。

環境変数の取得

[コード]

1
2
3
4
5
import os

print(os.environ)

print(os.environ['PATH'])

実行結果は下記のとおりです。

[実行結果]

environ({'ALLUSERSPROFILE': 'C:\\ProgramData', 'APPDATA': 'C:\\Users\\xxxx\\AppData\\Roaming',
（以下省略）
C:\ProgramData\Oracle\Java\javapath;C:\Util\Anaconda3;
（以下省略）

環境変数の設定が必要な処理をする場合に、実行前に設定を確認し問題があればアラート表示するなどの使い方ができると思います。

Python - 数値型

数値型のいろいろな表現をまとめてみました。

int型

[コード]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 10進数
num = 12
# => 12

# 8進数
num = 0o16
# => 14

# 16進数
num = 0xf2
# => 242

# 2進数
num = 0b1100
# => 12

float型

[コード]

1
2
3
4
5
6
7
8
9
10
num = 1.2
# => 1.2

# 指数表記 1.2×10の３乗
num = 1.2e3
# => 1200.0

# 指数表記 1.2×10のマイナス３乗
num = 1.2e-3
# => 0.0012

Python - グローバル変数、ローカル変数の表示

組み込みのglobals関数、locals関数を使うと、現在のスコープのグローバル変数、ローカル変数を表示することができます。

グローバル変数、ローカル変数の表示

[コード]

1
2
3
4
5
6
7
8
9
10
GLOBAL_AVR = 123
def func():
    x = 3
    print('グローバル変数')
    print(globals())
    print()
    print('ローカル変数')
    print(locals())

func()

実行結果は下記のとおりです。

グローバル変数
{'__name__': '__main__', '__doc__': None, '__package__': None, '__loader__': , '__spec__': None, '__annotations__': {}, '__builtins__': , 'GLOBAL_AVR': 123, 'func': }

ローカル変数
{'x': 3}

Python Image - 画像を編集する

Imageライブラリを使うとPythonから画像を編集することができます。

インストール

下記のコマンドを実行し、Imageライブラリをインストールします。

1
pip install PIL

ログ出力

基本的な画像処理をまとめて実行します。

[コード]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from PIL import Image

# 画像ファイルを開く
image = Image.open('a.jpg')

# 画像のサイズを変更(横幅・縦幅)
image = image.resize([256, 256])
# 画像ファイルを出力
image.save('b.png')

# 回転させる
image = image.rotate(45)
# 画像ファイルを出力
image.save('c.png')

# 反転させる
image = image.transpose(Image.FLIP_LEFT_RIGHT)  # 左右対称
#image = image.transpose(Image.FLIP_TOP_BUTTOM) # 上下対象
# 画像ファイルを出力
image.save('d.png')

# 切り貼り
part = image.crop((0, 0, 64, 64))   # 左、上、右、下
image.paste(part, (10, 10))         # 左、上
image.save('e.png')

Python Logging - ログを出力する

Pythonでログの出力・制御を行うためには、loggingモジュールを使用します。

標準出力とファイルへの出力や、重要度に応じた出力の制御、書式の設定などが可能です。

ログ出力

今回は、ファイルと標準出力にログを出力します。

またログの書式設定とログローテーションの設定も行います。

[コード]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import logging
import logging.handlers
import sys

# 適当な名前を付けてLoggerオブジェクトを生成
logger = logging.getLogger('test_log')

# デバッグレベルを指定
#  logging.CRITICAL（一番高いレベル）
#  logging.ERROR
#  logging.WARNING
#  logging.INFO
#  logging.DEBUG（一番低いレベル）
logger.setLevel(logging.DEBUG)

# ログ出力の書式設定
formatter = logging.Formatter('%(asctime)s[%(levelname)s] %(message)s')

# ファイル出力の設定（ログローテーション設定あり）
#file_handler = logging.FileHandler('test.log')
file_handler = logging.handlers.RotatingFileHandler('test.log', maxBytes=1024, backupCount=5)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)

# 標準出力の設定
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)

logger.debug('debug test.')
logger.info('info test.')
logger.warning('warning test.')
logger.error('error test.')
logger.critical('critical test.')

実行結果は下記のとおりです。

[実行結果]

2020-04-26 18:40:15,631[DEBUG] debug test.
2020-04-26 18:40:15,636[INFO] info test.
2020-04-26 18:40:15,639[WARNING] warning test.
2020-04-26 18:40:15,644[ERROR] error test.
2020-04-26 18:40:15,648[CRITICAL] critical test.

test.logファイルにも同様の内容が出力されます。