from __future__ import annotations import asyncio import json import logging from contextlib import suppress from datetime import UTC, datetime from typing import Any from pybit.unified_trading import WebSocket from llm_trade_bot.agent import AgentRunner from llm_trade_bot.bybit_gateway import BybitGateway from llm_trade_bot.config import AppConfig from llm_trade_bot.database import Database from llm_trade_bot.scheduler import next_scheduled_time from llm_trade_bot.telegram_bot import TelegramBot LOGGER = logging.getLogger(__name__) class TradeBotService: def __init__(self, config: AppConfig): self.config = config self.database = Database(config.storage.database_path) self.bybit = BybitGateway(config.bybit, config.scanner) self.agent = AgentRunner(config, self.database, self.bybit) self.telegram = TelegramBot(config.telegram, self) self._run_lock = asyncio.Lock() self._paused = False self._tasks: set[asyncio.Task[Any]] = set() self._stopping = asyncio.Event() self._websocket: WebSocket | None = None self._loop: asyncio.AbstractEventLoop | None = None async def start(self) -> None: await self.database.open() self._paused = (await self.database.get_state("paused", "false")) == "true" await self.bybit.startup_check() await self.telegram.start() self._loop = asyncio.get_running_loop() self._start_private_websocket() self._spawn(self._scheduler_loop(), "scheduler") self._spawn(self._outbox_loop(), "outbox") self._spawn(self._reconcile_loop(), "reconciler") await self.database.queue_message("🤖 LLM Trade Bot запущен в режиме Bybit Demo.") async def stop(self) -> None: self._stopping.set() for task in self._tasks: task.cancel() if self._tasks: await asyncio.gather(*self._tasks, return_exceptions=True) if self._websocket is not None: with suppress(Exception): self._websocket.exit() with suppress(Exception): await self.telegram.stop() await self.database.close() async def wait(self) -> None: await self._stopping.wait() def _spawn(self, awaitable: Any, name: str) -> None: task = asyncio.create_task(awaitable, name=name) self._tasks.add(task) task.add_done_callback(self._task_done) def _task_done(self, task: asyncio.Task[Any]) -> None: self._tasks.discard(task) if not task.cancelled() and (error := task.exception()) is not None: LOGGER.error("Background task %s failed", task.get_name(), exc_info=error) async def run_agent(self, trigger: str) -> str: if self._paused: raise RuntimeError("Сервис на паузе") if self._run_lock.locked(): raise RuntimeError("Агент уже выполняет другой запуск") async with self._run_lock: await self.database.queue_message(f"🔎 Начат анализ ({trigger}).") try: return await self.agent.run(trigger) except Exception as error: await self.database.queue_message(f"❌ Ошибка анализа: {str(error)[:1000]}") raise async def request_run(self) -> str: if self._paused: return "Сервис на паузе. Сначала выполните /resume." if self._run_lock.locked(): return "Агент уже работает." self._spawn(self.run_agent("telegram"), "manual-agent-run") return "Внеплановый анализ запущен." async def pause(self) -> str: self._paused = True await self.database.set_state("paused", "true") return "Новые анализы и входы приостановлены. Защитные ордера Bybit остаются активны." async def resume(self) -> str: self._paused = False await self.database.set_state("paused", "false") return "Работа возобновлена." async def status_text(self) -> str: next_run = next_scheduled_time(self.config.schedule) last = await self.database.last_run() return ( f"Режим: Bybit Demo\n" f"Состояние: {'пауза' if self._paused else 'активен'}\n" f"Агент: {'работает' if self._run_lock.locked() else 'ожидает'}\n" f"Следующий запуск: {next_run.isoformat()}\n" f"Последний запуск: {json.dumps(last, ensure_ascii=False, default=str) if last else 'нет'}" ) async def positions_text(self) -> str: state = await self.bybit.account_state() return "Позиции и активы:\n" + json.dumps( {"wallet": state["wallet"], "positions": state["positions"]}, ensure_ascii=False, default=str )[:3800] async def orders_text(self) -> str: orders = await self.bybit.open_orders() return "Открытые ордера:\n" + json.dumps(orders, ensure_ascii=False, default=str)[:3800] async def last_run_text(self) -> str: last = await self.database.last_run() return json.dumps(last, ensure_ascii=False, default=str)[:4000] if last else "Запусков ещё не было." async def notes_text(self) -> str: notes = await self.database.recent_notes(self.config.storage.max_notes, 3500) return "Заметки:\n" + ("\n\n".join(f"• {item}" for item in notes) if notes else "нет") async def _scheduler_loop(self) -> None: while not self._stopping.is_set(): scheduled = next_scheduled_time(self.config.schedule) delay = max(0, (scheduled.astimezone(UTC) - datetime.now(UTC)).total_seconds()) try: await asyncio.wait_for(self._stopping.wait(), timeout=delay) return except TimeoutError: pass if not self._paused: try: await self.run_agent(f"schedule:{scheduled.isoformat()}") except Exception: LOGGER.exception("Scheduled agent run failed") async def _outbox_loop(self) -> None: while not self._stopping.is_set(): for item in await self.database.pending_outbox(): try: await self.telegram.send_to_admins(item.text) await self.database.mark_outbox_sent(item.id) except Exception as error: # noqa: BLE001 - outbox must retry any connector failure LOGGER.warning("Telegram outbox send failed: %s", error) await self.database.mark_outbox_failed(item.id, str(error)) try: await asyncio.wait_for(self._stopping.wait(), timeout=5) except TimeoutError: pass async def _reconcile_loop(self) -> None: while not self._stopping.is_set(): try: await self._cancel_expired_orders() await self._apply_pending_trailing_stops() except Exception: LOGGER.exception("Bybit reconciliation failed") try: await asyncio.wait_for(self._stopping.wait(), timeout=60) except TimeoutError: pass async def _cancel_expired_orders(self) -> None: for order in await self.database.expired_managed_orders(): try: await self.bybit.cancel_order( category=order["category"], symbol=order["symbol"], orderId=order["order_id"] ) await self.database.mark_managed_order_status(order["order_link_id"], "Cancelled") await self.database.queue_message( f"⌛ Просроченный Limit-ордер отменён: {order['category']} {order['symbol']}" ) except Exception as error: if "not found" not in str(error).lower() and "too late" not in str(error).lower(): raise async def _apply_pending_trailing_stops(self) -> None: for managed in await self.database.managed_orders_with_pending_trailing(): history = await self.bybit.order_history("linear", 50) matching = next( (item for item in history if item.get("orderLinkId") == managed["order_link_id"]), None ) if matching is None or matching.get("orderStatus") != "Filled": continue order_link_id = managed["order_link_id"] symbol = managed["symbol"] trailing_distance = managed["trailing_distance"] if order_link_id is None or symbol is None or trailing_distance is None: continue await self.bybit.set_trading_stop( category="linear", symbol=symbol, positionIdx=0, tpslMode="Full", trailingStop=trailing_distance, ) await self.database.mark_managed_order_status(order_link_id, "TrailingActive") await self.database.queue_message( f"🛡 Trailing-stop установлен: {symbol} distance={trailing_distance}" ) def _start_private_websocket(self) -> None: websocket = WebSocket( testnet=False, demo=True, channel_type="private", api_key=self.config.bybit.api_key, api_secret=self.config.bybit.api_secret, ) websocket.order_stream(callback=self._websocket_callback("order")) websocket.execution_stream(callback=self._websocket_callback("execution")) websocket.position_stream(callback=self._websocket_callback("position")) websocket.wallet_stream(callback=self._websocket_callback("wallet")) self._websocket = websocket def _websocket_callback(self, kind: str): def callback(message: dict[str, Any]) -> None: if self._loop is None or self._loop.is_closed(): return future = asyncio.run_coroutine_threadsafe(self._record_ws_event(kind, message), self._loop) future.add_done_callback(lambda completed: completed.exception() if not completed.cancelled() else None) return callback async def _record_ws_event(self, kind: str, message: dict[str, Any]) -> None: await self.database.add_event(f"bybit_ws_{kind}", message) if kind == "order": for order in message.get("data", []): order_link_id = order.get("orderLinkId") order_status = order.get("orderStatus") if order_link_id and order_link_id.startswith("ltb-") and order_status: await self.database.mark_managed_order_status(order_link_id, order_status) if kind in {"order", "execution"}: await self.database.queue_message( f"Bybit {kind}: {json.dumps(message.get('data', message), ensure_ascii=False, default=str)[:3000]}" )