策略名称：动态ETF轮动策略（基于大盘情绪监控）
策略说明：
本策略通过监控大盘情绪（成交量变化）和ETF的相对表现，动态选择表现最好的ETF进行投资。策略的核心思想是，当大盘情绪良好且某些ETF表现优异时，选择买入；当大盘情绪不佳或所有ETF表现均不理想时，选择空仓。策略结合了成交量监控、ETF轮动和动态调仓机制，旨在捕捉市场趋势并控制风险。
策略逻辑：
1. **参数设置与初始化**：
- 设置大盘情绪监控开关（`g.MonitorSwitch`），动态调整大盘热度参考指标（`g.use_dynamic_target_market`）。
- 定义交易标的（ETF及其对应的指数），并确保交易标的在上市足够时间后才参与交易。
- 初始化交易信号（`g.signal`）和市场热度（`g.emotion_rate`）。
2. **大盘情绪监控**：
- 使用成交量监控大盘情绪。计算大盘成交量的移动平均线（MA），并判断当前成交量是否连续多日低于或高于均线。
- 当成交量连续多日低于均线时，认为大盘情绪不佳，发出空仓信号；当成交量连续多日高于均线时，认为大盘情绪良好，允许开仓。
3. **ETF轮动与信号生成**：
- 计算每个ETF的周期涨幅和均线差值，筛选出表现最好的ETF。
- 当大盘情绪良好且某些ETF表现优异时，选择买入表现最好的ETF；当大盘情绪不佳或所有ETF表现均不理想时，选择空仓。
4. **交易执行**：
- 根据生成的交易信号，执行买入、卖出或调仓操作。
- 买入时，选择表现最好的ETF；卖出时，清空所有持仓。
策略参数：
- **大盘情绪监控开关（g.MonitorSwitch）**：True（开启监控）
- **动态调整大盘热度参考指标（g.use_dynamic_target_market）**：True（动态调整）
- **大盘指数（g.target_market）**：399001.XSHE（深证成指）
- **成交量监控周期（g.lag0）**：7天
- **成交量连续低于均线的天数（g.lag）**：6天
- **均线周期（g.lag1）**：13天
- **涨幅计算周期（g.lag2）**：13天
- **ETF标的**：包括深证100、沪深300、上证50、上证180、红利ETF、深红利、创业板、中证500等。
策略特点：
- **大盘情绪监控**：通过成交量监控大盘情绪，避免在大盘情绪不佳时持有仓位。
- **ETF轮动**：动态选择表现最好的ETF进行投资，捕捉市场趋势。
- **动态调仓**：根据市场情况动态调整仓位，避免在市场弱势时持有过多仓位。
适用场景：
- **趋势明显的市场**：当市场处于明显的上升或下降趋势时，策略表现较好。
- **成交量活跃的市场**：在成交量活跃的市场环境下，策略的信号更为有效。
风险提示：
- **滞后性**：策略基于历史数据生成信号，可能存在一定的滞后性。
- **市场波动**：在市场波动较大时，策略可能会频繁调仓，增加交易成本。
- **ETF流动性风险**：部分ETF可能存在流动性不足的问题，影响交易执行。
总结：
本策略通过监控大盘情绪和ETF的相对表现，动态选择表现最好的ETF进行投资，适合在趋势明显且成交量活跃的市场环境下使用。策略的核心在于通过成交量监控和ETF轮动机制，动态调整仓位，以期在市场上升时获得收益，在市场下跌时减少损失。
策略源码
# 标题：蒋老师吐槽了之后，我也想说说对ETF轮动策略的看法
# 作者：智习
# 标题：策略coding吐槽帖——以“网红”ETF轮动为例
# 作者：jqz1226
# 标题：无杠杆，稳定盈利的etf轮动，06年开始3000%收益
# 作者：热爱大自然
# 标题：etf轮动的优化，680%收益
# 作者：热爱大自然
# 标题：无杠杆，回撤更小的etf轮动
# 作者：智习
from jqdata import *
import pandas as pd
import talib
def initialize(context):
set_params()
#
set_option("avoid_future_data", True)
set_option('use_real_price', True) # 用真实价格交易
set_benchmark('000300.XSHG')
log.set_level('order', 'error')
#
# 将滑点设置为0
set_slippage(FixedSlippage(0))
# 手续费: 采用系统默认设置
set_order_cost(OrderCost(open_tax=0, close_tax=0, \
open_commission=0.0005, close_commission=0.0005,\
close_today_commission=0, min_commission=5), type='fund')
# 开盘前运行
run_daily(before_market_open, time='before_open', reference_security='000300.XSHG')
# 11:30 计算大盘信号
run_daily(get_signal, time='11:30')
# 14:40 进行交易
run_daily(ETFtrade, time='14:40')
# 设置参数
def set_params():
g.MonitorSwitch = True #大盘止损开关，设False表示关闭，True表示打开
g.use_dynamic_target_market = True # 是否动态改变大盘热度参考指标
# g.target_market = '000300.XSHG'
g.target_market = '399001.XSHE'
g.signal = 'BUY' # 交易信号初始化
g.emotion_rate = 0 # 市场热度
g.lag = 6 # 大盘成交量连续跌破均线的天数，发出空仓信号
g.lag0 = 7 # 大盘成交量监控周期
g.lag1 = 13 # 比价均线周期
g.lag2 = 13 # 价格涨幅计算周期
#
g.buy = [] # 购买股票列表
# 指数、基金对, 所有想交易的etf都可以，会自动过滤掉交易时没有上市的
g.ETF_targets = {
'399632.XSHE':'159901.XSHE',#深证100
'000300.XSHG':'510300.XSHG',#沪深300
'000016.XSHG':'510050.XSHG',#上证50
'000010.XSHG':'510180.XSHG',#上证180
'000015.XSHG':'510880.XSHG',#红利ETF
'399324.XSHE':'159905.XSHE',#深红利
'399006.XSHE':'159915.XSHE',#创业板
#'399673.XSHE':'159949.XSHE',#创业板50
'000905.XSHG':'510500.XSHG'#中证500
}
#
stocks_info = "\n股票池:\n"
for security in g.ETF_targets.values():
s_info = get_security_info(security)
stocks_info += "【%s】%s 上市日期:%s\n" % (s_info.code, s_info.display_name, s_info.start_date)
log.info(stocks_info)
def get_before_after_trade_days(date, count, is_before=True):
"""
来自： https://www.joinquant.com/view/community/detail/c9827c6126003147912f1b47967052d9?type=1
date :查询日期
count : 前后追朔的数量
is_before : True , 前count个交易日 ; False ,后count个交易日
返回 : 基于date的日期, 向前或者向后count个交易日的日期 ,一个datetime.date 对象
"""
all_date = pd.Series(get_all_trade_days())
if isinstance(date, str):
date = datetime.datetime.strptime(date, '%Y-%m-%d').date()
if isinstance(date, datetime.datetime):
date = date.date()
if is_before:
return all_date[all_date <= date].tail(count).values[0] else: return all_date[all_date >= date].head(count).values[-1]
def before_market_open(context):
# 确保交易标的已经上市g.lag1个交易日以上
yesterday = context.previous_date
list_date = get_before_after_trade_days(yesterday, g.lag1) # 今天的前g.lag1个交易日的日期
g.ETFList = {}
all_funds = get_all_securities(types='fund', date=yesterday) # 上个交易日之前上市的所有基金
all_idxes = get_all_securities(types='index', date=yesterday) # 上个交易日之前就已经存在的指数
for idx in g.ETF_targets:
if idx in all_idxes.index:
if all_idxes.loc[idx].start_date <= list_date: # 指数已经在要求的日期前上市
symbol = g.ETF_targets[idx]
if symbol in all_funds.index:
if all_funds.loc[symbol].start_date <= list_date: # 对应的基金也已经在要求的日期前上市 g.ETFList[idx] = symbol # 则列入可交易对象中 # return # 每日交易时 def ETFtrade(context): if g.signal == 'CLEAR': for stock in context.portfolio.positions: log.info("清仓: %s" % stock) order_target(stock, 0) elif g.signal == 'BUY': holdings = set(context.portfolio.positions.keys()) # 现在持仓的 targets = set(g.buy) # 想买的目标 # # 1. 卖出不在targets中的 sells = holdings - targets for code in sells: log.info("卖出: %s" % code) order_target(code, 0) # ratio = len(targets) cash = context.portfolio.total_value / ratio # 2. 交集部分调仓 adjusts = holdings & targets for code in adjusts: # 手续费最低5元，只有交易5000元以上时，交易成本才会低于千分之一，才去调仓 if abs(cash - context.portfolio.positions[code].value) > 5000:
log.info('调仓: %s' % code)
order_target_value(code, cash)
# 3. 新的，买入
purchases = targets - holdings
for code in purchases:
log.info('买入: %s' % code)
order_target_value(code, cash)
#
current_returns = 100 * context.portfolio.returns
log.info("当前收益：%.2f%%，当前持仓: %s", current_returns, list(context.portfolio.positions.keys()))
# 获取信号
def get_signal(context):
if g.signal == 'CLEAR':
for stock in context.portfolio.positions:
log.info("清仓遗留基金: %s" % stock)
order_target(stock, 0)
# 创建保持计算结果的DataFrame
df_etf = pd.DataFrame(columns=['基金代码', '对应指数', '周期涨幅', '均线差值'])
current_data = get_current_data()
for mkt_idx in g.ETFList:
security = g.ETFList[mkt_idx] # 指数对应的基金
# 获取股票的收盘价
close_data = attribute_history(security, g.lag1, '1d', ['close'], df=False)
# 获取股票现价
current_price = current_data[security].last_price
# 获取股票的阶段收盘价涨幅
cp_increase = (current_price / close_data['close'][g.lag2 - g.lag1] - 1) * 100
# 取得平均价格
ma_n1 = close_data['close'].mean()
# 计算前一收盘价与均值差值
pre_price = (current_price / ma_n1 - 1) * 100
df_etf = df_etf.append({'基金代码': security, '对应指数': mkt_idx, '周期涨幅': cp_increase, '均线差值': pre_price},
ignore_index=True)
# 按照涨幅降序排列
if len(df_etf) == 0:
log.info("大盘信号：没有可以交易的品种，清仓")
g.signal = 'CLEAR'
return
df_etf.sort_values(by='周期涨幅', ascending=False, inplace=True)
if g.use_dynamic_target_market:
g.target_market = df_etf['对应指数'].iloc[-1] # 用涨幅最小的指数来处理
if EmotionMonitor() == -1: # 大盘情绪不好发出空仓信号
log.info("交易信号:大盘成交量持续低均线，空仓")
g.signal = 'CLEAR'
return
log.info("\n行情统计,%s 热度%f%%:\n%s" % (g.target_market, g.emotion_rate, df_etf))
# -----------------------------------------------------------------
if df_etf['周期涨幅'].iloc[0] < 0.1 or df_etf['均线差值'].iloc[0] < 0:
log.info('交易信号:所有品种均不符合要求，空仓')
g.signal = 'CLEAR'
return
# 大盘符合要求， 有基金品种也符合要求，买
g.buy = [df_etf['基金代码'].iloc[0],]
log.info("交易信号:持有 %s" % g.buy)
g.signal = 'BUY'
return
# 大盘行情监控函数
def EmotionMonitor():
if g.MonitorSwitch == False:
return 1
_cnt = g.lag0 + max(g.lag, 3)
volume = attribute_history(security=g.target_market, count=_cnt, unit='1d', fields='volume')['volume']
v_ma_lag0 = talib.MA(volume, g.lag0) # 前 g.lag0 - 1 个会变成nan
#
g.emotion_rate = round((volume[-1] / v_ma_lag0[-1] - 1) * 100, 2) # 最后一天的成交量比成交量均值高%
#
vol_diff = (volume - v_ma_lag0)
if vol_diff[-1] >= 0:
ret_val = 1 if (vol_diff[-3:] >= 0).all() else 0 # 大盘成交量，最近3天都站在均线之上
else:
ret_val = -1 if (vol_diff[-g.lag:] < 0).all() else 0 # 大盘成交量，最近g.lag天都在均线之下
#
return ret_val
'''
=================================================
每日收盘后
=================================================
'''
def after_trading_end(context):
log.info('今日持仓：%s',context.portfolio.positions.keys())
print("总权益:{:.2f}万".format(context.portfolio.total_value/10000))
return
2025-02-20
