基金评价指标1——收益率、回撤、下行标准差、痛苦指数、夏普比率、索蒂诺比率

大家好，欢迎来到IT知识分享网。

基金的评价指标有许多，这一篇博客分享的是最常见的评价指标，包括以下内容：

1. 单日涨跌幅
2. 累计收益率
3. 年化收益率
4. 最大回撤：从历史上最高点到之后的最低点的最大跌幅
5. 痛苦指数：创新高才会不痛苦，因此这个指标描述的是从上次创新高到现在的痛苦程度
6. 涨跌幅_年化标准差：表示净值波动的程度
7. 涨跌幅_年化下行标准差：由于向上波动代表盈利，向下波动代表亏损。而我们更在意亏损/风险时就会需要这个指标，计算方式是以日涨跌幅<平均涨跌幅的数据为样本，计算这些下跌超过平均数的标准差
8. 夏普比率：(年化收益率-无风险收益率) / 年化标准差
9. 索蒂诺比率：(年化收益率-无风险收益率) / 年化标准差

以上指标我们都使用pandas的向量化操作加速计算

各个模块

1. 用生成随机数作为模拟的净值序列

import pandas as pd import numpy as np from copy import deepcopy # 生成数据 _size = 500 net_value = 1 + pd.Series(np.random.normal(0, 0.01, size=_size)).cumsum() net_value.index = pd.date_range("2024-01-01", periods=_size) collect_df = pd.DataFrame({ 
   "单位净值": net_value}) 

2. 收益率相关计算

one_year_count = 365 # 表示一年又多少个数据点 # 收益率相关指标 collect_df['单日涨跌幅'] = collect_df['单位净值'] / collect_df['单位净值'].shift(1) - 1 collect_df["累计收益率"] = collect_df['单位净值'] - collect_df['单位净值'].iloc[0] collect_df['年化收益率'] = collect_df['累计收益率'] / (collect_df.index - collect_df.index[0]).days * one_year_count 

3. 风险指标计算

collect_df["最大回撤"] = collect_df['单位净值'].rolling(collect_df.shape[0], min_periods=0).apply(cal_max_draw_down) collect_df["痛苦指数"] = -(collect_df['单位净值'].cummax() - collect_df['单位净值']) / collect_df['单位净值'].cummax() collect_df["涨跌幅_年化标准差"] = collect_df['单日涨跌幅'].rolling(collect_df.shape[0], min_periods=0).std() * np.sqrt(252) collect_df['涨跌幅_年化下行标准差'] = collect_df['单日涨跌幅'].rolling( collect_df.shape[0], min_periods=0).apply(lambda x: x[x < x.mean()].std() * np.sqrt(one_year_count)) 

其中下行标准差有时也被认为小于0的样本，就是：

collect_df['涨跌幅_年化下行标准差'] = collect_df['单日涨跌幅'].rolling( collect_df.shape[0], min_periods=0).apply(lambda x: x[x < 0].std() * np.sqrt(one_year_count)) 

4. 综合评价

rf=0.03 collect_df["夏普比率"] = (collect_df['年化收益率'] - rf) / collect_df['涨跌幅_年化标准差'] collect_df['索蒂诺比率'] = (collect_df['年化收益率'] - rf) / collect_df['涨跌幅_年化下行标准差'] 

5. 异常的INF值处理

计算实际的基金净值如夏普比率，由于一开始基金从成立后有一段时间是无法交易的，所以刚开始分母的标准差大概率为0，此时夏普就是 inf，使用replace()替换掉这些inf值

collect_df.replace(-np.inf, float("nan"), inplace=True) collect_df.replace(np.inf, float("nan"), inplace=True) 

完整示例程序

import pandas as pd import numpy as np from copy import deepcopy def cal_max_draw_down(net_series): """计算最大回撤""" # (累计最大值 - 某个值) / 某个值，就找到了相对过去的最大值，某个值的最大跌幅 _history_low_index = np.argmax((np.maximum.accumulate(net_series) - net_series) / net_series) if _history_low_index == 0: # 没有回撤 return 0 else: _history_max_index = np.argmax(net_series[:_history_low_index]) # 最大的下标 return -(net_series[_history_max_index] - net_series[_history_low_index]) / net_series[_history_max_index] def analyze_net_value(pure_value_series: pd.Series, rf=0.03, one_year_count=365, normalize_first_data=True): """计算各项指标 :param pure_value_series: 净值序列数据, index为日期, value为净值 :param rf: 无风险收益率 :param one_year_count: 一年的天数 :param normalize_first_data: 是否归一化第一个数据 """ pure_value_series = deepcopy(pure_value_series) pure_value_series.index = pd.to_datetime(pure_value_series.index) pure_value_series.sort_index(inplace=True) pure_value_series.dropna(inplace=True) # 开始计算 if normalize_first_data: pure_value_series = pure_value_series / pure_value_series.iloc[0] collect_df = pd.DataFrame({ 
   "单位净值": pure_value_series}) # 收益率相关指标 collect_df['单日涨跌幅'] = collect_df['单位净值'] / collect_df['单位净值'].shift(1) - 1 collect_df["累计收益率"] = collect_df['单位净值'] - collect_df['单位净值'].iloc[0] collect_df['年化收益率'] = collect_df['累计收益率'] / (collect_df.index - collect_df.index[0]).days * one_year_count # 风控相关指标 collect_df["最大回撤"] = collect_df['单位净值'].rolling(collect_df.shape[0], min_periods=0).apply(cal_max_draw_down) collect_df["痛苦指数"] = -(collect_df['单位净值'].cummax() - collect_df['单位净值']) / collect_df['单位净值'].cummax() collect_df["涨跌幅_年化标准差"] = collect_df['单日涨跌幅'].rolling(collect_df.shape[0], min_periods=0).std() * np.sqrt(252) collect_df['涨跌幅_年化下行标准差'] = collect_df['单日涨跌幅'].rolling( collect_df.shape[0], min_periods=0).apply(lambda x: x[x < x.mean()].std() * np.sqrt(one_year_count)) # 综合评价 collect_df["夏普比率"] = (collect_df['年化收益率'] - rf) / collect_df['涨跌幅_年化标准差'] collect_df['索蒂诺比率'] = (collect_df['年化收益率'] - rf) / collect_df['涨跌幅_年化下行标准差'] # 异常值替换 collect_df.replace(-np.inf, float("nan"), inplace=True) collect_df.replace(np.inf, float("nan"), inplace=True) return collect_df def run_main(): # 生成数据 _size = 500 net_value = 1 + pd.Series(np.random.normal(0, 0.01, size=_size)).cumsum() net_value.index = pd.date_range("2024-01-01", periods=_size) # 计算各项指标 collect_df = analyze_net_value(net_value) print(collect_df.tail()) if __name__ == '__main__': run_main()