PythonでThreadPoolExecutorを使って並列処理による高速化

鼻の頭にポツポツと玉のような汗が吹き出る季節がやってきた。

ふと外の景色に目をやると、分厚い雨のカーテンにさえぎられてハッキリとは見えない。

このような梅雨の時期は、自然と戯れに出かけることができないから日記のネタが無い。

そもそも日記を書く気力も無い。誰も見てないからね。

そうなることは自ずと分かっていたので、

予め、1ヶ月分の日記を予め書いておいたよ！

……日記とは一体。

梅雨の季節なので室内のスポーツ施設で「卓球」を教えながら遊んでいるが、子供が上手になってきたので楽しくなってきた。

ThreadPoolExecutorを利用して並列処理する
その他のPythonエラー
1. Passthrough is not supported, GL is swiftshader
2. DevToolsのログを出力しない
まとめ

毎年毎年、後退しているのが

「体力」
「年収」

と

「株の自動売買」へのやる気だ。

とりあえず、まずはデイトレードから自動売買を開始しようと、SBI取引のシステムをSeleniumで作り中。

エキスパートが教えるSelenium最前線

戸田広, 島根義和, 高橋陽太郎, 沖田邦夫, 松尾和昭, 宮田淳平

1,584円(04/18 04:59時点)

発売日: 2018/05/15

Amazon 楽天市場

Amazonの情報を掲載しています

指値注文を出してしてしまうと資金が足りないので全シグナルに指値注文ができない問題がある。

このため、始値と購入予定額を比較して購入対象だと判断したら成行注文を出すシステムを構築したい。

3ヶ月前に、成行注文を出す予定の銘柄リストをメールで受け取るシステムまでは構築した。

PythonでProxyサーバ経由で購入銘柄情報をメール送信する（システムトレード）

今までのあらすじ。誰かが言った。人生の悲劇は2つしかない。ひとつは、金のない悲劇もうひとつは、金のある悲劇「ユダヤ人大富豪の教え」の著書・本田健氏は言った。お金持ちになるには5つの...

しかし

処理が遅い

どれくらい遅いって、9時に市場がオープンして、処理完了のメールが届くのが9時50分だからね。

ちょ、もはや始値じゃ無いじゃんｗｗ

遅い理由は、

約420件のシグナルに対して、一つ一つの銘柄の始値を比較しているから。

これ、早くしないと駄目だわ……。

ThreadPoolExecutorを利用して並列処理する

高速で動かすには並列処理を行えば良い。

Python 3.2から追加されたconcurrent.futuresモジュール。

ThreadPoolExecutorを使うと、スレッドプールで処理を実行できる。

Pythonには他にthreadingとmultiprocessingというモジュールがあり、これらが１つのスレッド・プロセスを扱うのに対して、concurrent.futuresモジュールは複数のスレッド・プロセスを扱うことを目的としている。

そして、このconcurrent.futuresモジュールには抽象クラスとしてExecutorクラスがあり、実装クラスとして次の2つのクラスが提供されている。

クラス	説明
ThreadPoolExecutor	スレッドを使って並列タスクを実行します。ネットワークアクセスなどCPUに負荷がかからない処理の並列実行に適しています。
ProcessPoolExecutor	プロセスを使って並列タスクを実行します。CPUに負荷がかかる計算処理などの並列実行に適しています。

難しい事は抜きにして、とりあえず株の売買で使うなら次のような書き方になる。

import os
import glob
import time
import shutil
from concurrent import futures
import functools
print = functools.partial(print, flush=True)
import time
from queue import Queue
import threading
LOCK = threading.Lock()

count = 0
q = Queue()

# 銘柄選定
def exec1(codes):
    rev_list = []
    print("[1]" + codes)
    # 始値と期待値の比較
    # ＜処理＞
    result = 1
    # ここまで
    if (result):
        q.put([codes, 100, "credit_buy", "LARGE"])
        rev_list.append(codes)
    return True, rev_list

# 購入処理
def exec2():
    global count
    while True:
       with LOCK:
          if q.empty():
             break
          else:
             [i, quantity, method, func] = q.get()
             # 購入処理
             # ＜処理＞
             # ここまで
             print("[2]" + str(count) + " : " + str(i))
             time.sleep(1.5)
             count += 1

def func(codes):
    rev_list = []
    # スレッドプール
    with futures.ThreadPoolExecutor(max_workers=8) as executor:
         # 1) executor.submit で逐次スレッド化
         for i in range(len(codes)):
             future1 = executor.submit(exec1, codes[i])
             future2 = executor.submit(exec2)
             rev, tmp_list = future1.result()
             if (len(tmp_list)):
                 rev_list.append(tmp_list)
    print(rev_list)

if __name__ == "__main__":
     with open("code.txt") as f:
          codes = f.read().splitlines()
     start = time.time()
     func(codes)
     process_time = time.time() - start
     print(process_time)

import os

import glob

import time

import shutil

from concurrent import futures

import functools

print = functools.partial(print, flush=True)

import time

from queue import Queue

import threading

LOCK = threading.Lock()

count = 0

q = Queue()

# 銘柄選定

def exec1(codes):

rev_list = []

print("[1]" + codes)

# 始値と期待値の比較

# ＜処理＞

result = 1

# ここまで

if (result):

q.put([codes, 100, "credit_buy", "LARGE"])

rev_list.append(codes)

return True, rev_list

# 購入処理

def exec2():

global count

while True:

with LOCK:

if q.empty():

break

else:

[i, quantity, method, func] = q.get()

# 購入処理

# ＜処理＞

# ここまで

print("[2]" + str(count) + " : " + str(i))

time.sleep(1.5)

count += 1

def func(codes):

rev_list = []

# スレッドプール

with futures.ThreadPoolExecutor(max_workers=8) as executor:

# 1) executor.submit で逐次スレッド化

for i in range(len(codes)):

future1 = executor.submit(exec1, codes[i])

future2 = executor.submit(exec2)

rev, tmp_list = future1.result()

if (len(tmp_list)):

rev_list.append(tmp_list)

print(rev_list)

if __name__ == "__main__":

with open("code.txt") as f:

codes = f.read().splitlines()

start = time.time()

func(codes)

process_time = time.time() - start

print(process_time)

一方のスレッドで規制価格かを確認し、もう一つのスレッドで売買を実施する。

Executorはwith文と組み合わせて使うことができ、そうするとwith文を抜けるとき勝手にshutdownしてくれる。便利だ。

「code.txt」は銘柄コードの一覧。予め Protra でバックテストした明日購入候補の一覧だ。

1001

1301

1305

1306

1308

1309

....

で、実行結果は次のようになる。

[1]1001
[1]1301
[1]1305
[1]1306
[2]0 : 1306
[1]1308
[1]1309
[1]1311
[1]1312
[1]1313
[1]1319
[1]1320
[2]1 : 1308
[2]2 : 1312
....
[['1306'], ['1308'], ['1312'], ['1320'], ['1322'], ['1324'], ['1326'], ['1328'], ['1330'], ['1332'], ['1344'], ['1346'], ['1348'], ['1352'], ['1356'], ['1358']]
24.029504537582397

[1]1001

[1]1301

[1]1305

[1]1306

[2]0 : 1306

[1]1308

[1]1309

[1]1311

[1]1312

[1]1313

[1]1319

[1]1320

[2]1 : 1308

[2]2 : 1312

....

[['1306'], ['1308'], ['1312'], ['1320'], ['1322'], ['1324'], ['1326'], ['1328'], ['1330'], ['1332'], ['1344'], ['1346'], ['1348'], ['1352'], ['1356'], ['1358']]

24.029504537582397

多分、これで大丈夫。

concurrent.futuresは以前から使っていたけど、スレッド処理というよりLOCK処理とQueueを使って取得した銘柄を受け渡し処理するのが大事。

なお、SBI証券は購入処理は1.5秒単位で行う必要があるらしい……ので注意を。

その他のPythonエラー

折角だから、色々とChromedriverを使ったselenium処理でWarningが出力されていたので、そこも修正する。

Passthrough is not supported, GL is swiftshader

ERROR:gpu_init.cc(440)] Passthrough is not supported, GL is swiftshader

1	ERROR:gpu_init.cc(440)] Passthrough is not supported, GL is swiftshader

というエラーが出ていた。

まんま「stackoverflow」に修正方法が書いてあった。

Passthrough is not supported, GL is disabled

I tried crawling a specific site using selenium and webdriver_manager.chrome, and my code crawled elements of that site totally. But after crawling, the followi...

ChromeでWebGLが使えない場合の対処法「–disable-software-rasterizer」をオプションとして追加。

        options = webdriver.ChromeOptions()
        options.add_argument('--headless')
        options.add_argument('--disable-gpu')
        options.add_argument('--disable-software-rasterizer')

options = webdriver.ChromeOptions()

options.add_argument('--headless')

options.add_argument('--disable-gpu')

options.add_argument('--disable-software-rasterizer')

うん。ログが出なくなった。

DevToolsのログを出力しない

さらに、次のようなログが表示され続けている。

DevTools listening on ws://127.0.0.1:59540/devtools/browser/*****

1	DevTools listening on ws://127.0.0.1:59540/devtools/browser/*****

これも「stackoverflow」を見ると書いてあった。

Python selenium: DevTools listening on ws://127.0.0.1

Today I got this message on the console when running selenium using the chromedriver. How do I suppress this? DevTools listening on ws://127.0.0.1:12740/devtoo...

要するに「enable-logging」オプション追加すると表示されなくなる。

全部合わせると次のように記載すればよい。

        options = webdriver.ChromeOptions()
        # ヘッドレスモードで起動
        options.add_argument('--headless')
        # Passthrough is not supportedのログを出力しない
        options.add_argument('--disable-gpu')
        options.add_argument('--disable-software-rasterizer')
        options.add_argument('--log-level=3')
        options.add_argument('--no-sandbox') # Bypass OS security model
        # enable-logging：DevToolsのログを出力しない
        options.add_experimental_option('excludeSwitches', ['enable-automation', 'enable-logging'])
        driver = webdriver.Chrome('../SBIcomm/chromedriver.exe', options=options)
        driver.get(url)

options = webdriver.ChromeOptions()

# ヘッドレスモードで起動

options.add_argument('--headless')

# Passthrough is not supportedのログを出力しない

options.add_argument('--disable-gpu')

options.add_argument('--disable-software-rasterizer')

options.add_argument('--log-level=3')

options.add_argument('--no-sandbox') # Bypass OS security model

# enable-logging：DevToolsのログを出力しない

options.add_experimental_option('excludeSwitches', ['enable-automation', 'enable-logging'])

driver = webdriver.Chrome('../SBIcomm/chromedriver.exe', options=options)

driver.get(url)

インターネットが発達して悩み事を共有できるようになってありがたいなぁ。

まとめ

実行時間は「50分」だったものが「12分」まで早くなった。

これなら実用に耐えられる。

とりあえず暫く不具合が発生しないか走らせてみて、少しずつ購入処理でも実装してみるかーー。

適切なストラテジーが無いんだよなーーー。