okx-codex-trader/scripts/analyze_eth_nextgen_micro_leverage_units.py

from __future__ import annotations

import json
from pathlib import Path

import pandas as pd


REPORT_DIR = Path("reports/eth-exploration")
INITIAL_EQUITY = 10_000.0


def max_drawdown(values: list[float]) -> float:
    peak = values[0]
    drawdown = 0.0
    for value in values:
        peak = max(peak, value)
        drawdown = max(drawdown, (peak - value) / peak if peak else 0.0)
    return drawdown


def metrics(equity: pd.Series) -> dict[str, float]:
    years = (equity.index[-1] - equity.index[0]).total_seconds() / 86_400 / 365
    total = float(equity.iloc[-1] / equity.iloc[0] - 1.0)
    annualized = (1.0 + total) ** (1.0 / years) - 1.0
    drawdown = max_drawdown([float(value) for value in equity])
    returns = equity.pct_change().dropna()
    std = float(returns.std(ddof=1)) if len(returns) > 1 else 0.0
    return {
        "total_return": total,
        "annualized_return": annualized,
        "max_drawdown": drawdown,
        "calmar": annualized / drawdown if drawdown else 0.0,
        "risk_reward_ratio": float(returns.mean()) / std * (365**0.5) if std else 0.0,
    }


def trade_stats(frame: pd.DataFrame) -> dict[str, float]:
    values = frame["net_return"].astype(float)
    wins = values[values > 0.0]
    losses = values[values < 0.0]
    gross_profit = float(wins.sum())
    gross_loss = abs(float(losses.sum()))
    avg_win = gross_profit / len(wins) if len(wins) else 0.0
    avg_loss = gross_loss / len(losses) if len(losses) else 0.0
    return {
        "trades": float(len(frame)),
        "win_rate": float((values > 0.0).mean()) if len(values) else 0.0,
        "payoff_ratio": avg_win / avg_loss if avg_loss else 0.0,
        "profit_factor": gross_profit / gross_loss if gross_loss else 0.0,
    }


def equity_from_grouped_trades(trades: pd.DataFrame) -> pd.Series:
    daily_return = trades.groupby("exit_date")["net_return"].sum().sort_index()
    index = pd.date_range(daily_return.index.min(), daily_return.index.max(), freq="D", tz="UTC")
    returns = daily_return.reindex(index.strftime("%Y-%m-%d")).fillna(0.0)
    returns.index = index
    equity = INITIAL_EQUITY * (1.0 + returns).cumprod()
    equity.iloc[0] = INITIAL_EQUITY * (1.0 + returns.iloc[0])
    return equity


def markdown_table(frame: pd.DataFrame) -> str:
    rows = [list(frame.columns), ["---" for _ in frame.columns]]
    rows.extend(frame.astype(object).values.tolist())
    return "\n".join("| " + " | ".join(format_cell(value) for value in row) + " |" for row in rows)


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


def main() -> int:
    raw = pd.read_csv(REPORT_DIR / "eth-nextgen-micro-selected-trades.csv")
    raw["entry_ts"] = pd.to_datetime(raw["entry_time"], utc=True)
    raw["exit_ts"] = pd.to_datetime(raw["exit_time"], utc=True)

    group_columns = ["source", "side", "entry_time", "exit_time", "entry_price", "exit_price", "exit_date"]
    grouped = (
        raw.groupby(group_columns, as_index=False)
        .agg(
            leg_count=("leg", "count"),
            legs=("leg", lambda values: ";".join(sorted(values))),
            net_return=("net_return", "sum"),
        )
        .sort_values(["entry_time", "exit_time", "source", "side"])
    )
    grouped["position_unit"] = grouped["leg_count"].astype(float)
    grouped.loc[grouped["source"] == "nextgen", "position_unit"] *= 0.5
    grouped.loc[grouped["source"] == "micro", "position_unit"] = 1.0

    grouped_path = REPORT_DIR / "eth-nextgen-micro-leverage-units-trades.csv"
    summary_path = REPORT_DIR / "eth-nextgen-micro-leverage-units-summary.json"
    report_path = REPORT_DIR / "eth-nextgen-micro-leverage-units.md"

    grouped.to_csv(grouped_path, index=False)

    equity = equity_from_grouped_trades(grouped)
    published = pd.read_csv(REPORT_DIR / "eth-nextgen-micro-portfolio-equity.csv")
    published = published[(published["name"] == "switch-l30-r96_q0.15_mf0.25_us") & (published["cost_model"] == "maker_taker")].copy()
    published["date"] = pd.to_datetime(published["date"], utc=True)
    published_equity = published.set_index("date")["equity"].sort_index()
    aligned = pd.concat([equity.rename("leverage_unit"), published_equity.rename("published")], axis=1).dropna()

    summary = {
        "raw_virtual_trade_events": int(len(raw)),
        "netted_trades": int(len(grouped)),
        "position_unit_counts": {str(key): int(value) for key, value in grouped["position_unit"].value_counts().sort_index().items()},
        "leg_count_counts": {str(key): int(value) for key, value in grouped["leg_count"].value_counts().sort_index().items()},
        "source_counts": {str(key): int(value) for key, value in grouped["source"].value_counts().items()},
        "side_counts": {str(key): int(value) for key, value in grouped["side"].value_counts().items()},
        "grouped_trade_stats": trade_stats(grouped),
        "grouped_equity_metrics": metrics(equity),
        "published_equity_metrics": metrics(published_equity),
        "max_abs_equity_diff_vs_published": float((aligned["leverage_unit"] - aligned["published"]).abs().max()),
        "max_abs_return_diff_vs_published": float((aligned["leverage_unit"].pct_change() - aligned["published"].pct_change()).abs().max()),
    }
    summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True), encoding="utf-8")

    top = grouped[grouped["leg_count"] > 1].head(12)[
        ["source", "side", "entry_time", "exit_time", "leg_count", "position_unit", "entry_price", "exit_price", "net_return"]
    ]
    lines = [
        "# ETH nextgen micro leverage-unit interpretation",
        "",
        "This report treats simultaneous same-price same-direction virtual legs as one netted trade with a larger position unit.",
        "",
        "## Summary",
        "",
        f"- Raw virtual trade events: `{summary['raw_virtual_trade_events']}`",
        f"- Netted trades: `{summary['netted_trades']}`",
        f"- Position unit counts: `{json.dumps(summary['position_unit_counts'], sort_keys=True)}`",
        f"- Source counts: `{json.dumps(summary['source_counts'], sort_keys=True)}`",
        f"- Side counts: `{json.dumps(summary['side_counts'], sort_keys=True)}`",
        f"- Grouped total return: `{summary['grouped_equity_metrics']['total_return']:.6g}`",
        f"- Published total return: `{summary['published_equity_metrics']['total_return']:.6g}`",
        f"- Max absolute equity difference vs published: `{summary['max_abs_equity_diff_vs_published']:.6g}`",
        "",
        "## Repeated-leg examples",
        "",
        markdown_table(top),
        "",
        "## Interpretation",
        "",
        "A nextgen trade with `leg_count=2` is not two unrelated discretionary orders. It is one ETH direction with `position_unit=1.0`, because each nextgen leg contributes 0.5 unit. A nextgen trade with `leg_count=1` is `position_unit=0.5`. Micro trades are `position_unit=1.0`.",
        "",
    ]
    report_path.write_text("\n".join(lines), encoding="utf-8")
    print(report_path)
    print(json.dumps(summary, indent=2, sort_keys=True))
    return 0


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