スレッド：1つのプロセス内の並行処理
スレッドは、1つのプログラム内の独立した実行の流れで、他のスレッドとメモリを共有します。10か所の市場から気配を取得し、ログを書き、画面の応答性も保つ必要があるトレーディングプログラムは、1つを待っても残りが止まらないようにスレッドを使います。Python の threading モジュールと、より便利な concurrent.futures.ThreadPoolExecutor がスレッドを起動・管理します。スレッドは変数を共有するため、通信は速い一方で、間違えると危険です。
# Ten slow "network calls" in a thread pool wait at the same time, so the total is about one call, not ten
import time
from concurrent.futures import ThreadPoolExecutor
def fetch_quote(symbol):
time.sleep(0.1) # stands in for waiting on a venue
return symbol, len(symbol) * 10.0
symbols = ["AAPL", "MSFT", "NVDA", "TSLA", "AMZN", "META", "GOOG", "NFLX", "ORCL", "INTC"]
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=10) as pool:
quotes = list(pool.map(fetch_quote, symbols))
elapsed = time.perf_counter() - start
print(quotes[0], len(quotes)) # ('AAPL', 40.0) 10
print(elapsed < 0.5) # True about 0.1 s in total, not 10 x 0.1 sGIL：待つスレッドと計算するスレッド
CPython にはグローバルインタプリタロック（GIL）があり、どの瞬間も Python のバイトコードを実行できるスレッドは1つだけです。スレッドはネットワーク、ファイル、time.sleep を待つ間はこれを手放すので、待ち時間は非常にうまく重なります。気配を取得する10スレッドは、ほぼ1スレッド分の時間で終わります。しかしCPU バウンドの Python コード（算術をする純粋な Python のループ）ではスレッドが交代で動くため、4スレッドは1スレッドより速くならず、むしろ少し遅くなることが多いです。NumPy や GIL を解放する C 拡張のように Python の外で動くコードは、スレッドで並列に動けます。純粋な Python の重い計算には、別プロセス（次のトピック）か、利用できるなら GIL をなくした Python 3.13 以降のフリースレッド版を使います。
# Splitting a list into chunks for workers: sizes differ by at most one, the first chunks get the extra items
def split_chunks(items, workers):
size, extra = divmod(len(items), workers)
chunks, start = [], 0
for index in range(workers):
stop = start + size + (1 if index < extra else 0)
chunks.append(list(items[start:stop]))
start = stop
return chunks
chunks = split_chunks(list(range(7)), 3)
print(chunks) # [[0, 1, 2], [3, 4], [5, 6]]
print([sum(chunk) for chunk in chunks]) # [3, 7, 11] one partial sum per worker
print(sum(sum(chunk) for chunk in chunks)) # 21共有データ、競合状態、ロック
同じ変数を更新するスレッドは、その値を壊すことがあります。total += amount は3つのステップ（読む、足す、書く）で、その間にスレッドの切り替えが起こりうるからです。2つのスレッドがどちらも100を読み、どちらも5を足し、どちらも105を書くと、更新が1つ失われます。これが競合状態（race condition）で、まれに予測できない形で現れるため、最悪の種類のバグです。対策はロックです。with lock: はブロックを一度に1つのスレッドだけに実行させ、読み取りと書き込みの両方をその中に入れる必要があります。さらに良いのは共有を避けることです。各ワーカーに自分の担当分と結果を持たせ、最後に結果を合わせるか、すでにスレッドセーフな queue.Queue で仕事と結果をやり取りします。
# A lock makes the read-modify-write a single step; the total comes out exact
import threading
totals = {"AAPL": 0}
lock = threading.Lock()
def safe_add(shared, key, amount, lock):
with lock:
shared[key] = shared.get(key, 0) + amount
return shared[key]
def worker():
for _ in range(10_000):
safe_add(totals, "AAPL", 1, lock)
threads = [threading.Thread(target=worker) for _ in range(4)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
print(totals["AAPL"]) # 40000 nothing lost: four threads x 10,000 additions# A back-of-envelope estimate before writing any threaded code
import math
def estimate_time(n_tasks, seconds, workers, kind):
if kind in ("io", "processes"):
return math.ceil(n_tasks / workers) * seconds
return n_tasks * seconds # "cpu_threads": the GIL makes the threads take turns
print(estimate_time(40, 0.5, 8, "io")) # 2.5 forty slow requests, eight at a time
print(estimate_time(40, 0.5, 8, "cpu_threads")) # 20.0 forty CPU-bound tasks in threads: no gain
print(estimate_time(40, 0.5, 8, "processes")) # 2.5 separate processes use separate coresasyncio のほうが通常はスケールし、await 間の競合もありません。配列の算術には NumPy でベクトル化し、CPU バウンドの Python にはプロセスを使います。スレッド間でデータを共有するなら、ロックする区間は短くし、ロックを持ったまま遅いコードを呼ばず、複数のロックは常に同じ順で取ります。そうしないと2つのスレッドが永遠に互いを待つことになります（デッドロック）。- スレッドはメモリを共有し、待ち時間（ネットワーク、ファイル）をうまく重ねる。1つの
ThreadPoolExecutorで多数の市場から同時に取得できる - GIL により Python のバイトコードは一度に1スレッドだけが実行する。CPU バウンドの純粋な Python は速くならないが、NumPy や C 拡張はこれを解放できる
- 共有の更新は読み取りと書き込みを
with lock:で囲んで守り、なるべく共有しない（チャンク、キュー）ようにし、デッドロックに注意する
次はマルチプロセッシングです。複数のプロセスで CPU の全コアを働かせます。
ラボ：採点される内容
split_chunksで仕事を分配する25点run_in_chunksでチャンクに関数を適用する25点safe_addで共有データを守る25点estimate_timeでスレッドの効果を見積もる25点