如何在聚宽（JoinQuant）中实现均线放量新高策略的分层选股与多因子评分？

在量化交易中，均线放量新高策略是一种经典的顺势突破策略。为了提高策略的鲁棒性并降低交易噪声，我们通常会引入**分层选股（Standard, Relaxed, Momentum）与多因子综合评分（Multi-Factor Scoring）**机制。本文将详细介绍如何在聚宽（JoinQuant）平台上实现这一套完整的量化选股与评分框架。

一、 策略核心逻辑设计

1. 三层分层选股逻辑

• 第一层：标准选股（Standard）：要求价格站上所有短期和长期均线（如MA5 > MA20 > MA60），且成交量明显放大（如今日成交量大于5日均值的1.5倍），价格创近20日新高。
• 第二层：放宽选股（Relaxed）：若标准池股票不足，放宽均线多头排列要求，仅要求价格在MA20之上，且创近10日新高，成交量大于5日均值即可。
• 第三层：动量选股（Momentum）：不强制要求新高，但要求个股具有极强的短期动量（如近3日涨幅前10%），且MACD处于金叉状态。

2. 五维多因子评分系统

对筛选出的股票，从以下五个维度进行打分（每项0-20分，总分100分）：

1. 均线偏离度（MA Score）：价格距离MA20的距离，过远扣分（防超买），适度偏离加分。
2. 动量效应（Momentum Score）：计算过去20日的收益率偏度（Skewness20）或变化速率（ROC12）。
3. 成交量爆发（Volume Score）：今日成交量与5日均量、120日均量的比值（DAVOL5）。
4. 突破强度（Breakout Score）：当前价格处于过去1年股价的位置（fifty_two_week_close_rank）。
5. 趋势指标（MACD Score）：MACD柱状图（MACDC）的绝对值及斜率。

二、 聚宽代码实现

以下是在聚宽回测环境中运行的完整策略代码：

import numpy as np
import pandas as pd
from jqdata import *
from jqlib.technical_analysis import * 

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 设定交易税费
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 策略参数
    g.max_hold = 5 # 最大持仓数
    
    # 每日运行
    run_daily(market_open, time='09:30')

def market_open(context):
    # 1. 获取全市场股票池（剔除ST、停牌、退市）
    all_stocks = get_all_securities(['stock'], context.current_dt.date())
    current_data = get_current_data()
    stock_list = [code for code, info in all_stocks.iterrows() 
                  if not current_data[code].paused 
                  and not current_data[code].is_st
                  and 'PT' not in current_data[code].name
                  and '退' not in current_data[code].name]
    
    # 2. 分层选股初筛
    selected_stocks = filter_stocks_layered(stock_list, context)
    
    if not selected_stocks:
        return
        
    # 3. 多因子评分
    scored_df = score_stocks(selected_stocks, context)
    
    # 4. 排序并调仓
    target_list = list(scored_df.sort_values(by='total_score', ascending=False).head(g.max_hold).index)
    
    # 卖出不在目标池的股票
    for stock in list(context.portfolio.positions.keys()):
        if stock not in target_list:
            order_target(stock, 0)
            log.info("卖出: %s" % stock)
            
    # 买入目标股票
    hold_num = len(context.portfolio.positions)
    if hold_num < g.max_hold:
        buy_list = [s for s in target_list if s not in context.portfolio.positions]
        if buy_list:
            cash_per_stock = context.portfolio.available_cash / (g.max_hold - hold_num)
            for stock in buy_list[:g.max_hold - hold_num]:
                order_value(stock, cash_per_stock)
                log.info("买入: %s" % stock)

def filter_stocks_layered(stock_list, context):
    """三层分层选股初筛"""
    # 获取历史数据
    h = history(65, '1d', 'close', stock_list, df=False)
    v = history(10, '1d', 'volume', stock_list, df=False)
    
    standard_pool = []
    relaxed_pool = []
    momentum_pool = []
    
    for stock in stock_list:
        close_prices = h[stock]
        volumes = v[stock]
        
        if len(close_prices) < 65:
            continue
            
        ma5 = np.mean(close_prices[-5:])
        ma20 = np.mean(close_prices[-20:])
        ma60 = np.mean(close_prices[-60:])
        vol_ma5 = np.mean(volumes[-5:])
        
        # 1. 标准层：均线多头 + 放量 + 20日新高
        if (close_prices[-1] > ma5 > ma20 > ma60) and \
           (volumes[-1] > 1.5 * vol_ma5) and \
           (close_prices[-1] >= np.max(close_prices[-20:])):
            standard_pool.append(stock)
            continue
            
        # 2. 放宽层：站上20日线 + 10日新高 + 放量
        if (close_prices[-1] > ma20) and \
           (volumes[-1] > vol_ma5) and \
           (close_prices[-1] >= np.max(close_prices[-10:])):
            relaxed_pool.append(stock)
            continue
            
        # 3. 动量层：短期强动量 + MACD金叉
        # 计算MACD
        macd_dif, macd_dea, macd_hist = MACD(stock, check_date=context.previous_date)
        if macd_hist and macd_hist[stock] > 0: # MACD柱状图大于0（金叉或红柱放大）
            roc3 = (close_prices[-1] - close_prices[-4]) / close_prices[-4]
            if roc3 > 0.05: # 3日涨幅大于5%
                momentum_pool.append(stock)
                
    # 优先返回标准池，不足则用放宽池和动量池补充
    log.info("标准池数量: %d, 放宽池: %d, 动量池: %d" % (len(standard_pool), len(relaxed_pool), len(momentum_pool)))
    if len(standard_pool) >= g.max_hold:
        return standard_pool
    elif len(standard_pool) + len(relaxed_pool) >= g.max_hold:
        return standard_pool + relaxed_pool
    else:
        return list(set(standard_pool + relaxed_pool + momentum_pool))

def score_stocks(stock_list, context):
    """五维多因子评分系统"""
    scores = pd.DataFrame(index=stock_list, columns=['total_score'])
    
    # 批量获取因子计算所需的基础数据
    h = history(250, '1d', 'close', stock_list, df=False)
    v = history(120, '1d', 'volume', stock_list, df=False)
    
    for stock in stock_list:
        close = h[stock]
        vol = v[stock]
        
        # 1. 均线偏离度评分 (MA Score)
        ma20 = np.mean(close[-20:])
        bias20 = (close[-1] - ma20) / ma20
        # 偏离度在 2% - 8% 之间得分最高，过高（超买）或过低扣分
        ma_score = 20 - min(abs(bias20 - 0.05) * 100, 20)
        
        # 2. 动量效应评分 (Momentum Score)
        # 计算20日收益率偏度
        returns = np.diff(close[-21:]) / close[-21:-1]
        skew = pd.Series(returns).skew()
        mom_score = 10 + np.clip(skew * 5, -10, 10) # 偏度越大，动量爆发力越强
        
        # 3. 成交量爆发评分 (Volume Score)
        vol_ma5 = np.mean(vol[-5:])
        vol_ma120 = np.mean(vol[-120:])
        davol5 = vol_ma5 / vol_ma120 if vol_ma120 > 0 else 1
        vol_score = min(davol5 * 5, 20) # 5日均量相比120日均量放大倍数
        
        # 4. 突破强度评分 (Breakout Score)
        # 计算当前价格在过去1年（250天）的位置百分比
        rank_52w = (close[-1] - np.min(close)) / (np.max(close) - np.min(close)) if (np.max(close) - np.min(close)) > 0 else 0.5
        breakout_score = rank_52w * 20
        
        # 5. 趋势指标评分 (MACD Score)
        _, _, macd_hist = MACD(stock, check_date=context.previous_date)
        hist_val = macd_hist[stock] if macd_hist else 0
        macd_score = 10 + np.clip(hist_val * 100, -10, 10)
        
        # 综合评分
        scores.loc[stock, 'total_score'] = ma_score + mom_score + vol_score + breakout_score + macd_score
        
    return scores

三、 策略优势与SEO优化建议

1. 避免未来函数：策略中所有数据获取（如 history、MACD）均使用 context.previous_date 或不包含当前未收盘 Bar 的历史序列，严格杜绝了未来函数。
2. 动态复权（真实价格）：通过 set_option('use_real_price', True) 开启真实价格回测，确保分红送股时持仓数量和现金的自动调整，使回测结果更贴近实盘。
3. 分层容错机制：当市场行情不佳、标准突破个股较少时，放宽层和动量层能够提供备选标的，避免策略频繁空仓。