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())