一、交易策略解释
核心思想
动量排名策略的核心思想基于"价格延续性"原理，即表现强势的资产在短期内倾向于继续表现强势，表现弱势的资产则倾向于继续表现弱势。 该策略通过计算各个交易品种在特定时间窗口内的价格变动率（收益率），对所有候选品种进行排名，选择收益率最高的N个品种建立多头头寸，实现择强而投的目标。
理论基础
动量效应是金融市场中经过大量研究证实的异象之一，主要学术依据包括：
- Jegadeesh和Titman（1993）的研究表明，过去3-12个月表现最好的股票在未来3-12个月继续跑赢大盘
- Moskowitz、Ooi和Pedersen（2012）的研究证实，动量效应存在于全球不同资产类别，包括期货市场
- Asness、Moskowitz和Pedersen（2013）进一步证明了动量策略在国际股票、货币、国债和商品市场的有效性
动量效应形成的原因可能包括：
- 投资者对新信息反应不足
- 行为偏差导致趋势延续
- 市场参与者追逐强势品种
- 交易商风险管理导致的自我强化趋势
策略适用场景
最佳使用环境：
- 趋势明显的市场环境
- 各品种之间相关性较低的市场
- 商品期货等波动较大的品种
- 中长期投资时间框架（如数周至数月）
- 有明确宏观经济驱动因素的市场
不宜使用的市场条件：
- 高度震荡的横盘市场
- 市场恐慌或流动性危机时期
- 高度相关的市场环境
- 基本面突变的极端事件（如疫情爆发初期）
- 政策频繁干预的市场
策略原理
动量排名策略的核心指标是价格动量，计算公式如下：
动量得分 = (当前价格 / 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("回测结束")