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() proxy_config_key = None if proxy: for key in ("ccxt_config", "ccxt_async_config"): config["exchange"].setdefault(key, {}) config["exchange"][key]["httpsProxy"] = proxy proxy_config_key = "httpsProxy" 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, "proxy_config_key": proxy_config_key, } 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']}`", f"- Proxy config key: `{payload['tmp_config']['proxy_config_key']}`", "", "## Data export", "", export_rows, "", "## Backtest", "", f"- Command: `{ ' '.join(backtest['command']) }`", f"- Result: `{status}` with exit `{backtest['returncode']}`", "", "## Full command 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-fix-{args.stamp}.json" md_path = REPORT_DIR / f"freqtrade-eth-skeleton-backtest-fix-{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())