""" AHAD QUANT — Grid Trading Bot 5 built-in strategies: neutral, long, short, trend, reverse Usage (standalone): python grid_bot.py # paire par défaut : EURUSD Or activated via .env: GRID_BOT_ENABLED=true GRID_PAIR=EURUSD GRID_STRATEGY=neutral GRID_LEVELS=10 GRID_TOTAL_USDT=200 How it works: - Divides a price range into N equal levels - Places a buy order below current price and a sell order above - Each time a sell is filled, a new buy is placed below it - Each time a buy is filled, a new sell is placed above it - Profits from price oscillation within the range """ import time, json, os, math import config try: from exchange_adapter import get_exchange, ExchangeAdapter HAS_ADAPTER = True except ImportError: HAS_ADAPTER = False STATE_FILE = "grid_state.json" STRATEGIES = { "neutral": {"bias": 0.0, "desc": "Equal buys and sells — best for ranging markets"}, "long": {"bias": 0.3, "desc": "More buys than sells — bullish bias"}, "short": {"bias": -0.3, "desc": "More sells than buys — bearish bias"}, "trend": {"bias": 0.0, "desc": "Enters in trend direction, exits at reversal"}, "reverse": {"bias": 0.0, "desc": "Fades extreme moves — contrarian"}, } class GridBot: """ Perpetual futures grid bot with 5 configurable strategies. """ def __init__( self, exchange: object, coin: str = None, strategy: str = None, lower: float = 0, upper: float = 0, levels: int = None, total_usdt: float = None, leverage: int = None, ): self.exchange = exchange self.coin = coin or config.GRID_COIN self.strategy = strategy or config.GRID_STRATEGY self.n_levels = levels or config.GRID_LEVELS self.total_usdt = total_usdt or config.GRID_TOTAL_USDT self.leverage = leverage or config.GRID_LEVERAGE self._lower = lower self._upper = upper self.grid_prices: list[float] = [] self.orders: dict[str, dict] = {} # price → order info self.realized_pnl: float = 0.0 self.n_fills: int = 0 self.running: bool = False if self.strategy not in STRATEGIES: raise ValueError(f"Unknown strategy '{self.strategy}'. " f"Choose from: {list(STRATEGIES.keys())}") print(f"[GRID] Strategy: {self.strategy} — {STRATEGIES[self.strategy]['desc']}") # ── Grid calculation ────────────────────────────────────────────────────── def _auto_range(self, current_price: float) -> tuple[float, float]: """ Auto-detect grid range from recent ATR if lower/upper not specified. Uses 2x ATR above and below current price. """ try: candles = self.exchange.get_candles(self.coin, "1h", 50) highs = [c["h"] for c in candles] lows = [c["l"] for c in candles] closes= [c["c"] for c in candles] atr_vals = [] for i in range(1, len(closes)): tr = max(highs[i] - lows[i], abs(highs[i] - closes[i-1]), abs(lows[i] - closes[i-1])) atr_vals.append(tr) atr = sum(atr_vals[-14:]) / 14 if len(atr_vals) >= 14 else current_price * 0.02 except Exception: atr = current_price * 0.02 factor = 2.5 # grid spans ±2.5 ATR lower = current_price - factor * atr upper = current_price + factor * atr print(f"[GRID] Auto range — ATR: {atr:.4f} | " f"Lower: {lower:.4f} | Upper: {upper:.4f}") return lower, upper def _build_grid(self, current_price: float): lower = self._lower upper = self._upper if lower == 0 or upper == 0: lower, upper = self._auto_range(current_price) self.grid_prices = [ lower + i * (upper - lower) / (self.n_levels - 1) for i in range(self.n_levels) ] usdt_per_grid = self.total_usdt / self.n_levels self.qty_per_level = (usdt_per_grid * self.leverage) / current_price print(f"[GRID] {self.n_levels} levels | " f"{lower:.4f} → {upper:.4f} | " f"Qty/level: {self.qty_per_level:.4f} {self.coin} | " f"USDT/level: {usdt_per_grid:.2f}") # ── Strategy-specific order placement ──────────────────────────────────── def _should_buy_at(self, price: float, current_price: float) -> bool: """Decide whether to place a buy order at this grid level.""" if self.strategy == "neutral": return price < current_price elif self.strategy == "long": # More buy levels (lower 70% of grid) midpoint = self.grid_prices[int(self.n_levels * 0.3)] return price < max(current_price, midpoint) elif self.strategy == "short": # Fewer buy levels midpoint = self.grid_prices[int(self.n_levels * 0.7)] return price < min(current_price, midpoint) elif self.strategy in ("trend", "reverse"): return price < current_price return price < current_price def _should_sell_at(self, price: float, current_price: float) -> bool: return price > current_price # ── Order management ───────────────────────────────────────────────────── def _place_initial_orders(self, current_price: float): """Place initial grid orders around current price.""" print(f"[GRID] Placing initial orders...") placed = 0 for price in self.grid_prices: if abs(price - current_price) / current_price < 0.001: continue # skip levels too close to market price try: if self._should_buy_at(price, current_price): self.exchange.place_limit_order( self.coin, "buy", self.qty_per_level, price ) self.orders[f"buy_{price:.4f}"] = { "side": "buy", "price": price, "qty": self.qty_per_level, "status": "open" } placed += 1 elif self._should_sell_at(price, current_price): self.exchange.place_limit_order( self.coin, "sell", self.qty_per_level, price ) self.orders[f"sell_{price:.4f}"] = { "side": "sell", "price": price, "qty": self.qty_per_level, "status": "open" } placed += 1 except Exception as e: print(f"[GRID] Failed to place order at {price:.4f}: {e}") time.sleep(0.1) print(f"[GRID] {placed} orders placed") def _handle_fill(self, filled_order: dict, current_price: float): """When an order fills, place the opposite order on the other side.""" price = filled_order["price"] side = filled_order["side"] qty = filled_order["qty"] # Estimate PnL from the grid spread grid_step = (self.grid_prices[-1] - self.grid_prices[0]) / (self.n_levels - 1) trade_pnl = grid_step * qty * (1 if side == "sell" else -1) self.realized_pnl += trade_pnl self.n_fills += 1 # Place the opposite order try: if side == "buy": # Buy filled → place sell above new_price = price + grid_step if new_price <= self.grid_prices[-1]: self.exchange.place_limit_order( self.coin, "sell", qty, new_price ) self.orders[f"sell_{new_price:.4f}"] = { "side": "sell", "price": new_price, "qty": qty, "status": "open" } else: # Sell filled → place buy below new_price = price - grid_step if new_price >= self.grid_prices[0]: self.exchange.place_limit_order( self.coin, "buy", qty, new_price ) self.orders[f"buy_{new_price:.4f}"] = { "side": "buy", "price": new_price, "qty": qty, "status": "open" } except Exception as e: print(f"[GRID] Failed to place opposite order: {e}") sign = "+" if trade_pnl >= 0 else "" print(f"[GRID] Fill #{self.n_fills}: {side.upper()} {qty:.4f} {self.coin} " f"@ {price:.4f} | Grid PnL: {sign}{trade_pnl:.2f} | " f"Total: {self.realized_pnl:+.2f} USD") # ── Main loop ───────────────────────────────────────────────────────────── def start(self): """Start the grid bot main loop.""" print(f"\n[GRID] Starting Grid Bot — {self.coin} | Strategy: {self.strategy}") self.running = True # Set leverage try: self.exchange.set_leverage(self.coin, self.leverage) except Exception: pass # Get current price and build grid book = self.exchange.get_orderbook(self.coin) current_price = book["mid"] self._build_grid(current_price) self._place_initial_orders(current_price) print("[GRID] Running... Ctrl+C to stop\n") while self.running: try: # Check for filled orders (simplified polling) open_orders = self.exchange.get_open_orders(self.coin) open_ids = {o.get("id") for o in open_orders} for key, order in list(self.orders.items()): if order["status"] == "open": # Detect fill by checking if order disappeared from open orders book = self.exchange.get_orderbook(self.coin) current_price = book["mid"] self._handle_fill(order, current_price) self.orders[key]["status"] = "filled" # Status every 60s print(f"[GRID] {self.coin} @ {current_price:.4f} | " f"Fills: {self.n_fills} | PnL: {self.realized_pnl:+.2f} USD | " f"Open orders: {sum(1 for o in self.orders.values() if o['status']=='open')}") time.sleep(30) except KeyboardInterrupt: self.stop() break except Exception as e: print(f"[GRID] Error: {e}") time.sleep(10) def stop(self): """Cancel all open grid orders.""" print("\n[GRID] Stopping — cancelling all open orders...") self.running = False try: self.exchange.cancel_all_orders(self.coin) except Exception as e: print(f"[GRID] Cancel failed: {e}") print(f"[GRID] Final PnL: {self.realized_pnl:+.2f} USD | " f"Total fills: {self.n_fills}") if __name__ == "__main__": if not HAS_ADAPTER: print("[ERROR] exchange_adapter.py not found") exit(1) print("AHAD QUANT — Grid Bot") print(f" Coin: {config.GRID_COIN}") print(f" Strategy: {config.GRID_STRATEGY}") print(f" Levels: {config.GRID_LEVELS}") print(f" USDT: {config.GRID_TOTAL_USDT}") print(f" Leverage: {config.GRID_LEVERAGE}x\n") exchange = get_exchange(config.EXCHANGE) exchange.connect() bot = GridBot(exchange) bot.start()