from __future__ import annotations from collections import deque from datetime import timedelta from decimal import Decimal, ROUND_DOWN, ROUND_UP from typing import Iterable from uuid import uuid4 from crypto_spot_bot.bybit import BybitClient, Instrument from crypto_spot_bot.config import Settings from crypto_spot_bot.models import Position, Signal, Ticker, Trade, utc_now from crypto_spot_bot.storage import Storage class BrokerError(RuntimeError): pass def _round_step(value: float, step: float) -> float: if step <= 0: return value value_decimal = Decimal(str(value)) step_decimal = Decimal(str(step)) rounded = (value_decimal / step_decimal).to_integral_value(rounding=ROUND_DOWN) return float(rounded * step_decimal) def _round_step_up(value: float, step: float) -> float: if step <= 0: return value value_decimal = Decimal(str(value)) step_decimal = Decimal(str(step)) rounded = (value_decimal / step_decimal).to_integral_value(rounding=ROUND_UP) return float(rounded * step_decimal) class PaperBroker: def __init__(self, settings: Settings, storage: Storage): self.settings = settings self.storage = storage self.positions = storage.open_positions() self.cash = float(storage.get_runtime("paper_cash", settings.starting_balance_usdt)) self.peak_equity = float(storage.get_runtime("peak_equity", settings.starting_balance_usdt)) self._entry_timestamps = deque() def open_positions(self) -> list[Position]: return list(self.positions) def positions_for_symbol(self, symbol: str) -> list[Position]: return [position for position in self.positions if position.symbol == symbol] def exposure(self) -> float: return sum(position.notional_usdt for position in self.positions) def symbol_exposure(self, symbol: str) -> float: return sum(position.notional_usdt for position in self.positions_for_symbol(symbol)) def equity(self, prices: dict[str, float]) -> float: value = self.cash for position in self.positions: value += position.mark_price(prices.get(position.symbol, position.entry_price)) return value def mark_equity(self, prices: dict[str, float]) -> dict[str, float]: state = self.account_state(prices) equity = state["equity"] self.peak_equity = max(self.peak_equity, equity) state["drawdown"] = max(0.0, self.peak_equity - equity) self.storage.set_runtime("paper_cash", self.cash) self.storage.set_runtime("peak_equity", self.peak_equity) self.storage.insert_equity(equity, self.cash, self.exposure(), state["drawdown"]) return state def account_state(self, prices: dict[str, float]) -> dict[str, float]: equity = self.equity(prices) return { "equity": equity, "cash": self.cash, "exposure": self.exposure(), "drawdown": max(0.0, self.peak_equity - equity), } def update_highs(self, tickers: dict[str, Ticker]) -> None: for position in self.positions: ticker = tickers.get(position.symbol) if not ticker: continue price = ticker.last_price if price > position.highest_price: position.highest_price = price if position.id is not None: self.storage.update_position_highest(position.id, price) def can_open( self, symbol: str, prices: dict[str, float], requested_notional: float | None = None, ) -> tuple[bool, str]: if not self._entry_rate_limit_allows(): return False, "достигнут лимит новых входов в минуту" if len(self.positions) >= self.settings.max_open_positions: return False, "достигнут общий лимит открытых позиций" if self.settings.strategy_mode == "trend_macd" and len(self.positions_for_symbol(symbol)) >= 1: return False, "DCA/усреднение отключено: позиция по паре уже открыта" dynamic_pair_limit = _symbol_position_limit(self.settings) if len(self.positions_for_symbol(symbol)) >= dynamic_pair_limit: return False, "достигнут лимит позиций по паре" requested = requested_notional if requested_notional is not None else self.settings.min_position_usdt symbol_room = max(0.0, self.settings.max_symbol_exposure_usdt - self.symbol_exposure(symbol)) if symbol_room < min(requested, self.settings.min_position_usdt): return False, "достигнут лимит экспозиции по паре" if self.cash <= self.settings.min_cash_reserve_usdt: return False, "недостаточно свободного USDT после резерва" if self.exposure() >= self.settings.max_total_exposure_usdt: return False, "достигнут лимит общей экспозиции" equity_state = self.mark_equity(prices) if equity_state["drawdown"] >= self.settings.max_daily_drawdown_usdt: return False, "достигнут лимит просадки" return True, "ok" def buy( self, signal: Signal, ticker: Ticker, instrument: Instrument | None, prices: dict[str, float], ) -> Position | None: requested_notional = self._signal_notional(signal) allowed, reason = self.can_open(ticker.symbol, prices, requested_notional) if not allowed: self.storage.event(f"{ticker.symbol}: покупка пропущена, {reason}", "WARN") return None return self._record_buy(signal, ticker, instrument, "демо-покупка") def _record_buy( self, signal: Signal, ticker: Ticker, instrument: Instrument | None, event_label: str, ) -> Position | None: fill_price = self._buy_price(ticker) minimum_budget = self._minimum_entry_budget(instrument, fill_price) budget = self._entry_budget(signal, ticker, minimum_notional=minimum_budget) if budget < max(self.settings.min_position_usdt, minimum_budget): self.storage.event(f"{ticker.symbol}: покупка пропущена, adaptive-лимит экспозиции исчерпан", "WARN") return None notional = budget / (1 + self.settings.taker_fee_rate) qty = _round_step(notional / fill_price, instrument.qty_step if instrument else 0) if instrument: qty = self._raise_qty_to_exchange_minimum(qty, fill_price, instrument, budget) if instrument and qty < instrument.min_order_qty: self.storage.event(f"{ticker.symbol}: количество ниже minOrderQty Bybit", "WARN") return None executed_notional = qty * fill_price if instrument and executed_notional < instrument.min_notional_value: self.storage.event(f"{ticker.symbol}: сумма ниже minNotionalValue Bybit", "WARN") return None fee = executed_notional * self.settings.taker_fee_rate if executed_notional + fee > self.cash: self.storage.event(f"{ticker.symbol}: недостаточно cash для комиссии", "WARN") return None stop_loss_percent = self._signal_percent(signal, "stop_loss_percent", self.settings.stop_loss_percent, 0.003, 0.08) take_profit_percent = self._signal_percent( signal, "take_profit_percent", self.settings.take_profit_percent, 0.003, 0.20 ) position = Position( id=None, symbol=ticker.symbol, qty=qty, entry_price=fill_price, notional_usdt=executed_notional, entry_fee_usdt=fee, stop_loss=fill_price * (1 - stop_loss_percent), take_profit=fill_price * (1 + take_profit_percent), highest_price=fill_price, entry_reason=signal.reason, entry_confidence=signal.confidence, entry_pattern=str(signal.diagnostics.get("pattern", {}).get("label", "")), entry_diagnostics=signal.diagnostics, ) position.id = self.storage.insert_position(position) self.positions.append(position) self._record_entry_timestamp() self.cash -= executed_notional + fee self.storage.set_runtime("paper_cash", self.cash) self.storage.insert_trade( Trade( id=None, symbol=ticker.symbol, side="BUY", qty=qty, entry_price=fill_price, fee_usdt=fee, net_pnl=-fee, reason=signal.reason, entry_pattern=position.entry_pattern, entry_confidence=position.entry_confidence, entry_diagnostics=position.entry_diagnostics, opened_at=position.opened_at, ) ) self.storage.event( f"{ticker.symbol}: {event_label} кол-во={qty:.8f} цена={fill_price:.8f} сумма={executed_notional:.2f} уверенность={signal.confidence:.2f}" ) return position def sell(self, position: Position, ticker: Ticker, reason: str) -> Trade: return self._record_sell(position, ticker, reason, "демо-продажа") def _record_sell( self, position: Position, ticker: Ticker, reason: str, event_label: str, ) -> Trade: fill_price = self._sell_price(ticker) exit_notional = position.qty * fill_price exit_fee = exit_notional * self.settings.taker_fee_rate gross_pnl = (fill_price - position.entry_price) * position.qty net_pnl = gross_pnl - position.entry_fee_usdt - exit_fee self.cash += exit_notional - exit_fee if position.id is not None: self.storage.close_position(position.id) self.positions = [item for item in self.positions if item.id != position.id] self.storage.set_runtime("paper_cash", self.cash) trade = Trade( id=None, symbol=position.symbol, side="SELL", qty=position.qty, entry_price=position.entry_price, exit_price=fill_price, gross_pnl=gross_pnl, fee_usdt=position.entry_fee_usdt + exit_fee, net_pnl=net_pnl, reason=reason, entry_pattern=position.entry_pattern, entry_confidence=position.entry_confidence, entry_diagnostics=position.entry_diagnostics, opened_at=position.opened_at, closed_at=utc_now(), ) trade.id = self.storage.insert_trade(trade) self.storage.event( f"{position.symbol}: {event_label} кол-во={position.qty:.8f} цена={fill_price:.8f} итог={net_pnl:.4f} причина={reason}" ) return trade def _buy_price(self, ticker: Ticker) -> float: base = ticker.ask if ticker.ask > 0 else ticker.last_price return base * (1 + self.settings.slippage_rate) def _sell_price(self, ticker: Ticker) -> float: base = ticker.bid if ticker.bid > 0 else ticker.last_price return base * (1 - self.settings.slippage_rate) def _signal_notional(self, signal: Signal) -> float: raw = signal.diagnostics.get("position_notional_usdt", self.settings.max_position_usdt) try: value = float(raw) except (TypeError, ValueError): value = self.settings.max_position_usdt low = max(0.0, self.settings.min_position_usdt) high = max(low, self.settings.max_position_usdt) return max(low, min(high, value)) def _signal_percent(self, signal: Signal, key: str, default: float, low: float, high: float) -> float: rules = signal.diagnostics.get("adaptive_rules") or {} raw = signal.diagnostics.get(key, rules.get(key, default) if isinstance(rules, dict) else default) try: value = float(raw) except (TypeError, ValueError): value = default return max(low, min(high, value)) def minimum_entry_budget(self, instrument: Instrument | None, ticker: Ticker | None = None) -> float: fill_price = self._buy_price(ticker) if ticker is not None else None return self._minimum_entry_budget(instrument, fill_price) def _minimum_entry_budget(self, instrument: Instrument | None, fill_price: float | None = None) -> float: minimum = max(0.0, self.settings.min_position_usdt) if instrument: exchange_notional = max(0.0, instrument.min_notional_value) if fill_price and fill_price > 0: minimum_qty = max(0.0, instrument.min_order_qty) if exchange_notional > 0: minimum_qty = max( minimum_qty, _round_step_up(exchange_notional / fill_price, instrument.qty_step), ) if minimum_qty > 0: exchange_notional = max(exchange_notional, minimum_qty * fill_price) if exchange_notional > 0: exchange_minimum = exchange_notional * (1 + self.settings.taker_fee_rate) * 1.002 + 0.01 minimum = max(minimum, exchange_minimum) return minimum def _raise_qty_to_exchange_minimum( self, qty: float, fill_price: float, instrument: Instrument, budget: float, ) -> float: minimum_qty = max(0.0, instrument.min_order_qty) if instrument.min_notional_value > 0 and fill_price > 0: minimum_qty = max( minimum_qty, _round_step_up(instrument.min_notional_value / fill_price, instrument.qty_step), ) if minimum_qty <= qty: return qty minimum_cost = minimum_qty * fill_price * (1 + self.settings.taker_fee_rate) if minimum_cost <= budget + 1e-9: return minimum_qty return qty def _entry_budget( self, signal: Signal, ticker: Ticker, extra_cap: float | None = None, minimum_notional: float = 0.0, ) -> float: available = max(0.0, self.cash - self.settings.min_cash_reserve_usdt) rules = signal.diagnostics.get("adaptive_rules") or {} target_total = self._adaptive_cap(rules, "target_total_exposure_usdt", self.settings.max_total_exposure_usdt) target_symbol = self._adaptive_cap(rules, "target_symbol_exposure_usdt", self.settings.max_symbol_exposure_usdt) exposure_room = max(0.0, target_total - self.exposure()) symbol_room = max(0.0, target_symbol - self.symbol_exposure(ticker.symbol)) requested = min( max(self._signal_notional(signal), minimum_notional), max(0.0, self.settings.max_position_usdt), ) caps = [requested, available, exposure_room, symbol_room] if extra_cap is not None: caps.append(max(0.0, extra_cap)) return max(0.0, min(caps)) def _adaptive_cap(self, rules: object, key: str, default: float) -> float: if not isinstance(rules, dict): return default try: value = float(rules.get(key, default)) except (TypeError, ValueError): value = default return max(0.0, min(default, value)) def _entry_rate_limit_allows(self) -> bool: limit = self.settings.max_entries_per_minute if limit <= 0: return True now = utc_now() cutoff = now - timedelta(seconds=60) while self._entry_timestamps and self._entry_timestamps[0] < cutoff: self._entry_timestamps.popleft() return len(self._entry_timestamps) < limit def _record_entry_timestamp(self) -> None: if self.settings.max_entries_per_minute <= 0: return self._entry_timestamps.append(utc_now()) class LiveBroker(PaperBroker): def __init__(self, settings: Settings, storage: Storage, client: BybitClient): super().__init__(settings, storage) if not settings.live_ready: raise BrokerError("Live mode is not unlocked by settings") self.client = client def buy( self, signal: Signal, ticker: Ticker, instrument: Instrument | None, prices: dict[str, float], ) -> Position | None: fill_price = self._buy_price(ticker) minimum_budget = self._minimum_entry_budget(instrument, fill_price) requested_notional = min( max(self._signal_notional(signal), minimum_budget), self.settings.live_order_max_usdt, ) allowed, reason = self.can_open(ticker.symbol, prices, requested_notional) if not allowed: self.storage.event(f"{ticker.symbol}: live BUY пропущен, {reason}", "WARN") return None budget = self._entry_budget( signal, ticker, self.settings.live_order_max_usdt, minimum_notional=minimum_budget, ) if budget < max(self.settings.min_position_usdt, minimum_budget): self.storage.event(f"{ticker.symbol}: live BUY skipped, adjusted budget below minimum", "WARN") return None signal.diagnostics["position_notional_usdt"] = budget notional = budget / (1 + self.settings.taker_fee_rate) response = self.client.place_spot_market_order( symbol=ticker.symbol, side="Buy", qty=notional, market_unit="quoteCoin", order_link_id=f"tb-buy-{uuid4().hex[:18]}", ) self.storage.event(f"{ticker.symbol}: реальная покупка отправлена orderId={response.get('orderId')}") return self._record_buy(signal, ticker, instrument, "реальная покупка, локальная запись") def sell(self, position: Position, ticker: Ticker, reason: str) -> Trade: response = self.client.place_spot_market_order( symbol=position.symbol, side="Sell", qty=position.qty, market_unit="baseCoin", order_link_id=f"tb-sell-{uuid4().hex[:18]}", ) self.storage.event( f"{position.symbol}: реальная продажа отправлена orderId={response.get('orderId')} причина={reason}" ) return self._record_sell(position, ticker, reason, "реальная продажа, локальная запись") def prices_from_tickers(tickers: Iterable[Ticker]) -> dict[str, float]: return {ticker.symbol: ticker.last_price for ticker in tickers} def _symbol_position_limit(settings: Settings) -> int: configured_limit = max(1, settings.max_positions_per_symbol) exposure_based_limit = max( 1, int(settings.max_symbol_exposure_usdt // max(settings.min_position_usdt, 0.01)), ) return min(configured_limit, exposure_based_limit)