_compute_indicator is the function which the RSI is calculated with:
def _compute_indicator(data_: pd.Series, period: int):
clean_data = data_.dropna()
delta = clean_data.diff()
delta_up = delta.clip(lower=0)
delta_down = delta.clip(upper=0).abs()
avg_gain = pd.Series(
np.full(len(delta_up), np.nan), index=delta_up.index.copy(), name=data_.name
)
avg_loss = pd.Series(
np.full(len(delta_down), np.nan),
index=delta_down.index.copy(),
name=data_.name,
)
avg_gain[period] = delta_up[: period + 1].mean()
avg_loss[period] = delta_down[: period + 1].mean()
for i in range(period + 1, avg_gain.shape[0]):
avg_gain.iloc[i] = (
avg_gain.iloc[i - 1] * (period - 1) + delta_up.iloc[i]
) / period
avg_loss.iloc[i] = (
avg_loss.iloc[i - 1] * (period - 1) + delta_down.iloc[i]
) / period
rs = avg_gain / avg_loss
rsi_calc = 100 - (100 / (1 + rs))
return rsi_calc
This is the Code where the function is implemented in:
if isinstance(data, pd.DataFrame):
indicator = data.apply(_compute_indicator, period=periods)
indicator = indicator.reindex(data.index)
else:
indicator = _compute_indicator(data, periods)
indicator = indicator.reindex(data.index)
return indicator
The Code is checking in an If-Else Statement if it is an pd.Dataframe or a Series-
Question: How to change the for loop to something quicker? Maybe a Matrix operation?
for i in range(period + 1, avg_gain.shape[0]):
avg_gain.iloc[i] = (
avg_gain.iloc[i - 1] * (period - 1) + delta_up.iloc[i]
) / period
avg_loss.iloc[i] = (
avg_loss.iloc[i - 1] * (period - 1) + delta_down.iloc[i]
) / period
This is the part which is super slow. But I have no Idea how to make it quicker