构建交易安全防线:自动止损策略(附源码)

# 说明：策略代码使用ptrade系统api实现# 版本：自动止损策略# 功能：按持仓类型（ETF/主板/科创板）设置差异化止损率，支持固定止损与移动止损import numpy as npimport mathfrom datetime import datetime, date, timedeltaimport pandas as pd# ==========================# 持仓类型识别# ==========================def get_security_type(code):    """    识别持仓类型    返回: 'ETF', 'STAR'(科创板), 'CHINEXT'(创业板), 'MAIN'(主板)    """    if not isinstance(code, str):        return 'MAIN'    code_clean = code.split('.')[0]    # [Bug1修复] 加括号明确 and/or 优先级，两个条件均需校验长度==6    if (code_clean.startswith('51') and len(code_clean) == 6) or \       (code_clean.startswith('15') and len(code_clean) == 6) or \       (code_clean.startswith('16') and len(code_clean) == 6):        return 'ETF'    if code_clean.startswith('688'):        return 'STAR'    if code_clean.startswith('30'):        return 'CHINEXT'    return 'MAIN'# ==================== 初始化模块 ====================def initialize(context):    log.info("========== 策略初始化开始 ==========")    # ---------- 止损方式 ----------    # 0: 关闭, 1: 固定比例止损, 2: 移动止损, 3: 联合止损(固定+移动任一触发)    g.stop_loss_method_etf = 1    g.stop_loss_method_main = 3    g.stop_loss_method_star = 3    g.stop_loss_method_chinext = 3    # ---------- 固定止损比例（成本价基准）----------    g.fixed_stop_loss_etf = 0.05       # ETF: 5% 止损    g.fixed_stop_loss_main = 0.05     # 主板: 5% 止损    g.fixed_stop_loss_star = 0.7     # 科创板: 7% 止损    g.fixed_stop_loss_chinext = 0.7  # 创业板: 7% 止损    # ---------- 移动止损比例（历史最高价基准）----------    g.trailing_stop_loss_etf = 0.07       # ETF: 7% 移动止损    g.trailing_stop_loss_main = 0.8      # 主板: 8% 移动止损    g.trailing_stop_loss_star = 0.10     # 科创板: 10% 移动止损    g.trailing_stop_loss_chinext = 0.10   # 创业板: 10% 移动止损    # ---------- 运行时变量 ----------    g.sold_today = set()                  # 今日卖出的标的    g.position_high_price = {}            # 各持仓的最高价（移动止损用）    # ---------- 交易调度 ----------    run_daily(context, check_positions, time='09:10')    run_daily(context, check_before_trade, time='09:25')    # ---------- 设置持仓最高价 ----------    for security, position in context.portfolio.positions.items():        if position.amount <= 0:            continue        high_price = max(position.cost_basis, position.last_sale_price) if hasattr(position, 'last_sale_price') and position.last_sale_price > 0 else position.cost_basis        g.position_high_price[security] = high_price    if g.position_high_price:        log.info(f"📊 持仓最高价已设置: { {k: f'{v:.3f}'for k, v in g.position_high_price.items()} }")    else:        log.info("📊 无历史持仓，运行时实时更新最高价")    log.info("========== 策略初始化完成 ==========")def update_all_high_prices(context):    """每分钟批量更新所有持仓的最高价"""    securities = list(context.portfolio.positions.keys())    minute_data =  get_history(            count=1, frequency='1m',            field=['open', 'high', 'low', 'close'],            security_list=securities, fq='pre', include=True, fill='pre', is_dict=True        )    all_data = minute_data    log.info(f"📊 最新分钟数据: {all_data}")    for security in securities:        position = context.portfolio.positions[security]        if position.amount <= 0:            continue        stock_info = all_data[security][0]        if stock_info is None:            continue        current_price = stock_info['high']        if current_price <= 0:            continue        update_position_high_price(context, security, current_price)def handle_data(context, data):    update_all_high_prices(context)    minute_stop_loss(context)def check_before_trade(context):    """    每日 09:25 开盘前初始化    """    g.sold_today = set()    log.info("📊 今日止损标的列表已清空")    check_positions(context)def check_positions(context):    """    每日 09:10 开盘后检查    """    for security, position in context.portfolio.positions.items():        if security not in g.position_high_price:            if position.enable_amount > 0:                high_price = max(position.cost_basis, position.last_sale_price) if hasattr(position, 'last_sale_price') and position.last_sale_price > 0 else position.cost_basis                g.position_high_price[security] = high_price                log.info(f"📊 盘前 补录持仓最高价: {security} = {high_price:.3f}")# ==================== 止损参数获取 ====================def get_stop_loss_params(security):    """    根据持仓类型获取止损参数    返回: (method, threshold_fixed, threshold_trailing)        method: 0=关闭, 1=固定止损, 2=移动止损, 3=联合止损(任一触发)        threshold_fixed: 固定止损比例        threshold_trailing: 移动止损比例    """    sec_type = get_security_type(security)    if sec_type == 'ETF':        return g.stop_loss_method_etf, g.fixed_stop_loss_etf, g.trailing_stop_loss_etf    elif sec_type == 'STAR':        return g.stop_loss_method_star, g.fixed_stop_loss_star, g.trailing_stop_loss_star    elif sec_type == 'CHINEXT':        return g.stop_loss_method_chinext, g.fixed_stop_loss_chinext, g.trailing_stop_loss_chinext    else:        return g.stop_loss_method_main, g.fixed_stop_loss_main, g.trailing_stop_loss_main# ==================== 更新持仓最高价 ====================def update_position_high_price(context, security, current_price):    """    更新持仓历史最高价    移动止损基于持仓以来的最高点回撤    """    if security not in g.position_high_price:        g.position_high_price[security] = current_price    else:        if current_price > g.position_high_price[security]:            g.position_high_price[security] = current_price# ==================== 获取持仓最高价 ====================def get_position_high_price(context, security):    """    获取持仓以来的最高价    由 handle_data 每分钟统一更新，此处仅读取    """    return g.position_high_price.get(security)# ==================== 分钟级止损主函数 ====================def minute_stop_loss(context):    """    分钟级止损检查    - 支持固定比例止损（基于成本价）    - 支持移动止损（基于持仓以来最高价）    - 仅在交易时间段执行    - 触发后全仓卖出并记录sold_today    """    current_time = context.blotter.current_dt.strftime('%H:%M')    if not (('09:30' <= current_time <= '11:30') or ('13:00' <= current_time <= '14:57')):        return    for security in list(context.portfolio.positions.keys()):        position = context.portfolio.positions[security]        if position.enable_amount <= 0:            continue        current_price = get_current_data([security])[security].last_price        if current_price <= 0 or math.isnan(current_price) or math.isinf(current_price):            continue        cost_basis = position.cost_basis        if cost_basis <= 0:            continue        sec_type = get_security_type(security)        method, threshold_fixed, threshold_trailing = get_stop_loss_params(security)        if method == 0:            continue        stop_triggered = False        stop_type = ""        if method == 1:            stop_price = cost_basis * (1 - threshold_fixed)            if current_price <= stop_price:                stop_triggered = True                stop_type = "固定止损"        elif method == 2:            high_price = get_position_high_price(context, security)            if high_price is not None and high_price > 0:                trailing_stop_price = high_price * (1 - threshold_trailing)                if current_price <= trailing_stop_price:                    stop_triggered = True                    stop_type = "移动止损"        elif method == 3:            fixed_triggered = current_price <= cost_basis * (1 - threshold_fixed)            high_price = get_position_high_price(context, security)            trailing_triggered = False            if high_price is not None and high_price > 0:                trailing_stop_price = high_price * (1 - threshold_trailing)                if current_price <= trailing_stop_price:                    trailing_triggered = True            if fixed_triggered or trailing_triggered:                stop_triggered = True                stop_type = "联合止损"        if stop_triggered:            security_name = get_name(security)            loss_pct = (current_price / cost_basis - 1) * 100            log.info(f"🚨 【{stop_type}】{security}{security_name} 类型:{sec_type} "                     f"当前价:{current_price:.3f} 成本:{cost_basis:.3f} 亏损:{loss_pct:.2f}%")            if order(security, -position.enable_amount): #smart_order_target_value(security, 0, context):                log.info(f"   ✅ 止损卖出成功")                g.sold_today.add(security)                if security in g.position_high_price:                    del g.position_high_price[security]            else:                log.warning(f"   ❌ 止损卖出失败")def get_name(security):    """获取证券名称，带异常处理"""    try:        result = get_stock_name(security)        if isinstance(result, dict):            return result.get(security, security)        return result    except Exception:        return securitydef get_current_data(stock=None):    """    PTrade 兼容聚宽 API：get_current_data()    实盘使用 get_snapshot，回测使用 get_history    返回值：一个dict, 其中 key 是股票代码, value 是拥有如下属性的对象        last_price : 最新价,09:30之前获取返回昨日收盘价        high_limit: 涨停价        low_limit: 跌停价        paused: 是否停牌, 当停牌、未上市或者退市后返回 True        is_st: 是否是 ST(包括ST, *ST)，是则返回 True，否则返回 False        day_open: 当天开盘价        name: 股票现在的名称    if stock is None and current_data is None:        security_list = g.etf_pool + [g.defensive_etf]        if is_trade():            current_data = _get_current_data_realtime(security_list)        else:            current_data = _get_current_data_backtest(security_list)    """    if isinstance(stock, str):        security_list = [stock]    else:        security_list = list(stock)    if is_trade():        return _get_current_data_realtime(security_list)    else:        return _get_current_data_backtest(security_list)def _get_current_data_realtime(security_list):    """    实盘：通过 get_snapshot 获取实时行情    """    current_data = {}    try:        snapshot = get_snapshot(security_list)    except Exception as e:        log.warning("get_snapshot 获取失败: %s" % str(e))        return _get_current_data_backtest(security_list)    for code in security_list:        info = snapshot.get(code, {})        stock_info = {            'last_price': float(info.get('last_px', 0) or 0),            'high_limit': float(info.get('up_px', 0) or 0),            'low_limit': float(info.get('down_px', 0) or 0),            'day_open': float(info.get('open_px', 0) or 0),            'paused': info.get('trade_status', 'TRADE') in ('HALT', 'SUSP', 'STOPT', 'SUSPENDED'),            'is_st': False,            'name': info.get('name', ''),        }        if stock_info['high_limit'] == 0 and stock_info['last_price'] > 0:            stock_info['high_limit'] = stock_info['last_price'] * 1.1        if stock_info['low_limit'] == 0 and stock_info['last_price'] > 0:            stock_info['low_limit'] = stock_info['last_price'] * 0.9        if 'ST' in stock_info['name'] or '*ST' in stock_info['name'] or '退' in stock_info['name']:            stock_info['is_st'] = True        stock_obj = type('StockInfo', (), stock_info)()        current_data[code] = stock_obj    return current_datadef _get_current_data_backtest(security_list):    """    回测：通过 get_history 获取日线和分钟数据    """    current_data = {}    for code in security_list:        stock_info = {}        day_df = get_history(            count=1,            frequency='1d',            field=['close', 'open', 'high', 'low', 'high_limit', 'low_limit', 'is_open', 'preclose'],            security_list=code,            fq='pre',            include=True        )        minute_df = get_history(            count=1,            frequency='1m',            field=['price', 'close'],            security_list=code,            fq='pre',            include=False,            fill='pre'        )        if day_df is not None and not day_df.empty:            row = day_df.iloc[-1]            stock_info['high_limit'] = float(row.get('high_limit', 0))            stock_info['low_limit'] = float(row.get('low_limit', 0))            stock_info['day_open'] = float(row.get('open', 0))            stock_info['paused'] = (int(row.get('is_open', 1)) == 0)            if minute_df is not None and not minute_df.empty:                if 'price' in minute_df.columns:                    stock_info['last_price'] = float(minute_df['price'].iloc[-1])                elif 'close' in minute_df.columns:                    stock_info['last_price'] = float(minute_df['close'].iloc[-1])                else:                    stock_info['last_price'] = float(row.get('close', 0))            else:                stock_info['last_price'] = float(row.get('close', 0))        else:            stock_info['high_limit'] = 0            stock_info['low_limit'] = 0            stock_info['day_open'] = 0            stock_info['last_price'] = 0            stock_info['paused'] = True        if stock_info['high_limit'] == 0 and stock_info['last_price'] > 0:            stock_info['high_limit'] = stock_info['last_price'] * 1.1        if stock_info['low_limit'] == 0 and stock_info['last_price'] > 0:            stock_info['low_limit'] = stock_info['last_price'] * 0.9        stock_info['is_st'] = False        try:            info = get_stock_info(code)            stock_info['name'] = info.get('stock_name', '') if isinstance(info, dict) else ''        except Exception:            stock_info['name'] = ''        if 'ST' in stock_info['name'] or '*ST' in stock_info['name'] or '退' in stock_info['name']:            stock_info['is_st'] = True        stock_obj = type('StockInfo', (), stock_info)()        current_data[code] = stock_obj    return current_data