风险提示：本内容为量化技术及平台操作的纯知识分享，不构成任何投资建议，不推荐任何具体证券或交易时机。部分内容可能理解包含错误，请自行核实。市场有风险，投资需谨慎
策略研究中，我们经常会看见2种常见的交易算法，因子的排序交易轮动，和因子的选择交易2种算法，因子的排序交易，按因子的排名来轮动交易的一种方式，因子的选择交易就是按因子的买入条件，排序条件，卖出条件选择标的交易，和因子排序交易是2种算法，常见的交易方式有动量交易，趋势交易，套利交易等，思路都不一样，今天利用例子来看怎么样建立一个动量的策略回测
点击第一个进入多因子排序算法，我目前是计算了680多个因子，基本覆盖常见的全部量化因子，闲理解一些策略的思路，因子的排序交易
"""
小果排序多因子回测系统
选择符合的买入因子，利用排序因子，对于bool值，True，False的因子类型，排序因子默认都是1
利用排序因子的权重，相关性计算，排序因子总权重，得到交易股票池，利用持股限制，买入数量
选择买入的股票池，对于持股没有在买入股票池的标的卖出，买入股票池里面的标的没有在持股买入
作者: xg_quant
因子值说明:
==========
- Bool类型因子 (金叉/死叉、价格在均线上等): 0=符合条件(True), 1=不符合条件(False)
- 数值类型因子 (连续上涨天数、连续下跌天数、价格等): 正常数值比较
策略逻辑:
==========
第一步: 根据买入条件选择符合条件的股票池 (and/or/not逻辑)初步形成交易股票池
第二步: 对选出的股票按排序因子进行排序，对因子采用正相关，负相关，权重得到全部排序因子的值，在排序
第三步: 根据因子加权排序得到的股票池作为交易股票池，安买入排名选择前排名的股票池
第四步: 先卖出后买入，卖出，持股的标的不在排序股票池卖出，排序股票池不在持股里面的标的买入，保持买入排名数量不变
因子条件说明:
==========
选择类型: and (全部满足) / or (至少满足一个) / not (不满足)
选择方向: 等于 / 大于 / 小于
值: 比较的目标值 (支持数值和Bool值0/1)
交易方式的说明，
小果排序多因子量化回测系统
第一步: 根据买入条件选择符合条件的股票池 (and/or/not逻辑)初步形成交易股票池
第二步: 对选出的股票按排序因子进行排序，对因子采用正相关，负相关，权重得到全部排序因子的值，在排序
第三步: 根据因子加权排序得到的股票池作为交易股票池，安买入排名选择前排名的股票池
第四步: 先卖出后买入，卖出，持股的标的不在排序股票池卖出，排序股票池不在持股里面的标的买入，保持买入排名数量不变
因子条件说明:
==========
选择类型: and (全部满足) / or (至少满足一个) / not (不满足)
选择方向: 等于 / 大于 / 小于
值: 比较的目标值 (支持数值和Bool)
重点需要处理一些排序因子，对于排序因子值是True，flase需要把因子值全部改成1好加权计算，因为0*其他数都是0
重点注意返回的是boll，True，和False,在做因子买卖选择的时候
True,其他值也是一样的意思表示真比如'T','true',字符串的'TRUE',
False 其他值也是应该意思表示假字符串的'F','false','FALSE'
下面是一个简单的教程，怎么样去建立一个动量排序的策略回测，第一步选择我们需要建立的股票池
点击添加就可以
第二部建立我们需要自定义因子，如果我系统里面默认的因子不满足的话可以自定义因子回测
选择基础因子加工，上面我给了默认的常见的计算公式，和通达信差不多
第三步建立选择买入因子形成第一步股票池，可以自己添加自己需要的因子去设计回测，检验自己的思路
后面就是排序的因子，可以按大到小，小到大形成交易的股票池排序
下面就是交易参数的设置，比如买入排名前N的，每一个标的交易多少形成最终的交易股票池
最后是轮动交易，先卖出后买入，卖出，持股的标的不在排序股票池卖出，排序股票池不在持股里面的标的买入，保持买入排名数量不变，点击界面的回测就可以，这样就非常简单的构建了一个动量策略的回测，检验了我们自己的策略思路
回测分析，可以很方便检验自己的策略思路，600多个因子支持
策略的年化分析
可以看见我们自己设计的算法的成交数据
持股明细等数据
今天就介绍完成了怎么样去检验动量策略的思路，后面介绍更多常见的交易算法思路，不懂的问我就可以，加我可以入量化交流群
可以开户找我，专业技术支持
常见的基础函数计算方式
'''
小果
微信:xg_quant
'''
import pandas as pd
import numpy as np
#------------------ 0级：核心工具函数 -------------------------------------------
import numpy as np
import pandas as pd
def RD(N, D=3):
"""四舍五入取3位小数"""
return np.round(N, D)
def RET(S, N=1):
"""返回序列倒数第N个值，默认返回最后一个"""
return np.array(S)[-N]
def ABS(S):
"""返回N的绝对值"""
return np.abs(S)
def MAX(S1, S2):
"""序列max"""
return np.maximum(S1, S2)
def MIN(S1, S2):
"""序列min"""
return np.minimum(S1, S2)
def IF(S, A, B):
"""序列布尔判断 return=A if S==True else B"""
return np.where(S, A, B)
def REF(S, N=1):
"""对序列整体下移动N，返回序列(shift后会产生NAN)"""
return pd.Series(S).shift(N).values
def DIFF(S, N=1):
"""前一个值减后一个值，前面会产生nan；np.diff(S)直接删除nan，会少一行"""
return pd.Series(S).diff(N).values
def STD(S, N):
"""求序列的N日标准差，返回序列"""
return pd.Series(S).rolling(N).std(ddof=0).values
def SUM(S, N):
"""对序列求N天累计和，返回序列；N=0对序列所有依次求和"""
return pd.Series(S).rolling(N).sum().values if N > 0 else pd.Series(S).cumsum().values
def CONST(S):
"""返回序列S最后的值组成常量序列"""
return np.full(len(S), S[-1])
def AND(S1, S2):
"""逻辑与运算"""
return np.logical_and(S1, S2)
def OR(S1, S2):
"""逻辑或运算"""
return np.logical_or(S1, S2)
def NOT(S1):
"""逻辑非运算"""
return np.logical_not(S1)
def RANGE(A, B, C):
"""期间函数：B <= A <= C"""
df = pd.DataFrame()
df['select'] = A.tolist()
df['select'] = df['select'].apply(lambda x: True if (x >= B and x <= C) else False)
return df['select']
def HHV(S, N):
"""HHV(C, 5) 最近5天收盘最高价"""
return pd.Series(S).rolling(N).max().values
def LLV(S, N):
"""LLV(C, 5) 最近5天收盘最低价"""
return pd.Series(S).rolling(N).min().values
def HHVBARS(S, N):
"""求N周期内S最高值到当前周期数，返回序列"""
return pd.Series(S).rolling(N).apply(lambda x: np.argmax(x[::-1]), raw=True).values
def LLVBARS(S, N):
"""求N周期内S最低值到当前周期数，返回序列"""
return pd.Series(S).rolling(N).apply(lambda x: np.argmin(x[::-1]), raw=True).values
def MA(S, N):
"""求序列的N日简单移动平均值，返回序列"""
return pd.Series(S).rolling(N).mean().values
def EMA(S, N):
"""指数移动平均，为了精度 S>4*N，EMA至少需要120周期；alpha=2/(span+1)"""
return pd.Series(S).ewm(span=N, adjust=False).mean().values
def SMA(S, N, M=1):
"""中国式的SMA，至少需要120周期才精确(雪球180周期)；alpha=1/(1+com)"""
return pd.Series(S).ewm(alpha=M/N, adjust=False).mean().values # com=N-M/M
def DMA(S, A):
"""求S的动态移动平均，A作平滑因子，必须 0<A<1 (此为核心函数，非指标)"""
return pd.Series(S).ewm(alpha=A, adjust=True).mean().values
def WMA(S, N):
"""通达信S序列的N日加权移动平均 Yn = (1*X1+2*X2+3*X3+...+n*Xn)/(1+2+3+...+Xn)"""
return pd.Series(S).rolling(N).apply(lambda x: x[::-1].cumsum().sum() * 2 / N / (N + 1), raw=True).values
def AVEDEV(S, N):
"""平均绝对偏差 (序列与其平均值的绝对差的平均值)"""
return pd.Series(S).rolling(N).apply(lambda x: (np.abs(x - x.mean())).mean()).values
def SLOPE(S, N):
"""返回S序列N周期回线性回归斜率"""
return pd.Series(S).rolling(N).apply(lambda x: np.polyfit(range(N), x, deg=1)[0], raw=True).values
def FORCAST(S, N):
"""返回S序列N周期回线性回归后的预测值"""
return pd.Series(S).rolling(N).apply(lambda x: np.polyval(np.polyfit(range(N), x, deg=1), N-1), raw=True).values
def LAST(S, A, B):
"""从前A日到前B日一直满足S_BOOL条件，要求A>B & A>0 & B>=0"""
return np.array(pd.Series(S).rolling(A+1).apply(lambda x: np.all(x[::-1][B:]), raw=True), dtype=bool)
#------------------ 1级：应用层函数(通过0级核心函数实现）--------------------------
def COUNT(S, N):
"""COUNT(CLOSE>O, N): 最近N天满足S_BOO的天数，True的天数"""
return SUM(S, N)
def EVERY(S, N):
"""EVERY(CLOSE>O, 5) 最近N天是否都是True"""
return IF(SUM(S, N) == N, True, False)
def EXIST(S, N):
"""EXIST(CLOSE>3010, N=5) n日内是否存在一天大于3000点"""
return IF(SUM(S, N) > 0, True, False)
def FILTER(S, N):
"""
FILTER函数，S满足条件后，将其后N周期内的数据置为0
例：FILTER(C==H,5) 涨停后，后5天不再发出信号
"""
for i in range(len(S)):
if S[i]:
S[i+1:i+1+N] = 0
return S
def BARSLAST(S):
"""上一次条件成立到当前的周期，BARSLAST(C/REF(C,1)>=1.1) 上一次涨停到今天的天数"""
M = np.concatenate(([0], np.where(S, 1, 0)))
for i in range(1, len(M)):
M[i] = 0 if M[i] else M[i-1] + 1
return M[1:]
def BARSLASTCOUNT(S):
"""统计连续满足S条件的周期数；BARSLASTCOUNT(CLOSE>OPEN)表示统计连续收阳的周期数"""
rt = np.zeros(len(S) + 1)
for i in range(len(S)):
rt[i+1] = rt[i] + 1 if S[i] else rt[i+1]
return rt[1:]
def BARSSINCEN(S, N):
"""N周期内第一次S条件成立到现在的周期数，N为常量"""
return pd.Series(S).rolling(N).apply(
lambda x: N-1-np.argmax(x) if np.argmax(x) or x[0] else 0,
raw=True
).fillna(0).values.astype(int)
def CROSS(S1, S2):
"""判断向上金叉穿越 CROSS(MA(C,5), MA(C,10))；判断向下死叉穿越 CROSS(MA(C,10), MA(C,5))"""
return np.concatenate(([False], np.logical_not((S1 > S2)[:-1]) & (S1 > S2)[1:]))
def CROSS_UP(S1, S2):
"""判断向上金叉穿越 CROSS(MA(C,5), MA(C,10))"""
return np.concatenate(([False], np.logical_not((S1 > S2)[:-1]) & (S1 > S2)[1:]))
def CROSS_DOWN(S1, S2):
"""判断向下死叉穿越 CROSS(MA(C,5), MA(C,10))"""
return np.concatenate(([False], np.logical_not((S1 < S2)[:-1]) & (S1 < S2)[1:]))
def LONGCROSS(S1, S2, N):
"""两条线维持一定周期后交叉，S1在N周期内都小于S2，本周期从S1下方向上穿过S2时返回1，否则返回0；N=1时等同于CROSS(S1, S2)"""
return np.array(np.logical_and(LAST(S1 < S2, N, 1), (S1 > S2)), dtype=bool)
def VALUEWHEN(S, X):
"""当S条件成立时，取X的当前值，否则取VALUEWHEN的上个成立时的X值"""
return pd.Series(np.where(S, X, np.nan)).ffill().values
#------------------ 扩展函数（来自第二个文件）-------------------------------------
def BACKSET(X, N):
"""
属于未来函数，将当前位置到若干周期前的数据设为1。
用法：BACKSET(X,N)，若X非0，则将当前位置到N周期前的数值设为1。
例如：BACKSET(CLOSE>OPEN,2) 若收阳则将该周期及前一周期数值设为1，否则为0
"""
result = np.zeros_like(X)
for i in range(len(X)):
if X[i] != 0:
start_index = max(0, i - N + 1)
result[start_index:i+1] = 1
return result
def ALIGNRIGHT(X):
"""
有效数据右对齐。
用法：ALIGNRIGHT(X) 有效数据向右移动，左边空出来的周期填充无效值
例如：TC:=IF(CURRBARSCOUNT=2 || CURRBARSCOUNT=5, DRAWNULL, C); XC:=ALIGNRIGHT(TC);
删除了两天的收盘价，并将剩余数据右移
"""
valid_indices = np.where(X != np.nan)[0]
invalid_count = len(X) - len(valid_indices)
result = np.empty_like(X)
result[:] = np.nan
result[invalid_count:len(valid_indices)+invalid_count] = X[valid_indices]
return result
def BARSCOUNT(X):
"""
有效数据周期数。
用法：BARSCOUNT(X) 第一个有效数据到当前的间隔周期数
注意：判断范围为指标或条件选股计算时公式使用的数据，
如果给画线指标的数据少(比如没有按下箭头取更多K线)或给条件选股给的数据少，这个有效值也可能少
"""
valid_indices = np.where(~np.isnan(X))[0]
if len(valid_indices) == 0:
return 0
first_valid_index = valid_indices[0]
current_index = len(X) - 1
bars_count = current_index - first_valid_index + 1
return bars_count
def BARSLASTS(X, N):
"""
倒数第N次成立时距今的周期数。
用法：BARSLASTS(X,N): X倒数第N满足到现在的周期数，N支持变量
"""
valid_indices = np.where(~np.isnan(X))[0]
if len(valid_indices) == 0:
return -1
last_n_indices = valid_indices[-N:]
if len(last_n_indices) < N:
return -1
current_index = len(X) - 1
bars_since_last_n = current_index - last_n_indices[-1] + 1
return bars_since_last_n
def ZIG(CLOSE, X=0.05):
"""
未来函数，计算之字转向。
用法：ZIG(CLOSE, 0.05) 5%之字转向
"""
ZIG_STATE_START = 0
ZIG_STATE_RISE = 1
ZIG_STATE_FALL = 2
x = X
k = CLOSE
peer_i = 0
candidate_i = None
scan_i = 0
peers = [0]
z = np.zeros(len(k))
state = ZIG_STATE_START
while True:
scan_i += 1
if scan_i == len(k) - 1:
if candidate_i is None:
peer_i = scan_i
peers.append(peer_i)
else:
if state == ZIG_STATE_RISE:
if k[scan_i] >= k[candidate_i]:
peer_i = scan_i
peers.append(peer_i)
else:
peer_i = candidate_i
peers.append(peer_i)
peer_i = scan_i
peers.append(peer_i)
elif state == ZIG_STATE_FALL:
if k[scan_i] <= k[candidate_i]:
peer_i = scan_i
peers.append(peer_i)
else:
peer_i = candidate_i
peers.append(peer_i)
peer_i = scan_i
peers.append(peer_i)
break
if state == ZIG_STATE_START:
if k[scan_i] >= k[peer_i] * (1 + x):
candidate_i = scan_i
state = ZIG_STATE_RISE
elif k[scan_i] <= k[peer_i] * (1 - x):
candidate_i = scan_i
state = ZIG_STATE_FALL
elif state == ZIG_STATE_RISE:
if k[scan_i] >= k[candidate_i]:
candidate_i = scan_i
elif k[scan_i] <= k[candidate_i] * (1 - x):
peer_i = candidate_i
peers.append(peer_i)
state = ZIG_STATE_FALL
candidate_i = scan_i
elif state == ZIG_STATE_FALL:
if k[scan_i] <= k[candidate_i]:
candidate_i = scan_i
elif k[scan_i] >= k[candidate_i] * (1 + x):
peer_i = candidate_i
peers.append(peer_i)
state = ZIG_STATE_RISE
candidate_i = scan_i
for i in range(len(peers) - 1):
peer_start_i = peers[i]
peer_end_i = peers[i + 1]
start_value = k[peer_start_i]
end_value = k[peer_end_i]
a = (end_value - start_value) / (peer_end_i - peer_start_i)
for j in range(peer_end_i - peer_start_i + 1):
z[j + peer_start_i] = start_value + a * j
return pd.Series(z)
