#!/usr/bin/env python3 from __future__ import annotations import argparse import json import sys from dataclasses import dataclass from itertools import product from pathlib import Path import pandas as pd ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from scripts import explore_ultrashort as explore ETH_SYMBOL = "ETH-USDT-SWAP" BTC_SYMBOL = "BTC-USDT-SWAP" BAR = "4H" YEARS = 10.0 OUTPUT_DIR = Path("reports/recent-regime") PREFIX = "regime-router-v2" PRIMARY_COST = "maker_taker" COSTS = { "maker_taker": 0.0021, "taker_taker": 0.0030, } HORIZONS = ( ("7d", pd.DateOffset(days=7)), ("14d", pd.DateOffset(days=14)), ("30d", pd.DateOffset(days=30)), ("90d", pd.DateOffset(days=90)), ("6m", pd.DateOffset(months=6)), ("1y", pd.DateOffset(years=1)), ("3y", pd.DateOffset(years=3)), ) @dataclass(frozen=True) class RouterSpec: name: str trend_sma: int btc_momentum_lookback: int eth_momentum_lookback: int vol_lookback: int corr_lookback: int ratio_lookback: int btc_trend_min: float btc_momentum_min: float eth_momentum_min: float max_btc_vol: float max_eth_vol: float min_corr: float ratio_z_entry: float stop_loss_pct: float take_profit_pct: float max_hold_bars: int def load_candles(symbol: str, bar: str, years: float) -> list[explore.Candle]: candles, _ = explore.load_cached_candles(explore.CANDLE_CACHE_DIR, symbol, bar) if not candles and bar in ("1H", "4H"): raw, _ = explore.load_cached_candles(explore.CANDLE_CACHE_DIR, symbol, "15m") candles = resample_candles(raw, symbol, {"1H": "1h", "4H": "4h"}[bar]) if not candles: raise FileNotFoundError(f"missing cached candles for {symbol} {bar}") requested = history_bars_for_years(bar, years) return candles[-requested:] if len(candles) > requested else candles def history_bars_for_years(bar: str, years: float) -> int: if bar == "1H": minutes = 60 elif bar == "4H": minutes = 240 elif bar.endswith("m"): minutes = int(bar[:-1]) else: raise ValueError(f"unsupported bar: {bar}") return int(years * explore.MINUTES_PER_YEAR / minutes) def resample_candles(candles: list[explore.Candle], symbol: str, rule: str) -> list[explore.Candle]: frame = pd.DataFrame( [ { "ts": pd.to_datetime(candle.ts, unit="ms", utc=True), "open": candle.open, "high": candle.high, "low": candle.low, "close": candle.close, "volume": candle.volume, } for candle in candles ] ).set_index("ts") out = frame.resample(rule, label="left", closed="left").agg( open=("open", "first"), high=("high", "max"), low=("low", "min"), close=("close", "last"), volume=("volume", "sum"), ).dropna() return [ explore.Candle( symbol=symbol, ts=int(index.timestamp() * 1000), open=float(row.open), high=float(row.high), low=float(row.low), close=float(row.close), volume=float(row.volume), ) for index, row in out.iterrows() ] def is_nan(value: float) -> bool: return value != value def exit_price_for_risk_hit(position: dict[str, object], candle: explore.Candle) -> float | None: side = str(position["side"]) stop_price = float(position["stop_price"]) take_profit_price = float(position["take_profit_price"]) if side == "long": if candle.open <= stop_price: return candle.open if candle.open >= take_profit_price: return candle.open if candle.low <= stop_price: return stop_price if candle.high >= take_profit_price: return take_profit_price else: if candle.open >= stop_price: return candle.open if candle.open <= take_profit_price: return candle.open if candle.high >= stop_price: return stop_price if candle.low <= take_profit_price: return take_profit_price return None def close_position( *, trades: list[dict[str, object]], exits: list[dict[str, object]], position: dict[str, object], account_equity: float, candle: explore.Candle, exit_price: float, leverage: int, ) -> tuple[float, bool]: margin_used = float(position["margin_used"]) exit_equity = explore.trade_equity( side=str(position["side"]), margin_used=margin_used, entry_price=float(position["entry_price"]), exit_price=exit_price, leverage=leverage, ) pnl = exit_equity - margin_used trades.append( { "side": "Long" if position["side"] == "long" else "Short", "regime": str(position["regime"]), "entry_time": explore._format_ts(int(position["entry_time"])), "exit_time": explore._format_ts(candle.ts), "entry_price": round(float(position["entry_price"]), 4), "exit_price": round(exit_price, 4), "pnl": round(pnl, 4), "return_pct": round(pnl / account_equity * 100.0, 6), "cost_weight": 1.0, } ) exits.append({"ts": candle.ts, "price": exit_price, "side": str(position["side"]), "regime": str(position["regime"])}) return account_equity + pnl, pnl > 0.0 def regime_side( *, index: int, eth_close: pd.Series, btc_close: pd.Series, eth_sma: pd.Series, btc_sma: pd.Series, eth_vol: pd.Series, btc_vol: pd.Series, corr: pd.Series, ratio_z: pd.Series, spec: RouterSpec, ) -> tuple[str, str]: values = ( eth_sma.iloc[index], btc_sma.iloc[index], eth_vol.iloc[index], btc_vol.iloc[index], corr.iloc[index], ratio_z.iloc[index], ) if any(is_nan(float(value)) for value in values): return "cash", "cash" btc_trend = btc_close.iloc[index] / btc_sma.iloc[index] - 1.0 eth_trend = eth_close.iloc[index] / eth_sma.iloc[index] - 1.0 btc_momentum = btc_close.iloc[index] / btc_close.iloc[index - spec.btc_momentum_lookback] - 1.0 eth_momentum = eth_close.iloc[index] / eth_close.iloc[index - spec.eth_momentum_lookback] - 1.0 calm = btc_vol.iloc[index] <= spec.max_btc_vol and eth_vol.iloc[index] <= spec.max_eth_vol coupled = corr.iloc[index] >= spec.min_corr if not calm or not coupled: return "cash", "cash" if ( btc_trend >= spec.btc_trend_min and btc_momentum >= spec.btc_momentum_min and eth_momentum >= spec.eth_momentum_min and ratio_z.iloc[index] <= spec.ratio_z_entry ): return "long", "btc_bull_eth_lag" if ( btc_trend <= -spec.btc_trend_min and btc_momentum <= -spec.btc_momentum_min and eth_momentum <= -spec.eth_momentum_min and ratio_z.iloc[index] >= -spec.ratio_z_entry ): return "short", "btc_bear_eth_lag" if abs(btc_trend) < spec.btc_trend_min and abs(eth_trend) < spec.btc_trend_min: return "cash", "weak_trend_cash" return "cash", "cash" def run_router_segment( *, eth: list[explore.Candle], btc: list[explore.Candle], spec: RouterSpec, leverage: int, ) -> explore.SegmentResult: eth_close = pd.Series([candle.close for candle in eth], dtype=float) btc_close = pd.Series([candle.close for candle in btc], dtype=float) eth_ret = eth_close.pct_change() btc_ret = btc_close.pct_change() ratio = eth_close / btc_close eth_sma = eth_close.rolling(spec.trend_sma).mean() btc_sma = btc_close.rolling(spec.trend_sma).mean() eth_vol = eth_ret.rolling(spec.vol_lookback).std(ddof=0) btc_vol = btc_ret.rolling(spec.vol_lookback).std(ddof=0) corr = eth_ret.rolling(spec.corr_lookback).corr(btc_ret) ratio_z = (ratio - ratio.rolling(spec.ratio_lookback).mean()) / ratio.rolling(spec.ratio_lookback).std(ddof=0) warmup = max( spec.trend_sma, spec.btc_momentum_lookback + 1, spec.eth_momentum_lookback + 1, spec.vol_lookback, spec.corr_lookback, spec.ratio_lookback, ) equity = explore.INITIAL_EQUITY ending_equity = equity peak_equity = equity max_drawdown = 0.0 wins = 0 trades: list[dict[str, object]] = [] entries: list[dict[str, object]] = [] exits: list[dict[str, object]] = [] equity_curve: list[dict[str, float | int]] = [] position: dict[str, object] | None = None pending_entry: tuple[str, str] | None = None pending_exit = False previous_signal_side = "cash" for index in range(warmup, len(eth)): candle = eth[index] if pending_exit and position is not None: equity, won = close_position( trades=trades, exits=exits, position=position, account_equity=equity, candle=candle, exit_price=candle.open, leverage=leverage, ) wins += 1 if won else 0 position = None pending_exit = False if pending_entry is not None and position is None and equity > 0.0: side, regime = pending_entry entry_price = candle.open position = { "side": side, "regime": regime, "entry_time": candle.ts, "entry_price": entry_price, "entry_index": index, "margin_used": equity, "stop_price": entry_price * (1.0 - spec.stop_loss_pct if side == "long" else 1.0 + spec.stop_loss_pct), "take_profit_price": entry_price * (1.0 + spec.take_profit_pct if side == "long" else 1.0 - spec.take_profit_pct), } entries.append({"ts": candle.ts, "price": entry_price, "side": side, "regime": regime}) pending_entry = None current_equity = equity if position is not None: exit_price = exit_price_for_risk_hit(position, candle) if exit_price is not None: equity, won = close_position( trades=trades, exits=exits, position=position, account_equity=equity, candle=candle, exit_price=exit_price, leverage=leverage, ) wins += 1 if won else 0 current_equity = equity position = None if position is not None: current_equity = explore.mark_to_market( side=str(position["side"]), margin_used=float(position["margin_used"]), entry_price=float(position["entry_price"]), mark_price=candle.close, leverage=leverage, ) peak_equity = max(peak_equity, current_equity) max_drawdown = max(max_drawdown, (peak_equity - current_equity) / peak_equity) equity_curve.append({"ts": candle.ts, "equity": current_equity, "close": candle.close}) ending_equity = current_equity if index == len(eth) - 1 or equity <= 0.0: continue side, regime = regime_side( index=index, eth_close=eth_close, btc_close=btc_close, eth_sma=eth_sma, btc_sma=btc_sma, eth_vol=eth_vol, btc_vol=btc_vol, corr=corr, ratio_z=ratio_z, spec=spec, ) if position is not None: held_bars = index - int(position["entry_index"]) if side == "cash" or side != position["side"] or held_bars >= spec.max_hold_bars: pending_exit = True previous_signal_side = side continue if side != "cash" and side != previous_signal_side: pending_entry = (side, regime) previous_signal_side = side trade_count = len(trades) return explore.SegmentResult( trade_count=trade_count, total_return=(ending_equity - explore.INITIAL_EQUITY) / explore.INITIAL_EQUITY, win_rate=wins / trade_count if trade_count else 0.0, max_drawdown=max_drawdown, trades=trades, open_position=position, candles=eth[warmup:], equity_curve=equity_curve, entries=entries, exits=exits, ) def specs() -> list[RouterSpec]: out: list[RouterSpec] = [] for trend_sma, momentum_lookback, max_vol, momentum_min, ratio_z_entry, max_hold in product( (60, 120), (18, 42), (0.030, 0.040), (0.012, 0.020), (0.25, 0.75), (12, 24), ): name = ( f"{PREFIX}-ts{trend_sma}-ml{momentum_lookback}-vol{max_vol:g}" f"-mom{momentum_min:g}-rz{ratio_z_entry:g}-h{max_hold}" ) out.append( RouterSpec( name=name, trend_sma=trend_sma, btc_momentum_lookback=momentum_lookback, eth_momentum_lookback=momentum_lookback // 2, vol_lookback=18, corr_lookback=42, ratio_lookback=42, btc_trend_min=0.008, btc_momentum_min=momentum_min, eth_momentum_min=momentum_min * 0.35, max_btc_vol=max_vol, max_eth_vol=max_vol * 1.35, min_corr=0.45, ratio_z_entry=ratio_z_entry, stop_loss_pct=0.010, take_profit_pct=0.018, max_hold_bars=max_hold, ) ) return out def cost_frame(result: explore.SegmentResult, cost: float, last_ts: int) -> pd.DataFrame: if not result.equity_curve: return pd.DataFrame([{"ts": pd.to_datetime(last_ts, unit="ms", utc=True), "equity": explore.INITIAL_EQUITY}]) rows = [{"ts": pd.to_datetime(result.equity_curve[0]["ts"], unit="ms", utc=True), "equity": explore.INITIAL_EQUITY}] equity = explore.INITIAL_EQUITY for trade in result.trades: equity *= 1.0 + float(trade["return_pct"]) / 100.0 - cost * float(trade.get("cost_weight", 1.0)) rows.append({"ts": pd.to_datetime(str(trade["exit_time"]), utc=True), "equity": equity}) end_time = pd.to_datetime(last_ts, unit="ms", utc=True) if pd.Timestamp(rows[-1]["ts"]) < end_time: rows.append({"ts": end_time, "equity": equity}) return pd.DataFrame(rows) def trade_stats(trades: list[dict[str, object]], cost: float, start: pd.Timestamp | None = None) -> dict[str, float | int]: scoped = [ trade for trade in trades if start is None or pd.to_datetime(str(trade["exit_time"]), utc=True) >= start ] returns = [float(trade["return_pct"]) / 100.0 - cost * float(trade.get("cost_weight", 1.0)) for trade in scoped] wins = [value for value in returns if value > 0.0] losses = [value for value in returns if value < 0.0] avg_win = sum(wins) / len(wins) if wins else 0.0 avg_loss = abs(sum(losses) / len(losses)) if losses else 0.0 gross_profit = sum(wins) gross_loss = abs(sum(losses)) return { "trades": len(returns), "win_rate": len(wins) / len(returns) if returns else 0.0, "profit_loss_ratio": avg_win / avg_loss if avg_loss else 0.0, "profit_factor": gross_profit / gross_loss if gross_loss else 0.0, } def horizon_frame(frame: pd.DataFrame, end_time: pd.Timestamp, offset: pd.DateOffset) -> tuple[pd.DataFrame, pd.Timestamp]: cutoff = end_time - offset before_cutoff = frame[frame["ts"] <= cutoff] if len(before_cutoff): start_equity = float(before_cutoff["equity"].iloc[-1]) after_cutoff = frame[frame["ts"] > cutoff] return ( pd.concat( [pd.DataFrame([{"ts": cutoff, "equity": start_equity}]), after_cutoff[["ts", "equity"]]], ignore_index=True, ), cutoff, ) return frame[["ts", "equity"]].copy(), pd.Timestamp(frame["ts"].iloc[0]) def horizon_rows(name: str, frame: pd.DataFrame, trades: list[dict[str, object]], cost: float, last_ts: int) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] end_time = pd.to_datetime(last_ts, unit="ms", utc=True) for label, offset in HORIZONS: current, start_time = horizon_frame(frame, end_time, offset) metrics = explore.annualized_metrics_from_equity(current, int(start_time.timestamp() * 1000), last_ts) rows.append( { "name": name, "horizon": label, "start": start_time.strftime("%Y-%m-%d %H:%M"), "end": end_time.strftime("%Y-%m-%d %H:%M"), "total_return": metrics["net_total_return"], "annualized_return": metrics["net_annualized_return"], "max_drawdown": metrics["net_max_drawdown"], "calmar": metrics["net_calmar"], **trade_stats(trades, cost, start_time), } ) return rows def regime_rows(name: str, trades: list[dict[str, object]], cost: float) -> list[dict[str, object]]: if not trades: return [{"name": name, "regime": "none", "trades": 0, "win_rate": 0.0, "profit_loss_ratio": 0.0, "profit_factor": 0.0}] rows: list[dict[str, object]] = [] for regime, group in pd.DataFrame(trades).groupby("regime"): rows.append({"name": name, "regime": regime, **trade_stats(group.to_dict("records"), cost)}) return rows def markdown_table(frame: pd.DataFrame) -> str: columns = list(frame.columns) rows = [columns, ["---" for _ in 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 report_text(command: str, output_files: list[Path], total: pd.DataFrame, horizon: pd.DataFrame, regime: pd.DataFrame) -> str: primary = total[total["cost_model"] == PRIMARY_COST].head(10) names = set(primary["name"]) horizon_top = horizon[(horizon["cost_model"] == PRIMARY_COST) & horizon["name"].isin(names)].copy() regime_top = regime[(regime["cost_model"] == PRIMARY_COST) & regime["name"].isin(names)].copy() lines = [ "# Recent BTC regime router v2", "", f"Run command: `{command}`", "", "Output files:", *[f"- `{path}`" for path in output_files], "", "BTC drives regime selection. ETH is the traded instrument. Router states are long, short, and cash; weak trend and low-volatility drag are explicitly routed to cash.", "", "## Top maker_taker routers", "", markdown_table( primary[ [ "name", "trades", "total_return", "annualized_return", "max_drawdown", "calmar", "win_rate", "profit_loss_ratio", "profit_factor", "min_recent_total_return", ] ] ), "", "## Required horizons", "", markdown_table( horizon_top[ [ "name", "horizon", "total_return", "annualized_return", "max_drawdown", "calmar", "trades", "win_rate", "profit_loss_ratio", "profit_factor", ] ] ), "", "## Regime split", "", markdown_table(regime_top[["name", "regime", "trades", "win_rate", "profit_loss_ratio", "profit_factor"]]), ] return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--bar", default=BAR) parser.add_argument("--years", type=float, default=YEARS) parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR) parser.add_argument("--max-candidates", type=int) args = parser.parse_args() eth_raw = load_candles(ETH_SYMBOL, args.bar, args.years) btc_raw = load_candles(BTC_SYMBOL, args.bar, args.years) eth, btc = explore.align_pair_candles(eth_raw, btc_raw) if not eth: raise RuntimeError("no aligned ETH/BTC candles") candidates = specs() if args.max_candidates is not None: candidates = candidates[: args.max_candidates] total_rows: list[dict[str, object]] = [] horizon_output: list[dict[str, object]] = [] regime_output: list[dict[str, object]] = [] for index, spec in enumerate(candidates, start=1): result = run_router_segment(eth=eth, btc=btc, spec=spec, leverage=explore.LEVERAGE) print(f"done {index}/{len(candidates)} {spec.name} trades={result.trade_count}", flush=True) for cost_model, cost in COSTS.items(): frame = cost_frame(result, cost, eth[-1].ts) start_ts = int(pd.Timestamp(frame["ts"].iloc[0]).timestamp() * 1000) end_ts = int(pd.Timestamp(frame["ts"].iloc[-1]).timestamp() * 1000) metrics = explore.annualized_metrics_from_equity(frame, start_ts, end_ts) current_horizons = horizon_rows(spec.name, frame, result.trades, cost, eth[-1].ts) min_recent = min(float(row["total_return"]) for row in current_horizons) total_rows.append( { "name": spec.name, "cost_model": cost_model, "symbol": ETH_SYMBOL, "signal_symbol": BTC_SYMBOL, "bar": args.bar, "first_candle": pd.Timestamp(frame["ts"].iloc[0]).strftime("%Y-%m-%d %H:%M"), "last_candle": pd.Timestamp(frame["ts"].iloc[-1]).strftime("%Y-%m-%d %H:%M"), "years": (end_ts - start_ts) / 86_400_000 / 365, "total_return": metrics["net_total_return"], "annualized_return": metrics["net_annualized_return"], "max_drawdown": metrics["net_max_drawdown"], "calmar": metrics["net_calmar"], "min_recent_total_return": min_recent, **trade_stats(result.trades, cost), **spec.__dict__, } ) for row in current_horizons: horizon_output.append({"cost_model": cost_model, **row}) for row in regime_rows(spec.name, result.trades, cost): regime_output.append({"cost_model": cost_model, **row}) total = pd.DataFrame(total_rows).sort_values( ["cost_model", "min_recent_total_return", "calmar", "annualized_return", "trades"], ascending=[True, False, False, False, True], ) horizon = pd.DataFrame(horizon_output) horizon["horizon"] = pd.Categorical(horizon["horizon"], categories=[label for label, _ in HORIZONS], ordered=True) horizon = horizon.sort_values(["cost_model", "name", "horizon"]) regime = pd.DataFrame(regime_output).sort_values(["cost_model", "name", "trades"], ascending=[True, True, False]) args.output_dir.mkdir(parents=True, exist_ok=True) total_path = args.output_dir / f"{PREFIX}-total.csv" horizon_path = args.output_dir / f"{PREFIX}-horizons.csv" regime_path = args.output_dir / f"{PREFIX}-regime.csv" top_path = args.output_dir / f"{PREFIX}-top10.csv" json_path = args.output_dir / f"{PREFIX}-summary.json" report_path = args.output_dir / f"{PREFIX}-report.md" total.to_csv(total_path, index=False) horizon.to_csv(horizon_path, index=False) regime.to_csv(regime_path, index=False) total[total["cost_model"] == PRIMARY_COST].head(10).to_csv(top_path, index=False) command = f"rtk .venv/bin/python {Path(__file__).as_posix()} --bar {args.bar} --years {args.years}" summary = { "report": PREFIX, "command": command, "primary_cost": PRIMARY_COST, "candidate_count": len(candidates), "horizons": [label for label, _ in HORIZONS], "top_maker_taker": total[total["cost_model"] == PRIMARY_COST].head(10).to_dict("records"), "output_files": [str(path) for path in [total_path, horizon_path, regime_path, top_path, json_path, report_path]], } json_path.write_text(json.dumps(summary, indent=2), encoding="utf-8") report_path.write_text( report_text(command, [total_path, horizon_path, regime_path, top_path, json_path, report_path], total, horizon, regime), encoding="utf-8", ) print(total[total["cost_model"] == PRIMARY_COST].head(10).to_string(index=False)) return 0 if __name__ == "__main__": raise SystemExit(main())