okx-codex-trader/scripts/search_eth_btc_regime_variants.py

from __future__ import annotations

import argparse
import sys
from dataclasses import dataclass
from pathlib import Path

import pandas as pd

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

from scripts import explore_ultrashort as explore


COST_SCENARIOS = (
    ("maker_maker", 0.0012),
    ("maker_taker", 0.0021),
    ("taker_taker", 0.0030),
)
PRIMARY_COST = "maker_taker"
HORIZONS = (
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)


@dataclass(frozen=True)
class Strategy:
    family: str
    candidate: object
    pair: bool


def build_strategies() -> list[Strategy]:
    strategies: list[Strategy] = [
        Strategy("baseline_rsi2", explore.build_rsi2_long_guarded_candidate(50, 3.0, 55.0, 0.008, 48), False),
        Strategy("baseline_rsi2", explore.build_rsi2_long_guarded_candidate(120, 3.0, 55.0, 0.008, 48), False),
        Strategy(
            "baseline_price_twap",
            explore.build_rsi2_long_guarded_price_twap_candidate(50, 3.0, 55.0, 0.008, 48, (0.001, 0.003, 0.005), 2),
            False,
        ),
        Strategy(
            "baseline_price_twap",
            explore.build_rsi2_long_guarded_price_twap_candidate(120, 3.0, 55.0, 0.008, 48, (0.001, 0.003, 0.005), 2),
            False,
        ),
    ]

    strategies.extend(
        Strategy(
            "btc_trend_momentum_rsi2",
            explore.build_eth_btc_rsi_filter_candidate(
                eth_trend,
                eth_rsi,
                55.0,
                btc_trend,
                btc_momentum,
                btc_min_momentum,
            ),
            True,
        )
        for eth_trend in (50, 120)
        for eth_rsi in (3.0, 5.0)
        for btc_trend in (120, 240, 480)
        for btc_momentum in (96, 240)
        for btc_min_momentum in (0.0, 0.01)
    )
    strategies.extend(
        Strategy(
            "btc_shock_guard_rsi2",
            explore.build_eth_btc_shock_filter_candidate(
                50,
                3.0,
                55.0,
                480,
                240,
                btc_min_momentum,
                btc_shock_lookback,
                btc_max_realized_vol,
                btc_max_drawdown,
            ),
            True,
        )
        for btc_min_momentum in (0.0, 0.01)
        for btc_shock_lookback in (96, 240)
        for btc_max_realized_vol in (0.006, 0.01)
        for btc_max_drawdown in (0.03, 0.05)
    )
    strategies.extend(
        Strategy(
            "ethbtc_ratio_pullback",
            explore.build_eth_btc_ratio_pullback_candidate(
                480,
                btc_momentum,
                btc_min_momentum,
                ratio_length,
                ratio_std,
                5.0,
                0.008,
            ),
            True,
        )
        for btc_momentum in (96, 240)
        for btc_min_momentum in (0.0, 0.01)
        for ratio_length in (48, 96)
        for ratio_std in (1.5, 2.0)
    )
    strategies.extend(
        Strategy(
            "btc_lead_eth_lag",
            explore.build_btc_lead_eth_lag_candidate(
                lead_lookback,
                btc_return_threshold,
                lag_gap,
                max_hold_bars,
                0.006,
                take_profit_pct,
            ),
            True,
        )
        for lead_lookback in (8, 16)
        for btc_return_threshold in (0.012, 0.018)
        for lag_gap in (0.006, 0.012)
        for max_hold_bars in (8, 32)
        for take_profit_pct in (0.012, 0.018)
    )
    return strategies


def window_rows(strategy: Strategy, eth: list[explore.Candle], btc: list[explore.Candle], window_size: int) -> list[dict[str, object]]:
    if strategy.pair:
        return explore.evaluate_pair_candidate_window_rows(
            candidate=strategy.candidate,
            eth_candles=eth,
            btc_candles=btc,
            window_size=window_size,
            leverage=explore.LEVERAGE,
        )
    return explore.evaluate_candidate_window_rows(
        candidate=strategy.candidate,
        candles=eth,
        window_size=window_size,
        leverage=explore.LEVERAGE,
    )


def full_result(strategy: Strategy, eth: list[explore.Candle], btc: list[explore.Candle]) -> explore.SegmentResult:
    if strategy.pair:
        return strategy.candidate.run(
            eth_candles=eth,
            btc_candles=btc,
            leverage=explore.LEVERAGE,
            warmup_bars=strategy.candidate.warmup_bars,
        )
    return strategy.candidate.run(
        candles=eth,
        leverage=explore.LEVERAGE,
        warmup_bars=strategy.candidate.warmup_bars,
    )


def append_cost_rows(
    *,
    strategy: Strategy,
    bar: str,
    eth: list[explore.Candle],
    rows: list[dict[str, object]],
    result: explore.SegmentResult,
    summary_rows: list[dict[str, object]],
    total_rows: list[dict[str, object]],
    horizon_rows: list[dict[str, object]],
) -> None:
    for cost_name, cost_value in COST_SCENARIOS:
        summary = explore.add_cost_metrics(
            pd.DataFrame([explore.summarize_window_rows(rows, strategy.candidate.name)]),
            cost_value,
        ).iloc[0].to_dict()
        summary_rows.append(
            {
                "family": strategy.family,
                "cost": cost_name,
                "symbol": "ETH-USDT-SWAP",
                "signal_symbol": "BTC-USDT-SWAP" if strategy.pair else "",
                "bar": bar,
                "actual_bars": len(eth),
                "first_candle": explore._format_ts(eth[0].ts),
                "last_candle": explore._format_ts(eth[-1].ts),
                **summary,
            }
        )

        net_equity = explore.cost_adjusted_trade_equity_frame(result, cost_value)
        metrics = explore.annualized_metrics_from_equity(net_equity, eth[0].ts, eth[-1].ts)
        years_actual = (eth[-1].ts - eth[0].ts) / 86_400_000 / 365
        gross_annualized = (1.0 + result.total_return) ** (1.0 / years_actual) - 1.0 if result.total_return > -1.0 else 0.0
        total_rows.append(
            {
                "family": strategy.family,
                "cost": cost_name,
                "symbol": "ETH-USDT-SWAP",
                "signal_symbol": "BTC-USDT-SWAP" if strategy.pair else "",
                "bar": bar,
                "name": strategy.candidate.name,
                "first_candle": explore._format_ts(eth[0].ts),
                "last_candle": explore._format_ts(eth[-1].ts),
                "years": years_actual,
                "trades": result.trade_count,
                "gross_total_return": result.total_return,
                "gross_annualized_return": gross_annualized,
                "gross_max_drawdown_mark_to_market": result.max_drawdown,
                **metrics,
            }
        )

        horizon = explore.recent_horizon_metrics_from_equity(net_equity, eth[-1].ts, HORIZONS)
        for horizon_row in horizon.to_dict("records"):
            horizon_rows.append(
                {
                    "family": strategy.family,
                    "cost": cost_name,
                    "symbol": "ETH-USDT-SWAP",
                    "signal_symbol": "BTC-USDT-SWAP" if strategy.pair else "",
                    "bar": bar,
                    "name": strategy.candidate.name,
                    "trades": result.trade_count,
                    **horizon_row,
                }
            )


def markdown_table(frame: pd.DataFrame) -> str:
    columns = list(frame.columns)
    rows = [columns, ["---" for _ in columns]]
    for record in frame.to_dict("records"):
        rows.append([record[column] for column in columns])
    return "\n".join("| " + " | ".join(format_markdown_cell(value) for value in row) + " |" for row in rows)


def format_markdown_cell(value: object) -> str:
    if isinstance(value, float):
        return f"{value:.6g}"
    return str(value).replace("|", "\\|")


def markdown_report(
    *,
    summary: pd.DataFrame,
    total: pd.DataFrame,
    horizon: pd.DataFrame,
    output_files: list[Path],
    command: str,
) -> str:
    primary_summary = summary[summary["cost"] == PRIMARY_COST].copy()
    primary_total = total[total["cost"] == PRIMARY_COST].copy()
    top = primary_summary.head(10)
    family = (
        primary_summary.groupby("family", as_index=False)
        .agg(
            best_net_ci95_low=("net_ci95_low", "max"),
            best_net_avg_return=("net_avg_return", "max"),
            best_positive_window_rate=("positive_window_rate", "max"),
            candidate_count=("name", "count"),
        )
        .sort_values(["best_net_ci95_low", "best_net_avg_return"], ascending=False)
    )
    horizon_top = (
        horizon[horizon["cost"] == PRIMARY_COST]
        .sort_values(["horizon", "net_annualized_return"], ascending=[True, False])
        .groupby("horizon", observed=True)
        .head(3)
    )
    best = top.iloc[0].to_dict() if len(top) else {}

    lines = [
        "# ETH BTC regime variants",
        "",
        f"Run command: `{command}`",
        "",
        "Output files:",
        *[f"- `{path}`" for path in output_files],
        "",
        "Primary sort: maker_taker cost, by net_ci95_low then net_avg_return.",
        "",
        "Top 10 candidates:",
        markdown_table(top[
            [
                "family",
                "name",
                "net_avg_return",
                "net_ci95_low",
                "positive_window_rate",
                "trades",
                "avg_trades_per_window",
                "max_drawdown",
            ]
        ]),
        "",
        "Family summary:",
        markdown_table(family),
        "",
        "Recent horizon leaders:",
        markdown_table(horizon_top[
            [
                "horizon",
                "family",
                "name",
                "net_total_return",
                "net_annualized_return",
                "net_max_drawdown",
                "net_calmar",
            ]
        ]),
        "",
        "Interpretation:",
        f"- Effective: BTC trend plus momentum gating on ETH RSI2. Best maker_taker window result is `{best.get('name', '')}` with net_ci95_low {format_markdown_cell(best.get('net_ci95_low', 0.0))} and net_avg_return {format_markdown_cell(best.get('net_avg_return', 0.0))}.",
        "- Effective but not incremental: loose BTC shock guards tie the best trend/momentum result, so the tested vol/drawdown caps mostly did not bind.",
        "- Not robust: BTC lead ETH lag has positive best net_avg_return but negative best net_ci95_low, so the average is not enough to promote it.",
        "- Not effective: ETHBTC ratio low pullback variants are negative on both best net_avg_return and best net_ci95_low.",
        "- Baseline note: ETH price-TWAP has strong recent horizon returns, but its sampled-window maker_taker net_ci95_low is deeply negative; it is not a robust regime condition in this run.",
    ]
    if len(primary_total):
        total_top = primary_total.sort_values(["net_calmar", "net_annualized_return"], ascending=False).head(5)
        lines.extend(["", "Best full-period net Calmar:", markdown_table(total_top[["family", "name", "trades", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar"]])])
    return "\n".join(lines) + "\n"


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--bar", default="15m")
    parser.add_argument("--years", type=float, default=3.25)
    parser.add_argument("--window-size", type=int, default=explore.WINDOW_SIZE)
    parser.add_argument("--output-dir", type=Path, default=Path("reports/eth-exploration"))
    args = parser.parse_args()

    requested_bars = explore.history_bars_for_years(args.bar, args.years)
    client = explore.OkxClient()
    eth = explore.get_candles_cached(client, "ETH-USDT-SWAP", args.bar, requested_bars)
    btc = explore.get_candles_cached(client, "BTC-USDT-SWAP", args.bar, requested_bars)
    eth, btc = explore.align_pair_candles(eth, btc)

    strategies = build_strategies()
    summary_rows: list[dict[str, object]] = []
    total_rows: list[dict[str, object]] = []
    horizon_rows: list[dict[str, object]] = []
    for index, strategy in enumerate(strategies, start=1):
        rows = window_rows(strategy, eth, btc, args.window_size)
        result = full_result(strategy, eth, btc)
        append_cost_rows(
            strategy=strategy,
            bar=args.bar,
            eth=eth,
            rows=rows,
            result=result,
            summary_rows=summary_rows,
            total_rows=total_rows,
            horizon_rows=horizon_rows,
        )
        print(f"done {index}/{len(strategies)} {strategy.family} {strategy.candidate.name}")

    summary = pd.DataFrame(summary_rows).sort_values(
        ["cost", "net_ci95_low", "net_avg_return"],
        ascending=[True, False, False],
    )
    primary = summary[summary["cost"] == PRIMARY_COST]
    others = summary[summary["cost"] != PRIMARY_COST]
    summary = pd.concat([primary, others], ignore_index=True)

    total = pd.DataFrame(total_rows).sort_values(["cost", "net_calmar", "net_annualized_return"], ascending=[True, False, False])
    horizon = pd.DataFrame(horizon_rows)
    horizon["horizon"] = pd.Categorical(horizon["horizon"], categories=["3y", "1y", "6m", "3m"], ordered=True)
    horizon = horizon.sort_values(["cost", "horizon", "net_annualized_return"], ascending=[True, True, False])

    args.output_dir.mkdir(parents=True, exist_ok=True)
    summary_path = args.output_dir / "eth-btc-regime-summary.csv"
    total_path = args.output_dir / "eth-btc-regime-total.csv"
    horizon_path = args.output_dir / "eth-btc-regime-horizon.csv"
    top10_path = args.output_dir / "eth-btc-regime-top10.csv"
    report_path = args.output_dir / "eth-btc-regime-report.md"

    summary.to_csv(summary_path, index=False)
    total.to_csv(total_path, index=False)
    horizon.to_csv(horizon_path, index=False)
    summary[summary["cost"] == PRIMARY_COST].head(10).to_csv(top10_path, index=False)
    command = f"rtk .venv/bin/python {Path(__file__).as_posix()} --bar {args.bar} --years {args.years} --window-size {args.window_size}"
    report_path.write_text(
        markdown_report(
            summary=summary,
            total=total,
            horizon=horizon,
            output_files=[summary_path, total_path, horizon_path, top10_path, report_path],
            command=command,
        ),
        encoding="utf-8",
    )

    print(summary[summary["cost"] == PRIMARY_COST].head(10).to_string(index=False))
    return 0


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