Abdoul Ahad Binizi
300 lines
12 KiB
Python
300 lines
12 KiB
Python
"""
|
|
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()
|