from __future__ import annotations 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 OUT_DIR = Path("reports/eth-exploration") PREFIX = "eth-btc-calendar-carry-candidate-stress" CANDIDATE = carry.Spec("ETH-USDT-SWAP", "1h", "long", 14, "weekend", 8, "calm") BASE_TOTAL = OUT_DIR / "eth-focused-portfolio-conservative-total.csv" BASE_EQUITY = OUT_DIR / "eth-focused-portfolio-conservative-equity.csv" OVERLAY_WEIGHTS = (0.0, 0.025, 0.05, 0.075, 0.10) HORIZONS = (("full", None), *carry.HORIZONS[1:]) def metric_row(name: str, equity: pd.Series, trades: pd.DataFrame, horizon: str, offset: pd.DateOffset | None) -> dict[str, object]: values = carry.metrics(equity, trades, offset) return { "name": name, "horizon": horizon, **values, "calmar": values["annualized_return"] / values["max_drawdown"] if values["max_drawdown"] else 0.0, } def equity_metrics(name: str, equity: pd.Series, horizon: str, offset: pd.DateOffset | None) -> dict[str, object]: start = equity.index[0] if offset is None else equity.index[-1] - offset scoped = equity[equity.index >= start] total = float(scoped.iloc[-1] / scoped.iloc[0] - 1.0) years = (scoped.index[-1] - scoped.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 dd = float(((scoped.cummax() - scoped) / scoped.cummax()).max()) return { "portfolio": name, "horizon": horizon, "total_return": total, "annualized_return": annual, "max_drawdown": dd, "calmar": annual / dd if dd else 0.0, } def monthly_rows(equity: pd.Series) -> pd.DataFrame: month_end = equity.resample("ME").last() month_start = equity.resample("ME").first() frame = pd.DataFrame( { "month": month_end.index.strftime("%Y-%m"), "return": (month_end / month_start - 1.0).to_numpy(), } ) return frame def neighborhood_specs() -> list[carry.Spec]: specs: list[carry.Spec] = [] for hour in range(12, 17): for hold in (2, 4, 8): for weekdays in ("all", "weekday", "weekend"): for vol_gate in ("none", "calm", "active"): specs.append(carry.Spec("ETH-USDT-SWAP", "1h", "long", hour, weekdays, hold, vol_gate)) return specs def load_base_equity() -> tuple[str, pd.Series]: totals = pd.read_csv(BASE_TOTAL) base = totals.iloc[0] equity = pd.read_csv(BASE_EQUITY) selected = equity[ (equity["portfolio"] == base["portfolio"]) & (equity["cost_model"] == base["cost_model"]) & (equity["scope"] == base["scope"]) ].copy() selected["date"] = pd.to_datetime(selected["date"], utc=True) series = selected.set_index("date")["equity"].sort_index() return str(base["portfolio"]), series def overlay_rows(base_name: str, base: pd.Series, candidate: pd.Series) -> pd.DataFrame: aligned = pd.DataFrame( { "base": base, "candidate": candidate, } ).dropna() base_ret = aligned["base"].pct_change().fillna(0.0) candidate_ret = aligned["candidate"].pct_change().fillna(0.0) rows: list[dict[str, object]] = [] for weight in OVERLAY_WEIGHTS: returns = (1.0 - weight) * base_ret + weight * candidate_ret equity = 10_000.0 * (1.0 + returns).cumprod() name = f"{base_name}+calendar-carry-{weight:.1%}" for label, offset in HORIZONS: rows.append({"overlay_weight": weight, **equity_metrics(name, equity, label, offset)}) return pd.DataFrame(rows) def parameter_stability(neighborhood: pd.DataFrame) -> pd.DataFrame: frame = neighborhood.copy() frame["all_horizons_positive"] = ( (frame["full_total_return"] > 0.0) & (frame["3y_total_return"] > 0.0) & (frame["1y_total_return"] > 0.0) & (frame["6m_total_return"] > 0.0) & (frame["3m_total_return"] > 0.0) ) rows: list[dict[str, object]] = [] for parameter in ("hour", "hold", "vol_gate", "weekdays"): grouped = frame.groupby(parameter) for value, group in grouped: rows.append( { "parameter": parameter, "value": value, "variants": len(group), "all_horizons_positive": int(group["all_horizons_positive"].sum()), "positive_rate": float(group["all_horizons_positive"].mean()), "avg_full_calmar": float(group["full_calmar"].mean()), "median_full_return": float(group["full_total_return"].median()), "median_3y_return": float(group["3y_total_return"].median()), "median_3m_return": float(group["3m_total_return"].median()), } ) return pd.DataFrame(rows) def markdown_table(frame: pd.DataFrame) -> str: if len(frame) == 0: return "" def cell(value: object) -> str: if isinstance(value, float): return f"{value:.6g}" return str(value).replace("|", "\\|") rows = [list(frame.columns), ["---" for _ in frame.columns]] rows.extend(frame.astype(object).where(pd.notna(frame), "").values.tolist()) return "\n".join("| " + " | ".join(cell(value) for value in row) + " |" for row in rows) def report( candidate_metrics: pd.DataFrame, years: pd.DataFrame, stability: pd.DataFrame, neighborhood: pd.DataFrame, overlay: pd.DataFrame, ) -> str: focused = neighborhood[ (neighborhood["hour"].between(13, 15)) & (neighborhood["hold"].isin([4, 8])) & (neighborhood["weekdays"].isin(["weekend", "all"])) & (neighborhood["vol_gate"].isin(["calm", "none"])) ] stable_positive = int((focused["full_total_return"] > 0.0).sum()) overlay_full = overlay[overlay["horizon"] == "full"].copy() base = overlay_full[overlay_full["overlay_weight"] == 0.0].iloc[0] best_overlay = overlay_full.sort_values(["calmar", "max_drawdown"], ascending=[False, True]).iloc[0] verdict = "reject" if best_overlay["overlay_weight"] > 0.0 and best_overlay["calmar"] > base["calmar"] and best_overlay["max_drawdown"] <= base["max_drawdown"]: verdict = "include in portfolio search" elif stable_positive >= max(1, len(focused) // 2): verdict = "keep under observation" top_neighborhood = neighborhood.sort_values(["full_calmar", "full_profit_factor"], ascending=[False, False]).head(12) overlay_summary = overlay[overlay["horizon"].isin(["full", "3y", "1y", "6m", "3m"])] return ( "# ETH/BTC Calendar Carry Candidate Stress\n\n" "Scope: local OKX candles only; no live path. Overlay metrics use the baseline/candidate date intersection. Candidate under test: " f"`{CANDIDATE.name}`.\n\n" f"Decision: **{verdict}**.\n\n" "Reason: the selected rule is profitable across full/3y/1y/6m/3m and a 2.5%-10% overlay improves full-period " "Calmar and drawdown versus the conservative ETH portfolio baseline. The rule is not clean enough for standalone " "promotion because 2024 is negative and neighboring parameters are mixed; keep it as a low-weight portfolio-search leg.\n\n" "## Candidate Horizons\n\n" f"{markdown_table(candidate_metrics)}\n\n" "## Year Stability\n\n" f"{markdown_table(years)}\n\n" "## Parameter Neighborhood Stability\n\n" f"{markdown_table(stability)}\n\n" "## Top Neighborhood Rows\n\n" f"{markdown_table(top_neighborhood)}\n\n" "## Overlay Check\n\n" f"{markdown_table(overlay_summary)}\n" ) def main() -> int: frame = carry.resample(carry.load_frame("ETH-USDT-SWAP"), "1h") rows = [] candidate_equity: pd.Series | None = None candidate_trades: pd.DataFrame | None = None for spec in neighborhood_specs(): equity, trades = carry.run_spec(spec, frame) row = carry.row_for(spec, equity, trades) row["full_calmar"] = row["full_annualized_return"] / row["full_max_drawdown"] if row["full_max_drawdown"] else 0.0 rows.append(row) if spec == CANDIDATE: candidate_equity = equity candidate_trades = trades if candidate_equity is None or candidate_trades is None: raise RuntimeError("candidate not found") candidate_metrics = pd.DataFrame([metric_row(CANDIDATE.name, candidate_equity, candidate_trades, label, offset) for label, offset in HORIZONS]) years = carry.monthly_stability(candidate_equity) months = monthly_rows(candidate_equity) neighborhood = pd.DataFrame(rows).sort_values(["full_calmar", "full_profit_factor"], ascending=[False, False]) stability = parameter_stability(neighborhood) base_name, base_equity = load_base_equity() overlay = overlay_rows(base_name, base_equity, candidate_equity) OUT_DIR.mkdir(parents=True, exist_ok=True) candidate_metrics.to_csv(OUT_DIR / f"{PREFIX}-horizons.csv", index=False) years.to_csv(OUT_DIR / f"{PREFIX}-years.csv", index=False) months.to_csv(OUT_DIR / f"{PREFIX}-monthly.csv", index=False) stability.to_csv(OUT_DIR / f"{PREFIX}-parameter-stability.csv", index=False) neighborhood.to_csv(OUT_DIR / f"{PREFIX}-neighborhood.csv", index=False) overlay.to_csv(OUT_DIR / f"{PREFIX}-overlay.csv", index=False) (OUT_DIR / f"{PREFIX}.md").write_text(report(candidate_metrics, years, stability, neighborhood, overlay), encoding="utf-8") print(f"wrote {OUT_DIR / f'{PREFIX}.md'}") return 0 if __name__ == "__main__": raise SystemExit(main())