(このプログラムはまだ動作確認もまともに行っていません。随時アップデートしていきます。リポジトリは記事の最後に。)
はじめに
FXのトレーディングアルゴリズムを強化学習で作ってみたい！
というわけで、界隈では人気のOpenAI Gymのenvでトレーディング環境を作ってみようと思います。
下準備
まずはgymをインストールしておきます。
pip install gym
環境の仕様
こちらのEUR/USDの1分足ヒストリカルデータを4ヶ月分ダウンロードし、
1分足、5分足、30分足、4時間足と4つの時間足を64本分作成します。
なぜこれらの時間足にしたかというと、MT4で指定可能な時間足で馴染みがあり、1分足->5分足は5倍、5分足->30分足は6倍と、各々5～8倍長期に設定されているためです。
とりあえず観測できる情報を自分と同じにしてみようというコンセプトです。
本当は自分が裁量でトレードするときは5分足と1時間足くらいしか見ないのですが、ここらへんは後々調整してみます。
コーディング
OpenAI gymの環境はgym.Envを継承して行います。
今回はFxEnvというクラスを作成します。
class FxEnv(gym.Env):
Envは _reset、_stepとaction_space、observation_spaceが必要なようです。
また、_renderがあると学習の様子をレンダリングできます。
__init__
action_spaceやobservation_space、その他の初期情報は__init__で指定します。
ヒストリカルデータはカレントディレクトリにダウンロードされていることが前提です。
（後々スクレイピングするように改良してもよいのですが月1のことなので…）
def __init__(self):
# 定数
self.STAY = 0
self.BUY = 1
self.SELL = 2
self.CLOSE = 3
# 対象となる通貨ペアの最大値
self.MAX_VALUE = 2
# 初期の口座資金
self.initial_balance = 10000
# CSVファイルのパス配列(最低4ヶ月分を昇順で)
self.csv_file_paths = []
now = datetime.datetime.now()
for _ in range(4):
now = now - relativedelta.relativedelta(months=1)
filename = 'DAT_MT_EURUSD_M1_{}.csv'.format(now.strftime('%Y%m'))
if not os.path.exists(filename):
print('ファイルが存在していません。下記からダウンロードしてください。', filename)
print('http://www.histdata.com/download-free-forex-historical-data/?/metatrader/1-minute-bar-quotes/EURUSD/')
else:
self.csv_file_paths.append(filename)
# スプレッド
self.spread = 0.5
# Point(1pipsの値)
self.point = 0.0001
# 利食いpips
self.take_profit_pips = 30
# 損切りpips
self.stop_loss_pips = 15
# ロット数
self.lots = 0.1
# 0～3のアクション。定数に詳細は記載している
self.action_space = gym.spaces.Discrete(4)
# 1分足、5分足、30分足、4時間足の5時系列データを64本分作る
self.observation_space = spaces.Box(low=0, high=self.MAX_VALUE, shape=numpy.shape([4, 64, 4]))
_reset
_resetでは4ヶ月分のCSVを読み込んでpandasのDataFrameに変換します。
def _reset(self):
self.info = AccountInformation(self.initial_balance)
# CSVを読み込む
self.data = pandas.DataFrame()
for path in self.csv_file_paths:
csv = pandas.read_csv(path,
names=['date', 'time', 'open', 'high', 'low', 'close', 'v'],
parse_dates={'datetime': ['date', 'time']},
)
csv.index = csv['datetime']
csv = csv.drop('datetime', axis=1)
csv = csv.drop('v', axis=1)
self.data = self.data.append(csv)
# 最後に読んだCSVのインデックスを開始インデックスとする
self.read_index = len(self.data) - len(csv)
# チケット一覧
self.tickets = []
_step
_stepではエージェントの行動を受けとり、環境や報酬を返します。
__initで指定したとおりaction_spaceは0～3の4通りなので、それに沿って報酬を決めていきます。
def _step(self, action):
current_data = self.data.iloc[self.read_index]
ask = current_data['close'] + self.spread * self.point
bid = current_data['close'] - self.spread * self.point
if action == self.STAY:
for ticket in self.tickets:
if ticket.order_type == self.BUY:
if bid > ticket.take_profit:
# 買いチケットを利確
profit = (ticket.take_profit - ticket.open_price) * ticket.lots
self.info.balance += profit
self.info.total_pips_buy += profit
elif bid < ticket.stop_loss:
# 買いチケットを損切り
profit = (ticket.stop_loss - ticket.open_price) * ticket.lots
self.info.balance += profit
self.info.total_pips_buy += profit
elif ticket.order_type == self.SELL:
if ask < ticket.take_profit:
# 売りチケットを利確
profit = (ticket.open_price - ticket.take_profit) * ticket.lots
self.info.balance += profit
self.info.total_pips_sell += profit
elif bid < ticket.stop_loss:
# 売りチケットを損切り
profit = (ticket.open_price - ticket.stop_loss) * ticket.lots
self.info.balance += profit
self.info.total_pips_sell += profit
elif action == self.BUY:
ticket = Ticket(self.BUY, ask, ask + self.take_profit_pips * self.point,
ask - self.stop_loss_pips * self.point, self.lots)
self.tickets.append(ticket)
pass
elif action == self.SELL:
ticket = Ticket(self.SELL, bid, bid - self.take_profit_pips * self.point,
bid + self.stop_loss_pips * self.point, self.lots)
self.tickets.append(ticket)
pass
elif action == self.CLOSE:
for ticket in self.tickets:
if ticket.order_type == self.BUY:
# 買いチケットをクローズ
profit = (bid - ticket.open_price) * ticket.lots
self.info.balance += profit
self.info.total_pips_buy += profit
elif ticket.order_type == self.SELL:
# 売りチケットをクローズ
profit = (ticket.open_price - ask) * ticket.lots
self.info.balance += profit
self.info.total_pips_sell += profit
# インデックスをインクリメント
self.read_index += 1
# obs, reward, done, infoを返す
return self.make_obs('ohlc_array'), self.info.balance, self.read_index >= len(self.data), self.info
観測情報の作成
先述の通り、1分足～4時間足を観測情報とするため、1分足からリサンプリングを行います。
pandasのohlc()は1つの列からohlcを作成してしまうため、元の1分足がohlc形式だとohlcそれぞれにohlcを作成してしまい16列に分裂してしまいます。
ここではcloseのみを使って長期のohlcを作成していますが、厳密ではないので後で修正します。
(openはfirst、highはhigh、lowはlow、closeはlastを使って集約すれば良いと予想)
m1 = numpy.array(target.iloc[-64:][target.columns])
m5 = numpy.array(target['close'].resample('5min').ohlc().dropna().iloc[-64:][target.columns])
m30 = numpy.array(target['close'].resample('30min').ohlc().dropna().iloc[-64:][target.columns])
h4 = numpy.array(target['close'].resample('4H').ohlc().dropna().iloc[-64:][target.columns])
return numpy.array([m1, m5, m30, h4])
gym.envに登録する
最初にpip installしたgymのディレクトリにenvディレクトリがあるので、
FxEnvというディレクトリを作成し、fx_env.pyと下記の__init__.pyを置きます。
from fx_env import FxEnv
また、envディレクトリに存在する__init.pyを開き、FxEnvを登録するコードを追記します。
とりあえず動作検証なので7200step（5日間分）だけ動かします。
# FxEnv
register(
id='FxEnv-v0',
entry_point='fx_env:FxEnv',
max_episode_steps=7200,
)
とりあえず動かしてみる
ここでは、ランダムで傍観、買い、売り、手仕舞いを繰り返すという単純なロジックで環境を利用してみましょう。
import gym
import random
env = gym.make('FxEnv-v0')
Episodes = 1
obs = []
for _ in range(Episodes):
observation = env.reset()
done = False
count = 0
while not done:
action = random.choice([0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 2, 3])
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)
実行すると、以下のような出力が得られます。
reward:10119.1893579 steps:7200
今回はランダムでもやや勝ったみたいですね。
次回以降は学習するロジックを組みながら環境のバグを取っていこうと思います。
学習のロジックは別の記事に上げますが、環境に関してはこの記事をアップデートする予定です。
今回のリポジトリはこちらです。
コメント