USDCNY即期均值顺势信号——基于Python的均值回归进阶策略

在聚宽注册了之后，发现这个宝藏网站提供了大量的入门策略的教学文章：动态情景多因子Alpha模型、双均线策略、多因子策略入门等等。这个网站主要是为股票的量化投资提供一个策略撰写、回测的平台。站内的量化课堂里存放的策略，不少思路值得借鉴。
本次练习手稿的就参考了网站上的均值回归高阶策略。

原文框架及思路如下：
【量化课堂】均值回归进阶策略https://www.joinquant.com/view/community/detail/e979915d5d2af85ae0a747621c3cf15a
        具体代码可以到聚宽上注册账号后，直接克隆到你账号下的策略列表内。
        这篇文章内提到了《指标效果的统计分析》，在文章内点击链接可能无法查看。有兴趣的可以点击下述链接查看：
指标效果的统计分析：思路之一https://zhuanlan.zhihu.com/p/23393617

策略思路：
本策略信号假设1：当价格偏离过去特定天数均值的n个（n为较小值）标准差时，表明价格有更多的同趋势运动的动能。
a) 如果向上偏离过去若干天移动均价，若后续价格在上下震动后，能在设定时间区间内的某一天，向上移动且突破一个标准差（你可以设定任意你觉得合适的倍数，或者通过回测统计获得合适的倍数参数），则该顺势信号被证实，标记为”赢“；否则该信号被证伪，为”输“。
b) 如果向下偏离过去若干天移动均价，若后续价格在上下震动后，能在设定时间区间内的某一天，向下移动且突破一个标准差，则该顺势信号被证实，标记为”赢“；否则该信号被证伪，为”输“。

本策略信号假设2：当价格偏离过去特定天数均值的n个（n为较大值）标准差时，价格可能无法维持长期偏离的动能，因此有较大的概率回归均值。
c) 如果向上偏离过去若干天移动均价，若后续价格在上下震动后，能在设定时间区间内的某一天，向下移动且突破一个标准差（你可以设定任意你觉得合适的倍数，或者通过回测统计获得合适的倍数参数），则该回归信号被证实，标记为”赢“；否则该回归信号证伪，为”输“。
d) 如果向下偏离过去若干天移动均价，若后续价格在上下震动后，能在设定时间区间内的某一天，向上移动且突破一个标准差，则该回归信号被证实，标记为”赢“；否则该信号被证伪，为”输yi

一、顺势信号

1.导入库类，根据顺势策略思路定义输赢函数

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
%matplotlib qt5#********************价格序列自上到下按时间的由远至今排序*********************************
ma_length=35                #选取当前价格前35天为计算移动均值的时间区间
after_days=10               #设定当前价格后10天为判断输赢的时间区间
win_times_sigma=1           #未来价格运行区间上限
lose_times_sigma=1          #未来价格运行区间下限
least_percentage=0.08       #寻找最优偏离倍数区间时，需要该区间输赢笔数超过总输赢笔数的最低比例
band_width=5                #每次均挑选相邻的5个偏离倍数，作为一簇数据组，在该簇数据中，寻找最优倍数区间
tick_width=0.1              #如保留小数后1位，则设为0.1；后续将【（价格-均价）/方差】的结果整除以0.1，等同于将该偏离度乘以10后取整
min_signal_std=0.5          #设定对偏离度的最低区间为偏离度的0.6个方差
max_signal_std=1.5          #设定对偏离度的最高区间为偏离度的1.5个方差，过度偏离可能意味着动能过大，反而容易冲顶回落或者触底反弹
#**************************************************************************************cny_df=pd.read_csv("c:\\Users\\你的电脑ID\\Desktop\\usdcny.csv",index_col='Date',parse_dates=True)
price_series=cny_df["收盘"].sort_index()#选取当前价格price往后若干天的价格区间my_list,计算以当前价格为原点，上下若干标准差为上下界。
def win_or_lose(price,my_list,sigma,symble):upper_bound=price+win_times_sigma*sigmalower_bound=price-lose_times_sigma*sigmaif symble>0:win_lose='lose'lose_win='win'else:win_lose='win'lose_win='lose'for future_price in my_list:if future_price>=upper_bound:return(lose_win)if future_price<=lower_bound:return(win_lose)return("even")

2.可视化数据并进行观测，从而调整并设定原始参数
下面的函数用来对所有价格的偏离倍数画图，通过该图我们可以相对容易的选定相对合理的价格偏离倍数区间，作为我们入选信号的筛选标准。据图，我们设定min_signal_std=0.6, max_signal_std=1.5。

#可视化观察usdcny的标准差大致范围
def plot_spot_std(price_series,i,ma_length,after_days,plot=False):my_data=[]while i+after_days<len(price_series):range_data=price_series[i-ma_length:i]mean=range_data.mean()sigma=range_data.std()difference_times_sigma=int(((price_series[i]-mean)/sigma)//tick_width)result=win_or_lose(price_series.iloc[i],price_series[i+1:i+after_days+1],sigma,difference_times_sigma)my_data.append((difference_times_sigma,result))i+=1my_data=pd.DataFrame(my_data)std=np.std(my_data[0])if plot:my_data[0].plot()plt.axhline(-min_signal_std * std,color='r')plt.axhline(min_signal_std * std,color='r')plt.axhline(-max_signal_std * std,color='g')plt.axhline(max_signal_std * std,color='g')return std# #这一步把所有交易日spot的偏离度进行统计，计算出该偏离度的标准差，选取限定标准差内的偏离度作为入选信号
std_of_spot_deviation=plot_spot_std(price_series,ma_length,ma_length,after_days,True)

3.收集输赢函数对应的偏离倍数数据集，统计输赢次数
由于后续画图需要，我在collect data内把价格的date信息也添加到数据集内，后续可以根据这一信息拼合数据集，从而对不同信号的数据着色。

def collect_data(price_series,i,ma_length,after_days,std):my_data=[]while i+after_days<len(price_series):range_data=price_series[i-ma_length:i]date=price_series.index[i]mea=range_data.mean()sigma=range_data.std()difference_times_sigma=int((((price_series[i]-mea)/sigma)//tick_width))if -max_signal_std *std<difference_times_sigma<-min_signal_std*std or min_signal_std*std<difference_times_sigma < max_signal_std*std:result=win_or_lose(price_series.iloc[i],price_series[i+1:i+after_days+1],sigma,difference_times_sigma)my_data.append((difference_times_sigma,result,date))i+=1return my_datadef compute_statistics(my_data):statistics={}for pair in my_data:result=pair[1]value=pair[0]if value not in statistics:statistics[value]={'win':0,'even':0,'lose':0}statistics[value][result]+=1return statistics#收集输赢数据，将输赢标签、偏离倍数的字典数据追加到列表内
spot_data=collect_data(price_series,ma_length,ma_length,after_days,std_of_spot_deviation)#对列表内的所有偏离倍数进行归总统计，同一倍数产生多少次输、多少次赢、多少次平局
statistics=compute_statistics(spot_data)#将该统计数据转化为数据框((DF)
spot_df=pd.DataFrame(statistics,index=pd.Series(['value','win','even','lose']))
spot_df.loc['value']=spot_df.columns
spot_stats=spot_df.T
spot_stats.sort_values(by='value',inplace=True)
spot_stats=spot_stats.reset_index(drop=True)

4.对统计数据进行可视化（散点图、柱图）
将原始价格数据(test3)，和贴有输赢标签的数据集(test2)，按照日期数据进行拼合。对拼合后的数据框，添加颜色信息列。并在后续画图时，通过该颜色信息，对散点图和柱图进行相应的着色，提高数据可视化的效率 (此处定义【赢】为红色，【输】为绿色，平局为白色）。


#*************************画价格走势图 & 输赢统计随时间轴分布的的散点图，同时在同轴的第二张图表画出价格偏离倍数随时间分布的柱状图********
fig1,(ax3,ax4)=plt.subplots(2,1,figsize=(15,10))
test2=pd.DataFrame(spot_data,columns=['ratio','win_lose','Date'])
test2.set_index(test2['Date'],inplace=True)
del test2['Date']
test3=pd.DataFrame(price_series).copy()
join_df=test3.merge(test2,how='outer',on='Date')
join_df=join_df.assign(color="")
join_df['ratio'].fillna(0,inplace=True)
join_df['color']=np.where(join_df['win_lose']=="even",'white',np.where(join_df['win_lose']=='win','red',np.where(join_df['win_lose']=='lose','green','white')))
color=join_df['color'].to_list()
ax3.scatter(join_df.index,join_df['收盘'],c=color,marker="*")
ax3.plot(join_df['收盘'],c='orange')
ax4.bar(join_df.index,join_df['ratio'],width=3.5,color=color)#**************************************画图完毕*********************************************

观察图二，可以看出价格偏离倍数的顺势信号，确实在连续下跌，或者连续上涨的时候，产生了足够多的【赢】信号。下面通过另外一张图辅助观察：

fig,(ax1,ax2)=plt.subplots(2,1,figsize=(15,10))values=spot_stats['value']
print(values)
wins=spot_stats['win']
loses=spot_stats['lose']
evens=spot_stats['even']
p1=ax1.bar(values,loses,width=0.6,color='green')
p2=ax1.bar(values,wins,bottom=loses,width=0.6,color='red')
p3=ax1.bar(values,evens,bottom=wins+loses,width=0.6,color='yellow')    ax1.legend((p1[0],p2[0],p3[0]),('lose','win','even'))
ax1.set_xticks(values)
ax1.set_xticklabels(labels=values,rotation=75)ax2.bar(values,wins/(loses+1),width=0.6,color='b')
ax2.set_xticks(values)
ax2.set_xticklabels(values,rotation=75)
ax1.set_title("Chart 3: Win Lose Bar Chart")
plt.tight_layout()

观察图三，可以看到当价格偏离倍数为负，也就是说价格处于下跌通道时，胜率（输赢比）是比较高的。根据图二，我们可以得出初步的感性认识：自2000年---2001年的那波下跌中，价格偏离倍数信号标签多为【赢】。

5.统计正负偏离倍数的输赢笔数

negative_range=spot_stats.loc[spot_stats['value'] < -10,:]
neg_win=negative_range['win'].sum()
neg_lose=negative_range['lose'].sum()
neg_even=negative_range['even'].sum()
print(f"negative range sumary: neg_win:{neg_win},neg_lose:{neg_lose},neg_even:{neg_even}")positive_range=spot_stats.loc[spot_stats['value'] > 10,:]
po_win=positive_range['win'].sum()
po_lose=positive_range['lose'].sum()
po_even=positive_range['even'].sum()
print(f"positive range sumary: po_win:{po_win},po_lose:{po_lose},po_even:{po_even}")

结果如下：
negative range sumary: neg_win:157.0,neg_lose:50.0,neg_even:122.0
positive range sumary: po_win:73.0,po_lose:51.0,po_even:91.0

6.对单次数据集，搜索最优价格偏离倍数的数据簇

#搜索最优偏离度的数据簇，簇的宽度初始化为5根柱。该簇数据内的数据必须超过设定的最低数据百分比（默认为8%）
#传入寻优函数的df默认按照从小到大排序
def get_best_ranges(df,tick_width,least_percentage,band_width):spot_best_ranges={}mydata=[]values=df['value']loses=df['lose']wins=df['win']evens=df['even']num_data=sum(wins)+sum(loses)+sum(evens)low_bound=int(df['value'].iloc[0])high_bound=int(df['value'].iloc[-1]-band_width+1)for n in range(low_bound,high_bound):statistics1=df[values>=float(n)]stat_in_range=statistics1[values<=float(n+band_width-1)]  #从入选的偏离度数据内，按照事先设定好的数据簇宽度选取数据ratio=float(sum(stat_in_range['win']))/float((sum(stat_in_range['lose'])+1))  #计算该数据簇的输赢比率range_data=float(sum(stat_in_range['win'])+sum(stat_in_range['lose'])+sum(stat_in_range['even'])) if range_data/num_data>=least_percentage:  #如果该数据簇内的样本个数超过设定的最低比例线，则添加到列表保存起来mydata.append({'low':n,'high':n+band_width,'ratio':ratio,'original data':stat_in_range})data_table=pd.DataFrame(mydata)sorted_table=data_table.sort_values('ratio',ascending=False)   #把转成数据框的列表按照输赢比率的降序进行排序spot_best_ranges=sorted_table.iloc[0]  return spot_best_rangesspot_best_ranges=get_best_ranges(spot_stats,tick_width,least_percentage,band_width)
print(spot_best_ranges)

由于入选信号（价格偏离倍数）数量不少，哪怕统计归总后的价格偏离倍数也有32个，在这32个信号里面，我们需要根据某些标准，挑选出胜率（赢的次数除以输的次数）比较高的信号。如果按照图三内，每根信号柱子上的【赢】次数除以【输】次数来进行划分，结论有效性存疑。
因为某些柱子内胜率不低，例如信号-23，胜率为+6（由于该信号的输次数是0，在算胜率时需要将【输】的笔数加上1，避免除零错误）。但是信号【-23】的总笔数12笔，占信号总的笔数544笔的2.2%。这个信号的发生比例太低，不足以对实际交易提供有力的支持。因此，我们需要设定最低的数据占比。本文设定比例为8%。
这样就能有效避免单根信号胜率较高但在总体数据内比重过低的弊端。因此我们通过数据簇的概念，来搜索包含最优胜率的数据簇。这样，我们后续就能在实际交易中，利用能产生最优胜率的足够大的信号区间来指导交易。本文设定的数据簇区间为5根信号柱(band_width)。

7.迭代不同长度的日期区间，搜索最优信号的数据簇

def find_best_dayset(price_series):best_dayset=[]for x in range(5,60,5):for y in range(5,30,5):ma_length=xafter_days=ystd_of_spot_deviation=plot_spot_std(price_series,ma_length,ma_length,after_days,False)spot_data=collect_data(price_series,ma_length,ma_length,after_days,std_of_spot_deviation)statistics=compute_statistics(spot_data)spot_df=pd.DataFrame(statistics,index=pd.Series(['value','win','even','lose']))spot_df.loc['value']=spot_df.columnsspot_stats=spot_df.Tspot_stats.sort_values(by='value',inplace=True)spot_stats=spot_stats.reset_index(drop=True)spot_best_ranges=get_best_ranges(spot_stats,tick_width,least_percentage,band_width)best_dayset.append({'Day_set':f"Moving days {x} ,afterdays {y}",'ratio':spot_best_ranges['ratio'],'low':spot_best_ranges['low'],'high':spot_best_ranges['high'],'original data':spot_best_ranges['original data']})return best_daysetfind_best_dayset=find_best_dayset(price_series)
result_df=pd.DataFrame(find_best_dayset)
sorted_result_df=result_df.sort_values('ratio',ascending=False)
print(sorted_result_df.iloc[0])

我们无法确认在USDCNY即期交易里，多长时间的移动平均价格偏离倍数，会对后续多长时间的价格波动产生有效信号。因此，本文通过迭代不同的moving average时间长度，以及不同长度的after days，来计算出不同组合内的最优信号。再将该信号数据进行排序后，获得最优信号相应的时间组合。然后再利用这一时间组合来修正在一开始设定的初始参数。
结果如下：

Day_set                                                                                                                                               Moving days 35 ,afterdays 10
ratio 7
low -18
high -13
original data       value   win  even  lose
5  -18.0  10.0  11.0   1.0
6  -17.0  13.0  13.0   3.0
7  -16.0  17.0   8.0   3.0
8  -15.0  18.0   7.0   2.0
9  -14.0  12.0  10.0   0.0

二、回归信号

1.修改输赢函数

def win_or_lose(price,my_list,sigma,symble):upper_bound=price+win_times_sigma*sigmalower_bound=price-lose_times_sigma*sigmaif symble>0:win_lose='win'lose_win='lose'else:win_lose='lose'lose_win='win'for future_price in my_list:if future_price>=upper_bound:return(lose_win)if future_price<=lower_bound:return(win_lose)return("even")

2.修改偏离倍数信号上下界
        ma_length=5
        after_days=10
        min_signal_std=1.5
        max_signal_std=3.5
        根据策略信号假设2，我们除了需要修改输赢函数，还需要修改产生信号的偏离倍数上下边界。能产生回归信号的价格，我们假定它已经偏离移动均值较大倍数。因此本文选择了偏离倍数的1.5倍--3.5倍方差为上下界。同时，按照顺势策略的迭代寻优思路，我们测算出回归信号的移动均值时间长度为5天，该长度产生的信号，需要根据后续10天的时间区间来判定输赢与否。

观察图5，可以看到回归信号确实主要分布在价格图的走势逆转顶点或者谷底。但是价格信号总数不多，而且胜率也比较低。多数胜率数值小于等于2。实话说，回归信号对交易的指导意义不大。

三、结论
在应用均值回归策略进阶版对USDCNY即期交易的历史数据进行分析时，我们可以发现USDCNY即期交易中，顺势信号的胜率是比较高的，而且在USDCNY下跌时胜率尤其高。但是如果想从USDCNY的即期中寻找反转信号（回归信号），需要面临胜率较低的事实。
如果是在JoinQuant中，采集了有效交易信号后，可以直接通过handle_data等交易处理函数来进行回测，并生成回测的收益图表和数据。对实际交易有着比较好的参考意义。
此外，本次练手并没有分出训练数据集和回测数据集。这当然是后续需要优化的地方。毕竟本文只是一份入门级别的练习手稿。
继续加油吧！