並列分散処理を用いたシステムトレーディング

はじめに

agentの行動として、 BUY、HOLD、SELLの三つの内一つを選択する。環境の戻り値として、状態（今現在保有しているポジションの価格、市場価格、手持ちのキャッシュ）、報酬（手持ちのキャッシュの変化値（含む益も含む））、終了（取引の終了か否か）、情報（ターミナルにディスプレイする情報）を返す。

マルチスレッドとマルチプロセス

マルチスレッドの特徴は以下の通り
・平行処理
・GILロックを持っているスレッドのみ、実行可能、他のスレッドは待機する。
・cpuのコア数に依存しない。

マルチプロセスの特徴は以下の通り
・並列処理
・メモリーが共有されていないので、値の受け渡しが大変。
・tensowflowだと、modelが動かないことがある。
・cpuのコア一つにプロセスを割り当てする。

cpuバウンドとはcpu内で行なっている数値計算などの処理

ソースコードはこちら

使用データについて

トレンド傾向の掴みやすさから、yahoo financeからGSPCの日足を使用した。

訓練データの期間：2015/1/1 - 2017/6/30
テストデータの期間：2017/7/1 - 2021/1/1

以下ソースコード

ループ

class Task:
    def __init__(self, name):
        self.name = name
        self.num = 0
        self.sum = 0.0

    def run(self):
        while True:
            sleep(1)
            if self.num == 3:
                break
            print('roop_name  = ' + self.name + '  :num ＝ '+ str(self.num) + '  :sum ＝ '+ str(self.sum))
            self.num += 1
            self.sum += random.random()


name = 'nomal-roop'
start = time.time()

for i in range(4):
    w = Task('roop_point_'+str(i))
    w.run()

end = time.time() - start
arr.append("name:" + name + " process_time:{0}".format(end) + "[s]")

roop_name  = roop_point_0  :num ＝ 0  :sum ＝ 0.0
roop_name  = roop_point_0  :num ＝ 1  :sum ＝ 0.642469181212962
roop_name  = roop_point_0  :num ＝ 2  :sum ＝ 1.5964812171373977
roop_name  = roop_point_1  :num ＝ 0  :sum ＝ 0.0
roop_name  = roop_point_1  :num ＝ 1  :sum ＝ 0.8876820994429431
roop_name  = roop_point_1  :num ＝ 2  :sum ＝ 1.627826300716026
roop_name  = roop_point_2  :num ＝ 0  :sum ＝ 0.0
roop_name  = roop_point_2  :num ＝ 1  :sum ＝ 0.03546302344611851
roop_name  = roop_point_2  :num ＝ 2  :sum ＝ 1.0239282875765587
roop_name  = roop_point_3  :num ＝ 0  :sum ＝ 0.0
roop_name  = roop_point_3  :num ＝ 1  :sum ＝ 0.602393530385244
roop_name  = roop_point_3  :num ＝ 2  :sum ＝ 1.555539488491399

マルチスレッド

class Task:
    def __init__(self, name):
        self.name = name
        self.num = 0
        self.sum = 0.0

    def run(self):
        while True:
            sleep(1)
            if self.num == 3:
                break
            print('roop_name  = ' + self.name + '  :num ＝ ' + str(self.num) + '  :sum ＝ ' + str(self.sum))
            self.num += 1
            self.sum += random.random()

name = 'thread-pool'
start = time.time()

thread_num = 4
threads = []
for i in range(thread_num):
    threads.append(Task(name=f'thread_{i}'))

datas = []
with ThreadPoolExecutor(max_workers = thread_num) as executor:
    for task in threads:
        job = lambda: task.run()
        datas.append(executor.submit(job))

end = time.time() - start
arr.append("name:" + name + " process_time:{0}".format(end) + "[s]")

roop_name  = thread_0  :num ＝ 0  :sum ＝ 0.0
roop_name  = thread_1  :num ＝ 0  :sum ＝ 0.0
roop_name  = thread_2  :num ＝ 0  :sum ＝ 0.0
roop_name  = thread_3  :num ＝ 0  :sum ＝ 0.0
roop_name  = thread_0  :num ＝ 1  :sum ＝ 0.7829927782861958
roop_name  = thread_2  :num ＝ 1  :sum ＝ 0.7264674393557742
roop_name  = thread_1  :num ＝ 1  :sum ＝ 0.4721450639806136
roop_name  = thread_3  :num ＝ 1  :sum ＝ 0.2746835685320669
roop_name  = thread_0  :num ＝ 2  :sum ＝ 0.8189509274906515
roop_name  = thread_1  :num ＝ 2  :sum ＝ 0.7522106668563098
roop_name  = thread_2  :num ＝ 2  :sum ＝ 1.3346477522815392
roop_name  = thread_3  :num ＝ 2  :sum ＝ 0.33216049073474685

関数のidについて

job = lambda: task.run()でラムダ式にさせないと、ただのループになってしまう。
このことについて調べた。

def double(n):
    return n * 2

for _ in range(5):
    lambda_ver = lambda: double(2)
    print(id(lambda_ver))

print(' ')

for _ in range(5):
    lambda_ver = double(2)
    print(id(lambda_ver))

140396010246352
140396010249808
140396009553344
140396009595376
140396009595952

140395963095440
140395963095440
140395963095440
140395963095440
140395963095440

と言うふうにlambda式を使わないと同一のオブジェクトidになってしまうことがわかる。
つまり、同一のオブジェクトをループさせていると言うことになっている。

マルチプロセス

class Task:
    def __init__(self, name):
        self.name = name
        self.num = 0
        self.sum = 0.0

    def run(self):
        while True:
            sleep(1)
            if self.num == 3:
                break
            print('roop_name  = ' + self.name + '  :num ＝ ' + str(self.num) + '  :sum ＝ ' + str(self.sum))
            self.num += 1
            self.sum += random.random()

name = 'multi-process'
start = time.time()

p_inst = [Task('work1'),Task('work2'),Task('work3'),Task('work4')]
Task = []
for i in p_inst:
    p = Process(target=i.run,)
    Task.append(p)
    p.start()

for w in Task:
    w.join()

end = time.time() - start
arr.append("name:" + name + " process_time:{0}".format(end) + "[s]")

roop_name  = work1  :num ＝ 0  :sum ＝ 0.0
roop_name  = work2  :num ＝ 0  :sum ＝ 0.0
roop_name  = work3  :num ＝ 0  :sum ＝ 0.0
roop_name  = work4  :num ＝ 0  :sum ＝ 0.0
roop_name  = work1  :num ＝ 1  :sum ＝ 0.32601928337976416
roop_name  = work2  :num ＝ 1  :sum ＝ 0.394125272204221
roop_name  = work3  :num ＝ 1  :sum ＝ 0.6268785557359723
roop_name  = work4  :num ＝ 1  :sum ＝ 0.09698983618263579
roop_name  = work1  :num ＝ 2  :sum ＝ 0.5691443562637948
roop_name  = work2  :num ＝ 2  :sum ＝ 0.6206409805829249
roop_name  = work3  :num ＝ 2  :sum ＝ 0.821770663159113
roop_name  = work4  :num ＝ 2  :sum ＝ 0.6171283310993135

実行時間

name:nomal-roop 　 process_time:16.021376371383667[s]
name:multi-thread  process_time:4.008574485778809[s]
name:multi-process process_time:4.034740686416626[s]

実装と結果

下記の図より、マルチプロセスとマルチスレッドの収益、勝率で大きな違いがないことがわかる。

マルチプロセスのソースコードはこちら

マルチスレッドのソースコードはこちら