BTCが暴落して楽しいことになっていますが、株式みたいな自動取引を目指してチャートを強化学習させようとしてみました。
どちらかというと、強化学習用に使用する OpenAI Gym の練習がしたかったので、色々と設定はガバガバです。
今回は強化学習でいう環境の実装までやってみました。
使用したデータ
Kaggleで配布されているcoincheckのBTCJPY過去チャートデータを使用しました。
データは１分ごとの価格が入っており、これを60ステップずつ飛ばして読む（=1時間毎の価格を使用）します。
今回は2017年1月1日から12ヶ月分のデータを使用しました。
OpenAI gym
強化学習にはPythonからOpenAI gymを使用します。
pipでインストールでき、Atariなど手軽に強化学習を試せるようになていますが、自分で環境を定義して強化学習をすることが可能です。
強化学習では環境の状態と、環境に影響させるエージェントの行動を定義する必要があります。
今回は簡易ですが、それぞれ以下のように設定しました。
-
環境
- 0: 1ステップ前の状態から価格が上昇している
- 1: 1ステップ前の状態から価格が下降している
- 2: 価格に変化なし
-
エージェントの行動
- 0: 買い注文（1単位購入する）
- 1: 売り注文（それまでに購入したBTCを全部売却）
- 2: 注文なし
gymでの環境定義は以下のようにしました。
import gym
import pandas as pd
import gym.spaces
from gym.utils import seeding
import datetime
class MyEnv(gym.Env):
metadata = {'render.modes': ['human', 'ansi']}
time_step = 60
MAX_STEPS = 12*24*30
def __init__(self):
super().__init__()
# action_space, observation_space, reward_range を設定する
self.action_space = gym.spaces.Discrete(3) # 買い、売り、そのまま
self.observation_space = gym.spaces.Box(
low=0, high=float("inf"), shape=(1,1)
)
self.seed = None
# チャート読み込み
df = pd.read_csv('data_check.csv', usecols=[0,7])
df['Timestamp'] = pd.to_datetime(df.Timestamp, unit='s')
df = df[df['Timestamp'] >= datetime.datetime(2017, 1, 1, 0, 0)]
self.data = df.reset_index(drop=True)
self.read_index = self.time_step
self.state = None
self.hold = 0
self.steps = 0
self.reward = 0
self._reset()
def _reset(self):
return self._observe()
def _observe(self):
# トレンド：0上昇、1下降、2そのまま
return [2]
def _step(self, action):
# 返り値: observation, reward, done(ゲーム終了したか),
# info(追加の情報の辞書)
index = self.read_index
hold = self.hold
df = self.data
# 買う
if action == 0:
self.state = action
self.hold += 1
reward = 0
# 売る
elif action == 1:
reward = hold * (df.iloc[index, 1] - df.iloc[index - self.time_step, 1])
self.hold = 0
# そのまま
elif action == 2:
reward = 0
pass
self.steps += 1
if (df.iloc[index, 1] - df.iloc[index - self.time_step, 1]) > 0:
observation = [0]
elif (df.iloc[index, 1] - df.iloc[index - self.time_step, 1]) < 0:
observation = [1]
else:
observation = [2]
self.reward += reward
self.done = self._is_done()
return observation, reward, self.done, {}
def _close(self):
pass
def _seed(self, seed=None):
self.np_random, seed = seeding.np_random(seed)
return [seed]
def _is_done(self):
# 今回は最大で self.MAX_STEPS までとした
if self.steps > self.MAX_STEPS:
return True
else:
return False
実行結果
この環境を以下のようにして実行します。
import random
import myenv
Episodes = 1
for _ in range(Episodes):
env = env = myenv.MyEnv()
done = False
count = 0
while not done:
action = random.choice([0, 1, 2])
observation, reward, done, info = env.step(action)
obs = obs + [observation]
# print observation,reward,done,info
count += 1
if done:
print('reward:', reward)
print('steps:', count)
print('hold:', env.hold)
print("")
実行結果は以下のようになりました。
…どうも環境の設定がよくないのもあり、いい感じの結果じゃないですね。
holdは購入して保有しているBTCです。なので、これを全ステップ後売却すればその分のrewardとして加算できるはずです。
reward: 156.0
steps: 8641
hold: 0
思うこと
今回はとりあえずgymで自前の環境が作ってみたかったので、この題材でやってみました。
本当は純粋な強化学習から踏み込んでDQNで実装したかったのですが、うまいこと環境の定義が思いつかなかったのでとりあえずこうなりました。
強化学習を考えた場合、マルコフ決定仮定の問題として定式化していく都合上、ある状態におけるそれまでの行動は考慮されません。これは時系列を保持するチャートからすれば非常に相性が悪いです。
DQNとして強化学習させる場合、やはりチャートをプロットしたものを画像として解いていくのがゴリ押しのように見えて、実は一番よい方法なのかもしれません。