from __future__ import annotations import argparse import sys from dataclasses import dataclass from itertools import combinations from pathlib import Path import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts import explore_ultrashort as explore from scripts import search_eth_price_twap_variants as price_twap OUTPUT_DIR = Path("reports/eth-exploration") YEARS = 10.0 PRIMARY_COST = "maker_taker" COSTS = { "maker_maker": 0.0012, "maker_taker": 0.0021, "taker_taker": 0.0030, } HORIZONS = ( ("3y", pd.DateOffset(years=3)), ("1y", pd.DateOffset(years=1)), ("6m", pd.DateOffset(months=6)), ("3m", pd.DateOffset(months=3)), ) @dataclass(frozen=True) class Strategy: family: str bar: str candidate: object pair: bool def twap_candidate_name(candidate: object) -> str: if isinstance(candidate, dict): return str(candidate["name"]) return str(candidate.name) def build_strategies() -> list[Strategy]: return [ Strategy( "eth_price_twap_deep", "15m", { "name": "rsi2-long-guarded-price-twap-o0.0030-0.0060-0.0090-v2-t160-l5.0-x50.0-sl0.012-mh48", "spec": { "trend_sma": 160, "rsi_threshold": 5.0, "exit_rsi": 50.0, "stop_loss_pct": 0.012, "max_hold_bars": 48, "entry_offsets": (0.003, 0.006, 0.009), "entry_valid_bars": 2, "fill_buffer": 0.0, }, }, False, ), Strategy( "eth_price_twap_mid", "15m", { "name": "rsi2-long-guarded-price-twap-o0.0010-0.0030-0.0050-v2-t160-l5.0-x55.0-sl0.008-mh48", "spec": { "trend_sma": 160, "rsi_threshold": 5.0, "exit_rsi": 55.0, "stop_loss_pct": 0.008, "max_hold_bars": 48, "entry_offsets": (0.001, 0.003, 0.005), "entry_valid_bars": 2, "fill_buffer": 0.0, }, }, False, ), Strategy( "eth_rsi2_market", "15m", explore.build_rsi2_long_guarded_candidate(240, 5.0, 45.0, 0.006, 48), False, ), Strategy( "eth_btc_rsi_filter", "15m", explore.build_eth_btc_rsi_filter_candidate(50, 3.0, 55.0, 120, 240, 0.0), True, ), Strategy( "eth_btc_rsi_filter", "15m", explore.build_eth_btc_rsi_filter_candidate(50, 3.0, 55.0, 480, 240, 0.0), True, ), Strategy( "btc_lead_eth_lag_15m", "15m", explore.build_btc_lead_eth_lag_candidate(8, 0.018, 0.006, 8, 0.006, 0.018), True, ), Strategy( "btc_lead_eth_lag_15m", "15m", explore.build_btc_lead_eth_lag_candidate(16, 0.024, 0.006, 32, 0.006, 0.018), True, ), Strategy( "btc_lead_eth_lag_5m", "5m", explore.build_btc_lead_eth_lag_candidate(16, 0.012, 0.006, 32, 0.006, 0.018), True, ), Strategy( "btc_lead_eth_lag_5m", "5m", explore.build_btc_lead_eth_lag_candidate(16, 0.012, 0.006, 8, 0.006, 0.018), True, ), ] def load_candles(symbol: str, bar: str, years: float) -> list[explore.Candle]: candles, exhausted = explore.load_cached_candles(explore.CANDLE_CACHE_DIR, symbol, bar) if not candles: raise FileNotFoundError(f"missing cached candles for {symbol} {bar}") requested = explore.history_bars_for_years(bar, years) return candles[-requested:] if len(candles) > requested else candles def run_strategy(strategy: Strategy, data: dict[tuple[str, str], list[explore.Candle]]) -> explore.SegmentResult: eth = data[("ETH-USDT-SWAP", strategy.bar)] if isinstance(strategy.candidate, dict): return price_twap.run_price_twap_segment( candles=eth, spec=strategy.candidate["spec"], roundtrip_cost_on_margin=0.0, ) if strategy.pair: btc = data[("BTC-USDT-SWAP", strategy.bar)] eth, btc = explore.align_pair_candles(eth, btc) return strategy.candidate.run( eth_candles=eth, btc_candles=btc, leverage=explore.LEVERAGE, warmup_bars=strategy.candidate.warmup_bars, ) return strategy.candidate.run( candles=eth, leverage=explore.LEVERAGE, warmup_bars=strategy.candidate.warmup_bars, ) def daily_equity(frame: pd.DataFrame, start: pd.Timestamp, end: pd.Timestamp) -> pd.Series: series = frame.set_index("ts")["equity"].sort_index() index = pd.date_range(start.normalize(), end.normalize(), freq="1D", tz="UTC") return series.reindex(index.union(series.index)).sort_index().ffill().reindex(index).ffill() def metrics_from_daily_equity(series: pd.Series) -> dict[str, float]: years = (series.index[-1] - series.index[0]).total_seconds() / 86_400 / 365 total_return = float(series.iloc[-1] / series.iloc[0] - 1.0) annualized_return = (1.0 + total_return) ** (1.0 / years) - 1.0 if total_return > -1.0 and years > 0.0 else 0.0 max_drawdown = explore.max_drawdown_from_equity([float(value) for value in series]) returns = series.pct_change().dropna() daily_std = float(returns.std(ddof=1)) if len(returns) > 1 else 0.0 sharpe = float(returns.mean()) / daily_std * (365 ** 0.5) if daily_std else 0.0 return { "net_total_return": total_return, "net_annualized_return": annualized_return, "net_max_drawdown": max_drawdown, "net_calmar": annualized_return / max_drawdown if max_drawdown else 0.0, "net_sharpe_daily": sharpe, } def horizon_rows(name: str, series: pd.Series) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] end_time = series.index[-1] for label, offset in HORIZONS: cutoff = end_time - offset horizon = series[series.index >= cutoff] if len(horizon) < 2: horizon = series rows.append( { "portfolio": name, "horizon": label, "horizon_start": horizon.index[0].strftime("%Y-%m-%d"), "horizon_end": horizon.index[-1].strftime("%Y-%m-%d"), **metrics_from_daily_equity(horizon), } ) return rows def combine_daily_returns( *, name: str, legs: tuple[str, ...], mode: str, daily: dict[str, pd.Series], strategy_metrics: dict[str, dict[str, float]], ) -> pd.Series: returns = pd.DataFrame({leg: daily[leg].pct_change().fillna(0.0) for leg in legs}).dropna() if mode == "equal": weights = pd.Series(1.0 / len(legs), index=legs) else: raw = pd.Series({leg: 1.0 / max(strategy_metrics[leg]["net_max_drawdown"], 0.01) for leg in legs}) weights = raw / raw.sum() equity = explore.INITIAL_EQUITY * (1.0 + returns.mul(weights, axis=1).sum(axis=1)).cumprod() equity.name = name return equity def monthly_rows(portfolio: str, series: pd.Series) -> pd.DataFrame: monthly = series.resample("ME").last() frame = pd.DataFrame( { "portfolio": portfolio, "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 markdown_table(frame: pd.DataFrame) -> str: rows = [list(frame.columns), ["---" for _ in frame.columns]] rows.extend(frame.astype(object).where(pd.notna(frame), "").values.tolist()) return "\n".join("| " + " | ".join(format_cell(value) for value in row) + " |" for row in rows) def format_cell(value: object) -> str: if isinstance(value, float): return f"{value:.6g}" return str(value).replace("|", "\\|") def markdown_report( *, command: str, paths: list[Path], strategy_total: pd.DataFrame, portfolio_total: pd.DataFrame, horizon: pd.DataFrame, worst_months: pd.DataFrame, correlation: pd.DataFrame, ) -> str: primary_total = portfolio_total[portfolio_total["cost_model"] == PRIMARY_COST].copy() top = primary_total.head(10) deep = strategy_total[ (strategy_total["cost_model"] == PRIMARY_COST) & (strategy_total["family"] == "eth_price_twap_deep") ].iloc[0] best = top.iloc[0] recent = horizon[ (horizon["cost_model"] == PRIMARY_COST) & (horizon["portfolio"] == best["portfolio"]) ] recent_ok = bool((recent[recent["horizon"].isin(["6m", "3m"])]["net_total_return"] > 0.0).all()) worth_small = bool( best["net_max_drawdown"] < deep["net_max_drawdown"] and best["net_annualized_return"] > 0.0 and recent_ok ) lines = [ "# ETH strategy portfolio 10y exploration", "", f"Run command: `{command}`", "", "Output files:", *[f"- `{path}`" for path in paths], "", "Scope: requested 10 years from cached continuous OKX candles; actual coverage is shown in the CSV files.", f"Baseline: ETH price-TWAP deep maker_taker annualized {deep['net_annualized_return']:.4f}, max DD {deep['net_max_drawdown']:.4f}.", "", "## Top 10 maker_taker portfolios", "", markdown_table( top[ [ "portfolio", "mode", "leg_count", "legs", "net_annualized_return", "net_max_drawdown", "net_calmar", "net_sharpe_daily", "worst_month_return", "avg_pair_corr", "lower_dd_than_deep", ] ] ), "", "## Recent horizons for top portfolio", "", markdown_table( recent[ [ "horizon", "horizon_start", "horizon_end", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", ] ] ), "", "## Worst months", "", markdown_table(worst_months[worst_months["cost_model"] == PRIMARY_COST].head(10)), "", "## Strategy return correlation", "", markdown_table(correlation), "", "## Live small allocation judgment", "", ( "Yes: the best maker_taker portfolio is more suitable than ETH price-TWAP deep for a small live allocation under this test." if worth_small else "No: the best maker_taker portfolio does not clear the drawdown plus recent-validity bar against ETH price-TWAP deep under this test." ), ] return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--years", type=float, default=YEARS) parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR) parser.add_argument("--max-leg-count", type=int, default=5) args = parser.parse_args() strategies = build_strategies() bars = sorted({strategy.bar for strategy in strategies}) data = { (symbol, bar): load_candles(symbol, bar, args.years) for bar in bars for symbol in ("ETH-USDT-SWAP", "BTC-USDT-SWAP") } results: dict[str, tuple[Strategy, explore.SegmentResult]] = {} for index, strategy in enumerate(strategies, start=1): key = f"{strategy.family}:{strategy.bar}:{twap_candidate_name(strategy.candidate)}" results[key] = (strategy, run_strategy(strategy, data)) print(f"done {index}/{len(strategies)} {key}") start = max( pd.to_datetime(result.equity_curve[0]["ts"], unit="ms", utc=True) for _, result in results.values() ) end = min( pd.to_datetime(result.equity_curve[-1]["ts"], unit="ms", utc=True) for _, result in results.values() ) strategy_rows: list[dict[str, object]] = [] daily_by_cost: dict[str, dict[str, pd.Series]] = {cost: {} for cost in COSTS} metrics_by_cost: dict[str, dict[str, dict[str, float]]] = {cost: {} for cost in COSTS} for key, (strategy, result) in results.items(): for cost_name, cost_value in COSTS.items(): frame = explore.cost_adjusted_trade_equity_frame(result, cost_value) daily = daily_equity(frame, start, end) metrics = metrics_from_daily_equity(daily) daily_by_cost[cost_name][key] = daily metrics_by_cost[cost_name][key] = metrics strategy_rows.append( { "strategy_key": key, "family": strategy.family, "bar": strategy.bar, "name": twap_candidate_name(strategy.candidate), "cost_model": cost_name, "roundtrip_cost_on_margin": cost_value, "first_candle": start.strftime("%Y-%m-%d %H:%M"), "last_candle": end.strftime("%Y-%m-%d %H:%M"), "years": (end - start).total_seconds() / 86_400 / 365, "trades": result.trade_count, "gross_total_return": result.total_return, "gross_max_drawdown_mark_to_market": result.max_drawdown, **metrics, } ) strategy_total = pd.DataFrame(strategy_rows) portfolio_rows: list[dict[str, object]] = [] horizon_output: list[dict[str, object]] = [] equity_frames: list[pd.DataFrame] = [] month_frames: list[pd.DataFrame] = [] keys = list(results.keys()) deep_key = next(key for key, (strategy, _) in results.items() if strategy.family == "eth_price_twap_deep") combo_index = 0 for cost_name, daily in daily_by_cost.items(): deep_dd = metrics_by_cost[cost_name][deep_key]["net_max_drawdown"] for leg_count in range(2, min(args.max_leg_count, len(keys)) + 1): for legs in combinations(keys, leg_count): if len({results[leg][0].family for leg in legs}) != leg_count: continue for mode in ("equal", "risk"): combo_index += 1 portfolio = f"{mode}-{leg_count}-c{combo_index:03d}-" + "+".join(results[leg][0].family for leg in legs) series = combine_daily_returns( name=portfolio, legs=legs, mode=mode, daily=daily, strategy_metrics=metrics_by_cost[cost_name], ) metrics = metrics_from_daily_equity(series) monthly = monthly_rows(portfolio, series) worst_month = float(monthly["return"].min()) returns = pd.DataFrame({leg: daily[leg].pct_change() for leg in legs}).dropna() corr = returns.corr() pair_corrs = [float(corr.loc[left, right]) for left, right in combinations(legs, 2)] avg_corr = float(pd.Series(pair_corrs).mean()) if pair_corrs else 0.0 portfolio_rows.append( { "portfolio": portfolio, "cost_model": cost_name, "mode": mode, "leg_count": leg_count, "legs": ";".join(legs), "first_candle": start.strftime("%Y-%m-%d %H:%M"), "last_candle": end.strftime("%Y-%m-%d %H:%M"), "years": (end - start).total_seconds() / 86_400 / 365, "worst_month_return": worst_month, "avg_pair_corr": avg_corr, "max_leg_drawdown": max(metrics_by_cost[cost_name][leg]["net_max_drawdown"] for leg in legs), "lower_dd_than_deep": metrics["net_max_drawdown"] < deep_dd, **metrics, } ) for row in horizon_rows(portfolio, series): horizon_output.append({"cost_model": cost_name, **row}) equity_frames.append( pd.DataFrame( { "portfolio": portfolio, "cost_model": cost_name, "date": series.index.strftime("%Y-%m-%d"), "equity": series.to_numpy(), } ) ) month_frames.append(monthly.assign(cost_model=cost_name)) portfolio_total = pd.DataFrame(portfolio_rows) portfolio_total = portfolio_total.sort_values( ["cost_model", "lower_dd_than_deep", "net_calmar", "net_annualized_return", "worst_month_return"], ascending=[True, False, False, False, False], ) primary = portfolio_total[portfolio_total["cost_model"] == PRIMARY_COST] other = portfolio_total[portfolio_total["cost_model"] != PRIMARY_COST] portfolio_total = pd.concat([primary, other], ignore_index=True) top_names = set(primary.head(10)["portfolio"]) horizon = pd.DataFrame(horizon_output) horizon["horizon"] = pd.Categorical(horizon["horizon"], categories=["3y", "1y", "6m", "3m"], ordered=True) horizon = horizon[horizon["portfolio"].isin(top_names)].sort_values(["cost_model", "portfolio", "horizon"]) equity = pd.concat(equity_frames, ignore_index=True) equity = equity[equity["portfolio"].isin(top_names)] monthly = pd.concat(month_frames, ignore_index=True) worst_months = monthly[monthly["portfolio"].isin(top_names)].sort_values("return").head(50) primary_daily = pd.DataFrame({key: daily_by_cost[PRIMARY_COST][key].pct_change() for key in keys}).dropna() correlation = primary_daily.corr().reset_index().rename(columns={"index": "strategy_key"}) args.output_dir.mkdir(parents=True, exist_ok=True) strategy_path = args.output_dir / "eth-strategy-portfolio-10y-strategies.csv" total_path = args.output_dir / "eth-strategy-portfolio-10y-total.csv" top10_path = args.output_dir / "eth-strategy-portfolio-10y-top10.csv" horizon_path = args.output_dir / "eth-strategy-portfolio-10y-horizon.csv" corr_path = args.output_dir / "eth-strategy-portfolio-10y-correlation.csv" worst_path = args.output_dir / "eth-strategy-portfolio-10y-worst-months.csv" equity_path = args.output_dir / "eth-strategy-portfolio-10y-equity.csv" report_path = args.output_dir / "eth-strategy-portfolio-10y-report.md" strategy_total.to_csv(strategy_path, index=False) portfolio_total.to_csv(total_path, index=False) primary.head(10).to_csv(top10_path, index=False) horizon.to_csv(horizon_path, index=False) correlation.to_csv(corr_path, index=False) worst_months.to_csv(worst_path, index=False) equity.to_csv(equity_path, index=False) command = f"rtk .venv/bin/python {Path(__file__).as_posix()} --years {args.years} --max-leg-count {args.max_leg_count}" report_path.write_text( markdown_report( command=command, paths=[strategy_path, total_path, top10_path, horizon_path, corr_path, worst_path, equity_path, report_path], strategy_total=strategy_total, portfolio_total=portfolio_total, horizon=horizon, worst_months=worst_months, correlation=correlation, ), encoding="utf-8", ) print(primary.head(10).to_string(index=False)) return 0 if __name__ == "__main__": raise SystemExit(main())