#!/usr/bin/env python3 """ BTC Reversal Bot v1 Paper trading and backtesting only. Does not place real orders. Core idea: Own BTC while short term momentum is confirmed upward. Move to USD when momentum is confirmed downward. Use hysteresis, confirmation candles, cooldown, fees, and a trend filter so the bot does not flip on every tiny price wiggle. """ from __future__ import annotations import argparse import csv import json import math import statistics import time import urllib.parse import urllib.request from dataclasses import dataclass, asdict from datetime import datetime, timezone, timedelta from typing import Iterable, List, Optional, Tuple @dataclass class Candle: ts: int low: float high: float open: float close: float volume: float @dataclass class Trade: ts: int side: str price: float fee: float usd_after: float btc_after: float reason: str @dataclass class Result: start_usd: float end_usd: float buy_hold_end_usd: float return_pct: float buy_hold_return_pct: float trades: int round_trips: int max_drawdown_pct: float fees_paid: float win_rate_pct: float params: dict def ema(values: List[float], period: int) -> List[float]: if period <= 1: return values[:] alpha = 2.0 / (period + 1.0) out = [values[0]] for v in values[1:]: out.append(alpha * v + (1.0 - alpha) * out[-1]) return out def rsi(values: List[float], period: int = 14) -> List[float]: out = [50.0] * len(values) if len(values) <= period: return out gains = [] losses = [] for i in range(1, period + 1): d = values[i] - values[i - 1] gains.append(max(d, 0.0)) losses.append(max(-d, 0.0)) avg_gain = sum(gains) / period avg_loss = sum(losses) / period out[period] = 100.0 if avg_loss == 0 else 100.0 - (100.0 / (1.0 + avg_gain / avg_loss)) for i in range(period + 1, len(values)): d = values[i] - values[i - 1] gain = max(d, 0.0) loss = max(-d, 0.0) avg_gain = ((avg_gain * (period - 1)) + gain) / period avg_loss = ((avg_loss * (period - 1)) + loss) / period out[i] = 100.0 if avg_loss == 0 else 100.0 - (100.0 / (1.0 + avg_gain / avg_loss)) return out def atr(candles: List[Candle], period: int = 14) -> List[float]: trs = [candles[0].high - candles[0].low] for i in range(1, len(candles)): c = candles[i] prev = candles[i - 1].close trs.append(max(c.high - c.low, abs(c.high - prev), abs(c.low - prev))) return ema(trs, period) def iso_utc(ts: int) -> str: return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() def fetch_coinbase_exchange_candles(days: int = 7, granularity: int = 60, product: str = "BTC-USD") -> List[Candle]: """Fetch public historical candles from Coinbase Exchange REST. Requests are chunked to stay under the historical candle limit. This endpoint is public and is used only for market data. """ end = datetime.now(timezone.utc).replace(second=0, microsecond=0) start = end - timedelta(days=days) chunk_seconds = granularity * 280 all_rows = {} cursor = start headers = {"User-Agent": "btc-reversal-paper-bot/1.0"} while cursor < end: chunk_end = min(cursor + timedelta(seconds=chunk_seconds), end) params = urllib.parse.urlencode({ "start": cursor.isoformat(), "end": chunk_end.isoformat(), "granularity": granularity, }) url = f"https://api.exchange.coinbase.com/products/{product}/candles?{params}" req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, timeout=20) as resp: rows = json.loads(resp.read().decode("utf-8")) except Exception as exc: raise RuntimeError(f"Coinbase candle download failed: {exc}") from exc if isinstance(rows, dict): raise RuntimeError(f"Coinbase returned an error: {rows}") for row in rows: if len(row) >= 6: c = Candle(int(row[0]), float(row[1]), float(row[2]), float(row[3]), float(row[4]), float(row[5])) all_rows[c.ts] = c cursor = chunk_end time.sleep(0.08) candles = [all_rows[k] for k in sorted(all_rows)] if len(candles) < 50: raise RuntimeError("Not enough candles downloaded to run the strategy.") return candles def load_csv(path: str) -> List[Candle]: candles = [] with open(path, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: raw_ts = row.get("ts") or row.get("timestamp") or row.get("time") or row.get("date") if raw_ts is None: raise ValueError("CSV needs ts, timestamp, time, or date column") try: ts = int(float(raw_ts)) except ValueError: ts = int(datetime.fromisoformat(raw_ts.replace("Z", "+00:00")).timestamp()) candles.append(Candle( ts=ts, low=float(row["low"]), high=float(row["high"]), open=float(row["open"]), close=float(row["close"]), volume=float(row.get("volume", 0.0)), )) candles.sort(key=lambda c: c.ts) return candles def save_csv(candles: Iterable[Candle], path: str) -> None: with open(path, "w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(["timestamp", "open", "high", "low", "close", "volume"]) for c in candles: w.writerow([iso_utc(c.ts), c.open, c.high, c.low, c.close, c.volume]) def backtest(candles: List[Candle], start_usd: float = 100.0, fee_rate: float = 0.006, fast: int = 5, slow: int = 20, rsi_period: int = 14, buy_rsi: float = 52.0, sell_rsi: float = 48.0, confirm: int = 2, cooldown: int = 2, min_edge_pct: float = 0.10, volatility_scale: float = 0.10) -> Tuple[Result, List[Trade], List[float]]: if len(candles) < max(slow, rsi_period) + 5: raise ValueError("Not enough candles") closes = [c.close for c in candles] fast_e = ema(closes, fast) slow_e = ema(closes, slow) rsi_v = rsi(closes, rsi_period) atr_v = atr(candles, 14) usd = start_usd btc = 0.0 fees = 0.0 trades: List[Trade] = [] equity_curve: List[float] = [] last_trade_i = -10**9 up_count = 0 down_count = 0 entry_total_cost: Optional[float] = None completed_pnls: List[float] = [] warmup = max(slow, rsi_period, 14) + 2 for i, c in enumerate(candles): price = c.close equity_curve.append(usd + btc * price) if i < warmup: continue spread_pct = ((fast_e[i] - slow_e[i]) / slow_e[i]) * 100.0 fast_slope_pct = ((fast_e[i] - fast_e[i - 1]) / fast_e[i - 1]) * 100.0 atr_pct = (atr_v[i] / price) * 100.0 if price else 0.0 adaptive_edge = min_edge_pct + atr_pct * volatility_scale bullish = spread_pct > adaptive_edge and fast_slope_pct > 0 and rsi_v[i] >= buy_rsi bearish = spread_pct < -adaptive_edge and fast_slope_pct < 0 and rsi_v[i] <= sell_rsi up_count = up_count + 1 if bullish else 0 down_count = down_count + 1 if bearish else 0 can_trade = (i - last_trade_i) >= cooldown if btc == 0.0 and can_trade and up_count >= confirm: fee = usd * fee_rate spend = usd - fee btc = spend / price fees += fee usd = 0.0 entry_total_cost = spend + fee last_trade_i = i trades.append(Trade(c.ts, "BUY", price, fee, usd, btc, f"up reversal: ema gap {spread_pct:.3f}%, rsi {rsi_v[i]:.1f}")) down_count = 0 elif btc > 0.0 and can_trade and down_count >= confirm: gross = btc * price fee = gross * fee_rate usd = gross - fee fees += fee btc = 0.0 if entry_total_cost is not None: completed_pnls.append(usd - entry_total_cost) entry_total_cost = None last_trade_i = i trades.append(Trade(c.ts, "SELL", price, fee, usd, btc, f"down reversal: ema gap {spread_pct:.3f}%, rsi {rsi_v[i]:.1f}")) up_count = 0 end_price = candles[-1].close if btc > 0: gross = btc * end_price fee = gross * fee_rate usd = gross - fee fees += fee btc = 0.0 if entry_total_cost is not None: completed_pnls.append(usd - entry_total_cost) trades.append(Trade(candles[-1].ts, "SELL_END", end_price, fee, usd, btc, "end of backtest")) buy_hold_fee_in = start_usd * fee_rate buy_hold_btc = (start_usd - buy_hold_fee_in) / candles[0].close buy_hold_gross = buy_hold_btc * candles[-1].close buy_hold_fee_out = buy_hold_gross * fee_rate buy_hold_end = buy_hold_gross - buy_hold_fee_out peak = equity_curve[0] max_dd = 0.0 for e in equity_curve: peak = max(peak, e) if peak > 0: max_dd = max(max_dd, (peak - e) / peak * 100.0) wins = sum(1 for p in completed_pnls if p > 0) win_rate = (wins / len(completed_pnls) * 100.0) if completed_pnls else 0.0 result = Result( start_usd=start_usd, end_usd=usd, buy_hold_end_usd=buy_hold_end, return_pct=(usd / start_usd - 1.0) * 100.0, buy_hold_return_pct=(buy_hold_end / start_usd - 1.0) * 100.0, trades=len(trades), round_trips=len(completed_pnls), max_drawdown_pct=max_dd, fees_paid=fees, win_rate_pct=win_rate, params={ "fee_rate": fee_rate, "fast": fast, "slow": slow, "buy_rsi": buy_rsi, "sell_rsi": sell_rsi, "confirm": confirm, "cooldown": cooldown, "min_edge_pct": min_edge_pct, "volatility_scale": volatility_scale, }, ) return result, trades, equity_curve def optimize(candles: List[Candle], start_usd: float, fee_rate: float) -> List[Result]: results = [] for fast, slow in [(3, 12), (5, 20), (8, 30), (10, 40)]: for confirm in [1, 2, 3]: for edge in [0.03, 0.07, 0.12, 0.20]: for buy_rsi, sell_rsi in [(50, 50), (52, 48), (55, 45)]: r, _, _ = backtest( candles, start_usd=start_usd, fee_rate=fee_rate, fast=fast, slow=slow, confirm=confirm, min_edge_pct=edge, buy_rsi=buy_rsi, sell_rsi=sell_rsi, ) results.append(r) results.sort(key=lambda r: (r.end_usd, -r.max_drawdown_pct), reverse=True) return results def print_result(r: Result) -> None: print("\nBTC Reversal Bot v1") print("Paper trading and backtesting only") print(f"Starting balance: ${r.start_usd:,.2f}") print(f"Strategy ending value: ${r.end_usd:,.2f}") print(f"Strategy return: {r.return_pct:,.2f}%") print(f"Buy and hold ending: ${r.buy_hold_end_usd:,.2f}") print(f"Buy and hold return: {r.buy_hold_return_pct:,.2f}%") print(f"Trades: {r.trades}") print(f"Round trips: {r.round_trips}") print(f"Win rate: {r.win_rate_pct:,.1f}%") print(f"Fees paid: ${r.fees_paid:,.4f}") print(f"Max drawdown: {r.max_drawdown_pct:,.2f}%") print("Parameters:", json.dumps(r.params, indent=2)) def write_trades(trades: List[Trade], path: str) -> None: with open(path, "w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(["time_utc", "side", "price", "fee", "usd_after", "btc_after", "reason"]) for t in trades: w.writerow([iso_utc(t.ts), t.side, t.price, t.fee, t.usd_after, t.btc_after, t.reason]) def synthetic_demo() -> List[Candle]: candles = [] base = 100000.0 start = int((datetime.now(timezone.utc) - timedelta(days=3)).timestamp()) for i in range(3 * 24 * 60): wave = 1500 * math.sin(i / 90.0) + 500 * math.sin(i / 19.0) drift = i * 0.10 close = base + wave + drift prev = base + 1500 * math.sin((i - 1) / 90.0) + 500 * math.sin((i - 1) / 19.0) + (i - 1) * 0.10 if i else close op = prev high = max(op, close) + 50 low = min(op, close) - 50 candles.append(Candle(start + i * 60, low, high, op, close, 10 + 2 * math.sin(i / 7.0))) return candles def main() -> None: p = argparse.ArgumentParser(description="BTC reversal paper trader and backtester") p.add_argument("--days", type=int, default=7, help="days of BTC history to fetch") p.add_argument("--start", type=float, default=100.0, help="paper starting USD") p.add_argument("--fee", type=float, default=0.006, help="fee rate per trade, e.g. 0.006 = 0.6 percent") p.add_argument("--csv", help="use local OHLCV CSV instead of downloading") p.add_argument("--save-data", help="save downloaded candles to CSV") p.add_argument("--trades", default="trades.csv", help="trade log CSV output") p.add_argument("--optimize", action="store_true", help="test a parameter grid") p.add_argument("--demo", action="store_true", help="run on synthetic data without internet") args = p.parse_args() if args.demo: candles = synthetic_demo() elif args.csv: candles = load_csv(args.csv) else: print(f"Downloading {args.days} days of BTC-USD one minute candles...") candles = fetch_coinbase_exchange_candles(days=args.days) print(f"Loaded {len(candles):,} candles from {iso_utc(candles[0].ts)} to {iso_utc(candles[-1].ts)}") if args.save_data: save_csv(candles, args.save_data) if args.optimize: ranked = optimize(candles, args.start, args.fee) print("\nTop 10 parameter sets by ending balance") for idx, r in enumerate(ranked[:10], 1): print(f"{idx:2d}. ${r.end_usd:8.2f} | return {r.return_pct:7.2f}% | DD {r.max_drawdown_pct:6.2f}% | trades {r.trades:4d} | {r.params}") best = ranked[0] r, trades, _ = backtest(candles, start_usd=args.start, **{ "fee_rate": best.params["fee_rate"], "fast": best.params["fast"], "slow": best.params["slow"], "buy_rsi": best.params["buy_rsi"], "sell_rsi": best.params["sell_rsi"], "confirm": best.params["confirm"], "cooldown": best.params["cooldown"], "min_edge_pct": best.params["min_edge_pct"], "volatility_scale": best.params["volatility_scale"], }) else: r, trades, _ = backtest(candles, start_usd=args.start, fee_rate=args.fee) print_result(r) write_trades(trades, args.trades) print(f"Trade log written to {args.trades}") if __name__ == "__main__": main()