问题描述
解决方案
这是一个基于 JoinQuant (聚宽) 平台的“追涨杀跌”策略(学术上通常称为动量策略 Momentum Strategy)。
策略逻辑
这个策略的核心思想是:强者恒强,弱者恒弱。
- 追涨(买入逻辑):计算股票池中所有股票过去 N 天(例如20天)的涨幅,买入涨幅最高的几只股票。
- 杀跌(卖出逻辑):如果持仓的股票涨幅不再排名靠前(动量减弱),或者跌出了目标列表,则将其卖出。
- 股票池:选用沪深300成分股(流动性好,避免小盘股陷阱)。
- 调仓频率:按天进行监测和调仓。
策略代码
# -*- coding: utf-8 -*-
from jqdata import *
def initialize(context):
"""
初始化函数,设定基准、手续费、全局变量等
"""
# 设定沪深300作为基准
set_benchmark('000300.XSHG')
# 开开启真实价格模式(动态复权)
set_option('use_real_price', True)
# 过滤掉order系列API产生的比error级别低的log
log.set_level('order', 'error')
# 设定手续费:买入万三,卖出万三加千一印花税
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
# --- 策略参数设置 ---
g.security_pool = '000300.XSHG' # 股票池:沪深300
g.momentum_days = 20 # 追涨周期:过去20天的涨幅
g.stock_num = 5 # 持仓数量:持有涨幅最好的5只
# 每天开盘时运行
run_daily(market_open, time='09:30')
def check_stocks(context):
"""
筛选股票:去除停牌、ST、退市、涨跌停的股票
"""
# 获取股票池成分股
stocks = get_index_stocks(g.security_pool)
# 获取当前时刻的股票数据(判断停牌、ST等)
current_data = get_current_data()
final_stocks = []
for code in stocks:
# 过滤停牌和ST
if current_data[code].paused or current_data[code].is_st:
continue
# 过滤涨停(无法买入)或跌停(无法卖出)
if current_data[code].day_open >= current_data[code].high_limit or \
current_data[code].day_open <= current_data[code].low_limit:
continue
final_stocks.append(code)
return final_stocks
def get_momentum_list(context, stocks):
"""
计算动量并排序
"""
score_list = []
# 获取历史收盘价数据
# count = g.momentum_days + 1 是为了计算 (今天收盘 - N天前收盘) / N天前收盘
# 注意:在盘中运行 '09:30' 时,取不到当天的收盘价,这里取的是截止到昨天的历史数据
hist_data = history(g.momentum_days + 1, unit='1d', field='close', security_list=stocks, df=True)
for code in stocks:
# 确保数据长度足够
if len(hist_data[code]) < g.momentum_days + 1:
continue
# 计算收益率:(昨天收盘价 - N天前收盘价) / N天前收盘价
prev_close = hist_data[code][-1]
start_close = hist_data[code][0]
if start_close == 0: # 避免除以0
continue
ret = (prev_close - start_close) / start_close
score_list.append((code, ret))
# 按收益率从大到小排序
score_list.sort(key=lambda x: x[1], reverse=True)
# 返回排名前 g.stock_num 的股票代码列表
top_stocks = [x[0] for x in score_list[:g.stock_num]]
return top_stocks
def market_open(context):
"""
每日交易逻辑
"""
# 1. 筛选符合条件的股票池
valid_stocks = check_stocks(context)
# 2. 获取动量最高的股票列表(追涨目标)
buy_list = get_momentum_list(context, valid_stocks)
# 3. 获取当前持仓
current_holdings = list(context.portfolio.positions.keys())
# 4. 卖出逻辑(杀跌):如果持仓股票不在买入列表中,说明它涨势不再强劲,卖出
for stock in current_holdings:
if stock not in buy_list:
order_target_value(stock, 0)
log.info("杀跌卖出: %s" % stock)
# 5. 买入逻辑(追涨):买入在榜单中的股票
# 计算每只股票应该分配的资金(等权重分配)
# 注意:这里使用 total_value 可能会导致满仓波动,实际操作可预留现金
position_count = len(buy_list)
if position_count > 0:
value_per_stock = context.portfolio.total_value / position_count
for stock in buy_list:
# 如果已经在持仓中,order_target_value 会自动调整仓位(加仓或减仓到目标值)
# 如果不在持仓中,则开仓
order_target_value(stock, value_per_stock)
# 记录日志(仅当是新开仓时记录比较清晰,这里简化处理)
if stock not in current_holdings:
log.info("追涨买入: %s" % stock)
代码详解
-
initialize:- 设置了基准为沪深300。
- 设置了
g.momentum_days = 20,意味着我们看过去20个交易日的表现。 - 设置了
g.stock_num = 5,即只持有表现最好的5只股票。
-
check_stocks:- 这是一个风控过滤函数。它剔除了停牌、ST(风险警示)以及开盘就涨跌停的股票,防止策略买入无法交易或高风险的标的。
-
get_momentum_list:- 这是策略的核心。它使用
history函数获取过去 N 天的价格。 - 计算公式:
(昨天收盘价 - N天前收盘价) / N天前收盘价。 - 将所有股票按涨幅从高到低排序,取前5名。这就是“追涨”的量化体现。
- 这是策略的核心。它使用
-
market_open:- 卖出(杀跌):检查当前持仓的股票,如果它已经不在前5名列表里了(说明它涨不动了,或者跌了),直接清仓。
- 买入(追涨):对前5名的股票进行等权重买入。
如何优化这个策略?
如果你想让这个策略更完善,可以考虑以下几点:
- 止损机制:目前的“杀跌”是基于排名的。你可以加入硬性止损,例如:如果单只股票亏损超过 8%,无视排名强制卖出。
- 大盘择时:在
market_open开头加入判断,如果沪深300指数跌破了60日均线(大盘走熊),则清空所有仓位,空仓避险。 - 调仓频率:目前是每天调仓,手续费可能较高。可以改为
run_weekly每周调仓一次。
Q&A
Q: 为什么选择沪深300而不是全市场股票?
A: 全市场包含很多小盘股,波动极大且容易受流动性影响。对于动量策略,沪深300成分股流动性好,回测结果更接近实盘真实情况。
Q: 什么是“动态复权”?
A: 代码中 set_option('use_real_price', True) 开启了真实价格模式。这在回测中非常重要,因为它能正确处理分红和拆股,避免因为股票除权导致的价格“假摔”触发错误的卖出信号。
Q: 这个策略最大的风险是什么?
A: 动量策略最大的风险是市场震荡。在趋势不明显的震荡市中,股票可能今天涨明天跌,导致策略频繁高买低卖(左右挨打),造成本金磨损。