okx-codex-trader/scripts/build_freqtrade_eth_informative_skeleton.py

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())