问题描述
解决方案
编写基于 ROC(Rate of Change,变动速率)指标的动量策略,核心思想是买入过去一段时间内涨幅最大(动量最强)的股票,并定期进行调仓。
策略逻辑说明
- 股票池:选取沪深300指数成分股。
- ROC指标计算:
- 公式:$ROC = \frac{Close_{today} - Close_{n}}{Close_{n}} \times 100%$
- 其中 $Close_{today}$ 为当前收盘价,$Close_{n}$ 为 $N$ 天前的收盘价。
- 本策略设定 $N=20$(即20日ROC)。
- 选股逻辑:
- 计算股票池内所有股票的 20日 ROC。
- 剔除停牌、ST、涨跌停股票。
- 选取 ROC 值最大的前 10 只股票。
- 交易逻辑:
- 调仓频率:按月调仓(每月第一个交易日)。
- 卖出:卖出不在新选出的 10 只股票名单中的持仓。
- 买入:买入新选出的股票,资金等权分配。
策略代码实现
# -*- coding: utf-8 -*-
# 导入聚宽函数库
import jqdata
import pandas as pd
def initialize(context):
"""
初始化函数,设定基准、全局变量、运行频率等
"""
# 设定沪深300作为基准
set_benchmark('000300.XSHG')
# 开启动态复权模式(真实价格)
set_option('use_real_price', True)
# 过滤掉order系列API产生的比error级别低的log
log.set_level('order', 'error')
# --- 策略参数设置 ---
g.index_security = '000300.XSHG' # 股票池基准
g.N = 20 # ROC计算周期 (20天)
g.stock_num = 10 # 持仓股票数量
# 设定交易费率:买入万分之三,卖出万分之三加千分之一印花税
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
# 设定按月运行,每月第一个交易日开盘后运行
run_monthly(rebalance, monthday=1, time='09:30')
def rebalance(context):
"""
调仓主函数
"""
# 1. 获取股票池(沪深300)
stocks = get_index_stocks(g.index_security)
# 2. 过滤股票(停牌、ST、涨跌停、上市时间不足)
stocks = filter_stocks(context, stocks)
# 3. 计算ROC并排序
# 获取过去 N+1 天的收盘价数据(因为要计算变动率,需要用到N天前的数据)
# history返回的是DataFrame,行是时间,列是股票代码
hist_data = history(g.N + 1, unit='1d', field='close', security_list=stocks)
# 计算ROC: (当前价格 - N天前价格) / N天前价格
# iloc[-1] 是最新一天数据,iloc[0] 是N天前数据
roc_series = (hist_data.iloc[-1] - hist_data.iloc[0]) / hist_data.iloc[0]
# 去除计算结果为NaN的股票
roc_series = roc_series.dropna()
# 按ROC从大到小排序,取前 g.stock_num 只
buy_list = roc_series.sort_values(ascending=False)[:g.stock_num].index.tolist()
log.info("今日选股(Top %d ROC): %s" % (g.stock_num, buy_list))
# 4. 执行交易
trade(context, buy_list)
def trade(context, buy_list):
"""
交易执行函数
"""
# 获取当前持仓
current_positions = list(context.portfolio.positions.keys())
# 卖出逻辑:如果当前持仓不在买入列表中,则卖出
for stock in current_positions:
if stock not in buy_list:
order_target_value(stock, 0)
log.info("卖出: %s" % stock)
# 买入逻辑:对买入列表中的股票进行等权买入
if len(buy_list) > 0:
# 计算每只股票应该分配的资金(总资产 / 目标持仓数)
# 注意:这里使用 total_value 而不是 available_cash,是为了保持仓位平衡
position_value = context.portfolio.total_value / len(buy_list)
for stock in buy_list:
order_target_value(stock, position_value)
log.info("买入/调仓: %s, 目标市值: %.2f" % (stock, position_value))
def filter_stocks(context, stock_list):
"""
股票过滤函数:过滤停牌、ST、涨跌停、新股
"""
current_data = get_current_data()
filtered_stocks = []
for stock in stock_list:
# 过滤停牌
if current_data[stock].paused:
continue
# 过滤ST
if current_data[stock].is_st:
continue
# 过滤涨停(无法买入)
if current_data[stock].last_price >= current_data[stock].high_limit:
continue
# 过滤跌停(无法卖出,虽然选股选不到跌停的,但作为通用过滤加上)
if current_data[stock].last_price <= current_data[stock].low_limit:
continue
# 过滤上市时间不足60天的次新股(可选)
# start_date = get_security_info(stock).start_date
# if (context.current_dt.date() - start_date).days < 60:
# continue
filtered_stocks.append(stock)
return filtered_stocks
代码关键点解析
-
数据获取 (
history):- 使用
history(g.N + 1, ...)获取过去 $N+1$ 天的数据。这是因为计算 $N$ 日变动率需要用到 $T$ 日(今天)和 $T-N$ 日($N$天前)的价格。 hist_data.iloc[-1]代表最近一个交易日的收盘价。hist_data.iloc[0]代表 $N$ 天前的收盘价。
- 使用
-
ROC 计算:
- 代码中使用了 Pandas 的向量化运算
(hist_data.iloc[-1] - hist_data.iloc[0]) / hist_data.iloc[0],这比使用for循环遍历每只股票效率高得多。
- 代码中使用了 Pandas 的向量化运算
-
过滤机制 (
filter_stocks):- 在量化回测中,必须剔除无法交易的股票(如停牌、涨停无法买入)以及风险过大的股票(如ST),否则回测结果会虚高且无法实盘。
-
交易执行 (
order_target_value):- 使用
order_target_value可以自动计算需要买卖的数量,将持仓调整到目标金额。如果已持有该股票且价格上涨,它会自动卖出部分以止盈(再平衡);如果未持有,则买入。
- 使用
Q&A
Q: 如何修改 ROC 的计算周期?
A: 修改 initialize 函数中的 g.N 变量即可。例如,想计算 60 日动量,将 g.N = 20 改为 g.N = 60。
Q: 为什么使用 order_target_value 而不是 order_value?
A: order_target_value 是目标市值下单。在定期调仓策略中,我们需要让每只股票的持仓市值保持均等(例如总资金的10%)。如果某只股票上个月涨了很多,占比超过了10%,order_target_value 会自动卖出多余部分;如果跌了,会自动补仓。而 order_value 只是单纯买入指定金额,无法实现仓位的动态再平衡。
Q: 这个策略适合什么市场环境?
A: ROC 动量策略属于趋势跟踪策略,适合在牛市或震荡上行的市场环境中表现较好。在快速轮动的震荡市或熊市中,可能会出现“追涨杀跌”导致回撤较大。建议结合大盘择时(如均线择时)来控制风险。