Numba

ترجمة الحلقات التي كتبتها

بعض الحسابات حلقة بطبيعتها: كل خطوة تعتمد على السابقة كما في المتوسط الموزون أسّيًا أو السحب أو عائد خيار يعتمد على المسار. ولا يستطيع NumPy توجيهها، وحلقة بايثون على مليون عنصر بطيئة لأن كل خطوة تمر بالمفسّر. وNumba مترجم في وقت التشغيل: تضع @njit فوق دالة وعند أول نداء يترجم Numba حلقة بايثون إلى شيفرة آلة للأنواع التي أُعطيت بالضبط. وتجري النداءات اللاحقة بسرعة قريبة من C، غالبًا أسرع بخمسين إلى مئة مرة من الحلقة نفسها في بايثون الصرفة، دون تغيير الشيفرة.

# 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

في وضع nopython الذي يختاره @njit يترجم Numba مجموعة جزئية من بايثون فقط ويرفض الباقي بدل الرجوع بصمت إلى شيفرة بطيئة. وهو يفهم الأعداد ومصفوفات NumPy والـ tuples والحلقات البسيطة وif وجزءًا كبيرًا من دوال NumPy. ولا يفهم pandas ولا كائنات بايثون العشوائية ولا القواميس بأنواع مختلطة. فاكتب الدالة الساخنة بالأسلوب البسيط: تأخذ مصفوفات وأعدادًا مفردة، وتحجز الناتج بـ 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

يدفع النداء الأول ثمن الترجمة، غالبًا ثانية أو أكثر، فقِس زمن دالة Numba بعد استدعائها مرة على بيانات صغيرة، واستعمل @njit(cache=True) لحفظ الشيفرة المترجمة بين التشغيلات. ويحتاج التسريع عملًا كافيًا: دالة تُستدعى مليون مرة على ثلاثة أرقام تقضي وقتها في كلفة النداء لا في الحلقة. ويفيد Numba في الحلقات على المصفوفات؛ ولا يفعل شيئًا لشيفرة هي أصلًا نداء NumPy واحد ولا لانتظار قاعدة بيانات ولا لخوارزمية سيئة. ويمكن لـ @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 product

• @njit يترجم دالة بايثون إلى شيفرة آلة عند أول نداء؛ وتصير الحلقات المعتمدة على الخطوات السابقة سريعة دون توجيه.
• اكتب لوضع nopython: مصفوفات وأعداد مفردة وناتج محجوز مسبقًا وfor i in range(n) ومراكم عددية؛ دون pandas أو حاويات بأنواع مختلطة.
• قِس الزمن بعد نداء تسخين واستعمل cache=True ولا تتوقع مكسبًا في شيفرة هي أصلًا نداء NumPy واحد، وتحقق من النتائج بنسخة بايثون العادية.

التالي: Cython، طريقة أخرى لترجمة حلقات بايثون بإضافة الأنواع وبناء وحدة امتداد.

المختبر: ما الذي يُقيَّم

• احتفظ بمجموع جارٍ في rolling_mean25 نقطة
• تتبّع القمة في max_drawdown25 نقطة
• عُدّ التقاطعات في count_crossovers25 نقطة
• املأ مصفوفة ناتج في ewma25 نقطة