动量排名策略

一、交易策略解释

核心思想

动量排名策略的核心思想基于"价格延续性"原理，即表现强势的资产在短期内倾向于继续表现强势，表现弱势的资产则倾向于继续表现弱势。该策略通过计算各个交易品种在特定时间窗口内的价格变动率（收益率），对所有候选品种进行排名，选择收益率最高的N个品种建立多头头寸，实现择强而投的目标。

理论基础

动量效应是金融市场中经过大量研究证实的异象之一，主要学术依据包括：

Jegadeesh和Titman（1993）的研究表明，过去3-12个月表现最好的股票在未来3-12个月继续跑赢大盘
Moskowitz、Ooi和Pedersen（2012）的研究证实，动量效应存在于全球不同资产类别，包括期货市场
Asness、Moskowitz和Pedersen（2013）进一步证明了动量策略在国际股票、货币、国债和商品市场的有效性

动量效应形成的原因可能包括：

投资者对新信息反应不足
行为偏差导致趋势延续
市场参与者追逐强势品种
交易商风险管理导致的自我强化趋势

策略适用场景

最佳使用环境：

趋势明显的市场环境
各品种之间相关性较低的市场
商品期货等波动较大的品种
中长期投资时间框架（如数周至数月）
有明确宏观经济驱动因素的市场

不宜使用的市场条件：

高度震荡的横盘市场
市场恐慌或流动性危机时期
高度相关的市场环境
基本面突变的极端事件（如疫情爆发初期）
政策频繁干预的市场

二、天勤介绍

天勤平台概述

天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特点：

丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。

策略开发流程

环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制

三、天勤实现策略

策略原理

动量排名策略的核心指标是价格动量，计算公式如下：

动量得分 = (当前价格 / N天前价格 - 1) × 100%

其中N是回看天数，决定了动量的时间窗口。本策略使用15天作为回看期，这个周期足够捕捉短期趋势，又不至于过度滞后。

排名机制：

计算所有候选品种的动量得分
按动量得分从高到低排序
选择排名最高的TOP_N个品种建立等权重头寸

交易逻辑

开仓信号：

当品种的动量排名进入前TOP_N名时（本例中为前2名）
立即建立固定规模的多头头寸（本例中每个品种100手）

平仓信号：

当品种的动量排名跌出前TOP_N名时
立即平掉全部头寸

头寸管理：

采用等资金分配方式，为每个入选品种分配相同的头寸规模
每日收盘后重新评估动量排名，调整持仓

策略不包含止损机制，完全依靠动量排名进行头寸管理，这是一个可以优化的方向。

回测

回测初始设置

测试周期： 2023 年 10 月 1 日至 2023 年 12 月 31 日
品种选择： 选择流动性好、波动适中的8个期货品种
- 上海期货交易所：螺纹钢、橡胶、铜、镍
- 大连商品交易所：豆粕、棕榈油
- 郑州商品交易所：甲醇、PTA
参数配置：
- 回看天数(LOOKBACK_DAYS)：15天
- 选择品种数(TOP_N)：2个
- 每个品种交易手数(POSITION_SIZE)：100手

回测结果

上述回测累计收益走势图

完整代码示例

#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Chaos"

from datetime import date
from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished

# 参数设置
SYMBOLS = [
    # 流动性好、波动适中的品种
    "SHFE.rb2405", "SHFE.ru2405",  # 螺纹橡胶
    "DCE.m2405", "DCE.p2405",      # 豆粕棕榈
    "CZCE.MA405", "CZCE.TA405",    # 甲醇PTA
    "SHFE.cu2405", "SHFE.ni2405"   # 铜镍
]
LOOKBACK_DAYS = 15  # 动量计算周期(回看天数)
TOP_N = 2  # 选择排名最高的N个品种做多
POSITION_SIZE = 100  # 每个品种的交易手数
START_DATE = date(2023, 10, 1)
END_DATE = date(2023, 12, 31)

# 创建API实例，设置回测区间和认证信息
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
            auth=TqAuth("快期账号", "快期密码"))

# 订阅行情数据
klines = {}  # 存储K线数据的字典
quotes = {}  # 存储报价数据的字典
positions = {}  # 存储持仓信息的字典
target_pos = {}  # 存储目标持仓任务的字典

for symbol in SYMBOLS:
    klines[symbol] = api.get_kline_serial(symbol, 24 * 60 * 60)  # 获取日线数据
    quotes[symbol] = api.get_quote(symbol)  # 获取实时报价
    target_pos[symbol] = TargetPosTask(api, symbol)  # 创建目标持仓任务

# 当前持仓的品种集合
current_holdings = set()

print(f"策略启动: 动量排名策略")
print(f"参数设置: 回看期={LOOKBACK_DAYS}天, 选择前{TOP_N}名品种, 每个品种{POSITION_SIZE}手")

def calculate_momentum():
    """计算所有品种的动量并排名"""
    momentum_scores = {}

    for symbol in SYMBOLS:
        # 确保有足够的历史数据
        if len(klines[symbol].close) >= LOOKBACK_DAYS:
            # 计算过去N天的收益率作为动量分数
            returns = (klines[symbol].close.iloc[-1] / klines[symbol].close.iloc[-LOOKBACK_DAYS] - 1) * 100
            momentum_scores[symbol] = returns

    # 按动量分数降序排序
    ranked_symbols = sorted(momentum_scores.items(), key=lambda x: x[1], reverse=True)
    return ranked_symbols

def update_positions(ranked_symbols):
    """更新持仓"""
    # 获取应该持有的品种
    new_holdings = set([symbol for symbol, score in ranked_symbols[:TOP_N]])

    # 平掉不在新持仓中的品种
    for symbol in current_holdings - new_holdings:
        target_pos[symbol].set_target_volume(0)
        print(f"平仓: {symbol}, 动量得分不再排名前{TOP_N}")

    # 建立新的持仓
    for symbol in new_holdings - current_holdings:
        target_pos[symbol].set_target_volume(POSITION_SIZE)
        print(f"开仓: {symbol}, 动量得分排名前{TOP_N}")

    # 更新当前持仓集合
    current_holdings.clear()
    current_holdings.update(new_holdings)

try:
    day_count = 0  # 用于追踪交易日

    while True:
        api.wait_update()  # 等待数据更新

        # 判断是否有新的一天开始
        for symbol in SYMBOLS:
            if api.is_changing(klines[symbol].iloc[-1], "datetime"):
                day_count += 1
                print(f"\n=== 第{day_count}个交易日 ===")

                # 确保有足够的历史数据
                if day_count >= LOOKBACK_DAYS:
                    # 计算动量并排名
                    ranked_symbols = calculate_momentum()

                    # 打印动量排名
                    print("\n动量排名:")
                    for symbol, score in ranked_symbols:
                        print(f"{symbol}: {score:.2f}%")

                    # 更新持仓
                    update_positions(ranked_symbols)
                break

except BacktestFinished as e:
    api.close()
    print("回测结束")