# source: https://raw.githubusercontent.com/whitneyjohn61/freqtrade-coint-pairs-trading/a0fba66a39a4ff55658b3ff7ea159fbbd7e425f7/user_data/strategies/EnhancedCointPairsStrategy_V02.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa """ Github_whitneyjohn61_freqtrade_coint_pairs_trading__EnhancedCointPairsStrategy_V02__20260502_113041 — Candidate L Phase 1+: dual-leg coint pair @ 4h + **β-churn gate** Same as V01 (dual-leg, β stakes, z entries/exits, time stop, orphan safety), plus: **β churn filter (default ON):** Rolling **mean absolute change** in hedge_ratio (`|Δβ|`) over `beta_churn_window` bars. When churn exceeds `beta_churn_max`, skip entries. Both are **hyperoptable** (`space="buy"`) alongside entry z / OLS / zscore windows. Set `ENABLE_BETA_STAB_FILTER = False` to match V01 behavior. Optional `config["cointpairs"]`: `{"traded": "...", "anchor": "..."}` (Binance USDT futures symbols). Defaults: BTC vs ETH. `exchange.pair_whitelist` must contain exactly those two pairs. """ from __future__ import annotations import numpy as np import pandas as pd from pandas import DataFrame from datetime import datetime from freqtrade.strategy import IStrategy, merge_informative_pair from freqtrade.strategy.parameters import DecimalParameter, IntParameter from freqtrade.persistence import Trade class Github_whitneyjohn61_freqtrade_coint_pairs_trading__EnhancedCointPairsStrategy_V02__20260502_113041(IStrategy): INTERFACE_VERSION = 3 timeframe = "4h" inf_tf = "4h" minimal_roi = {"0": 100} stoploss = -0.99 use_custom_stoploss = False can_short = True ENABLE_VOL_FILTER: bool = False ENABLE_SPREAD_TRAIL: bool = False ENABLE_BETA_STAB_FILTER: bool = True startup_candle_count: int = 500 entry_zscore = DecimalParameter(1.5, 3.0, default=2.0, decimals=2, space="buy") exit_zscore = DecimalParameter(0.0, 1.0, default=0.5, decimals=2, space="sell") zscore_window = IntParameter(12, 180, default=84, space="buy") ols_window = IntParameter(120, 270, default=180, space="buy") beta_churn_window = IntParameter(8, 24, default=12, space="buy") beta_churn_max = DecimalParameter(0.003, 0.015, default=0.0085, decimals=4, space="buy") max_hold_candles = IntParameter(240, 480, default=360, space="sell") SPREAD_VOL_WINDOW: int = 6 SPREAD_VOL_PCT_LOOKBACK: int = 180 VOL_PERCENTILE: float = 0.90 TRAIL_K: float = 2.0 ORPHAN_MAX_CANDLES: int = 6 def __init__(self, config: dict) -> None: super().__init__(config) cp = config.get("cointpairs") or {} self._traded = str(cp.get("traded", "BTC/USDT:USDT")) self._anchor = str(cp.get("anchor", "ETH/USDT:USDT")) self._spread_extreme: dict[str, dict[str, float]] = {} self._had_partner: set[int] = set() def informative_pairs(self) -> list[tuple[str, str]]: return [ (self._traded, self.inf_tf), (self._anchor, self.inf_tf), ] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata["pair"] if pair == self._traded: other = self.dp.get_pair_dataframe(self._anchor, self.inf_tf) if other.empty: return self._nan_frame(dataframe) other = other[["date", "close"]].rename(columns={"close": "anchor_close"}) dataframe = merge_informative_pair( dataframe, other, self.timeframe, self.inf_tf, ffill=True, date_column="date", ) acol = f"anchor_close_{self.inf_tf}" log_y = np.log(dataframe["close"]) log_x = np.log(dataframe[acol]) else: other = self.dp.get_pair_dataframe(self._traded, self.inf_tf) if other.empty: return self._nan_frame(dataframe) other = other[["date", "close"]].rename(columns={"close": "traded_close"}) dataframe = merge_informative_pair( dataframe, other, self.timeframe, self.inf_tf, ffill=True, date_column="date", ) tcol = f"traded_close_{self.inf_tf}" log_y = np.log(dataframe[tcol]) log_x = np.log(dataframe["close"]) dataframe["hedge_ratio"] = self._rolling_hedge_ratio(log_y, log_x, int(self.ols_window.value)) dataframe["spread"] = log_y - dataframe["hedge_ratio"] * log_x zw = int(self.zscore_window.value) sm = dataframe["spread"].rolling(zw).mean() sd = dataframe["spread"].rolling(zw).std().replace(0, np.nan) dataframe["z_score"] = (dataframe["spread"] - sm) / sd bw = int(self.beta_churn_window.value) beta_churn = dataframe["hedge_ratio"].diff().abs().rolling(bw, min_periods=max(4, bw // 2)).mean() dataframe["beta_churn"] = beta_churn if self.ENABLE_BETA_STAB_FILTER: ok = beta_churn <= float(self.beta_churn_max.value) dataframe["beta_ok"] = ok.astype(np.int8) else: dataframe["beta_ok"] = 1 spread_vol = dataframe["spread"].rolling(self.SPREAD_VOL_WINDOW).std() dataframe["spread_vol"] = spread_vol vol_thr = spread_vol.rolling(self.SPREAD_VOL_PCT_LOOKBACK).quantile(self.VOL_PERCENTILE) if self.ENABLE_VOL_FILTER: dataframe["spread_vol_ok"] = (spread_vol <= vol_thr) | vol_thr.isna() else: dataframe["spread_vol_ok"] = 1 return dataframe def _nan_frame(self, dataframe: DataFrame) -> DataFrame: dataframe["z_score"] = np.nan dataframe["spread_vol_ok"] = 1 dataframe["beta_ok"] = 0 return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata["pair"] ok = dataframe["z_score"].notna() & (dataframe["volume"] > 0) if self.ENABLE_VOL_FILTER: ok = ok & (dataframe["spread_vol_ok"] == 1) if self.ENABLE_BETA_STAB_FILTER: ok = ok & (dataframe["beta_ok"] == 1) ez = float(self.entry_zscore.value) if pair == self._traded: dataframe.loc[ok & (dataframe["z_score"] < -ez), "enter_long"] = 1 dataframe.loc[ok & (dataframe["z_score"] > ez), "enter_short"] = 1 else: dataframe.loc[ok & (dataframe["z_score"] > ez), "enter_long"] = 1 dataframe.loc[ok & (dataframe["z_score"] < -ez), "enter_short"] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata["pair"] has_signal = dataframe["z_score"].notna() xz = float(self.exit_zscore.value) if pair == self._traded: dataframe.loc[has_signal & (dataframe["z_score"] > -xz), "exit_long"] = 1 dataframe.loc[has_signal & (dataframe["z_score"] < xz), "exit_short"] = 1 else: dataframe.loc[has_signal & (dataframe["z_score"] < xz), "exit_long"] = 1 dataframe.loc[has_signal & (dataframe["z_score"] > -xz), "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: open_tr = [t for t in Trade.get_open_trades() if t.pair == self._traded] open_an = [t for t in Trade.get_open_trades() if t.pair == self._anchor] if pair == self._traded and open_tr: return False if pair == self._anchor and open_an: return False if len(open_tr) + len(open_an) >= 2: return False return True def _session_key(self, trade: Trade) -> str: ts = pd.Timestamp(trade.open_date_utc).floor(str(self.timeframe)) return str(ts.value) def _is_short_spread_leg(self, trade: Trade) -> bool: if trade.pair == self._traded: return bool(trade.is_short) return not bool(trade.is_short) def _tf_seconds(self) -> int: return int(pd.Timedelta(self.timeframe).total_seconds()) def _both_pair_legs_open(self) -> bool: pairs = {t.pair for t in Trade.get_open_trades() if t.pair in (self._traded, self._anchor)} return self._traded in pairs and self._anchor in pairs def custom_exit( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> str | None: tf_sec = self._tf_seconds() trade_duration_candles = int((current_time - trade.open_date_utc).total_seconds() / tf_sec) if self._both_pair_legs_open(): self._had_partner.add(int(trade.id)) if not self._both_pair_legs_open(): if int(trade.id) not in self._had_partner: if trade_duration_candles >= self.ORPHAN_MAX_CANDLES: return "orphan_close" return None return "partner_closed" if self.ENABLE_SPREAD_TRAIL: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df.empty: return None sub = df.loc[df["date"] <= current_time] if sub.empty: return None row = sub.iloc[-1] spread = float(row["spread"]) sv = row["spread_vol"] vol = float(sv) if not pd.isna(sv) else float(sub["spread"].tail(self.SPREAD_VOL_WINDOW).std()) if np.isnan(vol) or vol <= 0: vol = 1e-12 sk = self._session_key(trade) if sk not in self._spread_extreme: self._spread_extreme[sk] = {"max": spread, "min": spread} else: self._spread_extreme[sk]["max"] = max(self._spread_extreme[sk]["max"], spread) self._spread_extreme[sk]["min"] = min(self._spread_extreme[sk]["min"], spread) ex = self._spread_extreme[sk] short_spread = self._is_short_spread_leg(trade) if short_spread: if spread < ex["max"] - self.TRAIL_K * vol: return "trail_spread" else: if spread > ex["min"] + self.TRAIL_K * vol: return "trail_spread" if trade_duration_candles >= int(self.max_hold_candles.value): return "time_stop" return None def leverage( self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str, side: str, **kwargs, ) -> float: return min(2.0, max_leverage) def custom_stake_amount( self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df.empty: return proposed_stake row = df.loc[df["date"] <= current_time] if row.empty: return proposed_stake beta = float(row["hedge_ratio"].iloc[-1]) if np.isnan(beta) or beta <= 0: return proposed_stake den = 1.0 + beta w_btc = 1.0 / den w_eth = beta / den w = w_btc if pair == self._traded else w_eth stake = proposed_stake * w if min_stake is not None: stake = max(stake, min_stake) return min(stake, max_stake) def _rolling_hedge_ratio(self, y: pd.Series, x: pd.Series, window: int) -> pd.Series: y_vals = y.values.astype(float) x_vals = x.values.astype(float) n = len(y_vals) betas = np.full(n, np.nan) for i in range(window - 1, n): y_w = y_vals[i - window + 1 : i + 1] x_w = x_vals[i - window + 1 : i + 1] if np.any(np.isnan(y_w)) or np.any(np.isnan(x_w)): continue x_mean = x_w.mean() y_mean = y_w.mean() x_demeaned = x_w - x_mean var_x = np.dot(x_demeaned, x_demeaned) if var_x < 1e-12: continue betas[i] = np.dot(x_demeaned, y_w - y_mean) / var_x return pd.Series(betas, index=y.index)