okx-codex-trader/scripts/search_long_short_fusion.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_short_bias_swing as swing
from scripts import search_short_overlay_mix as overlay
from scripts.search_short_bias_overlay import markdown_table


OUTPUT_DIR = Path("reports/long-short-fusion")
PREFIX = "fusion"
INITIAL_EQUITY = 10_000.0
HORIZONS = (
    ("full", None),
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)


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


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


def horizon_return_fields(series: pd.Series) -> dict[str, float]:
    rows = {row["horizon"]: row for row in horizon_rows("", "", series)}
    return {
        "h3y_return": float(rows["3y"]["total_return"]),
        "h1y_return": float(rows["1y"]["total_return"]),
        "h6m_return": float(rows["6m"]["total_return"]),
        "h3m_return": float(rows["3m"]["total_return"]),
    }


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


def component_returns(series: pd.Series) -> pd.Series:
    return series.pct_change().fillna(0.0)


def combine_components(components: dict[str, pd.Series], weights: dict[str, float]) -> pd.Series:
    frame = pd.DataFrame({name: component_returns(series) for name, series in components.items()}).dropna()
    combined = sum(frame[name] * weights.get(name, 0.0) for name in frame.columns)
    equity = INITIAL_EQUITY * (1.0 + combined).cumprod()
    equity.name = "equity"
    return equity


def riskoff_long_equity(series: pd.Series, gate: pd.Series, multiplier: float) -> pd.Series:
    aligned = pd.DataFrame({"returns": component_returns(series), "gate": gate.reindex(series.index).ffill().fillna(0.0)}).dropna()
    scale = aligned["gate"].where(aligned["gate"] <= 0.0, multiplier).where(aligned["gate"] > 0.0, 1.0)
    equity = INITIAL_EQUITY * (1.0 + aligned["returns"] * scale).cumprod()
    equity.name = "equity"
    return equity


def btc_risk_short(years: float) -> tuple[str, pd.Series, dict[str, float | int], pd.Series]:
    params = overlay.BtcRiskShort(
        family="btc_risk_pair",
        bar="1h",
        btc_trend=1440,
        btc_lookback=336,
        symbol_trend=720,
        vol_lookback=336,
        btc_max_momentum=-0.005,
        btc_min_drop=0.025,
        min_btc_vol=0.012,
        symbol_max_momentum=-0.010,
        short_symbols=("ETH-USDT-SWAP",),
    )
    frames = overlay.load_frames(years)
    closes = pd.DataFrame({symbol: frames[(symbol, params.bar)]["close"] for symbol in ("BTC-USDT-SWAP", "ETH-USDT-SWAP")}).dropna()
    weights = overlay.short_weights(closes, params)
    equity = daily(overlay.equity_from_weights(closes, weights))
    risk_state = daily((weights["ETH-USDT-SWAP"].abs() > 0.0).astype(float))
    return params.name, equity, overlay.trade_stats(weights, closes), risk_state


def swing_short(strategy: swing.Strategy, years: float) -> tuple[str, pd.Series, dict[str, float | int]]:
    frame = swing.resample_frame(swing.load_15m_frame(strategy.symbol, years), strategy.bar)
    btc_frame = swing.resample_frame(swing.load_15m_frame("BTC-USDT-SWAP", years), strategy.bar) if strategy.symbol == "ETH-USDT-SWAP" else None
    result = swing.run_strategy(strategy, frame, btc_frame)
    equity = swing.daily_equity(result)
    return strategy.name, equity, swing.trade_metrics(result["trades"])


def build_components(years: float) -> dict[str, tuple[str, pd.Series, dict[str, float | int]]]:
    base_name, base_equity, base_stats = overlay.rotation_base(years)
    btc_risk_name, btc_risk_equity, btc_risk_stats, risk_state = btc_risk_short(years)
    eth_name, eth_equity, eth_stats = swing_short(
        swing.Strategy(
            family="vol_expansion_short",
            symbol="ETH-USDT-SWAP",
            bar="4H",
            params={
                "fast": 20,
                "slow": 80,
                "entry": 20,
                "exit": 10,
                "atr": 14,
                "stop_atr": 3.0,
                "take_atr": 6.0,
                "max_hold": 120,
                "vol_window": 120,
                "vol_quantile": 0.8,
            },
        ),
        years,
    )
    btc_name, btc_equity, btc_stats = swing_short(
        swing.Strategy(
            family="vol_expansion_short",
            symbol="BTC-USDT-SWAP",
            bar="4H",
            params={
                "fast": 30,
                "slow": 120,
                "entry": 20,
                "exit": 10,
                "atr": 14,
                "stop_atr": 3.0,
                "take_atr": 6.0,
                "max_hold": 120,
                "vol_window": 120,
                "vol_quantile": 0.8,
            },
        ),
        years,
    )
    components = {
        "long_rotation": (base_name, daily(base_equity), base_stats),
        "long_rotation_riskoff70": (f"{base_name}-riskoff70", riskoff_long_equity(daily(base_equity), risk_state, 0.70), base_stats),
        "long_rotation_riskoff50": (f"{base_name}-riskoff50", riskoff_long_equity(daily(base_equity), risk_state, 0.50), base_stats),
        "long_rotation_riskoff25": (f"{base_name}-riskoff25", riskoff_long_equity(daily(base_equity), risk_state, 0.25), base_stats),
        "long_rotation_riskoff00": (f"{base_name}-riskoff00", riskoff_long_equity(daily(base_equity), risk_state, 0.0), base_stats),
        "btc_risk_short": (btc_risk_name, btc_risk_equity, btc_risk_stats),
        "eth_4h_vol_short": (eth_name, eth_equity, eth_stats),
        "btc_4h_vol_short": (btc_name, btc_equity, btc_stats),
        "eth_4h_vol_short_gated": (f"{eth_name}-btc_risk_gated", gated_equity(eth_equity, risk_state), eth_stats),
        "btc_4h_vol_short_gated": (f"{btc_name}-btc_risk_gated", gated_equity(btc_equity, risk_state), btc_stats),
    }
    return components


def gated_equity(series: pd.Series, gate: pd.Series) -> pd.Series:
    aligned = pd.DataFrame({"returns": component_returns(series), "gate": gate.reindex(series.index).ffill().fillna(0.0)}).dropna()
    equity = INITIAL_EQUITY * (1.0 + aligned["returns"] * aligned["gate"]).cumprod()
    equity.name = "equity"
    return equity


def monthly_rows(name: str, kind: str, series: pd.Series) -> pd.DataFrame:
    monthly = series.resample("ME").last()
    frame = pd.DataFrame(
        {
            "name": name,
            "kind": kind,
            "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 score(row: dict[str, object]) -> float:
    return float(row["annualized_return"]) - float(row["max_drawdown"]) + 0.5 * float(row["h1y_return"]) + 0.25 * float(row["h6m_return"])


def report_text(command: str, paths: list[Path], selected: pd.DataFrame, components: pd.DataFrame, horizons: pd.DataFrame, monthly: pd.DataFrame) -> str:
    return "\n".join(
        [
            "# Long Short Fusion Search",
            "",
            f"Run command: `{command}`",
            "",
            "Output files:",
            *[f"- `{path}`" for path in paths],
            "",
            "Objective: combine existing long rotation and short-biased legs into bidirectional portfolios. This is research only; no live service was changed.",
            "",
            "## Selected Fusion Candidates",
            "",
            markdown_table(selected),
            "",
            "## Component Curves",
            "",
            markdown_table(components),
            "",
            "## Horizons",
            "",
            markdown_table(horizons[horizons["name"].isin(set(selected["name"]))]),
            "",
            "## Recent Monthly Returns",
            "",
            markdown_table(monthly[monthly["name"].isin(set(selected["name"]))].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)
    components = build_components(args.years)
    component_series = {key: value[1] for key, value in components.items()}
    rows: list[dict[str, object]] = []
    horizon_output: list[dict[str, object]] = []
    monthly_output: list[pd.DataFrame] = []

    component_rows = []
    for key, (name, series, stats) in components.items():
        row = {
            "name": key,
            "source": name,
            "kind": "component",
            "long_weight": 1.0 if key == "long_rotation" else 0.0,
            "long_variant": key if key.startswith("long_rotation") else "",
            "btc_risk_short_weight": 1.0 if key == "btc_risk_short" else 0.0,
            "eth_4h_vol_short_weight": 1.0 if key == "eth_4h_vol_short" else 0.0,
            "btc_4h_vol_short_weight": 1.0 if key == "btc_4h_vol_short" else 0.0,
            "eth_4h_vol_short_gated_weight": 1.0 if key == "eth_4h_vol_short_gated" else 0.0,
            "btc_4h_vol_short_gated_weight": 1.0 if key == "btc_4h_vol_short_gated" else 0.0,
            **horizon_return_fields(series),
            **metrics(series),
            **stats,
        }
        component_rows.append(row)
        horizon_output.extend(horizon_rows(key, "component", series))
        monthly_output.append(monthly_rows(key, "component", series))

    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

    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
                                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,
                                }
                                equity = combine_components(component_series, weights)
                                name = (
                                    f"fusion-{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}"
                                )
                                row = {
                                    "name": name,
                                    "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,
                                    "gross_exposure": long_weight + short_weight,
                                    "short_exposure": short_weight,
                                    **horizon_return_fields(equity),
                                    **metrics(equity),
                                    "trades": int(sum(int(components[key][2].get("trades", 0)) for key, weight in weights.items() if weight > 0.0)),
                                    "win_rate": 0.0,
                                    "profit_factor": 0.0,
                                }
                                row["score"] = score(row)
                                row["selected"] = "no"
                                rows.append(row)
                                horizon_output.extend(horizon_rows(name, "fusion", equity))
                                monthly_output.append(monthly_rows(name, "fusion", equity))

    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)
    selected = pd.concat([low_drawdown, high_return]).drop_duplicates("name").head(12).copy()
    selected["selected"] = "yes"
    total.loc[total["name"].isin(set(selected["name"])), "selected"] = "yes"

    component_frame = pd.DataFrame(component_rows)
    horizons = pd.DataFrame(horizon_output)
    monthly = pd.concat(monthly_output, ignore_index=True)

    total_path = args.output_dir / f"{PREFIX}-total.csv"
    selected_path = args.output_dir / f"{PREFIX}-selected.csv"
    component_path = args.output_dir / f"{PREFIX}-components.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.to_csv(total_path, index=False)
    selected.to_csv(selected_path, index=False)
    component_frame.to_csv(component_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.py --years {args.years}",
            [total_path, selected_path, component_path, horizon_path, monthly_path, report_path],
            selected,
            component_frame,
            horizons,
            monthly,
        ),
        encoding="utf-8",
    )
    print(report_path)
    print(selected.head(8).to_string(index=False))
    return 0


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