okx-codex-trader/scripts/search_eth_btc_calendar_carry.py

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

import pandas as pd


DATA_DIR = Path("data/okx-candles")
OUT_DIR = Path("reports/eth-exploration")
PREFIX = "eth-btc-calendar-carry"
SYMBOLS = ("ETH-USDT-SWAP", "BTC-USDT-SWAP")
INITIAL_EQUITY = 10_000.0
FEE = 0.0004
ROUNDTRIP_FEE = FEE * 2
HORIZONS = (
    ("full", None),
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)
WEEKDAY_SETS = {
    "all": set(range(7)),
    "weekday": set(range(5)),
    "weekend": {5, 6},
}


@dataclass(frozen=True)
class Spec:
    symbol: str
    bar: str
    side: str
    hour: int
    weekdays: str
    hold: int
    vol_gate: str

    @property
    def name(self) -> str:
        token = self.symbol.split("-")[0].lower()
        return f"{token}-{self.bar}-{self.side}-h{self.hour:02d}-{self.weekdays}-hold{self.hold}-vol{self.vol_gate}"


def load_frame(symbol: str) -> pd.DataFrame:
    frame = pd.read_csv(DATA_DIR / symbol / "15m.csv")
    frame["ts"] = pd.to_datetime(frame["ts"], unit="ms", utc=True)
    return frame.sort_values("ts").drop_duplicates("ts", keep="last").set_index("ts")


def resample(frame: pd.DataFrame, bar: str) -> pd.DataFrame:
    rule = {"15m": "15min", "1h": "1h", "4h": "4h"}[bar]
    return (
        frame.resample(rule, label="left", closed="left")
        .agg(open=("open", "first"), high=("high", "max"), low=("low", "min"), close=("close", "last"), volume=("volume", "sum"))
        .dropna()
    )


def build_specs() -> list[Spec]:
    specs: list[Spec] = []
    for symbol in SYMBOLS:
        for bar, hours, holds in (
            ("1h", range(24), (2, 4, 8)),
            ("4h", range(0, 24, 4), (1, 2, 4)),
        ):
            for side in ("short", "long"):
                for hour in hours:
                    for weekdays in WEEKDAY_SETS:
                        for hold in holds:
                            for vol_gate in ("none", "calm", "active"):
                                specs.append(Spec(symbol, bar, side, hour, weekdays, hold, vol_gate))
    return specs


def trade_return(side: str, entry: float, exit_: float) -> float:
    gross = exit_ / entry - 1.0 if side == "long" else entry / exit_ - 1.0
    return gross - ROUNDTRIP_FEE


def marked_equity(equity: float, side: str, entry: float, mark: float) -> float:
    gross = mark / entry - 1.0 if side == "long" else entry / mark - 1.0
    return equity * (1.0 + gross - FEE)


def run_spec(spec: Spec, frame: pd.DataFrame) -> tuple[pd.Series, pd.DataFrame]:
    returns = frame["close"].pct_change()
    vol = returns.rolling(96).std(ddof=0)
    vol_rank = vol.rolling(720).rank(pct=True)
    allowed_weekdays = WEEKDAY_SETS[spec.weekdays]
    entry_signal = (frame.index.hour == spec.hour) & pd.Series(frame.index.weekday, index=frame.index).isin(allowed_weekdays)
    if spec.vol_gate == "calm":
        entry_signal &= vol_rank <= 0.5
    elif spec.vol_gate == "active":
        entry_signal &= vol_rank >= 0.5

    trades: list[dict[str, object]] = []
    warmup = 800
    opens = frame["open"].to_numpy(dtype=float)
    signal_indices = [int(value) for value in entry_signal.iloc[warmup:].to_numpy().nonzero()[0] + warmup]
    last_exit_index = -1
    equity = INITIAL_EQUITY
    equity_points: list[tuple[pd.Timestamp, float]] = [(frame.index[warmup].normalize(), equity)]

    for signal_index in signal_indices:
        entry_index = signal_index + 1
        exit_index = entry_index + spec.hold
        if entry_index <= last_exit_index or exit_index >= len(frame):
            continue
        entry_price = opens[entry_index]
        exit_price = opens[exit_index]
        net = trade_return(spec.side, entry_price, exit_price)
        before = equity
        equity *= 1.0 + net
        trades.append(
            {
                "entry_time": frame.index[entry_index],
                "exit_time": frame.index[exit_index],
                "side": spec.side,
                "return": net,
                "pnl": equity - before,
            }
        )
        equity_points.append((frame.index[exit_index], equity))
        last_exit_index = exit_index

    series = pd.Series({ts: value for ts, value in equity_points}).sort_index()
    daily_index = pd.date_range(frame.index[warmup].normalize(), frame.index[-1].normalize(), freq="1D", tz="UTC")
    daily = series.reindex(daily_index.union(series.index)).sort_index().ffill().reindex(daily_index)
    return daily, pd.DataFrame(trades)


def metrics(equity: pd.Series, trades: pd.DataFrame, offset: pd.DateOffset | None) -> dict[str, float | int]:
    start = equity.index[0] if offset is None else equity.index[-1] - offset
    scoped = equity[equity.index >= start]
    scoped_trades = trades[trades["entry_time"] >= scoped.index[0]] if len(trades) else trades
    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
    drawdown = float(((scoped.cummax() - scoped) / scoped.cummax()).max())
    rets = scoped_trades["return"].astype(float).tolist() if len(scoped_trades) else []
    wins = [value for value in rets if value > 0.0]
    losses = [value for value in rets if value < 0.0]
    pf = sum(wins) / abs(sum(losses)) if losses else (999.0 if wins else 0.0)
    return {
        "total_return": total,
        "annualized_return": annual,
        "max_drawdown": drawdown,
        "win_rate": len(wins) / len(rets) if rets else 0.0,
        "profit_factor": pf,
        "trades": len(rets),
    }


def row_for(spec: Spec, equity: pd.Series, trades: pd.DataFrame) -> dict[str, object]:
    row: dict[str, object] = {
        "name": spec.name,
        "symbol": spec.symbol,
        "bar": spec.bar,
        "side": spec.side,
        "hour": spec.hour,
        "weekdays": spec.weekdays,
        "hold": spec.hold,
        "vol_gate": spec.vol_gate,
    }
    for label, offset in HORIZONS:
        for key, value in metrics(equity, trades, offset).items():
            row[f"{label}_{key}"] = value
    return row


def monthly_stability(equity: pd.Series) -> pd.DataFrame:
    month_end = equity.resample("ME").last()
    month_start = equity.resample("ME").first()
    monthly = month_end / month_start - 1.0
    rows = []
    for year, values in monthly.groupby(monthly.index.year):
        clean = values.dropna()
        if len(clean) == 0:
            continue
        rows.append(
            {
                "year": int(year),
                "months": int(len(clean)),
                "total_return": float((1.0 + clean).prod() - 1.0),
                "positive_month_rate": float((clean > 0.0).mean()),
                "worst_month": float(clean.min()),
                "best_month": float(clean.max()),
            }
        )
    return pd.DataFrame(rows)


def markdown_table(frame: pd.DataFrame) -> str:
    def cell(value: object) -> str:
        if isinstance(value, float):
            return f"{value:.4f}"
        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(totals: pd.DataFrame, selected: pd.Series, selected_metrics: pd.DataFrame, stability: pd.DataFrame) -> str:
    positive = totals[
        (totals["full_total_return"] > 0.0)
        & (totals["3y_total_return"] > 0.0)
        & (totals["1y_total_return"] > 0.0)
        & (totals["6m_total_return"] > 0.0)
        & (totals["3m_total_return"] > 0.0)
    ]
    verdict = "not worth continuing"
    if (
        selected["full_profit_factor"] >= 1.08
        and selected["full_trades"] >= 80
        and selected["3m_trades"] >= 5
        and float(stability["positive_month_rate"].mean()) >= 0.50
    ):
        verdict = "worth continuing only as a small calendar-anomaly research branch; not suitable as a short-biased or bidirectional promotion yet"

    keep = [
        "name",
        "symbol",
        "bar",
        "side",
        "hour",
        "weekdays",
        "hold",
        "vol_gate",
        "full_total_return",
        "full_annualized_return",
        "full_max_drawdown",
        "full_win_rate",
        "full_profit_factor",
        "full_trades",
        "3y_total_return",
        "1y_total_return",
        "6m_total_return",
        "3m_total_return",
    ]
    directional = (
        positive.sort_values(["calmar", "full_profit_factor"], ascending=[False, False])
        .groupby(["symbol", "side"], as_index=False)
        .head(1)
    )
    directional_table = markdown_table(directional[keep]) if len(directional) else "No all-window-positive directional candidates."
    short_table = markdown_table(positive[positive["side"] == "short"].sort_values(["calmar", "full_profit_factor"], ascending=[False, False])[keep])
    if not len(positive[positive["side"] == "short"]):
        short_table = "No short all-window-positive candidates."

    return (
        "# ETH/BTC Calendar Carry Search\n\n"
        "Scope: local OKX candles only; no live path; single-entry fixed-hold time bucket rules; no crash-follow, no ETH/BTC relative momentum, no staged entry.\n\n"
        f"Selected candidate: `{selected['name']}`.\n\n"
        f"All-window-positive candidates: {len(positive)} / {len(totals)}.\n\n"
        f"Verdict: {verdict}.\n\n"
        "## Selected Metrics\n\n"
        f"{markdown_table(selected_metrics)}\n\n"
        "## Year Stability\n\n"
        f"{markdown_table(stability)}\n\n"
        "## Directional Check\n\n"
        f"{directional_table}\n\n"
        "## Short Candidates\n\n"
        f"{short_table}\n\n"
        "## Top 10 Candidates\n\n"
        f"{markdown_table(totals.head(10)[keep])}\n"
    )


def main() -> int:
    frames = {
        (symbol, bar): resample(load_frame(symbol), bar)
        for symbol in SYMBOLS
        for bar in ("1h", "4h")
    }
    rows = []
    selected_equity: pd.Series | None = None
    selected_trades: pd.DataFrame | None = None
    selected_spec: Spec | None = None
    equities: dict[str, tuple[Spec, pd.Series, pd.DataFrame]] = {}

    specs = build_specs()
    for index, spec in enumerate(specs, 1):
        equity, trades = run_spec(spec, frames[(spec.symbol, spec.bar)])
        row = row_for(spec, equity, trades)
        rows.append(row)
        equities[spec.name] = (spec, equity, trades)
        if index % 1000 == 0:
            print(f"done {index}/{len(specs)}", flush=True)

    totals = pd.DataFrame(rows)
    totals["calmar"] = totals["full_annualized_return"] / totals["full_max_drawdown"].replace(0.0, pd.NA)
    eligible = totals[
        (totals["full_total_return"] > 0.0)
        & (totals["3y_total_return"] > 0.0)
        & (totals["1y_total_return"] > 0.0)
        & (totals["6m_total_return"] > 0.0)
        & (totals["3m_total_return"] > 0.0)
        & (totals["full_trades"] >= 50)
    ]
    ranked = (eligible if len(eligible) else totals).sort_values(
        ["calmar", "full_profit_factor", "3m_total_return", "full_trades"],
        ascending=[False, False, False, False],
    )
    totals = totals.sort_values(["calmar", "full_profit_factor", "3m_total_return"], ascending=[False, False, False])
    selected_name = str(ranked.iloc[0]["name"])
    selected_spec, selected_equity, selected_trades = equities[selected_name]
    selected = totals[totals["name"] == selected_name].iloc[0]
    selected_rows = []
    for label, offset in HORIZONS:
        row = {"period": label}
        row.update(metrics(selected_equity, selected_trades, offset))
        selected_rows.append(row)
    selected_metrics = pd.DataFrame(selected_rows)
    stability = monthly_stability(selected_equity)

    OUT_DIR.mkdir(parents=True, exist_ok=True)
    totals_path = OUT_DIR / f"{PREFIX}-totals.csv"
    stability_path = OUT_DIR / f"{PREFIX}-stability.csv"
    report_path = OUT_DIR / f"{PREFIX}-report.md"
    totals.to_csv(totals_path, index=False)
    stability.to_csv(stability_path, index=False)
    report_path.write_text(report(totals, selected, selected_metrics, stability), encoding="utf-8")
    print(selected_metrics.to_string(index=False))
    print(f"wrote {totals_path}, {stability_path}, {report_path}")
    return 0


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