okx-codex-trader/scripts/compare_eth_twap_evaluation.py

from __future__ import annotations

import importlib.util
import sys
from pathlib import Path

import pandas as pd

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from okx_codex_trader.models import Candle
from scripts import search_eth_price_twap_variants as search
from scripts.explore_ultrashort import (
    CANDLE_CACHE_DIR,
    INITIAL_EQUITY,
    LEVERAGE,
    build_rsi2_long_guarded_price_twap_candidate,
    cost_adjusted_trade_equity_frame,
    history_bars_for_years,
    load_cached_candles,
    recent_horizon_metrics_from_equity,
)

SYMBOL = "ETH-USDT-SWAP"
BAR = "15m"
MAKER_TAKER_COST = 0.0021
OUTPUT_DIR = Path("reports/eth-exploration")
CSV_PATH = OUTPUT_DIR / "eth-twap-evaluation-diff.csv"
MD_PATH = OUTPUT_DIR / "eth-twap-evaluation-diff.md"
HORIZONS = (
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)
DEEP_SPEC = {
    "trend_sma": 50,
    "rsi_threshold": 3.0,
    "exit_rsi": 55.0,
    "stop_loss_pct": 0.008,
    "max_hold_bars": 48,
    "entry_offsets": (0.002, 0.005, 0.008),
    "entry_valid_bars": 3,
    "fill_buffer": 0.0,
}


def load_report_module():
    path = Path(__file__).resolve().with_name("generate_ultrashort_report.py")
    spec = importlib.util.spec_from_file_location("generate_ultrashort_report", path)
    if spec is None or spec.loader is None:
        raise RuntimeError("cannot load generate_ultrashort_report.py")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def cached_history(years: float) -> list[Candle]:
    candles, _ = load_cached_candles(CANDLE_CACHE_DIR, SYMBOL, BAR)
    if not candles:
        raise RuntimeError(f"missing cached candles: {SYMBOL} {BAR}")
    requested_bars = history_bars_for_years(BAR, years)
    return candles[-requested_bars:] if len(candles) > requested_bars else candles


def equity_frame_from_search_result(result) -> pd.DataFrame:
    return search.equity_frame(result)


def metric_rows(label: str, candles: list[Candle], trade_count: int, gross_equity: pd.DataFrame, net_equity: pd.DataFrame) -> list[dict[str, object]]:
    gross_return = float(gross_equity["equity"].iloc[-1] / gross_equity["equity"].iloc[0] - 1.0)
    net_return = float(net_equity["equity"].iloc[-1] / net_equity["equity"].iloc[0] - 1.0)
    rows: list[dict[str, object]] = [
        {
            "evaluation": label,
            "window": "total",
            "first_candle": search._format_ts(candles[0].ts),
            "last_candle": search._format_ts(candles[-1].ts),
            "bars": len(candles),
            "trades": trade_count,
            "gross_return": gross_return,
            "maker_taker_net_return": net_return,
        }
    ]
    horizons = recent_horizon_metrics_from_equity(net_equity, candles[-1].ts, HORIZONS)
    for row in horizons.to_dict("records"):
        rows.append(
            {
                "evaluation": label,
                "window": row["horizon"],
                "first_candle": row["horizon_start"],
                "last_candle": row["horizon_end"],
                "bars": "",
                "trades": trade_count,
                "gross_return": "",
                "maker_taker_net_return": float(row["net_total_return"]),
            }
        )
    return rows


def pct(value: object) -> str:
    if value == "":
        return ""
    return f"{float(value) * 100:.2f}%"


def markdown_table(frame: pd.DataFrame) -> str:
    display = frame.copy()
    display["gross_return"] = display["gross_return"].map(pct)
    display["maker_taker_net_return"] = display["maker_taker_net_return"].map(pct)
    columns = [
        "evaluation",
        "window",
        "first_candle",
        "last_candle",
        "bars",
        "trades",
        "gross_return",
        "maker_taker_net_return",
    ]
    rows = [["" if pd.isna(value) else str(value) for value in row] for row in display[columns].itertuples(index=False, name=None)]
    return "\n".join(
        [
            "| " + " | ".join(columns) + " |",
            "| " + " | ".join("---" for _ in columns) + " |",
            *["| " + " | ".join(row) + " |" for row in rows],
        ]
    )


def main() -> int:
    report = load_report_module()

    report_candles = report.load_cached_history(report.load_explore_module(), SYMBOL, BAR, history_bars_for_years(BAR, 10.0))
    report_candidate = build_rsi2_long_guarded_price_twap_candidate(
        int(DEEP_SPEC["trend_sma"]),
        float(DEEP_SPEC["rsi_threshold"]),
        float(DEEP_SPEC["exit_rsi"]),
        float(DEEP_SPEC["stop_loss_pct"]),
        int(DEEP_SPEC["max_hold_bars"]),
        tuple(DEEP_SPEC["entry_offsets"]),
        int(DEEP_SPEC["entry_valid_bars"]),
        float(DEEP_SPEC["fill_buffer"]),
    )
    report_gross = report_candidate.run(
        candles=report_candles,
        leverage=LEVERAGE,
        warmup_bars=report_candidate.warmup_bars,
    )
    report_gross_equity = cost_adjusted_trade_equity_frame(report_gross, 0.0)
    report_net = cost_adjusted_trade_equity_frame(report_gross, MAKER_TAKER_COST)

    search_candles = cached_history(search.YEARS)
    search_gross = search.run_price_twap_segment(
        candles=search_candles,
        spec=DEEP_SPEC,
        roundtrip_cost_on_margin=0.0,
    )
    search_gross_equity = equity_frame_from_search_result(search_gross)
    search_net_result = search.run_price_twap_segment(
        candles=search_candles,
        spec=DEEP_SPEC,
        roundtrip_cost_on_margin=MAKER_TAKER_COST,
    )
    search_net = equity_frame_from_search_result(search_net_result)

    rows = [
        *metric_rows("main_report_10y_eval", report_candles, report_gross.trade_count, report_gross_equity, report_net),
        *metric_rows("search_script_3y_eval", search_candles, search_gross.trade_count, search_gross_equity, search_net),
    ]
    frame = pd.DataFrame(rows)

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    frame.to_csv(CSV_PATH, index=False)
    MD_PATH.write_text(
        "\n".join(
            [
                "# ETH RSI2 Price TWAP deep evaluation diff",
                "",
                "Parameter under test: `trend_sma=50, rsi_threshold=3.0, exit_rsi=55.0, stop_loss_pct=0.008, max_hold_bars=48, entry_offsets=(0.002, 0.005, 0.008), entry_valid_bars=3, fill_buffer=0.0`.",
                "",
                markdown_table(frame),
                "",
                "## Readout",
                "",
                "- `main_report_10y_eval` matches `scripts/generate_ultrashort_report.py`: load up to 10 years, run the shared `explore_ultrashort` candidate gross, then apply maker_taker cost with `cost_weight` per closed trade.",
                "- `search_script_3y_eval` matches `scripts/search_eth_price_twap_variants.py`: load 3 years, run its local price-TWAP segment, and deduct maker_taker cost inside each close.",
                "- The prior search grid did not include the main-report ETH deep parameter. It searched `trend_sma in {80,160}`, `rsi_threshold in {5,8,10}`, `entry_valid_bars in {2,4}`, and offset sets `(0.001,0.003,0.005)` / `(0.003,0.006,0.009)`, not `trend_sma=50`, `rsi_threshold=3`, deep offsets `(0.002,0.005,0.008)`, `entry_valid_bars=3`.",
                "- Therefore the sign mismatch is primarily parameter universe plus total-sample scope: the main report's cost table is 10-year total, while the search table was a different 3-year grid sorted by maker_taker `net_calmar` then `net_annualized_return`.",
                "",
                "Adopt the main-report evaluation口径 for reporting selected strategies: shared `explore_ultrashort` runner, 10-year cached history for total cost sensitivity, and explicit recent horizons for 3y/1y/6m/3m.",
                "",
            ]
        ),
        encoding="utf-8",
    )
    print(CSV_PATH)
    print(MD_PATH)
    print(frame.to_string(index=False))
    return 0


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