from __future__ import annotations import argparse import json import sys from pathlib import Path import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts import search_eth_nextgen_micro_portfolio as base OUTPUT_DIR = Path("reports/eth-exploration") PREFIX = "eth-nextgen-micro-direction-b" PRIMARY_COST = "maker_taker" COST_MODELS = { "maker_taker": 0.0021, "taker_taker": 0.0030, } MICRO_NAMES = ( "atr-compress-expand-r96-q0.15-sl0.008-tp0.016-mf0.25-us", "atr-compress-expand-r96-q0.15-sl0.008-tp0.016-mf0.4-us", "atr-compress-expand-r48-q0.15-sl0.008-tp0.016-mf0.25-us", "atr-compress-expand-r48-q0.15-sl0.008-tp0.016-mf0.4-us", ) LOOKBACKS = (30, 60, 90) HORIZONS = ( ("full", None), ("3y", pd.DateOffset(years=3)), ("1y", pd.DateOffset(years=1)), ("6m", pd.DateOffset(months=6)), ("3m", pd.DateOffset(months=3)), ("30d", pd.DateOffset(days=30)), ("21d", pd.DateOffset(days=21)), ("14d", pd.DateOffset(days=14)), ) def short_micro_name(name: str) -> str: return base.short_micro_name(name) def metrics_from_daily(series: pd.Series) -> dict[str, float]: return base.metrics_from_daily(series) def trade_return_stats(trades: list[base.LegReturn]) -> dict[str, float]: values = [trade.value for trade in trades] wins = [value for value in values if value > 0.0] losses = [value for value in values if value < 0.0] gross_profit = sum(wins) gross_loss_abs = abs(sum(losses)) avg_win = gross_profit / len(wins) if wins else 0.0 avg_loss_abs = gross_loss_abs / len(losses) if losses else 0.0 return { "trades": len(values), "win_rate": len(wins) / len(values) if values else 0.0, "avg_return": sum(values) / len(values) if values 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, } def horizon_rows(name: str, cost_model: str, series: pd.Series, trades: list[base.LegReturn]) -> list[dict[str, object]]: end = series.index[-1] rows: list[dict[str, object]] = [] for label, offset in HORIZONS: horizon = series if offset is None else series[series.index >= end - offset] if len(horizon) < 2: horizon = series start = horizon.index[0] selected = [trade for trade in trades if trade.exit_date >= start] years = max((horizon.index[-1] - start).total_seconds() / 86_400 / 365, 1 / 365) rows.append( { "name": name, "cost_model": cost_model, "horizon": label, "horizon_start": start.strftime("%Y-%m-%d"), "horizon_end": horizon.index[-1].strftime("%Y-%m-%d"), "trades_per_year": len(selected) / years, **trade_return_stats(selected), **metrics_from_daily(horizon), } ) return rows def evaluate( *, name: str, cost_model: str, kind: str, micro_name: str, lookback_days: int, micro_weight: float, series: pd.Series, trades: list[base.LegReturn], ) -> tuple[dict[str, object], list[dict[str, object]]]: horizons = horizon_rows(name, cost_model, series, trades) full = next(row for row in horizons if row["horizon"] == "full") recent = [row for row in horizons if row["horizon"] != "full"] return ( { "name": name, "cost_model": cost_model, "kind": kind, "micro_name": micro_name, "lookback_days": lookback_days, "micro_weight": micro_weight, "min_recent_total_return": min(float(row["net_total_return"]) for row in recent), **{key: full[key] for key in ( "trades", "trades_per_year", "win_rate", "avg_return", "payoff_ratio", "profit_factor", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", )}, }, horizons, ) def build_direction_b( *, cost_model: str, nextgen_series: pd.Series, nextgen_trades: list[base.LegReturn], micro_candidates: dict[str, tuple[pd.Series, list[base.LegReturn]]], ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: nextgen_returns = nextgen_series.pct_change().fillna(0.0) rows: list[dict[str, object]] = [] horizon_rows_out: list[dict[str, object]] = [] equity_rows: list[pd.DataFrame] = [] baseline_row, baseline_horizons = evaluate( name=base.NEXTGEN_BASELINE, cost_model=cost_model, kind="baseline", micro_name="", lookback_days=0, micro_weight=0.0, series=nextgen_series, trades=nextgen_trades, ) rows.append(baseline_row) horizon_rows_out.extend(baseline_horizons) equity_rows.append(base.equity_frame(base.NEXTGEN_BASELINE, cost_model, nextgen_series)) for micro_name, (micro_series, micro_trades) in micro_candidates.items(): if micro_name not in MICRO_NAMES: continue micro_returns = micro_series.pct_change().fillna(0.0) for lookback in LOOKBACKS: nextgen_regime = nextgen_series / nextgen_series.shift(lookback) - 1.0 micro_regime = micro_series / micro_series.shift(lookback) - 1.0 active = ((nextgen_regime < 0.0) & (micro_regime > 0.0)).shift(1).fillna(False).astype(bool) switch_name = f"switch-l{lookback}-{short_micro_name(micro_name)}" switch_series = base.returns_to_equity(switch_name, nextgen_returns.where(~active, micro_returns)) switch_trades = base.combine_trade_returns( nextgen_returns=[row for row in nextgen_trades if not bool(active.reindex([row.exit_date]).fillna(False).iloc[0])], micro_returns=micro_trades, nextgen_weight=1.0, micro_weight=1.0, micro_mask=active, ) row, horizons = evaluate( name=switch_name, cost_model=cost_model, kind="recent_regime_switch", micro_name=micro_name, lookback_days=lookback, micro_weight=1.0, series=switch_series, trades=switch_trades, ) rows.append(row) horizon_rows_out.extend(horizons) equity_rows.append(base.equity_frame(switch_name, cost_model, switch_series)) for micro_weight in (0.25, 0.40): overlay_name = f"riskoff-overlay-l{lookback}-m{micro_weight:.2f}-{short_micro_name(micro_name)}" overlay_series = base.returns_to_equity(overlay_name, nextgen_returns + micro_returns.where(active, 0.0) * micro_weight) overlay_trades = base.combine_trade_returns( nextgen_returns=nextgen_trades, micro_returns=micro_trades, nextgen_weight=1.0, micro_weight=micro_weight, micro_mask=active, ) row, horizons = evaluate( name=overlay_name, cost_model=cost_model, kind="riskoff_overlay", micro_name=micro_name, lookback_days=lookback, micro_weight=micro_weight, series=overlay_series, trades=overlay_trades, ) rows.append(row) horizon_rows_out.extend(horizons) equity_rows.append(base.equity_frame(overlay_name, cost_model, overlay_series)) summary = pd.DataFrame(rows).sort_values( ["cost_model", "kind", "net_calmar", "min_recent_total_return", "net_total_return"], ascending=[True, True, False, False, False], ) horizons = pd.DataFrame(horizon_rows_out) horizons["horizon"] = pd.Categorical(horizons["horizon"], categories=[label for label, _ in HORIZONS], ordered=True) horizons = horizons.sort_values(["cost_model", "name", "horizon"]) equity = pd.concat(equity_rows, ignore_index=True) return summary, horizons, equity def markdown_table(frame: pd.DataFrame) -> str: return base.markdown_table(frame) def pct_columns(frame: pd.DataFrame, columns: tuple[str, ...]) -> pd.DataFrame: output = frame.copy() for column in columns: output[column] = output[column].map(lambda value: f"{float(value) * 100:.2f}%") return output def write_report(summary: pd.DataFrame, horizons: pd.DataFrame, output_files: list[Path], command: str) -> str: primary = summary[(summary["cost_model"] == PRIMARY_COST) & (summary["kind"] != "baseline")].copy() stress = summary[(summary["cost_model"] == "taker_taker") & (summary["kind"] != "baseline")].copy() top_primary = primary.sort_values(["kind", "net_calmar", "min_recent_total_return"], ascending=[True, False, False]).head(12) top_switch = primary[primary["kind"] == "recent_regime_switch"].sort_values(["net_calmar", "min_recent_total_return"], ascending=[False, False]).head(5) top_overlay = primary[primary["kind"] == "riskoff_overlay"].sort_values(["net_calmar", "min_recent_total_return"], ascending=[False, False]).head(8) taker_positive = stress[stress["net_total_return"] > 0.0].sort_values(["kind", "net_calmar"], ascending=[True, False]).head(10) best = top_switch.iloc[0] best_horizons = horizons[(horizons["cost_model"] == PRIMARY_COST) & (horizons["name"] == best["name"])].copy() frequency_ok = bool((best_horizons[best_horizons["horizon"].isin(["30d", "21d", "14d"])]["trades"] >= 2).all()) short_window_ok = bool((best_horizons[best_horizons["horizon"].isin(["30d", "21d", "14d"])]["net_total_return"] > 0.0).all()) stress_match = stress[stress["name"] == best["name"]] taker_ok = bool(len(stress_match) and float(stress_match.iloc[0]["net_total_return"]) > 0.0) recommendation = ( "建议作为当前 BB squeeze 的并行只读观察策略；不建议直接列为近期实盘候选。" if frequency_ok and taker_ok and not short_window_ok else "建议只读观察，不建议进入实盘候选。" ) lines = [ "# ETH nextgen + microstructure direction B", "", f"Run command: `{command}`", "", "Scope: lookback 30/60/90, selected ATR compression/expansion US-session micro legs, recent_regime_switch first and small riskoff_overlay checks second.", "Read-only local backtest only; no secrets, private exchange API, order path, executor, or remote service was touched.", "", "Output files:", *[f"- `{path}`" for path in output_files], "", "## Verdict", "", recommendation, f"Best switch candidate: `{best['name']}`. Recent 30d/21d/14d trade counts are {best_horizons[best_horizons['horizon'].isin(['30d', '21d', '14d'])]['trades'].astype(int).tolist()}; taker/taker full-sample positive: `{taker_ok}`; all short windows positive: `{short_window_ok}`.", "", "## Top maker_taker candidates", "", markdown_table( pct_columns( top_primary[ [ "name", "kind", "lookback_days", "micro_weight", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", "trades", "trades_per_year", "win_rate", "avg_return", "payoff_ratio", "profit_factor", "min_recent_total_return", ] ], ("net_total_return", "net_annualized_return", "net_max_drawdown", "win_rate", "avg_return", "min_recent_total_return"), ) ), "", "## Riskoff overlay checks", "", markdown_table( pct_columns( top_overlay[ [ "name", "lookback_days", "micro_weight", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", "trades", "trades_per_year", "win_rate", "avg_return", "payoff_ratio", "profit_factor", "min_recent_total_return", ] ], ("net_total_return", "net_annualized_return", "net_max_drawdown", "win_rate", "avg_return", "min_recent_total_return"), ) ), "", "## Taker/taker positive candidates", "", markdown_table( pct_columns( taker_positive[ [ "name", "kind", "lookback_days", "micro_weight", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", "trades", "trades_per_year", ] ], ("net_total_return", "net_annualized_return", "net_max_drawdown"), ) ), "", "## Best switch windows", "", markdown_table( pct_columns( best_horizons[ [ "horizon", "horizon_start", "horizon_end", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", "trades", "trades_per_year", "win_rate", "avg_return", "payoff_ratio", "profit_factor", ] ], ("net_total_return", "net_annualized_return", "net_max_drawdown", "win_rate", "avg_return"), ) ), ] return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR) args = parser.parse_args() existing_equity = pd.read_csv(args.output_dir / "eth-btc-nextgen-equity.csv") base_equity = existing_equity[(existing_equity["cost_model"] == PRIMARY_COST) & (existing_equity["name"] == base.NEXTGEN_BASELINE)] if base_equity.empty: raise KeyError(f"missing existing nextgen equity for {base.NEXTGEN_BASELINE}") index = pd.DatetimeIndex(pd.to_datetime(base_equity["date"], utc=True)) summary_frames: list[pd.DataFrame] = [] horizon_frames: list[pd.DataFrame] = [] equity_frames: list[pd.DataFrame] = [] for cost_model, roundtrip_cost in COST_MODELS.items(): nextgen_series, nextgen_trades = base.load_nextgen(index, roundtrip_cost) micro_candidates = base.load_micro_candidates(index, roundtrip_cost) summary, horizons, equity = build_direction_b( cost_model=cost_model, nextgen_series=nextgen_series, nextgen_trades=nextgen_trades, micro_candidates=micro_candidates, ) summary_frames.append(summary) horizon_frames.append(horizons) equity_frames.append(equity) summary = pd.concat(summary_frames, ignore_index=True) horizons = pd.concat(horizon_frames, ignore_index=True) equity = pd.concat(equity_frames, ignore_index=True) args.output_dir.mkdir(parents=True, exist_ok=True) summary_path = args.output_dir / f"{PREFIX}-summary.csv" horizon_path = args.output_dir / f"{PREFIX}-horizons.csv" equity_path = args.output_dir / f"{PREFIX}-equity.csv" json_path = args.output_dir / f"{PREFIX}-top.json" report_path = args.output_dir / f"{PREFIX}-report.md" output_files = [summary_path, horizon_path, equity_path, json_path, report_path] summary.to_csv(summary_path, index=False) horizons.to_csv(horizon_path, index=False) equity.to_csv(equity_path, index=False) primary = summary[(summary["cost_model"] == PRIMARY_COST) & (summary["kind"] != "baseline")] json_path.write_text(json.dumps(primary.sort_values(["kind", "net_calmar"], ascending=[True, False]).head(20).to_dict("records"), indent=2), encoding="utf-8") command = f"rtk .venv/bin/python {Path(__file__).as_posix()}" report_path.write_text(write_report(summary, horizons, output_files, command), encoding="utf-8") print(primary.sort_values(["kind", "net_calmar"], ascending=[True, False]).head(12).to_string(index=False)) return 0 if __name__ == "__main__": raise SystemExit(main())