okx-codex-trader/scripts/validate_calendar_carry_fusion_leg.py

from __future__ import annotations

import argparse
import sys
from pathlib import Path

import pandas as pd

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

from scripts import search_eth_btc_calendar_carry as carry
from scripts import search_long_short_fusion as fusion
from scripts.search_short_bias_overlay import markdown_table


OUT_DIR = Path("reports/long-short-fusion")
PREFIX = "calendar-carry-fusion-leg"
CANDIDATE = carry.Spec("ETH-USDT-SWAP", "1h", "long", 14, "weekend", 8, "calm")
SELECTED_PATH = OUT_DIR / "fusion-selected.csv"
WEIGHTS = (0.00, 0.025, 0.05, 0.075, 0.10, 0.125)
HORIZONS = (
    ("full", None),
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)


def selected_baseline() -> pd.Series:
    selected = pd.read_csv(SELECTED_PATH)
    return selected.iloc[0]


def baseline_equity(row: pd.Series, years: float) -> pd.Series:
    components = fusion.build_components(years)
    series = {key: value[1] for key, value in components.items()}
    weights = {
        str(row["long_variant"]): float(row["long_weight"]),
        "btc_risk_short": float(row["btc_risk_short_weight"]),
        "eth_4h_vol_short": float(row["eth_4h_vol_short_weight"]),
        "btc_4h_vol_short": float(row["btc_4h_vol_short_weight"]),
        "eth_4h_vol_short_gated": float(row["eth_4h_vol_short_gated_weight"]),
        "btc_4h_vol_short_gated": float(row["btc_4h_vol_short_gated_weight"]),
    }
    return fusion.combine_components(series, weights)


def candidate_equity() -> pd.Series:
    frame = carry.resample(carry.load_frame("ETH-USDT-SWAP"), "1h")
    equity, _ = carry.run_spec(CANDIDATE, frame)
    return equity


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


def horizon_slice(series: pd.Series, offset: pd.DateOffset | None) -> pd.Series:
    if offset is None:
        return series
    scoped = series[series.index >= series.index[-1] - offset]
    return scoped if len(scoped) >= 2 else series


def monthly_returns(series: pd.Series) -> pd.Series:
    monthly = series.resample("ME").last()
    return monthly / monthly.shift(1).fillna(series.iloc[0]) - 1.0


def yearly_returns(series: pd.Series) -> pd.Series:
    monthly = monthly_returns(series)
    return monthly.groupby(monthly.index.year).apply(lambda values: float((1.0 + values).prod() - 1.0))


def build_portfolios(base: pd.Series, candidate: pd.Series) -> dict[float, pd.Series]:
    aligned = pd.DataFrame({"base": base, "candidate": candidate}).dropna()
    base_returns = aligned["base"].pct_change().fillna(0.0)
    candidate_returns = aligned["candidate"].pct_change().fillna(0.0)
    portfolios: dict[float, pd.Series] = {}
    for weight in WEIGHTS:
        returns = base_returns + weight * candidate_returns
        portfolios[weight] = fusion.INITIAL_EQUITY * (1.0 + returns).cumprod()
    return portfolios


def horizon_rows(portfolios: dict[float, pd.Series]) -> pd.DataFrame:
    baseline = portfolios[0.0]
    rows: list[dict[str, object]] = []
    baseline_metrics = {label: metrics(horizon_slice(baseline, offset)) for label, offset in HORIZONS}
    for weight, equity in portfolios.items():
        for label, offset in HORIZONS:
            scoped = horizon_slice(equity, offset)
            values = metrics(scoped)
            base_values = baseline_metrics[label]
            rows.append(
                {
                    "calendar_weight": weight,
                    "horizon": label,
                    "start": scoped.index[0].strftime("%Y-%m-%d"),
                    "end": scoped.index[-1].strftime("%Y-%m-%d"),
                    **values,
                    "delta_total": values["total_return"] - base_values["total_return"],
                    "delta_annualized": values["annualized_return"] - base_values["annualized_return"],
                    "delta_max_drawdown": values["max_drawdown"] - base_values["max_drawdown"],
                    "delta_calmar": values["calmar"] - base_values["calmar"],
                }
            )
    return pd.DataFrame(rows)


def deterioration_rows(portfolios: dict[float, pd.Series]) -> pd.DataFrame:
    baseline_monthly = monthly_returns(portfolios[0.0])
    baseline_yearly = yearly_returns(portfolios[0.0])
    rows: list[dict[str, object]] = []
    for weight, equity in portfolios.items():
        monthly = monthly_returns(equity)
        yearly = yearly_returns(equity)
        month_delta = (monthly - baseline_monthly).dropna()
        year_delta = (yearly - baseline_yearly).dropna()
        rows.append(
            {
                "calendar_weight": weight,
                "worse_years": int((year_delta < 0.0).sum()),
                "total_years": int(len(year_delta)),
                "worst_year_delta": float(year_delta.min()) if len(year_delta) else 0.0,
                "worse_months": int((month_delta < 0.0).sum()),
                "total_months": int(len(month_delta)),
                "worst_month_delta": float(month_delta.min()) if len(month_delta) else 0.0,
            }
        )
    return pd.DataFrame(rows)


def detail_rows(portfolios: dict[float, pd.Series], period: str) -> pd.DataFrame:
    baseline = monthly_returns(portfolios[0.0]) if period == "month" else yearly_returns(portfolios[0.0])
    rows: list[dict[str, object]] = []
    for weight, equity in portfolios.items():
        values = monthly_returns(equity) if period == "month" else yearly_returns(equity)
        delta = (values - baseline).dropna()
        for index, value in values.dropna().items():
            rows.append(
                {
                    "calendar_weight": weight,
                    period: index.strftime("%Y-%m") if period == "month" else int(index),
                    "return": float(value),
                    "baseline_return": float(baseline.loc[index]),
                    "delta": float(delta.loc[index]),
                    "worse_than_baseline": bool(delta.loc[index] < 0.0),
                }
            )
    return pd.DataFrame(rows)


def verdict(horizons: pd.DataFrame, deterioration: pd.DataFrame) -> tuple[str, float]:
    full = horizons[horizons["horizon"] == "full"].copy()
    positive = full[
        (full["calendar_weight"] > 0.0)
        & (full["delta_total"] > 0.0)
        & (full["delta_calmar"] > 0.0)
        & (full["delta_max_drawdown"] <= 0.0)
    ]
    stable = deterioration[(deterioration["calendar_weight"] > 0.0) & (deterioration["worse_years"] <= 2)]
    eligible = positive[positive["calendar_weight"].isin(set(stable["calendar_weight"]))]
    if len(eligible) == 0:
        return "do not include in the next main portfolio search", 0.0
    max_weight = float(eligible["calendar_weight"].max())
    return "include as a low-weight leg in the next main portfolio search", max_weight


def report(
    command: str,
    baseline: pd.Series,
    horizons: pd.DataFrame,
    deterioration: pd.DataFrame,
    years: pd.DataFrame,
    months: pd.DataFrame,
) -> str:
    decision, max_weight = verdict(horizons, deterioration)
    summary = horizons[horizons["horizon"].isin(["full", "3y", "1y", "6m", "3m"])]
    year_check = years[years["calendar_weight"] > 0.0]
    month_check = months[months["calendar_weight"] > 0.0].sort_values("delta").head(20)
    return "\n".join(
        [
            "# Calendar Carry Fusion Leg Validation",
            "",
            f"Run command: `{command}`",
            "",
            "Scope: research-only validation under `reports/long-short-fusion`; no live path changed.",
            f"Baseline: first row of `{SELECTED_PATH}` rebuilt from long-short fusion research components.",
            f"Baseline date intersection: {baseline.index[0].strftime('%Y-%m-%d')} to {baseline.index[-1].strftime('%Y-%m-%d')}.",
            f"Candidate leg: `{CANDIDATE.name}`.",
            "",
            f"Decision: **{decision}**.",
            f"Suggested calendar weight upper bound: **{max_weight:.3f}**.",
            "",
            "## Horizon Weight Search",
            "",
            markdown_table(summary),
            "",
            "## Deterioration Summary",
            "",
            markdown_table(deterioration),
            "",
            "## Year Check",
            "",
            markdown_table(year_check),
            "",
            "## Worst Monthly Deltas",
            "",
            markdown_table(month_check),
            "",
        ]
    )


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

    selected = selected_baseline()
    base = baseline_equity(selected, args.years)
    candidate = candidate_equity()
    portfolios = build_portfolios(base, candidate)
    baseline = portfolios[0.0]

    horizons = horizon_rows(portfolios)
    deterioration = deterioration_rows(portfolios)
    years = detail_rows(portfolios, "year")
    months = detail_rows(portfolios, "month")

    args.output_dir.mkdir(parents=True, exist_ok=True)
    horizon_path = args.output_dir / f"{PREFIX}-horizons.csv"
    deterioration_path = args.output_dir / f"{PREFIX}-deterioration.csv"
    years_path = args.output_dir / f"{PREFIX}-years.csv"
    months_path = args.output_dir / f"{PREFIX}-months.csv"
    report_path = args.output_dir / f"{PREFIX}.md"

    horizons.to_csv(horizon_path, index=False)
    deterioration.to_csv(deterioration_path, index=False)
    years.to_csv(years_path, index=False)
    months.to_csv(months_path, index=False)
    report_path.write_text(
        report(
            f"rtk .venv/bin/python scripts/validate_calendar_carry_fusion_leg.py --years {args.years}",
            baseline,
            horizons,
            deterioration,
            years,
            months,
        ),
        encoding="utf-8",
    )
    print(report_path)
    print(horizons.to_string(index=False))
    print(deterioration.to_string(index=False))
    return 0


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