Find the maximum price change over any 1 to 5 day rolling window

Question

The code below solves the following task

Find the maximum price change over any 1 to 5 day rolling window, over a 1000 day period.

To be clear "any 1 to 5 day rolling window" means t[i:i + j] where j can range from 1 to 5. Rather than t[i:i + 5] if it were just "any 5 day rolling window".

I used NumPy native functions to do this. But I've used a for-loop for the inner iteration.

import numpy as np
import numpy.random as npr

prices = npr.random([1000,1])*1000

max_array = np.zeros([(prices.size-5),1])
for index, elem in np.ndenumerate(prices[:-5,:]):
    local_max = 0.0
    for i in range(1,6,1):
        price_return = prices[(index[0] + i),0] / elem
        local_max = max(local_max, price_return)
    max_array[index[0]] = local_max
global_max = np.amax(max_array)

Can I somehow eliminate the inner for loop and use NumPy vectorization instead?

I also don't particularly like using "index[0]" to extract the actual index of the current loop from the tuple object that is returned into the variable "index" via the call:

for index, elem in np.ndenumerate(prices[:-5,:]):

Answer 1

1. Put this in a function.
2. Make the input a 1d array.

def maximum(prices):
    max_array = np.zeros(prices.size - 5)
    for index, elem in np.ndenumerate(prices[:-5]):
        local_max = 0.0
        for i in range(1,6,1):
            price_return = prices[index[0] + i] / elem
            local_max = max(local_max, price_return)
        max_array[index] = local_max
    return np.amax(max_array)

3. Flip the enumeration around.

def maximum(prices):
    max_array = np.zeros(5)
    for i in range(1, 6):
        local_max = 0.0
        for index, elem in np.ndenumerate(prices[:-5]):
            price_return = prices[index[0] + i] / elem
            local_max = max(local_max, price_return)
        max_array[i - 1,] = local_max
    return np.amax(max_array)

4. Use numpy to slice the array. And then use .max.

def maximum(prices):
    size, = prices.shape
    size -= 4
    max_array = np.zeros(5)
    for i in range(5):
        max_array[i,] = (prices[i:size + i] / prices[:size]).max()
    return np.amax(max_array)

5. Use max and a comprehension.

Now the inner loop is vectorized and it's a lot easier to see what it's doing.

def maximum(prices, windows=5):
    size, = prices.shape
    size -= windows - 1
    return max(
        (prices[i:size + i] / prices[:size]).max()
        for i in range(windows)
    )

Answer 2

A first pass will see (as Vishesh suggests) collapsing one dimension, and properly unpacking the index tuple:

import numpy as np

# Window size
W = 5

prices = np.arange(1, 10)  # npr.random([1000,1])*1000

max_array = np.zeros(prices.size-W)
for (index,), elem in np.ndenumerate(prices[:-W]):
    local_max = 0
    for i in range(1, W+1):
        price_return = prices[index + i] / elem
        local_max = max(local_max, price_return)
    max_array[index] = local_max
global_max = np.amax(max_array)

I've also replaced your random array with a simple one so that this is repeatable and verifiable. You should make your window size parametric or at least configurable via constant.

Unfortunately, none of the native Numpy options are great. Your best bet is to use Pandas:

import numpy as np
import pandas as pd

# Window size
W = 5

rolled = (
    pd.Series(prices)
    .rolling(window=W)
    .min()
    .to_numpy()
)
max_array = prices[1-W:] / rolled[W-1:-1]

I have tested this to be equivalent. Note that you need to start with a rolling minimum in the denominator to get a maximum numerator.

Answer 3

This line is equivalent to your code which you were likely trying to make and using 2d array for

 np.max(
    prices[np.arange(0, len(prices) - 5).reshape(-1, 1) + np.arange(1, 6)].squeeze()
    / prices[np.arange(0, len(prices) - 5)]
)

Hmm, it's not easy to understand. Let's break it into chunks.

np.arange(0, len(prices) - 5).reshape(-1, 1) 

gives

array([[  0],
       [  1],
       [  2],
       [  3],
       .....
       .....
       [  994]])

Next, here, something called as "broadcasting" is taking place.

np.arange(0, len(prices) - 5).reshape(-1, 1) + np.arange(1, 6)

gives

array([[  1,   2,   3,   4,   5],
       [  2,   3,   4,   5,   6],
       [  3,   4,   5,   6,   7],
       ...,
       [993, 994, 995, 996, 997],
       [994, 995, 996, 997, 998],
       [995, 996, 997, 998, 999]])

After getting the right index we take the prices at those indices, but it will return a 3d matrix. Therefore we use "squeeze" to remove one of the dimensions.

prices[np.arange(0, len(prices) - 5).reshape(-1, 1) + np.arange(1, 6)].squeeze()

Finally we use vector division and return the maximum of the matrix.

Alternate method: Well numpy is good but this broadcasting and some stuff is often unclear to me and always keeps me in the dilemma if my matrices are being multiplied right.Sure I can use assert but there's a Python package "Numba" which optimizes your Python code.You should definitely check it out.