書いたループをコンパイルする
計算の中には、指数加重平均、ドローダウン、経路依存オプションのペイオフのように、各ステップが前のステップに依存する、本質的にループのものがあります。NumPy ではベクトル化できず、100万要素の Python ループは、各ステップがインタプリタを通るため遅くなります。Numba はジャストインタイムコンパイラです。関数の上に @njit を付けると、最初の呼び出し時に、渡された型に合わせて Python のループが機械語に翻訳されます。以降の呼び出しは C に近い速度で動き、コードを変えずに、素の Python の同じループより50〜100倍速くなることもよくあります。
# An exponentially weighted average is a loop: each value depends on the one before
import numpy as np
try:
from numba import njit # one decorator compiles the function when it is installed
except ImportError:
def njit(func): # without Numba the same code simply runs as Python
return func
@njit
def ewma(values, alpha):
out = np.empty(len(values))
out[0] = values[0]
for i in range(1, len(values)):
out[i] = alpha * values[i] + (1.0 - alpha) * out[i - 1]
return out
print(ewma(np.array([1.0, 2.0, 3.0]), 0.5)) # [1. 1.5 2.25]Numba が好むスタイル
@njit が選ぶ nopython モードでは、Numba は Python のサブセットだけをコンパイルし、それ以外は黙って遅いコードに戻るのではなく拒否します。数値、NumPy 配列、タプル、単純なループ、if、NumPy の関数の多くを理解します。pandas、任意の Python オブジェクト、型が混在する辞書は理解しません。そのためホットな関数は素朴なスタイルで書きます。配列とスカラーを受け取り、出力を np.empty や np.zeros で確保し、for i in range(n) でループし、スカラーの累積変数（total、peak）を持ちます。毎回同じ型の引数で呼び出してください。型の組み合わせが変わると新たなコンパイルが起きます。
# A drawdown is a running peak and a running worst case: two scalars and one loop
import numpy as np
try:
from numba import njit
except ImportError:
def njit(func):
return func
@njit
def max_drawdown(equity):
peak = 0.0
worst = 0.0
for i in range(len(equity)):
value = equity[i]
if value > peak:
peak = value
drawdown = (peak - value) / peak
if drawdown > worst:
worst = drawdown
return worst
curve = np.array([100.0, 120.0, 90.0, 130.0, 117.0])
print(max_drawdown(curve)) # 0.25 from 120 down to 90
print(max_drawdown(np.array([1.0, 2.0, 3.0]))) # 0.0 a curve that only risesコンパイル時間、キャッシュ、Numba が役立たない場面
最初の呼び出しでコンパイル代を払います（1秒以上かかることもあります）。そのため Numba 関数は、小さなデータで一度呼び出してから計測し、@njit(cache=True) で実行間にコンパイル済みコードを保持します。高速化には十分な仕事量が必要です。3つの数値に対して100万回呼ばれる関数は、ループではなく呼び出しのオーバーヘッドに時間を使います。Numba が助けるのは配列に対するループです。すでに NumPy の1回の呼び出しであるコード、データベースの待ち時間、悪いアルゴリズムには何もしません。@njit(parallel=True) と prange で独立した反復をコアに分散でき、fastmath=True は浮動小数の厳密さを少し犠牲にして速度を得ます。結果が許容誤差内に収まると確認してから有効にしましょう。
# Counting crossings: Numba compiles any loop like this, so keep the logic in scalars
import numpy as np
try:
from numba import njit
except ImportError:
def njit(func):
return func
@njit(cache=False)
def count_crossovers(fast, slow):
count = 0
previous = 0.0
for i in range(len(fast)):
gap = fast[i] - slow[i]
if i > 0 and previous * gap < 0:
count += 1
previous = gap
return count
fast = np.array([1.0, 3.0, 2.0, 1.0, 4.0])
slow = np.full(5, 2.0)
print(count_crossovers(fast, slow)) # 2 below, above, touch, below, above: two strict sign changes# Options change how a function is compiled; here they are ignored if Numba is not installed
import numpy as np
try:
from numba import njit, prange
except ImportError:
prange = range
def njit(*args, **kwargs):
if args and callable(args[0]):
return args[0]
return lambda func: func
@njit(cache=True) # keep the compiled code on disk between runs
def total_notional(prices, quantities):
total = 0.0
for i in prange(len(prices)): # prange runs in parallel only with parallel=True
total += prices[i] * quantities[i]
return total
prices = np.array([184.25, 410.5, 99.0])
quantities = np.array([100.0, 50.0, 200.0])
print(total_notional(prices, quantities)) # 58750.0
print(total_notional(prices, quantities) == float(prices @ quantities)) # True same answer as the vectorised dot productfn.py_func を使います）。これらのラボのテストがしているとおりです。開発中は境界チェックを有効にします。@njitは最初の呼び出しで Python 関数を機械語にコンパイルする。前のステップに依存するループもベクトル化なしで速くなる- nopython モード向けに書く。配列とスカラー、事前に確保した出力、
for i in range(n)、スカラーの累積変数。pandas や型混在のコンテナは使わない - ウォームアップ後に計測し、
cache=Trueを使う。すでに NumPy 1回の呼び出しのコードには効果を期待せず、結果は素の Python 版と照合する
次は Cython です。型を追加し、拡張モジュールをビルドして Python のループをコンパイルするもう1つの方法です。
ラボ：採点される内容
rolling_meanで累計を持ち回る25点max_drawdownでピークを追跡する25点count_crossoversで交差を数える25点ewmaで出力配列を埋める25点