from numpy.lib.stride_tricks import as_strided as strided
def get_sliding_window(narray, window, return2D=0):
s0,s1 = narray.strides
m,n = narray.shape
out = strided(narray,shape=(m-window+1,window,n),strides=(s0,s0,s1))
if return2D==1:
return out.reshape(a.shape[0]-W+1,-1)
else:
return out
def ts_rank_np(df,window=20):
df_w = get_sliding_window(df.values,window=window)
out = np.sum(np.array([d_w[:,-1,:] > d_w[:,i,:] for i in range(window-1)]),axis=0)
padding = np.full(shape=(window-1,df.shape[1]),fill_value= np.nan)
return pd.DataFrame(np.vstack((padding,out)),index = df.index,columns = df.columns)
def ts_max_rank_np(df,window=20):
df_w = get_sliding_window(df.values,window=window)
out =np.argsort(d_w,axis=1)[:,0,:]
padding = np.full(shape=(window-1,df.shape[1]),fill_value= np.nan)
return pd.DataFrame(np.vstack((padding,out)),index = df.index,columns = df.columns)
def ts_min_rank_np(df,window=20):
df_w = get_sliding_window(df.values,window=window)
out =np.argsort(d_w,axis=1)[:,-1,:]
padding = np.full(shape=(window-1,df.shape[1]),fill_value= np.nan)
return pd.DataFrame(np.vstack((padding,out)),index = df.index,columns = df.columns)
@jit
def test(df,window=20):
df_w = get_sliding_window(df.values,window=window)
out = np.sum(df_w[:,np.repeat(-1,19),:] > df_w[:,:-1,:],axis=1)
padding = np.full(shape=(window-1,df.shape[1]),fill_value= np.nan)
return pd.DataFrame(np.vstack((padding,out)),index = df.index,columns = df.columns)
def backtest_old(alpha_raw ,daily_ret,alpha_id='5dr'):
factor = alpha_raw.copy()
daily_ret = daily_ret.loc[factor.index,factor.columns]
# 剔除未上市合约
factor[np.isnan(daily_ret)] = np.nan
# 因子归一化，减均值除绝对值之和
factor = factor.sub(factor.mean(axis=1),axis=0)
factor_neutral = factor.div(factor.abs().sum(axis=1),axis=0)
ret = daily_ret.shift(-2,axis=0)
# ic_mean = factor_neutral.corrwith(ret).mean()
ret_matrix = (factor_neutral*ret)
# 得到多空收益率序列
long_short_ret = ret_matrix.sum(axis=1)
long_short_net_value = long_short_ret.cumsum()+1
# 计算回撤
drawdown = (long_short_net_value.groupby(long_short_net_value.index.year).cummax() - long_short_net_value) # 绝对值计算
max_drawdown = drawdown.groupby(drawdown.index.year).max()
drawdown_all = (long_short_net_value.cummax() - long_short_net_value) # 绝对值计算
max_drawdown_all = drawdown_all.max()
# 计算日度夏普比率
sharpe = long_short_ret.groupby(long_short_ret.index.year).mean() / long_short_ret.groupby(long_short_ret.index.year).std()
sharpe_all = long_short_ret.mean() / long_short_ret.std()
# 年化收益率
annual_ret = long_short_ret.groupby(long_short_ret.index.year).sum()
annual_ret_all = long_short_ret.sum() / len(long_short_ret) * 252
# 双边换手率
turnover = factor_neutral.fillna(0).diff().abs().sum(axis=1)
turnover_mean = turnover.groupby(turnover.index.year).mean()
turnover_mean_all = turnover.mean()
# Long short
long_count = (factor_neutral > 0).sum(axis=1).groupby(factor_neutral.index.year).mean()
short_count = (factor_neutral < 0).sum(axis=1).groupby(factor_neutral.index.year).mean()
long = factor_neutral[factor_neutral > 0].sum(axis=1).groupby(factor_neutral.index.year).mean()
short = factor_neutral[factor_neutral < 0].sum(axis=1).groupby(factor_neutral.index.year).mean()
long_count_all = (factor_neutral > 0).sum(axis=1).mean()
short_count_all = (factor_neutral < 0).sum(axis=1).mean()
long_all = factor_neutral[factor_neutral > 0].sum(axis=1).mean()
short_all = factor_neutral[factor_neutral < 0].sum(axis=1).mean()
result = pd.concat([long,short,long_count,short_count,sharpe,annual_ret,turnover_mean,max_drawdown],axis=1)
result.columns = ['long','short','long(num)','short(num)','sharpe','returns','turnover','drawdown']
result_all = pd.DataFrame([[long_all,short_all,long_count_all,short_count_all,sharpe_all,annual_ret_all,turnover_mean_all,max_drawdown_all]])
result_all.columns = result.columns
result_all.index = ['all']
result = pd.concat([result,result_all])
return alpha_summary(alpha_name = alpha_id,daily_pnl=long_short_ret,
factor_neutral=factor_neutral,
ret_matrix=ret_matrix,results= result)
