一、交易策略解释
核心思想
基于分形的趋势突破策略是一种技术分析交易方法，它基于金融市场中经常出现的分形模式来识别潜在的市场转折点和趋势延续信号。策略核心是捕捉价格在突破关键分形水平后的动量变化，结合趋势确认进行交易。它的基本原理包括：
分形识别：通过分析连续K线的高低点关系，识别出牛市分形和熊市分形。牛市分形出现在某根K线的低点低于两侧K线的低点时，熊市分形出现在某根K线的高点高于两侧K线的高点时。
趋势确认：使用移动平均线（MA）交叉判断当前整体趋势方向，确保交易顺应主要趋势。短期MA位于长期MA上方表示上升趋势，反之则为下降趋势。
突破入场：当价格突破牛市分形的高点且处于上升趋势时产生多头信号；当价格跌破熊市分形的低点且处于下降趋势时产生空头信号。
风险管理：使用ATR指标动态设置止损位置，并采用固定盈亏比设置止盈目标，同时结合均线和对立分形反转提供额外的出场信号。
理论基础
分形市场假说(FMH)：由埃德加·彼得斯于1991年提出，该理论认为市场价格表现出分形特性，在不同时间尺度上显示出相似的模式。与传统的有效市场假说不同，FMH考虑了投资者持有不同的投资时间周期和对信息的不同解读方式，这使得市场呈现分形特征。
比尔·威廉姆斯的交易理论：威廉姆斯将分形概念应用于交易，提出通过识别价格高低点的特定排列来寻找市场转折点。他的理论强调，这些分形点往往代表市场中的重要支撑和阻力位，可以作为交易决策的参考点。
策略适用场景
具有明显趋势的市场：该策略最适合应用于有清晰趋势的市场，因为它主要捕捉趋势中的突破机会。在横盘整理市场中可能会产生较多假信号。
流动性较好的期货合约：策略适用于股指期货、大宗商品期货等流动性较好的合约，这些市场价格变动更加连续，减少了分形突破后的跳空风险。
中等波动性环境：在极低波动性环境下，分形识别效果不佳；而在极高波动性环境下，止损容易被触及。因此策略适合于波动性适中的市场条件。
短中期交易：该策略更适合1小时至日线级别的时间周期，适合短中期交易风格，而非高频交易或长期投资。
策略原理
分形识别算法是策略的核心部分，我们使用以下算法识别牛市和熊市分形：
def identify_fractals(klines): """识别牛市和熊市分形""" bull_fractals = [] # 牛市分形列表 bear_fractals = [] # 熊市分形列表 # 至少需要2*FRACTAL_WINDOW+1个K线才能形成分形 if len(klines) < 2*FRACTAL_WINDOW+1: return bull_fractals, bear_fractals # 遍历K线，寻找分形 for i in range(FRACTAL_WINDOW, len(klines)-FRACTAL_WINDOW): # 牛市分形：中间K线的低点低于两侧K线的低点 is_bull_fractal = True for j in range(1, FRACTAL_WINDOW+1): if klines.low.iloc[i] >= klines.low.iloc[i-j] or klines.low.iloc[i] >= klines.low.iloc[i+j]: is_bull_fractal = False break if is_bull_fractal: bull_fractals.append((i, klines.low.iloc[i], klines.high.iloc[i])) # 熊市分形：中间K线的高点高于两侧K线的高点 is_bear_fractal = True for j in range(1, FRACTAL_WINDOW+1): if klines.high.iloc[i] <= klines.high.iloc[i-j] or klines.high.iloc[i] <= klines.high.iloc[i+j]: is_bear_fractal = False break if is_bear_fractal: bear_fractals.append((i, klines.low.iloc[i], klines.high.iloc[i])) return bull_fractals, bear_fractals这个函数实现了基本的分形识别逻辑：
牛市分形：中间K线的低点低于前后各FRACTAL_WINDOW根K线的低点
熊市分形：中间K线的高点高于前后各FRACTAL_WINDOW根K线的高点
返回的分形信息包含索引位置、低点和高点价格，便于后续的交易信号判断。
趋势确认指标
趋势确认使用移动平均线交叉系统：
短期MA(6)：捕捉较为敏感的价格变化
长期MA(12)：提供更稳定的趋势参考
当短期MA位于长期MA上方时，认为市场处于上升趋势；反之则认为是下降趋势
波动性衡量和止损设置
使用ATR(10)指标测量市场波动性，并将其作为止损距离的动态调整依据：
多头止损 = 分形低点 - ATR * ATR_MULTIPLIER
空头止损 = 分形高点 + ATR * ATR_MULTIPLIER
ATR_MULTIPLIER参数(1.2)用于调整止损的宽松程度，较大的乘数会给价格更多的波动空间，但也会增加每笔交易的风险。
交易逻辑
多头入场条件
多头入场需同时满足以下条件：
- 已识别出牛市分形
- 当前价格突破该分形的高点
- 短期MA位于长期MA上方(上升趋势确认)
- 分形不是最新形成的K线(需要有确认)
空头入场条件
空头入场需同时满足以下条件：
- 已识别出熊市分形
- 当前价格跌破该分形的低点
- 短期MA位于长期MA下方(下降趋势确认)
- 分形不是最新形成的K线(需要有确认)
多头平仓条件
多头持仓在以下任一情况下平仓：
- 价格达到止损位(分形低点 - ATR*1.2)
- 价格达到止盈位(入场价 + (入场价-止损价)*2.0)
- 价格跌破短期MA(趋势反转信号)
- 出现新的熊市分形且价格跌破该分形的低点(分形反转信号)
空头平仓条件
空头持仓在以下任一情况下平仓：
- 价格达到止损位(分形高点 + ATR*1.2)
- 价格达到止盈位(入场价 - (止损价-入场价)*2.0)
- 价格突破短期MA(趋势反转信号)
- 出现新的牛市分形且价格突破该分形的高点(分形反转信号)
回测
回测初始设置
- 测试周期： 2023 年 4 月 10 日 - 2023 年 4 月 26 日
- 交易品种： SHFE.au2306
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
代码
#!/usr/bin/env python # -*- coding: utf-8 -*- __author__ = "Chaos" from datetime import date import pandas as pd from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished from tqsdk.ta import ATR, MA # ===== 全局参数设置 ===== SYMBOL = "SHFE.au2306" POSITION_SIZE = 30 # 每次交易手数 START_DATE = date(2023, 4, 10) # 回测开始日期 END_DATE = date(2023, 4, 26) # 回测结束日期 # 1小时K线 KLINE_PERIOD = 60 * 60 # 分形识别参数 FRACTAL_WINDOW = 2 # 分形窗口大小 # 确认指标参数 SHORT_MA_PERIOD = 6 # 短期MA周期 LONG_MA_PERIOD = 12 # 长期MA周期 VOLUME_PERIOD = 5 # 成交量平均周期 ATR_PERIOD = 10 # ATR计算周期 # 风控参数 ATR_MULTIPLIER = 1.2 # 止损ATR乘数 TP_RATIO = 1.8 # 止盈比例 # ===== 全局变量 ===== current_direction = 0 # 当前持仓方向：1=多头，-1=空头，0=空仓 entry_price = 0 # 开仓价格 stop_loss_price = 0 # 止损价格 take_profit_price = 0 # 止盈价格 last_bull_fractal = None # 最近的牛市分形 last_bear_fractal = None # 最近的熊市分形 # ===== 分形识别函数 ===== def identify_fractals(klines): """识别牛市和熊市分形""" bull_fractals = [] # 牛市分形列表 [(index, low_price, high_price), ...] bear_fractals = [] # 熊市分形列表 [(index, low_price, high_price), ...] # 至少需要2*FRACTAL_WINDOW+1个K线才能形成分形 if len(klines) < 2 * FRACTAL_WINDOW + 1: return bull_fractals, bear_fractals # 遍历K线，寻找分形 start_idx = max(FRACTAL_WINDOW, len(klines) - 20) end_idx = len(klines) - FRACTAL_WINDOW for i in range(start_idx, end_idx): # 牛市分形：中间K线的低点低于两侧K线的低点 is_bull_fractal = True for j in range(1, FRACTAL_WINDOW + 1): if klines.low.iloc[i] >= klines.low.iloc[i - j] or klines.low.iloc[i] >= klines.low.iloc[i + j]: is_bull_fractal = False break if is_bull_fractal: bull_fractals.append((i, klines.low.iloc[i], klines.high.iloc[i])) # 熊市分形：中间K线的高点高于两侧K线的高点 is_bear_fractal = True for j in range(1, FRACTAL_WINDOW + 1): if klines.high.iloc[i] <= klines.high.iloc[i - j] or klines.high.iloc[i] <= klines.high.iloc[i + j]: is_bear_fractal = False break if is_bear_fractal: bear_fractals.append((i, klines.low.iloc[i], klines.high.iloc[i])) return bull_fractals, bear_fractals # ===== 策略开始 ===== print("开始运行基于分形的趋势突破期货策略...") print(f"品种: {SYMBOL}, 回测周期: {START_DATE} 至 {END_DATE}") # 创建API实例 api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE), auth=TqAuth("快期账号", "快期密码")) # 订阅合约的K线数据 klines = api.get_kline_serial(SYMBOL, KLINE_PERIOD) # 1小时K线 # 创建目标持仓任务 target_pos = TargetPosTask(api, SYMBOL) try: while True: # 等待更新 api.wait_update() # 如果K线更新 if api.is_changing(klines.iloc[-1], "datetime"): # 确保有足够的数据计算指标 if len(klines) < max(SHORT_MA_PERIOD, LONG_MA_PERIOD, ATR_PERIOD) + 5: print(f"数据不足，当前K线数量: {len(klines)}") continue # 计算确认指标 klines["short_ma"] = MA(klines, SHORT_MA_PERIOD).ma klines["long_ma"] = MA(klines, LONG_MA_PERIOD).ma klines["volume_avg"] = klines.volume.rolling(VOLUME_PERIOD).mean().fillna(0) atr = ATR(klines, ATR_PERIOD).atr # 识别分形 bull_fractals, bear_fractals = identify_fractals(klines) # 更新最近的分形 if bull_fractals: last_bull_fractal = bull_fractals[-1] print( f"发现新牛市分形: 索引={last_bull_fractal[0]}, 低点={last_bull_fractal[1]}, 高点={last_bull_fractal[2]}") if bear_fractals: last_bear_fractal = bear_fractals[-1] print( f"发现新熊市分形: 索引={last_bear_fractal[0]}, 低点={last_bear_fractal[1]}, 高点={last_bear_fractal[2]}") # 获取当前价格和指标数据 current_price = float(klines.close.iloc[-1]) current_volume = float(klines.volume.iloc[-1]) current_short_ma = float(klines.short_ma.iloc[-1]) current_long_ma = float(klines.long_ma.iloc[-1]) current_volume_avg = float(klines.volume_avg.iloc[-1] if not pd.isna(klines.volume_avg.iloc[-1]) else 0) current_atr = float(atr.iloc[-1] if not pd.isna(atr.iloc[-1]) else 0) # 确定当前趋势 uptrend = current_short_ma > current_long_ma downtrend = current_short_ma < current_long_ma # 输出当前状态 current_time = pd.to_datetime(klines.datetime.iloc[-1], unit='ns') print(f"时间: {current_time.strftime('%Y-%m-%d %H:%M')}") print(f"价格: {current_price}, ATR: {current_atr:.2f}") print(f"短期MA: {current_short_ma:.2f}, 长期MA: {current_long_ma:.2f}") print(f"趋势方向: {'上升' if uptrend else '下降' if downtrend else '盘整'}") print(f"当前成交量: {current_volume}, 平均成交量: {current_volume_avg:.2f}") # ===== 交易逻辑 ===== # 空仓状态 - 寻找开仓机会 if current_direction == 0: # 多头入场信号 if (last_bull_fractal and last_bull_fractal[0] < len(klines) - 1 and # 确认分形不是最新K线 current_price > last_bull_fractal[2] and # 价格突破牛市分形高点 uptrend): # 上升趋势确认 current_direction = 1 entry_price = current_price target_pos.set_target_volume(POSITION_SIZE) # 设置止损和止盈 stop_loss_price = last_bull_fractal[1] - current_atr * ATR_MULTIPLIER take_profit_price = entry_price + (entry_price - stop_loss_price) * TP_RATIO print(f"多头开仓: 价格={entry_price}, 止损={stop_loss_price:.2f}, 止盈={take_profit_price:.2f}") # 空头入场信号 elif (last_bear_fractal and last_bear_fractal[0] < len(klines) - 1 and # 确认分形不是最新K线 current_price < last_bear_fractal[1] and # 价格跌破熊市分形低点 downtrend): # 下降趋势确认 current_direction = -1 entry_price = current_price target_pos.set_target_volume(-POSITION_SIZE) # 设置止损和止盈 stop_loss_price = last_bear_fractal[2] + current_atr * ATR_MULTIPLIER take_profit_price = entry_price - (stop_loss_price - entry_price) * TP_RATIO print(f"空头开仓: 价格={entry_price}, 止损={stop_loss_price:.2f}, 止盈={take_profit_price:.2f}") # 多头持仓 - 检查平仓条件 elif current_direction == 1: # 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}%") # 2. 止盈条件 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}%") # 3. 价格跌破短期MA elif current_price < current_short_ma: profit_pct = (current_price - entry_price) / entry_price * 100 target_pos.set_target_volume(0) current_direction = 0 print(f"多头MA平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%") # 4. 出现新的熊市分形且价格跌破该分形的低点 elif (last_bear_fractal and last_bear_fractal[0] > klines.index[-10] and # 确保是较新的分形 current_price < last_bear_fractal[1]): 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: # 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}%") # 2. 止盈条件 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}%") # 3. 价格突破短期MA elif current_price > current_short_ma: profit_pct = (entry_price - current_price) / entry_price * 100 target_pos.set_target_volume(0) current_direction = 0 print(f"空头MA平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%") # 4. 出现新的牛市分形且价格突破该分形的高点 elif (last_bull_fractal and last_bull_fractal[0] > klines.index[-10] and # 确保是较新的分形 current_price > last_bull_fractal[2]): 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()