okx-codex-trader/scripts/evaluate_expansion_portfolio.py

from __future__ import annotations

import argparse
import sys
from pathlib import Path

import pandas as pd

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

from scripts import explore_ultrashort as explore
from scripts import refine_expansion_mean_reversion_regime as mean_regime
from scripts import refine_expansion_rotation_risk as rotation_risk
from scripts.search_eth_btc_nextgen_variants import markdown_table
from scripts.search_expansion_mean_reversion import ROUNDTRIP_TAKER_COST_ON_MARGIN, daily_equity, load_base_candles, resample_candles


OUTPUT_DIR = Path("reports/strategy-expansion")
INITIAL_EQUITY = 10_000.0
HORIZONS = (
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)


def daily_from_intraday(equity: pd.Series) -> pd.Series:
    daily = equity.resample("1D").last().ffill()
    index = pd.date_range(equity.index[0].normalize(), equity.index[-1].normalize(), freq="1D", tz="UTC")
    return daily.reindex(index).ffill()


def metrics(series: pd.Series) -> dict[str, float]:
    years = (series.index[-1] - series.index[0]).total_seconds() / 86_400 / 365
    total_return = float(series.iloc[-1] / series.iloc[0] - 1.0)
    annualized_return = (1.0 + total_return) ** (1.0 / years) - 1.0 if total_return > -1.0 and years > 0.0 else 0.0
    drawdown = float((series.cummax() - series).div(series.cummax()).max())
    return {
        "total_return": total_return,
        "annualized_return": annualized_return,
        "max_drawdown": drawdown,
        "calmar": annualized_return / drawdown if drawdown else 0.0,
    }


def horizon_rows(name: str, series: pd.Series) -> list[dict[str, object]]:
    rows: list[dict[str, object]] = []
    end = series.index[-1]
    for label, offset in HORIZONS:
        horizon = series[series.index >= end - offset]
        rows.append(
            {
                "name": name,
                "horizon": label,
                "start": horizon.index[0].strftime("%Y-%m-%d"),
                "end": horizon.index[-1].strftime("%Y-%m-%d"),
                **metrics(horizon),
            }
        )
    return rows


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


def worst_rolling_months(name: str, monthly: pd.DataFrame, windows: tuple[int, ...] = (6, 12)) -> list[dict[str, object]]:
    rows: list[dict[str, object]] = []
    returns = monthly.set_index("month")["return"]
    for window in windows:
        rolled = (1.0 + returns).rolling(window).apply(lambda values: float(values.prod()), raw=True) - 1.0
        worst_month = str(rolled.idxmin())
        rows.append(
            {
                "name": name,
                "window_months": window,
                "start_month": str(returns.index[returns.index.get_loc(worst_month) - window + 1]),
                "end_month": worst_month,
                "return": float(rolled.loc[worst_month]),
            }
        )
    return rows


def rotation_risk_daily(years: float, row_index: int) -> tuple[str, pd.Series]:
    row = pd.read_csv(OUTPUT_DIR / "rotation-risk-top.csv").iloc[row_index]
    base = rotation_risk.params_from_row(row)
    params = rotation_risk.RiskParams(
        base=base,
        leverage=float(row["leverage"]),
        exposure=float(row["exposure"]),
        vol_target=float(row["vol_target"]),
    )
    frames = rotation_risk.rotation.load_symbol_bar_frames(years)
    closes = rotation_risk.rotation.aligned_closes(frames, base)
    signal_weights = rotation_risk.rotation.target_weights(closes, base)
    weights = rotation_risk.apply_risk_controls(closes, signal_weights, params)
    return params.name, daily_from_intraday(rotation_risk.equity_curve(closes, weights, params))


def mean_reversion_eq_days30_daily(years: float) -> tuple[str, pd.Series]:
    eth_15m = load_base_candles("ETH-USDT-SWAP", years)
    btc_15m = load_base_candles("BTC-USDT-SWAP", years)
    eth_4h = resample_candles(eth_15m, "4H")
    btc_4h = resample_candles(btc_15m, "4H")
    eth_daily = resample_candles(eth_15m, "1D")
    btc_daily = resample_candles(btc_15m, "1D")
    candles, btc_candles = explore.align_pair_candles(eth_4h, btc_4h)
    params = mean_regime.BaseParams()
    baseline = mean_regime.run_segment(candles, btc_candles, eth_daily, btc_daily, params, mean_regime.Regime("baseline", "baseline", ""))
    baseline_frame = explore.cost_adjusted_trade_equity_frame(baseline, ROUNDTRIP_TAKER_COST_ON_MARGIN)
    baseline_start = pd.to_datetime(baseline.equity_curve[0]["ts"], unit="ms", utc=True)
    baseline_end = pd.to_datetime(baseline.equity_curve[-1]["ts"], unit="ms", utc=True)
    baseline_daily = daily_equity(baseline_frame, baseline_start, baseline_end)
    regime = mean_regime.Regime("eq_days30", "equity_momentum", "baseline only after previous 30 closed-trade net equity days are positive")
    allowed = mean_regime.baseline_equity_regime_series(regime, candles, baseline, baseline_daily)
    result = mean_regime.run_segment(candles, btc_candles, eth_daily, btc_daily, params, regime, allowed)
    frame = explore.cost_adjusted_trade_equity_frame(result, ROUNDTRIP_TAKER_COST_ON_MARGIN)
    start = pd.to_datetime(result.equity_curve[0]["ts"], unit="ms", utc=True)
    end = pd.to_datetime(result.equity_curve[-1]["ts"], unit="ms", utc=True)
    return regime.name, daily_equity(frame, start, end)


def portfolio_series(name: str, returns: pd.DataFrame, weights: dict[str, float]) -> pd.Series:
    weighted = returns[list(weights)].mul(pd.Series(weights), axis=1).sum(axis=1)
    equity = INITIAL_EQUITY * (1.0 + weighted).cumprod()
    equity.name = name
    return equity


def report_text(paths: list[Path], totals: pd.DataFrame, correlations: pd.DataFrame, horizons: pd.DataFrame, rolling: pd.DataFrame, monthly: pd.DataFrame) -> str:
    rotation = totals[totals["name"] == "rotation-risk"].iloc[0]
    portfolios = totals[totals["kind"] == "portfolio"].sort_values(["calmar", "total_return"], ascending=[False, False])
    best = portfolios.iloc[0]
    verdict = "优于单独 rotation-risk" if float(best["calmar"]) > float(rotation["calmar"]) and float(best["max_drawdown"]) <= float(rotation["max_drawdown"]) else "不优于单独 rotation-risk"
    lines = [
        "# Expansion portfolio evaluation",
        "",
        "Run command: `rtk .venv/bin/python scripts/evaluate_expansion_portfolio.py`",
        "",
        "Output files:",
        *[f"- `{path}`" for path in paths],
        "",
        "Candidates: rotation-risk top ranked row from `rotation-risk-top.csv`; mean-reversion-regime `eq_days30`.",
        "Portfolio construction: daily return blend on common dates, no extra fees beyond each strategy equity curve.",
        "",
        "## Verdict",
        "",
        f"{verdict}. Best blend by Calmar is `{best['name']}`.",
        "",
        "## Totals",
        "",
        markdown_table(totals),
        "",
        "## Correlation",
        "",
        markdown_table(correlations),
        "",
        "## Horizons",
        "",
        markdown_table(horizons),
        "",
        "## Worst rolling windows",
        "",
        markdown_table(rolling),
        "",
        "## Monthly returns",
        "",
        markdown_table(monthly.tail(180)),
        "",
    ]
    return "\n".join(lines)


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

    rotation_name, rotation_daily = rotation_risk_daily(args.years, args.rotation_row)
    mean_name, mean_daily = mean_reversion_eq_days30_daily(args.years)
    aligned = pd.DataFrame({"rotation-risk": rotation_daily, "mean-reversion-eq_days30": mean_daily}).dropna()
    returns = aligned.pct_change().fillna(0.0)
    series = {
        "rotation-risk": INITIAL_EQUITY * (1.0 + returns["rotation-risk"]).cumprod(),
        "mean-reversion-eq_days30": INITIAL_EQUITY * (1.0 + returns["mean-reversion-eq_days30"]).cumprod(),
        "portfolio-50-50": portfolio_series("portfolio-50-50", returns, {"rotation-risk": 0.50, "mean-reversion-eq_days30": 0.50}),
        "portfolio-70-30": portfolio_series("portfolio-70-30", returns, {"rotation-risk": 0.70, "mean-reversion-eq_days30": 0.30}),
    }

    total_rows = []
    horizon_output = []
    monthly_frames = []
    rolling_rows = []
    for name, equity in series.items():
        monthly = monthly_rows(name, equity)
        total_rows.append(
            {
                "name": name,
                "kind": "portfolio" if name.startswith("portfolio") else "single",
                "source_strategy": rotation_name if name == "rotation-risk" else mean_name if name == "mean-reversion-eq_days30" else "",
                "start": equity.index[0].strftime("%Y-%m-%d"),
                "end": equity.index[-1].strftime("%Y-%m-%d"),
                "years": (equity.index[-1] - equity.index[0]).total_seconds() / 86_400 / 365,
                **metrics(equity),
            }
        )
        horizon_output.extend(horizon_rows(name, equity))
        monthly_frames.append(monthly)
        rolling_rows.extend(worst_rolling_months(name, monthly))

    correlations = pd.DataFrame(
        [
            {
                "leg_a": "rotation-risk",
                "leg_b": "mean-reversion-eq_days30",
                "common_start": aligned.index[0].strftime("%Y-%m-%d"),
                "common_end": aligned.index[-1].strftime("%Y-%m-%d"),
                "daily_return_correlation": float(returns["rotation-risk"].corr(returns["mean-reversion-eq_days30"])),
                "equity_curve_correlation": float(aligned["rotation-risk"].corr(aligned["mean-reversion-eq_days30"])),
            }
        ]
    )
    totals = pd.DataFrame(total_rows)
    horizons = pd.DataFrame(horizon_output)
    monthly = pd.concat(monthly_frames, ignore_index=True)
    rolling = pd.DataFrame(rolling_rows)

    args.output_dir.mkdir(parents=True, exist_ok=True)
    totals_path = args.output_dir / "portfolio-totals.csv"
    correlations_path = args.output_dir / "portfolio-correlations.csv"
    horizons_path = args.output_dir / "portfolio-horizons.csv"
    monthly_path = args.output_dir / "portfolio-monthly-returns.csv"
    rolling_path = args.output_dir / "portfolio-rolling-worst.csv"
    report_path = args.output_dir / "portfolio-report.md"
    paths = [totals_path, correlations_path, horizons_path, monthly_path, rolling_path, report_path]
    totals.to_csv(totals_path, index=False)
    correlations.to_csv(correlations_path, index=False)
    horizons.to_csv(horizons_path, index=False)
    monthly.to_csv(monthly_path, index=False)
    rolling.to_csv(rolling_path, index=False)
    report_path.write_text(report_text(paths, totals, correlations, horizons, rolling, monthly), encoding="utf-8")
    print(totals.to_string(index=False))
    return 0


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