以下是一个基于上升旗形买入、5日均线卖出的量化策略,适用于JoinQuant平台。策略每日开盘时判断信号,以开盘价交易,避免未来函数。

# 导入函数库
import numpy as np
import pandas as pd

def initialize(context):
    """
    初始化函数,设定要操作的股票、参数等
    """
    # 股票池:沪深300成分股(可自行修改)
    g.stocks = get_index_stocks('000300.XSHG')
    # 最大持仓数量
    g.max_hold = 5
    # 均线周期
    g.ma_days = 5
    # 运行每日策略(开盘后立即执行)
    run_daily(handle, time='open')

def handle(context):
    """
    每日开盘执行:先卖出,后买入
    """
    # 获取当前持仓
    current_positions = context.portfolio.positions
    # 获取所有可用股票(剔除ST、停牌等)
    stock_list = filter_valid_stocks(g.stocks, context)

    # ========== 卖出部分:跌破5日均线 ==========
    stocks_to_sell = []
    for stock in current_positions:
        # 判断是否跌破5日均线(基于昨日数据)
        if sell_signal(stock, context):
            stocks_to_sell.append(stock)

    for stock in stocks_to_sell:
        # 清空该股票持仓
        order_target_percent(stock, 0)
        log.info(f"卖出 {stock}:跌破5日均线")

    # ========== 买入部分:上升旗形突破 ==========
    # 获取现金占总资产比例
    cash_ratio = context.portfolio.available_cash / context.portfolio.total_value
    if cash_ratio <= 0.01:      # 现金不足1%则不买入
        return

    # 当前持仓股票数
    hold_count = len(current_positions)
    # 还可买入的股票数量
    can_buy_count = g.max_hold - hold_count
    if can_buy_count <= 0:
        return

    # 选出未持仓且出现买入信号的股票
    buy_candidates = []
    for stock in stock_list:
        if stock in current_positions:
            continue
        if buy_signal(stock, context):
            buy_candidates.append(stock)

    # 如果候选过多,按信号强度排序(此处简单取前N只)
    if len(buy_candidates) > can_buy_count:
        # 可自行添加排序逻辑,如成交量放大比例等
        buy_candidates = buy_candidates[:can_buy_count]

    # 买入:等比例分配剩余资金
    if buy_candidates:
        cash = context.portfolio.available_cash
        percent_per_stock = 0.95 / g.max_hold   # 留5%现金防止误差
        for stock in buy_candidates:
            order_target_percent(stock, percent_per_stock)
            log.info(f"买入 {stock}:上升旗形突破")

def filter_valid_stocks(stock_list, context):
    """
    过滤ST、停牌、上市不满60天的股票
    """
    current_date = context.current_dt.strftime('%Y-%m-%d')
    # 获取ST状态
    st_data = get_extras('is_st', stock_list, start_date=current_date, end_date=current_date, df=True)
    # 获取上市日期
    listed_date = get_fundamentals(query(valuation.code, valuation.market_cap, income.total_operation_income,
                                         indicator.adjust_factor).filter(valuation.code.in_(stock_list)))
    # 这里简化处理,实际可详细过滤
    valid = []
    for stock in stock_list:
        if st_data.loc[current_date, stock]:   # ST剔除
            continue
        # 简单用attribute_history检查是否有足够数据(例如60天)
        df = attribute_history(stock, 60, '1d', fields=['close'], skip_paused=True)
        if df is None or len(df) < 60:
            continue
        valid.append(stock)
    return valid

def sell_signal(stock, context):
    """
    卖出信号:昨日收盘价 < 昨日5日均线
    """
    df = attribute_history(stock, g.ma_days + 1, '1d', fields=['close'], skip_paused=True)
    if df is None or len(df) < g.ma_days + 1:
        return False
    # 昨日收盘价
    last_close = df['close'][-1]
    # 昨日5日均线(需要前5日收盘价,包括昨日)
    ma5 = df['close'][-g.ma_days:].mean()
    return last_close < ma5

def buy_signal(stock, context):
    """
    买入信号:上升旗形整理后突破
    简化条件:
    1. 最近20日涨幅 > 5%
    2. 最近10日最高价、最低价均呈下降趋势(线性回归斜率<0)
    3. 通道宽度(最高-最低)稳定,变异系数 < 0.1
    4. 最近5日均量 < 最近10日均量 * 0.9(缩量整理)
    5. 昨日收盘价突破过去10日(不含昨日)最高点
    6. 昨日成交量 > 最近5日均量 * 1.1(放量突破)
    """
    # 获取近30日数据(需包含足够天数)
    df = attribute_history(stock, 40, '1d', fields=['close','high','low','volume'], skip_paused=True)
    if df is None or len(df) < 25:
        return False

    # 1. 最近20日涨幅
    ret_20 = df['close'][-1] / df['close'][-20] - 1
    if ret_20 < 0.05:
        return False

    # 提取最近10日作为整理期
    df10 = df.iloc[-10:]        # 最近10个交易日
    if len(df10) < 10:
        return False

    highs = df10['high'].values
    lows = df10['low'].values
    vols = df10['volume'].values

    # 2. 线性回归判断趋势
    x = np.arange(len(highs))
    # 最高价斜率
    coeff_high = np.polyfit(x, highs, 1)[0]
    # 最低价斜率
    coeff_low = np.polyfit(x, lows, 1)[0]
    if coeff_high >= 0 or coeff_low >= 0:
        return False

    # 3. 通道宽度稳定性
    widths = highs - lows
    mean_w = np.mean(widths)
    std_w = np.std(widths)
    if mean_w == 0 or std_w / mean_w > 0.1:
        return False

    # 4. 成交量萎缩
    vol_ma5 = df['volume'].tail(5).mean()
    vol_ma10 = df['volume'].tail(10).mean()
    if vol_ma5 > 0.9 * vol_ma10:   # 缩量不足
        return False

    # 5. 突破过去10日(不含昨日)最高点
    # 过去9日最高价最大值(不含昨日)
    high_max_prev = df.iloc[-11:-1]['high'].max()   # 倒数第11到倒数第2
    last_close = df['close'][-1]
    if last_close <= high_max_prev:
        return False

    # 6. 放量突破
    last_vol = df['volume'][-1]
    if last_vol <= vol_ma5 * 1.1:
        return False

    # 附加:可增加平行性检验,此处简化
    return True

策略说明

· 买入逻辑:识别上升旗形整理后的放量突破。判断条件包括:前期上涨、整理期高点低点同步下移、通道宽度稳定、成交量先缩后放、价格突破整理区高点。
· 卖出逻辑:股价跌破5日均线时清仓。
· 交易时间:每日开盘时根据昨日数据产生信号,以开盘价执行。
· 股票池:沪深300成分股(可自行修改),并自动过滤ST、停牌、次新股。
· 资金管理:最多同时持有5只股票,每只分配等比例资金。

注意事项

· 策略中的形态参数(如涨幅阈值、变异系数、放量倍数)可根据历史回测优化调整。
· 由于简化处理,上升旗形的识别并非100%准确,实盘前建议充分测试。
· 可结合更多过滤器(如大盘环境、行业)提高胜率。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐