""" Structure Flow Swing Strategy v4.2 ================================== 纯价格行为学震荡策略 — 借鉴 v2.2b 的 swing point + K线形态框架 核心逻辑（模拟手工交易）： 1. 用 swing point 识别近期高低点，自动形成交易区间 2. 价格到支撑 + K线止跌形态（bullish pinbar/engulfing）→ 做多 3. 价格到阻力 + K线滞涨 → 平多（反向开空同理） 4. 不在震荡判定上设严苛门槛，价格够到边界+形态确认就做 5. 趋势突破时，突破边界导致亏损，但可控 v4.1 系列教训（2026-06-10）： 用滚动window + 碰壁验证 + ATR比例止损 → 全是技术指标思维，不是价格行为学 唯一正收益的是最简单的版本（1H + 形态 + 固定-3%止损） """ from datetime import datetime import numpy as np import pandas as pd from pandas import DataFrame from freqtrade.strategy import IStrategy, IntParameter from freqtrade.persistence import Trade class StructureFlowSwingV42(IStrategy): can_short = True stoploss = -0.20 use_custom_stoploss = True minimal_roi = {"0": 100} max_open_trades = 1 timeframe = "1h" # ── 价格行为学参数 ── swing_window = IntParameter(5, 11, default=11, space="buy") # swing point 检测窗口 entry_zone_pct = IntParameter(2, 8, default=5, space="buy") # 入场范围（距S/R 0.5%） # 固定参数 cooldown = 6 # 冷却6根1H（6小时） # ================================================================ # Swing Point 检测（v2.2b 同款） # ================================================================ def _detect_swing_points( self, high: pd.Series, low: pd.Series, window: int = 5, ) -> tuple[pd.Series, pd.Series]: n = len(high) sh = pd.Series(np.nan, index=high.index, dtype=float) sl = pd.Series(np.nan, index=low.index, dtype=float) for i in range(window, n - window): if high.iloc[i] > high.iloc[i - window:i].max() and high.iloc[i] > high.iloc[i + 1:i + window + 1].max(): sh.iloc[i] = high.iloc[i] if low.iloc[i] < low.iloc[i - window:i].min() and low.iloc[i] < low.iloc[i + 1:i + window + 1].min(): sl.iloc[i] = low.iloc[i] return sh, sl # ================================================================ # K线形态检测（v2.2b 同款） # ================================================================ @staticmethod def _detect_pinbar(open: pd.Series, high: pd.Series, low: pd.Series, close: pd.Series) -> pd.Series: body = (open - close).abs() total_range = (high - low) upper_wick = high - open.where(open > close, close) lower_wick = open.where(open < close, close) - low bullish_pin = (lower_wick > body * 2) & (lower_wick > upper_wick * 2) & (close > open) bearish_pin = (upper_wick > body * 2) & (upper_wick > lower_wick * 2) & (close < open) return bullish_pin, bearish_pin @staticmethod def _detect_engulfing(open: pd.Series, close: pd.Series) -> tuple[pd.Series, pd.Series]: prev_open = open.shift(1) prev_close = close.shift(1) bullish_eng = (close > open) & (prev_close < prev_open) & (close > prev_open) & (open < prev_close) bearish_eng = (close < open) & (prev_close > prev_open) & (close < prev_open) & (open > prev_close) return bullish_eng, bearish_eng # ================================================================ # 主指标 # ================================================================ def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: sw = self.swing_window.value # ── Swing Points ── sh, sl = self._detect_swing_points(dataframe["high"], dataframe["low"], sw) # 向前填充最近的 swing high / low 作为动态 S/R dataframe["swing_high"] = sh.ffill() dataframe["swing_low"] = sl.ffill() # ── 区间宽度（用于止盈参考） ── dataframe["zone_width"] = np.where( dataframe["swing_high"].notna() & dataframe["swing_low"].notna() & (dataframe["swing_high"] > dataframe["swing_low"]), (dataframe["swing_high"] - dataframe["swing_low"]) / dataframe["swing_low"], np.nan, ) # ── K线形态 ── bull_pin, bear_pin = self._detect_pinbar( dataframe["open"], dataframe["high"], dataframe["low"], dataframe["close"] ) bull_eng, bear_eng = self._detect_engulfing(dataframe["open"], dataframe["close"]) dataframe["bullish_signal"] = bull_pin | bull_eng dataframe["bearish_signal"] = bear_pin | bear_eng # ── 距边界的距离 ── dataframe["dist_to_swing_low"] = np.where( dataframe["swing_low"].notna() & (dataframe["swing_low"] > 0), (dataframe["close"] - dataframe["swing_low"]) / dataframe["close"], np.nan, ) dataframe["dist_to_swing_high"] = np.where( dataframe["swing_high"].notna() & (dataframe["swing_high"] > 0), (dataframe["swing_high"] - dataframe["close"]) / dataframe["close"], np.nan, ) for col in ["dist_to_swing_low", "dist_to_swing_high"]: dataframe[col] = dataframe[col].fillna(999) return dataframe # ================================================================ # 入场 # ================================================================ def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: entry_zone = self.entry_zone_pct.value / 1000.0 # ── 做多：价格在 swing low 附近 + 止跌形态 ── long_conds = ( (dataframe["dist_to_swing_low"] < entry_zone) & (dataframe["dist_to_swing_low"] > 0) & dataframe["bullish_signal"] ) long_recent = long_conds.rolling(self.cooldown, min_periods=1).max().shift(1) == 0 dataframe.loc[long_conds & long_recent, "enter_long"] = 1 # ── 做空：价格在 swing high 附近 + 止涨形态 ── short_conds = ( (dataframe["dist_to_swing_high"] < entry_zone) & (dataframe["dist_to_swing_high"] > 0) & dataframe["bearish_signal"] ) short_recent = short_conds.rolling(self.cooldown, min_periods=1).max().shift(1) == 0 dataframe.loc[short_conds & short_recent, "enter_short"] = 1 return dataframe # ================================================================ # 出场 # ================================================================ def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe # ================================================================ # 止损：区间宽度 × 0.5（自适应） # ================================================================ def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> float: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or len(dataframe) == 0: return -0.02 if not trade.is_short else 0.02 z_width = dataframe.iloc[-1].get("zone_width", np.nan) if pd.notna(z_width) and z_width > 0.005: stop_pct = min(max(z_width * 0.5, 0.005), 0.05) else: stop_pct = 0.015 if not trade.is_short: return max((trade.open_rate * (1 - stop_pct) / current_rate) - 1.0, -0.20) else: return min(1.0 - (trade.open_rate * (1 + stop_pct) / current_rate), 0.20) # ================================================================ # 止盈：到对侧边界 + K线形态确认 → 平仓 # ================================================================ def custom_exit( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> str | None: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or len(dataframe) == 0: return None last = dataframe.iloc[-1] swing_high = last.get("swing_high", np.nan) swing_low = last.get("swing_low", np.nan) bull_sig = last.get("bullish_signal", False) bear_sig = last.get("bearish_signal", False) # ── 做多：到阻力附近 + 滞涨形态 → 平仓 ── if not trade.is_short: if pd.notna(swing_high) and swing_high > 0: near_high = current_rate >= swing_high * 0.99 if near_high and bear_sig: return "exit_signal" if near_high and current_profit > 0: return "exit_signal" # ── 做空：到支撑附近 + 止跌形态 → 平仓 ── else: if pd.notna(swing_low) and swing_low > 0: near_low = current_rate <= swing_low * 1.01 if near_low and bull_sig: return "exit_signal" if near_low and current_profit > 0: return "exit_signal" return None