from __future__ import annotations import argparse from dataclasses import dataclass from pathlib import Path import pandas as pd DATA_DIR = Path("data/okx-candles") OUTPUT_DIR = Path("reports/eth-exploration") PREFIX = "eth-btc-wick-rejection" INITIAL_EQUITY = 10_000.0 FEE = 0.0004 ROUNDTRIP_FEE = FEE * 2 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 Spec: symbol: str bar: str vol_window: int min_vol_rank: float min_upper_wick: float max_close_pos: float stop: float take: float hold: int @property def name(self) -> str: base = self.symbol.split("-")[0].lower() return ( f"{base}-upper-wick-{self.bar}-vw{self.vol_window}-vr{self.min_vol_rank:g}" f"-uw{self.min_upper_wick:g}-cp{self.max_close_pos:g}" f"-sl{self.stop:g}-tp{self.take:g}-h{self.hold}" ) def load_frame(symbol: str) -> pd.DataFrame: frame = pd.read_csv(DATA_DIR / symbol / "15m.csv") frame["ts"] = pd.to_datetime(frame["ts"], unit="ms", utc=True) return frame.sort_values("ts").drop_duplicates("ts", keep="last").set_index("ts") def resample(frame: pd.DataFrame, bar: str) -> pd.DataFrame: rule = {"15m": "15min", "1h": "1h", "4h": "4h"}[bar] return ( frame.resample(rule, label="left", closed="left") .agg(open=("open", "first"), high=("high", "max"), low=("low", "min"), close=("close", "last"), volume=("volume", "sum")) .dropna() ) def signal(frame: pd.DataFrame, spec: Spec) -> pd.Series: candle_range = (frame["high"] - frame["low"]).replace(0, pd.NA) upper_wick = (frame["high"] - frame[["open", "close"]].max(axis=1)) / candle_range close_pos = (frame["close"] - frame["low"]) / candle_range vol_rank = frame["volume"].rolling(spec.vol_window).rank(pct=True) weak_body = frame["close"] <= frame["open"] entry = ( weak_body & (upper_wick >= spec.min_upper_wick) & (close_pos <= spec.max_close_pos) & (vol_rank >= spec.min_vol_rank) ) return entry.fillna(False) def trade_return(entry: float, exit_: float) -> float: return entry / exit_ - 1.0 - ROUNDTRIP_FEE def run_spec(frame: pd.DataFrame, spec: Spec) -> tuple[pd.Series, list[dict[str, object]]]: entry = signal(frame, spec) warmup = spec.vol_window + 2 equity = INITIAL_EQUITY position: dict[str, object] | None = None pending_entry = False trades: list[dict[str, object]] = [] curve: list[tuple[pd.Timestamp, float]] = [] rows = list(frame.itertuples()) for index in range(warmup, len(rows)): candle = rows[index] ts = frame.index[index] if pending_entry and position is None and equity > 0: position = { "entry_time": ts, "entry_index": index, "entry_price": float(candle.open), "stop": float(candle.open) * (1.0 + spec.stop), "take": float(candle.open) * (1.0 - spec.take), } pending_entry = False mark = equity if position is not None: stop_hit = candle.high >= float(position["stop"]) take_hit = candle.low <= float(position["take"]) held = index - int(position["entry_index"]) if stop_hit or take_hit or held >= spec.hold: exit_price = float(position["stop"] if stop_hit else position["take"] if take_hit else candle.close) net = trade_return(float(position["entry_price"]), exit_price) equity *= 1.0 + net trades.append({"entry_time": position["entry_time"], "exit_time": ts, "return": net}) position = None mark = equity else: gross = float(position["entry_price"]) / candle.close - 1.0 mark = equity * (1.0 + gross - FEE) curve.append((ts, mark)) if index == len(rows) - 1 or position is not None: continue if bool(entry.iloc[index]): pending_entry = True series = pd.Series({ts: value for ts, value in curve}).sort_index() daily = series.resample("1D").last().ffill() daily = pd.concat([pd.Series([INITIAL_EQUITY], index=[daily.index[0].normalize()]), daily]).sort_index() return daily.groupby(level=0).last(), trades def period_metrics(equity: pd.Series, trades: list[dict[str, object]], offset: pd.DateOffset | None) -> dict[str, object]: start = equity.index[0] if offset is None else equity.index[-1] - offset scoped = equity[equity.index >= start] scoped_trades = [trade for trade in trades if pd.Timestamp(trade["entry_time"]) >= scoped.index[0]] total = float(scoped.iloc[-1] / scoped.iloc[0] - 1.0) years = (scoped.index[-1] - scoped.index[0]).total_seconds() / 86_400 / 365 annual = (1.0 + total) ** (1.0 / years) - 1.0 if total > -1.0 and years > 0 else 0.0 drawdown = float(((scoped.cummax() - scoped) / scoped.cummax()).max()) 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] profit_factor = sum(wins) / abs(sum(losses)) if losses else (999.0 if wins else 0.0) return { "total_return": total, "annualized_return": annual, "max_drawdown": drawdown, "win_rate": len(wins) / len(returns) if returns else 0.0, "profit_factor": profit_factor, "trades": len(returns), } def monthly_rows(name: str, equity: pd.Series) -> pd.DataFrame: monthly = equity.resample("ME").last() frame = pd.DataFrame( { "name": name, "month": monthly.index.strftime("%Y-%m"), "start_equity": monthly.shift(1).fillna(equity.iloc[0]).to_numpy(), "end_equity": monthly.to_numpy(), } ) frame["return"] = frame["end_equity"] / frame["start_equity"] - 1.0 return frame def correlation_to_existing(candidate_monthly: pd.DataFrame) -> float | None: path = Path("reports/eth-exploration/eth-nextgen-micro-portfolio-monthly.csv") if not path.exists(): return None base = pd.read_csv(path) base = base[base["name"] == "equal-2-c0003"][["month", "return"]].rename(columns={"return": "base_return"}) joined = candidate_monthly[["month", "return"]].merge(base, on="month", how="inner") return None if len(joined) < 6 else float(joined["return"].corr(joined["base_return"])) def row_for_spec(spec: Spec, equity: pd.Series, trades: list[dict[str, object]]) -> dict[str, object]: row: dict[str, object] = {"name": spec.name, "symbol": spec.symbol, "bar": spec.bar} for label, offset in HORIZONS: metrics = period_metrics(equity, trades, offset) for key, value in metrics.items(): row[f"{label}_{key}"] = value return row def build_specs() -> list[Spec]: specs: list[Spec] = [] for symbol in ("ETH-USDT-SWAP", "BTC-USDT-SWAP"): for bar in ("15m", "1h", "4h"): holds = (12, 24) if bar == "15m" else (8, 16) for vol_window in (96, 192): for min_vol_rank in (0.90, 0.95): for min_upper_wick in (0.45, 0.60): for stop, take in ((0.012, 0.018), (0.02, 0.03)): for hold in holds: specs.append(Spec(symbol, bar, vol_window, min_vol_rank, min_upper_wick, 0.35, stop, take, hold)) return specs def markdown_table(frame: pd.DataFrame) -> str: def cell(value: object) -> str: return f"{value:.4f}" if isinstance(value, float) else str(value).replace("|", "\\|") rows = [list(frame.columns), ["---" for _ in frame.columns]] rows.extend(frame.astype(object).where(pd.notna(frame), "").values.tolist()) return "\n".join("| " + " | ".join(cell(value) for value in row) + " |" for row in rows) def report_text( totals: pd.DataFrame, selected: pd.Series | None, corr: float | None, paths: list[Path], min_full_trades: int, min_3m_trades: int, verdict: str, ) -> str: keep = [ "name", "symbol", "bar", "full_total_return", "full_annualized_return", "full_max_drawdown", "full_win_rate", "full_profit_factor", "full_trades", "3y_total_return", "1y_total_return", "6m_total_return", "3m_total_return", "3m_trades", ] if selected is None: period_table = "No reference candidate." else: period_table = markdown_table( pd.DataFrame( [ { "period": label, "total_return": selected[f"{label}_total_return"], "annualized_return": selected[f"{label}_annualized_return"], "max_drawdown": selected[f"{label}_max_drawdown"], "win_rate": selected[f"{label}_win_rate"], "profit_factor": selected[f"{label}_profit_factor"], "trades": selected[f"{label}_trades"], } for label, _ in HORIZONS ] ) ) corr_text = "n/a" if corr is None else f"{corr:.4f}" return "\n".join( [ "# ETH/BTC Wick Rejection Light Screen", "", "Scope: read-only local OKX candles; short-only single-candle upper-wick rejection; no staged entry, relative momentum, crash-follow, calendar/time bucket, trend-exhaustion, or false-breakout reversal.", "", f"Output files: {', '.join(f'`{path}`' for path in paths)}", f"Trade-count floor: full>={min_full_trades}, 3m>={min_3m_trades}.", f"Decision: {verdict}", f"Monthly return correlation vs `equal-2-c0003` nextgen micro portfolio: {corr_text}.", "", "## Reference Metrics", "", period_table, "", "## Top Candidates", "", markdown_table(totals[keep].head(12)), "", ] ) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR) parser.add_argument("--min-full-trades", type=int, default=100) parser.add_argument("--min-3m-trades", type=int, default=5) args = parser.parse_args() raw = {symbol: load_frame(symbol) for symbol in ("ETH-USDT-SWAP", "BTC-USDT-SWAP")} frames = {(symbol, bar): resample(raw[symbol], bar) for symbol in raw for bar in ("15m", "1h", "4h")} rows: list[dict[str, object]] = [] equity_by_name: dict[str, pd.Series] = {} for spec in build_specs(): equity, trades = run_spec(frames[(spec.symbol, spec.bar)], spec) rows.append(row_for_spec(spec, equity, trades)) equity_by_name[spec.name] = equity totals = pd.DataFrame(rows).sort_values( ["full_total_return", "1y_total_return", "full_profit_factor"], ascending=[False, False, False], ) trade_eligible = totals[(totals["full_trades"] >= args.min_full_trades) & (totals["3m_trades"] >= args.min_3m_trades)] viable = trade_eligible[ (trade_eligible["full_total_return"] > 0) & (trade_eligible["1y_total_return"] > 0) & (trade_eligible["6m_total_return"] > 0) & (trade_eligible["3m_total_return"] > 0) ] if not viable.empty: selected = viable.iloc[0] verdict = "Light-screen pass only: trade count and all required return windows are positive; edge still needs independent validation." elif not trade_eligible.empty: selected = trade_eligible.iloc[0] verdict = "Rejected: trade count is sufficient, but no trade-sufficient candidate has positive full/1y/6m/3m returns." else: selected = None verdict = "Rejected: no candidate met the trade-count floor." monthly = pd.DataFrame() corr = None if selected is not None: monthly = monthly_rows(str(selected["name"]), equity_by_name[str(selected["name"])]) corr = correlation_to_existing(monthly) args.output_dir.mkdir(parents=True, exist_ok=True) totals_path = args.output_dir / f"{PREFIX}-totals.csv" monthly_path = args.output_dir / f"{PREFIX}-monthly.csv" report_path = args.output_dir / f"{PREFIX}-report.md" totals.to_csv(totals_path, index=False) monthly.to_csv(monthly_path, index=False) report_path.write_text( report_text(totals, selected, corr, [totals_path, monthly_path, report_path], args.min_full_trades, args.min_3m_trades, verdict), encoding="utf-8", ) print(totals.head(10).to_string(index=False)) print(f"wrote {totals_path}, {monthly_path}, {report_path}") return 0 if __name__ == "__main__": raise SystemExit(main())