PTrade 策略风控：如何实现动态止盈止损？(附完整Python代码)

这是一个基于PTrade量化交易平台的策略示例。

策略逻辑说明

该策略包含以下核心部分：

1. 开仓逻辑（示例）：为了演示止盈止损功能，我设置了一个简单的双均线策略（5日均线上穿10日均线）作为买入信号。
2. 固定比例止损：当当前价格低于持仓成本价的一定比例（如5%）时，触发止损卖出。
3. 动态止盈（回撤止盈）：
   • 记录持仓以来的历史最高价。
   • 当当前价格从历史最高价回撤达到一定比例（如8%）时，触发止盈卖出。
   • 这种方式可以让利润在上涨趋势中奔跑，而在趋势反转时及时锁定利润。

PTrade 策略代码

def initialize(context):
    """
    初始化函数，设置参数和全局变量
    """
    # 设置回测资金和费率（可选，视券商环境而定）
    set_commission(commission_ratio=0.0003, min_commission=5.0)
    
    # 设定要操作的股票池基准，这里以沪深300为例
    g.index_code = '000300.SS'
    
    # --- 策略参数设置 ---
    # 固定止损比例 (例如 0.05 代表亏损 5% 止损)
    g.stop_loss_pct = 0.05
    
    # 动态回撤止盈比例 (例如 0.08 代表从最高点回撤 8% 止盈)
    g.trailing_stop_pct = 0.08
    
    # 用于存储持仓股票的最高价，格式：{ '股票代码': 最高价 }
    g.highest_prices = {}

def before_trading_start(context, data):
    """
    盘前处理，每天开盘前运行一次
    """
    # 获取沪深300成分股作为股票池
    # 注意：get_index_stocks 建议在盘前调用
    stocks = get_index_stocks(g.index_code)
    set_universe(stocks)

def handle_data(context, data):
    """
    盘中处理，每个单位时间（如每天或每分钟）运行一次
    """
    # 1. 止盈止损检查（优先处理持仓）
    check_stop_loss_and_take_profit(context, data)
    
    # 2. 开仓逻辑（示例：双均线金叉）
    buy_logic(context, data)

def check_stop_loss_and_take_profit(context, data):
    """
    检查持仓，执行固定止损和动态止盈
    """
    # 获取当前所有持仓
    positions = context.portfolio.positions
    # 获取持仓代码列表
    holding_list = list(positions.keys())
    
    for stock in holding_list:
        # 如果数据中没有该股票行情（如停牌），跳过
        if stock not in data:
            continue
            
        position = positions[stock]
        
        # 如果可用仓位为0，跳过
        if position.enable_amount == 0:
            continue
            
        current_price = data[stock]['close']
        cost_price = position.cost_basis
        
        # --- 更新最高价逻辑 ---
        # 如果该股票不在记录中，或者当前价格创出新高，则更新最高价
        if stock not in g.highest_prices:
            g.highest_prices[stock] = max(cost_price, current_price)
        else:
            if current_price > g.highest_prices[stock]:
                g.highest_prices[stock] = current_price
        
        highest_price = g.highest_prices[stock]
        
        # --- 1. 固定止损逻辑 ---
        # 当前价格 < 成本价 * (1 - 止损比例)
        if current_price < cost_price * (1 - g.stop_loss_pct):
            log.info("触发固定止损: %s, 成本价: %.2f, 当前价: %.2f" % (stock, cost_price, current_price))
            order_target(stock, 0)
            # 卖出后清除最高价记录
            if stock in g.highest_prices:
                del g.highest_prices[stock]
            continue # 已卖出，进入下一只股票循环

        # --- 2. 动态回撤止盈逻辑 ---
        # 当前价格 < 历史最高价 * (1 - 回撤比例)
        # 且为了保证是盈利状态，通常要求当前价格高于成本价（可选）
        if current_price < highest_price * (1 - g.trailing_stop_pct):
            log.info("触发动态止盈: %s, 历史最高: %.2f, 当前价: %.2f" % (stock, highest_price, current_price))
            order_target(stock, 0)
            # 卖出后清除最高价记录
            if stock in g.highest_prices:
                del g.highest_prices[stock]

def buy_logic(context, data):
    """
    简单的双均线买入逻辑
    """
    # 获取股票池
    security_list = list(context.universe)
    
    # 获取历史数据，取11天是为了计算10日均线和前一日的均线
    # count=11, 频率1d
    history = get_history(11, '1d', 'close', security_list)
    
    # 遍历股票池寻找买入机会
    # 注意：实际交易中建议限制每日买入数量，避免资金过于分散
    buy_count = 0 
    
    for stock in security_list:
        # 限制每次handle_data最多买入几只，防止循环过慢
        if buy_count >= 5:
            break
            
        # 如果已有持仓，暂不加仓
        if context.portfolio.positions[stock].amount > 0:
            continue
            
        # 确保数据存在且未停牌
        if stock in data and not math.isnan(data[stock]['close']):
            # 提取该股票的收盘价序列
            # 注意：get_history返回的数据结构处理
            try:
                # 兼容不同版本的pandas返回
                if 'code' in history.columns:
                    closes = history[history['code'] == stock]['close'].values
                else:
                    closes = history[stock].values
            except:
                continue

            if len(closes) < 11:
                continue
                
            # 计算均线
            ma5_curr = closes[-5:].mean()
            ma10_curr = closes[-10:].mean()
            
            ma5_prev = closes[-6:-1].mean()
            ma10_prev = closes[-11:-1].mean()
            
            # 金叉判断：昨天MA5 < MA10，今天MA5 > MA10
            if ma5_prev < ma10_prev and ma5_curr > ma10_curr:
                # 获取可用资金
                cash = context.portfolio.cash
                # 简单的资金管理：每次使用剩余资金的20%买入
                if cash > 10000: 
                    order_value(stock, cash * 0.2)
                    log.info("均线金叉买入: %s" % stock)
                    
                    # 买入时初始化该股票的最高价为当前价
                    g.highest_prices[stock] = data[stock]['close']
                    buy_count += 1

# 引入math库用于判断nan
import math

代码关键点解析

1. g.highest_prices 字典：

   • 这是实现动态止盈的核心。我们使用一个全局字典来记录每只持仓股票自买入以来的最高价格。
   • 在 check_stop_loss_and_take_profit 函数中，每次行情更新时，都会比较 current_price 和字典中记录的价格，取较大值更新进去。
   • 重要：当股票卖出（order_target(stock, 0)）后，必须使用 del 从字典中删除该记录，否则下次买入同一只股票时会使用旧的最高价导致逻辑错误。

2. 固定止损逻辑：

   • 公式：current_price < cost_price * (1 - g.stop_loss_pct)
   • cost_basis 是 PTrade 自动维护的持仓成本价。如果价格跌破成本线的 5%，直接清仓。

3. 动态止盈（移动止损）逻辑：

   • 公式：current_price < highest_price * (1 - g.trailing_stop_pct)
   • 假设你 10 元买入，价格涨到 12 元，highest_price 更新为 12。如果设置回撤 8%，则止盈触发价为 12 * (1 - 0.08) = 11.04 元。如果价格继续涨到 15 元，触发价自动提高到 15 * 0.92 = 13.8 元。

4. Python 版本兼容性：

   • 代码中使用了 %s 和 %f 进行字符串格式化，而不是 f-string，这是为了兼容 PTrade 可能使用的 Python 3.5 或更低版本环境。
   • 避免了使用 import os 等禁止库。

Q&A

Q: 如何调整止损和止盈的幅度？
A: 修改 initialize 函数中的 g.stop_loss_pct（固定止损）和 g.trailing_stop_pct（动态回撤）的值即可。

Q: 为什么买入时要初始化 g.highest_prices？
A: 为了防止买入后价格立即下跌，导致字典中没有数据而报错或逻辑失效。买入时将最高价设为当前价是初始状态。

Q: 这个策略适合回测还是实盘？
A: 代码结构同时兼容回测和实盘。但在实盘中，建议将 get_history 的 count 设置得更大一些以确保指标计算稳定，并注意实盘的成交量限制。