from __future__ import annotations import json from datetime import UTC, datetime from pathlib import Path ROOT = Path(__file__).resolve().parents[1] STRATEGY_PATH = ROOT / "freqtrade" / "user_data" / "strategies" / "EthFocusedInformativeDry.py" REPORT_DIR = ROOT / "reports" / "eth-exploration" STRATEGY_SOURCE = '''from __future__ import annotations from datetime import datetime import pandas as pd from freqtrade.persistence import Trade from freqtrade.strategy import IStrategy class EthFocusedInformativeDry(IStrategy): INTERFACE_VERSION = 3 timeframe = "5m" can_short = False startup_candle_count = 480 process_only_new_candles = True minimal_roi = {"0": 100.0} stoploss = -0.02 use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False eth_rsi_trend_sma = 120 eth_rsi_length = 2 eth_rsi_threshold = 3.0 eth_exit_rsi = 55.0 btc_trend_sma = 480 btc_momentum_lookback = 240 btc_min_momentum = 0.0 lead_lookback_15m = 8 lead_lookback_5m = 16 btc_return_threshold_15m = 0.018 btc_return_threshold_5m = 0.012 lag_gap = 0.006 lead_lag_max_hold_bars = 8 lead_lag_stop_loss = -0.006 lead_lag_take_profit = 0.018 rsi_filter_leverage = 3.0 lead_lag_leverage = 3.0 def informative_pairs(self) -> list[tuple[str, str]]: return [ ("BTC/USDT:USDT", "5m"), ("BTC/USDT:USDT", "15m"), ("ETH/USDT:USDT", "15m"), ] def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: dataframe["eth_return_5m"] = dataframe["close"].pct_change(self.lead_lookback_5m) if self.dp: btc_5m = self.dp.get_pair_dataframe(pair="BTC/USDT:USDT", timeframe="5m") btc_5m["btc_return"] = btc_5m["close"].pct_change(self.lead_lookback_5m) dataframe = self._merge_informative(dataframe, btc_5m, "btc", "5m") btc_15m = self.dp.get_pair_dataframe(pair="BTC/USDT:USDT", timeframe="15m") btc_15m["btc_trend"] = btc_15m["close"].rolling(self.btc_trend_sma).mean() btc_15m["btc_momentum"] = btc_15m["close"].pct_change(self.btc_momentum_lookback) btc_15m["btc_return"] = btc_15m["close"].pct_change(self.lead_lookback_15m) dataframe = self._merge_informative(dataframe, btc_15m, "btc", "15m") eth_15m = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe="15m") eth_15m["eth_trend"] = eth_15m["close"].rolling(self.eth_rsi_trend_sma).mean() eth_15m["eth_rsi2"] = self._rsi(eth_15m["close"], self.eth_rsi_length) eth_15m["eth_return"] = eth_15m["close"].pct_change(self.lead_lookback_15m) dataframe = self._merge_informative(dataframe, eth_15m, "eth", "15m") return dataframe def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: rsi_filter = ( (dataframe["eth_close_15m"] > dataframe["eth_trend_15m"]) & (dataframe["eth_rsi2_15m"] <= self.eth_rsi_threshold) & (dataframe["btc_close_15m"] > dataframe["btc_trend_15m"]) & (dataframe["btc_momentum_15m"] >= self.btc_min_momentum) ) lead_lag_15m = ( (dataframe["btc_return_15m"] >= self.btc_return_threshold_15m) & ((dataframe["btc_return_15m"] - dataframe["eth_return_15m"]) >= self.lag_gap) ) lead_lag_5m = ( (dataframe["btc_return_5m"] >= self.btc_return_threshold_5m) & ((dataframe["btc_return_5m"] - dataframe["eth_return_5m"]) >= self.lag_gap) ) dataframe.loc[rsi_filter, ["enter_long", "enter_tag"]] = (1, "eth_btc_rsi_filter_15m") dataframe.loc[lead_lag_15m, ["enter_long", "enter_tag"]] = (1, "btc_lead_eth_lag_15m") dataframe.loc[lead_lag_5m, ["enter_long", "enter_tag"]] = (1, "btc_lead_eth_lag_5m") return dataframe def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: dataframe.loc[ (dataframe["eth_rsi2_15m"] >= self.eth_exit_rsi) | (dataframe["btc_close_15m"] <= dataframe["btc_trend_15m"]), ["exit_long", "exit_tag"], ] = (1, "rsi_or_btc_trend_exit") return dataframe def custom_exit( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> str | None: if trade.enter_tag not in {"btc_lead_eth_lag_15m", "btc_lead_eth_lag_5m"}: return None held_bars = int((current_time - trade.open_date_utc).total_seconds() // (5 * 60)) if current_profit <= self.lead_lag_stop_loss: return "lead_lag_stop" if current_profit >= self.lead_lag_take_profit: return "lead_lag_take_profit" if held_bars >= self.lead_lag_max_hold_bars: return "lead_lag_max_hold" return None def leverage( self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: if entry_tag in {"btc_lead_eth_lag_15m", "btc_lead_eth_lag_5m"}: return min(self.lead_lag_leverage, max_leverage) return min(self.rsi_filter_leverage, max_leverage) @staticmethod def _merge_informative( dataframe: pd.DataFrame, informative: pd.DataFrame, prefix: str, timeframe: str, ) -> pd.DataFrame: minutes = {"5m": 5, "15m": 15}[timeframe] informative = informative.copy() informative["merge_date"] = informative["date"] + pd.to_timedelta(minutes, unit="m") columns = ["merge_date", "open", "high", "low", "close", "volume"] columns += [column for column in informative.columns if column.startswith(f"{prefix}_")] informative = informative[columns].rename( columns={ column: f"{prefix}_{column}_{timeframe}" for column in columns if column != "merge_date" and not column.startswith(f"{prefix}_") } ) informative = informative.rename( columns={ column: f"{column}_{timeframe}" for column in informative.columns if column.startswith(f"{prefix}_") and not column.endswith(f"_{timeframe}") } ) merged = pd.merge_asof( dataframe.sort_values("date"), informative.sort_values("merge_date"), left_on="date", right_on="merge_date", direction="backward", ).ffill() return merged.drop(columns=[column for column in merged.columns if column.startswith("merge_date")]) @staticmethod def _rsi(close: pd.Series, length: int) -> pd.Series: deltas = close.diff() gains = deltas.clip(lower=0.0) losses = -deltas.clip(upper=0.0) values = [float("nan")] * len(close) if len(close) <= length: return pd.Series(values, index=close.index) average_gain = float(gains.iloc[1 : length + 1].mean()) average_loss = float(losses.iloc[1 : length + 1].mean()) for index in range(length, len(close)): if index > length: average_gain = ((average_gain * (length - 1)) + float(gains.iloc[index])) / length average_loss = ((average_loss * (length - 1)) + float(losses.iloc[index])) / length if pd.isna(average_gain) or pd.isna(average_loss): continue if average_loss == 0.0: values[index] = 100.0 if average_gain > 0.0 else 50.0 continue relative_strength = average_gain / average_loss values[index] = 100.0 - (100.0 / (1.0 + relative_strength)) return pd.Series(values, index=close.index) ''' def build_payload(generated_at: str) -> dict[str, object]: return { "generated_at": generated_at, "mode": "backtest_comparison_skeleton_only", "strategy": str(STRATEGY_PATH.relative_to(ROOT)), "scope": { "real_trading": False, "config_changed": False, "existing_strategy_changed": False, "base_pair": "ETH/USDT:USDT", "informative_pairs": ["BTC/USDT:USDT 5m", "BTC/USDT:USDT 15m", "ETH/USDT:USDT 15m"], "base_timeframe": "5m", }, "legs": [ { "tag": "eth_btc_rsi_filter_15m", "timeframe": "15m", "entry": "ETH close > ETH SMA120, ETH RSI2 <= 3, BTC close > BTC SMA480, BTC momentum240 >= 0", "exit": "ETH RSI2 >= 55 or BTC close <= BTC SMA480", "leverage": 3.0, }, { "tag": "btc_lead_eth_lag_15m", "timeframe": "15m", "entry": "BTC return8 >= 0.018 and BTC return8 - ETH return8 >= 0.006", "exit": "stop -0.006, take profit 0.018, or max 8 base 5m bars", "leverage": 3.0, }, { "tag": "btc_lead_eth_lag_5m", "timeframe": "5m", "entry": "BTC return16 >= 0.012 and BTC return16 - ETH return16 >= 0.006", "exit": "stop -0.006, take profit 0.018, or max 8 base 5m bars", "leverage": 3.0, }, ], "data_export_commands": [ "rtk uv run python scripts/export_freqtrade_data.py --symbol ETH-USDT-SWAP --bar 5m", "rtk uv run python scripts/export_freqtrade_data.py --symbol BTC-USDT-SWAP --bar 5m", "rtk uv run python scripts/export_freqtrade_data.py --symbol ETH-USDT-SWAP --bar 15m", "rtk uv run python scripts/export_freqtrade_data.py --symbol BTC-USDT-SWAP --bar 15m", ], "backtesting_command": ( "rtk freqtrade backtesting --config freqtrade/config-okx-futures.json " "--userdir freqtrade/user_data --strategy EthFocusedInformativeDry " "--timeframe 5m --pairs ETH/USDT:USDT --timerange 20230101-" ), "notes": [ "The skeleton is for backtest comparison only and does not modify config.", "The strategy models signal legs with entry tags on one ETH futures pair; it is not a multi-position portfolio allocator.", "The maker-dependent ETH robust TWAP leg is intentionally excluded.", ], } def build_markdown(payload: dict[str, object]) -> str: lines = [ "# Freqtrade ETH informative skeleton", "", "Purpose: backtest comparison only. No live or dry-run trading command was executed, and no config file was changed.", "", "## Generated files", "", f"- Strategy: `{payload['strategy']}`", f"- JSON report: `{payload['json_report']}`", f"- Markdown report: `{payload['markdown_report']}`", "", "## Strategy mapping", "", "| Entry tag | Timeframe | Entry | Exit |", "| --- | --- | --- | --- |", ] for leg in payload["legs"]: lines.append(f"| `{leg['tag']}` | `{leg['timeframe']}` | {leg['entry']} | {leg['exit']} |") lines.extend( [ "", "## Data export", "", "Export cached OKX candles into Freqtrade JSON futures files before backtesting:", "", ] ) for command in payload["data_export_commands"]: lines.append(f"```bash\n{command}\n```") lines.extend( [ "", "## Backtesting", "", "Run this only as a backtest comparison against exported data:", "", f"```bash\n{payload['backtesting_command']}\n```", "", "This uses the existing config path but does not require editing it. The `--pairs ETH/USDT:USDT` argument keeps the run focused on the ETH base pair while BTC is used only as informative data.", "", "## Boundaries", "", ] ) for note in payload["notes"]: lines.append(f"- {note}") lines.append("") return "\n".join(lines) def main() -> int: generated_at = datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") json_path = REPORT_DIR / f"freqtrade-eth-skeleton-{stamp}.json" md_path = REPORT_DIR / f"freqtrade-eth-skeleton-{stamp}.md" REPORT_DIR.mkdir(parents=True, exist_ok=True) STRATEGY_PATH.parent.mkdir(parents=True, exist_ok=True) STRATEGY_PATH.write_text(STRATEGY_SOURCE, encoding="utf-8") payload = build_payload(generated_at) payload["json_report"] = str(json_path.relative_to(ROOT)) payload["markdown_report"] = str(md_path.relative_to(ROOT)) json_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") md_path.write_text(build_markdown(payload), encoding="utf-8") print(STRATEGY_PATH.relative_to(ROOT)) print(json_path.relative_to(ROOT)) print(md_path.relative_to(ROOT)) return 0 if __name__ == "__main__": raise SystemExit(main())