okx-codex-trader/scripts/stress_eth_btc_calendar_carry_candidate.py

from __future__ import annotations

import sys
from pathlib import Path

import pandas as pd

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

from scripts import search_eth_btc_calendar_carry as carry


OUT_DIR = Path("reports/eth-exploration")
PREFIX = "eth-btc-calendar-carry-candidate-stress"
CANDIDATE = carry.Spec("ETH-USDT-SWAP", "1h", "long", 14, "weekend", 8, "calm")
BASE_TOTAL = OUT_DIR / "eth-focused-portfolio-conservative-total.csv"
BASE_EQUITY = OUT_DIR / "eth-focused-portfolio-conservative-equity.csv"
OVERLAY_WEIGHTS = (0.0, 0.025, 0.05, 0.075, 0.10)
HORIZONS = (("full", None), *carry.HORIZONS[1:])


def metric_row(name: str, equity: pd.Series, trades: pd.DataFrame, horizon: str, offset: pd.DateOffset | None) -> dict[str, object]:
    values = carry.metrics(equity, trades, offset)
    return {
        "name": name,
        "horizon": horizon,
        **values,
        "calmar": values["annualized_return"] / values["max_drawdown"] if values["max_drawdown"] else 0.0,
    }


def equity_metrics(name: str, equity: pd.Series, horizon: str, offset: pd.DateOffset | None) -> dict[str, object]:
    start = equity.index[0] if offset is None else equity.index[-1] - offset
    scoped = equity[equity.index >= start]
    total = float(scoped.iloc[-1] / scoped.iloc[0] - 1.0)
    years = (scoped.index[-1] - scoped.index[0]).total_seconds() / 31_536_000
    annual = (1.0 + total) ** (1.0 / years) - 1.0 if total > -1.0 and years > 0.0 else -1.0
    dd = float(((scoped.cummax() - scoped) / scoped.cummax()).max())
    return {
        "portfolio": name,
        "horizon": horizon,
        "total_return": total,
        "annualized_return": annual,
        "max_drawdown": dd,
        "calmar": annual / dd if dd else 0.0,
    }


def monthly_rows(equity: pd.Series) -> pd.DataFrame:
    month_end = equity.resample("ME").last()
    month_start = equity.resample("ME").first()
    frame = pd.DataFrame(
        {
            "month": month_end.index.strftime("%Y-%m"),
            "return": (month_end / month_start - 1.0).to_numpy(),
        }
    )
    return frame


def neighborhood_specs() -> list[carry.Spec]:
    specs: list[carry.Spec] = []
    for hour in range(12, 17):
        for hold in (2, 4, 8):
            for weekdays in ("all", "weekday", "weekend"):
                for vol_gate in ("none", "calm", "active"):
                    specs.append(carry.Spec("ETH-USDT-SWAP", "1h", "long", hour, weekdays, hold, vol_gate))
    return specs


def load_base_equity() -> tuple[str, pd.Series]:
    totals = pd.read_csv(BASE_TOTAL)
    base = totals.iloc[0]
    equity = pd.read_csv(BASE_EQUITY)
    selected = equity[
        (equity["portfolio"] == base["portfolio"])
        & (equity["cost_model"] == base["cost_model"])
        & (equity["scope"] == base["scope"])
    ].copy()
    selected["date"] = pd.to_datetime(selected["date"], utc=True)
    series = selected.set_index("date")["equity"].sort_index()
    return str(base["portfolio"]), series


def overlay_rows(base_name: str, base: pd.Series, candidate: pd.Series) -> pd.DataFrame:
    aligned = pd.DataFrame(
        {
            "base": base,
            "candidate": candidate,
        }
    ).dropna()
    base_ret = aligned["base"].pct_change().fillna(0.0)
    candidate_ret = aligned["candidate"].pct_change().fillna(0.0)
    rows: list[dict[str, object]] = []
    for weight in OVERLAY_WEIGHTS:
        returns = (1.0 - weight) * base_ret + weight * candidate_ret
        equity = 10_000.0 * (1.0 + returns).cumprod()
        name = f"{base_name}+calendar-carry-{weight:.1%}"
        for label, offset in HORIZONS:
            rows.append({"overlay_weight": weight, **equity_metrics(name, equity, label, offset)})
    return pd.DataFrame(rows)


def parameter_stability(neighborhood: pd.DataFrame) -> pd.DataFrame:
    frame = neighborhood.copy()
    frame["all_horizons_positive"] = (
        (frame["full_total_return"] > 0.0)
        & (frame["3y_total_return"] > 0.0)
        & (frame["1y_total_return"] > 0.0)
        & (frame["6m_total_return"] > 0.0)
        & (frame["3m_total_return"] > 0.0)
    )
    rows: list[dict[str, object]] = []
    for parameter in ("hour", "hold", "vol_gate", "weekdays"):
        grouped = frame.groupby(parameter)
        for value, group in grouped:
            rows.append(
                {
                    "parameter": parameter,
                    "value": value,
                    "variants": len(group),
                    "all_horizons_positive": int(group["all_horizons_positive"].sum()),
                    "positive_rate": float(group["all_horizons_positive"].mean()),
                    "avg_full_calmar": float(group["full_calmar"].mean()),
                    "median_full_return": float(group["full_total_return"].median()),
                    "median_3y_return": float(group["3y_total_return"].median()),
                    "median_3m_return": float(group["3m_total_return"].median()),
                }
            )
    return pd.DataFrame(rows)


def markdown_table(frame: pd.DataFrame) -> str:
    if len(frame) == 0:
        return ""

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

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


def report(
    candidate_metrics: pd.DataFrame,
    years: pd.DataFrame,
    stability: pd.DataFrame,
    neighborhood: pd.DataFrame,
    overlay: pd.DataFrame,
) -> str:
    focused = neighborhood[
        (neighborhood["hour"].between(13, 15))
        & (neighborhood["hold"].isin([4, 8]))
        & (neighborhood["weekdays"].isin(["weekend", "all"]))
        & (neighborhood["vol_gate"].isin(["calm", "none"]))
    ]
    stable_positive = int((focused["full_total_return"] > 0.0).sum())
    overlay_full = overlay[overlay["horizon"] == "full"].copy()
    base = overlay_full[overlay_full["overlay_weight"] == 0.0].iloc[0]
    best_overlay = overlay_full.sort_values(["calmar", "max_drawdown"], ascending=[False, True]).iloc[0]
    verdict = "reject"
    if best_overlay["overlay_weight"] > 0.0 and best_overlay["calmar"] > base["calmar"] and best_overlay["max_drawdown"] <= base["max_drawdown"]:
        verdict = "include in portfolio search"
    elif stable_positive >= max(1, len(focused) // 2):
        verdict = "keep under observation"

    top_neighborhood = neighborhood.sort_values(["full_calmar", "full_profit_factor"], ascending=[False, False]).head(12)
    overlay_summary = overlay[overlay["horizon"].isin(["full", "3y", "1y", "6m", "3m"])]
    return (
        "# ETH/BTC Calendar Carry Candidate Stress\n\n"
        "Scope: local OKX candles only; no live path. Overlay metrics use the baseline/candidate date intersection. Candidate under test: "
        f"`{CANDIDATE.name}`.\n\n"
        f"Decision: **{verdict}**.\n\n"
        "Reason: the selected rule is profitable across full/3y/1y/6m/3m and a 2.5%-10% overlay improves full-period "
        "Calmar and drawdown versus the conservative ETH portfolio baseline. The rule is not clean enough for standalone "
        "promotion because 2024 is negative and neighboring parameters are mixed; keep it as a low-weight portfolio-search leg.\n\n"
        "## Candidate Horizons\n\n"
        f"{markdown_table(candidate_metrics)}\n\n"
        "## Year Stability\n\n"
        f"{markdown_table(years)}\n\n"
        "## Parameter Neighborhood Stability\n\n"
        f"{markdown_table(stability)}\n\n"
        "## Top Neighborhood Rows\n\n"
        f"{markdown_table(top_neighborhood)}\n\n"
        "## Overlay Check\n\n"
        f"{markdown_table(overlay_summary)}\n"
    )


def main() -> int:
    frame = carry.resample(carry.load_frame("ETH-USDT-SWAP"), "1h")
    rows = []
    candidate_equity: pd.Series | None = None
    candidate_trades: pd.DataFrame | None = None

    for spec in neighborhood_specs():
        equity, trades = carry.run_spec(spec, frame)
        row = carry.row_for(spec, equity, trades)
        row["full_calmar"] = row["full_annualized_return"] / row["full_max_drawdown"] if row["full_max_drawdown"] else 0.0
        rows.append(row)
        if spec == CANDIDATE:
            candidate_equity = equity
            candidate_trades = trades

    if candidate_equity is None or candidate_trades is None:
        raise RuntimeError("candidate not found")

    candidate_metrics = pd.DataFrame([metric_row(CANDIDATE.name, candidate_equity, candidate_trades, label, offset) for label, offset in HORIZONS])
    years = carry.monthly_stability(candidate_equity)
    months = monthly_rows(candidate_equity)
    neighborhood = pd.DataFrame(rows).sort_values(["full_calmar", "full_profit_factor"], ascending=[False, False])
    stability = parameter_stability(neighborhood)
    base_name, base_equity = load_base_equity()
    overlay = overlay_rows(base_name, base_equity, candidate_equity)

    OUT_DIR.mkdir(parents=True, exist_ok=True)
    candidate_metrics.to_csv(OUT_DIR / f"{PREFIX}-horizons.csv", index=False)
    years.to_csv(OUT_DIR / f"{PREFIX}-years.csv", index=False)
    months.to_csv(OUT_DIR / f"{PREFIX}-monthly.csv", index=False)
    stability.to_csv(OUT_DIR / f"{PREFIX}-parameter-stability.csv", index=False)
    neighborhood.to_csv(OUT_DIR / f"{PREFIX}-neighborhood.csv", index=False)
    overlay.to_csv(OUT_DIR / f"{PREFIX}-overlay.csv", index=False)
    (OUT_DIR / f"{PREFIX}.md").write_text(report(candidate_metrics, years, stability, neighborhood, overlay), encoding="utf-8")
    print(f"wrote {OUT_DIR / f'{PREFIX}.md'}")
    return 0


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