from __future__ import annotations import time from dataclasses import dataclass from datetime import datetime from decimal import Decimal, InvalidOperation, ROUND_DOWN from typing import Iterable from .audit_log import AuditLogger from .config import Settings from .execution import build_client_order_id from .exchange import BinanceSpotClient from .models import AccountSnapshot, Candle, OrderSide, SymbolMetadata, TradingMode from .spot_order_fills import record_order_fills from .storage import Storage BPS = Decimal("10000") def _audit_log_dir(settings: Settings): return settings.database_path.parent / "logs" def _audit_candidate(candidate: AutoBuyCandidate | None) -> dict | None: if candidate is None: return None return { "symbol": candidate.symbol, "score": candidate.score, "reference_price": candidate.reference_price, "spread_bps": candidate.spread_bps, "reasons": list(candidate.reasons), } def _audit_formula() -> dict[str, str]: return { "spread_bps": "(ask - bid) / ((ask + bid) / 2) * 10000", "short_sma": "sum(last 3 closes) / 3", "long_sma": "sum(last 6 closes) / 6", "baseline_volume": "sum(previous 5 volumes) / 5", "score": "momentum flags + volume confirmation + positive 4h momentum percent", } def _balance_parts(account: AccountSnapshot, asset: str) -> tuple[Decimal, Decimal]: balance = account.balances.get(asset.upper()) if balance is None: return Decimal("0"), Decimal("0") return balance.free, balance.locked def _is_finite_count(value: object) -> bool: return isinstance(value, int) and not isinstance(value, bool) def _is_finite_decimal(value: object) -> bool: return isinstance(value, Decimal) and value.is_finite() def _is_bool(value: object) -> bool: return isinstance(value, bool) def _is_integral_count(value: object) -> bool: if _is_finite_count(value): return True return isinstance(value, Decimal) and value.is_finite() and value == value.to_integral_value() def _all_finite(*values: object) -> bool: return all(_is_finite_decimal(value) for value in values) def _all_non_negative(*values: Decimal) -> bool: return all(value >= 0 for value in values) def _valid_text(value: object) -> bool: return isinstance(value, str) and bool(value.strip()) def _has_metadata_shape(metadata: object) -> bool: required_text_fields = ("symbol", "base_asset", "quote_asset") required_decimal_fields = ("min_notional", "min_qty", "step_size", "tick_size") return all(_valid_text(getattr(metadata, field, None)) for field in required_text_fields) and all( isinstance(getattr(metadata, field, None), Decimal) for field in required_decimal_fields ) @dataclass(frozen=True) class AutoBuyMarketSnapshot: symbol: str bid: Decimal ask: Decimal candles: list[Candle] def __post_init__(self) -> None: object.__setattr__(self, "symbol", self.symbol.upper()) @dataclass(frozen=True) class AutoBuyCandidate: symbol: str score: Decimal reference_price: Decimal spread_bps: Decimal reasons: tuple[str, ...] def __post_init__(self) -> None: object.__setattr__(self, "symbol", self.symbol.upper()) class AutoBuyScanner: def __init__(self, symbols: tuple[str, ...], min_score: Decimal, max_spread_bps: Decimal) -> None: self.symbols = tuple(symbol.upper() for symbol in symbols) self.min_score = min_score self.max_spread_bps = max_spread_bps def select(self, snapshots: Iterable[AutoBuyMarketSnapshot]) -> AutoBuyCandidate | None: by_symbol = {snapshot.symbol: snapshot for snapshot in snapshots} candidates = [self._score(by_symbol[symbol]) for symbol in self.symbols if symbol in by_symbol] eligible = [candidate for candidate in candidates if candidate is not None and candidate.score >= self.min_score] if not eligible: return None return max(eligible, key=lambda candidate: (candidate.score, -self.symbols.index(candidate.symbol))) def _score(self, snapshot: AutoBuyMarketSnapshot) -> AutoBuyCandidate | None: if not snapshot.bid.is_finite() or not snapshot.ask.is_finite(): return None if snapshot.bid <= 0 or snapshot.ask <= 0 or snapshot.ask < snapshot.bid: return None candles = snapshot.candles if len(candles) < 6: return None midpoint = (snapshot.bid + snapshot.ask) / Decimal("2") spread_bps = ((snapshot.ask - snapshot.bid) / midpoint) * BPS if spread_bps > self.max_spread_bps: return None recent_candles = candles[-6:] closes = [candle.close for candle in recent_candles] volumes = [candle.volume for candle in recent_candles] if any(not close.is_finite() for close in closes): return None if any(not volume.is_finite() for volume in volumes): return None if any(close <= 0 for close in closes): return None if any(volume <= 0 for volume in volumes): return None score = Decimal("0") reasons: list[str] = [] if closes[-1] > closes[-2]: score += Decimal("1") reasons.append("positive 1h momentum") if closes[-1] > closes[-5]: score += Decimal("1") reasons.append("positive 4h momentum") short_sma = sum(closes[-3:]) / Decimal("3") long_sma = sum(closes[-6:]) / Decimal("6") if short_sma > long_sma: score += Decimal("1") reasons.append("short SMA above long SMA") baseline_volume = sum(volumes[-6:-1]) / Decimal("5") if volumes[-1] <= baseline_volume: return None score += Decimal("1") reasons.append("volume above baseline") momentum = ((closes[-1] - closes[-5]) / closes[-5]) * Decimal("100") if momentum > 0: score += momentum.quantize(Decimal("0.0001")) return AutoBuyCandidate( symbol=snapshot.symbol, score=score, reference_price=snapshot.ask, spread_bps=spread_bps, reasons=tuple(reasons), ) @dataclass(frozen=True) class AutoBuyRiskSettings: daily_limit: int max_spread_bps: Decimal = Decimal("10") max_24h_rally_percent: Decimal = Decimal("10") quote_reserve: Decimal = Decimal("0") existing_base_value_limit: Decimal | None = None def __post_init__(self) -> None: if not (_is_finite_count(self.daily_limit) or (isinstance(self.daily_limit, Decimal) and self.daily_limit.is_finite())): raise ValueError("daily_limit must be finite") if not _is_integral_count(self.daily_limit): raise ValueError("daily_limit must be an integer") if self.daily_limit <= 0: raise ValueError("daily_limit must be positive") object.__setattr__(self, "daily_limit", int(self.daily_limit)) if not isinstance(self.max_spread_bps, Decimal): raise ValueError("max_spread_bps must be a Decimal") if not self.max_spread_bps.is_finite(): raise ValueError("max_spread_bps must be finite") if self.max_spread_bps <= 0: raise ValueError("max_spread_bps must be positive") if not isinstance(self.max_24h_rally_percent, Decimal): raise ValueError("max_24h_rally_percent must be a Decimal") if not self.max_24h_rally_percent.is_finite(): raise ValueError("max_24h_rally_percent must be finite") if self.max_24h_rally_percent <= 0: raise ValueError("max_24h_rally_percent must be positive") if not isinstance(self.quote_reserve, Decimal): raise ValueError("quote_reserve must be a Decimal") if not self.quote_reserve.is_finite(): raise ValueError("quote_reserve must be finite") if self.quote_reserve < 0: raise ValueError("quote_reserve must be non-negative") if self.existing_base_value_limit is not None: if not isinstance(self.existing_base_value_limit, Decimal): raise ValueError("existing_base_value_limit must be a Decimal") if not self.existing_base_value_limit.is_finite(): raise ValueError("existing_base_value_limit must be finite") if self.existing_base_value_limit <= 0: raise ValueError("existing_base_value_limit must be positive") @dataclass(frozen=True) class AutoBuyRiskContext: mode: TradingMode enabled: bool candidate: AutoBuyCandidate metadata: SymbolMetadata account: AccountSnapshot daily_attempts: int has_recent_symbol_attempt: bool quote_amount: Decimal price_change_percent_24h: Decimal @dataclass(frozen=True) class AutoBuyRiskDecision: approved: bool reason: str candidate: AutoBuyCandidate class AutoBuyRiskEngine: def __init__(self, settings: AutoBuyRiskSettings) -> None: self.settings = settings def evaluate(self, context: AutoBuyRiskContext) -> AutoBuyRiskDecision: candidate = context.candidate if context.mode is not TradingMode.LIVE: return AutoBuyRiskDecision(False, "live mode required", candidate) if not _is_bool(context.enabled): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if context.enabled is not True: return AutoBuyRiskDecision(False, "live auto buyer disabled", candidate) if not _is_integral_count(context.daily_attempts) or context.daily_attempts < 0: return AutoBuyRiskDecision(False, "invalid risk input", candidate) if context.daily_attempts >= self.settings.daily_limit: return AutoBuyRiskDecision(False, "daily auto-buy limit reached", candidate) if not _is_bool(context.has_recent_symbol_attempt): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if context.has_recent_symbol_attempt: return AutoBuyRiskDecision(False, "symbol cooldown active", candidate) if not _has_metadata_shape(context.metadata): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if candidate.symbol != context.metadata.symbol: return AutoBuyRiskDecision(False, "metadata symbol mismatch", candidate) if not _all_finite(candidate.score, candidate.reference_price, candidate.spread_bps): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if candidate.score < 0 or candidate.reference_price <= 0 or candidate.spread_bps < 0: return AutoBuyRiskDecision(False, "invalid risk input", candidate) if candidate.spread_bps > self.settings.max_spread_bps: return AutoBuyRiskDecision(False, "spread above max", candidate) if not _all_finite(context.price_change_percent_24h): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if context.price_change_percent_24h >= self.settings.max_24h_rally_percent: return AutoBuyRiskDecision(False, "24h rally above max", candidate) if not _all_finite(context.quote_amount, context.metadata.min_notional): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if context.quote_amount <= 0 or context.metadata.min_notional <= 0: return AutoBuyRiskDecision(False, "invalid risk input", candidate) if context.quote_amount < context.metadata.min_notional: return AutoBuyRiskDecision(False, "quote amount below min notional", candidate) quote_free, quote_locked = _balance_parts(context.account, context.metadata.quote_asset) if not _all_finite(quote_free, quote_locked): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if not _all_non_negative(quote_free, quote_locked): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if quote_free < context.quote_amount + self.settings.quote_reserve: return AutoBuyRiskDecision(False, "insufficient quote balance", candidate) base_free, base_locked = _balance_parts(context.account, context.metadata.base_asset) if not _all_finite(base_free, base_locked): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if not _all_non_negative(base_free, base_locked): return AutoBuyRiskDecision(False, "invalid risk input", candidate) base_value = (base_free + base_locked) * candidate.reference_price existing_base_value_limit = ( self.settings.existing_base_value_limit if self.settings.existing_base_value_limit is not None else context.quote_amount ) if base_value >= existing_base_value_limit: return AutoBuyRiskDecision(False, "existing base balance", candidate) if not _all_finite(context.metadata.min_qty, context.metadata.step_size, context.metadata.tick_size): return AutoBuyRiskDecision(False, "invalid risk input", candidate) if context.metadata.min_qty <= 0 or context.metadata.step_size <= 0 or context.metadata.tick_size <= 0: return AutoBuyRiskDecision(False, "missing exchange filters", candidate) return AutoBuyRiskDecision(True, "approved", candidate) @dataclass(frozen=True) class AutoBuyExecutionResult: run_id: int status: str symbol: str reason: str def __post_init__(self) -> None: object.__setattr__(self, "symbol", self.symbol.upper()) def _round_down(value: Decimal, increment: Decimal) -> Decimal: if not isinstance(increment, Decimal) or not increment.is_finite() or increment <= 0: raise ValueError("increment must be positive") if not isinstance(value, Decimal) or not value.is_finite(): raise ValueError("value must be a finite Decimal") units = (value / increment).to_integral_value(rounding=ROUND_DOWN) return (units * increment).quantize(increment, rounding=ROUND_DOWN) def _average_fill_price(payload: dict) -> Decimal: try: executed_qty = Decimal(str(payload["executedQty"])) quote_qty = Decimal(str(payload["cummulativeQuoteQty"])) except (KeyError, InvalidOperation, ValueError): raise ValueError("invalid fill quantities") from None if not executed_qty.is_finite() or not quote_qty.is_finite() or executed_qty <= 0 or quote_qty <= 0: raise ValueError("invalid fill quantities") return quote_qty / executed_qty def _sellable_executed_quantity(payload: dict, base_asset: str, step_size: Decimal) -> Decimal: try: executed_qty = Decimal(str(payload["executedQty"])) except (KeyError, InvalidOperation, ValueError): raise ValueError("invalid fill quantities") from None if not executed_qty.is_finite() or executed_qty <= 0: raise ValueError("invalid fill quantities") base_commission = Decimal("0") fills = payload.get("fills") if fills is not None: if not isinstance(fills, list): raise ValueError("invalid fill quantities") for fill in fills: if not isinstance(fill, dict): raise ValueError("invalid fill quantities") commission_asset = str(fill.get("commissionAsset", "")).upper() if commission_asset != base_asset.upper(): continue try: commission = Decimal(str(fill["commission"])) except (KeyError, InvalidOperation, ValueError): raise ValueError("invalid fill quantities") from None if not commission.is_finite() or commission < 0: raise ValueError("invalid fill quantities") base_commission += commission sellable_qty = executed_qty - base_commission if not sellable_qty.is_finite() or sellable_qty <= 0: raise ValueError("invalid fill quantities") return _round_down(sellable_qty, step_size) def _validate_positive_decimal(name: str, value: object) -> Decimal: if not isinstance(value, Decimal): raise ValueError(f"{name} must be a Decimal") if not value.is_finite(): raise ValueError(f"{name} must be finite") if value <= 0: raise ValueError(f"{name} must be positive") return value def _validate_percent_less_than_100(name: str, value: object) -> Decimal: value = _validate_positive_decimal(name, value) if value >= Decimal("100"): raise ValueError(f"{name} must be less than 100") return value def _payload_status(payload: dict) -> str: status = payload.get("status") if isinstance(status, str) and status: return status return "MISSING" def _rollback_executed_quantity(payload: dict) -> Decimal | None: try: quantity = Decimal(str(payload["executedQty"])) except (KeyError, InvalidOperation, ValueError): return None if not quantity.is_finite() or quantity <= 0: return None return quantity def _decimal_payload_field_matches(payload: dict, field: str, expected: Decimal) -> bool: if field not in payload: return True try: quantity = Decimal(str(payload[field])) except (InvalidOperation, ValueError): return False return quantity.is_finite() and quantity == expected def _valid_market_buy_payload(payload: dict, symbol: str, client_order_id: str) -> bool: if _payload_status(payload) != "FILLED": return False if payload.get("symbol") != symbol: return False if payload.get("clientOrderId") != client_order_id: return False if "side" in payload and payload.get("side") != "BUY": return False if "type" in payload and payload.get("type") != "MARKET": return False return True def _oco_order_reports(payload: dict) -> list[dict]: reports = payload.get("orderReports") if reports is None: return [] if not isinstance(reports, list): return [] return [report for report in reports if isinstance(report, dict)] def _valid_oco_report( report: dict, symbol: str, side: str, order_type: str, quantity: Decimal, price: Decimal, stop_price: Decimal | None = None, ) -> bool: if report.get("symbol") != symbol: return False if "side" in report and report.get("side") != side: return False if "type" in report and report.get("type") != order_type: return False if "status" in report and report.get("status") not in {"NEW", "PENDING_NEW"}: return False quantity_matches = all(_decimal_payload_field_matches(report, field, quantity) for field in ("origQty", "quantity")) price_matches = _decimal_payload_field_matches(report, "price", price) stop_matches = True if stop_price is None else _decimal_payload_field_matches(report, "stopPrice", stop_price) return quantity_matches and price_matches and stop_matches def _valid_protective_oco_payload( payload: dict, symbol: str, list_client_order_id: str, take_profit_client_order_id: str, stop_client_order_id: str, quantity: Decimal, take_profit_price: Decimal, stop_price: Decimal, stop_limit_price: Decimal, ) -> bool: status = payload.get("status") if isinstance(status, str) and status not in {"NEW", "EXECUTING", "EXEC_STARTED"}: return False list_status_type = payload.get("listStatusType") if isinstance(list_status_type, str) and list_status_type not in {"EXEC_STARTED", "ALL_DONE"}: return False list_order_status = payload.get("listOrderStatus") if isinstance(list_order_status, str) and list_order_status not in {"EXECUTING", "ALL_DONE"}: return False if list_status_type is None and list_order_status is None and "orderReports" not in payload: return False if payload.get("symbol") != symbol: return False if payload.get("listClientOrderId") != list_client_order_id: return False if "orderReports" in payload and not isinstance(payload.get("orderReports"), list): return False reports = _oco_order_reports(payload) if "orderReports" in payload and not reports: return False if not reports: return True take_profit_reports = [report for report in reports if report.get("clientOrderId") == take_profit_client_order_id] stop_reports = [report for report in reports if report.get("clientOrderId") == stop_client_order_id] if not take_profit_reports: take_profit_reports = [report for report in reports if report.get("type") == "LIMIT_MAKER"] if not stop_reports: stop_reports = [report for report in reports if report.get("type") == "STOP_LOSS_LIMIT"] if len(take_profit_reports) != 1 or len(stop_reports) != 1: return False return _valid_oco_report( take_profit_reports[0], symbol, "SELL", "LIMIT_MAKER", quantity, take_profit_price, ) and _valid_oco_report( stop_reports[0], symbol, "SELL", "STOP_LOSS_LIMIT", quantity, stop_limit_price, stop_price, ) def _valid_rollback_payload(payload: dict, symbol: str, client_order_id: str, quantity: Decimal) -> bool: if _payload_status(payload) != "FILLED": return False if payload.get("symbol") != symbol: return False if payload.get("clientOrderId") != client_order_id: return False rollback_quantity = _rollback_executed_quantity(payload) return rollback_quantity is not None and rollback_quantity >= quantity class AutoBuyExecutionEngine: def __init__( self, exchange, storage, stop_loss_percent: Decimal = Decimal("2"), stop_limit_buffer_percent: Decimal = Decimal("0.2"), audit_logger: AuditLogger | None = None, protective_order_max_attempts: int = 3, take_profit_percent: Decimal = Decimal("3"), ) -> None: self.exchange = exchange self.storage = storage self.take_profit_percent = _validate_positive_decimal("take_profit_percent", take_profit_percent) self.stop_loss_percent = _validate_percent_less_than_100("stop_loss_percent", stop_loss_percent) self.stop_limit_buffer_percent = _validate_percent_less_than_100( "stop_limit_buffer_percent", stop_limit_buffer_percent, ) if not isinstance(protective_order_max_attempts, int) or isinstance(protective_order_max_attempts, bool): raise ValueError("protective_order_max_attempts must be a positive integer") if protective_order_max_attempts <= 0: raise ValueError("protective_order_max_attempts must be a positive integer") self.protective_order_max_attempts = protective_order_max_attempts self.audit_logger = audit_logger or AuditLogger() def execute( self, candidate: AutoBuyCandidate, metadata: SymbolMetadata, quote_amount: Decimal, run_id: int | None = None, ) -> AutoBuyExecutionResult: symbol = candidate.symbol if run_id is None: run_id = self.storage.record_auto_buy_run( symbol=symbol, status="buy_submitted", score=candidate.score, reason=", ".join(candidate.reasons), quote_amount=quote_amount, ) else: self.storage.update_auto_buy_run_status(run_id, "buy_submitted", ", ".join(candidate.reasons)) buy_client_order_id = build_client_order_id("bqab-buy") try: buy_payload = self.exchange.create_market_order(symbol, OrderSide.BUY, quote_amount, buy_client_order_id) except Exception: return self._fail_unprotected( run_id, symbol, "buy submission failed", raw={"message": "buy submission failed"}, ) if not isinstance(buy_payload, dict): return self._fail_unprotected( run_id, symbol, "invalid fill quantities", raw={"reason": "invalid fill quantities"}, ) self._record_event( run_id, event_type="market_buy", symbol=symbol, client_order_id=buy_client_order_id, status=buy_payload.get("status", "UNKNOWN"), raw=buy_payload, ) record_order_fills(self.storage, symbol, buy_payload) if not _valid_market_buy_payload(buy_payload, symbol, buy_client_order_id): return self._fail_unprotected( run_id, symbol, "invalid buy response", raw={"reason": "invalid buy response", "buyPayload": buy_payload}, ) try: executed_qty = _sellable_executed_quantity(buy_payload, metadata.base_asset, metadata.step_size) except ValueError: return self._fail_unprotected( run_id, symbol, "invalid fill quantities", raw={"reason": "invalid fill quantities", "buyPayload": buy_payload}, ) if executed_qty < metadata.min_qty: return self._fail_unprotected( run_id, symbol, "executed quantity below min quantity", raw={ "reason": "executed quantity below min quantity", "executedQty": executed_qty, "minQty": metadata.min_qty, }, ) try: average_price = _average_fill_price(buy_payload) except ValueError: self._record_event( run_id, event_type="protective_oco_failed", symbol=symbol, client_order_id=None, status="FAILED", raw={"message": "invalid fill quantities", "buyPayload": buy_payload}, ) return self._rollback(run_id, symbol, executed_qty, reason_prefix="invalid fill quantities") try: take_profit_price = _round_down( average_price * (Decimal("1") + (self.take_profit_percent / Decimal("100"))), metadata.tick_size, ) stop_price = _round_down( average_price * (Decimal("1") - (self.stop_loss_percent / Decimal("100"))), metadata.tick_size, ) stop_limit_price = _round_down( stop_price * (Decimal("1") - (self.stop_limit_buffer_percent / Decimal("100"))), metadata.tick_size, ) except ValueError as exc: self._record_event( run_id, event_type="protective_oco_failed", symbol=symbol, client_order_id=None, status="FAILED", raw={"error_type": type(exc).__name__, "message": "protective oco calculation failed"}, ) return self._rollback(run_id, symbol, executed_qty, reason_prefix="protective oco calculation failed") for _attempt in range(1, self.protective_order_max_attempts + 1): list_client_order_id = build_client_order_id("bqab-oco") take_profit_client_order_id = build_client_order_id("bqab-tp") stop_client_order_id = build_client_order_id("bqab-stop") try: oco_payload = self.exchange.create_oco_sell( symbol, executed_qty, take_profit_price, stop_price, stop_limit_price, list_client_order_id, take_profit_client_order_id, stop_client_order_id, ) except Exception as exc: self._record_event( run_id, event_type="protective_oco_failed", symbol=symbol, client_order_id=list_client_order_id, status="FAILED", raw={"error_type": type(exc).__name__, "message": "protective oco failed"}, ) continue if not isinstance(oco_payload, dict): self._record_event( run_id, event_type="protective_oco_failed", symbol=symbol, client_order_id=list_client_order_id, status="FAILED", raw={"message": "protective oco response invalid"}, ) continue oco_status = str(oco_payload.get("listOrderStatus") or oco_payload.get("listStatusType") or "UNKNOWN") if not _valid_protective_oco_payload( oco_payload, symbol, list_client_order_id, take_profit_client_order_id, stop_client_order_id, executed_qty, take_profit_price, stop_price, stop_limit_price, ): self._record_event( run_id, event_type="protective_oco_failed", symbol=symbol, client_order_id=list_client_order_id, status="FAILED", raw={"message": "protective oco response invalid"}, ) continue self._record_event( run_id, event_type="protective_oco", symbol=symbol, client_order_id=list_client_order_id, status=oco_status, raw=oco_payload, ) return self._finish(run_id, "protected", symbol, "protective oco accepted") return self._rollback(run_id, symbol, executed_qty) def _rollback(self, run_id: int, symbol: str, quantity: Decimal, reason_prefix: str = "protective oco failed") -> AutoBuyExecutionResult: rollback_client_order_id = build_client_order_id("bqab-rb") try: rollback_payload = self.exchange.create_market_sell_quantity(symbol, quantity, rollback_client_order_id) except Exception as exc: self._record_event( run_id, event_type="rollback_failed", symbol=symbol, client_order_id=rollback_client_order_id, status="CRITICAL", raw={"error_type": type(exc).__name__, "message": "rollback failed"}, ) return self._finish(run_id, "rollback_failed", symbol, f"{reason_prefix}; rollback failed") if not isinstance(rollback_payload, dict): self._record_event( run_id, event_type="rollback_failed", symbol=symbol, client_order_id=rollback_client_order_id, status="CRITICAL", raw={"message": "rollback response invalid"}, ) return self._finish(run_id, "rollback_failed", symbol, f"{reason_prefix}; rollback failed") rollback_status = _payload_status(rollback_payload) if not _valid_rollback_payload(rollback_payload, symbol, rollback_client_order_id, quantity): self._record_event( run_id, event_type="rollback_failed", symbol=symbol, client_order_id=rollback_client_order_id, status="CRITICAL", raw={"message": "rollback response invalid"}, ) return self._finish(run_id, "rollback_failed", symbol, f"{reason_prefix}; rollback failed") self._record_event( run_id, event_type="rollback_sell", symbol=symbol, client_order_id=rollback_client_order_id, status=rollback_status, raw=rollback_payload, ) record_order_fills(self.storage, symbol, rollback_payload) return self._finish(run_id, "rollback_completed", symbol, f"{reason_prefix}; rollback completed") def _fail_unprotected(self, run_id: int, symbol: str, reason: str, raw: dict) -> AutoBuyExecutionResult: self._record_event( run_id, event_type="rollback_failed", symbol=symbol, client_order_id=None, status="CRITICAL", raw=raw, ) return self._finish(run_id, "rollback_failed", symbol, reason) def _finish(self, run_id: int, status: str, symbol: str, reason: str) -> AutoBuyExecutionResult: try: self.storage.update_auto_buy_run_status(run_id, status, reason) except Exception: pass self.audit_logger.log( "auto_buy.execution", params={"symbol": symbol, "run_id": run_id}, result={"status": status, "reason": reason}, ) return AutoBuyExecutionResult(run_id=run_id, status=status, symbol=symbol, reason=reason) def _record_event( self, run_id: int, event_type: str, symbol: str | None, client_order_id: str | None, status: str | None, raw: dict, ) -> None: try: self.storage.record_auto_buy_event( run_id, event_type=event_type, symbol=symbol, client_order_id=client_order_id, status=status, raw=raw, ) except Exception: pass def _kline_to_candle(symbol: str, row: list) -> Candle: return Candle( symbol=symbol, open_time=int(row[0]), close_time=int(row[6]), open=Decimal(str(row[1])), high=Decimal(str(row[2])), low=Decimal(str(row[3])), close=Decimal(str(row[4])), volume=Decimal(str(row[5])), ) def _local_day_start_ms(now_ms: int) -> int: current = datetime.fromtimestamp(now_ms / 1000) day_start = current.replace(hour=0, minute=0, second=0, microsecond=0) return int(day_start.timestamp() * 1000) def run_live_auto_buy_once( settings: Settings, exchange_client=None, now_ms: int | None = None, ) -> AutoBuyExecutionResult: storage = Storage(settings.database_path) storage.initialize() now_ms = now_ms if now_ms is not None else int(time.time() * 1000) audit_logger = AuditLogger(_audit_log_dir(settings)) if settings.mode is not TradingMode.LIVE: run_id = storage.record_auto_buy_run( symbol=None, status="rejected", score=Decimal("0"), reason="live mode required", quote_amount=settings.auto_buy_quote_amount, created_at_ms=now_ms, ) return AutoBuyExecutionResult(run_id, "rejected", "", "live mode required") if not settings.live_auto_buyer_enabled: run_id = storage.record_auto_buy_run( symbol=None, status="rejected", score=Decimal("0"), reason="live auto buyer disabled", quote_amount=settings.auto_buy_quote_amount, created_at_ms=now_ms, ) return AutoBuyExecutionResult(run_id, "rejected", "", "live auto buyer disabled") exchange = exchange_client if exchange_client is not None else BinanceSpotClient( api_key=settings.api_key, api_secret=settings.api_secret.get_secret_value(), base_url=settings.rest_base_url, base_urls=settings.rest_base_urls, recv_window=settings.recv_window, proxy_url=settings.proxy_url, audit_logger=audit_logger, ) try: snapshots: list[AutoBuyMarketSnapshot] = [] metadata_by_symbol: dict[str, SymbolMetadata] = {} price_change_percent_24h_by_symbol: dict[str, Decimal] = {} for symbol in settings.auto_buy_symbols: normalized_symbol = symbol.upper() metadata = exchange.exchange_info(normalized_symbol) ticker = exchange.symbol_order_book_ticker(normalized_symbol) ticker_24hr = exchange.ticker_24hr(normalized_symbol) kline_rows = exchange.klines(normalized_symbol, interval="1h", limit=6) metadata_by_symbol[normalized_symbol] = metadata price_change_percent_24h_by_symbol[normalized_symbol] = Decimal(str(ticker_24hr["priceChangePercent"])) snapshots.append( AutoBuyMarketSnapshot( symbol=normalized_symbol, bid=Decimal(str(ticker["bidPrice"])), ask=Decimal(str(ticker["askPrice"])), candles=[_kline_to_candle(normalized_symbol, row) for row in kline_rows], ) ) candidate = AutoBuyScanner( settings.auto_buy_symbols, settings.auto_buy_min_score, settings.auto_buy_max_spread_bps, ).select(snapshots) audit_logger.log( "auto_buy.scan", params={"symbols": list(settings.auto_buy_symbols)}, formula=_audit_formula(), result={"candidate": _audit_candidate(candidate)}, ) if candidate is None: run_id = storage.record_auto_buy_run( symbol=None, status="no_candidate", score=Decimal("0"), reason="no candidate passed scanner", quote_amount=settings.auto_buy_quote_amount, created_at_ms=now_ms, ) return AutoBuyExecutionResult(run_id, "no_candidate", "", "no candidate passed scanner") account = exchange.account() day_start_ms = _local_day_start_ms(now_ms) recent_since_ms = now_ms - (settings.auto_buy_cooldown_seconds * 1000) decision = AutoBuyRiskEngine( AutoBuyRiskSettings( daily_limit=settings.auto_buy_daily_limit, max_spread_bps=settings.auto_buy_max_spread_bps, max_24h_rally_percent=settings.auto_buy_max_24h_rally_percent, existing_base_value_limit=settings.auto_buy_existing_base_value_limit, ) ).evaluate( risk_context := AutoBuyRiskContext( mode=settings.mode, enabled=settings.live_auto_buyer_enabled, candidate=candidate, metadata=metadata_by_symbol[candidate.symbol], account=account, daily_attempts=storage.count_auto_buy_attempts_since(day_start_ms), has_recent_symbol_attempt=storage.has_recent_auto_buy_attempt(candidate.symbol, recent_since_ms), quote_amount=settings.auto_buy_quote_amount, price_change_percent_24h=price_change_percent_24h_by_symbol[candidate.symbol], ) ) audit_logger.log( "auto_buy.risk", params={"symbol": candidate.symbol}, formula={ "spread_bps": "(ask - bid) / ((ask + bid) / 2) * 10000", "base_value": "(base_free + base_locked) * ask", "required_quote": "quote_amount + quote_reserve", "price_change_percent_24h": "ticker_24hr.priceChangePercent", }, result={ "approved": decision.approved, "reason": decision.reason, "score": candidate.score, "quote_amount": risk_context.quote_amount, "existing_base_value_limit": settings.auto_buy_existing_base_value_limit, "daily_attempts": risk_context.daily_attempts, "daily_limit": settings.auto_buy_daily_limit, "spread_bps": candidate.spread_bps, "max_spread_bps": settings.auto_buy_max_spread_bps, "price_change_percent_24h": risk_context.price_change_percent_24h, "max_24h_rally_percent": settings.auto_buy_max_24h_rally_percent, }, ) except Exception: run_id = storage.record_auto_buy_run( symbol=None, status="rejected", score=Decimal("0"), reason="pre-buy data error", quote_amount=settings.auto_buy_quote_amount, created_at_ms=now_ms, ) return AutoBuyExecutionResult(run_id, "rejected", "", "pre-buy data error") if not decision.approved: run_id = storage.record_auto_buy_run( symbol=candidate.symbol, status="rejected", score=candidate.score, reason=decision.reason, quote_amount=settings.auto_buy_quote_amount, created_at_ms=now_ms, ) return AutoBuyExecutionResult(run_id, "rejected", candidate.symbol, decision.reason) reservation = storage.reserve_auto_buy_attempt( symbol=candidate.symbol, score=candidate.score, reason=", ".join(candidate.reasons), quote_amount=settings.auto_buy_quote_amount, daily_limit=settings.auto_buy_daily_limit, day_start_ms=day_start_ms, recent_since_ms=recent_since_ms, created_at_ms=now_ms, ) if not reservation.approved: run_id = storage.record_auto_buy_run( symbol=candidate.symbol, status="rejected", score=candidate.score, reason=reservation.reason, quote_amount=settings.auto_buy_quote_amount, created_at_ms=now_ms, ) return AutoBuyExecutionResult(run_id, "rejected", candidate.symbol, reservation.reason) return AutoBuyExecutionEngine( exchange=exchange, storage=storage, stop_loss_percent=settings.auto_buy_stop_loss_percent, stop_limit_buffer_percent=settings.auto_buy_stop_limit_buffer_percent, audit_logger=audit_logger, take_profit_percent=settings.auto_buy_take_profit_percent, ).execute(candidate, metadata_by_symbol[candidate.symbol], settings.auto_buy_quote_amount, run_id=reservation.run_id)