from __future__ import annotations import asyncio import logging import math import sqlite3 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, apply_profit_only_exit_policy, torch_model_readiness_reasons, ) from crypto_spot_bot.storage import Storage from crypto_spot_bot.time_series import TimeSeriesForecaster, _barrier_outcome logger = logging.getLogger(__name__) 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, shadow_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.shadow_forecaster = shadow_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 self._last_reconciliation_at: datetime | None = None self._last_prune_at: datetime | None = None self._consecutive_loop_errors = 0 self._orderbook_feature_cache_key: tuple[tuple[str, int], ...] = () self._orderbook_feature_cache: dict[str, dict[int, dict[str, float]]] = {} async def start(self) -> None: if self.running: return self.market.reset_stop() if not self.market.symbols: await self.market.bootstrap() if isinstance(self.broker, LiveBroker): try: await asyncio.to_thread(self.broker.reconcile, self.market.instruments) self._last_reconciliation_at = utc_now() except Exception as exc: self.broker.reconciliation_state = { "status": "error", "blocking": True, "discrepancies": [{"code": "initial_reconciliation_failed", "message": str(exc)}], } self.storage.event(f"Initial live reconciliation failed: {exc}", "ERROR") 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._safe_event("Бот запущен") # Maintenance must never delay the first market decision after startup. # The bounded telemetry prune starts after the configured interval. self._last_prune_at = utc_now() 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._safe_event("Бот остановлен") def _safe_event(self, message: str, level: str = "INFO") -> None: try: self.storage.event(message, level) except sqlite3.Error: logger.exception("Could not persist non-critical bot event: %s", message) 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) await self._maintain_runtime() 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() self._consecutive_loop_errors = 0 except asyncio.CancelledError: raise except Exception as exc: self._consecutive_loop_errors += 1 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()): freshness = self.market.symbol_freshness(position.symbol) if not freshness["ok"]: self._record_signal( Signal( position.symbol, "HOLD", 0.0, "market data is stale; exchange protective stop remains authoritative", {"market_freshness": freshness}, ) ) continue 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) if ticker is not None: signal = apply_profit_only_exit_policy(self.settings, position, ticker, signal) self._record_signal(signal) if signal.action == "SELL" and ticker is not None: await asyncio.to_thread(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, mode=self.settings.trading_mode, ), self.storage.latest_equity(mode=self.settings.trading_mode), ) for symbol in self.market.symbols: freshness = self.market.symbol_freshness(symbol) if not freshness["ok"]: self._record_signal( Signal( symbol, "HOLD", 0.0, "market data is stale; new entries blocked", {"market_freshness": freshness, "checks": {"market_fresh": False}}, ) ) continue 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._record_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._record_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._record_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._record_signal(signal) if signal.action == "BUY" and ticker is not None: position = await asyncio.to_thread( self.broker.buy, signal, ticker, instrument, prices, ) if position is not None: self._entry_cooldown_until[symbol] = utc_now() def _record_signal(self, signal: Signal) -> None: self.storage.insert_signal(signal, self.settings.hold_signal_sample_seconds) async def _maintain_runtime(self) -> None: now = utc_now() if isinstance(self.broker, LiveBroker): age = ( (now - self._last_reconciliation_at).total_seconds() if self._last_reconciliation_at else float("inf") ) if age >= self.settings.live_reconciliation_interval_seconds: try: await asyncio.to_thread(self.broker.reconcile, self.market.instruments) except Exception as exc: self.broker.reconciliation_state = { "status": "error", "blocking": True, "discrepancies": [ {"code": "periodic_reconciliation_failed", "message": str(exc)} ], "checked_at": utc_now().isoformat(), } self.storage.event(f"Periodic live reconciliation failed: {exc}", "ERROR") finally: self._last_reconciliation_at = utc_now() prune_age = ( (now - self._last_prune_at).total_seconds() if self._last_prune_at else float("inf") ) if prune_age >= self.settings.storage_prune_interval_seconds: await asyncio.to_thread(self.storage.prune, self.settings.storage_retention_days) self._last_prune_at = 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, ) candidate = Signal( position.symbol, "SELL", 0.5, f"{self.settings.strategy_mode}: старая paper-позиция вне списка разрешенных пар", { "emergency_exit": True, "emergency_exit_type": "symbol_removed_from_universe", }, ) decision = apply_profit_only_exit_policy(self.settings, position, synthetic_ticker, candidate) self._record_signal(decision) if decision.action == "SELL": self.broker.sell(position, synthetic_ticker, decision.reason) self.storage.event( f"{position.symbol}: старая paper-позиция закрыта при переходе на {self.settings.strategy_mode}" ) else: self.storage.event( f"{position.symbol}: старая paper-позиция сохранена политикой profit-only", "WARN", ) 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 or ( self.settings.strategy_mode == "torch_forecast" and self.settings.time_series_trend_fallback_enabled and self.settings.time_series_fallback_mode == "legacy" ) ) 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: cache_key = tuple( (symbol, rows[-1].timestamp if rows else 0) for symbol, rows in sorted(self.market.candles.items()) ) earliest_timestamp = min( (rows[0].timestamp for rows in self.market.candles.values() if rows), default=0, ) if cache_key != self._orderbook_feature_cache_key: orderbook_features, _manifest = self.storage.recent_aggregated_orderbook_features( interval=self.settings.base_interval, symbols=self.market.symbols, after_timestamp_ms=earliest_timestamp, min_samples_per_bucket=20, ) self._orderbook_feature_cache = orderbook_features self._orderbook_feature_cache_key = cache_key else: orderbook_features = self._orderbook_feature_cache 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, []), orderbook_features=orderbook_features, ).as_dict() self.market.forecasts = forecasts self._update_shadow_forecasts(orderbook_features) def _update_shadow_forecasts( self, orderbook_features: dict[str, dict[int, dict[str, float]]], ) -> None: if self.shadow_forecaster is None: self.market.shadow_forecasts = {} return model_sha256 = self.shadow_forecaster.artifact_sha256() if not model_sha256: self.market.shadow_forecasts = {} return forecasts: dict[str, dict] = {} for symbol in self.market.symbols: candles = self.market.candles.get(symbol, []) forecast = self.shadow_forecaster.forecast( candles, symbol=symbol, market_candles=self.market.candles, trend_candles=self.market.trend_candles.get(symbol, []), orderbook_features=orderbook_features, ).as_dict() forecast["shadow"] = True forecast["model_sha256"] = model_sha256 forecasts[symbol] = forecast self._record_and_settle_shadow(symbol, candles, forecast, model_sha256) self.market.shadow_forecasts = forecasts def _record_and_settle_shadow( self, symbol: str, candles: list, forecast: dict, model_sha256: str, ) -> None: if candles and forecast.get("usable"): probability = float( forecast.get("probability_take_profit_first") if forecast.get("probability_take_profit_first") is not None else forecast.get("probability_up", 0.5) ) expected = float(forecast.get("expected_return_percent", 0.0) or 0.0) eligible = bool( not forecast.get("block_entry") and expected >= float(forecast.get("calibrated_min_edge_percent", 0.0) or 0.0) and probability >= float(forecast.get("calibrated_min_probability_up", 0.5) or 0.5) ) self.storage.insert_shadow_prediction( model_sha256=model_sha256, symbol=symbol, forecast_timestamp_ms=candles[-1].timestamp, horizon=max(1, int(forecast.get("horizon", 1) or 1)), reference_price=float(candles[-1].close), expected_return_percent=expected, probability_up=probability, eligible_signal=eligible, ) if not candles: return indexes = {candle.timestamp: index for index, candle in enumerate(candles)} round_trip_cost = 2.0 * ( float(self.settings.taker_fee_rate) + float(self.settings.slippage_rate) ) for row in self.storage.pending_shadow_predictions( model_sha256=model_sha256, symbol=symbol, ): index = indexes.get(int(row.get("forecast_timestamp_ms", 0) or 0)) horizon = max(1, int(row.get("horizon", 1) or 1)) if index is None or index + horizon >= len(candles): continue outcome = _barrier_outcome( candles, end_index=index, horizon=horizon, stop_loss_percent=float(self.settings.stop_loss_percent), take_profit_percent=float(self.settings.take_profit_percent), round_trip_cost=round_trip_cost, ) if outcome is None: continue actual_log_return, take_profit_first = outcome actual_return_percent = (math.exp(actual_log_return) - 1.0) * 100.0 self.storage.settle_shadow_prediction( int(row["id"]), actual_return_percent=actual_return_percent, take_profit_first=take_profit_first >= 0.5, ) def status(self) -> BotStatus: live_ready = self.settings.live_ready if isinstance(self.broker, LiveBroker): live_ready = live_ready and not self.broker.reconciliation_state.get("blocking", True) return BotStatus( running=self.running, mode=self.settings.trading_mode, live_trading_ready=live_ready, symbols=self.market.symbols, started_at=self.started_at, last_loop_at=self.last_loop_at, message=self.message, ) def readiness_snapshot(self) -> dict: reasons: list[str] = [] now = utc_now() if not self.running: reasons.append("bot_not_running") max_loop_age = max(30.0, self.settings.effective_loop_interval_seconds * 4) loop_age = (now - self.last_loop_at).total_seconds() if self.last_loop_at else None if loop_age is None or loop_age > max_loop_age: reasons.append("decision_loop_stale") stale_symbols = [ symbol for symbol in self.market.symbols if not self.market.symbol_freshness(symbol)["ok"] ] if stale_symbols: reasons.append("stale_market_data") if self._consecutive_loop_errors >= 3: reasons.append("repeated_loop_errors") if self.settings.strategy_mode == "torch_forecast": invalid_models = [] for symbol in self.market.symbols: forecast = self.market.forecasts.get(symbol, {}) if torch_model_readiness_reasons(self.settings, forecast): invalid_models.append(symbol) if invalid_models: if self.settings.time_series_trend_fallback_enabled: forecast_fallback_active = True else: forecast_fallback_active = False reasons.append("forecast_model_not_ready") else: forecast_fallback_active = False else: invalid_models = [] forecast_fallback_active = False reconciliation: dict = {} if isinstance(self.broker, LiveBroker): reconciliation = dict(self.broker.reconciliation_state) if reconciliation.get("blocking", True): reasons.append("live_reconciliation_blocking") return { "ready": not reasons, "mode": self.settings.trading_mode, "reasons": reasons, "loop_age_seconds": round(loop_age, 3) if loop_age is not None else None, "stale_symbols": stale_symbols, "consecutive_loop_errors": self._consecutive_loop_errors, "reconciliation": reconciliation, "forecast_model_ready": not invalid_models, "forecast_fallback_active": forecast_fallback_active, "forecast_invalid_symbols": invalid_models, } 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()