okx-codex-trader/scripts/summarize_eth_recent_bearish_candidates.py

from __future__ import annotations

import json
from pathlib import Path

import pandas as pd


REPORT_DIR = Path("reports/eth-exploration")
DATA_META = Path("data/okx-candles/ETH-USDT-SWAP/15m.meta.json")
SUMMARY_CSV = REPORT_DIR / "eth-recent-bearish-candidates-summary.csv"
REPORT_MD = REPORT_DIR / "eth-recent-bearish-candidates-report.md"
HORIZONS = ("full", "3y", "1y", "6m", "3m")

CANDIDATES = (
    {
        "candidate": "crash_follow_short",
        "source": REPORT_DIR / "eth-bearish-price-proxy-totals.csv",
        "name": "crash_follow-1H-f20-s120-lb8-th0.035-sl0.02-tp0.06-h96-btc_riskoff",
        "direction": "short",
        "logic": "BTC risk-off gate + ETH downside continuation; enter short after sharp ETH drop below EMA regime.",
        "readonly_observe": "yes",
        "reason": "All requested windows are positive, but full-window MDD is high and PF is thin.",
    },
    {
        "candidate": "trend_exhaustion_short",
        "source": REPORT_DIR / "eth-bearish-failure-confirmation-totals.csv",
        "name": "trend_exhaustion-1H-f50-s240-lb8-th0.012-sl0.03-tp0.045-h72-none",
        "direction": "short",
        "logic": "Short failed relief rallies inside a falling EMA regime.",
        "readonly_observe": "yes",
        "reason": "All requested windows are positive, but 3m PF is near flat and historical drawdown remains material.",
    },
    {
        "candidate": "calendar_short_anomaly",
        "source": REPORT_DIR / "eth-btc-calendar-carry-totals.csv",
        "name": "eth-1h-short-h22-weekend-hold4-volcalm",
        "direction": "short",
        "logic": "Fixed-hold ETH weekend short bucket under calm volatility.",
        "readonly_observe": "no",
        "reason": "Positive windows exist, but full return is small relative to MDD and edge is calendar-fragile.",
    },
)


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


def metric(row: pd.Series, horizon: str, key: str) -> object:
    return row[f"{horizon}_{key}"]


def load_selected(candidate: dict[str, object]) -> pd.Series:
    frame = pd.read_csv(Path(candidate["source"]))
    selected = frame[frame["name"] == candidate["name"]]
    if len(selected) != 1:
        raise ValueError(f"missing candidate {candidate['name']} in {candidate['source']}")
    return selected.iloc[0]


def data_window() -> str:
    payload = json.loads(DATA_META.read_text(encoding="utf-8"))
    first = pd.to_datetime(int(payload["first_ts"]), unit="ms", utc=True).strftime("%Y-%m-%d %H:%M UTC")
    last = pd.to_datetime(int(payload["last_ts"]), unit="ms", utc=True).strftime("%Y-%m-%d %H:%M UTC")
    return f"{first} to {last}; rows={payload['rows']}"


def markdown_table(frame: pd.DataFrame) -> str:
    rows = [list(frame.columns), ["---" for _ in frame.columns]]
    rows.extend(frame.astype(object).where(pd.notna(frame), "").values.tolist())
    return "\n".join("| " + " | ".join(str(value).replace("|", "\\|") for value in row) + " |" for row in rows)


def main() -> int:
    summary_rows: list[dict[str, object]] = []
    decision_rows: list[dict[str, object]] = []

    for candidate in CANDIDATES:
        row = load_selected(candidate)
        for horizon in HORIZONS:
            summary_rows.append(
                {
                    "candidate": candidate["candidate"],
                    "strategy_name": candidate["name"],
                    "direction": candidate["direction"],
                    "horizon": "available_full" if horizon == "full" else horizon,
                    "total_return": metric(row, horizon, "total_return"),
                    "max_drawdown": metric(row, horizon, "max_drawdown"),
                    "trades": int(metric(row, horizon, "trades")),
                    "profit_factor": metric(row, horizon, "profit_factor"),
                    "win_rate": metric(row, horizon, "win_rate"),
                    "readonly_observe": candidate["readonly_observe"],
                }
            )
        decision_rows.append(
            {
                "candidate": candidate["candidate"],
                "direction": candidate["direction"],
                "logic": candidate["logic"],
                "readonly_observe": candidate["readonly_observe"],
                "reason": candidate["reason"],
            }
        )

    summary = pd.DataFrame(summary_rows)
    SUMMARY_CSV.parent.mkdir(parents=True, exist_ok=True)
    summary.to_csv(SUMMARY_CSV, index=False)

    display = summary.copy()
    display["total_return"] = display["total_return"].map(pct)
    display["max_drawdown"] = display["max_drawdown"].map(pct)
    display["profit_factor"] = display["profit_factor"].map(lambda value: f"{value:.3f}")
    display["win_rate"] = display["win_rate"].map(pct)

    report = (
        "# ETH Recent Bearish Candidate Review\n\n"
        f"Data window: {data_window()}.\n\n"
        "Scope: non-BB-squeeze candidates only; local candle/search outputs only; no live executor, deploy, or order path touched. "
        "The local ETH history is shorter than 10y, so `available_full` is the maximum available window.\n\n"
        "## Decision\n\n"
        f"{markdown_table(pd.DataFrame(decision_rows))}\n\n"
        "## Metrics\n\n"
        f"{markdown_table(display[['candidate', 'direction', 'horizon', 'total_return', 'max_drawdown', 'trades', 'profit_factor', 'win_rate', 'readonly_observe']])}\n\n"
        "## Source Files\n\n"
        "- `reports/eth-exploration/eth-bearish-price-proxy-totals.csv`\n"
        "- `reports/eth-exploration/eth-bearish-failure-confirmation-totals.csv`\n"
        "- `reports/eth-exploration/eth-btc-calendar-carry-totals.csv`\n"
    )
    REPORT_MD.write_text(report, encoding="utf-8")
    print(f"wrote {SUMMARY_CSV} and {REPORT_MD}")
    return 0


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