471 lines
16 KiB
Python
471 lines
16 KiB
Python
# ============================================================================
|
||
# Structure Flow Strategy v1.1
|
||
# 纯价格结构策略 — 零技术指标，价格行为学驱动
|
||
#
|
||
# 版本变化 v1.0 → v1.1:
|
||
# - 主时间框架：5M → 1H（匹配用户实际交易尺度）
|
||
# - 信息时间框架：D1 → 4H → 1H
|
||
# - 启用做空（can_short=True），适配无杠杆合约交易
|
||
# - 硬止损改为结构失效点 + 1% 缓冲（不再用固定百分比）
|
||
# - 修复 custom_stoploss 动态跟踪结构位
|
||
#
|
||
# 设计哲学:
|
||
# 趋势由 HH/HL 定义，支撑阻力由 Swing Point 定义，
|
||
# 止损由结构失效点定义，出场由结构反转定义。
|
||
#
|
||
# 多时间框架:
|
||
# D1 → 宏观结构方向
|
||
# 4H → 中期结构位 + 入场区域判定
|
||
# 1H → K线形态确认入场时机
|
||
# ============================================================================
|
||
|
||
from datetime import datetime
|
||
import numpy as np
|
||
import pandas as pd
|
||
from pandas import DataFrame
|
||
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, informative
|
||
from freqtrade.persistence import Trade
|
||
|
||
|
||
class StructureFlowStrategy(IStrategy):
|
||
"""
|
||
Structure Flow Strategy v1.1 — 纯价格结构策略
|
||
|
||
不使用任何技术指标（无 EMA、ATR、RSI、MACD、布林带等）。
|
||
一切信号来源于价格本身的 OHLC 数据和由此推导的结构信息。
|
||
|
||
趋势判断:
|
||
HH + HL → 上升趋势（Bullish Structure）
|
||
LH + LL → 下降趋势（Bearish Structure）
|
||
|
||
入场逻辑:
|
||
做多: D1上升结构 + 价格在4H Swing区间下半区 + 1H看涨K线形态
|
||
做空: D1下降结构 + 价格在4H Swing区间上半区 + 1H看跌K线形态
|
||
|
||
止损逻辑:
|
||
初始止损: 4H 最近 Swing Low（做多）/ Swing High（做空）+ 1% 缓冲
|
||
动态止损: custom_stoploss 随新 Swing Point 形成而跟踪
|
||
"""
|
||
|
||
# ── 基础配置 ──────────────────────────────────────────
|
||
|
||
timeframe = "1h"
|
||
can_short = True # v1.1 启用做空，适配无杠杆合约
|
||
stoploss = -0.05 # 硬止损 5%，实际由 custom_stoploss 动态管理
|
||
use_custom_stoploss = True
|
||
minimal_roi = {"0": 100} # 不设时间止盈，出场由结构决定
|
||
max_open_trades = 1
|
||
|
||
# 回测参数
|
||
startup_candle_count = 40 # 需要更多历史数据（1H 级别）
|
||
|
||
# ── 可调参数 ──────────────────────────────────────────
|
||
|
||
swing_lookback_d1 = IntParameter(
|
||
2, 10, default=5, space="buy",
|
||
)
|
||
swing_lookback_h4 = IntParameter(
|
||
2, 10, default=5, space="buy",
|
||
)
|
||
|
||
# Pin Bar 确认强度：影线至少是实体的 N 倍
|
||
pin_bar_wick_ratio = DecimalParameter(
|
||
1.5, 4.0, default=2.0, space="buy",
|
||
)
|
||
|
||
# 结构止损缓冲（%）：止损设在结构位之外一点，避免被噪音扫损
|
||
sl_buffer_pct = DecimalParameter(
|
||
0.005, 0.03, default=0.01, space="sell",
|
||
optimize=True,
|
||
)
|
||
|
||
# ================================================================
|
||
# 工具函数 — 纯价格计算，不依赖任何技术指标
|
||
# ================================================================
|
||
|
||
@staticmethod
|
||
def _detect_swing_points(
|
||
high: pd.Series,
|
||
low: pd.Series,
|
||
lookback: int,
|
||
) -> tuple[pd.Series, pd.Series]:
|
||
"""
|
||
检测 Swing High 和 Swing Low。
|
||
|
||
纯价格比较：
|
||
- Swing High: 当前高点 > 左右各 lookback 根K线的所有高点
|
||
- Swing Low: 当前低点 < 左右各 lookback 根K线的所有低点
|
||
"""
|
||
n = len(high)
|
||
is_swing_high = np.full(n, False)
|
||
is_swing_low = np.full(n, False)
|
||
|
||
for i in range(lookback, n - lookback):
|
||
window_high = high.iloc[i - lookback : i + lookback + 1]
|
||
window_low = low.iloc[i - lookback : i + lookback + 1]
|
||
|
||
if high.iloc[i] == window_high.max():
|
||
is_swing_high[i] = True
|
||
if low.iloc[i] == window_low.min():
|
||
is_swing_low[i] = True
|
||
|
||
return (
|
||
pd.Series(is_swing_high, index=high.index),
|
||
pd.Series(is_swing_low, index=low.index),
|
||
)
|
||
|
||
@staticmethod
|
||
def _build_structure(
|
||
high: pd.Series,
|
||
low: pd.Series,
|
||
close: pd.Series,
|
||
swing_high: pd.Series,
|
||
swing_low: pd.Series,
|
||
) -> DataFrame:
|
||
"""
|
||
从 Swing Points 构建市场结构信息。
|
||
|
||
返回值包含:
|
||
trend_up / trend_down：当前处于上升/下降结构
|
||
support：最近 Swing Low 价格
|
||
resistance：最近 Swing High 价格
|
||
in_demand：价格在下半区（做多区域）
|
||
in_supply：价格在上半区（做空区域）
|
||
"""
|
||
n = len(high)
|
||
|
||
trend_up_arr = np.full(n, False)
|
||
trend_down_arr = np.full(n, False)
|
||
nearest_support = np.full(n, np.nan)
|
||
nearest_resistance = np.full(n, np.nan)
|
||
in_demand_zone = np.full(n, False)
|
||
in_supply_zone = np.full(n, False)
|
||
|
||
sh_prices: list[float] = []
|
||
sl_prices: list[float] = []
|
||
|
||
for i in range(n):
|
||
# ── 更新 Swing Point 队列 ──
|
||
if swing_high.iloc[i] and not np.isnan(high.iloc[i]):
|
||
sh_prices.append(high.iloc[i])
|
||
if len(sh_prices) > 4:
|
||
sh_prices.pop(0)
|
||
|
||
if swing_low.iloc[i] and not np.isnan(low.iloc[i]):
|
||
sl_prices.append(low.iloc[i])
|
||
if len(sl_prices) > 4:
|
||
sl_prices.pop(0)
|
||
|
||
# ── 趋势判断 ──
|
||
if len(sh_prices) >= 2 and len(sl_prices) >= 2:
|
||
latest_sh, prev_sh = sh_prices[-1], sh_prices[-2]
|
||
latest_sl, prev_sl = sl_prices[-1], sl_prices[-2]
|
||
|
||
if latest_sh > prev_sh and latest_sl > prev_sl:
|
||
trend_up_arr[i] = True
|
||
trend_down_arr[i] = False
|
||
elif latest_sh < prev_sh and latest_sl < prev_sl:
|
||
trend_up_arr[i] = False
|
||
trend_down_arr[i] = True
|
||
else:
|
||
if i > 0:
|
||
trend_up_arr[i] = trend_up_arr[i - 1]
|
||
trend_down_arr[i] = trend_down_arr[i - 1]
|
||
elif i > 0:
|
||
trend_up_arr[i] = trend_up_arr[i - 1]
|
||
trend_down_arr[i] = trend_down_arr[i - 1]
|
||
|
||
# ── 最近支撑/阻力 ──
|
||
if sl_prices:
|
||
nearest_support[i] = sl_prices[-1]
|
||
elif i > 0:
|
||
nearest_support[i] = nearest_support[i - 1]
|
||
|
||
if sh_prices:
|
||
nearest_resistance[i] = sh_prices[-1]
|
||
elif i > 0:
|
||
nearest_resistance[i] = nearest_resistance[i - 1]
|
||
|
||
# ── 入场区域：用 Swing 区间中点划分 ──
|
||
if (
|
||
not np.isnan(nearest_support[i])
|
||
and not np.isnan(nearest_resistance[i])
|
||
and nearest_resistance[i] > nearest_support[i]
|
||
):
|
||
mid = (nearest_support[i] + nearest_resistance[i]) / 2.0
|
||
in_demand_zone[i] = low.iloc[i] <= mid
|
||
in_supply_zone[i] = high.iloc[i] >= mid
|
||
elif i > 0:
|
||
in_demand_zone[i] = in_demand_zone[i - 1]
|
||
in_supply_zone[i] = in_supply_zone[i - 1]
|
||
|
||
result = DataFrame(
|
||
{
|
||
"trend_up": trend_up_arr,
|
||
"trend_down": trend_down_arr,
|
||
"support": nearest_support,
|
||
"resistance": nearest_resistance,
|
||
"in_demand": in_demand_zone,
|
||
"in_supply": in_supply_zone,
|
||
},
|
||
index=high.index,
|
||
)
|
||
return result
|
||
|
||
@staticmethod
|
||
def _detect_candle_patterns(
|
||
o: pd.Series,
|
||
h: pd.Series,
|
||
l: pd.Series,
|
||
c: pd.Series,
|
||
pin_ratio: float,
|
||
) -> tuple[pd.Series, pd.Series, pd.Series, pd.Series]:
|
||
"""
|
||
检测 K 线形态 — 纯 OHLC 计算。
|
||
"""
|
||
body = abs(c - o)
|
||
upper_wick = h - np.maximum(o, c)
|
||
lower_wick = np.minimum(o, c) - l
|
||
total_range = h - l
|
||
|
||
valid_range = total_range > 0
|
||
valid_body = body > 0
|
||
|
||
bullish_pin = (
|
||
valid_range
|
||
& valid_body
|
||
& (lower_wick >= pin_ratio * body)
|
||
& (upper_wick <= 0.5 * body)
|
||
)
|
||
|
||
bearish_pin = (
|
||
valid_range
|
||
& valid_body
|
||
& (upper_wick >= pin_ratio * body)
|
||
& (lower_wick <= 0.5 * body)
|
||
)
|
||
|
||
prev_body = body.shift(1)
|
||
prev_o = o.shift(1)
|
||
prev_c = c.shift(1)
|
||
|
||
bullish_engulf = (
|
||
(c > o)
|
||
& (prev_c < prev_o)
|
||
& (body > prev_body)
|
||
)
|
||
|
||
bearish_engulf = (
|
||
(c < o)
|
||
& (prev_c > prev_o)
|
||
& (body > prev_body)
|
||
)
|
||
|
||
return (
|
||
pd.Series(bullish_pin, index=c.index),
|
||
pd.Series(bearish_pin, index=c.index),
|
||
pd.Series(bullish_engulf, index=c.index),
|
||
pd.Series(bearish_engulf, index=c.index),
|
||
)
|
||
|
||
# ================================================================
|
||
# 信息时间框架 — D1 宏观结构
|
||
# ================================================================
|
||
|
||
@informative("1d")
|
||
def populate_indicators_1d(
|
||
self, dataframe: DataFrame, metadata: dict
|
||
) -> DataFrame:
|
||
sh, sl = self._detect_swing_points(
|
||
dataframe["high"], dataframe["low"],
|
||
self.swing_lookback_d1.value,
|
||
)
|
||
|
||
structure = self._build_structure(
|
||
dataframe["high"], dataframe["low"], dataframe["close"],
|
||
sh, sl,
|
||
)
|
||
|
||
dataframe["trend_up"] = structure["trend_up"]
|
||
dataframe["trend_down"] = structure["trend_down"]
|
||
|
||
return dataframe
|
||
|
||
# ================================================================
|
||
# 信息时间框架 — 4H 中期结构
|
||
# ================================================================
|
||
|
||
@informative("4h")
|
||
def populate_indicators_4h(
|
||
self, dataframe: DataFrame, metadata: dict
|
||
) -> DataFrame:
|
||
sh, sl = self._detect_swing_points(
|
||
dataframe["high"], dataframe["low"],
|
||
self.swing_lookback_h4.value,
|
||
)
|
||
|
||
structure = self._build_structure(
|
||
dataframe["high"], dataframe["low"], dataframe["close"],
|
||
sh, sl,
|
||
)
|
||
|
||
dataframe["trend_up"] = structure["trend_up"]
|
||
dataframe["trend_down"] = structure["trend_down"]
|
||
dataframe["support"] = structure["support"]
|
||
dataframe["resistance"] = structure["resistance"]
|
||
dataframe["in_demand"] = structure["in_demand"]
|
||
dataframe["in_supply"] = structure["in_supply"]
|
||
|
||
return dataframe
|
||
|
||
# ================================================================
|
||
# 主时间框架 — 1H K线形态
|
||
# ================================================================
|
||
|
||
def populate_indicators(
|
||
self, dataframe: DataFrame, metadata: dict
|
||
) -> DataFrame:
|
||
"""
|
||
1H 一小时线：检测 K 线形态。
|
||
"""
|
||
bullish_pin, bearish_pin, bullish_engulf, bearish_engulf = (
|
||
self._detect_candle_patterns(
|
||
dataframe["open"],
|
||
dataframe["high"],
|
||
dataframe["low"],
|
||
dataframe["close"],
|
||
self.pin_bar_wick_ratio.value,
|
||
)
|
||
)
|
||
|
||
dataframe["bullish_pinbar"] = bullish_pin
|
||
dataframe["bearish_pinbar"] = bearish_pin
|
||
dataframe["bullish_engulfing"] = bullish_engulf
|
||
dataframe["bearish_engulfing"] = bearish_engulf
|
||
|
||
dataframe["bullish_signal"] = bullish_pin | bullish_engulf
|
||
dataframe["bearish_signal"] = bearish_pin | bearish_engulf
|
||
|
||
return dataframe
|
||
|
||
# ================================================================
|
||
# 入场信号
|
||
# ================================================================
|
||
|
||
def populate_entry_trend(
|
||
self, dataframe: DataFrame, metadata: dict
|
||
) -> DataFrame:
|
||
"""
|
||
入场逻辑（1H 时间框架）。
|
||
|
||
做多条件:
|
||
1. D1 上升结构（trend_up_1d）
|
||
2. 4H 下半区 / 需求区域（in_demand_4h）
|
||
3. 1H 看涨 K 线形态（bullish_signal）
|
||
|
||
做空条件:
|
||
1. D1 下降结构（trend_down_1d）
|
||
2. 4H 上半区 / 供给区域（in_supply_4h）
|
||
3. 1H 看跌 K 线形态（bearish_signal）
|
||
"""
|
||
|
||
# ── NaN 安全处理 ──
|
||
bool_cols = [
|
||
"trend_up_1d", "trend_down_1d",
|
||
"trend_up_4h", "trend_down_4h",
|
||
"in_demand_4h", "in_supply_4h",
|
||
"bullish_signal", "bearish_signal",
|
||
]
|
||
for col in bool_cols:
|
||
if col in dataframe.columns:
|
||
dataframe[col] = dataframe[col].fillna(False).infer_objects(copy=False)
|
||
|
||
# ── 做多 ──
|
||
long_conditions = (
|
||
dataframe["trend_up_1d"]
|
||
& dataframe["in_demand_4h"]
|
||
& dataframe["bullish_signal"]
|
||
)
|
||
dataframe.loc[long_conditions, "enter_long"] = 1
|
||
|
||
# ── 做空 ──
|
||
short_conditions = (
|
||
dataframe["trend_down_1d"]
|
||
& dataframe["in_supply_4h"]
|
||
& dataframe["bearish_signal"]
|
||
)
|
||
dataframe.loc[short_conditions, "enter_short"] = 1
|
||
|
||
return dataframe
|
||
|
||
# ================================================================
|
||
# 出场信号
|
||
# ================================================================
|
||
|
||
def populate_exit_trend(
|
||
self, dataframe: DataFrame, metadata: dict
|
||
) -> DataFrame:
|
||
"""
|
||
出场逻辑 — 由结构反转触发。
|
||
"""
|
||
|
||
# 做多出场：D1 不再上升 或 4H 不再上升
|
||
exit_long = (
|
||
~dataframe["trend_up_1d"].fillna(True)
|
||
)
|
||
dataframe.loc[exit_long, "exit_long"] = 1
|
||
|
||
# 做空出场：D1 不再下降 或 4H 不再下降
|
||
exit_short = (
|
||
dataframe["trend_up_1d"].fillna(False)
|
||
)
|
||
dataframe.loc[exit_short, "exit_short"] = 1
|
||
|
||
return dataframe
|
||
|
||
# ================================================================
|
||
# 动态止损 — 基于结构失效
|
||
# ================================================================
|
||
|
||
def custom_stoploss(
|
||
self,
|
||
pair: str,
|
||
trade: Trade,
|
||
current_time: datetime,
|
||
current_rate: float,
|
||
current_profit: float,
|
||
after_fill: bool,
|
||
**kwargs,
|
||
) -> float | None:
|
||
"""
|
||
结构止损：止损位设在最近的 4H Swing Low（做多）或 Swing High（做空），
|
||
加上缓冲距离（sl_buffer_pct）。
|
||
|
||
随着行情发展，新的 Swing Point 形成，止损自动跟随。
|
||
"""
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
if dataframe is None or len(dataframe) == 0:
|
||
return None
|
||
|
||
last = dataframe.iloc[-1]
|
||
buffer = self.sl_buffer_pct.value
|
||
|
||
if trade.is_short:
|
||
resistance = last.get("resistance_4h")
|
||
if resistance is not None and not (isinstance(resistance, float) and np.isnan(resistance)):
|
||
# 做空止损：resistance × (1 + buffer)，在当前价格上方
|
||
sl_price = float(resistance) * (1 + buffer)
|
||
sl_ratio = (current_rate - sl_price) / current_rate
|
||
if sl_ratio < 0: # 止损在当前价格上方（做空方向正确）
|
||
return max(sl_ratio, -0.25) # 不超过 25%
|
||
else:
|
||
support = last.get("support_4h")
|
||
if support is not None and not (isinstance(support, float) and np.isnan(support)):
|
||
# 做多止损：support × (1 - buffer)，在当前价格下方
|
||
sl_price = float(support) * (1 - buffer)
|
||
sl_ratio = (sl_price - current_rate) / current_rate
|
||
if sl_ratio < 0: # 止损在当前价格下方（做多方向正确）
|
||
return max(sl_ratio, -0.25)
|
||
|
||
return None
|