JoinQuant 平台如何编写 Supertrend 超级趋势交易策略？

这是一个基于 Supertrend（超级趋势） 指标的趋势跟踪策略。

策略原理

Supertrend 是一种结合了 ATR（平均真实波幅）和价格中枢（High+Low）/ 2 的趋势跟踪指标。

1. 计算 ATR：衡量市场波动率。
2. 计算上下轨：
   • 上轨（Basic Upper）= (最高价 + 最低价) / 2 + Multiplier * ATR
   • 下轨（Basic Lower）= (最高价 + 最低价) / 2 - Multiplier * ATR
3. 趋势判断：
   • 当收盘价突破上轨时，趋势转为看多（Supertrend 线变为下轨支撑线）。
   • 当收盘价跌破下轨时，趋势转为看空（Supertrend 线变为上轨压力线）。
4. 交易逻辑：
   • 买入信号：当前价格高于 Supertrend 线，且之前为空头趋势。
   • 卖出信号：当前价格低于 Supertrend 线，且之前为多头趋势。

策略代码

# -*- coding: utf-8 -*-
import numpy as np
import pandas as pd
from jqdata import *

def initialize(context):
    """
    初始化函数，设定基准、手续费、滑点及策略参数
    """
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', 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')
    
    # 设定滑点
    set_slippage(PriceRelatedSlippage(0.002), type='stock')
    
    # --- 策略参数设置 ---
    # 操作的标的：这里以平安银行为例，实际使用可替换为指数ETF或股票池
    g.security = '000001.XSHE' 
    
    # Supertrend 参数
    g.atr_period = 10   # ATR 周期
    g.multiplier = 3.0  # ATR 倍数
    
    # 每日开盘时运行
    run_daily(market_open, time='open')

def get_supertrend(df, period=10, multiplier=3.0):
    """
    计算 Supertrend 指标
    输入: 包含 high, low, close 的 DataFrame
    输出: 包含 supertrend 值和 trend 方向的 DataFrame
    """
    # 1. 计算 ATR (Average True Range)
    df['tr0'] = abs(df['high'] - df['low'])
    df['tr1'] = abs(df['high'] - df['close'].shift(1))
    df['tr2'] = abs(df['low'] - df['close'].shift(1))
    df['tr'] = df[['tr0', 'tr1', 'tr2']].max(axis=1)
    df['atr'] = df['tr'].rolling(window=period).mean()
    
    # 2. 计算基础上下轨
    hl2 = (df['high'] + df['low']) / 2
    df['basic_upper'] = hl2 + (multiplier * df['atr'])
    df['basic_lower'] = hl2 - (multiplier * df['atr'])
    
    # 3. 初始化最终上下轨和趋势
    df['final_upper'] = 0.0
    df['final_lower'] = 0.0
    df['supertrend'] = 0.0
    df['trend'] = 1 # 1 为多头, -1 为空头
    
    # 4. 迭代计算最终轨道 (需要用到前一天的值，所以使用循环)
    # 注意：为了性能，这里只遍历最后一部分数据，但为了演示逻辑完整性，这里遍历整个获取的数据段
    # 实际回测中，df 长度不宜过长，建议取 period * 5 左右的长度即可
    
    for i in range(period, len(df)):
        # --- 计算 Final Upper Band ---
        if (df['basic_upper'].iloc[i] < df['final_upper'].iloc[i-1]) or \
           (df['close'].iloc[i-1] > df['final_upper'].iloc[i-1]):
            df.at[df.index[i], 'final_upper'] = df['basic_upper'].iloc[i]
        else:
            df.at[df.index[i], 'final_upper'] = df['final_upper'].iloc[i-1]
            
        # --- 计算 Final Lower Band ---
        if (df['basic_lower'].iloc[i] > df['final_lower'].iloc[i-1]) or \
           (df['close'].iloc[i-1] < df['final_lower'].iloc[i-1]):
            df.at[df.index[i], 'final_lower'] = df['basic_lower'].iloc[i]
        else:
            df.at[df.index[i], 'final_lower'] = df['final_lower'].iloc[i-1]
            
        # --- 计算 Trend 和 Supertrend ---
        prev_trend = df['trend'].iloc[i-1]
        prev_final_upper = df['final_upper'].iloc[i-1]
        prev_final_lower = df['final_lower'].iloc[i-1]
        curr_close = df['close'].iloc[i]
        curr_final_upper = df['final_upper'].iloc[i]
        curr_final_lower = df['final_lower'].iloc[i]
        
        if prev_trend == 1: # 之前是多头
            if curr_close < curr_final_lower:
                df.at[df.index[i], 'trend'] = -1 # 转空
                df.at[df.index[i], 'supertrend'] = curr_final_upper
            else:
                df.at[df.index[i], 'trend'] = 1 # 保持多
                df.at[df.index[i], 'supertrend'] = curr_final_lower
        else: # 之前是空头
            if curr_close > curr_final_upper:
                df.at[df.index[i], 'trend'] = 1 # 转多
                df.at[df.index[i], 'supertrend'] = curr_final_lower
            else:
                df.at[df.index[i], 'trend'] = -1 # 保持空
                df.at[df.index[i], 'supertrend'] = curr_final_upper
                
    return df

def market_open(context):
    security = g.security
    
    # 获取历史数据
    # 需要足够的数据来预热 ATR 和 迭代计算，这里取 100 天
    hist_len = 100
    hdata = attribute_history(security, hist_len, '1d', ['high', 'low', 'close'])
    
    # 如果数据不足（例如新股），则跳过
    if len(hdata) < hist_len:
        return

    # 计算 Supertrend
    st_data = get_supertrend(hdata, period=g.atr_period, multiplier=g.multiplier)
    
    # 获取当前最新的信号
    current_trend = st_data['trend'].iloc[-1]
    current_supertrend = st_data['supertrend'].iloc[-1]
    current_close = st_data['close'].iloc[-1]
    
    # 获取上一期的信号（用于判断是否刚刚发生反转）
    prev_trend = st_data['trend'].iloc[-2]
    
    # 获取当前仓位
    position = context.portfolio.positions[security]
    
    # --- 交易逻辑 ---
    
    # 信号：从空头转为多头 (金叉)
    if prev_trend == -1 and current_trend == 1:
        log.info("Supertrend 翻红 (看多)，买入 %s，当前价: %.2f, 支撑线: %.2f" % (security, current_close, current_supertrend))
        # 全仓买入
        order_target_value(security, context.portfolio.available_cash)
        
    # 信号：从多头转为空头 (死叉)
    elif prev_trend == 1 and current_trend == -1:
        log.info("Supertrend 翻绿 (看空)，卖出 %s，当前价: %.2f, 压力线: %.2f" % (security, current_close, current_supertrend))
        # 清仓
        order_target(security, 0)
        
    # 绘制图表 (可选)
    record(Price=current_close, Supertrend=current_supertrend)

常见问题 Q&A

Q: 为什么 get_supertrend 函数里要用循环？
A: Supertrend 的计算具有递归性质。当前的“最终上轨/下轨”不仅取决于当天的价格，还取决于前一天的最终轨道值。例如，如果趋势是向上的，下轨只能持平或上升，不能下降，这需要对比前一天的值来确定。这种逻辑很难直接用 Pandas 的向量化操作实现，因此使用循环是最准确的方法。

Q: 如何修改策略操作多只股票？
A: 可以将 g.security 改为一个列表（例如 g.securities = get_index_stocks('000300.XSHG')），然后在 market_open 函数中使用 for security in g.securities: 循环，对每只股票分别计算指标和下单。

Q: 为什么获取 100 天的历史数据？
A: ATR 计算需要 g.atr_period（例如10天）的数据，而 Supertrend 的轨道计算需要一定的数据量来“稳定”下来的（类似于 EMA 的预热）。获取 100 天数据可以确保计算出的指标值是准确和稳定的。