指数算法运算文件指数算法运算文件指数算法运算文件指数算法运算文件指数算法运算文件

#2023/10/23 #交付 #文件说明： # 1，下面很多算法函数，如果想检验算法的可行性，可以直接单独调用函数进行验证 # 2，该文件采用的是根据时间维度进行模块划分 分为有 all模块，cycle周期模块,year年度区分模块（compute_present_product，cycle_data，cycle_data_year）拼接输出 import pandas as pd import numpy as np import datetime import warnings '''此处忽略，pandas计算过程的一些 runwarning报错''' warnings.filterwarnings("ignore") np.seterr(divide='ignore', invalid='ignore') '''路径，此处路径文件是按照发给我的 全部产品——单位净值表中的sheet1来做的''' path='./file/finacial_text.xlsx' def file_load_deal(path): '''读取文件地址，并转换时间列为index，直接获取原数据''' df = pd.read_excel(path) df = df.set_index('Unnamed: 0') names = df.keys() return df,names def file_load_add_time(path): '''读取文件地址，并转换时间列为index,此处读取文件是将空白日期填充''' df = pd.read_excel(path) df['Unnamed: 0'] = pd.to_datetime(df['Unnamed: 0']) df = df.set_index('Unnamed: 0') df = df.resample('D').asfreq() return df def insert_excel(data): # 创建DataFrame df = pd.DataFrame(data) # 将DataFrame写入Excel文件 df.to_excel('./file/output.xlsx', index=False) def calculate_max_drawdown(net_values,dates): """ 计算最大回撤率 参数: net_values (list or numpy.ndarray): 日净值的列表或数组 返回值: float: 最大回撤率 """ net_values = np.array(net_values) # 计算每个时点的最大净值 max_values = np.maximum.accumulate(net_values) # 计算每个时点的回撤率 drawdowns = (max_values - net_values) / max_values max_drawdown_start = np.argmax(drawdowns) max_drawdown_end = np.argmax(net_values[:max_drawdown_start+1]) # 获取最大回撤发生的日期范围 start_date = dates[max_drawdown_end] end_date = dates[max_drawdown_start] # 计算最大回撤率 max_drawdown = np.max(drawdowns) return max_drawdown, start_date, end_date def calculate_volatility(returns): """ 计算波动率的函数 参数: returns (list or numpy.ndarray): 日收益率的列表或数组 返回值: float: 波动率值 """ # 将收益率转换为numpy数组 returns = np.array(returns) # 计算标准差 volatility = np.std(returns) return volatility def calculate_downside_volatility(returns, mean): """ 计算所有上，下行波动率 参数: returns (list or numpy.ndarray): 收益率的列表或数组 mean (float): 收益率的均值 返回值: float: 所有下行波动率 """ returns = np.array(returns) N = len(returns) downside_volatility = np.sqrt(np.sum((returns - mean)**2) / N) return downside_volatility def calculate_max_drawdown_recovery_time(net_values): """ 计算最长回撤修复时间 参数: net_values (list or numpy.ndarray): 日净值的列表或数组 返回值: int: 最长回撤修复时间（以天为单位） """ net_values = np.array(net_values) # 找到最大回撤的结束点的索引 max_drawdown_end = np.argmax(net_values) # 从最大回撤的结束点开始，找到之后的最高点的索引 subsequent_high = np.argmax(net_values[max_drawdown_end:]) # 计算最长回撤修复时间 recovery_time = subsequent_high - max_drawdown_end return recovery_time def date_range(max_date): max_date = datetime.datetime.strptime(str(max_date), '%Y-%m-%d').date() #近1个月的日期 laster_one_date=max_date-datetime.timedelta(30) #近3个月的日期 laster_three_date=max_date-datetime.timedelta(90) #近6个月的日期 laster_six_date=max_date-datetime.timedelta(180) #近12个月的日期 laster_twelve_date=max_date-datetime.timedelta(360) return [laster_one_date,laster_three_date,laster_six_date,laster_twelve_date] def remove_inf_nan(your_list): arr = np.array(your_list) mask = np.isfinite(arr) filtered_arr = arr[mask] filtered_list = filtered_arr.tolist() return filtered_list def year_data_range(): return [ ("2023-01-01","2023-12-31"), ("2022-01-01", "2022-12-31"), ("2021-01-01", "2021-12-31"), ("2020-01-01", "2020-12-31"), ("2019-01-01", "2019-12-31"), ("2018-01-01", "2018-12-31"), ] def compute_present_product(name,df): '''本周数据，所有数据''' # 1) 单位净值指数 max_date = df.index.max() max_date_row = df.loc[max_date] max_date_value_exponent = max_date_row['中证500指数'] # 指数 max_date_value = max_date_row[name] # 1 产品当前净值 # 获取当前最近日期的上周5的年月日 max_date = datetime.datetime.strptime(str(max_date), '%Y-%m-%d').date() current_weekday = max_date.weekday() # 计算距离上周五的天数 days_to_friday = (current_weekday + 2) % 7 # 计算上周五的日期 last_friday = str(max_date - datetime.timedelta(days=days_to_friday)) # 上周五指数 last_friday_date_row = df.loc[last_friday].fillna(0) last_friday_value_index = last_friday_date_row['中证500指数'] last_friday_value = last_friday_date_row[name] current_week_earnings_value = max_date_value / last_friday_value - 1 if last_friday_value!=0 else 0 # 2 当周产品收益 current_week_earnings_index = max_date_value_exponent / last_friday_value_index - 1 if last_friday_value_index != 0 else 0 # 3 当周指数收益 overage_pruduct = current_week_earnings_index - current_week_earnings_value #4 当周产品超额 # 总天数计算 unit_product = df[name].dropna() start_value=unit_product.values[0] start_index=unit_product.index[0] start_exponent_value=df.loc[start_index]['中证500指数'] total_days=(pd.to_datetime(max_date)-pd.to_datetime(start_index)).days #5 总天数计算 all_product_earning=max_date_value/start_value - 1 if start_value!=0 else 0 #6 产品总收益 all_exponent_earning=max_date_value_exponent/start_exponent_value - 1 if start_exponent_value != 0 else 0 #7 指数总收益 all_year_earning=(1+all_product_earning)**(365/total_days)-1 #8 all年化收益率 all_year_allexponent_earning = (1 + all_exponent_earning) ** (365 / total_days) - 1 #9 all指数年化收益率 more_up_year = all_year_earning - all_year_allexponent_earning #10 all年化超额收益率 prices = df[start_index:][name].dropna() returns = [(prices.values[i] - prices.values[i - 1]) / prices.values[i - 1] for i in range(1, len(prices))] volatility = calculate_volatility(returns) #11 all波动率 sharprate=(more_up_year-0.03)/volatility #12 all夏普比率 mean = np.mean(returns) diff_up = [item for item in returns if item > mean ] diff_dowm = [item for item in returns if item < mean ] downside_volatility_up=calculate_downside_volatility(diff_up,mean) #13 all上行波动率 downside_volatility_down=calculate_downside_volatility(diff_dowm,mean) #14 all下行波动率 max_drawdown,start_day,end_day = calculate_max_drawdown(prices,prices.index) #15 最大回撤率值 ,20 最大回撤日期，end_day max_duration = -calculate_max_drawdown_recovery_time(prices) #21 all最长回撤修复时间 cama_rate=all_exponent_earning/max_drawdown if max_drawdown!=0 else 0 #16 all卡玛比率 # 使用cov()方法计算协方差 exponent_cov = df[start_index:]['中证500指数'] product1_cov = df[star