策略研究中，我们经常会看见2种常见的交易算法，因子的排序交易轮动，和因子的选择交易2种算法，因子的排序交易，按因子的排名来轮动交易的一种方式，因子的选择交易就是按因子的买入条件，排序条件，卖出条件选择标的交易，和因子排序交易是2种算法，常见的交易方式有动量交易，趋势交易，套利交易等，思路都不一样，今天利用例子来看怎么样建立一个动量的策略回测

点击第一个进入多因子排序算法，我目前是计算了680多个因子，基本覆盖常见的全部量化因子，闲理解一些策略的思路，因子的排序交易


"""
小果排序多因子回测系统
选择符合的买入因子，利用排序因子，对于bool值，True，False的因子类型，排序因子默认都是1
利用排序因子的权重，相关性计算，排序因子总权重，得到交易股票池，利用持股限制，买入数量
选择买入的股票池，对于持股没有在买入股票池的标的卖出，买入股票池里面的标的没有在持股买入
因子值说明:
==========
- 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多个因子支持

可以看见我们自己设计的算法的成交数据

今天就介绍完成了怎么样去检验动量策略的思路，后面介绍更多常见的交易算法思路，


'''
小果
'''
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)


动量策略回测交易全解析：信号、执行与绩效评估

投资顾问

算法

多因子排序交易与因子选择交易是两种常见量化策略，本文通过680+因子库演示动量策略回测构建流程，涵盖买入条件筛选、排序因子加权、轮动交易参数设置及回测分析，帮助投资者快速验证策略思路。

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

动量策略回测交易全解析：信号、执行与绩效评估

动量策略回测交易全解析：信号、执行与绩效评估

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐