okx-codex-trader/scripts/validate_eth_nextgen_micro_switch.py

from __future__ import annotations

import json
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
from scripts import search_eth_btc_nextgen_variants as nextgen
from scripts import search_eth_microstructure_variants as micro


REPORT_DIR = Path("reports/eth-exploration")
TARGET_NAME = "switch-l30-r96_q0.15_mf0.25_us"
COST_MODEL = "maker_taker"
ROUNDTRIP_COST_ON_MARGIN = 0.0021
NEXTGEN_LEGS = (
    "btc_trend_eth_rsi:15m:eth-btc-rsi-filter-et50-l3.0-x55.0-bt480-bm240-br0.0",
    "btc_shock_guard_eth_rsi:15m:eth-btc-shock-filter-et50-l3.0-x55.0-bt480-bm240-br0.01-sw96-sv0.01-sd0.05",
)
MICRO_NAME = "atr-compress-expand-r96-q0.15-sl0.008-tp0.016-mf0.25-us"
LOOKBACK_DAYS = 30
HORIZONS = (
    ("full", None),
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)


@dataclass(frozen=True)
class ReturnEvent:
    source: str
    exit_date: pd.Timestamp
    value: float


def daily_equity(frame: pd.DataFrame, index: pd.DatetimeIndex) -> pd.Series:
    series = frame.set_index("ts")["equity"].sort_index()
    daily = series.reindex(index.union(series.index)).sort_index().ffill().reindex(index).ffill()
    daily.iloc[0] = explore.INITIAL_EQUITY
    return daily


def events_from_nextgen(result: object) -> list[ReturnEvent]:
    events = []
    for trade in result.trades:
        net = (float(trade["return_pct"]) / 100.0 - ROUNDTRIP_COST_ON_MARGIN * float(trade.get("cost_weight", 1.0))) * 0.5
        events.append(ReturnEvent("nextgen", pd.to_datetime(str(trade["exit_time"]), utc=True).normalize(), net))
    return events


def events_from_micro(result: micro.SegmentResult) -> list[ReturnEvent]:
    events = []
    for trade in result.trades:
        net = float(trade["return_pct"]) / 100.0 - ROUNDTRIP_COST_ON_MARGIN * float(trade["cost_weight"])
        events.append(ReturnEvent("micro", pd.to_datetime(str(trade["exit_time"]), utc=True).normalize(), net))
    return events


def stats(events: list[ReturnEvent]) -> dict[str, float]:
    values = [event.value for event in events]
    wins = [value for value in values if value > 0.0]
    losses = [value for value in values if value < 0.0]
    gross_profit = sum(wins)
    gross_loss = abs(sum(losses))
    avg_win = gross_profit / len(wins) if wins else 0.0
    avg_loss = gross_loss / len(losses) if losses else 0.0
    return {
        "trades": float(len(values)),
        "win_rate": len(wins) / len(values) if values else 0.0,
        "payoff_ratio": avg_win / avg_loss if avg_loss else 0.0,
        "profit_factor": gross_profit / gross_loss if gross_loss else 0.0,
    }


def metrics(series: pd.Series) -> dict[str, float]:
    years = (series.index[-1] - series.index[0]).total_seconds() / 86_400 / 365
    total = float(series.iloc[-1] / series.iloc[0] - 1.0)
    annualized = (1.0 + total) ** (1.0 / years) - 1.0
    drawdown = explore.max_drawdown_from_equity([float(value) for value in series])
    returns = series.pct_change().dropna()
    daily_std = float(returns.std(ddof=1)) if len(returns) > 1 else 0.0
    risk_reward = float(returns.mean()) / daily_std * (365**0.5) if daily_std else 0.0
    return {
        "net_total_return": total,
        "net_annualized_return": annualized,
        "net_max_drawdown": drawdown,
        "net_calmar": annualized / drawdown if drawdown else 0.0,
        "risk_reward_ratio": risk_reward,
    }


def monthly(series: pd.Series) -> pd.DataFrame:
    month_end = series.resample("ME").last()
    frame = pd.DataFrame(
        {
            "month": month_end.index.strftime("%Y-%m"),
            "start_equity": month_end.shift(1).fillna(series.iloc[0]).to_numpy(),
            "end_equity": month_end.to_numpy(),
        }
    )
    frame["return"] = frame["end_equity"] / frame["start_equity"] - 1.0
    return frame


def horizon_rows(series: pd.Series, events: list[ReturnEvent]) -> pd.DataFrame:
    end = series.index[-1]
    month_rows = monthly(series)
    rows = []
    for label, offset in HORIZONS:
        horizon = series if offset is None else series[series.index >= end - offset]
        start = horizon.index[0]
        selected_events = [event for event in events if event.exit_date >= start]
        selected_months = month_rows[month_rows["month"] >= start.strftime("%Y-%m")]
        worst = selected_months.sort_values("return").iloc[0]
        rows.append(
            {
                "horizon": label,
                "horizon_start": start.strftime("%Y-%m-%d"),
                "horizon_end": horizon.index[-1].strftime("%Y-%m-%d"),
                "worst_month": str(worst["month"]),
                "worst_month_return": float(worst["return"]),
                **stats(selected_events),
                **metrics(horizon),
            }
        )
    return pd.DataFrame(rows)


def max_abs_diff(left: pd.Series, right: pd.Series) -> float:
    return float((left.astype(float) - right.astype(float)).abs().max())


def compare_named_columns(left: pd.DataFrame, right: pd.DataFrame, columns: list[str]) -> dict[str, float]:
    return {column: max_abs_diff(left[column], right[column]) for column in columns}


def build_replay() -> tuple[pd.Series, pd.Series, pd.Series, list[ReturnEvent], dict[str, object]]:
    published_equity = pd.read_csv(REPORT_DIR / "eth-btc-nextgen-equity.csv")
    base = published_equity[(published_equity["name"] == "equal-2-c0003") & (published_equity["cost_model"] == COST_MODEL)]
    index = pd.DatetimeIndex(pd.to_datetime(base["date"], utc=True))

    strategies = {
        f"{strategy.family}:{strategy.bar}:{strategy.candidate.name}": strategy
        for strategy in nextgen.build_strategies()
    }
    data = {
        (symbol, "15m"): nextgen.load_candles(symbol, "15m", 10.0)
        for symbol in ("ETH-USDT-SWAP", "BTC-USDT-SWAP")
    }
    leg_series = []
    nextgen_events = []
    leg_trade_counts = {}
    for key in NEXTGEN_LEGS:
        result = nextgen.run_strategy(strategies[key], data)
        leg_trade_counts[key] = result.trade_count
        leg_series.append(daily_equity(explore.cost_adjusted_trade_equity_frame(result, ROUNDTRIP_COST_ON_MARGIN), index))
        nextgen_events.extend(events_from_nextgen(result))
    nextgen_returns = pd.DataFrame([series.pct_change().fillna(0.0) for series in leg_series]).T.mean(axis=1)
    nextgen_series = explore.INITIAL_EQUITY * (1.0 + nextgen_returns).cumprod()
    nextgen_series.iloc[0] = explore.INITIAL_EQUITY

    candles = micro._load_candles(micro.SYMBOL, micro.BAR)
    requested = int(10.0 * 365 * 24 * 60 / 15)
    candles = candles[-requested:]
    variant = {variant.name: variant for variant in micro.build_variants()}[MICRO_NAME]
    micro_result = variant.run(candles)
    micro_series = daily_equity(micro.cost_equity_frame(micro_result, ROUNDTRIP_COST_ON_MARGIN), index)
    micro_events = events_from_micro(micro_result)

    nextgen_regime = nextgen_series / nextgen_series.shift(LOOKBACK_DAYS) - 1.0
    micro_regime = micro_series / micro_series.shift(LOOKBACK_DAYS) - 1.0
    active = ((nextgen_regime < 0.0) & (micro_regime > 0.0)).shift(1).fillna(False).astype(bool)
    switch_returns = nextgen_series.pct_change().fillna(0.0).where(~active, micro_series.pct_change().fillna(0.0))
    switch_series = explore.INITIAL_EQUITY * (1.0 + switch_returns).cumprod()
    switch_series.iloc[0] = explore.INITIAL_EQUITY

    selected_events = [
        event for event in nextgen_events if not bool(active.reindex([event.exit_date]).fillna(False).iloc[0])
    ]
    selected_events.extend(
        event for event in micro_events if bool(active.reindex([event.exit_date]).fillna(False).iloc[0])
    )
    meta = {
        "active_days": int(active.sum()),
        "inactive_days": int((~active).sum()),
        "nextgen_trade_counts": leg_trade_counts,
        "micro_trade_count": micro_result.trade_count,
        "selected_nextgen_events": sum(1 for event in selected_events if event.source == "nextgen"),
        "selected_micro_events": sum(1 for event in selected_events if event.source == "micro"),
    }
    return switch_series, nextgen_series, micro_series, selected_events, meta


def write_report(payload: dict[str, object]) -> str:
    checks = payload["checks"]
    passed = all(float(value) <= 1e-9 for group in checks.values() for value in group.values())
    lines = [
        "# ETH nextgen micro switch validation",
        "",
        f"Target: `{TARGET_NAME}` / `{COST_MODEL}`",
        "",
        "This is a read-only replay from raw local candles and strategy functions. It does not call OKX private APIs and does not place orders.",
        "",
        "## Result",
        "",
        f"Validation status: `{'pass' if passed else 'fail'}`",
        "",
        "## Differences",
        "",
        "```json",
        json.dumps(checks, indent=2, sort_keys=True),
        "```",
        "",
        "## Replay Metadata",
        "",
        "```json",
        json.dumps(payload["meta"], indent=2, sort_keys=True),
        "```",
        "",
        "## Freqtrade Cross-Check Boundary",
        "",
        "The available Freqtrade run is not an equivalent check for this target. Standard Freqtrade backtesting produces one executable position stream, while this target selects between two independently compounded virtual engines using prior-day equity state. The valid next comparison is to implement this exact state machine as a Freqtrade strategy or compare a generated single signal stream candle-by-candle.",
        "",
    ]
    return "\n".join(lines)


def main() -> int:
    switch_series, _, _, selected_events, meta = build_replay()
    replay_monthly = monthly(switch_series)
    replay_horizons = horizon_rows(switch_series, selected_events)
    replay_summary = {**stats(selected_events), **metrics(switch_series)}

    published_summary = pd.read_csv(REPORT_DIR / "eth-nextgen-micro-portfolio-summary.csv")
    published_summary = published_summary[(published_summary["name"] == TARGET_NAME) & (published_summary["cost_model"] == COST_MODEL)].iloc[0]
    published_equity = pd.read_csv(REPORT_DIR / "eth-nextgen-micro-portfolio-equity.csv")
    published_equity = published_equity[(published_equity["name"] == TARGET_NAME) & (published_equity["cost_model"] == COST_MODEL)].copy()
    published_monthly = pd.read_csv(REPORT_DIR / "eth-nextgen-micro-portfolio-monthly.csv")
    published_monthly = published_monthly[(published_monthly["name"] == TARGET_NAME) & (published_monthly["cost_model"] == COST_MODEL)].reset_index(drop=True)
    published_horizons = pd.read_csv(REPORT_DIR / "eth-nextgen-micro-portfolio-horizons.csv")
    published_horizons = published_horizons[(published_horizons["name"] == TARGET_NAME) & (published_horizons["cost_model"] == COST_MODEL)].reset_index(drop=True)

    replay_equity = pd.DataFrame({"date": switch_series.index.strftime("%Y-%m-%d"), "equity": switch_series.to_numpy()})
    replay_horizons = replay_horizons.reset_index(drop=True)

    numeric_summary = [
        "trades",
        "win_rate",
        "payoff_ratio",
        "profit_factor",
        "net_total_return",
        "net_annualized_return",
        "net_max_drawdown",
        "net_calmar",
        "risk_reward_ratio",
    ]
    payload = {
        "target": TARGET_NAME,
        "cost_model": COST_MODEL,
        "checks": {
            "summary": {
                column: abs(float(replay_summary[column]) - float(published_summary[column]))
                for column in numeric_summary
            },
            "equity": compare_named_columns(replay_equity, published_equity.reset_index(drop=True), ["equity"]),
            "monthly": compare_named_columns(replay_monthly, published_monthly, ["start_equity", "end_equity", "return"]),
            "horizons": compare_named_columns(
                replay_horizons,
                published_horizons,
                [
                    "worst_month_return",
                    "trades",
                    "win_rate",
                    "payoff_ratio",
                    "profit_factor",
                    "net_total_return",
                    "net_annualized_return",
                    "net_max_drawdown",
                    "net_calmar",
                    "risk_reward_ratio",
                ],
            ),
        },
        "meta": meta,
    }
    json_path = REPORT_DIR / "eth-nextgen-micro-switch-validation.json"
    md_path = REPORT_DIR / "eth-nextgen-micro-switch-validation.md"
    json_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
    md_path.write_text(write_report(payload), encoding="utf-8")
    print(md_path)
    print(json.dumps(payload["checks"], indent=2, sort_keys=True))
    return 0


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