同じ仕事を速くするのではなく、仕事を減らす
プロファイラがホットループを指したら、最大の効果は同じ仕事を絞り出すことではなく、仕事の量を変えることから得られます。100万の位置のそれぞれで250個の価格の窓を足し合わせるループは2億5000万回の加算をしますが、窓をスライドさせる版は200万回で済みます。この違い、つまり計算量は、内側の1行の工夫など比べものになりません。まず、ループが何を繰り返しているかを見ましょう。リストの検索、ゼロから再計算される合計、変わらないのに毎回作り直される値などです。
# Membership in a list scans it; in a set it is a hash lookup. Same answer, very different work
import time
ids = list(range(20_000))
as_set = set(ids)
wanted = list(range(19_000, 20_000))
start = time.perf_counter()
found_list = sum(1 for x in wanted if x in ids) # scans up to 20,000 items for each of 1,000 lookups
slow = time.perf_counter() - start
start = time.perf_counter()
found_set = sum(1 for x in wanted if x in as_set)
fast = time.perf_counter() - start
print(found_list, found_set) # 1000 1000
print(slow > 20 * fast) # True the set is at least twenty times faster hereスライド、一度だけ検索、ループの外へ
ほとんどのホットループは3つのパターンで片付きます。再計算せずにスライドする:移動和は窓に入る値を足して出ていく値を引くので、窓の長さにかかわらず各位置の費用は2回の演算です。検索を一度だけ作る:ペアのリストを dict にするのは1回の走査で済み、その後の検索は一瞬です。クエリのたびにリストを走査する必要はありません。変わらないものはループの外へ:ループ内で計算していても外の値にしか依存しないもの(定数、len(x)、属性の参照)はループの前に出します。このラボのように、読み取りや演算の回数のカウンターでテストしましょう。回数はマシンの速さに依存しないので、仕事量の安定した尺度になります。
# A sliding sum does two operations per position; recomputing each window does `window` of them
def moving_sum_slow(values, window):
operations = 0
sums = []
for i in range(len(values) - window + 1):
total = 0
for v in values[i : i + window]:
total += v
operations += 1
sums.append(total)
return sums, operations
def moving_sum_fast(values, window):
operations = window
total = sum(values[:window])
sums = [total]
for i in range(window, len(values)):
total += values[i] - values[i - window]
operations += 2
sums.append(total)
return sums, operations
values = list(range(1_000))
slow, slow_ops = moving_sum_slow(values, 250)
fast, fast_ops = moving_sum_fast(values, 250)
print(slow == fast) # True the same answers
print(slow_ops, fast_ops) # 187750 1750 about a hundred times less work# Hoisting: a value that never changes is computed once, outside the loop
import math
calls = {"count": 0}
def annualisation_factor(): # imagine something costly that never changes
calls["count"] += 1
return math.sqrt(252)
def scale_slow(returns):
return [r * annualisation_factor() for r in returns]
def scale_fast(returns):
factor = annualisation_factor() # hoisted out of the loop
return [r * factor for r in returns]
returns = [0.001 * i for i in range(2_000)]
slow = scale_slow(returns)
slow_calls, calls["count"] = calls["count"], 0
fast = scale_fast(returns)
print(slow == fast) # True the same numbers
print(slow_calls, calls["count"]) # 2000 1 one call instead of two thousandループは NumPy に任せる
仕事が本当に多数の数値に対する算術なら、ループを Python の外へ出します。NumPy 配列は数値を1つのメモリブロックに格納し、コンパイル済みのコードですべてに演算を適用するので、np.log(prices[1:] / prices[:-1]) は Python のループを、通常は数十倍速い1回の呼び出しに置き換えます。これがベクトル化で、数値計算のコードで最初に試す最適化です。要素ごとの if がない配列全体の処理で最も効果が出ます。条件が必要なときは np.where やブールマスクを使います。メモリにも注意してください。ベクトル化はデータと同じ大きさの一時配列を作ります。
# Vectorising: one NumPy call replaces a Python loop and gives the same numbers
import math
import time
import numpy as np
rng = np.random.default_rng(3)
prices = 100 * np.exp(np.cumsum(rng.normal(0, 0.001, size=300_000)))
start = time.perf_counter()
loop = [math.log(prices[i] / prices[i - 1]) for i in range(1, len(prices))]
loop_time = time.perf_counter() - start
start = time.perf_counter()
vector = np.log(prices[1:] / prices[:-1])
vector_time = time.perf_counter() - start
print(np.allclose(loop, vector)) # True the same returns
print(loop_time > 5 * vector_time) # True at least five times faster here- まず仕事の量を減らす。窓は足し直さずスライドさせ、すでにあるものを再計算しない
- 適切な構造を使う。検索には(一度だけ作った)集合や dict を使い、クエリごとにリストを走査しない。定数はループの外へ
- 数値ループは NumPy でベクトル化し、テストでは秒数でなく仕事量を数え、速い版を遅くて明らかに正しい版と照らし続ける
次は Numba です。デコレータで Python の数値ループを機械語にコンパイルします。
ラボ:採点される内容
moving_sumで窓をスライドさせる25点has_duplicatesで集合を使う25点lookup_pricesで辞書を一度だけ作る25点log_returnsをベクトル化する25点