okx-codex-trader/scripts/search_eth_btc_wick_rejection.py

from __future__ import annotations

import argparse
from dataclasses import dataclass
from pathlib import Path

import pandas as pd


DATA_DIR = Path("data/okx-candles")
OUTPUT_DIR = Path("reports/eth-exploration")
PREFIX = "eth-btc-wick-rejection"
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)),
)


@dataclass(frozen=True)
class Spec:
    symbol: str
    bar: str
    vol_window: int
    min_vol_rank: float
    min_upper_wick: float
    max_close_pos: float
    stop: float
    take: float
    hold: int

    @property
    def name(self) -> str:
        base = self.symbol.split("-")[0].lower()
        return (
            f"{base}-upper-wick-{self.bar}-vw{self.vol_window}-vr{self.min_vol_rank:g}"
            f"-uw{self.min_upper_wick:g}-cp{self.max_close_pos:g}"
            f"-sl{self.stop:g}-tp{self.take:g}-h{self.hold}"
        )


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 signal(frame: pd.DataFrame, spec: Spec) -> pd.Series:
    candle_range = (frame["high"] - frame["low"]).replace(0, pd.NA)
    upper_wick = (frame["high"] - frame[["open", "close"]].max(axis=1)) / candle_range
    close_pos = (frame["close"] - frame["low"]) / candle_range
    vol_rank = frame["volume"].rolling(spec.vol_window).rank(pct=True)
    weak_body = frame["close"] <= frame["open"]
    entry = (
        weak_body
        & (upper_wick >= spec.min_upper_wick)
        & (close_pos <= spec.max_close_pos)
        & (vol_rank >= spec.min_vol_rank)
    )
    return entry.fillna(False)


def trade_return(entry: float, exit_: float) -> float:
    return entry / exit_ - 1.0 - ROUNDTRIP_FEE


def run_spec(frame: pd.DataFrame, spec: Spec) -> tuple[pd.Series, list[dict[str, object]]]:
    entry = signal(frame, spec)
    warmup = spec.vol_window + 2
    equity = INITIAL_EQUITY
    position: dict[str, object] | None = None
    pending_entry = False
    trades: list[dict[str, object]] = []
    curve: list[tuple[pd.Timestamp, float]] = []
    rows = list(frame.itertuples())

    for index in range(warmup, len(rows)):
        candle = rows[index]
        ts = frame.index[index]
        if pending_entry and position is None and equity > 0:
            position = {
                "entry_time": ts,
                "entry_index": index,
                "entry_price": float(candle.open),
                "stop": float(candle.open) * (1.0 + spec.stop),
                "take": float(candle.open) * (1.0 - spec.take),
            }
            pending_entry = False

        mark = equity
        if position is not None:
            stop_hit = candle.high >= float(position["stop"])
            take_hit = candle.low <= float(position["take"])
            held = index - int(position["entry_index"])
            if stop_hit or take_hit or held >= spec.hold:
                exit_price = float(position["stop"] if stop_hit else position["take"] if take_hit else candle.close)
                net = trade_return(float(position["entry_price"]), exit_price)
                equity *= 1.0 + net
                trades.append({"entry_time": position["entry_time"], "exit_time": ts, "return": net})
                position = None
                mark = equity
            else:
                gross = float(position["entry_price"]) / candle.close - 1.0
                mark = equity * (1.0 + gross - FEE)

        curve.append((ts, mark))
        if index == len(rows) - 1 or position is not None:
            continue
        if bool(entry.iloc[index]):
            pending_entry = True

    series = pd.Series({ts: value for ts, value in curve}).sort_index()
    daily = series.resample("1D").last().ffill()
    daily = pd.concat([pd.Series([INITIAL_EQUITY], index=[daily.index[0].normalize()]), daily]).sort_index()
    return daily.groupby(level=0).last(), trades


def period_metrics(equity: pd.Series, trades: list[dict[str, object]], 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]
    scoped_trades = [trade for trade in trades if pd.Timestamp(trade["entry_time"]) >= scoped.index[0]]
    total = float(scoped.iloc[-1] / scoped.iloc[0] - 1.0)
    years = (scoped.index[-1] - scoped.index[0]).total_seconds() / 86_400 / 365
    annual = (1.0 + total) ** (1.0 / years) - 1.0 if total > -1.0 and years > 0 else 0.0
    drawdown = float(((scoped.cummax() - scoped) / scoped.cummax()).max())
    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]
    profit_factor = 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(returns) if returns else 0.0,
        "profit_factor": profit_factor,
        "trades": len(returns),
    }


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


def correlation_to_existing(candidate_monthly: pd.DataFrame) -> float | None:
    path = Path("reports/eth-exploration/eth-nextgen-micro-portfolio-monthly.csv")
    if not path.exists():
        return None
    base = pd.read_csv(path)
    base = base[base["name"] == "equal-2-c0003"][["month", "return"]].rename(columns={"return": "base_return"})
    joined = candidate_monthly[["month", "return"]].merge(base, on="month", how="inner")
    return None if len(joined) < 6 else float(joined["return"].corr(joined["base_return"]))


def row_for_spec(spec: Spec, equity: pd.Series, trades: list[dict[str, object]]) -> dict[str, object]:
    row: dict[str, object] = {"name": spec.name, "symbol": spec.symbol, "bar": spec.bar}
    for label, offset in HORIZONS:
        metrics = period_metrics(equity, trades, offset)
        for key, value in metrics.items():
            row[f"{label}_{key}"] = value
    return row


def build_specs() -> list[Spec]:
    specs: list[Spec] = []
    for symbol in ("ETH-USDT-SWAP", "BTC-USDT-SWAP"):
        for bar in ("15m", "1h", "4h"):
            holds = (12, 24) if bar == "15m" else (8, 16)
            for vol_window in (96, 192):
                for min_vol_rank in (0.90, 0.95):
                    for min_upper_wick in (0.45, 0.60):
                        for stop, take in ((0.012, 0.018), (0.02, 0.03)):
                            for hold in holds:
                                specs.append(Spec(symbol, bar, vol_window, min_vol_rank, min_upper_wick, 0.35, stop, take, hold))
    return specs


def markdown_table(frame: pd.DataFrame) -> str:
    def cell(value: object) -> str:
        return f"{value:.4f}" if isinstance(value, float) else 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_text(
    totals: pd.DataFrame,
    selected: pd.Series | None,
    corr: float | None,
    paths: list[Path],
    min_full_trades: int,
    min_3m_trades: int,
    verdict: str,
) -> str:
    keep = [
        "name",
        "symbol",
        "bar",
        "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",
        "3m_trades",
    ]
    if selected is None:
        period_table = "No reference candidate."
    else:
        period_table = markdown_table(
            pd.DataFrame(
                [
                    {
                        "period": label,
                        "total_return": selected[f"{label}_total_return"],
                        "annualized_return": selected[f"{label}_annualized_return"],
                        "max_drawdown": selected[f"{label}_max_drawdown"],
                        "win_rate": selected[f"{label}_win_rate"],
                        "profit_factor": selected[f"{label}_profit_factor"],
                        "trades": selected[f"{label}_trades"],
                    }
                    for label, _ in HORIZONS
                ]
            )
        )
    corr_text = "n/a" if corr is None else f"{corr:.4f}"
    return "\n".join(
        [
            "# ETH/BTC Wick Rejection Light Screen",
            "",
            "Scope: read-only local OKX candles; short-only single-candle upper-wick rejection; no staged entry, relative momentum, crash-follow, calendar/time bucket, trend-exhaustion, or false-breakout reversal.",
            "",
            f"Output files: {', '.join(f'`{path}`' for path in paths)}",
            f"Trade-count floor: full>={min_full_trades}, 3m>={min_3m_trades}.",
            f"Decision: {verdict}",
            f"Monthly return correlation vs `equal-2-c0003` nextgen micro portfolio: {corr_text}.",
            "",
            "## Reference Metrics",
            "",
            period_table,
            "",
            "## Top Candidates",
            "",
            markdown_table(totals[keep].head(12)),
            "",
        ]
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
    parser.add_argument("--min-full-trades", type=int, default=100)
    parser.add_argument("--min-3m-trades", type=int, default=5)
    args = parser.parse_args()

    raw = {symbol: load_frame(symbol) for symbol in ("ETH-USDT-SWAP", "BTC-USDT-SWAP")}
    frames = {(symbol, bar): resample(raw[symbol], bar) for symbol in raw for bar in ("15m", "1h", "4h")}

    rows: list[dict[str, object]] = []
    equity_by_name: dict[str, pd.Series] = {}
    for spec in build_specs():
        equity, trades = run_spec(frames[(spec.symbol, spec.bar)], spec)
        rows.append(row_for_spec(spec, equity, trades))
        equity_by_name[spec.name] = equity

    totals = pd.DataFrame(rows).sort_values(
        ["full_total_return", "1y_total_return", "full_profit_factor"],
        ascending=[False, False, False],
    )
    trade_eligible = totals[(totals["full_trades"] >= args.min_full_trades) & (totals["3m_trades"] >= args.min_3m_trades)]
    viable = trade_eligible[
        (trade_eligible["full_total_return"] > 0)
        & (trade_eligible["1y_total_return"] > 0)
        & (trade_eligible["6m_total_return"] > 0)
        & (trade_eligible["3m_total_return"] > 0)
    ]
    if not viable.empty:
        selected = viable.iloc[0]
        verdict = "Light-screen pass only: trade count and all required return windows are positive; edge still needs independent validation."
    elif not trade_eligible.empty:
        selected = trade_eligible.iloc[0]
        verdict = "Rejected: trade count is sufficient, but no trade-sufficient candidate has positive full/1y/6m/3m returns."
    else:
        selected = None
        verdict = "Rejected: no candidate met the trade-count floor."
    monthly = pd.DataFrame()
    corr = None
    if selected is not None:
        monthly = monthly_rows(str(selected["name"]), equity_by_name[str(selected["name"])])
        corr = correlation_to_existing(monthly)

    args.output_dir.mkdir(parents=True, exist_ok=True)
    totals_path = args.output_dir / f"{PREFIX}-totals.csv"
    monthly_path = args.output_dir / f"{PREFIX}-monthly.csv"
    report_path = args.output_dir / f"{PREFIX}-report.md"
    totals.to_csv(totals_path, index=False)
    monthly.to_csv(monthly_path, index=False)
    report_path.write_text(
        report_text(totals, selected, corr, [totals_path, monthly_path, report_path], args.min_full_trades, args.min_3m_trades, verdict),
        encoding="utf-8",
    )
    print(totals.head(10).to_string(index=False))
    print(f"wrote {totals_path}, {monthly_path}, {report_path}")
    return 0


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