Data Handling and Visualization › Performance Optimization

# Extract NumPy arrays from DataFrame
prices_array = multi_prices.values
dates_array = multi_prices.index.values

# Vectorized calculation avoids a Python-level element loop
returns_numpy = np.diff(prices_array, axis=0) / prices_array[:-1]

from numba import jit


@jit(nopython=True)
def calculate_drawdowns(prices):
    r"""Calculate drawdown series from price array.

    Drawdown measures the percentage decline from a historical peak. At time $t$, the drawdown is computed as:

    $$
    DD_t = \frac{P_t - \text{Peak}_t}{\text{Peak}_t}
    $$

    where:

    - $DD_t$: drawdown at time $t$ (a non-positive value where more negative numbers indicate larger percentage declines from the peak)
    - $P_t$: current price at time $t$
    - $\text{Peak}_t$: maximum price observed from the start of the series up to and including time $t$ (where $\text{Peak}_t = \max_{s \leq t} P_s$)

    The drawdown is zero when the price equals the within-sample running high, possibly tying an earlier high, and becomes negative when the price falls below the peak. More negative values indicate larger declines. For example, a drawdown of $-0.20$ means the current price is 20% below its historical high. If the peak was $100 and the current price is $80: the drawdown is $(80-100)/100 = -0.20$.
    """
    # Float64 arithmetic prevents unsigned integer subtraction from wrapping.
    prices = prices.astype(np.float64)
    n = len(prices)
    if n == 0:
        raise ValueError("prices must be nonempty")
    for price in prices:
        if not np.isfinite(price) or price <= 0:
            raise ValueError("prices must contain only finite positive values")
    drawdowns = np.empty(n)
    peak = prices[0]

    for i in range(n):
        if prices[i] > peak:
            peak = prices[i]
        drawdowns[i] = (prices[i] - peak) / peak

    return drawdowns


# Apply to our price data
aapl_prices = multi_prices["AAPL"].values
drawdowns = calculate_drawdowns(aapl_prices)

import time


def calculate_drawdowns_python(prices):
    """Pure Python implementation for comparison."""
    prices = prices.astype(np.float64)
    n = len(prices)
    if n == 0:
        raise ValueError("prices must be nonempty")
    for price in prices:
        if not np.isfinite(price) or price <= 0:
            raise ValueError("prices must contain only finite positive values")
    drawdowns = np.empty(n)
    peak = prices[0]

    for i in range(n):
        if prices[i] > peak:
            peak = prices[i]
        drawdowns[i] = (prices[i] - peak) / peak

    return drawdowns


# Separate illustrative workload: Gaussian simple-return shocks,
# not the Student-t log-shock data used in the earlier analysis.
large_prices = np.cumprod(1 + np.random.randn(100_000) * 0.02) * 100

# Warm this input signature; compilation cost is excluded from timed calls.
numba_result = calculate_drawdowns(large_prices)
python_result = calculate_drawdowns_python(large_prices)
np.testing.assert_allclose(numba_result, python_result, rtol=1e-12, atol=1e-15)
boundary_prices = np.array([100, 80, 120, 60], dtype=np.uint64)
expected_drawdowns = np.array([0.0, -0.2, 0.0, -0.5])
for kernel in (calculate_drawdowns_python, calculate_drawdowns):
    np.testing.assert_allclose(kernel(boundary_prices), expected_drawdowns)

# Time both implementations
start = time.perf_counter()
for _ in range(100):
    _ = calculate_drawdowns_python(large_prices)
python_time = time.perf_counter() - start

start = time.perf_counter()
for _ in range(100):
    _ = calculate_drawdowns(large_prices)
numba_time = time.perf_counter() - start

# Cython code (would be in a .pyx file)
# cimport numpy as np
# import numpy as np
#
# def calculate_drawdowns_cython(np.ndarray[np.float64_t, ndim=1] prices):
#     cdef int n = prices.shape[0]
#     cdef int i
#     cdef np.ndarray[np.float64_t, ndim=1] drawdowns
#     cdef double peak
#
#     if n == 0:
#         raise ValueError("prices must be nonempty")
#     for i in range(n):
#         if not np.isfinite(prices[i]) or prices[i] <= 0:
#             raise ValueError("prices must contain only finite positive values")
#     drawdowns = np.empty(n)
#     peak = prices[0]
#
#     for i in range(n):
#         if prices[i] > peak:
#             peak = prices[i]
#         drawdowns[i] = (prices[i] - peak) / peak
#
#     return drawdowns