2423 MACD 低波价值涨停轮动策略大盘择时，逻辑简单

MACD 低波价值涨停轮动策略

核心逻辑：

本策略融合技术指标择时、价值选股与涨停板管理，构建多维度风控体系。策略运行分为三大模块：

MACD 大盘择时系统：

基于沪深 300 指数的 MACD (12,26,9) 指标
信号规则：
- DIF 上穿 DEA：允许股票持仓
- MACD 柱转负：强制清仓转货币 ETF
- 特殊机制：月末定期调仓触发超额收益再平衡

价值选股模型：

A[全市场股票] --> B[剔除次新/KCB/ST]
B --> C[PB<1+ROA>15%]
C --> D[市值>500亿]
D --> E[Beta<0.7]

组合构建：选取 5 只低估值 + 高盈利质量 + 低波动标的

涨停板管理系统：

持仓监控：自动识别昨日涨停标的
尾盘决策：14:00 检查未封涨停即止盈
特殊保护：涨停板打开即时触发市价单退出

核心指标：

MACD 动量：沪深 300 指数 500 分钟线 MACD
价值因子：ROA (资产回报率) 前 15% + PB<1
波动控制：年度 Beta 值 < 0.7（相比沪深 300）
流动性门槛：流通市值 > 500 亿

策略特点：

三维风控体系：
- 大盘择时（MACD）
- 个股质量（ROA+PB）
- 波动控制（Beta + 市值）
特殊机制设计：
- 月末定期再平衡（每月 1 日执行）
- 涨停板特殊处理规则
- 货币基金 (511880) 作为现金管理工具
交易细节优化：
- 组合分散：5 只等权配置
- 冲击成本控制：0.02 固定滑点
- 尾盘集中交易：14:00 关键时点决策
风险收益特征：
- Beta 暴露：主动控制组合系统风险
- 风格偏向：大市值价值型股票
- 极端行情保护：MACD 负值时全仓货币基金

（注：策略中 PB<1 的设定使其具有显著深度价值特征，配合 ROA 筛选避免落入价值陷阱）

策略代码

import pandas as pd
import talib as tb
import numpy as np
from jqdata import *
from jqfactor import get_factor_values


def initialize(context):
    set_benchmark('000300.XSHG') 
    log.set_level('order', 'error')
    set_option('use_real_price', True)
    set_option('avoid_future_data', True)# 设置是否开启避免未来数据模式
    set_slippage(FixedSlippage(0.02))# 设置滑点
    # 股票类交易手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, \
                             open_commission=0.00015, close_commission=0.00015,\
                             close_today_commission=0, min_commission=0), type='stock')
    # 持仓数量
    g.stock_num =5
    # 空仓选用
    g.etf_A = '511880.XSHG'
    # 轮动选用
    g.etf_B = '510300.XSHG'
    #相对300指数波动
    g.beta=0.7
    # MACD择时
    g.no_trading_today_signal =True
    # 设置交易时间，每天运行
    run_daily(prepare_stock_list, time='9:05', reference_security='000300.XSHG')
    run_daily(get_macd,time='9:30', reference_security='000300.XSHG')
    run_monthly(my_Trader,1, time='9:35', reference_security='000300.XSHG')
    run_daily(check_limit_up, time='14:00', reference_security='000300.XSHG')
    run_daily(my_trade_stocknum, '15:00')

def my_trade_stocknum(context):
    record(stocknum=len(context.portfolio.positions))

关键函数解锁后查看：

#显示筛查出股票的：名称，代码，市值
def slist(context,stock_list):    
    current_data = get_current_data()
    for stock in stock_list:
        df = get_fundamentals(query(valuation).filter(valuation.code == stock))
        print('股票代码：{0},  名称：{1},  总市值:{2:.2f},  流通市值:{3:.2f},  PE:{4:.2f},股价：{5:.2f}'.format(stock,get_security_info(stock).display_name,df['market_cap'][0],df['circulating_market_cap'][0],df['pb_ratio'][0],current_data[stock].last_price))

#1-1 准备股票池
# 如果持有股票昨天处于涨停的，则放入涨停列表，只要今天打开涨停就卖出，这个每天执行
def prepare_stock_list(context):
    #获取昨日涨停列
    g.high_limit_list=[]
    for stock in context.portfolio.positions.keys():
        df = get_price(stock, end_date=context.previous_date, frequency='daily', fields=['close','high_limit'], count=1)
        if df['close'][0] >= df['high_limit'][0]*0.98:#如果昨天有股票涨停，则放入列表
            g.high_limit_list.append(stock)
    
#1-5 调整昨日涨停股票
def check_limit_up(context):
    if g.high_limit_list != []:
        current_data = get_current_data()
        #对昨日涨停股票观察到尾盘如不涨停则提前卖出，如果涨停即使不在应买入列表仍暂时持有
        for stock in g.high_limit_list:
            if current_data[stock].last_price <   current_data[stock].high_limit:
                log.info("[%s]涨停打开，卖出" % (stock))
                order_target(stock, 0)
                order_value(g.etf_A, context.portfolio.cash)
            else:
                log.info("[%s]涨停，继续持有" % (stock))            
 
# 过滤科创北交股票
def filter_kcbj_stock(stock_list):
    for stock in stock_list[:]:
        if stock[0] == '4' or stock[0] == '8' or stock[:2] == '68':
            stock_list.remove(stock)
    return stock_list

# 过滤停牌股票
def filter_paused_stock(stock_list):
	current_data = get_current_data()
	return [stock for stock in stock_list if not current_data[stock].paused]


# 过滤ST及其他具有退市标签的股票
def filter_st_stock(stock_list):
	current_data = get_current_data()
	return [stock for stock in stock_list
			if not current_data[stock].is_st
			and 'ST' not in current_data[stock].name
			and '*' not in current_data[stock].name
			and '退' not in current_data[stock].name]


# 过滤涨停的股票
def filter_limit_stock(context, stock_list):
	last_prices = history(1, unit='1m', field='close', security_list=stock_list)
	current_data = get_current_data()
	# 已存在于持仓的股票即使涨停也不过滤，避免此股票再次可买，但因被过滤而导致选择别的股票
	return [stock for stock in stock_list if stock in context.portfolio.positions.keys()
			or current_data[stock].low_limit < last_prices[stock][-1] < current_data[stock].high_limit]
# 过滤次新股
def filter_new_stock(context, stock_list):
    return [stock for stock in stock_list if (context.previous_date - datetime.timedelta(days=300)) > get_security_info(stock).start_date]

  
#macd
def get_macd_M(stock_list,check_date):
    macd_list = {}
    if isinstance(check_date,str):
        check_date = datetime.datetime.strptime(check_date, "%Y-%m-%d %H:%M:%S")
    if isinstance(stock_list,str):
        stock_list = [stock_list]
    for stock in stock_list:
        array = get_bars(security=stock, 
                         count=500, 
                         unit='1M',
                         fields=['close'],
                         include_now=False,
                         end_dt=check_date, 
                         fq_ref_date=check_date)
        close_list = array['close']
        dif, dea, macd = tb.MACD(close_list, 
                                 fastperiod=12, 
                                 slowperiod=26, 
                                 signalperiod=9)
        last_dif = dif[-1]
        last_dea = dea[-1]
        last_macd = macd[-1]
        macd_dic = (last_dif, last_dea, last_macd*2)
        macd_list[stock] = macd_dic
    return macd_list
    

#历史BETA
def get_beta(today, stock_list):
    time0 = today  # 请根据实际情况调整开始日期
    if time0.day==29:
        time1day=28
    else:
        time1day=time0.day
    time1 = datetime.datetime((time0.year)-1,time0.month,time1day)
    print(time1)
    score_list =[]
    # 计算沪深300指数的方差
    index_data = get_price('000300.XSHG', start_date=time1.strftime('%Y-%m-%d') , end_date=time0.strftime('%Y-%m-%d'), frequency='daily', fields=['close'],panel=False)
    index_returns = index_data['close'].pct_change()
    index_var = index_returns.var()
    for stock in stock_list:   
        stock_data = get_price(stock, start_date=time1.strftime('%Y-%m-%d') , end_date=time0.strftime('%Y-%m-%d'), frequency='daily', fields=['close'],panel=False)
        stock_returns = stock_data['close'].pct_change()
        cov_matrix = index_returns.cov(stock_returns)
        cov = cov_matrix 
    # 计算比率
        ratio = cov / index_var
        score_list.append(ratio)
    df = pd.DataFrame(columns=['code','score'])
    df['code'] = stock_list
    df['score'] = score_list
    df = df.dropna()
    df = df.query(f'score<{g.beta}')


    filter_list = list(df.code)
    
    return filter_list

2025-02-25

{{userData.name}}已认证

MACD 低波价值涨停轮动策略

核心逻辑：

核心指标：

策略特点：

策略代码

2364.基于XGBoost6m滚动选股策略 42706

2419 五因子动态加权小市值周频策略机器学习线性回归小市值

2086 北向资金布林带动态调仓策略.py 策略源码分享

2051 北上资金持股比选股策略（北向港资外资）.py

2311 低价小盘简单有效.py 量化交易策略源码

2084 ETF网格交易策略.py

2304 首板低开策略.py

2092 敢于直接实盘的量化交易策略—八仙过海V2.py

2314 四年多翻9倍，年化73%，成长因子评分筛选.py

2302 7年40倍绩优低价超跌缩量小盘扩容到50只

2332 多因子选股策略，非线性关系市值（不是小市值）4只绩优股组合自动化量化交易策略代码

2350 多因子成长股月轮动策略一种新思路，超额376%！ 40560 量化策略自动化交易策略代码

优惠活动

配件商城

法律声明

在线工单

建议提交