from __future__ import annotations import asyncio from datetime import datetime from crypto_spot_bot.analytics import risk_guard_snapshot from crypto_spot_bot.config import Settings from crypto_spot_bot.execution import LiveBroker, PaperBroker from crypto_spot_bot.learning import TradeLearner from crypto_spot_bot.market_data import MarketData from crypto_spot_bot.models import BotStatus, Signal, Ticker, utc_now from crypto_spot_bot.patterns import PatternAnalyzer from crypto_spot_bot.strategy import SpotStrategy from crypto_spot_bot.storage import Storage from crypto_spot_bot.time_series import TimeSeriesForecaster class CryptoSpotBot: def __init__( self, settings: Settings, storage: Storage, market: MarketData, broker: PaperBroker | LiveBroker, strategy: SpotStrategy, pattern_analyzer: PatternAnalyzer, learner: TradeLearner, forecaster: TimeSeriesForecaster | None = None, llm_advisor=None, ): self.settings = settings self.storage = storage self.market = market self.broker = broker self.strategy = strategy self.pattern_analyzer = pattern_analyzer self.learner = learner self.forecaster = forecaster self.llm_advisor = llm_advisor self.running = False self.started_at: datetime | None = None self.last_loop_at: datetime | None = None self.message = "бот остановлен" self._entry_cooldown_until: dict[str, datetime] = {} self._loop_task: asyncio.Task | None = None self._ws_task: asyncio.Task | None = None async def start(self) -> None: if self.running: return self.market.reset_stop() if not self.market.symbols: await self.market.bootstrap() self._close_paper_positions_outside_symbol_universe() self._update_patterns() self._update_forecasts() self.learner.refresh() self.running = True self.started_at = utc_now() self.message = "бот работает" self.storage.event("Бот запущен") if self.settings.websocket_enabled: self._ws_task = asyncio.create_task(self.market.websocket_loop()) self._loop_task = asyncio.create_task(self._run_loop()) async def stop(self) -> None: self.running = False self.message = "бот остановлен" self.market.stop() tasks = [task for task in (self._loop_task, self._ws_task) if task] for task in tasks: task.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) self.storage.event("Бот остановлен") async def _run_loop(self) -> None: while self.running: try: rest_refresh_seconds = self._rest_refresh_seconds() if self._needs_rest_refresh(rest_refresh_seconds): await asyncio.to_thread(self.market.refresh_rest) self.broker.update_highs(self.market.tickers) self._update_patterns() self._update_forecasts() self.learner.refresh() await self._process_exits() await self._process_entries() self.broker.mark_equity(self.market.prices()) self.last_loop_at = utc_now() except asyncio.CancelledError: raise except Exception as exc: self.message = f"ошибка цикла: {exc}" self.storage.event(self.message, "ERROR") await asyncio.sleep(self.settings.effective_loop_interval_seconds) def _needs_rest_refresh(self, rest_refresh_seconds: float) -> bool: if self.market.last_rest_refresh_at is None: return True age = (utc_now() - self.market.last_rest_refresh_at).total_seconds() return age >= rest_refresh_seconds def _rest_refresh_seconds(self) -> float: if self.settings.websocket_enabled: return 10.0 if self.settings.fast_trading_enabled else 20.0 return self.settings.effective_loop_interval_seconds async def _process_exits(self) -> None: prices = self.market.prices() reduction_candidate_id = self._reduction_candidate_id(prices) for position in list(self.broker.open_positions()): ticker = self.market.tickers.get(position.symbol) candles = self.market.candles.get(position.symbol, []) forecast = self.market.forecasts.get(position.symbol, {}) adaptive_rules = self._with_exposure_context(self.learner.rules_for(position.symbol, position.entry_pattern)) adaptive_rules["reduce_now"] = position.id is not None and position.id == reduction_candidate_id learning = {"adaptive_rules": adaptive_rules} signal = self.strategy.exit_signal(position, candles, ticker, learning, forecast) self.storage.insert_signal(signal) if signal.action == "SELL" and ticker is not None: self.broker.sell(position, ticker, signal.reason) self._entry_cooldown_until[position.symbol] = utc_now() async def _process_entries(self) -> None: prices = self.market.prices() risk_guard = risk_guard_snapshot( self.settings, self.storage.closed_trades(self.settings.learning_lookback_trades), self.storage.latest_equity(), ) for symbol in self.market.symbols: cooldown_since = self._entry_cooldown_until.get(symbol) if cooldown_since: age = (utc_now() - cooldown_since).total_seconds() cooldown_seconds = self.settings.effective_entry_cooldown_seconds if age < cooldown_seconds: self.storage.insert_signal( Signal( symbol, "HOLD", 0.0, "пауза между входами по паре", {"cooldown_remaining_seconds": cooldown_seconds - age}, ) ) continue self._entry_cooldown_until.pop(symbol, None) ticker = self.market.tickers.get(symbol) candles = self.market.candles.get(symbol, []) trend_candles = self.market.trend_candles.get(symbol, []) open_count = len(self.broker.positions_for_symbol(symbol)) instrument = self.market.instruments.get(symbol) pattern = self.market.patterns.get(symbol, {}) forecast = self.market.forecasts.get(symbol, {}) learning = self.learner.adjustment_for(symbol, str(pattern.get("label", ""))).as_dict() learning["adaptive_rules"] = self._with_exposure_context(learning.get("adaptive_rules") or {}) account = self.broker.account_state(prices) account["risk_guard"] = risk_guard account["symbol"] = symbol account["symbol_exposure_usdt"] = self.broker.symbol_exposure(symbol) account["open_positions_for_symbol"] = open_count account["exchange_min_entry_usdt"] = self.broker.minimum_entry_budget(instrument, ticker) if risk_guard.get("block_new_entries"): self.storage.insert_signal( Signal( symbol, "HOLD", 0.0, "risk_guard: new entries blocked", { "strategy_mode": self.settings.strategy_mode, "risk_guard": risk_guard, "checks": {"risk_guard_ok": False}, }, ) ) continue symbol_guard = self._risk_guard_for_symbol(risk_guard, symbol) if symbol_guard.get("block_new_entries"): self.storage.insert_signal( Signal( symbol, "HOLD", 0.0, "risk_guard: symbol blocked", { "strategy_mode": self.settings.strategy_mode, "risk_guard": risk_guard, "symbol_guard": symbol_guard, "checks": {"risk_guard_symbol_ok": False}, }, ) ) continue llm = {} if ( self.settings.llm_advisor_enabled and self.llm_advisor is not None and ticker is not None and len(candles) >= 200 ): llm = ( await asyncio.to_thread( self.llm_advisor.advice_for, symbol=symbol, candles=candles, ticker=ticker, pattern=pattern, learning=learning, open_positions_for_symbol=open_count, account=account, ) ).as_dict() signal = self.strategy.entry_signal( symbol, candles, ticker, open_count, pattern, learning, llm, forecast, account, trend_candles, ) self.storage.insert_signal(signal) if signal.action == "BUY" and ticker is not None: position = self.broker.buy( signal, ticker, instrument, prices, ) if position is not None: self._entry_cooldown_until[symbol] = utc_now() @staticmethod def _risk_guard_for_symbol(risk_guard: dict, symbol: str) -> dict: rows = risk_guard.get("symbols") if not isinstance(rows, list): return {} symbol_upper = symbol.upper() for row in rows: if isinstance(row, dict) and str(row.get("symbol", "")).upper() == symbol_upper: return row return {} def _with_exposure_context(self, rules: dict) -> dict: enriched = dict(rules) current_exposure = self.broker.exposure() target_total = float(enriched.get("target_total_exposure_usdt", self.settings.max_total_exposure_usdt) or 0.0) target_total = max(0.0, min(self.settings.max_total_exposure_usdt, target_total)) enriched["current_total_exposure_usdt"] = round(current_exposure, 6) enriched["target_total_exposure_usdt"] = round(target_total, 6) enriched["over_target_exposure"] = current_exposure > target_total + self.settings.min_position_usdt return enriched def _close_paper_positions_outside_symbol_universe(self) -> None: if self.settings.strategy_mode not in {"trend_macd", "torch_forecast"} or self.settings.trading_mode != "paper": return allowed_symbols = set(self.market.symbols or self.settings.symbols) for position in list(self.broker.open_positions()): if position.symbol in allowed_symbols: continue synthetic_ticker = Ticker( symbol=position.symbol, last_price=position.entry_price, bid=position.entry_price, ask=position.entry_price, turnover_24h=0.0, volume_24h=0.0, change_24h=0.0, ) self.broker.sell( position, synthetic_ticker, f"{self.settings.strategy_mode}: закрыта старая paper-позиция вне списка разрешенных пар", ) self.storage.event( f"{position.symbol}: старая paper-позиция закрыта при переходе на {self.settings.strategy_mode}" ) def _reduction_candidate_id(self, prices: dict[str, float]) -> int | None: rules = self._with_exposure_context(self.learner.state.adaptive_rules or {}) if not rules.get("reduce_exposure") or not rules.get("over_target_exposure"): return None positions = self.broker.open_positions() if not positions: return None def loss_ratio(position) -> float: mark = prices.get(position.symbol, position.entry_price) if position.notional_usdt <= 0: return 0.0 return position.unrealized_pnl(mark) / position.notional_usdt worst = min(positions, key=loss_ratio) return worst.id def _update_patterns(self) -> None: patterns_needed = ( self.settings.pattern_analysis_enabled or self.settings.grid_trading_enabled or self.settings.rebound_trading_enabled ) if self.settings.strategy_mode == "trend_macd" or not patterns_needed: self.market.patterns = {} return patterns: dict[str, dict] = {} for symbol in self.market.symbols: result = self.pattern_analyzer.analyze( self.market.candles.get(symbol, []), self.market.tickers.get(symbol), ) patterns[symbol] = result.as_dict() self.market.patterns = patterns def _update_forecasts(self) -> None: if ( self.forecaster is None or not self.settings.time_series_forecast_enabled ): self.market.forecasts = {} return forecasts: dict[str, dict] = {} for symbol in self.market.symbols: forecasts[symbol] = self.forecaster.forecast( self.market.candles.get(symbol, []), symbol=symbol, market_candles=self.market.candles, trend_candles=self.market.trend_candles.get(symbol, []), ).as_dict() self.market.forecasts = forecasts def status(self) -> BotStatus: return BotStatus( running=self.running, mode=self.settings.trading_mode, live_trading_ready=self.settings.live_ready, symbols=self.market.symbols, started_at=self.started_at, last_loop_at=self.last_loop_at, message=self.message, ) def account_snapshot(self) -> dict[str, float]: prices = self.market.prices() state = self.broker.account_state(prices) state["starting_balance"] = self.settings.starting_balance_usdt state["net_pnl"] = state["equity"] - self.settings.starting_balance_usdt state["net_pnl_percent"] = ( (state["net_pnl"] / self.settings.starting_balance_usdt) * 100 if self.settings.starting_balance_usdt else 0.0 ) return state def positions_snapshot(self) -> list[dict]: prices = self.market.prices() items: list[dict] = [] for position in self.broker.open_positions(): mark_price = prices.get(position.symbol, position.entry_price) item = position.as_dict(mark_price=mark_price) exit_signal = self.strategy.exit_signal( position=position, candles=self.market.candles.get(position.symbol, []), ticker=self.market.tickers.get(position.symbol), learning=self.learner.state.as_dict(), forecast=self.market.forecasts.get(position.symbol, {}), ) diagnostics = exit_signal.diagnostics or {} fallback_stop_loss = position.stop_loss if self.settings.stop_loss_exit_enabled else None item["exit_plan"] = { "action": exit_signal.action, "reason": exit_signal.reason, "confidence": exit_signal.confidence, "stop_loss": diagnostics.get("stop_loss", fallback_stop_loss), "take_profit": diagnostics.get("take_profit", position.take_profit), "trailing_stop": diagnostics.get("trailing_stop"), "atr_trailing_stop": diagnostics.get("atr_trailing_stop"), "highest_price": diagnostics.get("highest_price", position.highest_price), "stop_loss_exit_enabled": diagnostics.get( "stop_loss_exit_enabled", self.settings.stop_loss_exit_enabled, ), } items.append(item) return items def learning_snapshot(self) -> dict: snapshot = self.learner.state.as_dict() snapshot["adaptive_rules"] = self._with_exposure_context(snapshot.get("adaptive_rules") or {}) return snapshot def llm_snapshot(self) -> dict: if self.llm_advisor is None: return {"enabled": False, "items": []} return self.llm_advisor.snapshot()