okx-codex-trader/scripts/run_freqtrade_eth_skeleton_backtest.py

from __future__ import annotations

import argparse
import json
import os
import shlex
import shutil
import subprocess
from datetime import UTC, datetime
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
REPORT_DIR = ROOT / "reports" / "eth-exploration"
BASE_CONFIG = ROOT / "freqtrade" / "config-okx-futures.json"
STRATEGY = ROOT / "freqtrade" / "user_data" / "strategies" / "EthFocusedInformativeDry.py"
TMP_ROOT = Path("/tmp/okx-codex-trader-freqtrade-eth-skeleton")
TMP_USERDIR = TMP_ROOT / "user_data"
TMP_DATA_DIR = TMP_USERDIR / "data" / "okx" / "futures"
TMP_STRATEGY_DIR = TMP_USERDIR / "strategies"
TMP_CONFIG = TMP_ROOT / "config-eth-skeleton-okx-futures.json"
PAIR = "ETH/USDT:USDT"
EXPORTS = (
    ("ETH-USDT-SWAP", "5m"),
    ("BTC-USDT-SWAP", "5m"),
    ("ETH-USDT-SWAP", "15m"),
    ("BTC-USDT-SWAP", "15m"),
)
FREQTRADE_COMMAND_ENV = "FREQTRADE_COMMAND"


def run_command(command: list[str]) -> dict[str, Any]:
    started_at = datetime.now(UTC)
    completed = subprocess.run(
        command,
        cwd=ROOT,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    finished_at = datetime.now(UTC)
    return {
        "command": command,
        "returncode": completed.returncode,
        "started_at": started_at.isoformat(),
        "finished_at": finished_at.isoformat(),
        "stdout": completed.stdout,
        "stderr": completed.stderr,
    }


def proxy_url() -> str | None:
    for name in ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy", "ALL_PROXY", "all_proxy"):
        value = os.environ.get(name)
        if value:
            return value
    return None


def write_tmp_config() -> dict[str, Any]:
    config = json.loads(BASE_CONFIG.read_text(encoding="utf-8"))
    config["timeframe"] = "5m"
    config["bot_name"] = "okx-codex-eth-skeleton-backtest"
    config["dry_run"] = True
    config["exchange"]["pair_whitelist"] = [PAIR]

    proxy = proxy_url()
    if proxy:
        for key in ("ccxt_config", "ccxt_async_config"):
            config["exchange"].setdefault(key, {})
            config["exchange"][key]["proxies"] = {"http": proxy, "https": proxy}

    TMP_ROOT.mkdir(parents=True, exist_ok=True)
    TMP_CONFIG.write_text(json.dumps(config, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return {
        "path": str(TMP_CONFIG),
        "pair_whitelist": config["exchange"]["pair_whitelist"],
        "timeframe": config["timeframe"],
        "dry_run": config["dry_run"],
        "proxy_injected": proxy is not None,
        "proxy_source": "environment" if proxy else None,
    }


def prepare_tmp_userdir() -> dict[str, str]:
    TMP_DATA_DIR.mkdir(parents=True, exist_ok=True)
    TMP_STRATEGY_DIR.mkdir(parents=True, exist_ok=True)
    target_strategy = TMP_STRATEGY_DIR / STRATEGY.name
    shutil.copy2(STRATEGY, target_strategy)
    return {
        "userdir": str(TMP_USERDIR),
        "data_dir": str(TMP_DATA_DIR),
        "strategy": str(target_strategy),
    }


def export_data() -> list[dict[str, Any]]:
    results = []
    for symbol, bar in EXPORTS:
        results.append(
            run_command(
                [
                    "rtk",
                    "uv",
                    "run",
                    "python",
                    "scripts/export_freqtrade_data.py",
                    "--symbol",
                    symbol,
                    "--bar",
                    bar,
                    "--output-dir",
                    str(TMP_DATA_DIR),
                ]
            )
        )
    return results


def run_backtest() -> dict[str, Any]:
    freqtrade_command = shlex.split(
        os.environ.get(FREQTRADE_COMMAND_ENV, "uvx --from freqtrade freqtrade")
    )
    return run_command(
        [
            "rtk",
            *freqtrade_command,
            "backtesting",
            "--config",
            str(TMP_CONFIG),
            "--userdir",
            str(TMP_USERDIR),
            "--strategy",
            "EthFocusedInformativeDry",
            "--timeframe",
            "5m",
            "--pairs",
            PAIR,
            "--timerange",
            "20230101-",
        ]
    )


def next_steps(payload: dict[str, Any]) -> list[str]:
    backtest = payload["backtest"]
    if backtest["returncode"] == 0:
        return ["Review the Freqtrade result table and compare trade count, drawdown, and total profit with the research backtest."]
    reason = f"{backtest['stderr']}\n{backtest['stdout']}"
    if "Failed to spawn" in reason or "No such file or directory" in reason:
        return [f"Set {FREQTRADE_COMMAND_ENV} to a runnable Freqtrade command, then rerun this script."]
    if "No data found" in reason or "No history data" in reason:
        return ["Verify the four exported futures JSON files under the temporary userdir data directory and rerun backtesting."]
    if "OperationalException" in reason or "ImportError" in reason:
        return ["Fix the reported Freqtrade strategy/config error directly, then rerun the same script."]
    return ["Use the full stderr/stdout captured in the JSON report to identify the first Freqtrade failure and rerun after that specific issue is fixed."]


def render_markdown(payload: dict[str, Any]) -> str:
    export_rows = "\n".join(
        f"- `{ ' '.join(item['command']) }`: exit `{item['returncode']}`"
        for item in payload["exports"]
    )
    backtest = payload["backtest"]
    status = "succeeded" if backtest["returncode"] == 0 else "failed"
    return "\n".join(
        [
            "# Freqtrade ETH skeleton backtest attempt",
            "",
            f"- Generated at: `{payload['generated_at']}`",
            "- Scope: backtest only; no live or dry-run trading process was started.",
            "- Repo config changed: `false`; temporary config was written under `/tmp`.",
            f"- Temporary userdir: `{payload['tmp_userdir']['userdir']}`",
            f"- Temporary config: `{payload['tmp_config']['path']}`",
            f"- Proxy injected: `{payload['tmp_config']['proxy_injected']}`",
            "",
            "## Data export",
            "",
            export_rows,
            "",
            "## Backtest",
            "",
            f"- Command: `{ ' '.join(backtest['command']) }`",
            f"- Result: `{status}` with exit `{backtest['returncode']}`",
            "",
            "## Full failure output",
            "",
            "### stderr",
            "",
            "```text",
            backtest["stderr"].rstrip(),
            "```",
            "",
            "### stdout",
            "",
            "```text",
            backtest["stdout"].rstrip(),
            "```",
            "",
            "## Next step",
            "",
            *[f"- {step}" for step in payload["next_steps"]],
            "",
        ]
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--stamp", default=datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ"))
    args = parser.parse_args()

    tmp_userdir = prepare_tmp_userdir()
    tmp_config = write_tmp_config()
    exports = export_data()
    backtest = run_backtest()
    payload: dict[str, Any] = {
        "generated_at": datetime.now(UTC).isoformat(),
        "mode": "freqtrade_eth_skeleton_backtest_attempt",
        "real_trading": False,
        "repo_config_changed": False,
        "strategy": str(STRATEGY.relative_to(ROOT)),
        "tmp_userdir": tmp_userdir,
        "tmp_config": tmp_config,
        "exports": exports,
        "backtest": backtest,
    }
    payload["next_steps"] = next_steps(payload)

    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    json_path = REPORT_DIR / f"freqtrade-eth-skeleton-backtest-{args.stamp}.json"
    md_path = REPORT_DIR / f"freqtrade-eth-skeleton-backtest-{args.stamp}.md"
    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(render_markdown(payload), encoding="utf-8")
    print(md_path)
    return 0 if backtest["returncode"] == 0 else 1


if __name__ == "__main__":
    raise SystemExit(main())