from __future__ import annotations from dataclasses import dataclass from math import sqrt from pathlib import Path import pandas as pd SYMBOLS = ("BTC-USDT-SWAP", "ETH-USDT-SWAP") BARS = ("3m", "5m", "15m") DATA_DIR = Path("data/okx-candles") OUT_DIR = Path("reports/short-bias") INITIAL_EQUITY = 10_000.0 LEVERAGE = 3.0 TAKER_FEE = 0.0004 MIN_TRADES = 30 BACKTEST_MONTHS = 36 @dataclass(frozen=True) class Candidate: family: str name: str warmup: int params: dict[str, float | int] @dataclass(frozen=True) class BacktestResult: equity: pd.DataFrame trades: pd.DataFrame def load_frame(symbol: str, bar: str) -> pd.DataFrame: frame = pd.read_csv(DATA_DIR / symbol / f"{bar}.csv") frame["dt"] = pd.to_datetime(frame["ts"], unit="ms", utc=True) frame = frame.sort_values("ts").drop_duplicates("ts", keep="last").reset_index(drop=True) cutoff = frame["dt"].iloc[-1] - pd.DateOffset(months=BACKTEST_MONTHS) return frame[frame["dt"] >= cutoff].reset_index(drop=True) def rsi(series: pd.Series, length: int) -> pd.Series: delta = series.diff() gain = delta.clip(lower=0.0).rolling(length).mean() loss = (-delta.clip(upper=0.0)).rolling(length).mean() rs = gain / loss return 100.0 - (100.0 / (1.0 + rs)) def build_candidates() -> list[Candidate]: candidates: list[Candidate] = [] for trend in (80, 160): for entry in (90, 94): for exit_rsi in (35,): for max_hold in (12, 32): params = {"trend": trend, "entry": entry, "exit": exit_rsi, "max_hold": max_hold} candidates.append(Candidate("rsi2_short", f"rsi2-short-t{trend}-e{entry}-x{exit_rsi}-h{max_hold}", trend + 3, params)) for lookback in (16, 32, 64): for stop, take in ((0.005, 0.010),): for max_hold in (12, 32): params = {"lookback": lookback, "stop": stop, "take": take, "max_hold": max_hold} candidates.append(Candidate("breakdown_short", f"breakdown-short-l{lookback}-sl{stop}-tp{take}-h{max_hold}", lookback + 1, params)) for window in (48, 96): for entry_z in (1.8, 2.2): for exit_z in (0.0,): for max_hold in (12, 32): params = {"window": window, "entry_z": entry_z, "exit_z": exit_z, "max_hold": max_hold} candidates.append(Candidate("vwap_revert_short", f"vwap-short-w{window}-e{entry_z}-x{exit_z}-h{max_hold}", window * 2, params)) for length in (40, 80): for std in (2.0,): for trend in (160,): for max_hold in (12, 32): params = {"length": length, "std": std, "trend": trend, "max_hold": max_hold} candidates.append(Candidate("bb_short", f"bb-short-l{length}-s{std}-t{trend}-h{max_hold}", max(length, trend) + 1, params)) return candidates def exit_equity(start_equity: float, entry: float, exit_price: float) -> tuple[float, float]: entry_notional = start_equity * LEVERAGE exit_notional = entry_notional * (exit_price / entry) pnl = start_equity * LEVERAGE * ((entry - exit_price) / entry) fees = TAKER_FEE * (entry_notional + exit_notional) return start_equity + pnl - fees, pnl - fees def mark_equity(start_equity: float, entry: float, mark: float) -> float: entry_fee = TAKER_FEE * start_equity * LEVERAGE pnl = start_equity * LEVERAGE * ((entry - mark) / entry) return start_equity + pnl - entry_fee def signal_columns(frame: pd.DataFrame, candidate: Candidate) -> tuple[pd.Series, pd.Series, pd.Series]: close = frame["close"] false = pd.Series(False, index=frame.index) if candidate.family == "rsi2_short": trend = close.rolling(int(candidate.params["trend"])).mean() value = rsi(close, 2) entry = (close < trend) & (value >= float(candidate.params["entry"])) exit_signal = value <= float(candidate.params["exit"]) return entry, exit_signal, false if candidate.family == "breakdown_short": lookback = int(candidate.params["lookback"]) low = frame["low"].shift(1).rolling(lookback).min() entry = close < low return entry, false, false if candidate.family == "vwap_revert_short": window = int(candidate.params["window"]) volume = frame["volume"] vwap = (close * volume).rolling(window).sum() / volume.rolling(window).sum() deviation = (close - vwap) / vwap zscore = deviation / deviation.rolling(window).std(ddof=0) entry = zscore >= float(candidate.params["entry_z"]) exit_signal = zscore <= float(candidate.params["exit_z"]) return entry, exit_signal, false length = int(candidate.params["length"]) middle = close.rolling(length).mean() upper = middle + float(candidate.params["std"]) * close.rolling(length).std(ddof=0) trend = close.rolling(int(candidate.params["trend"])).mean() entry = (close > upper) & (close < trend) exit_signal = close <= middle return entry, exit_signal, false def backtest(frame: pd.DataFrame, candidate: Candidate) -> BacktestResult: entry_signal, exit_signal, _ = signal_columns(frame, candidate) ts_values = frame["ts"].to_numpy() dt_values = frame["dt"].to_numpy() open_values = frame["open"].to_numpy(dtype=float) high_values = frame["high"].to_numpy(dtype=float) low_values = frame["low"].to_numpy(dtype=float) close_values = frame["close"].to_numpy(dtype=float) entry_values = entry_signal.fillna(False).to_numpy(dtype=bool) exit_values = exit_signal.fillna(False).to_numpy(dtype=bool) equity = INITIAL_EQUITY peak = equity equity_rows: list[dict[str, object]] = [] trade_rows: list[dict[str, object]] = [] position: dict[str, float | int] | None = None pending_entry = False pending_exit = False for index in range(candidate.warmup, len(frame)): if pending_exit and position is not None: new_equity, pnl = exit_equity(equity, float(position["entry"]), float(open_values[index])) trade_rows.append( { "entry_time": dt_values[int(position["entry_index"])], "exit_time": dt_values[index], "side": "short", "entry": float(position["entry"]), "exit": float(open_values[index]), "pnl": pnl, "return": new_equity / equity - 1.0, "bars": index - int(position["entry_index"]), } ) equity = new_equity position = None pending_exit = False if pending_entry and position is None and equity > 0.0: position = {"entry": float(open_values[index]), "entry_index": index} pending_entry = False current_equity = equity if position is not None: entry = float(position["entry"]) stop = float(candidate.params.get("stop", 0.0)) take = float(candidate.params.get("take", 0.0)) exit_price: float | None = None if stop and float(high_values[index]) >= entry * (1.0 + stop): exit_price = entry * (1.0 + stop) elif take and float(low_values[index]) <= entry * (1.0 - take): exit_price = entry * (1.0 - take) if exit_price is not None: new_equity, pnl = exit_equity(equity, entry, exit_price) trade_rows.append( { "entry_time": dt_values[int(position["entry_index"])], "exit_time": dt_values[index], "side": "short", "entry": entry, "exit": exit_price, "pnl": pnl, "return": new_equity / equity - 1.0, "bars": index - int(position["entry_index"]), } ) equity = new_equity current_equity = equity position = None else: current_equity = mark_equity(equity, entry, float(close_values[index])) peak = max(peak, current_equity) equity_rows.append({"ts": ts_values[index], "dt": dt_values[index], "equity": current_equity, "drawdown": (peak - current_equity) / peak}) if index == len(frame) - 1 or equity <= 0.0: continue if position is not None: held = index - int(position["entry_index"]) if bool(exit_values[index]) or held >= int(candidate.params["max_hold"]): pending_exit = True continue if bool(entry_values[index]): pending_entry = True return BacktestResult(pd.DataFrame(equity_rows), pd.DataFrame(trade_rows)) def annualized(total_return: float, start: pd.Timestamp, end: pd.Timestamp) -> float: years = (end - start).total_seconds() / 31_536_000 if years <= 0.0: return 0.0 if total_return <= -1.0: return -1.0 return (1.0 + total_return) ** (1.0 / years) - 1.0 def horizon_return(equity: pd.DataFrame, months: int) -> float: end = equity["dt"].iloc[-1] cutoff = end - pd.DateOffset(months=months) before = equity[equity["dt"] <= cutoff] start_equity = float(before["equity"].iloc[-1]) if len(before) else float(equity["equity"].iloc[0]) return float(equity["equity"].iloc[-1]) / start_equity - 1.0 def monthly_returns(equity: pd.DataFrame, key: dict[str, object]) -> pd.DataFrame: monthly = equity.set_index("dt")["equity"].resample("ME").last().ffill().pct_change().dropna() rows = [{**key, "month": index.strftime("%Y-%m"), "return": value} for index, value in monthly.items()] return pd.DataFrame(rows) def summarize(symbol: str, bar: str, candidate: Candidate, result: BacktestResult) -> dict[str, object]: equity = result.equity trades = result.trades total_return = float(equity["equity"].iloc[-1] / equity["equity"].iloc[0] - 1.0) ann = annualized(total_return, equity["dt"].iloc[0], equity["dt"].iloc[-1]) max_dd = float(equity["drawdown"].max()) wins = trades[trades["return"] > 0.0] if len(trades) else trades losses = trades[trades["return"] < 0.0] if len(trades) else trades gross_profit = float(wins["return"].sum()) if len(wins) else 0.0 gross_loss = abs(float(losses["return"].sum())) if len(losses) else 0.0 avg_win = float(wins["return"].mean()) if len(wins) else 0.0 avg_loss = abs(float(losses["return"].mean())) if len(losses) else 0.0 return { "symbol": symbol, "bar": bar, "family": candidate.family, "candidate": candidate.name, "total_return": total_return, "annualized_return": ann, "max_drawdown": max_dd, "calmar": ann / max_dd if max_dd else 0.0, "trades": int(len(trades)), "win_rate": float(len(wins) / len(trades)) if len(trades) else 0.0, "payoff_ratio": avg_win / avg_loss if avg_loss else 0.0, "profit_factor": gross_profit / gross_loss if gross_loss else 0.0, "short_trade_ratio": 1.0 if len(trades) else 0.0, "return_3y": horizon_return(equity, 36), "return_1y": horizon_return(equity, 12), "return_6m": horizon_return(equity, 6), "return_3m": horizon_return(equity, 3), "first_time": equity["dt"].iloc[0].strftime("%Y-%m-%d %H:%M"), "last_time": equity["dt"].iloc[-1].strftime("%Y-%m-%d %H:%M"), } def worth_label(row: pd.Series) -> str: if row["trades"] >= MIN_TRADES and row["total_return"] > 0.0 and row["calmar"] > 0.25 and row["return_1y"] > 0.0: return "continue" return "stop" def main() -> int: OUT_DIR.mkdir(parents=True, exist_ok=True) candidates = build_candidates() total_rows: list[dict[str, object]] = [] monthly_frames: list[pd.DataFrame] = [] for symbol in SYMBOLS: for bar in BARS: frame = load_frame(symbol, bar) for candidate in candidates: result = backtest(frame, candidate) if len(result.equity) == 0: continue row = summarize(symbol, bar, candidate, result) if row["trades"] < MIN_TRADES: continue total_rows.append(row) monthly_frames.append(monthly_returns(result.equity, {key: row[key] for key in ("symbol", "bar", "family", "candidate")})) totals = pd.DataFrame(total_rows) totals = totals.sort_values(["calmar", "annualized_return", "profit_factor"], ascending=False).reset_index(drop=True) totals["worth_continuing"] = totals.apply(worth_label, axis=1) top3 = totals.head(3).copy() monthly = pd.concat(monthly_frames, ignore_index=True) if monthly_frames else pd.DataFrame() top_monthly = monthly.merge(top3[["symbol", "bar", "family", "candidate"]], on=["symbol", "bar", "family", "candidate"], how="inner") totals.to_csv(OUT_DIR / "intraday-totals.csv", index=False) top3.to_csv(OUT_DIR / "intraday-top3.csv", index=False) monthly.to_csv(OUT_DIR / "intraday-monthly.csv", index=False) top_monthly.to_csv(OUT_DIR / "intraday-top3-monthly.csv", index=False) lines = [ "# Short-Bias Intraday Search", "", f"Universe: {', '.join(SYMBOLS)} on {', '.join(BARS)}. All candidates are short-only; short trade ratio is therefore 1.0 for traded rows.", f"Cost model: {TAKER_FEE:.2%} single-side taker fee on leveraged notional, charged on entry and exit; leverage {LEVERAGE:g}x.", "", "## Top 3", "", ] for index, row in top3.iterrows(): lines.append( f"{index + 1}. {row.symbol} {row.bar} {row.candidate}: total {row.total_return:.2%}, annualized {row.annualized_return:.2%}, " f"max DD {row.max_drawdown:.2%}, Calmar {row.calmar:.2f}, trades {int(row.trades)}, win {row.win_rate:.2%}, " f"PF {row.profit_factor:.2f}, 3y {row.return_3y:.2%}, 1y {row.return_1y:.2%}, 6m {row.return_6m:.2%}, 3m {row.return_3m:.2%}, " f"decision {row.worth_continuing}." ) lines.extend( [ "", "## Continue", "", "Continue only candidates marked `continue`; the rest fail the direct requirement of positive full-period and recent 1y net return with enough trades.", ] ) (OUT_DIR / "intraday-report.md").write_text("\n".join(lines) + "\n", encoding="utf-8") print(top3.to_string(index=False)) return 0 if __name__ == "__main__": raise SystemExit(main())