Reducción media y duración media de la reducción en Python

Question

Estoy intentando usar Python para que me dé más información sobre las detracciones que sólo la detracción máxima y la duración de la detracción máxima. Me gustaría determinar el número de detracciones que se han producido (más allá de un cierto umbral de recuento de días), la detracción media y la duración media de la detracción. He encontrado este pregunta con una respuesta sobre la extracción máxima, y la longitud de la extracción máxima, pero después de leer los comentarios, no estoy seguro de qué hacer con él. También he encontrado este pregunta que parece dar una reducción máxima diferente, así que estoy un poco confundido. Creo que el segundo es lo que estoy buscando, pero no quiero un máximo; Quiero una reducción media que ha durado más de un número de días (digamos cinco días).

Mi conjunto de datos es un marco de datos Pandas con precios. Esto es lo que tengo hasta ahora, pero ahora estoy atascado en cómo proceder:

def avg_dd(df, column='close'):

    df['simple_ret'] = df[column].pct_change().fillna(0)
    df['cum_ret'] = (1 + df['simple_ret']).cumprod() - 1
    df['nav'] = ((1 + df['cum_ret']) * 100).fillna(100)
    df['hwm'] = df['nav'].cummax()
    df['dd'] = df['nav'] / df['hwm'] - 1

A partir de aquí, mi idea era utilizar el hwm como un índice que se incrementa cada vez que alcanza un nuevo máximo, y la distancia entre ellos era la duración de esa reducción temporal.

¿Alguien tiene una fuente o referencia que pueda ayudarme?

Answer 1

He creado una solución que espero que te funcione. No estoy seguro de si esto es exactamente lo que tenía en mente. De todos modos, primero voy a crear algunos datos de prueba al azar. Esto nos da una serie que de alguna manera se asemeja a un activo real:

import pandas as pd
import numpy as np

np.random.seed(1)
rand = np.random.normal(size=750)
delta_S = (0.05 * 1 / 252 + 0.2 * rand * np.sqrt(1 / 252))
df = pd.DataFrame(S, columns=["S"], index=pd.bdate_range("2010-01-01","2014-01-01")[:750]).add(1).cumprod().mul(100)

A continuación calculamos la reducción. He excluido la primera observación y he creado un indicador is_dd cada vez que la reducción llega a 0. Con ello se pretende delimitar un periodo de reducción, es decir, cada periodo de reducción llega a su fin cuando alcanzamos la marca cero. A continuación, identifico los regímenes/períodos de reducción del riesgo mediante la función ne y un desplazamiento de la serie temporal. Por último, elimino las observaciones que son simplemente cero porque no son de interés aquí.

dd = df / df.cummax() - 1
dd = dd.iloc[1:]
dd["is_dd"] = np.where(np.isclose(dd, 0), 1, 0)
dd["regime"] = dd["is_dd"].ne(dd["is_dd"].shift(1)).cumsum()
dd = dd[~np.isclose(dd["S"], 0)]
dd.drop(columns="is_dd", inplace=True)

A continuación, puede ejecutar algunos análisis, como la reducción media, la duración, etc. Mi solución para obtener periodos de al menos 5 días no es muy clara, pero funciona:

def days_to_trough(x):
    t_date = x[x == x.min()].index
    return (t_date - x.index[0]).days

def days_of_drawdown(x):
    return (x.index[-1] - x.index[0]).days

atleast_5 = dd.groupby("regime")["S"].count().gt(5).replace(False, np.nan).dropna().index
dd[dd["regime"].isin(atleast_5)].groupby("regime")["S"].agg([days_of_drawdown, "min", "mean", days_to_trough]).round(3).T

Tenga en cuenta que si llama a la función min En función de esto, obtendrá una reducción máxima del 15% (véase el régimen 51), que coincide con lo que se obtiene mediante el simple cálculo de la MDD de la serie. Se utiliza el valor de 2011-04-14 y 2011-08-12 en mi serie de tiempo artificial y se calcula como 129.503421/152.829372 - 1 .

Por último, un poco de viz:

fig, ax = plt.subplots(figsize=(13,5))
dd["S"].plot(title="MaxDD vs regimes", ax=ax)
xpos = dd.reset_index().groupby("regime").first()["index"]

for x in xpos:
    ax.axvline(x=x, color='r', linestyle='-', alpha=0.5)

Cada línea vertical delimita un nuevo periodo de detracción:

Espero que le sirva de ayuda.

Answer 2

Un enfoque ligeramente diferente al de @oronimbus. Esperemos que, entre ambas respuestas, puedas lograr tu objetivo.

La siguiente función toma un Pandas Dataframe, df con los precios (no los rendimientos) en una columna denominada close realiza todos los cálculos necesarios y devuelve el número de caídas, la profundidad media de todas las caídas y el tiempo medio (en días) para recuperar la caída (es decir, alcanzar un nuevo máximo). Además de la información que devuelve la función, todos los datos necesarios para los cálculos se añaden al marco de datos como columnas para que pueda examinarlos y entender lo que hice para llegar a la salida.

import pandas as pd
import numpy as np

def avg_dd(df, column='close'):

    df['simple_ret'] = df[column].pct_change().fillna(0)
    df['cum_ret'] = (1 + df['simple_ret']).cumprod() - 1
    df['nav'] = ((1 + df['cum_ret']) * 100).fillna(100)
    df['hwm'] = df['nav'].cummax()
    df['dd'] = df['nav'] / df['hwm'] - 1
    df['hwm_idx'] = (df['nav']
                     .expanding(min_periods=1)
                     .apply(lambda x: x.argmax())
                     .fillna(0)
                     .astype(int))
    df['dd_length'] = (df['hwm_idx'] - df['hwm_idx'].shift(1) - 1).fillna(0)
    df['dd_length'] = df['dd_length'][df['dd_length'] > 5]
    df['dd_length'].fillna(0, inplace=True)
    dd_end_idx = df['hwm_idx'].loc[df['dd_length'] != 0]
    temp_dd_days = df['dd_length'].loc[df['dd_length'] != 0]
    dd_start_idx = dd_end_idx - temp_dd_days
    temp_dd = [min(df['dd'].loc[df.index[int(dd_start_idx[i])]:
                                df.index[int(dd_end_idx[i])]])
               for i in range(len(dd_end_idx))]
    num_dd = len(temp_dd)
    avg_dd = np.average(temp_dd)
    avg_dd_length = (df['dd_length'][df['dd_length'] > 0]).mean()

    return num_dd, avg_dd, avg_dd_length

El retorno de la función es una tupla con la información buscada. Feliz de explicar lo que está pasando en el código si es necesario, pero creo que es bastante fácil de averiguar.

¡Buena suerte!