numpy 技巧

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)