445 lines
18 KiB
Python
445 lines
18 KiB
Python
"""
|
|
盘整背驰策略 (PanZhengBeiChi Strategy)
|
|
|
|
基于缠论的盘整背驰进行交易：
|
|
- 盘整背驰：同级别走势中，Ai与Ai+2比较力度减弱
|
|
- 顶背驰（卖点）：价格创新高或接近，但MACD力度明显减弱
|
|
- 底背驰（买点）：价格创新低或接近，但MACD力度明显减弱
|
|
|
|
使用命令：
|
|
freqtrade backtesting -c ./user_data/Chan/config/Local_Test.json \
|
|
--strategy PanZhengBeiChiStrategy --strategy-path ./user_data/Chan/strategies \
|
|
--timerange=20250301-
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
import numpy as np
|
|
import pandas as pd
|
|
import talib.abstract as ta
|
|
from pandas import DataFrame
|
|
from technical.util import resample_to_interval, resampled_merge
|
|
from freqtrade.strategy import IStrategy
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PanZhengBeiChiStrategy(IStrategy):
|
|
"""
|
|
盘整背驰策略
|
|
|
|
核心逻辑：
|
|
1. 在5分钟级别识别同级别走势段（Ai）
|
|
2. 比较Ai与Ai+2的MACD力度，判断盘整背驰
|
|
3. 盘整顶背驰（i+2为偶数）-> 卖出
|
|
4. 盘整底背驰（i+2为奇数）-> 买入
|
|
"""
|
|
INTERFACE_VERSION: int = 3
|
|
|
|
# === 基础配置 ===
|
|
timeframe = '1m'
|
|
informative_timeframe = '5m'
|
|
can_short = True
|
|
can_long = True
|
|
|
|
startup_candle_count: int = 2000 # 需要足够的数据来识别走势段
|
|
|
|
# === 止损止盈配置 ===
|
|
stoploss = -0.02 # 2% 硬止损
|
|
use_custom_stoploss = False
|
|
|
|
# Trailing stop
|
|
trailing_stop = True
|
|
trailing_stop_positive = 0.008 # 回撤 0.8% 触发退出
|
|
trailing_stop_positive_offset = 0.015 # 盈利 1.5% 后才开始追踪
|
|
trailing_only_offset_is_reached = True
|
|
|
|
# ROI - 调整止盈策略
|
|
minimal_roi = {
|
|
"0": 0.015, # 1.5% 立即止盈（更保守）
|
|
"30": 0.01, # 30分钟后 1%
|
|
"120": 0.008, # 2小时后 0.8%
|
|
}
|
|
|
|
order_types = {
|
|
"entry": "market",
|
|
"exit": "market",
|
|
"stoploss": "market",
|
|
"stoploss_on_exchange": False,
|
|
}
|
|
|
|
# === 盘整背驰参数 ===
|
|
same_level_timeframe = 5 # 5分钟级别
|
|
pivot_window = 4 # 转折点确认窗口（增大减少噪音）
|
|
min_segment_length = 5 # 最小段长度（K线数）（增大减少假信号）
|
|
|
|
# 背驰判断参数（更严格）
|
|
beichi_price_threshold = 1.10 # 价格涨幅/跌幅阈值（允许10%范围内，更严格）
|
|
beichi_macd_threshold = 0.75 # MACD力度阈值（低于75%即背驰，更严格）
|
|
|
|
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""计算指标并识别盘整背驰"""
|
|
ticker = self.get_ticker_indicator()
|
|
|
|
# Resample 到 5m 进行同级别分解
|
|
dataframe_5m = resample_to_interval(dataframe, ticker * self.same_level_timeframe)
|
|
|
|
# 在 5m 上计算指标
|
|
dataframe_5m = self.add_indicators_5m(dataframe_5m)
|
|
|
|
# 识别盘整背驰
|
|
dataframe_5m = self.identify_panzheng_beichi(dataframe_5m)
|
|
|
|
# 合并回 1m dataframe
|
|
dataframe = resampled_merge(dataframe, dataframe_5m)
|
|
|
|
# 在 1m 上也计算基础指标
|
|
dataframe = self.add_indicators_1m(dataframe)
|
|
|
|
return dataframe
|
|
|
|
def add_indicators_5m(self, dataframe: DataFrame) -> DataFrame:
|
|
"""在5m级别计算指标"""
|
|
# MACD 用于识别背驰
|
|
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
|
|
dataframe['macd'] = macd['macd']
|
|
dataframe['macdsignal'] = macd['macdsignal']
|
|
dataframe['macdhist'] = macd['macdhist']
|
|
|
|
# EMA 用于识别趋势
|
|
dataframe['ema12'] = ta.EMA(dataframe, timeperiod=12)
|
|
dataframe['ema26'] = ta.EMA(dataframe, timeperiod=26)
|
|
dataframe['ema50'] = ta.EMA(dataframe, timeperiod=50)
|
|
dataframe['ema200'] = ta.EMA(dataframe, timeperiod=200)
|
|
|
|
# EMA趋势方向
|
|
dataframe['ema_trend_up'] = (dataframe['ema12'] > dataframe['ema26']) & (dataframe['ema26'] > dataframe['ema50'])
|
|
dataframe['ema_trend_dn'] = (dataframe['ema12'] < dataframe['ema26']) & (dataframe['ema26'] < dataframe['ema50'])
|
|
|
|
# 价格与EMA200关系
|
|
dataframe['price_above_ema200'] = dataframe['close'] > dataframe['ema200']
|
|
dataframe['price_below_ema200'] = dataframe['close'] < dataframe['ema200']
|
|
|
|
# 趋势强度
|
|
dataframe['ema12_slope'] = dataframe['ema12'].diff(5) / dataframe['ema12'].shift(5)
|
|
dataframe['ema26_slope'] = dataframe['ema26'].diff(5) / dataframe['ema26'].shift(5)
|
|
|
|
# 强趋势判断
|
|
dataframe['strong_uptrend'] = (
|
|
(dataframe['ema12_slope'] > 0) &
|
|
(dataframe['ema26_slope'] > 0) &
|
|
(dataframe['price_above_ema200'])
|
|
)
|
|
dataframe['strong_downtrend'] = (
|
|
(dataframe['ema12_slope'] < 0) &
|
|
(dataframe['ema26_slope'] < 0) &
|
|
(dataframe['price_below_ema200'])
|
|
)
|
|
|
|
# RSI
|
|
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
|
|
|
|
# ATR 用于波动率过滤
|
|
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
|
|
dataframe['atr_mean'] = dataframe['atr'].rolling(window=20).mean()
|
|
dataframe['volatility_ok'] = dataframe['atr'] > dataframe['atr_mean'] * 0.8
|
|
|
|
return dataframe
|
|
|
|
def add_indicators_1m(self, dataframe: DataFrame) -> DataFrame:
|
|
"""在1m级别计算基础指标"""
|
|
dataframe['rsi_1m'] = ta.RSI(dataframe, timeperiod=14)
|
|
dataframe['volume_mean'] = dataframe['volume'].rolling(window=20).mean()
|
|
|
|
# MACD 用于1m级别确认
|
|
macd_1m = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
|
|
dataframe['macd_1m'] = macd_1m['macd']
|
|
dataframe['macdsignal_1m'] = macd_1m['macdsignal']
|
|
dataframe['macdhist_1m'] = macd_1m['macdhist']
|
|
|
|
# MACD交叉
|
|
dataframe['macd_cross_up'] = (
|
|
(dataframe['macd_1m'] > dataframe['macdsignal_1m']) &
|
|
(dataframe['macd_1m'].shift(1) <= dataframe['macdsignal_1m'].shift(1))
|
|
)
|
|
dataframe['macd_cross_dn'] = (
|
|
(dataframe['macd_1m'] < dataframe['macdsignal_1m']) &
|
|
(dataframe['macd_1m'].shift(1) >= dataframe['macdsignal_1m'].shift(1))
|
|
)
|
|
|
|
return dataframe
|
|
|
|
def identify_panzheng_beichi(self, dataframe: DataFrame) -> DataFrame:
|
|
"""
|
|
识别盘整背驰
|
|
|
|
核心逻辑：
|
|
1. 识别局部转折点（高低点）
|
|
2. 构建同级别走势段（Ai）
|
|
3. 比较Ai与Ai+2的MACD力度
|
|
4. 判断盘整背驰：价格涨幅相近但MACD力度减弱
|
|
"""
|
|
df = dataframe.copy()
|
|
window = self.pivot_window
|
|
lookback = window + 1
|
|
|
|
# 初始化列
|
|
df['ai_index'] = -1
|
|
df['ai_type'] = 0 # 1: 上涨, -1: 下跌
|
|
df['ai_high'] = np.nan
|
|
df['ai_low'] = np.nan
|
|
df['ai_macd_max'] = np.nan
|
|
df['ai_macd_min'] = np.nan
|
|
df['beichi_long'] = False # 盘整底背驰（买入信号）
|
|
df['beichi_short'] = False # 盘整顶背驰（卖出信号）
|
|
|
|
# 识别局部高点
|
|
df['temp_high'] = df['high'].shift(window)
|
|
df['is_pivot_high'] = (
|
|
(df['temp_high'] == df['temp_high'].rolling(window=lookback).max()) &
|
|
(df['temp_high'].notna())
|
|
)
|
|
|
|
# 识别局部低点
|
|
df['temp_low'] = df['low'].shift(window)
|
|
df['is_pivot_low'] = (
|
|
(df['temp_low'] == df['temp_low'].rolling(window=lookback).min()) &
|
|
(df['temp_low'].notna())
|
|
)
|
|
|
|
# 逐行处理，识别走势段和背驰
|
|
ai_list = []
|
|
current_ai_start = None
|
|
current_ai_type = None
|
|
last_pivot_idx = None
|
|
|
|
for i in range(window, len(df)):
|
|
# 检查新的转折点
|
|
is_new_pivot = False
|
|
pivot_type = None
|
|
|
|
if df.iloc[i]['is_pivot_high']:
|
|
is_new_pivot = True
|
|
pivot_type = 'high'
|
|
elif df.iloc[i]['is_pivot_low']:
|
|
is_new_pivot = True
|
|
pivot_type = 'low'
|
|
|
|
if is_new_pivot and last_pivot_idx is not None:
|
|
# 完成一个走势段
|
|
if current_ai_start is not None:
|
|
seg_df = df.iloc[current_ai_start:last_pivot_idx]
|
|
if len(seg_df) >= self.min_segment_length:
|
|
high_val = seg_df['high'].max()
|
|
low_val = seg_df['low'].min()
|
|
macd_max = seg_df['macd'].max()
|
|
macd_min = seg_df['macd'].min()
|
|
|
|
# 判断走势类型
|
|
if current_ai_type is None:
|
|
if high_val > df.iloc[current_ai_start]['close']:
|
|
current_ai_type = 1
|
|
else:
|
|
current_ai_type = -1
|
|
|
|
ai_info = {
|
|
'start': current_ai_start,
|
|
'end': last_pivot_idx,
|
|
'type': current_ai_type,
|
|
'high': high_val,
|
|
'low': low_val,
|
|
'macd_max': macd_max,
|
|
'macd_min': macd_min,
|
|
}
|
|
ai_list.append(ai_info)
|
|
|
|
# 标记该段
|
|
df.iloc[current_ai_start:last_pivot_idx, df.columns.get_loc('ai_index')] = len(ai_list) - 1
|
|
df.iloc[current_ai_start:last_pivot_idx, df.columns.get_loc('ai_type')] = current_ai_type
|
|
df.iloc[current_ai_start:last_pivot_idx, df.columns.get_loc('ai_high')] = high_val
|
|
df.iloc[current_ai_start:last_pivot_idx, df.columns.get_loc('ai_low')] = low_val
|
|
df.iloc[current_ai_start:last_pivot_idx, df.columns.get_loc('ai_macd_max')] = macd_max
|
|
df.iloc[current_ai_start:last_pivot_idx, df.columns.get_loc('ai_macd_min')] = macd_min
|
|
|
|
# 判断背驰（Ai与Ai+2比较）
|
|
if len(ai_list) >= 3:
|
|
ai = ai_list[-3] # Ai
|
|
ai_plus_2 = ai_list[-1] # Ai+2
|
|
|
|
if ai['type'] == ai_plus_2['type']:
|
|
# 上涨段：比较向上力度
|
|
if ai['type'] == 1:
|
|
price_chg = (ai_plus_2['high'] - ai_plus_2['low']) / ai_plus_2['low'] if ai_plus_2['low'] > 0 else 0
|
|
price_chg_prev = (ai['high'] - ai['low']) / ai['low'] if ai['low'] > 0 else 0
|
|
macd_chg = ai_plus_2['macd_max']
|
|
macd_chg_prev = ai['macd_max']
|
|
|
|
# 顶背驰：价格涨幅相近但MACD力度减弱
|
|
if price_chg <= price_chg_prev * self.beichi_price_threshold and \
|
|
macd_chg < macd_chg_prev * self.beichi_macd_threshold:
|
|
idx = len(ai_list) - 1 # i+2的索引
|
|
if idx % 2 == 0: # 偶数 -> 卖出
|
|
df.iloc[last_pivot_idx, df.columns.get_loc('beichi_short')] = True
|
|
else: # 奇数 -> 买入
|
|
df.iloc[last_pivot_idx, df.columns.get_loc('beichi_long')] = True
|
|
|
|
# 下跌段：比较向下力度
|
|
else:
|
|
price_chg = abs((ai_plus_2['high'] - ai_plus_2['low']) / ai_plus_2['low']) if ai_plus_2['low'] > 0 else 0
|
|
price_chg_prev = abs((ai['high'] - ai['low']) / ai['low']) if ai['low'] > 0 else 0
|
|
macd_chg = abs(ai_plus_2['macd_min'])
|
|
macd_chg_prev = abs(ai['macd_min'])
|
|
|
|
# 底背驰：价格跌幅相近但MACD力度减弱
|
|
if price_chg <= price_chg_prev * self.beichi_price_threshold and \
|
|
macd_chg < macd_chg_prev * self.beichi_macd_threshold:
|
|
idx = len(ai_list) - 1
|
|
if idx % 2 == 0: # 偶数 -> 卖出
|
|
df.iloc[last_pivot_idx, df.columns.get_loc('beichi_short')] = True
|
|
else: # 奇数 -> 买入
|
|
df.iloc[last_pivot_idx, df.columns.get_loc('beichi_long')] = True
|
|
|
|
# 更新当前段信息
|
|
if pivot_type == 'high':
|
|
current_ai_type = -1 # 高点后向下
|
|
else:
|
|
current_ai_type = 1 # 低点后向上
|
|
current_ai_start = last_pivot_idx
|
|
|
|
if is_new_pivot:
|
|
last_pivot_idx = i
|
|
|
|
return df
|
|
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
ticker = self.get_ticker_indicator()
|
|
resample_col = f"resample_{ticker * self.same_level_timeframe}_"
|
|
|
|
# 5m 级别指标列名
|
|
beichi_long_col = f"{resample_col}beichi_long"
|
|
beichi_short_col = f"{resample_col}beichi_short"
|
|
ai_type_col = f"{resample_col}ai_type"
|
|
rsi_5m_col = f"{resample_col}rsi"
|
|
ema_trend_up_col = f"{resample_col}ema_trend_up"
|
|
ema_trend_dn_col = f"{resample_col}ema_trend_dn"
|
|
volatility_ok_col = f"{resample_col}volatility_ok"
|
|
strong_uptrend_col = f"{resample_col}strong_uptrend"
|
|
strong_downtrend_col = f"{resample_col}strong_downtrend"
|
|
price_above_ema200_col = f"{resample_col}price_above_ema200"
|
|
price_below_ema200_col = f"{resample_col}price_below_ema200"
|
|
|
|
# === 做多入场 ===
|
|
# 条件：盘整底背驰 + 强上升趋势确认
|
|
dataframe.loc[
|
|
(
|
|
# 核心信号：盘整底背驰
|
|
(dataframe[beichi_long_col] == True) &
|
|
|
|
# 强上升趋势确认（更严格）
|
|
(dataframe[strong_uptrend_col] == True) &
|
|
|
|
# RSI 确认（更严格：只在大趋势中操作）
|
|
(dataframe[rsi_5m_col] > 40) &
|
|
(dataframe[rsi_5m_col] < 60) &
|
|
|
|
# 1m 指标确认
|
|
(dataframe['macd_1m'] > dataframe['macdsignal_1m']) &
|
|
|
|
# 成交量确认
|
|
(dataframe['volume'] > dataframe['volume_mean'] * 1.5)
|
|
),
|
|
['enter_long', 'enter_tag']
|
|
] = (1, "pzbc_long")
|
|
|
|
# === 做空入场 ===
|
|
# 条件：盘整顶背驰 + 强下降趋势确认（更严格）
|
|
dataframe.loc[
|
|
(
|
|
# 核心信号：盘整顶背驰
|
|
(dataframe[beichi_short_col] == True) &
|
|
|
|
# 强下降趋势确认
|
|
(dataframe[strong_downtrend_col] == True) &
|
|
|
|
# RSI 确认
|
|
(dataframe[rsi_5m_col] > 40) &
|
|
(dataframe[rsi_5m_col] < 60) &
|
|
|
|
# 1m 指标确认
|
|
(dataframe['macd_1m'] < dataframe['macdsignal_1m']) &
|
|
|
|
# 成交量确认
|
|
(dataframe['volume'] > dataframe['volume_mean'] * 1.5)
|
|
),
|
|
['enter_short', 'enter_tag']
|
|
] = (1, "pzbc_short")
|
|
|
|
return dataframe
|
|
|
|
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
出场逻辑
|
|
|
|
多头出场：
|
|
1. 出现盘整顶背驰
|
|
2. 趋势转弱
|
|
|
|
空头出场：
|
|
1. 出现盘整底背驰
|
|
2. 趋势转弱
|
|
"""
|
|
ticker = self.get_ticker_indicator()
|
|
resample_col = f"resample_{ticker * self.same_level_timeframe}_"
|
|
|
|
beichi_long_col = f"{resample_col}beichi_long"
|
|
beichi_short_col = f"{resample_col}beichi_short"
|
|
ai_type_col = f"{resample_col}ai_type"
|
|
rsi_5m_col = f"{resample_col}rsi"
|
|
ema_trend_dn_col = f"{resample_col}ema_trend_dn"
|
|
strong_downtrend_col = f"{resample_col}strong_downtrend"
|
|
strong_uptrend_col = f"{resample_col}strong_uptrend"
|
|
|
|
# === 多头出场 ===
|
|
dataframe.loc[
|
|
(
|
|
# 出现盘整顶背驰 -> 退出多头
|
|
(dataframe[beichi_short_col] == True) |
|
|
|
|
# 趋势转弱
|
|
(
|
|
(dataframe[ai_type_col] == -1) &
|
|
(dataframe[rsi_5m_col] > 55)
|
|
) |
|
|
|
|
# 强下跌趋势
|
|
(dataframe[strong_downtrend_col] == True)
|
|
),
|
|
['exit_long', 'exit_tag']
|
|
] = (1, "pzbc_exit_long")
|
|
|
|
# === 空头出场 ===
|
|
dataframe.loc[
|
|
(
|
|
# 出现盘整底背驰 -> 退出空头
|
|
(dataframe[beichi_long_col] == True) |
|
|
|
|
# 趋势转弱
|
|
(
|
|
(dataframe[ai_type_col] == 1) &
|
|
(dataframe[rsi_5m_col] < 45)
|
|
) |
|
|
|
|
# 强上涨趋势
|
|
(dataframe[strong_uptrend_col] == True)
|
|
),
|
|
['exit_short', 'exit_tag']
|
|
] = (1, "pzbc_exit_short")
|
|
|
|
return dataframe
|
|
|
|
def get_ticker_indicator(self) -> int:
|
|
"""获取 timeframe 的分钟数"""
|
|
return int(self.timeframe[:-1])
|
|
|