2317 阴线止损与涨停板策略 5年多因子超额收益1055%，K线风险过滤，无未来函数.py 量化交易策略源码

策略名称：阴线止损与涨停板策略

本策略结合了涨停板捕捉和大阴线止损的逻辑，通过筛选近期表现优异的股票，并在特定条件下进行买入和卖出操作。策略的核心思想是捕捉涨停板的短期机会，同时通过大阴线止损控制风险。

策略核心逻辑：

1. 股票池筛选：

   • 筛选上市375天以上的股票，排除科创板、ST股、停牌股、涨停或跌停的股票。

   • 通过营业收入增长率、盈利增长率、PEG等因子筛选出表现优异的股票。

2. 交易决策：

   • 每周第一个交易日开盘时，根据筛选出的股票调整持仓，买入新筛选的股票，卖出不再符合条件的股票。

   • 每日14:00检查昨日涨停的股票，若今日未涨停则卖出。

   • 每日开盘时检查昨日出现大阴线的股票，若开盘跌幅超过7%则卖出。

3. 风险控制：

   • 通过大阴线止损，避免在极端行情中遭受过大损失。

   • 限制持仓数量，避免过度集中风险。

策略特点：

• 涨停板捕捉：通过筛选近期涨停的股票，捕捉短期上涨机会。

• 大阴线止损：通过识别大阴线，及时止损，控制风险。

• 动态仓位管理：限制持仓数量，避免过度集中风险。

适用场景：

• 适用于A股市场，能够有效捕捉涨停板的短期机会。

• 适合短线交易者，通过快速交易获取收益。

风险提示：

• 涨停板策略在震荡市中可能产生较多的错误信号，导致频繁交易。

• 策略依赖于历史数据的有效性，需定期优化参数以适应市场变化。

---

策略代码结构：

1. 初始化函数（initialize）：

   • 设定基准指数（中证500）。

   • 配置交易手续费和滑点。

   • 定义每日运行的函数（开盘前、开盘时、收盘后）。

2. 开盘前运行函数（prepare_stock_list）：

   • 筛选符合条件的股票，更新持仓列表和历史持仓列表。

3. 开盘时运行函数（weekly_adjustment）：

   • 根据筛选出的股票调整持仓，买入新筛选的股票，卖出不再符合条件的股票。

4. 收盘后运行函数（check_limit_up）：

   • 检查昨日涨停的股票，若今日未涨停则卖出。

5. 开盘时运行函数（check_csy）：

   • 检查昨日出现大阴线的股票，若开盘跌幅超过7%则卖出。

6. 辅助函数：

   • get_single_factor_list：根据指定因子筛选股票。

   • sorted_by_circulating_market_cap：按流通市值排序股票。

   • get_stock_list：根据营业收入增长率、盈利增长率、PEG等因子筛选股票。

   • filter_paused_stock：过滤停牌股票。

   • filter_st_stock：过滤ST股。

   • get_recent_limit_up_stock：获取近期涨停的股票。

   • filter_limit_stock：过滤涨停股票。

   • filter_kcb_stock：过滤科创板股票。

   • order_target_value_：自定义下单函数。

   • open_position：买入指定价值的股票。

   • close_position：卖出指定持仓的股票。

---

策略优化建议：

1. 参数优化：

   • 可对涨停板筛选的天数（如20天）进行优化，以适应不同的市场周期。

   • 调整大阴线止损的条件（如7%），以更好地控制风险。

2. 风险控制：

   • 增加最大回撤限制，当回撤超过一定比例时暂停交易。

   • 设置单笔交易的最大亏损比例，避免单笔交易亏损过大。

3. 多因子结合：

   • 可结合其他技术指标（如成交量、MACD等）进行综合筛选，提高策略的稳定性。

---

总结：

本策略通过涨停板捕捉和大阴线止损的结合，适合短线交易者在A股市场中获取收益。策略逻辑清晰，适合有一定编程基础的投资者进行进一步优化和实盘测试。

策略回测

 

策略源码

# 风险及免责提示：该策略由聚宽用户在聚宽社区分享，仅供学习交流使用。
# 标题：5年超额收益1055%，K线处理大阴线，无未来


import statsmodels.api as sm  # 导入statsmodels库，用于统计分析
from jqdata import *  # 从jqdata模块导入所有内容，提供数据获取接口
from jqfactor import get_factor_values  # 导入获取因子值的函数
import datetime  # 导入datetime库，用于处理日期和时间

# 初始化函数
def initialize(context):
    # 设定基准
    set_benchmark('000905.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", 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.stock_num = 5      # 设置策略的目标持股数量为5
    g.limit_days = 20    # 设置检查涨停股票的天数为20天
    g.hold_list = []     # 初始化当前持仓列表
    g.history_hold_list = []    # 初始化历史持仓列表
    g.not_buy_again_list = []   # 初始化短期内不重复买入的股票列表
    g.switch=0                  # 初始化一个开关变量，用于控制某些逻辑
    # 设置交易时间，每天运行
    # 每天开盘前 9:05 运行prepare_stock_list函数，准备股票列表
    run_daily(prepare_stock_list, time='9:05', reference_security='000300.XSHG')
    # 每周的第一个交易日 9:30 运行weekly_adjustment函数，进行周调整
    run_weekly(weekly_adjustment, weekday=1, time='9:30', reference_security='000300.XSHG')
    # 每天14:00运行check_limit_up函数，检查昨日涨停今天未涨停的股票并卖出
    run_daily(check_limit_up, time='14:00', reference_security='000300.XSHG')
    # 每天开盘时9:30运行check_csy函数，检查昨天长上影的股票并卖出
    run_daily(check_csy, time='09:30', reference_security='000300.XSHG')

# 1-1 选股模块
def get_single_factor_list(context, stock_list, jqfactor, sort, p1, p2):
    # 这个函数用于根据指定的因子（jqfactor）对股票列表（stock_list）进行排序，并返回排序后位于p1到p2百分比范围内的股票列表。
    yesterday = context.previous_date    # 获取策略运行前一天的日期，用于获取因子数据
    # 使用get_factor_values函数获取指定股票列表和因子的数据，end_date指定获取数据的日期，count=1表示获取最新的数据
    # 从获取的因子数据中提取特定因子的值，并去除NaN值，然后根据sort参数指定的顺序进行排序
    s_score = get_factor_values(stock_list, jqfactor, end_date=yesterday, count=1
                                )[jqfactor].iloc[0].dropna().sort_values(ascending=sort)
    # 计算开始和结束的索引位置，然后根据这些索引从排序后的股票列表中选择股票，列表切片操作用于选择特定范围内的股票
    return s_score.index[int(p1 * len(stock_list)):int(p2 * len(stock_list))].tolist()

# 定义一个函数，用于根据流通市值对股票列表进行排序，并返回市值最大的前5只股票
def sorted_by_circulating_market_cap(stock_list, n_limit_top=5):
    # 创建一个查询对象，用于查询股票的代码和流通市值
    q = query(
        valuation.code,  # 指定查询字段为股票代码
    ).filter(
        valuation.code.in_(stock_list),      # 过滤条件，只选择列表中的股票
        indicator.eps > 0                    # 过滤条件，只选择每股收益大于0的股票
    ).order_by(
        valuation.circulating_market_cap.asc()  # 排序条件，按流通市值升序排序
    ).limit(
        n_limit_top    # 限制条件，限制返回的股票数量为前N名
    )
    # 获取流通市值最大的前N只股票，get_fundamentals(q) 执行查询，返回查询结果，.tolist() 将股票代码转换为列表形式
    return get_fundamentals(q)['code'].tolist() 

关键函数解锁后查看：

# 2-1 过滤停牌股票
def filter_paused_stock(stock_list):
    current_data = get_current_data()   # 获取当前所有股票的数据
    return [stock for stock in stock_list if not current_data[stock].paused]   # 使用列表推导式，返回不在停牌状态的股票列表

# 2-2 过滤ST及其他具有退市标签的股票
def filter_st_stock(stock_list):
    current_data = get_current_data()       # 获取当前所有股票的数据
    return [stock for stock in stock_list if not (      # 使用列表推导式，返回不是ST股且名称中不包含特定退市标签的股票列表
            current_data[stock].is_st or
            'ST' in current_data[stock].name or
            '*' in current_data[stock].name or
            '退' in current_data[stock].name)]

# 2-3 获取最近rencent_days个交易日内有涨停的股票列表
def get_recent_limit_up_stock(context, stock_list, recent_days):
    yesterday = context.previous_date    # 获取策略运行前一天的日期
    # 获取指定股票列表在最近recent_days天内的每日收盘价、涨停价和是否停牌的数据
    h = get_price(stock_list, end_date=yesterday, frequency='daily', fields=['close', 'high_limit', 'paused'],
                  count=recent_days, panel=False)
    # 查询涨停且未停牌的股票，然后按股票代码分组，计算每只股票涨停的次数
    s_limit = h.query('close==high_limit and paused==0').groupby('code')['high_limit'].count()
    return s_limit.index.tolist()        # 返回涨停次数大于0的股票列表

# 2-4 过滤涨停的股票
def filter_limit_stock(context, stock_list):
    current_data = get_current_data()        # 获取当前所有股票的数据，包括今日的开盘价、最高价、最低价和最新价
    holdings = list(context.portfolio.positions)      # 获取当前持仓的股票列表
    # 过滤掉不在持仓列表中，或者今日未涨停（即最新价不在涨停价和跌停价之间）的股票
    return [stock for stock in stock_list if (stock in holdings) or
            current_data[stock].low_limit < current_data[stock].last_price < current_data[stock].high_limit]

# 2-6 过滤科创板
def filter_kcb_stock(stock_list):
    return [stock for stock in stock_list if not stock.startswith('68')]   # 使用列表推导式，过滤68开头的科创版股票

# 3-1 交易模块-自定义下单
def order_target_value_(security, value):
    if value == 0:   # 检查目标价值是否为0，即是否需要清仓
        log.debug("Selling out %s" % security)   # 日志记录清仓信息
    else:
        log.debug("Order %s to value %f" % (security, value)) # 日志记录下单到特定价值的信息
    return order_target_value(security, value)   # 调用聚宽平台的order_target_value函数执行交易

# 3-2 交易模块-开仓
def open_position(security, value):
    _order = order_target_value_(security, value)  # 尝试对指定股票进行下单到特定价值
    if _order is not None and _order.filled > 0:        # 检查订单是否创建成功且有成交
        return True  # 开仓成功
    return False  # 开仓失败

# 3-3 交易模块-平仓
def close_position(position):
    security = position.security  # 获取持仓的股票代码
    _order = order_target_value_(security, 0)    # 下单清仓 ；可能会因停牌失败
    if _order is not None:  # 检查订单是否创建成功
        if _order.status == OrderStatus.held and _order.filled == _order.amount:  # 检查订单状态是否为全部成交且订单数量等于下单数量
            return True  # 平仓成功
    return False  # 平仓失败

2025-02-21