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", ] HORIZON_COLUMNS = [ "name", "kind", "horizon", "start", "end", "total_return", "annualized_return", "max_drawdown", "calmar", "improved_vs_no_calendar", "delta_total_return", "delta_calmar", "delta_max_drawdown", ] MONTHLY_COLUMNS = ["name", "kind", "month", "start_equity", "end_equity", "return"] 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: 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", ] if selected.empty: selected_table = "No selected candidates. The strict recent filter requires positive 1y, 6m, and 3m returns." top_horizons = pd.DataFrame() monthly_tail = pd.DataFrame() else: selected_table = markdown_table(selected[summary_cols]) top_horizons = horizons[horizons["name"].isin(set(selected["name"]))] monthly_tail = monthly.tail(96) 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", "", selected_table, "", "## Top Candidate Horizons", "", markdown_table(top_horizons), "", "## Recent Monthly Returns", "", markdown_table(monthly_tail), "", ] ) 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() if horizons.empty: horizons = pd.DataFrame(columns=HORIZON_COLUMNS) if monthly.empty: monthly = pd.DataFrame(columns=MONTHLY_COLUMNS) 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())