okx-codex-trader/scripts/stress_eth_bearish_price_proxy_candidate.py

from __future__ import annotations

import argparse
import sys
from pathlib import Path

import pandas as pd

sys.path.append(str(Path(__file__).resolve().parent))

from search_eth_bearish_price_proxy import (  # noqa: E402
    BTC_SYMBOL,
    DATA_DIR,
    HORIZONS,
    SYMBOL,
    Spec,
    joined_frames,
    load_frame,
    markdown_table,
    period_metrics,
    resample,
    row_for_spec,
    run_spec,
)


OUTPUT_DIR = Path("reports/eth-exploration")
BASE_SPEC = Spec("crash_follow", "1H", 20, 120, 8, 0.035, 0.02, 0.06, 96, "btc_riskoff")


def pct(value: object) -> object:
    return value if not isinstance(value, float) else f"{value:.2%}"


def horizon_table(equity: pd.Series, trades: list[dict[str, object]]) -> pd.DataFrame:
    return pd.DataFrame(
        [
            {"period": label, **period_metrics(equity, trades, offset)}
            for label, offset in HORIZONS
        ]
    )


def scoped_metrics(equity: pd.Series, trades: list[dict[str, object]], start: pd.Timestamp, end: pd.Timestamp) -> dict[str, object]:
    scoped = equity[(equity.index >= start) & (equity.index <= end)]
    scoped_trades = [
        trade
        for trade in trades
        if start <= pd.Timestamp(trade["entry_time"]).normalize() <= end
    ]
    if len(scoped) < 2:
        return {
            "total_return": 0.0,
            "max_drawdown": 0.0,
            "win_rate": 0.0,
            "profit_factor": 0.0,
            "trades": 0,
        }
    returns = [float(trade["return"]) for trade in scoped_trades]
    wins = [value for value in returns if value > 0]
    losses = [value for value in returns if value < 0]
    return {
        "total_return": float(scoped.iloc[-1] / scoped.iloc[0] - 1),
        "max_drawdown": float(((scoped.cummax() - scoped) / scoped.cummax()).max()),
        "win_rate": len(wins) / len(returns) if returns else 0.0,
        "profit_factor": sum(wins) / abs(sum(losses)) if losses else (999.0 if wins else 0.0),
        "trades": len(returns),
    }


def yearly_table(equity: pd.Series, trades: list[dict[str, object]]) -> pd.DataFrame:
    rows = []
    for year, scoped in equity.groupby(equity.index.year):
        rows.append({"year": int(year), **scoped_metrics(equity, trades, scoped.index[0], scoped.index[-1])})
    return pd.DataFrame(rows)


def monthly_table(equity: pd.Series, trades: list[dict[str, object]]) -> pd.DataFrame:
    rows = []
    months = equity.index.tz_convert(None).to_period("M")
    for month, scoped in equity.groupby(months):
        rows.append({"month": str(month), **scoped_metrics(equity, trades, scoped.index[0], scoped.index[-1])})
    return pd.DataFrame(rows)


def regime_labels(frame: pd.DataFrame) -> pd.DataFrame:
    daily_close = frame["close"].resample("1D").last().ffill()
    daily_return = daily_close.pct_change()
    trend_90d = daily_close / daily_close.shift(90) - 1
    realized_vol_30d = daily_return.rolling(30).std() * (365 ** 0.5)
    vol_bucket = pd.qcut(realized_vol_30d.dropna(), 3, labels=["low_vol", "mid_vol", "high_vol"])
    labels = pd.DataFrame(index=daily_close.index)
    labels["trend"] = "unclassified"
    labels.loc[trend_90d > 0, "trend"] = "bull"
    labels.loc[trend_90d <= 0, "trend"] = "bear"
    labels["volatility"] = "unclassified"
    labels.loc[vol_bucket.index, "volatility"] = vol_bucket.astype(str)
    return labels


def trade_segment_table(trades: list[dict[str, object]], labels: pd.DataFrame, column: str) -> pd.DataFrame:
    rows = []
    for name in sorted(labels[column].unique()):
        returns = []
        for trade in trades:
            day = pd.Timestamp(trade["entry_time"]).normalize()
            if day in labels.index and labels.at[day, column] == name:
                returns.append(float(trade["return"]))
        wins = [value for value in returns if value > 0]
        losses = [value for value in returns if value < 0]
        rows.append(
            {
                "segment": name,
                "total_return": float(pd.Series([1 + value for value in returns]).prod() - 1) if returns else 0.0,
                "win_rate": len(wins) / len(returns) if returns else 0.0,
                "profit_factor": sum(wins) / abs(sum(losses)) if losses else (999.0 if wins else 0.0),
                "trades": len(returns),
            }
        )
    return pd.DataFrame(rows)


def worst_contiguous_months(months: pd.DataFrame) -> pd.DataFrame:
    returns = months[["month", "total_return"]].reset_index(drop=True)
    worst = None
    for start in range(len(returns)):
        compounded = 1.0
        for end in range(start, len(returns)):
            compounded *= 1 + float(returns.at[end, "total_return"])
            row = {
                "start_month": returns.at[start, "month"],
                "end_month": returns.at[end, "month"],
                "months": end - start + 1,
                "total_return": compounded - 1,
            }
            if worst is None or row["total_return"] < worst["total_return"]:
                worst = row

    streaks = []
    start = None
    compounded = 1.0
    for index, row in returns.iterrows():
        value = float(row["total_return"])
        if value < 0:
            start = index if start is None else start
            compounded *= 1 + value
        elif start is not None:
            streaks.append((start, index - 1, compounded - 1))
            start = None
            compounded = 1.0
    if start is not None:
        streaks.append((start, len(returns) - 1, compounded - 1))
    longest = max(streaks, key=lambda item: (item[1] - item[0] + 1, -item[2])) if streaks else None
    rows = [{"type": "worst_any_span", **worst}] if worst else []
    if longest:
        rows.append(
            {
                "type": "longest_losing_streak",
                "start_month": returns.at[longest[0], "month"],
                "end_month": returns.at[longest[1], "month"],
                "months": longest[1] - longest[0] + 1,
                "total_return": longest[2],
            }
        )
    return pd.DataFrame(rows)


def neighbor_specs() -> list[Spec]:
    specs = []
    for threshold in (0.03, 0.035, 0.04):
        for stop in (0.015, 0.02, 0.025):
            for take in (0.05, 0.06, 0.07):
                for hold in (72, 96, 120):
                    for gate in ("btc_riskoff", "eth_riskoff", "none"):
                        specs.append(
                            Spec(
                                BASE_SPEC.family,
                                BASE_SPEC.bar,
                                BASE_SPEC.fast,
                                BASE_SPEC.slow,
                                BASE_SPEC.lookback,
                                threshold,
                                stop,
                                take,
                                hold,
                                gate,
                            )
                        )
    return specs


def stability_table(frame: pd.DataFrame) -> pd.DataFrame:
    rows = []
    for spec in neighbor_specs():
        equity, trades = run_spec(spec, frame)
        rows.append(row_for_spec(spec, equity, trades))
    return pd.DataFrame(rows).sort_values(
        ["full_total_return", "full_profit_factor", "3m_total_return"],
        ascending=[False, False, False],
    )


def summarize_stability(stability: pd.DataFrame) -> pd.DataFrame:
    return pd.DataFrame(
        [
            {
                "group": "all_neighbors",
                "count": len(stability),
                "positive_full": int((stability["full_total_return"] > 0).sum()),
                "positive_all_windows": int(
                    (
                        (stability["full_total_return"] > 0)
                        & (stability["3y_total_return"] > 0)
                        & (stability["1y_total_return"] > 0)
                        & (stability["6m_total_return"] > 0)
                        & (stability["3m_total_return"] > 0)
                    ).sum()
                ),
                "min_full_return": float(stability["full_total_return"].min()),
                "p10_full_return": float(stability["full_total_return"].quantile(0.10)),
                "median_full_return": float(stability["full_total_return"].median()),
                "min_full_pf": float(stability["full_profit_factor"].min()),
                "p10_full_pf": float(stability["full_profit_factor"].quantile(0.10)),
                "max_full_dd": float(stability["full_max_drawdown"].max()),
                "p90_full_dd": float(stability["full_max_drawdown"].quantile(0.90)),
            }
        ]
    )


def report(
    csv_path: Path,
    horizon: pd.DataFrame,
    yearly: pd.DataFrame,
    monthly: pd.DataFrame,
    trend: pd.DataFrame,
    volatility: pd.DataFrame,
    worst_months: pd.DataFrame,
    stability: pd.DataFrame,
    stability_summary: pd.DataFrame,
) -> str:
    keep = [
        "name",
        "full_total_return",
        "full_max_drawdown",
        "full_profit_factor",
        "full_trades",
        "3y_total_return",
        "1y_total_return",
        "6m_total_return",
        "3m_total_return",
    ]
    base = stability[stability["name"] == BASE_SPEC.name].iloc[0]
    verdict = "reject as an independent candidate"
    reason = "full-sample max drawdown is 44.90% and the neighborhood left tail is structurally weak"
    return (
        "# ETH Bearish Price-Proxy Candidate Stress\n\n"
        f"Candidate: `{BASE_SPEC.name}`\n\n"
        f"Scope: existing local OKX candles only under `{DATA_DIR}`; no live API and no order path.\n\n"
        f"Parameter-neighborhood CSV: `{csv_path}`\n\n"
        f"Conclusion: {verdict}; {reason}.\n\n"
        "## Full / Recent Windows\n\n"
        f"{markdown_table(horizon)}\n\n"
        "## Years\n\n"
        f"{markdown_table(yearly)}\n\n"
        "## Months\n\n"
        f"{markdown_table(monthly)}\n\n"
        "## Bull / Bear Segments\n\n"
        f"{markdown_table(trend)}\n\n"
        "## Volatility Segments\n\n"
        f"{markdown_table(volatility)}\n\n"
        "## Worst Consecutive Months\n\n"
        f"{markdown_table(worst_months)}\n\n"
        "## Parameter Neighborhood Stability\n\n"
        f"Base row: full return {base['full_total_return']:.4f}, full DD {base['full_max_drawdown']:.4f}, full PF {base['full_profit_factor']:.4f}.\n\n"
        f"{markdown_table(stability_summary)}\n\n"
        "Top neighbors:\n\n"
        f"{markdown_table(stability[keep].head(12))}\n\n"
        "Left-tail neighbors:\n\n"
        f"{markdown_table(stability[keep].tail(12))}\n"
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
    args = parser.parse_args()

    eth_15m = load_frame(SYMBOL)
    btc_15m = load_frame(BTC_SYMBOL)
    frame = joined_frames(resample(eth_15m, BASE_SPEC.bar), resample(btc_15m, BASE_SPEC.bar))

    equity, trades = run_spec(BASE_SPEC, frame)
    horizon = horizon_table(equity, trades)
    yearly = yearly_table(equity, trades)
    monthly = monthly_table(equity, trades)
    labels = regime_labels(frame)
    trend = trade_segment_table(trades, labels, "trend")
    volatility = trade_segment_table(trades, labels, "volatility")
    worst_months = worst_contiguous_months(monthly)
    stability = stability_table(frame)
    stability_summary = summarize_stability(stability)

    args.output_dir.mkdir(parents=True, exist_ok=True)
    csv_path = args.output_dir / "eth-bearish-price-proxy-candidate-stability.csv"
    report_path = args.output_dir / "eth-bearish-price-proxy-candidate-stress.md"
    stability.to_csv(csv_path, index=False)
    report_path.write_text(
        report(csv_path, horizon, yearly, monthly, trend, volatility, worst_months, stability, stability_summary),
        encoding="utf-8",
    )
    print(f"wrote {csv_path} and {report_path}")
    return 0


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