gavindiaz/ahad-quant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
ahad-quant/dca_bot.py
T

294 lines
12 KiB
Python
Raw Permalink Normal View History

Initial commit - AHAD QUANT v1
2026-06-25 14:00:20 +03:00
"""
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()
Reference in New Issue Copy Permalink