okx-codex-trader/scripts/summarize_current_strategy_recent_activity.py

from __future__ import annotations

import sys
from pathlib import Path

import pandas as pd

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

from scripts.search_eth_bearish_price_proxy import (
    BTC_SYMBOL,
    SYMBOL,
    Spec,
    joined_frames,
    load_frame,
    period_metrics,
    resample,
    run_spec,
)
from scripts.search_eth_bb_squeeze_t_gates import (
    Variant as TGateVariant,
    _align_pair,
    _load_candles as load_tgate_candles,
    cost_equity_frame as tgate_cost_equity_frame,
    run_variant as run_tgate_variant,
)
from scripts.search_live_bb_squeeze_exit_variants import (
    PRIMARY_COST,
    Variant as LiveVariant,
    _load_candles as load_live_candles,
    cost_equity_frame as live_cost_equity_frame,
    run_variant as run_live_variant,
)


OUTPUT_DIR = Path("reports/eth-exploration")
SUMMARY_CSV = OUTPUT_DIR / "current-strategy-recent-activity.csv"
REPORT_MD = OUTPUT_DIR / "current-strategy-recent-activity-report.md"
WINDOWS = (30, 14, 7)


def scoped_return(frame: pd.DataFrame, days: int) -> float:
    cutoff = frame["ts"].iloc[-1] - pd.Timedelta(days=days)
    scoped = frame[frame["ts"] >= cutoff]
    if len(scoped) < 2:
        return 0.0
    return float(scoped["equity"].iloc[-1] / scoped["equity"].iloc[0] - 1.0)


def count_trades(trades: list[dict[str, object]], days: int, end: pd.Timestamp) -> int:
    cutoff = end - pd.Timedelta(days=days)
    return sum(pd.Timestamp(trade["entry_time"], tz="UTC") >= cutoff for trade in trades)


def live_rows() -> list[dict[str, object]]:
    candles = load_live_candles("ETH-USDT-SWAP", "15m")
    result = run_live_variant(candles, LiveVariant(0.0005, 1))
    frame = live_cost_equity_frame(result, dict((PRIMARY_COST, 0.0021))[PRIMARY_COST] if False else 0.0021)
    end = pd.to_datetime(candles[-1].ts, unit="ms", utc=True)
    return [
        {
            "strategy": "live_bb_squeeze_mxbuf0.0005",
            "window_days": days,
            "total_return": scoped_return(frame, days),
            "trades": count_trades(result.trades, days, end),
            "last_candle": end.isoformat(),
        }
        for days in WINDOWS
    ]


def tgate_rows() -> list[dict[str, object]]:
    eth = load_tgate_candles("ETH-USDT-SWAP", "15m")
    btc = load_tgate_candles("BTC-USDT-SWAP", "15m")
    eth, btc = _align_pair(eth, btc)
    variant = TGateVariant(96, 960, 0.25, 0.01, 0.035, "both", "btc-up", 0.006, 0.25, 24, 96, "btc_against", 0.0, 1, 0.006, 0.25, 0.008)
    result, gate_stats = run_tgate_variant(eth, btc, variant)
    frame = tgate_cost_equity_frame(result, 0.0021)
    end = pd.to_datetime(eth[-1].ts, unit="ms", utc=True)
    return [
        {
            "strategy": "bb_squeeze_t_gated_tre96",
            "window_days": days,
            "total_return": scoped_return(frame, days),
            "trades": count_trades(result.trades, days, end),
            "last_candle": end.isoformat(),
            "reentry_entries_full": gate_stats["reentry_entries"],
        }
        for days in WINDOWS
    ]


def crash_follow_rows() -> list[dict[str, object]]:
    eth = load_frame(SYMBOL)
    btc = load_frame(BTC_SYMBOL)
    frame = joined_frames(resample(eth, "1H"), resample(btc, "1H"))
    spec = Spec("crash_follow", "1H", 20, 120, 8, 0.035, 0.02, 0.06, 96, "btc_riskoff")
    equity, trades = run_spec(spec, frame)
    end = equity.index[-1]
    rows = []
    for days in WINDOWS:
        metrics = period_metrics(equity, trades, pd.DateOffset(days=days))
        rows.append(
            {
                "strategy": "crash_follow_short",
                "window_days": days,
                "total_return": metrics["total_return"],
                "max_drawdown": metrics["max_drawdown"],
                "trades": metrics["trades"],
                "profit_factor": metrics["profit_factor"],
                "win_rate": metrics["win_rate"],
                "last_candle": end.isoformat(),
            }
        )
    return 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:
    rows = live_rows() + tgate_rows() + crash_follow_rows()
    summary = pd.DataFrame(rows)
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    summary.to_csv(SUMMARY_CSV, index=False)

    display = summary.copy()
    display["total_return"] = display["total_return"].map(lambda value: f"{value:.2%}")
    if "max_drawdown" in display:
        display["max_drawdown"] = display["max_drawdown"].map(lambda value: "" if pd.isna(value) else f"{value:.2%}")
    if "profit_factor" in display:
        display["profit_factor"] = display["profit_factor"].map(lambda value: "" if pd.isna(value) else f"{value:.3f}")
    if "win_rate" in display:
        display["win_rate"] = display["win_rate"].map(lambda value: "" if pd.isna(value) else f"{value:.2%}")

    REPORT_MD.write_text(
        "# Current Strategy Recent Activity\n\n"
        "Scope: current live BB squeeze, selected T-gated BB squeeze observer, and crash-follow short observer.\n\n"
        f"{markdown_table(display)}\n",
        encoding="utf-8",
    )
    print(f"wrote {SUMMARY_CSV} and {REPORT_MD}")
    print(summary.to_string(index=False))
    return 0


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