""" AHAD QUANT — DCA Bot (Dollar Cost Averaging with Safety Orders) 5 built-in strategies: classic, aggressive, safe, trend, reverse Usage (standalone): python dca_bot.py Or activated via .env: DCA_BOT_ENABLED=true DCA_PAIR=EURUSD DCA_STRATEGY=classic DCA_BASE_ORDER_USDT=100 DCA_SAFETY_ORDER_USDT=50 DCA_MAX_SAFETY_ORDERS=5 DCA_PRICE_DEVIATION=0.015 DCA_TAKE_PROFIT_PCT=0.02 How it works: 1. Opens a base order at market price 2. If price drops by DCA_PRICE_DEVIATION, adds a safety order (larger) 3. Safety orders scale up (each one buys more than the last) 4. Adjusts average entry price downward 5. Takes profit when price recovers to TP% above average entry 6. Repeats indefinitely """ import time, json, os import config try: from exchange_adapter import get_exchange, ExchangeAdapter HAS_ADAPTER = True except ImportError: HAS_ADAPTER = False STRATEGIES = { "classic": {"scale": 1.5, "dev_scale": 1.0, "desc": "Standard DCA"}, "aggressive": {"scale": 2.0, "dev_scale": 1.2, "desc": "Larger safety orders, faster recovery"}, "safe": {"scale": 1.2, "dev_scale": 0.8, "desc": "Smaller orders, more levels"}, "trend": {"scale": 1.5, "dev_scale": 1.0, "desc": "Only DCA in trend direction"}, "reverse": {"scale": 1.5, "dev_scale": 1.0, "desc": "Fades extremes — contrarian"}, } STATE_FILE = "dca_state.json" class DCABot: """ Dollar Cost Averaging bot with configurable safety orders. """ def __init__( self, exchange: object, coin: str = None, strategy: str = None, base_order_usdt: float = None, safety_order_usdt: float = None, max_safety_orders: int = None, price_deviation: float = None, take_profit_pct: float = None, ): self.exchange = exchange self.coin = coin or config.DCA_COIN self.strategy = strategy or config.DCA_STRATEGY self.base_order_usdt = base_order_usdt or config.DCA_BASE_ORDER_USDT self.safety_order_usdt= safety_order_usdt or config.DCA_SAFETY_ORDER_USDT self.max_safety = max_safety_orders or config.DCA_MAX_SAFETY_ORDERS self.price_deviation = price_deviation or config.DCA_PRICE_DEVIATION self.take_profit_pct = take_profit_pct or config.DCA_TAKE_PROFIT_PCT cfg = STRATEGIES.get(self.strategy, STRATEGIES["classic"]) self.scale_factor = cfg["scale"] # each safety order = prev × scale self.dev_scale = cfg["dev_scale"]# deviation multiplier per level # Active deal state self.active_deal: dict | None = None self.completed_deals: int = 0 self.total_pnl: float = 0.0 self.running: bool = False self._load_state() print(f"[DCA] Strategy: {self.strategy} — {cfg['desc']}") # ── Persistence ────────────────────────────────────────────────────────── def _load_state(self): if os.path.exists(STATE_FILE): with open(STATE_FILE) as f: s = json.load(f) self.active_deal = s.get("active_deal") self.completed_deals = s.get("completed_deals", 0) self.total_pnl = s.get("total_pnl", 0.0) if self.active_deal: print(f"[DCA] Resumed deal — avg entry: " f"{self.active_deal['avg_entry']:.4f} | " f"safety orders used: {self.active_deal['n_safety']}") def _save_state(self): with open(STATE_FILE, "w") as f: json.dump({ "active_deal": self.active_deal, "completed_deals": self.completed_deals, "total_pnl": self.total_pnl, }, f, indent=2) # ── Deal management ────────────────────────────────────────────────────── def _open_deal(self, price: float) -> bool: """Open a new DCA deal with the base order.""" qty = self.base_order_usdt / price try: result = self.exchange.place_market_order(self.coin, "buy", qty) if not result.get("success"): return False except Exception as e: print(f"[DCA] Failed to open deal: {e}") return False self.active_deal = { "base_price": price, "avg_entry": price, "total_qty": qty, "total_cost": self.base_order_usdt, "n_safety": 0, "next_so_price": price * (1 - self.price_deviation * self.dev_scale), "tp_price": price * (1 + self.take_profit_pct), } self._save_state() print(f"[DCA] Deal opened — {self.coin} @ {price:.4f} | " f"Qty: {qty:.4f} | TP: {self.active_deal['tp_price']:.4f}") return True def _add_safety_order(self, current_price: float, deal: dict) -> bool: """Add a safety order at current price.""" n = deal["n_safety"] + 1 # Safety order size scales up geometrically so_usdt = self.safety_order_usdt * (self.scale_factor ** (n - 1)) qty = so_usdt / current_price try: result = self.exchange.place_market_order(self.coin, "buy", qty) if not result.get("success"): return False except Exception as e: print(f"[DCA] Safety order failed: {e}") return False # Update deal state total_qty = deal["total_qty"] + qty total_cost = deal["total_cost"] + so_usdt avg_entry = total_cost / total_qty # Next SO price (increasing deviation per level) next_dev = self.price_deviation * self.dev_scale * (n + 1) next_so = avg_entry * (1 - next_dev) tp_price = avg_entry * (1 + self.take_profit_pct) deal.update({ "avg_entry": avg_entry, "total_qty": total_qty, "total_cost": total_cost, "n_safety": n, "next_so_price": next_so, "tp_price": tp_price, }) self._save_state() print(f"[DCA] Safety order #{n} — {self.coin} @ {current_price:.4f} | " f"Qty: {qty:.4f} | Avg entry: {avg_entry:.4f} | " f"TP now: {tp_price:.4f}") return True def _close_deal(self, current_price: float, deal: dict, reason: str = "tp") -> float: """Close the full DCA position and calculate PnL.""" qty = deal["total_qty"] try: self.exchange.place_market_order(self.coin, "sell", qty) except Exception as e: print(f"[DCA] Close failed: {e}") return 0.0 pnl = (current_price - deal["avg_entry"]) * qty pnl -= deal["total_cost"] * config.FEE_RATE * 2 # fees (config.FEE_RATE) self.total_pnl += pnl self.completed_deals += 1 self.active_deal = None self._save_state() sign = "+" if pnl >= 0 else "" print(f"[DCA] Deal #{self.completed_deals} closed ({reason}) — " f"PnL: {sign}{pnl:.2f} USD | Total PnL: {self.total_pnl:+.2f} USD") return pnl # ── Strategy-specific entry logic ───────────────────────────────────────── def _should_open_deal(self, current_price: float) -> bool: """Strategy-specific entry condition.""" if self.strategy == "trend": # Only open if price is above 20-period MA (uptrend) try: candles = self.exchange.get_candles(self.coin, "1h", 25) ma20 = sum(c["c"] for c in candles[-20:]) / 20 return current_price > ma20 except Exception: return True elif self.strategy == "reverse": # Only open if RSI is oversold (< 30) try: candles = self.exchange.get_candles(self.coin, "1h", 20) closes = [c["c"] for c in candles] changes = [closes[i] - closes[i-1] for i in range(1, len(closes))] gains = [max(c, 0) for c in changes] losses = [abs(min(c, 0)) for c in changes] avg_gain = sum(gains[-14:]) / 14 avg_loss = sum(losses[-14:]) / 14 rsi = 100 - (100 / (1 + avg_gain / avg_loss)) if avg_loss else 100 return rsi < 35 except Exception: return True return True # classic, aggressive, safe: always open # ── Main loop ───────────────────────────────────────────────────────────── def start(self): """Start the DCA bot main loop.""" print(f"\n[DCA] Starting DCA Bot — {self.coin} | Strategy: {self.strategy}") print(f" Base order: ${self.base_order_usdt} | " f"Safety order: ${self.safety_order_usdt} | " f"Max safety orders: {self.max_safety} | " f"Deviation: {self.price_deviation:.1%} | " f"TP: {self.take_profit_pct:.1%}") self.running = True while self.running: try: book = self.exchange.get_orderbook(self.coin) price = book["mid"] if self.active_deal is None: # No active deal — check if we should open one if self._should_open_deal(price): self._open_deal(price) else: print(f"[DCA] Waiting for entry signal — " f"{self.coin} @ {price:.4f}") else: deal = self.active_deal # Check TP if price >= deal["tp_price"]: self._close_deal(price, deal, "tp") time.sleep(5) continue # Check if safety order needed if (deal["n_safety"] < self.max_safety and price <= deal["next_so_price"]): self._add_safety_order(price, deal) else: # Status update unrealized = (price - deal["avg_entry"]) * deal["total_qty"] pct = (price - deal["avg_entry"]) / deal["avg_entry"] sign = "+" if pct >= 0 else "" print(f"[DCA] {self.coin} @ {price:.4f} | " f"Avg: {deal['avg_entry']:.4f} | " f"Unrealized: {sign}{unrealized:.2f} ({sign}{pct:.2%}) | " f"SO: {deal['n_safety']}/{self.max_safety} | " f"TP: {deal['tp_price']:.4f}") time.sleep(60) except KeyboardInterrupt: self.stop() break except Exception as e: print(f"[DCA] Error: {e}") time.sleep(15) def stop(self): print(f"\n[DCA] Stopped — Completed deals: {self.completed_deals} | " f"Total PnL: {self.total_pnl:+.2f} USD") self.running = False if __name__ == "__main__": if not HAS_ADAPTER: print("[ERROR] exchange_adapter.py not found") exit(1) exchange = get_exchange(config.EXCHANGE) exchange.connect() bot = DCABot(exchange) bot.start()