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