okx-codex-trader/scripts/search_long_short_fusion_with_calendar.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 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


OUTPUT_DIR = Path("reports/long-short-fusion")
PREFIX = "fusion-calendar"
CANDIDATE = carry.Spec("ETH-USDT-SWAP", "1h", "long", 14, "weekend", 8, "calm")
CALENDAR_WEIGHTS = (0.0, 0.025, 0.05, 0.075, 0.10, 0.125)
HORIZONS = fusion.HORIZONS
TOTAL_COLUMNS = [
    "name",
    "selected",
    "long_variant",
    "long_weight",
    "btc_risk_short_weight",
    "eth_4h_vol_short_weight",
    "btc_4h_vol_short_weight",
    "eth_4h_vol_short_gated_weight",
    "btc_4h_vol_short_gated_weight",
    "calendar_weight",
    "gross_exposure",
    "short_exposure",
    "total_return",
    "annualized_return",
    "max_drawdown",
    "calmar",
    "h3y_return",
    "h1y_return",
    "h6m_return",
    "h3m_return",
    "full_improved_vs_no_calendar",
    "3y_improved_vs_no_calendar",
    "1y_improved_vs_no_calendar",
    "6m_improved_vs_no_calendar",
    "3m_improved_vs_no_calendar",
    "score",
]


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


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 horizon_metrics(series: pd.Series) -> dict[str, dict[str, float]]:
    return {label: fusion.metrics(horizon_slice(series, offset)) for label, offset in HORIZONS}


def row_from_metrics(name: str, metrics_by_horizon: dict[str, dict[str, float]]) -> dict[str, object]:
    full = metrics_by_horizon["full"]
    return {
        "name": name,
        **full,
        "h3y_return": metrics_by_horizon["3y"]["total_return"],
        "h1y_return": metrics_by_horizon["1y"]["total_return"],
        "h6m_return": metrics_by_horizon["6m"]["total_return"],
        "h3m_return": metrics_by_horizon["3m"]["total_return"],
        "h3y_calmar": metrics_by_horizon["3y"]["calmar"],
        "h1y_calmar": metrics_by_horizon["1y"]["calmar"],
        "h6m_calmar": metrics_by_horizon["6m"]["calmar"],
        "h3m_calmar": metrics_by_horizon["3m"]["calmar"],
        "h3y_max_drawdown": metrics_by_horizon["3y"]["max_drawdown"],
        "h1y_max_drawdown": metrics_by_horizon["1y"]["max_drawdown"],
        "h6m_max_drawdown": metrics_by_horizon["6m"]["max_drawdown"],
        "h3m_max_drawdown": metrics_by_horizon["3m"]["max_drawdown"],
    }


def improvement_fields(
    current: dict[str, dict[str, float]],
    baseline: dict[str, dict[str, float]] | None,
) -> dict[str, object]:
    fields: dict[str, object] = {}
    for label in ("full", "3y", "1y", "6m", "3m"):
        if baseline is None:
            fields[f"{label}_improved_vs_no_calendar"] = False
            fields[f"{label}_delta_calmar"] = 0.0
            fields[f"{label}_delta_max_drawdown"] = 0.0
            fields[f"{label}_delta_total_return"] = 0.0
            continue
        delta_calmar = current[label]["calmar"] - baseline[label]["calmar"]
        delta_drawdown = current[label]["max_drawdown"] - baseline[label]["max_drawdown"]
        delta_total = current[label]["total_return"] - baseline[label]["total_return"]
        fields[f"{label}_improved_vs_no_calendar"] = delta_calmar > 0.0 and delta_drawdown <= 0.0
        fields[f"{label}_delta_calmar"] = delta_calmar
        fields[f"{label}_delta_max_drawdown"] = delta_drawdown
        fields[f"{label}_delta_total_return"] = delta_total
    return fields


def candidate_name(
    long_key: str,
    long_weight: float,
    btc_risk_weight: float,
    eth_swing_weight: float,
    btc_swing_weight: float,
    eth_gated_weight: float,
    btc_gated_weight: float,
    calendar_weight: float,
) -> str:
    return (
        f"fusion-cal-{long_key.replace('long_rotation', 'lr')}-l{long_weight:.2f}"
        f"-brs{btc_risk_weight:.2f}"
        f"-eth4hs{eth_swing_weight:.2f}"
        f"-btc4hs{btc_swing_weight:.2f}"
        f"-eg{eth_gated_weight:.2f}"
        f"-bg{btc_gated_weight:.2f}"
        f"-cal{calendar_weight:.3f}"
    )


def build_equity(series: dict[str, pd.Series], row: pd.Series) -> pd.Series:
    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"]),
        "calendar_carry": float(row["calendar_weight"]),
    }
    return fusion.combine_components(series, weights)


def horizon_rows(name: str, kind: str, series: pd.Series, baseline: pd.Series | None) -> list[dict[str, object]]:
    rows: list[dict[str, object]] = []
    baseline_metrics = horizon_metrics(baseline) if baseline is not None else None
    for label, offset in HORIZONS:
        scoped = horizon_slice(series, offset)
        values = fusion.metrics(scoped)
        base_values = baseline_metrics[label] if baseline_metrics is not None else None
        row = {
            "name": name,
            "kind": kind,
            "horizon": label,
            "start": scoped.index[0].strftime("%Y-%m-%d"),
            "end": scoped.index[-1].strftime("%Y-%m-%d"),
            **values,
        }
        if base_values is None:
            row.update(
                {
                    "improved_vs_no_calendar": False,
                    "delta_total_return": 0.0,
                    "delta_calmar": 0.0,
                    "delta_max_drawdown": 0.0,
                }
            )
        else:
            row.update(
                {
                    "improved_vs_no_calendar": values["calmar"] > base_values["calmar"]
                    and values["max_drawdown"] <= base_values["max_drawdown"],
                    "delta_total_return": values["total_return"] - base_values["total_return"],
                    "delta_calmar": values["calmar"] - base_values["calmar"],
                    "delta_max_drawdown": values["max_drawdown"] - base_values["max_drawdown"],
                }
            )
        rows.append(row)
    return rows


def report_text(command: str, paths: list[Path], selected: pd.DataFrame, horizons: pd.DataFrame, monthly: pd.DataFrame) -> str:
    top_horizons = horizons[horizons["name"].isin(set(selected["name"]))]
    summary_cols = [
        "name",
        "calendar_weight",
        "total_return",
        "annualized_return",
        "max_drawdown",
        "calmar",
        "full_improved_vs_no_calendar",
        "3y_improved_vs_no_calendar",
        "1y_improved_vs_no_calendar",
        "6m_improved_vs_no_calendar",
        "3m_improved_vs_no_calendar",
    ]
    return "\n".join(
        [
            "# Long Short Fusion Search With Calendar Carry",
            "",
            f"Run command: `{command}`",
            "",
            "Output files:",
            *[f"- `{path}`" for path in paths],
            "",
            "Scope: research-only search under `reports/long-short-fusion`; no live path changed.",
            f"Calendar carry leg: `{CANDIDATE.name}`.",
            f"Calendar weights searched: `{', '.join(f'{value:.3f}' for value in CALENDAR_WEIGHTS)}`.",
            "",
            "## Selected Candidates",
            "",
            markdown_table(selected[summary_cols]),
            "",
            "## Top Candidate Horizons",
            "",
            markdown_table(top_horizons),
            "",
            "## Recent Monthly Returns",
            "",
            markdown_table(monthly.tail(96)),
            "",
        ]
    )


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

    args.output_dir.mkdir(parents=True, exist_ok=True)
    fusion.overlay.SYMBOLS = ("BTC-USDT-SWAP", "ETH-USDT-SWAP")
    components = fusion.build_components(args.years)
    component_series = {key: value[1] for key, value in components.items()}
    component_series["calendar_carry"] = calendar_equity()

    if args.focused:
        long_keys = ("long_rotation_riskoff00",)
        long_weights = (1.08, 1.10, 1.12, 1.15, 1.18, 1.20)
        btc_risk_weights = (0.06, 0.08, 0.10, 0.12)
        eth_swing_weights = (0.00, 0.04, 0.06, 0.08, 0.10, 0.12)
        btc_swing_weights = (0.00, 0.02, 0.04, 0.06)
        eth_gated_weights = (0.00, 0.06, 0.08, 0.10, 0.12)
        btc_gated_weights = (0.00, 0.02, 0.03, 0.04, 0.06)
        max_short_weight = 0.30
        max_gross_exposure = 1.45
    else:
        long_keys = ("long_rotation", "long_rotation_riskoff70", "long_rotation_riskoff50", "long_rotation_riskoff25", "long_rotation_riskoff00")
        long_weights = (0.70, 0.85, 1.00, 1.10, 1.20)
        btc_risk_weights = (0.00, 0.02, 0.04, 0.06, 0.08, 0.10, 0.12)
        eth_swing_weights = (0.00, 0.02, 0.04, 0.06, 0.08, 0.10, 0.12)
        btc_swing_weights = (0.00, 0.02, 0.04, 0.06)
        eth_gated_weights = (0.00, 0.04, 0.08, 0.12)
        btc_gated_weights = (0.00, 0.03, 0.06)
        max_short_weight = 0.40
        max_gross_exposure = 1.45

    rows: list[dict[str, object]] = []
    no_calendar_metrics: dict[tuple[object, ...], dict[str, dict[str, float]]] = {}

    for long_key in long_keys:
        for long_weight in long_weights:
            for btc_risk_weight in btc_risk_weights:
                for eth_swing_weight in eth_swing_weights:
                    for btc_swing_weight in btc_swing_weights:
                        for eth_gated_weight in eth_gated_weights:
                            for btc_gated_weight in btc_gated_weights:
                                if eth_swing_weight > 0.0 and eth_gated_weight > 0.0:
                                    continue
                                if btc_swing_weight > 0.0 and btc_gated_weight > 0.0:
                                    continue
                                short_weight = btc_risk_weight + eth_swing_weight + btc_swing_weight + eth_gated_weight + btc_gated_weight
                                if short_weight <= 0.0 or short_weight > max_short_weight:
                                    continue
                                if long_weight + short_weight > max_gross_exposure:
                                    continue

                                base_key = (
                                    long_key,
                                    long_weight,
                                    btc_risk_weight,
                                    eth_swing_weight,
                                    btc_swing_weight,
                                    eth_gated_weight,
                                    btc_gated_weight,
                                )
                                for calendar_weight in CALENDAR_WEIGHTS:
                                    weights = {
                                        long_key: long_weight,
                                        "btc_risk_short": btc_risk_weight,
                                        "eth_4h_vol_short": eth_swing_weight,
                                        "btc_4h_vol_short": btc_swing_weight,
                                        "eth_4h_vol_short_gated": eth_gated_weight,
                                        "btc_4h_vol_short_gated": btc_gated_weight,
                                        "calendar_carry": calendar_weight,
                                    }
                                    equity = fusion.combine_components(component_series, weights)
                                    metrics_by_horizon = horizon_metrics(equity)
                                    if calendar_weight == 0.0:
                                        no_calendar_metrics[base_key] = metrics_by_horizon
                                    name = candidate_name(
                                        long_key,
                                        long_weight,
                                        btc_risk_weight,
                                        eth_swing_weight,
                                        btc_swing_weight,
                                        eth_gated_weight,
                                        btc_gated_weight,
                                        calendar_weight,
                                    )
                                    row = {
                                        **row_from_metrics(name, metrics_by_horizon),
                                        "kind": "fusion",
                                        "long_variant": long_key,
                                        "long_weight": long_weight,
                                        "btc_risk_short_weight": btc_risk_weight,
                                        "eth_4h_vol_short_weight": eth_swing_weight,
                                        "btc_4h_vol_short_weight": btc_swing_weight,
                                        "eth_4h_vol_short_gated_weight": eth_gated_weight,
                                        "btc_4h_vol_short_gated_weight": btc_gated_weight,
                                        "calendar_weight": calendar_weight,
                                        "gross_exposure": long_weight + short_weight + calendar_weight,
                                        "short_exposure": short_weight,
                                        "calendar_leg": CANDIDATE.name,
                                        "trades": int(
                                            sum(int(components[key][2].get("trades", 0)) for key, weight in weights.items() if key in components and weight > 0.0)
                                        ),
                                        "win_rate": 0.0,
                                        "profit_factor": 0.0,
                                    }
                                    row.update(improvement_fields(metrics_by_horizon, no_calendar_metrics.get(base_key)))
                                    row["score"] = fusion.score(row)
                                    row["selected"] = "no"
                                    rows.append(row)

    total = pd.DataFrame(rows).sort_values(
        ["h1y_return", "h6m_return", "h3m_return", "max_drawdown", "annualized_return"],
        ascending=[False, False, False, True, False],
    )
    recent_ok = total[(total["h1y_return"] > 0.0) & (total["h6m_return"] > 0.0) & (total["h3m_return"] > 0.0)].copy()
    low_drawdown = recent_ok[recent_ok["max_drawdown"] <= 0.10].sort_values("score", ascending=False).head(10)
    high_return = recent_ok.sort_values(["annualized_return", "max_drawdown"], ascending=[False, True]).head(10)
    improved = recent_ok[
        (recent_ok["calendar_weight"] > 0.0)
        & (recent_ok["full_improved_vs_no_calendar"])
        & (recent_ok["1y_improved_vs_no_calendar"])
    ].sort_values("score", ascending=False).head(10)
    selected = pd.concat([low_drawdown, high_return, improved]).drop_duplicates("name").head(12).copy()
    selected["selected"] = "yes"
    total.loc[total["name"].isin(set(selected["name"])), "selected"] = "yes"

    horizon_output: list[dict[str, object]] = []
    monthly_output: list[pd.DataFrame] = []
    for _, row in selected.iterrows():
        equity = build_equity(component_series, row)
        baseline_row = row.copy()
        baseline_row["calendar_weight"] = 0.0
        baseline = build_equity(component_series, baseline_row)
        horizon_output.extend(horizon_rows(str(row["name"]), "fusion", equity, baseline))
        monthly_output.append(fusion.monthly_rows(str(row["name"]), "fusion", equity))

    horizons = pd.DataFrame(horizon_output)
    monthly = pd.concat(monthly_output, ignore_index=True) if monthly_output else pd.DataFrame()

    total_path = args.output_dir / f"{PREFIX}-total.csv"
    selected_path = args.output_dir / f"{PREFIX}-selected.csv"
    horizon_path = args.output_dir / f"{PREFIX}-horizons.csv"
    monthly_path = args.output_dir / f"{PREFIX}-monthly.csv"
    report_path = args.output_dir / f"{PREFIX}-report.md"
    total[TOTAL_COLUMNS].to_csv(total_path, index=False)
    selected.to_csv(selected_path, index=False)
    horizons.to_csv(horizon_path, index=False)
    monthly.to_csv(monthly_path, index=False)
    report_path.write_text(
        report_text(
            f"rtk .venv/bin/python scripts/search_long_short_fusion_with_calendar.py --years {args.years}{' --focused' if args.focused else ''}",
            [total_path, selected_path, horizon_path, monthly_path, report_path],
            selected,
            horizons,
            monthly,
        ),
        encoding="utf-8",
    )
    print(report_path)
    print(selected.head(8).to_string(index=False))
    return 0


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