from __future__ import annotations import argparse import sys from pathlib import Path import pandas as pd sys.path.append(str(Path(__file__).resolve().parent)) from search_eth_bearish_price_proxy import ( # noqa: E402 BTC_SYMBOL, DATA_DIR, HORIZONS, SYMBOL, Spec, joined_frames, load_frame, markdown_table, period_metrics, resample, row_for_spec, run_spec, ) OUTPUT_DIR = Path("reports/eth-exploration") BASE_SPEC = Spec("crash_follow", "1H", 20, 120, 8, 0.035, 0.02, 0.06, 96, "btc_riskoff") def pct(value: object) -> object: return value if not isinstance(value, float) else f"{value:.2%}" def horizon_table(equity: pd.Series, trades: list[dict[str, object]]) -> pd.DataFrame: return pd.DataFrame( [ {"period": label, **period_metrics(equity, trades, offset)} for label, offset in HORIZONS ] ) def scoped_metrics(equity: pd.Series, trades: list[dict[str, object]], start: pd.Timestamp, end: pd.Timestamp) -> dict[str, object]: scoped = equity[(equity.index >= start) & (equity.index <= end)] scoped_trades = [ trade for trade in trades if start <= pd.Timestamp(trade["entry_time"]).normalize() <= end ] if len(scoped) < 2: return { "total_return": 0.0, "max_drawdown": 0.0, "win_rate": 0.0, "profit_factor": 0.0, "trades": 0, } returns = [float(trade["return"]) for trade in scoped_trades] wins = [value for value in returns if value > 0] losses = [value for value in returns if value < 0] return { "total_return": float(scoped.iloc[-1] / scoped.iloc[0] - 1), "max_drawdown": float(((scoped.cummax() - scoped) / scoped.cummax()).max()), "win_rate": len(wins) / len(returns) if returns else 0.0, "profit_factor": sum(wins) / abs(sum(losses)) if losses else (999.0 if wins else 0.0), "trades": len(returns), } def yearly_table(equity: pd.Series, trades: list[dict[str, object]]) -> pd.DataFrame: rows = [] for year, scoped in equity.groupby(equity.index.year): rows.append({"year": int(year), **scoped_metrics(equity, trades, scoped.index[0], scoped.index[-1])}) return pd.DataFrame(rows) def monthly_table(equity: pd.Series, trades: list[dict[str, object]]) -> pd.DataFrame: rows = [] months = equity.index.tz_convert(None).to_period("M") for month, scoped in equity.groupby(months): rows.append({"month": str(month), **scoped_metrics(equity, trades, scoped.index[0], scoped.index[-1])}) return pd.DataFrame(rows) def regime_labels(frame: pd.DataFrame) -> pd.DataFrame: daily_close = frame["close"].resample("1D").last().ffill() daily_return = daily_close.pct_change() trend_90d = daily_close / daily_close.shift(90) - 1 realized_vol_30d = daily_return.rolling(30).std() * (365 ** 0.5) vol_bucket = pd.qcut(realized_vol_30d.dropna(), 3, labels=["low_vol", "mid_vol", "high_vol"]) labels = pd.DataFrame(index=daily_close.index) labels["trend"] = "unclassified" labels.loc[trend_90d > 0, "trend"] = "bull" labels.loc[trend_90d <= 0, "trend"] = "bear" labels["volatility"] = "unclassified" labels.loc[vol_bucket.index, "volatility"] = vol_bucket.astype(str) return labels def trade_segment_table(trades: list[dict[str, object]], labels: pd.DataFrame, column: str) -> pd.DataFrame: rows = [] for name in sorted(labels[column].unique()): returns = [] for trade in trades: day = pd.Timestamp(trade["entry_time"]).normalize() if day in labels.index and labels.at[day, column] == name: returns.append(float(trade["return"])) wins = [value for value in returns if value > 0] losses = [value for value in returns if value < 0] rows.append( { "segment": name, "total_return": float(pd.Series([1 + value for value in returns]).prod() - 1) if returns else 0.0, "win_rate": len(wins) / len(returns) if returns else 0.0, "profit_factor": sum(wins) / abs(sum(losses)) if losses else (999.0 if wins else 0.0), "trades": len(returns), } ) return pd.DataFrame(rows) def worst_contiguous_months(months: pd.DataFrame) -> pd.DataFrame: returns = months[["month", "total_return"]].reset_index(drop=True) worst = None for start in range(len(returns)): compounded = 1.0 for end in range(start, len(returns)): compounded *= 1 + float(returns.at[end, "total_return"]) row = { "start_month": returns.at[start, "month"], "end_month": returns.at[end, "month"], "months": end - start + 1, "total_return": compounded - 1, } if worst is None or row["total_return"] < worst["total_return"]: worst = row streaks = [] start = None compounded = 1.0 for index, row in returns.iterrows(): value = float(row["total_return"]) if value < 0: start = index if start is None else start compounded *= 1 + value elif start is not None: streaks.append((start, index - 1, compounded - 1)) start = None compounded = 1.0 if start is not None: streaks.append((start, len(returns) - 1, compounded - 1)) longest = max(streaks, key=lambda item: (item[1] - item[0] + 1, -item[2])) if streaks else None rows = [{"type": "worst_any_span", **worst}] if worst else [] if longest: rows.append( { "type": "longest_losing_streak", "start_month": returns.at[longest[0], "month"], "end_month": returns.at[longest[1], "month"], "months": longest[1] - longest[0] + 1, "total_return": longest[2], } ) return pd.DataFrame(rows) def neighbor_specs() -> list[Spec]: specs = [] for threshold in (0.03, 0.035, 0.04): for stop in (0.015, 0.02, 0.025): for take in (0.05, 0.06, 0.07): for hold in (72, 96, 120): for gate in ("btc_riskoff", "eth_riskoff", "none"): specs.append( Spec( BASE_SPEC.family, BASE_SPEC.bar, BASE_SPEC.fast, BASE_SPEC.slow, BASE_SPEC.lookback, threshold, stop, take, hold, gate, ) ) return specs def stability_table(frame: pd.DataFrame) -> pd.DataFrame: rows = [] for spec in neighbor_specs(): equity, trades = run_spec(spec, frame) rows.append(row_for_spec(spec, equity, trades)) return pd.DataFrame(rows).sort_values( ["full_total_return", "full_profit_factor", "3m_total_return"], ascending=[False, False, False], ) def summarize_stability(stability: pd.DataFrame) -> pd.DataFrame: return pd.DataFrame( [ { "group": "all_neighbors", "count": len(stability), "positive_full": int((stability["full_total_return"] > 0).sum()), "positive_all_windows": int( ( (stability["full_total_return"] > 0) & (stability["3y_total_return"] > 0) & (stability["1y_total_return"] > 0) & (stability["6m_total_return"] > 0) & (stability["3m_total_return"] > 0) ).sum() ), "min_full_return": float(stability["full_total_return"].min()), "p10_full_return": float(stability["full_total_return"].quantile(0.10)), "median_full_return": float(stability["full_total_return"].median()), "min_full_pf": float(stability["full_profit_factor"].min()), "p10_full_pf": float(stability["full_profit_factor"].quantile(0.10)), "max_full_dd": float(stability["full_max_drawdown"].max()), "p90_full_dd": float(stability["full_max_drawdown"].quantile(0.90)), } ] ) def report( csv_path: Path, horizon: pd.DataFrame, yearly: pd.DataFrame, monthly: pd.DataFrame, trend: pd.DataFrame, volatility: pd.DataFrame, worst_months: pd.DataFrame, stability: pd.DataFrame, stability_summary: pd.DataFrame, ) -> str: keep = [ "name", "full_total_return", "full_max_drawdown", "full_profit_factor", "full_trades", "3y_total_return", "1y_total_return", "6m_total_return", "3m_total_return", ] base = stability[stability["name"] == BASE_SPEC.name].iloc[0] verdict = "reject as an independent candidate" reason = "full-sample max drawdown is 44.90% and the neighborhood left tail is structurally weak" return ( "# ETH Bearish Price-Proxy Candidate Stress\n\n" f"Candidate: `{BASE_SPEC.name}`\n\n" f"Scope: existing local OKX candles only under `{DATA_DIR}`; no live API and no order path.\n\n" f"Parameter-neighborhood CSV: `{csv_path}`\n\n" f"Conclusion: {verdict}; {reason}.\n\n" "## Full / Recent Windows\n\n" f"{markdown_table(horizon)}\n\n" "## Years\n\n" f"{markdown_table(yearly)}\n\n" "## Months\n\n" f"{markdown_table(monthly)}\n\n" "## Bull / Bear Segments\n\n" f"{markdown_table(trend)}\n\n" "## Volatility Segments\n\n" f"{markdown_table(volatility)}\n\n" "## Worst Consecutive Months\n\n" f"{markdown_table(worst_months)}\n\n" "## Parameter Neighborhood Stability\n\n" f"Base row: full return {base['full_total_return']:.4f}, full DD {base['full_max_drawdown']:.4f}, full PF {base['full_profit_factor']:.4f}.\n\n" f"{markdown_table(stability_summary)}\n\n" "Top neighbors:\n\n" f"{markdown_table(stability[keep].head(12))}\n\n" "Left-tail neighbors:\n\n" f"{markdown_table(stability[keep].tail(12))}\n" ) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR) args = parser.parse_args() eth_15m = load_frame(SYMBOL) btc_15m = load_frame(BTC_SYMBOL) frame = joined_frames(resample(eth_15m, BASE_SPEC.bar), resample(btc_15m, BASE_SPEC.bar)) equity, trades = run_spec(BASE_SPEC, frame) horizon = horizon_table(equity, trades) yearly = yearly_table(equity, trades) monthly = monthly_table(equity, trades) labels = regime_labels(frame) trend = trade_segment_table(trades, labels, "trend") volatility = trade_segment_table(trades, labels, "volatility") worst_months = worst_contiguous_months(monthly) stability = stability_table(frame) stability_summary = summarize_stability(stability) args.output_dir.mkdir(parents=True, exist_ok=True) csv_path = args.output_dir / "eth-bearish-price-proxy-candidate-stability.csv" report_path = args.output_dir / "eth-bearish-price-proxy-candidate-stress.md" stability.to_csv(csv_path, index=False) report_path.write_text( report(csv_path, horizon, yearly, monthly, trend, volatility, worst_months, stability, stability_summary), encoding="utf-8", ) print(f"wrote {csv_path} and {report_path}") return 0 if __name__ == "__main__": raise SystemExit(main())