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