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 okx_codex_trader.models import Candle from okx_codex_trader.sampled_report import SegmentResult, mark_to_market, trade_equity from scripts import explore_ultrashort as explore OUTPUT_DIR = Path("reports/eth-exploration") PREFIX = "eth-btc-nextgen" YEARS = 10.0 COSTS = { "maker_taker": 0.0021, "taker_taker": 0.0030, } PRIMARY_COST = "maker_taker" HORIZONS = ( ("full", None), ("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: explore.PairCandidate def close_trade( *, trades: list[dict[str, object]], exits: list[dict[str, object]], position: dict[str, object], candle: Candle, exit_price: float, leverage: int, ) -> tuple[float, bool]: exit_equity = trade_equity( side=str(position["side"]), margin_used=float(position["margin_used"]), entry_price=float(position["entry_price"]), exit_price=exit_price, leverage=leverage, ) pnl = exit_equity - float(position["margin_used"]) trades.append( { "side": "Long" if position["side"] == "long" else "Short", "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 / float(position["margin_used"]) * 100, 4), } ) exits.append({"ts": candle.ts, "price": exit_price, "side": position["side"]}) return exit_equity, pnl > 0.0 def run_btc_impulse_eth_follow_segment( *, eth_candles: list[Candle], btc_candles: list[Candle], leverage: int, warmup_bars: int, lookback: int, btc_threshold: float, eth_min_follow: float, stop_loss_pct: float, take_profit_pct: float, max_hold_bars: int, ) -> SegmentResult: 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_side: str | None = None pending_exit = False for index in range(warmup_bars, len(eth_candles)): candle = eth_candles[index] if pending_exit and position is not None: equity, won = close_trade( trades=trades, exits=exits, position=position, candle=candle, exit_price=candle.open, leverage=leverage, ) wins += 1 if won else 0 position = None pending_exit = False if pending_side is not None and position is None and equity > 0.0: position = { "side": pending_side, "entry_time": candle.ts, "entry_price": candle.open, "entry_index": index, "margin_used": equity, "stop_price": candle.open * (1.0 - stop_loss_pct if pending_side == "long" else 1.0 + stop_loss_pct), "take_profit_price": candle.open * (1.0 + take_profit_pct if pending_side == "long" else 1.0 - take_profit_pct), } entries.append({"ts": candle.ts, "price": candle.open, "side": pending_side}) pending_side = None current_equity = equity if position is not None: side = str(position["side"]) stop_hit = (side == "long" and candle.low <= float(position["stop_price"])) or ( side == "short" and candle.high >= float(position["stop_price"]) ) take_hit = (side == "long" and candle.high >= float(position["take_profit_price"])) or ( side == "short" and candle.low <= float(position["take_profit_price"]) ) if stop_hit or take_hit: exit_price = float(position["stop_price"] if stop_hit else position["take_profit_price"]) equity, won = close_trade( trades=trades, exits=exits, position=position, 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 = 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_candles) - 1 or equity <= 0.0: continue if position is not None: if index - int(position["entry_index"]) >= max_hold_bars: pending_exit = True continue btc_return = btc_candles[index].close / btc_candles[index - lookback].close - 1.0 eth_return = candle.close / eth_candles[index - lookback].close - 1.0 if btc_return >= btc_threshold and eth_return >= eth_min_follow: pending_side = "long" elif btc_return <= -btc_threshold and eth_return <= -eth_min_follow: pending_side = "short" trade_count = len(trades) return 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_candles[warmup_bars:], equity_curve=equity_curve, entries=entries, exits=exits, ) def run_btc_regime_ratio_revert_segment( *, eth_candles: list[Candle], btc_candles: list[Candle], leverage: int, warmup_bars: int, btc_trend_sma: int, ratio_length: int, ratio_z: float, stop_loss_pct: float, max_hold_bars: int, ) -> SegmentResult: eth_close = pd.Series([candle.close for candle in eth_candles], dtype=float) btc_close = pd.Series([candle.close for candle in btc_candles], dtype=float) btc_trend = btc_close.rolling(btc_trend_sma).mean().tolist() ratio = eth_close / btc_close ratio_mean = ratio.rolling(ratio_length).mean() ratio_std = ratio.rolling(ratio_length).std(ddof=0) zscore = ((ratio - ratio_mean) / ratio_std).tolist() 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_side: str | None = None pending_exit = False for index in range(warmup_bars, len(eth_candles)): candle = eth_candles[index] if pending_exit and position is not None: equity, won = close_trade( trades=trades, exits=exits, position=position, candle=candle, exit_price=candle.open, leverage=leverage, ) wins += 1 if won else 0 position = None pending_exit = False if pending_side is not None and position is None and equity > 0.0: position = { "side": pending_side, "entry_time": candle.ts, "entry_price": candle.open, "entry_index": index, "margin_used": equity, "stop_price": candle.open * (1.0 - stop_loss_pct if pending_side == "long" else 1.0 + stop_loss_pct), } entries.append({"ts": candle.ts, "price": candle.open, "side": pending_side}) pending_side = None current_equity = equity if position is not None: side = str(position["side"]) stop_hit = (side == "long" and candle.low <= float(position["stop_price"])) or ( side == "short" and candle.high >= float(position["stop_price"]) ) if stop_hit: equity, won = close_trade( trades=trades, exits=exits, position=position, candle=candle, exit_price=float(position["stop_price"]), leverage=leverage, ) wins += 1 if won else 0 current_equity = equity position = None if position is not None: current_equity = 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_candles) - 1 or equity <= 0.0: continue current_trend = btc_trend[index] current_z = zscore[index] if current_trend != current_trend or current_z != current_z: continue btc_up = btc_candles[index].close > float(current_trend) if position is not None: side = str(position["side"]) held = index - int(position["entry_index"]) if (side == "long" and current_z >= 0.0) or (side == "short" and current_z <= 0.0) or held >= max_hold_bars: pending_exit = True continue if btc_up and current_z <= -ratio_z: pending_side = "long" elif not btc_up and current_z >= ratio_z: pending_side = "short" trade_count = len(trades) return 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_candles[warmup_bars:], equity_curve=equity_curve, entries=entries, exits=exits, ) def build_btc_impulse_eth_follow_candidate( lookback: int, btc_threshold: float, eth_min_follow: float, stop_loss_pct: float, take_profit_pct: float, max_hold_bars: int, ) -> explore.PairCandidate: return explore.PairCandidate( f"btc-impulse-eth-follow-l{lookback}-b{btc_threshold}-e{eth_min_follow}-sl{stop_loss_pct}-tp{take_profit_pct}-mh{max_hold_bars}", lookback, lambda eth_candles, btc_candles, leverage, warmup_bars: run_btc_impulse_eth_follow_segment( eth_candles=eth_candles, btc_candles=btc_candles, leverage=leverage, warmup_bars=warmup_bars, lookback=lookback, btc_threshold=btc_threshold, eth_min_follow=eth_min_follow, stop_loss_pct=stop_loss_pct, take_profit_pct=take_profit_pct, max_hold_bars=max_hold_bars, ), ) def build_btc_regime_ratio_revert_candidate( btc_trend_sma: int, ratio_length: int, ratio_z: float, stop_loss_pct: float, max_hold_bars: int, ) -> explore.PairCandidate: return explore.PairCandidate( f"btc-regime-ratio-revert-t{btc_trend_sma}-r{ratio_length}-z{ratio_z}-sl{stop_loss_pct}-mh{max_hold_bars}", max(btc_trend_sma, ratio_length), lambda eth_candles, btc_candles, leverage, warmup_bars: run_btc_regime_ratio_revert_segment( eth_candles=eth_candles, btc_candles=btc_candles, leverage=leverage, warmup_bars=warmup_bars, btc_trend_sma=btc_trend_sma, ratio_length=ratio_length, ratio_z=ratio_z, stop_loss_pct=stop_loss_pct, max_hold_bars=max_hold_bars, ), ) def build_strategies() -> list[Strategy]: strategies: list[Strategy] = [] strategies.extend( Strategy( "btc_trend_eth_rsi", "15m", explore.build_eth_btc_rsi_filter_candidate(eth_trend, eth_rsi, exit_rsi, btc_trend, btc_momentum, btc_min_momentum), ) for eth_trend in (50, 120) for eth_rsi in (3.0, 5.0) for exit_rsi in (55.0,) for btc_trend in (240, 480) for btc_momentum in (96, 240) for btc_min_momentum in (0.0, 0.01) ) strategies.extend( Strategy( "btc_shock_guard_eth_rsi", "15m", explore.build_eth_btc_shock_filter_candidate( 50, 3.0, 55.0, btc_trend, btc_momentum, btc_min_momentum, shock_lookback, max_vol, max_dd, ), ) for btc_trend in (480,) for btc_momentum in (96, 240) for btc_min_momentum in (0.0, 0.01) for shock_lookback in (96, 240) for max_vol in (0.006, 0.010) for max_dd in (0.03, 0.05) ) strategies.extend( Strategy( "btc_lead_eth_lag", bar, explore.build_btc_lead_eth_lag_candidate(lookback, btc_threshold, lag_gap, max_hold, stop_loss, take_profit), ) for bar in ("5m", "15m") for lookback in (8, 16) for btc_threshold in ((0.010, 0.014) if bar == "5m" else (0.018, 0.024)) for lag_gap in (0.006, 0.010) for max_hold in (8, 32) for stop_loss in (0.006,) for take_profit in (0.012, 0.018) ) strategies.extend( Strategy( "btc_impulse_eth_follow", bar, build_btc_impulse_eth_follow_candidate(lookback, btc_threshold, eth_min_follow, stop_loss, take_profit, max_hold), ) for bar in ("5m", "15m") for lookback in (8, 16) for btc_threshold in ((0.008, 0.012) if bar == "5m" else (0.015, 0.020)) for eth_min_follow in (0.002, 0.006) for stop_loss in (0.006,) for take_profit in (0.012, 0.018) for max_hold in (8, 16, 32) ) strategies.extend( Strategy( "ethbtc_ratio_pullback", "15m", explore.build_eth_btc_ratio_pullback_candidate(480, btc_momentum, btc_min_momentum, ratio_length, ratio_std, ratio_rsi, stop_loss), ) for btc_momentum in (96, 240) for btc_min_momentum in (0.0, 0.01) for ratio_length in (48, 96) for ratio_std in (1.5, 2.0) for ratio_rsi in (5.0,) for stop_loss in (0.008,) ) strategies.extend( Strategy( "btc_regime_ratio_revert", "15m", build_btc_regime_ratio_revert_candidate(btc_trend, ratio_length, ratio_z, stop_loss, max_hold), ) for btc_trend in (240, 480) for ratio_length in (48, 96) for ratio_z in (1.5, 2.0) for stop_loss in (0.008,) for max_hold in (16, 32) ) return strategies def load_candles(symbol: str, bar: str, years: float) -> list[Candle]: candles, _ = 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[Candle]]) -> SegmentResult: eth, btc = explore.align_pair_candles( data[("ETH-USDT-SWAP", strategy.bar)], data[("BTC-USDT-SWAP", strategy.bar)], ) return strategy.candidate.run( eth_candles=eth, btc_candles=btc, 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 trade_stats(result: SegmentResult, roundtrip_cost_on_margin: float) -> dict[str, float]: returns = [ float(trade["return_pct"]) / 100.0 - roundtrip_cost_on_margin * float(trade.get("cost_weight", 1.0)) for trade in result.trades ] 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 = abs(sum(losses) / len(losses)) if losses else 0.0 gross_profit = sum(wins) gross_loss_abs = abs(sum(losses)) return { "win_rate": len(wins) / len(returns) if returns else 0.0, "payoff_ratio": avg_win / avg_loss_abs if avg_loss_abs else 0.0, "profit_factor": gross_profit / gross_loss_abs if gross_loss_abs else 0.0, "expectancy_per_trade": sum(returns) / len(returns) if returns else 0.0, } 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 risk_reward = 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, "risk_reward_ratio": risk_reward, } 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: horizon = series if offset is None else series[series.index >= end_time - offset] if len(horizon) < 2: horizon = series rows.append( { "name": 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 monthly_rows(name: str, series: pd.Series) -> pd.DataFrame: monthly = series.resample("ME").last() frame = pd.DataFrame( { "name": name, "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 portfolio_equity( *, name: str, legs: tuple[str, ...], mode: str, daily: dict[str, pd.Series], strategy_metrics: dict[str, dict[str, float]], ) -> tuple[pd.Series, 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, weights 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, monthly_summary: pd.DataFrame, worst_months: pd.DataFrame, ) -> str: primary_strategies = strategy_total[strategy_total["cost_model"] == PRIMARY_COST].head(10) primary_portfolios = portfolio_total[portfolio_total["cost_model"] == PRIMARY_COST].head(10) top_name = str(primary_portfolios.iloc[0]["name"]) if len(primary_portfolios) else "" top_horizon = horizon[(horizon["cost_model"] == PRIMARY_COST) & (horizon["name"] == top_name)] lines = [ "# ETH BTC nextgen non-maker exploration", "", f"Run command: `{command}`", "", "Output files:", *[f"- `{path}`" for path in paths], "", "Scope: ETH-only execution, BTC-driven signals, market/taker style fills. No maker-dependent TWAP legs are included.", "Costs: maker_taker=0.0021 and taker_taker=0.0030 roundtrip on margin at 3x.", "", "## Top maker_taker strategies", "", markdown_table( primary_strategies[ [ "strategy_key", "family", "bar", "trades", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", "win_rate", "payoff_ratio", "profit_factor", "risk_reward_ratio", "worst_month_return", ] ] ), "", "## Top maker_taker portfolios", "", markdown_table( primary_portfolios[ [ "name", "mode", "leg_count", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", "risk_reward_ratio", "worst_month_return", "min_horizon_total_return", "legs", ] ] ), "", "## Horizon metrics for top portfolio", "", markdown_table( top_horizon[ [ "horizon", "horizon_start", "horizon_end", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", "risk_reward_ratio", ] ] ), "", "## Monthly summary for top portfolios", "", markdown_table(monthly_summary[monthly_summary["cost_model"] == PRIMARY_COST].head(20)), "", "## Worst months", "", markdown_table(worst_months.head(20)), ] 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("--top-strategy-count", type=int, default=24) parser.add_argument("--max-leg-count", type=int, default=4) 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, SegmentResult]] = {} for index, strategy in enumerate(strategies, start=1): key = f"{strategy.family}:{strategy.bar}:{strategy.candidate.name}" results[key] = (strategy, run_strategy(strategy, data)) print(f"done {index}/{len(strategies)} {key}", flush=True) 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]] = [] horizon_output: list[dict[str, object]] = [] monthly_frames: list[pd.DataFrame] = [] 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_model, 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) monthly = monthly_rows(key, daily) stats = trade_stats(result, cost_value) daily_by_cost[cost_model][key] = daily metrics_by_cost[cost_model][key] = metrics strategy_rows.append( { "strategy_key": key, "cost_model": cost_model, "roundtrip_cost_on_margin": cost_value, "family": strategy.family, "bar": strategy.bar, "name": strategy.candidate.name, "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, "worst_month_return": float(monthly["return"].min()), **stats, **metrics, } ) for row in horizon_rows(key, daily): horizon_output.append({"kind": "strategy", "cost_model": cost_model, **row}) monthly_frames.append(monthly.assign(kind="strategy", cost_model=cost_model)) strategy_total = pd.DataFrame(strategy_rows).sort_values( ["cost_model", "net_calmar", "net_annualized_return", "net_max_drawdown"], ascending=[True, False, False, True], ) primary_strategy_keys = list(strategy_total[strategy_total["cost_model"] == PRIMARY_COST].head(args.top_strategy_count)["strategy_key"]) keys_by_family: dict[str, list[str]] = {} for key in primary_strategy_keys: keys_by_family.setdefault(results[key][0].family, []).append(key) selected_keys = [keys[0] for keys in keys_by_family.values()] for key in primary_strategy_keys: if key not in selected_keys: selected_keys.append(key) selected_keys = selected_keys[: args.top_strategy_count] portfolio_rows: list[dict[str, object]] = [] equity_frames: list[pd.DataFrame] = [] combo_index = 0 for cost_model, daily in daily_by_cost.items(): for leg_count in range(2, min(args.max_leg_count, len(selected_keys)) + 1): for legs in combinations(selected_keys, leg_count): if len({results[leg][0].family for leg in legs}) != leg_count: continue for mode in ("equal", "risk"): combo_index += 1 name = f"{mode}-{leg_count}-c{combo_index:04d}" series, weights = portfolio_equity( name=name, legs=legs, mode=mode, daily=daily, strategy_metrics=metrics_by_cost[cost_model], ) metrics = metrics_from_daily_equity(series) monthly = monthly_rows(name, series) current_horizons = horizon_rows(name, series) min_horizon_return = min(float(row["net_total_return"]) for row in current_horizons) portfolio_rows.append( { "name": name, "cost_model": cost_model, "roundtrip_cost_on_margin": COSTS[cost_model], "mode": mode, "leg_count": leg_count, "legs": ";".join(legs), "weights": ";".join(f"{leg}={weights[leg]:.8f}" for leg in 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, "trades": sum(results[leg][1].trade_count for leg in legs), "win_rate": float(pd.Series([trade_stats(results[leg][1], COSTS[cost_model])["win_rate"] for leg in legs]).mean()), "payoff_ratio": float(pd.Series([trade_stats(results[leg][1], COSTS[cost_model])["payoff_ratio"] for leg in legs]).mean()), "profit_factor": float(pd.Series([trade_stats(results[leg][1], COSTS[cost_model])["profit_factor"] for leg in legs]).mean()), "worst_month_return": float(monthly["return"].min()), "min_horizon_total_return": min_horizon_return, **metrics, } ) for row in current_horizons: horizon_output.append({"kind": "portfolio", "cost_model": cost_model, **row}) monthly_frames.append(monthly.assign(kind="portfolio", cost_model=cost_model)) equity_frames.append( pd.DataFrame( { "name": name, "cost_model": cost_model, "date": series.index.strftime("%Y-%m-%d"), "equity": series.to_numpy(), } ) ) portfolio_total = pd.DataFrame(portfolio_rows).sort_values( ["cost_model", "net_calmar", "net_annualized_return", "min_horizon_total_return", "net_max_drawdown"], ascending=[True, False, False, False, True], ) primary = portfolio_total[portfolio_total["cost_model"] == PRIMARY_COST] others = portfolio_total[portfolio_total["cost_model"] != PRIMARY_COST] portfolio_total = pd.concat([primary, others], ignore_index=True) top_names = set(primary.head(25)["name"]) horizon = pd.DataFrame(horizon_output) horizon["horizon"] = pd.Categorical(horizon["horizon"], categories=["full", "3y", "1y", "6m", "3m"], ordered=True) horizon = horizon[(horizon["kind"] == "strategy") | (horizon["name"].isin(top_names))].sort_values(["cost_model", "kind", "name", "horizon"]) monthly = pd.concat(monthly_frames, ignore_index=True) monthly_summary = ( monthly[monthly["name"].isin(top_names)] .groupby(["kind", "cost_model", "name"], as_index=False) .agg( months=("return", "count"), positive_month_rate=("return", lambda values: float((values > 0.0).mean())), avg_month_return=("return", "mean"), median_month_return=("return", "median"), worst_month_return=("return", "min"), best_month_return=("return", "max"), ) .sort_values(["cost_model", "kind", "worst_month_return"], ascending=[True, True, False]) ) worst_months = monthly[monthly["name"].isin(top_names)].sort_values("return").head(100) equity = pd.concat(equity_frames, ignore_index=True) equity = equity[equity["name"].isin(top_names)] args.output_dir.mkdir(parents=True, exist_ok=True) strategy_path = args.output_dir / f"{PREFIX}-strategies.csv" portfolio_path = args.output_dir / f"{PREFIX}-portfolios.csv" top_path = args.output_dir / f"{PREFIX}-top10.csv" horizon_path = args.output_dir / f"{PREFIX}-horizon.csv" monthly_path = args.output_dir / f"{PREFIX}-monthly-summary.csv" worst_path = args.output_dir / f"{PREFIX}-worst-months.csv" equity_path = args.output_dir / f"{PREFIX}-equity.csv" report_path = args.output_dir / f"{PREFIX}-report.md" strategy_total.to_csv(strategy_path, index=False) portfolio_total.to_csv(portfolio_path, index=False) primary.head(10).to_csv(top_path, index=False) horizon.to_csv(horizon_path, index=False) monthly_summary.to_csv(monthly_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} " f"--top-strategy-count {args.top_strategy_count} --max-leg-count {args.max_leg_count}" ) report_path.write_text( markdown_report( command=command, paths=[strategy_path, portfolio_path, top_path, horizon_path, monthly_path, worst_path, equity_path, report_path], strategy_total=strategy_total, portfolio_total=portfolio_total, horizon=horizon, monthly_summary=monthly_summary, worst_months=worst_months, ), encoding="utf-8", ) print(primary.head(10).to_string(index=False)) return 0 if __name__ == "__main__": raise SystemExit(main())