はじめに
近年、人工知能ブームにより、人工知能を使ったトレーディング手法が盛んです。そこで、今回は深層強化学習を用いたシステムトレーディングを実施しました。
まず、基本的な深層強化学習を用いたトレーディングモデルです。agentの行動として、 BUY、HOLD、SELLの三つの内一つを選択します。環境の戻り値として、状態（今現在保有しているポジションの価格、市場価格、手持ちのキャッシュ）、報酬（手持ちのキャッシュの変化値（含む益も含む））、終了（取引の終了か否か）、情報（ターミナルにディスプレイする情報）を返します。
使用データについて
トレンド傾向の掴みやすさから、yahoo financeからGSPCの日足を使用しました。
訓練データの期間：2015/1/1 - 2017/6/30
テストデータの期間：2017/7/1 - 2021/1/1
以下ソースコード
MuZero1
AlphaZeroでは、ルールブックが与えたれていました。しかし、現実の世界は、複雑で、簡単なルールを獲得することが困難です。そこで、MuZeroでは、MuZeroは、環境全体もしくはルールブックそのものをモデル化するのではなく、エージェントの判断にとって重要な側面だけをモデル化する(ダイナミクス関数)ことに成功しました。
モデルベースアルゴリズムとは
DQN、R2D2、Agent57などに代表されるモデルフリーアルゴリズムは、環境に対するエージェントの試行錯誤を経験という形で学習(Q学習)を行います。ただこれだと、環境の変化に弱いという構造的な欠陥があります。
そこで、MuZero、World Modelなどに代表されるモデルベースアルゴリズムは、環境そのものの正確な予測に基づき、エージェントが行動を予想します。MuZeroでは、エージェントの意思決定プロセスにとって重要な側面のモデル化を行なっています。
MuZeroの全体図
MuZeroは、表現関数(representation function)、予測関数(prediction function)、ダイナミクス関数(dynamics function)の3つモデルを相互に影響し合いながら予測を行います。AlphaZeroでは、予測を行うためのルールブックが与えたれ、予測のたびに予測関数で予測を行い、ルールブックで確認を行うということを行っていました。しかし、MuZeroでは、ルールブックもダイナミクス関数で予測を行います。これを行うことで、atariなどのゲームに対して、対応できるようになりました。
※観測はゲーム(現実空間)から与えられた、状態と行動の軌跡です。隠し状態は、抽象空間上(エージェントが次の手を探査しているイメージ空間、エージェントの頭の中)で、起きている状態を指します。
表現関数: o → h → s
予測関数: s → f → p,v
ダイナミクス関数: s → g → r,s'
o:観測 s:隠し状態 p 方策: v:価値 r:即時報酬
MuZeroが環境に及ぼす作用
MuZeroでは、モンテカルロ木探査をステップ毎実施します。まず、アクション選択時に、ルートノードからの各アクションの検索ポリシーπtからサンプリングします。アクション実行後、新しい観測ot+1を生成し、ut+1に報酬を与えます。エピソードの最後に、軌跡データはgameとしてリプレイバッファに保存します。
モデルの学習過程
まず、学習開始時にリプレイバッファからgame(軌跡)をサンプリングします。表現関数に、サンプリングしたgame(軌跡)から過去の観測o1、...、otを入力し隠し状態を出力します。予測関数に隠し状態を入力し、価値とポリシーを出力します。ダイナミクス関数は、前のステップから隠し状態sk+kの実際のアクションを入力として即時報酬と隠し状態を出力します。ここから、損失関数を導出し損失関数が最小になるように学習を行います。
MuZeroの性能
MuZeroの機能をテストするために、囲碁、チェス、将棋、Atariの4つのゲームを行いました。どのパフォーマンスでも、 MuZeroの性能の高さが表れています。
MuZeroのアルゴリズムの要約
for gameは、囲碁、チェス、将棋を指しています。
for ganeral MDPsはatariを指しています。
以下で使われているコードは1に付属している擬似コードから引用してます。
下図は、擬似コード内のエントリーポイント関数であるmuzeroから始まるclassの相互作用を表しています。
モデルフリーの場合は、学習と行動(self_paly)を繰り返し試行錯誤をします。
MuZeroの場合は、一気に行動(self_paly)を済ませ、その後学習を行います。
引用: MuZero: The Walkthrough (Part 1/3) | by David Foster | Applied Data Science | Medium
エントリーポイント関数であるmuzeroから全ての操作が始まります。
リプレイバッファ
ReplayBufferはgameのストレージクラスです。
SharedStorageはネットワークのストレージクラスです。
学習時に使用します。
window_sizeはReplayBufferに保存できる最大のgame数
batch_sizeは学習時に抽出できるgameの数
class ReplayBuffer:
def __init__(self, config: MuZeroConfig):
self.window_size = config.window_size
self.batch_size = config.batch_size
self.buffer = []
def save_game(self, game):
if len(self.buffer) > self.window_size:
self.buffer.pop(0)
self.buffer.append(game)
def sample_batch(self, num_unroll_steps: int, td_steps: int):
games = [self.sample_game() for _ in range(self.batch_size)]
game_pos = [(g, self.sample_position(g)) for g in games]
return [(g.make_image(i), g.history[i:i + num_unroll_steps],
g.make_target(i, num_unroll_steps, td_steps, g.to_play()))
for (g, i) in game_pos]
# バッファーから均一または優先度に応じてサンプルゲーム。
def sample_game(self) -> Game:
return self.buffer[0]
# ゲームから均一に、または優先度に応じて位置をサンプリングします。
def sample_position(self, game) -> int:
return -1
make_uniform_networkはnetworksを生成する関数
class SharedStorage:
def __init__(self):
self._networks = {}
def latest_network(self) -> Network:
if self._networks:
return self._networks[max(self._networks.keys())]
else:
# policy -> uniform, value -> 0, reward -> 0
return make_uniform_network()
def save_network(self, step: int, network: Network):
self._networks[step] = network
##### End Helpers ########
モデル
initial_inferenceは観測を入力として、検索初期のみに使用します。(表現関数と予測関数)
recurrent_inferenceは隠れ状態を入力として、検索中の抽象空間のみで使用します。(ダイナミック関数と予測関数)
training_stepsは、トレーニング回数に応じて、行動の選択確率を変更します。
S(y)=\frac{e^{\frac{y}{T}}}{\Sigma{e^{\frac{y}{T}}}}
atari環境では、以下のようにTemperatureが変更します。
training_steps < 500e3 → 1.0
training_steps < 750e3 → 0.5
training_steps >= 750e3 → 0.25
class Network:
# representation + prediction function
def initial_inference(self, image) -> NetworkOutput:
'''output: value, reward, policy_logits, hidden_state'''
return NetworkOutput(0, 0, {}, [])
# dynamics + prediction function
def recurrent_inference(self, hidden_state, action) -> NetworkOutput:
'''output: value, reward, policy_logits, hidden_state'''
return NetworkOutput(0, 0, {}, [])
# このネットワークの重みを返します。
def get_weights(self):
return []
# ネットワークが訓練されたステップ/バッチの数。
def training_steps(self) -> int:
return 0
self_play
self_playは、各セルフプレイジョブは他のすべてのジョブとは独立しています。最新のネットワークスナップショットを取得し、ゲームを作成し、共有リプレイバッファに書き込むことでトレーニングジョブで利用できるようにします。
def run_selfplay(config: MuZeroConfig,
storage: SharedStorage,
replay_buffer: ReplayBuffer):
while True:
network = storage.latest_network()
game = play_game(config, network)
replay_buffer.save_game(game)
play_gameでは、各ゲームは、最初のボードの位置から開始し、ゲームの終了に達するまで動きを生成するためにモンテカルロツリー検索を繰り返し実行することによって生成されます。
def play_game(config: MuZeroConfig, network: Network) -> Game:
game = config.new_game()
while not game.terminal() and len(game.history) < config.max_moves:
# 検索ツリーのルートでは、表現関数を使用して、
# 現在の観測値を指定して隠し状態を取得します。
root = Node(0)
current_observation = game.make_image(-1)
expand_node(root, game.to_play(), game.legal_actions(),
network.initial_inference(current_observation))
add_exploration_noise(config, root)
# 次に、アクションシーケンスとネットワークによって
# 学習されたモデルのみを使用してモンテカルロツリー検索を実行します。
run_mcts(config, root, game.action_history(), network)
action = select_action(config, len(game.history), root, network)
game.apply(action)
game.store_search_statistics(root)
return game
学習
train_networkは今まで、ReplayBufferに貯めたgameを使い学習を行います。
各gameのmake_targetから、target_value, target_reward, target_policyを生成します。
value, reward, policy_logitsはnetworkから生成します。
def train_network(config: MuZeroConfig, storage: SharedStorage,
replay_buffer: ReplayBuffer):
network = Network()
learning_rate = config.lr_init * config.lr_decay_rate**(
tf.train.get_global_step() / config.lr_decay_steps)
optimizer = tf.train.MomentumOptimizer(learning_rate, config.momentum)
for i in range(config.training_steps):
if i % config.checkpoint_interval == 0:
storage.save_network(i, network)
batch = replay_buffer.sample_batch(config.num_unroll_steps,
config.td_steps)
update_weights(optimizer, network, batch, config.weight_decay)
storage.save_network(config.training_steps, network)
def update_weights(optimizer: tf.train.Optimizer, network: Network, batch,
weight_decay: float):
loss = 0
for image, actions, targets in batch:
# Initial step, from the real observation.
value, reward, policy_logits, hidden_state = network.
initial_inference(
image)
predictions = [(1.0, value, reward, policy_logits)]
# Recurrent steps, from action and previous hidden state.
for action in actions:
value, reward, policy_logits, hidden_state = network.
recurrent_inference(hidden_state, action)
predictions.append((1.0 / len(actions), value,
reward, policy_logits))
hidden_state = tf.scale_gradient(hidden_state, 0.5)
for prediction, target in zip(predictions, targets):
gradient_scale, value, reward, policy_logits = prediction
target_value, target_reward, target_policy = target
l = (
scalar_loss(value, target_value) +
scalar_loss(reward, target_reward) +
tf.nn.softmax_cross_entropy_with_logits(
logits=policy_logits, labels=target_policy))
loss += tf.scale_gradient(l, gradient_scale)
for weights in network.get_weights():
loss += weight_decay * tf.nn.l2_loss(weights)
optimizer.minimize(loss)
売買ルール
1.空売りは認めない
2.ポジションを持っている場合、追加注文を出せない。
3.最後のステップでポジションを全て売却する。
4.ポジションは全買い、全売り
5.所持金は1000000ドル
実装と結果
ソースコードはこちら
学習環境
テスト環境
自信作です。
ソースコードはこちら
-
Schrittwieser, J., Antonoglou, I., Hubert, T., Simonyan, K., Sifre, L., Schmitt, S., Guez, A., Lockhart, E., Hass- abis, D., Graepel, T., Lillicrap, T., and Silver, D. Mas- tering atari, go, chess and shogi by planning with a learned model. 2020. ↩ ↩2 ↩3
-
Ioannis Antonoglou, Julian Schrittwieser1 Sherjil Ozair Thomas Hubert David Silver. PLANNING IN STOCHASTIC ENVIRONMENTS WITH A LEARNED MODEL. 2022. ↩