346 lines
18 KiB
Python
346 lines
18 KiB
Python
# --- Do not remove these libs ---
|
|
from statistics import median
|
|
from freqtrade.strategy import IStrategy, stoploss_from_absolute
|
|
import sys
|
|
import os
|
|
# 添加父目录到系统路径
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
from ChanLun import ChanLun
|
|
from ChanLun_Classifier import ChanLunClassifier
|
|
from ChanEnum import Chan_FX_TYPE, Chan_KLC_FX, Chan_BI_DIR, Chan_KLC_FX
|
|
from ChanPY import ChanPY
|
|
# --------------------------------
|
|
from technical.util import resample_to_interval, resampled_merge
|
|
import talib.abstract as ta
|
|
import numpy as np
|
|
import pandas as pd
|
|
from pandas import DataFrame
|
|
from datetime import datetime, timedelta
|
|
from freqtrade.persistence import Trade, Order
|
|
from typing import Optional
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
### Now you can use logger.info('asfd') to log
|
|
# freqtrade plot-dataframe --strategy ChanLun_MACD --datadir user_data/data/binance -c ./user_data/ChanLun_MACD.json --timerange=20250309-
|
|
|
|
# freqtrade trade -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies
|
|
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies --timerange=20250812-
|
|
# freqtrade download-data -c ./user_data/Chan/config/ChanLun_MACD.json -t 5m --pairs BTC/USDT:USDT --timerange=20240101-
|
|
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_MACD.json -e 200 --timerange=20250201-20250401
|
|
|
|
# sudo docker compose run --rm chanlun_btc backtesting -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies --timerange=20250721-
|
|
# sudo docker compose run --rm chanlun_btc download-data -c ./user_data/Chan/config/ChanLun_MACD.json --pairs BTC/USDT:USDT -t 1m --timerange 20240101-
|
|
# sudo docker compose run --rm chanlun_btc trade -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies
|
|
|
|
class ChanLun_MACD(IStrategy):
|
|
# 标准 Freqtrade 策略：使用 归零轴 + 背离/隐性形态 进行交易
|
|
INTERFACE_VERSION: int = 3
|
|
|
|
# 基本参数
|
|
timeframe = '5m'
|
|
startup_candle_count = 300
|
|
can_short = True
|
|
|
|
# ROI/止损（止损由自定义 ATR 控制，此处设大）
|
|
minimal_roi = {"0": 0.1}
|
|
stoploss = -0.3
|
|
use_custom_stoploss = True
|
|
|
|
# 使用市价单，避免回测限价成交不充分导致信号丢单
|
|
order_types = {
|
|
"entry": "market",
|
|
"exit": "market",
|
|
"stoploss": "market",
|
|
"stoploss_on_exchange": False,
|
|
"stoploss_on_exchange_interval": 60,
|
|
}
|
|
|
|
# 过滤：ATR 太小不进场
|
|
# 为确保先跑出单，暂不限制 ATR（回测确认后再收紧）
|
|
min_atr_value = 0.0
|
|
# 可调参数
|
|
eps_zero_param = 0.06
|
|
div_shift = 2
|
|
zero_recent_lookback = 3
|
|
|
|
# ============ 指标计算 ============
|
|
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
# MACD
|
|
macd = ta.MACD(dataframe)
|
|
dataframe['macd'] = macd['macd']
|
|
dataframe['macdsignal'] = macd['macdsignal']
|
|
dataframe['macdhist'] = macd['macdhist']
|
|
|
|
# EMA24/EMA52（若无ema24，使用ema26近似）
|
|
dataframe['ema24'] = ta.EMA(dataframe, timeperiod=24)
|
|
dataframe['ema52'] = ta.EMA(dataframe, timeperiod=52)
|
|
|
|
# ATR
|
|
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
|
|
|
|
# 归零轴判定
|
|
eps_zero = self.eps_zero_param # 可调
|
|
dataframe['zero_cross'] = (dataframe['macd'].shift(1) * dataframe['macd'] <= 0)
|
|
dataframe['zero_near'] = (dataframe['macd'].abs() <= eps_zero) | ((dataframe['macd'].abs() <= eps_zero) & (dataframe['macdsignal'].abs() <= eps_zero))
|
|
dataframe['zero_axis'] = dataframe['zero_cross'] | dataframe['zero_near']
|
|
|
|
# 价格触碰均线
|
|
band24 = 0.006 # 放宽贴近阈值
|
|
band52 = 0.008
|
|
dataframe['near_ema24'] = (dataframe['ema24'] > 0) & ((dataframe['close'] - dataframe['ema24']).abs() / dataframe['ema24'] <= band24)
|
|
dataframe['near_ema52'] = (dataframe['ema52'] > 0) & ((dataframe['close'] - dataframe['ema52']).abs() / dataframe['ema52'] <= band52)
|
|
|
|
# 背离/隐性背离（可调间隔），先简化到 MACD 快线
|
|
sh = self.div_shift
|
|
dataframe['bull_div'] = (dataframe['close'] < dataframe['close'].shift(sh)) & (dataframe['macd'] > dataframe['macd'].shift(sh)) & (dataframe['macd'] < 0)
|
|
dataframe['bear_div'] = (dataframe['close'] > dataframe['close'].shift(sh)) & (dataframe['macd'] < dataframe['macd'].shift(sh)) & (dataframe['macd'] > 0)
|
|
dataframe['hidden_bull'] = (dataframe['close'] > dataframe['close'].shift(sh)) & (dataframe['macd'] < dataframe['macd'].shift(sh)) & (dataframe['macd'] < 0)
|
|
dataframe['hidden_bear'] = (dataframe['close'] < dataframe['close'].shift(sh)) & (dataframe['macd'] > dataframe['macd'].shift(sh)) & (dataframe['macd'] > 0)
|
|
|
|
# 近N根出现过归零轴（解决“同一根同时满足”过严问题）
|
|
zr = dataframe['zero_axis']
|
|
for i in range(1, self.zero_recent_lookback):
|
|
zr = zr | dataframe['zero_axis'].shift(i)
|
|
dataframe['zero_recent'] = zr.fillna(False)
|
|
|
|
# 近零处快线穿越慢线（补充触发源）
|
|
near_zero_now = dataframe['macd'].abs() <= (eps_zero * 2)
|
|
cross_up = (dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macd'].shift(1) <= dataframe['macdsignal'].shift(1))
|
|
cross_dn = (dataframe['macd'] < dataframe['macdsignal']) & (dataframe['macd'].shift(1) >= dataframe['macdsignal'].shift(1))
|
|
dataframe['zero_cross_up_near'] = near_zero_now & cross_up
|
|
dataframe['zero_cross_dn_near'] = near_zero_now & cross_dn
|
|
|
|
# ====== 结构：基于 ChanMACD 的 UnitTF 结束点（与零轴定义一致） ======
|
|
try:
|
|
from ChanKLU import ChanKLU
|
|
from ChanMACD import ChanMACD
|
|
klu_list = []
|
|
prev = None
|
|
for idx, row in dataframe.iterrows():
|
|
klu = ChanKLU(
|
|
time=idx,
|
|
open=float(row.get('open', 0) or 0),
|
|
high=float(row.get('high', 0) or 0),
|
|
low=float(row.get('low', 0) or 0),
|
|
close=float(row.get('close', 0) or 0),
|
|
volume=float(row.get('volume', 0) or 0),
|
|
)
|
|
klu.set_pre(prev)
|
|
if prev:
|
|
prev.set_next(klu)
|
|
klu.ema24 = float(row.get('ema24', 0) or 0)
|
|
klu.ema52 = float(row.get('ema52', 0) or 0)
|
|
klu.set_indicators({'macd': row.get('macd'), 'macdsignal': row.get('macdsignal'), 'macdhist': row.get('macdhist')})
|
|
klu.set_idx(len(klu_list))
|
|
klu_list.append(klu)
|
|
prev = klu
|
|
cm = ChanMACD(klu_list)
|
|
end_up = {}
|
|
end_dn = {}
|
|
for u in cm.unittf_list:
|
|
if not getattr(u, 'end_klu', None):
|
|
continue
|
|
dirv = getattr(u, 'dir', 0)
|
|
timev = u.end_klu.time
|
|
if dirv >= 0:
|
|
end_up[timev] = True
|
|
else:
|
|
end_dn[timev] = True
|
|
dataframe['unit_end_up'] = dataframe.index.to_series().apply(lambda t: bool(end_up.get(t, False))).astype(bool)
|
|
dataframe['unit_end_dn'] = dataframe.index.to_series().apply(lambda t: bool(end_dn.get(t, False))).astype(bool)
|
|
except Exception:
|
|
dataframe['unit_end_up'] = False
|
|
dataframe['unit_end_dn'] = False
|
|
|
|
# ====== 高位空 / 低位多 形态（高位横盘柱衰/低位横盘柱回升） ======
|
|
eps_high = max(eps_zero * 2, 0.08)
|
|
H = 5
|
|
N = 3
|
|
macd_high = (dataframe['macd'] > eps_high)
|
|
macd_low = (dataframe['macd'] < -eps_high)
|
|
# 黄白线高/低位区（结合快慢线）
|
|
dataframe['macd_high_zone'] = (dataframe['macd'] > eps_high) & (dataframe['macdsignal'] > eps_high)
|
|
dataframe['macd_low_zone'] = (dataframe['macd'] < -eps_high) & (dataframe['macdsignal'] < -eps_high)
|
|
hist_down = (dataframe['macdhist'].diff() < 0)
|
|
hist_up = (dataframe['macdhist'].diff() > 0)
|
|
dataframe['hs_window'] = macd_high.rolling(H).sum() == H
|
|
dataframe['ls_window'] = macd_low.rolling(H).sum() == H
|
|
dataframe['hist_down_streak'] = hist_down.rolling(N).sum() == N
|
|
dataframe['hist_up_streak'] = hist_up.rolling(N).sum() == N
|
|
dataframe['high_short_setup'] = (dataframe['hs_window'] & dataframe['hist_down_streak']).fillna(False)
|
|
dataframe['low_long_setup'] = (dataframe['ls_window'] & dataframe['hist_up_streak']).fillna(False)
|
|
|
|
# ====== 基于枢轴点(局部高低点)的直方图背离/隐性背离检测 ======
|
|
# 枢轴点定义：高点 high[i] > high[i-1] 且 >= high[i+1]；低点相反
|
|
pivot_high = (dataframe['high'] > dataframe['high'].shift(1)) & (dataframe['high'] >= dataframe['high'].shift(-1))
|
|
pivot_low = (dataframe['low'] < dataframe['low'].shift(1)) & (dataframe['low'] <= dataframe['low'].shift(-1))
|
|
# 直方图峰/谷
|
|
hist_peak = (dataframe['macdhist'] > dataframe['macdhist'].shift(1)) & (dataframe['macdhist'] >= dataframe['macdhist'].shift(-1))
|
|
hist_trough = (dataframe['macdhist'] < dataframe['macdhist'].shift(1)) & (dataframe['macdhist'] <= dataframe['macdhist'].shift(-1))
|
|
# 仅在对应象限判定
|
|
hist_peak_pos = hist_peak & (dataframe['macd'] > 0)
|
|
hist_trough_neg = hist_trough & (dataframe['macd'] < 0)
|
|
# 抽取序列上的上一枢轴值
|
|
ph_price = dataframe['high'].where(pivot_high)
|
|
pl_price = dataframe['low'].where(pivot_low)
|
|
ph_hist = dataframe['macdhist'].where(hist_peak_pos)
|
|
pl_hist = dataframe['macdhist'].where(hist_trough_neg)
|
|
prev_ph_price = ph_price.shift(1).ffill()
|
|
prev_pl_price = pl_price.shift(1).ffill()
|
|
prev_ph_hist = ph_hist.shift(1).ffill()
|
|
prev_pl_hist = pl_hist.shift(1).ffill()
|
|
# 经典背离
|
|
bear_div_pivot = pivot_high & hist_peak_pos & (dataframe['high'] > prev_ph_price) & (dataframe['macdhist'] < prev_ph_hist)
|
|
bull_div_pivot = pivot_low & hist_trough_neg & (dataframe['low'] < prev_pl_price) & (dataframe['macdhist'] > prev_pl_hist)
|
|
# 隐性背离（顺势）
|
|
hidden_bear_pivot = pivot_high & hist_peak_pos & (dataframe['high'] < prev_ph_price) & (dataframe['macdhist'] > prev_ph_hist)
|
|
hidden_bull_pivot = pivot_low & hist_trough_neg & (dataframe['low'] > prev_pl_price) & (dataframe['macdhist'] < prev_pl_hist)
|
|
dataframe['bear_div_pivot'] = bear_div_pivot.fillna(False)
|
|
dataframe['bull_div_pivot'] = bull_div_pivot.fillna(False)
|
|
dataframe['hidden_bear_pivot'] = hidden_bear_pivot.fillna(False)
|
|
dataframe['hidden_bull_pivot'] = hidden_bull_pivot.fillna(False)
|
|
|
|
# ====== 统计日志（便于回测定位信号规模） ======
|
|
try:
|
|
pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A'
|
|
cnt_zero_recent = int(dataframe['zero_recent'].fillna(False).sum())
|
|
cnt_zcup = int(dataframe['zero_cross_up_near'].fillna(False).sum())
|
|
cnt_zcdn = int(dataframe['zero_cross_dn_near'].fillna(False).sum())
|
|
cnt_bull_div = int(dataframe['bull_div'].fillna(False).sum())
|
|
cnt_bear_div = int(dataframe['bear_div'].fillna(False).sum())
|
|
cnt_hbull = int(dataframe['hidden_bull'].fillna(False).sum())
|
|
cnt_hbear = int(dataframe['hidden_bear'].fillna(False).sum())
|
|
cnt_u_end_up = int(dataframe.get('unit_end_up', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
cnt_u_end_dn = int(dataframe.get('unit_end_dn', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
cnt_hs = int(dataframe.get('high_short_setup', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
cnt_ll = int(dataframe.get('low_long_setup', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
cnt_bear_div_p = int(dataframe.get('bear_div_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
cnt_bull_div_p = int(dataframe.get('bull_div_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
cnt_hbear_p = int(dataframe.get('hidden_bear_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
cnt_hbull_p = int(dataframe.get('hidden_bull_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
zones_high = int(dataframe.get('macd_high_zone', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
zones_low = int(dataframe.get('macd_low_zone', pd.Series(False, index=dataframe.index)).fillna(False).sum())
|
|
logger.info(f"[{pair}] IND zr={cnt_zero_recent} zcup={cnt_zcup} zcdn={cnt_zcdn} div(bull={cnt_bull_div},bear={cnt_bear_div},hb={cnt_hbull},hs={cnt_hbear}) piv(bull={cnt_bull_div_p},bear={cnt_bear_div_p},hb={cnt_hbull_p},hs={cnt_hbear_p}) zones(high={zones_high},low={zones_low}) unit_end(up={cnt_u_end_up},dn={cnt_u_end_dn}) setup(hs={cnt_hs},ll={cnt_ll})")
|
|
except Exception:
|
|
pass
|
|
|
|
# 进场模板（在 populate_entry_trend 中使用）
|
|
return dataframe
|
|
|
|
# ============ 入场/出场信号 ============
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
dataframe['enter_long'] = 0
|
|
dataframe['enter_short'] = 0
|
|
|
|
# 做多：柱形枢轴底背离/隐性多 + 低位区 或 近零轴（允许只要枢轴+近零即可）
|
|
s_false = pd.Series(False, index=dataframe.index)
|
|
bull_hist = (
|
|
dataframe.get('bull_div_pivot', s_false).fillna(False)
|
|
|
|
|
dataframe.get('hidden_bull_pivot', s_false).fillna(False)
|
|
)
|
|
low_zone = dataframe.get('macd_low_zone', s_false).fillna(False)
|
|
# 放宽近零阈值以确保产生成交
|
|
near_zero = (dataframe['macd'].abs() <= (self.eps_zero_param * 2.0)).fillna(False)
|
|
long_cond = (bull_hist & (low_zone | near_zero | dataframe['zero_recent']))
|
|
dataframe.loc[long_cond, 'enter_long'] = 1
|
|
|
|
# 做空：柱形枢轴顶背离/隐性空 + 高位区 或 近零轴（允许只要枢轴+近零即可）
|
|
bear_hist = (
|
|
dataframe.get('bear_div_pivot', s_false).fillna(False)
|
|
|
|
|
dataframe.get('hidden_bear_pivot', s_false).fillna(False)
|
|
)
|
|
high_zone = dataframe.get('macd_high_zone', s_false).fillna(False)
|
|
short_cond = (bear_hist & (high_zone | near_zero | dataframe['zero_recent']))
|
|
dataframe.loc[short_cond, 'enter_short'] = 1
|
|
|
|
# ====== 入场标签与统计（聚焦柱形+区位） ======
|
|
try:
|
|
long_highzone = (bull_hist & low_zone).fillna(False)
|
|
long_nearzero = (bull_hist & near_zero).fillna(False)
|
|
short_highzone = (bear_hist & high_zone).fillna(False)
|
|
short_nearzero = (bear_hist & near_zero).fillna(False)
|
|
|
|
dataframe['enter_tag'] = ''
|
|
dataframe['long_tag_tmp'] = np.select(
|
|
[long_highzone, long_nearzero],
|
|
['HIST_BULL_DIV_HIGHZONE', 'HIST_BULL_DIV_NEARZERO'],
|
|
default=''
|
|
)
|
|
dataframe['short_tag_tmp'] = np.select(
|
|
[short_highzone, short_nearzero],
|
|
['HIST_BEAR_DIV_HIGHZONE', 'HIST_BEAR_DIV_NEARZERO'],
|
|
default=''
|
|
)
|
|
dataframe.loc[dataframe['enter_long'] == 1, 'enter_tag'] = dataframe.loc[dataframe['enter_long'] == 1, 'long_tag_tmp'].replace('', 'OTHER')
|
|
dataframe.loc[dataframe['enter_short'] == 1, 'enter_tag'] = dataframe.loc[dataframe['enter_short'] == 1, 'short_tag_tmp'].replace('', 'OTHER')
|
|
|
|
pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A'
|
|
cnt_long = int((dataframe['enter_long'] == 1).sum())
|
|
cnt_short = int((dataframe['enter_short'] == 1).sum())
|
|
cnt_l_hz = int(long_highzone.sum()); cnt_l_nz = int(long_nearzero.sum())
|
|
cnt_s_hz = int(short_highzone.sum()); cnt_s_nz = int(short_nearzero.sum())
|
|
# 最终可下单信号数量（enter_* 列）
|
|
el = int((dataframe.get('enter_long', 0) == 1).sum())
|
|
es = int((dataframe.get('enter_short', 0) == 1).sum())
|
|
logger.info(f"[{pair}] SIG long={cnt_long} short={cnt_short} long_parts(hz={cnt_l_hz},nz={cnt_l_nz}) short_parts(hz={cnt_s_hz},nz={cnt_s_nz}) ENTER(el={el},es={es})")
|
|
except Exception:
|
|
pass
|
|
|
|
# 不追加 UnitTF 入场，聚焦柱形+区位组合
|
|
try:
|
|
idx_long = list(dataframe.index[dataframe['enter_long'] == 1])
|
|
idx_short = list(dataframe.index[dataframe['enter_short'] == 1])
|
|
def _fmt(ts_list):
|
|
return [str(ts_list[i]) for i in range(min(5, len(ts_list)))] + (["..."] if len(ts_list) > 10 else []) + [str(ts_list[i]) for i in range(max(0, len(ts_list)-5), len(ts_list))] if ts_list else []
|
|
pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A'
|
|
logger.info(f"[{pair}] ENTER_LONG idx samples: {_fmt(idx_long)}")
|
|
logger.info(f"[{pair}] ENTER_SHORT idx samples: {_fmt(idx_short)}")
|
|
except Exception:
|
|
pass
|
|
return dataframe
|
|
|
|
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
dataframe['exit_long'] = 0
|
|
dataframe['exit_short'] = 0
|
|
|
|
# 退出：MACD 反向穿越零轴 或 触碰 EMA52 失败
|
|
long_exit = (dataframe['macd'] < 0) | (dataframe['near_ema52'] & (dataframe['macd'] < dataframe['macdsignal']))
|
|
short_exit = (dataframe['macd'] > 0) | (dataframe['near_ema52'] & (dataframe['macd'] > dataframe['macdsignal']))
|
|
dataframe.loc[long_exit, 'exit_long'] = 1
|
|
dataframe.loc[short_exit, 'exit_short'] = 1
|
|
return dataframe
|
|
|
|
# ============ 过滤与止损 ============
|
|
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
|
|
time_in_force: str, current_time: datetime, entry_tag: str | None,
|
|
side: str, **kwargs) -> bool:
|
|
# 暂时放开所有过滤，确保先产生成交，再逐步收紧
|
|
return True
|
|
|
|
def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
|
|
# 首次进场保存 ATR 作为 1x 止损距离
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
|
|
if dataframe is None or len(dataframe) == 0:
|
|
return None
|
|
last = dataframe.iloc[-1].squeeze()
|
|
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
|
|
entry_atr = float(last.get('atr', 0) or 0)
|
|
trade.set_custom_data(key="entry_atr", value=entry_atr)
|
|
logger.info(f"保存开仓ATR: {entry_atr}")
|
|
return None
|
|
|
|
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
|
|
current_rate: float, current_profit: float, after_fill: bool,
|
|
**kwargs) -> float | None:
|
|
# 1x ATR 止损
|
|
entry_atr = trade.get_custom_data(key="entry_atr")
|
|
if entry_atr is None:
|
|
# 兜底：5%
|
|
return -0.05
|
|
if trade.is_short:
|
|
stop_price = trade.open_rate + float(entry_atr)
|
|
else:
|
|
stop_price = trade.open_rate - float(entry_atr)
|
|
return stoploss_from_absolute(stop_price, current_rate, is_short=trade.is_short) |