3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
编写基于 ROC(Rate of Change,变动速率)指标的动量策略,核心思想是买入过去一段时间内涨幅最大(动量最强)的股票,并定期进行调仓。
# -*- 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 计算:
(hist_data.iloc[-1] - hist_data.iloc[0]) / hist_data.iloc[0],这比使用 for 循环遍历每只股票效率高得多。过滤机制 (filter_stocks):
交易执行 (order_target_value):
order_target_value 可以自动计算需要买卖的数量,将持仓调整到目标金额。如果已持有该股票且价格上涨,它会自动卖出部分以止盈(再平衡);如果未持有,则买入。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 动量策略属于趋势跟踪策略,适合在牛市或震荡上行的市场环境中表现较好。在快速轮动的震荡市或熊市中,可能会出现“追涨杀跌”导致回撤较大。建议结合大盘择时(如均线择时)来控制风险。