注册    登录
关注
Py学习  »  Python

年化81.8%，回撤13.29%，夏普2.45，三重指数平滑移动平均线的应用，附tqsdk案例python代码

七年实现财富自由 • 3 月前 • 183 次点击  
原创内容第971篇，专注AGI+，AI量化投资、个人成长与财富自由。
年化81.8%，回撤13.29%，夏普2.45。
指标的核心思想是通过对价格进行三次指数平滑来过滤掉短期市场波动（噪音），从而更清晰地识别中长期的趋势方向、动量的变化以及潜在的转折点（背离）。

from datetime import dateimport pandas as pdfrom tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinishedfrom tqsdk.ta import ATR
import osdef get_env_var(name, default=None):    value = os.environ.get(name)    if value is None and default is None:        raise EnvironmentError(f"环境变量 {name} 未设置")    return value or defaultusername = get_env_var('TQ_USERNAME')password = get_env_var('TQ_PASSWORD')
# ===== 全局参数设置 =====SYMBOL = "CFFEX.IC2306"  # 黄金期货合约POSITION_SIZE = 30  # 基础持仓手数START_DATE = date(2022, 11, 1)  # 回测开始日期END_DATE = date(2023, 4, 30)  # 回测结束日期
# TRIX指标参数TRIX_PERIOD = 12  # TRIX计算周期SIGNAL_PERIOD = 9  # 信号线计算周期MA_PERIOD = 60  # 长期移动平均线周期，用于趋势过滤
# 信号阈值参数SIGNAL_THRESHOLD = 0.05  # TRIX与信号线差值的阈值，避免微小交叉
# 风控参数ATR_PERIOD = 14  # ATR计算周期STOP_LOSS_MULTIPLIER = 2.0  # 止损ATR倍数TAKE_PROFIT_MULTIPLIER = 3.0  # 止盈ATR倍数MAX_HOLDING_DAYS = 15  # 最大持仓天数
# ===== 全局变量 =====current_direction = 0  # 当前持仓方向：1=多头，-1=空头，0=空仓entry_price = 0  # 开仓价格stop_loss_price = 0  # 止损价格entry_date = None  # 开仓日期trade_count = 0  # 交易次数win_count = 0  # 盈利次数

# ===== TRIX指标计算函数 =====def calculate_trix(close_prices, period):    """计算TRIX指标和信号线"""    # 第一重EMA    ema1 = close_prices.ewm(span=period, adjust=False).mean()    # 第二重EMA


    
    ema2 = ema1.ewm(span=period, adjust=False).mean()    # 第三重EMA    ema3 = ema2.ewm(span=period, adjust=False).mean()    # 计算TRIX    trix = 100 * (ema3 / ema3.shift(1) - 1)    # 计算信号线    signal = trix.rolling(SIGNAL_PERIOD).mean()
    return trix, signal

# ===== 策略开始 =====print("开始运行TRIX指标期货策略...")print(f"品种: {SYMBOL}, 回测周期: {START_DATE} - {END_DATE}")print(f"TRIX参数: 周期={TRIX_PERIOD}, 信号线周期={SIGNAL_PERIOD}")
# 创建API实例api = TqApi(web_gui=True,backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),            auth=TqAuth(username, password))
# 订阅合约的K线数据klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24)  # 日线数据
# 创建目标持仓任务target_pos = TargetPosTask(api, SYMBOL)
try:    while True:        # 等待更新        api.wait_update()
        # 如果K线有更新        if api.is_changing(klines.iloc[-1], "datetime"):            # 确保有足够的数据计算指标            if len(klines) max(TRIX_PERIOD, SIGNAL_PERIOD, MA_PERIOD, ATR_PERIOD) + 10:                continue
            # 计算TRIX指标和信号线            klines['trix'], klines['signal'] = calculate_trix(klines.close, TRIX_PERIOD)
            # 计算长期移动平均线，用于趋势过滤            klines['ma'] = klines.close.rolling(window=MA_PERIOD).mean()
            # 计算ATR用于设置止损            atr_data = ATR(klines, ATR_PERIOD)
            # 获取最新数据            current_price = float(klines.close.iloc[-1])            current_datetime = pd.to_datetime(klines.datetime.iloc[-1], unit='ns')            current_trix = float(klines.trix.iloc[-1])            previous_trix = float(klines.trix.iloc[-2])            current_signal = float(klines.signal.iloc[-1])            previous_signal = float(klines.signal.iloc[-2])            current_ma = float(klines.ma.iloc[-1])            current_atr = float(atr_data.atr.iloc[-1])
            # 计算TRIX与信号线的差值            trix_diff = current_trix - current_signal            previous_trix_diff = previous_trix - previous_signal
            # 输出调试信息            print(f"日期: {current_datetime.strftime('%Y-%m-%d')}, 价格: {current_price:.2f}")            print(f"TRIX: {current_trix:.4f}, 信号线: {current_signal:.4f}, 差值: {trix_diff:.4f}")
            # ===== 交易逻辑 =====
            # 空仓状态 - 寻找开仓机会            if current_direction == 0:                # 多头开仓条件：TRIX上穿信号线                if


    
 previous_trix < previous_signal and current_trix > current_signal:                    current_direction = 1                    target_pos.set_target_volume(POSITION_SIZE)                    entry_price = current_price                    stop_loss_price = entry_price - STOP_LOSS_MULTIPLIER * current_atr                    take_profit_price = entry_price + TAKE_PROFIT_MULTIPLIER * current_atr                    print(f"多头开仓: 价格={entry_price}, 止损={stop_loss_price:.2f}, 止盈={take_profit_price:.2f}")
                # 空头开仓条件：TRIX下穿信号线                elif previous_trix > previous_signal and current_trix < current_signal:                    current_direction = -1                    target_pos.set_target_volume(-POSITION_SIZE)                    entry_price = current_price                    stop_loss_price = entry_price + STOP_LOSS_MULTIPLIER * current_atr                    take_profit_price = entry_price - TAKE_PROFIT_MULTIPLIER * current_atr                    print(f"空头开仓: 价格={entry_price}, 止损={stop_loss_price:.2f}, 止盈={take_profit_price:.2f}")
            # 多头持仓 - 检查平仓条件            elif current_direction == 1:                # 止损条件                if current_price <= stop_loss_price:                    profit_pct = (current_price - entry_price) / entry_price * 100                    target_pos.set_target_volume(0)                    current_direction = 0                    print(f"多头止损平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
                # 止盈条件                elif current_price >= take_profit_price:                    profit_pct = (current_price - entry_price) / entry_price * 100                    target_pos.set_target_volume(0)                    current_direction = 0                    print(f"多头止盈平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
                # 信号平仓：TRIX下穿信号线                elif previous_trix > previous_signal and current_trix < current_signal:                    profit_pct = (current_price - entry_price) / entry_price * 100                    target_pos.set_target_volume(0)                    current_direction = 0                    print(f"多头信号平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
            # 空头持仓 - 检查平仓条件            elif current_direction == -1:                # 止损条件                if current_price >= stop_loss_price:                    profit_pct = (entry_price - current_price) / entry_price * 100                    target_pos.set_target_volume(0)                    current_direction = 0                    print(f"空头止损平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
                # 止盈条件                elif current_price <= take_profit_price:                    profit_pct = (entry_price - current_price) / entry_price * 100


    
                    target_pos.set_target_volume(0)                    current_direction = 0                    print(f"空头止盈平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
                # 信号平仓：TRIX上穿信号线                elif previous_trix < previous_signal and current_trix > current_signal:                    profit_pct = (entry_price - current_price) / entry_price * 100                    target_pos.set_target_volume(0)                    current_direction = 0                    print(f"空头信号平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
except BacktestFinished as e:    print("回测结束")    api.close()
代码所在位置：
代码已经提交，请前往星球下载：
AI量化实验室——量化投资的星辰大海
一些思考，关于规则量化和因子模型。
规则的逻辑是因子和阈值，比如咱们昨天策略里的aroon指标>70，来刻画一种走势状态。
规则当然是一目了然，缺点是比较机械，为何是70呢？
aroon指标本身也是一个因子，我们可以轻松计算它与未来N期收益率之间的相关性（IC值）。在单因子回测的时候，我们通常会设计百分位的阈值，而不是具体某个值，比如因子值>过去一段时间80%分位点时，做多等，然后通过止赢/止损位来平仓。
这样做比传统某一个阈值确实更科学，而且更普适。
另外一个就是N期收益率的问题，在多因子合成时建模时，需要一个标签。关于标签的设计：通过是未来N天的收益率。
这里的标签其实存在问题，比如，你的策略设定的是2%止损位，那在这个5天里，有没有可能，已经下行了2%？逻辑上当然可能，也可能突破了止盈位。
另外就是如果预测的是未来5天，那策略的持有期就应该设定为至少5天。
还有5天是一个离散值，有没有可能前面涨不少，然后第5天回落，或者前面跌不少，然后第5天回调。这都是有可能的。
问题就在于，使用一个第几天的标签，这个离散的状态刻画不了连续的价格。
代码和数据下载：AI量化实验室——2025量化投资的星辰大海

AI量化实验室 星球，已经运行三年多，1800+会员。

aitrader代码，因子表达式引擎、遗传算法（Deap）因子挖掘引擎等，支持vnpy，qlib，backtrader和bt引擎，内置多个年化30%+的策略，每周五迭代一次，代码和数据在星球全部开源。

点击 “查看原文”，直接访问策略集合。

扩展  •  历史文章   

EarnMore（赚得更多）基于RL的投资组合管理框架：一致的股票表示，可定制股票池管理。（附论文+代码）

机器学习驱动的策略开发通过流程 | 普通人阶层跃迁的可能路径？

年化30.24%，最大回撤19%，综合动量多因子评分策略再升级（python代码+数据）

三秒钟创建一个年化28%，夏普比1.25的策略（python系统已开放源代码下载）

6年年化收益46%，最大回撤率为16%的策略（附python代码）

Python社区是高质量的Python/Django开发社区
本文地址：http://www.python88.com/topic/185645
登录后回复
关于   移动版
Py学习 - 专注于Python技术发展的社区(原Django社区)
沪ICP备11025650号