145 lines
5.9 KiB
Python
Executable File
145 lines
5.9 KiB
Python
Executable File
import asyncio
|
|
import logging
|
|
import traceback
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
from src.strategies.dca import DCAStrategy
|
|
from src.bot.binance_client import BinanceClientWrapper
|
|
from src.bot.db import TradeDatabase
|
|
from src.integrations.telegram_notifier import TelegramNotifier
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class TradingEngine:
|
|
"""Core async trading engine for DCA bot."""
|
|
|
|
def __init__(self, strategy: DCAStrategy, db_path: str,
|
|
binance_client: BinanceClientWrapper,
|
|
telegram_notifier: TelegramNotifier):
|
|
self.strategy = strategy
|
|
self.db = TradeDatabase(db_path)
|
|
self.client = binance_client
|
|
self.telegram = telegram_notifier
|
|
self.is_running = False
|
|
self.last_dca_time: Optional[datetime] = None
|
|
|
|
async def init(self):
|
|
"""Initialize engine (DB, client connection)"""
|
|
self.db.init()
|
|
await self.client.connect()
|
|
logger.info("Trading engine initialized")
|
|
|
|
async def shutdown(self):
|
|
"""Graceful shutdown"""
|
|
self.is_running = False
|
|
await self.client.disconnect()
|
|
self.db.close()
|
|
logger.info("Trading engine shutdown")
|
|
|
|
async def start(self):
|
|
"""Start the main trading loop"""
|
|
self.is_running = True
|
|
logger.info(f"Trading engine started for {self.strategy.trading_pair}")
|
|
|
|
try:
|
|
while self.is_running:
|
|
await self._check_and_execute_dca()
|
|
await self._monitor_stop_losses()
|
|
await asyncio.sleep(30) # Check every 30 seconds
|
|
except Exception as e:
|
|
logger.error(f"Engine FATAL error: {e}")
|
|
logger.error(traceback.format_exc())
|
|
raise
|
|
|
|
async def _check_and_execute_dca(self):
|
|
"""Check if DCA trade should execute and place order"""
|
|
try:
|
|
# Check if interval has passed
|
|
if not self.strategy.should_execute_dca(self.last_dca_time):
|
|
return
|
|
|
|
logger.info("DCA interval reached - preparing order...")
|
|
|
|
# Get current price
|
|
current_price = await self._get_current_price()
|
|
logger.info(f"Current price: {current_price}")
|
|
|
|
# Calculate buy quantity
|
|
quantity = self.strategy.calculate_buy_quantity(current_price)
|
|
stop_loss = self.strategy.calculate_stop_loss_price(current_price)
|
|
|
|
logger.info(f"Placing order: {quantity} {self.strategy.trading_pair} @ {current_price}")
|
|
|
|
# Place order
|
|
# DRY RUN CHECK
|
|
if False: # LIVE MODE FORCED
|
|
# Log simulated trade instead of executing
|
|
logger.info(f"DRY RUN: Would place {quantity} {self.strategy.trading_pair} at {current_price}")
|
|
order = {"orderId": "DRY_RUN_" + str(int(datetime.utcnow().timestamp())), "status": "SIMULATED"}
|
|
else:
|
|
logger.info("Calling Binance API...")
|
|
order = await self.client.place_order(
|
|
symbol=self.strategy.trading_pair,
|
|
side="BUY",
|
|
quantity=quantity,
|
|
price=current_price
|
|
)
|
|
logger.info(f"Binance API Response: {order}")
|
|
|
|
# Store in DB
|
|
self.db.create_position(
|
|
symbol=self.strategy.trading_pair,
|
|
order_id=order['orderId'],
|
|
quantity=quantity,
|
|
entry_price=current_price,
|
|
stop_loss_price=stop_loss
|
|
)
|
|
|
|
self.last_dca_time = datetime.utcnow()
|
|
|
|
msg = f"✅ DCA Buy Order\nPair: {self.strategy.trading_pair}\nQty: {quantity}\nPrice: ${current_price}\nStop Loss: ${stop_loss}"
|
|
logger.info(msg)
|
|
await self.telegram.send_alert(msg)
|
|
logger.info("DCA execution complete")
|
|
|
|
except Exception as e:
|
|
logger.error(f"DCA execution error: {type(e).__name__}: {e}")
|
|
logger.error(traceback.format_exc())
|
|
await self.telegram.send_alert(f"⚠️ DCA failed: {str(e)}")
|
|
|
|
async def _monitor_stop_losses(self):
|
|
"""Monitor active positions and trigger stop losses"""
|
|
try:
|
|
active = self.db.get_active_positions()
|
|
|
|
for position in active:
|
|
current_price = await self._get_current_price()
|
|
|
|
if self._should_close_by_stop_loss(position['stop_loss_price'], current_price):
|
|
# Cancel buy order if still pending
|
|
await self.client.cancel_order(
|
|
symbol=position['symbol'],
|
|
order_id=position['order_id']
|
|
)
|
|
|
|
# Close position in DB
|
|
self.db.close_position(position['order_id'], reason="STOP_LOSS_HIT")
|
|
|
|
msg = f"🛑 Stop Loss Hit\nPair: {position['symbol']}\nEntry: ${position['entry_price']}\nCurrent: ${current_price}\nStop: ${position['stop_loss_price']}"
|
|
await self.telegram.send_alert(msg)
|
|
logger.warning(msg)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Stop loss monitoring error: {type(e).__name__}: {e}")
|
|
logger.error(traceback.format_exc())
|
|
|
|
def _should_close_by_stop_loss(self, stop_loss_price: float, current_price: float) -> bool:
|
|
"""Determine if stop loss should trigger"""
|
|
return current_price <= stop_loss_price
|
|
|
|
async def _get_current_price(self) -> float:
|
|
"""Fetch current BTC price"""
|
|
logger.debug(f"Fetching price for {self.strategy.trading_pair}...")
|
|
ticker = await self.client.client.get_symbol_ticker(symbol=self.strategy.trading_pair)
|
|
return float(ticker['price'])
|