Backtesting_API_Project/Backtest.py at main · Matthieu-Monnot/Backtesting_API_Project · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
import json
import numpy as np
import pandas as pd


class Stats:
    def __init__(self, poids_ts, dfs_dict):
        self.poids_ts = poids_ts
        self.dfs_dict = dfs_dict
        self.rf_rate = 0.2
        self.scale = 9
        self.r_indice = self.calculate_index_returns()
        self.setup_metrics()

    def calculate_returns_from_dfs(self):
        df_closes = pd.DataFrame()
        df_concat = pd.DataFrame()
        for key, df in self.dfs_dict.items():
            df_closes = pd.DataFrame(df['Close'].items(), columns=['Date', key]).set_index('Date')
            df_closes[key] = df_closes[key].astype(float)
            df_concat = pd.concat([df_closes, df_concat], axis=1)
        df_returns = df_concat.pct_change().fillna(0)
        df_returns = df_returns.reset_index(drop=True)
        return df_returns

    def calculate_index_returns(self):
        df_returns = self.calculate_returns_from_dfs()
        df_poids = self.poids_ts.reset_index(drop=True)
        df_index_returns = (df_returns * df_poids).sum(axis=1)
        return df_index_returns.to_frame(name='Index_Return')
        #return self.poids_ts
        #return df_returns

    def setup_metrics(self):
        self.r_annual = self.annualize_rets(self.r_indice, self.scale)
        self.vol_annual = self.annualize_vol()
        self.sharpe_r = self.sharpe_ratio()
        self.skew = self.skewness()
        self.kurt = self.kurtosis()
        self.semi_deviation = self.semideviation()
        self.var_hist = self.var_historic()
        self.max_draw = self.compute_drawdowns()
        self.downside_vol = self.downside_volatility()
        self.sortino_ratio = self.sortino_ratio()
        self.calmar_ratio = self.calmar_ratio()

    @staticmethod
    def annualize_rets(r, scale):
        compounded_growth = (1 + r).prod()
        n_periods = r.shape[0]
        return compounded_growth ** (scale / n_periods) - 1

    def annualize_vol(self):
        return self.r_indice.std() * np.sqrt(self.scale)

    def sharpe_ratio(self):
        rf_per_period = (1 + self.rf_rate) ** (1 / self.scale) - 1
        excess_ret = self.r_indice - rf_per_period
        return self.annualize_rets(excess_ret, self.scale) / self.annualize_vol()

    def skewness(self):
        demeaned_r = self.r_indice - self.r_indice.mean()
        return (demeaned_r ** 3).mean() / (self.r_indice.std(ddof=0) ** 3)

    def kurtosis(self):
        demeaned_r = self.r_indice - self.r_indice.mean()
        return (demeaned_r ** 4).mean() / (self.r_indice.std(ddof=0) ** 4)

    def semideviation(self):
        return self.r_indice[self.r_indice < 0].std(ddof=0)

    def var_historic(self, level=5):
        return np.percentile(self.r_indice, level)

    def compute_drawdowns(self):
        peaks = self.r_indice.cummax()
        drawdowns = (self.r_indice - peaks) / peaks
        return drawdowns.min()

    def downside_volatility(self):
        downside = np.minimum(self.r_indice - self.rf_rate, 0)
        return np.sqrt(np.mean(downside ** 2))

    def sortino_ratio(self):
        rf_per_period = (1 + self.rf_rate) ** (1 / self.scale) - 1
        excess_return = self.r_indice - rf_per_period
        return self.annualize_rets(excess_return, self.scale) / self.downside_volatility()

    def calmar_ratio(self):
        return self.r_annual / -self.max_draw

    def to_json(self):
        stats_dict = {
            'Rendement Annuel': self.r_annual.item() if isinstance(self.r_annual, pd.Series) else self.r_annual,
            'Volatilite Annuelle': self.vol_annual.item() if isinstance(self.vol_annual,
                                                                        pd.Series) else self.vol_annual,
            'Ratio de Sharpe': self.sharpe_r.item() if isinstance(self.sharpe_r, pd.Series) else self.sharpe_r,
            'Skewness': self.skew.item() if isinstance(self.skew, pd.Series) else self.skew,
            'Kurtosis': self.kurt.item() if isinstance(self.kurt, pd.Series) else self.kurt,
            'Semi-Deviation': self.semi_deviation.item() if isinstance(self.semi_deviation,
                                                                       pd.Series) else self.semi_deviation,
            'VaR Historique': self.var_hist.item() if isinstance(self.var_hist, pd.Series) else self.var_hist,
            'Drawdown Maximal': self.max_draw.item() if isinstance(self.max_draw, pd.Series) else self.max_draw,
            'Volatilite a la Baisse': self.downside_vol.item() if isinstance(self.downside_vol,
                                                                             pd.Series) else self.downside_vol,
            'Ratio de Sortino': self.sortino_ratio.item() if isinstance(self.sortino_ratio,
                                                                        pd.Series) else self.sortino_ratio,
            'Ratio de Calmar': self.calmar_ratio.item() if isinstance(self.calmar_ratio,
                                                                      pd.Series) else self.calmar_ratio
        }
        return json.dumps(stats_dict, indent=4)


if __name__ == "__main__":
    statistiques = Stats()