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, SymbolMetadata, TradingMode from .spot_cost_basis import evaluate_sell_cost_guard from .spot_order_fills import record_order_fills from .storage import Storage BPS = Decimal("10000") SKIPPABLE_CANDIDATE_REJECTION_REASONS = { "symbol cooldown active", "open sell order exists", "no free balance", "quantity below min quantity", "notional below min notional", "sell below cost before minimum hold", } def _audit_log_dir(settings: Settings): return settings.database_path.parent / "logs" def _audit_candidate(candidate: AutoSellCandidate | 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": "downside flags + bearish volume spike + positive downside 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_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 _is_finite_decimal(value: object) -> bool: return isinstance(value, Decimal) and value.is_finite() 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 AutoSellMarketSnapshot: 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 AutoSellCandidate: 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 AutoSellScanner: 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[AutoSellMarketSnapshot]) -> AutoSellCandidate | None: ranked = self.rank(snapshots) return ranked[0] if ranked else None def rank(self, snapshots: Iterable[AutoSellMarketSnapshot]) -> list[AutoSellCandidate]: 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] return sorted(eligible, key=lambda candidate: (candidate.score, -self.symbols.index(candidate.symbol)), reverse=True) def _score(self, snapshot: AutoSellMarketSnapshot) -> AutoSellCandidate | 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:] opens = [candle.open for candle in recent_candles] closes = [candle.close for candle in recent_candles] volumes = [candle.volume for candle in recent_candles] if any(not value.is_finite() for value in [*opens, *closes, *volumes]): return None if any(value <= 0 for value in [*opens, *closes, *volumes]): return None score = Decimal("0") reasons: list[str] = [] if closes[-1] < closes[-2]: score += Decimal("1") reasons.append("negative 1h momentum") if closes[-1] < closes[-5]: score += Decimal("1") reasons.append("negative 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 below long SMA") baseline_volume = sum(volumes[-6:-1]) / Decimal("5") if closes[-1] < opens[-1] and volumes[-1] > baseline_volume: score += Decimal("1") reasons.append("bearish volume spike") downside = ((closes[-5] - closes[-1]) / closes[-5]) * Decimal("100") if downside > 0: score += downside.quantize(Decimal("0.0001")) return AutoSellCandidate( symbol=snapshot.symbol, score=score, reference_price=snapshot.bid, spread_bps=spread_bps, reasons=tuple(reasons), ) @dataclass(frozen=True) class AutoSellRiskSettings: daily_limit: int max_spread_bps: Decimal = Decimal("10") 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") @dataclass(frozen=True) class AutoSellRiskContext: mode: TradingMode enabled: bool candidate: AutoSellCandidate metadata: SymbolMetadata account: AccountSnapshot daily_attempts: int has_recent_symbol_attempt: bool has_open_sell_order: bool quantity: Decimal @dataclass(frozen=True) class AutoSellRiskDecision: approved: bool reason: str candidate: AutoSellCandidate class AutoSellRiskEngine: def __init__(self, settings: AutoSellRiskSettings) -> None: self.settings = settings def evaluate(self, context: AutoSellRiskContext) -> AutoSellRiskDecision: candidate = context.candidate if context.mode is not TradingMode.LIVE: return AutoSellRiskDecision(False, "live mode required", candidate) if not _is_bool(context.enabled): return AutoSellRiskDecision(False, "invalid risk input", candidate) if context.enabled is not True: return AutoSellRiskDecision(False, "live auto seller disabled", candidate) if not _is_integral_count(context.daily_attempts) or context.daily_attempts < 0: return AutoSellRiskDecision(False, "invalid risk input", candidate) if context.daily_attempts >= self.settings.daily_limit: return AutoSellRiskDecision(False, "daily auto-sell limit reached", candidate) if not _is_bool(context.has_recent_symbol_attempt): return AutoSellRiskDecision(False, "invalid risk input", candidate) if context.has_recent_symbol_attempt: return AutoSellRiskDecision(False, "symbol cooldown active", candidate) if not _has_metadata_shape(context.metadata): return AutoSellRiskDecision(False, "invalid risk input", candidate) if candidate.symbol != context.metadata.symbol: return AutoSellRiskDecision(False, "metadata symbol mismatch", candidate) if not _all_finite(candidate.score, candidate.reference_price, candidate.spread_bps): return AutoSellRiskDecision(False, "invalid risk input", candidate) if candidate.score < 0 or candidate.reference_price <= 0 or candidate.spread_bps < 0: return AutoSellRiskDecision(False, "invalid risk input", candidate) if candidate.spread_bps > self.settings.max_spread_bps: return AutoSellRiskDecision(False, "spread above max", candidate) if not _is_bool(context.has_open_sell_order): return AutoSellRiskDecision(False, "invalid risk input", candidate) if context.has_open_sell_order: return AutoSellRiskDecision(False, "open sell order exists", candidate) if not _all_finite(context.quantity, context.metadata.min_qty, context.metadata.step_size, context.metadata.min_notional): return AutoSellRiskDecision(False, "invalid risk input", candidate) if context.metadata.min_qty <= 0 or context.metadata.step_size <= 0 or context.metadata.min_notional <= 0: return AutoSellRiskDecision(False, "missing exchange filters", candidate) if context.quantity <= 0: return AutoSellRiskDecision(False, "no free balance", candidate) if context.quantity < context.metadata.min_qty: return AutoSellRiskDecision(False, "quantity below min quantity", candidate) notional = context.quantity * candidate.reference_price if notional < context.metadata.min_notional: return AutoSellRiskDecision(False, "notional below min notional", candidate) base_free, base_locked = _balance_parts(context.account, context.metadata.base_asset) if not _all_finite(base_free, base_locked): return AutoSellRiskDecision(False, "invalid risk input", candidate) if not _all_non_negative(base_free, base_locked): return AutoSellRiskDecision(False, "invalid risk input", candidate) if context.quantity > base_free: return AutoSellRiskDecision(False, "quantity exceeds free balance", candidate) return AutoSellRiskDecision(True, "approved", candidate) @dataclass(frozen=True) class AutoSellExecutionResult: 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 _payload_status(payload: dict) -> str: status = payload.get("status") if isinstance(status, str) and status: return status return "MISSING" def _sell_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 _valid_market_sell_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 if "side" in payload and payload.get("side") != "SELL": return False if "type" in payload and payload.get("type") != "MARKET": return False executed_quantity = _sell_executed_quantity(payload) return executed_quantity is not None and executed_quantity >= quantity class AutoSellExecutionEngine: def __init__(self, exchange, storage, audit_logger: AuditLogger | None = None) -> None: self.exchange = exchange self.storage = storage self.audit_logger = audit_logger or AuditLogger() def execute( self, candidate: AutoSellCandidate, metadata: SymbolMetadata, quantity: Decimal, run_id: int | None = None, ) -> AutoSellExecutionResult: symbol = candidate.symbol if run_id is None: run_id = self.storage.record_auto_sell_run( symbol=symbol, status="sell_submitted", score=candidate.score, reason=", ".join(candidate.reasons), quantity=quantity, ) else: self.storage.update_auto_sell_run_status(run_id, "sell_submitted", ", ".join(candidate.reasons)) client_order_id = build_client_order_id("bqas-sell") try: sell_payload = self.exchange.create_market_sell_quantity(symbol, quantity, client_order_id) except Exception: self._record_event( run_id, event_type="sell_failed", symbol=symbol, client_order_id=client_order_id, status="CRITICAL", raw={"message": "sell submission failed"}, ) return self._finish(run_id, "sell_failed", symbol, "sell submission failed") if not isinstance(sell_payload, dict): self._record_event( run_id, event_type="sell_failed", symbol=symbol, client_order_id=client_order_id, status="CRITICAL", raw={"message": "sell response invalid"}, ) return self._finish(run_id, "sell_failed", symbol, "invalid sell response") status = _payload_status(sell_payload) self._record_event( run_id, event_type="market_sell", symbol=symbol, client_order_id=client_order_id, status=status, raw=sell_payload, ) record_order_fills(self.storage, symbol, sell_payload) if not _valid_market_sell_payload(sell_payload, symbol, client_order_id, quantity): return self._finish(run_id, "sell_failed", symbol, "invalid sell response") return self._finish(run_id, "sold", symbol, "market sell filled") def _finish(self, run_id: int, status: str, symbol: str, reason: str) -> AutoSellExecutionResult: try: self.storage.update_auto_sell_run_status(run_id, status, reason) except Exception: pass self.audit_logger.log( "auto_sell.execution", params={"symbol": symbol, "run_id": run_id}, result={"status": status, "reason": reason}, ) return AutoSellExecutionResult(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_sell_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 _has_open_sell_order(open_orders: list[dict]) -> bool: return any(str(order.get("side", "")).upper() == "SELL" for order in open_orders) def run_live_auto_sell_once( settings: Settings, exchange_client=None, now_ms: int | None = None, ) -> AutoSellExecutionResult: 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_sell_run( symbol=None, status="rejected", score=Decimal("0"), reason="live mode required", quantity=Decimal("0"), created_at_ms=now_ms, ) return AutoSellExecutionResult(run_id, "rejected", "", "live mode required") if not settings.live_auto_seller_enabled: run_id = storage.record_auto_sell_run( symbol=None, status="rejected", score=Decimal("0"), reason="live auto seller disabled", quantity=Decimal("0"), created_at_ms=now_ms, ) return AutoSellExecutionResult(run_id, "rejected", "", "live auto seller 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: account = exchange.account() snapshots: list[AutoSellMarketSnapshot] = [] metadata_by_symbol: dict[str, SymbolMetadata] = {} open_orders_by_symbol: dict[str, list[dict]] = {} quantity_by_symbol: dict[str, Decimal] = {} had_free_balance = False for symbol in settings.auto_sell_symbols: normalized_symbol = symbol.upper() metadata = exchange.exchange_info(normalized_symbol) metadata_by_symbol[normalized_symbol] = metadata base_free, _base_locked = _balance_parts(account, metadata.base_asset) quantity = _round_down(base_free, metadata.step_size) quantity_by_symbol[normalized_symbol] = quantity if quantity <= 0: continue had_free_balance = True ticker = exchange.symbol_order_book_ticker(normalized_symbol) kline_rows = exchange.klines(normalized_symbol, interval="1h", limit=6) open_orders = exchange.open_orders(normalized_symbol) open_orders_by_symbol[normalized_symbol] = open_orders snapshots.append( AutoSellMarketSnapshot( 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], ) ) if not had_free_balance: run_id = storage.record_auto_sell_run( symbol=None, status="rejected", score=Decimal("0"), reason="no free balance", quantity=Decimal("0"), created_at_ms=now_ms, ) return AutoSellExecutionResult(run_id, "rejected", "", "no free balance") candidates = AutoSellScanner( settings.auto_sell_symbols, settings.auto_sell_min_score, settings.auto_sell_max_spread_bps, ).rank(snapshots) audit_logger.log( "auto_sell.scan", params={"symbols": list(settings.auto_sell_symbols)}, formula=_audit_formula(), result={"candidates": [_audit_candidate(candidate) for candidate in candidates]}, ) if not candidates: run_id = storage.record_auto_sell_run( symbol=None, status="no_candidate", score=Decimal("0"), reason="no candidate passed scanner", quantity=Decimal("0"), created_at_ms=now_ms, ) return AutoSellExecutionResult(run_id, "no_candidate", "", "no candidate passed scanner") day_start_ms = _local_day_start_ms(now_ms) recent_since_ms = now_ms - (settings.auto_sell_cooldown_seconds * 1000) risk_engine = AutoSellRiskEngine( AutoSellRiskSettings( daily_limit=settings.auto_sell_daily_limit, max_spread_bps=settings.auto_sell_max_spread_bps, ) ) daily_attempts = storage.count_auto_sell_attempts_since(day_start_ms) rejected: list[AutoSellRiskDecision] = [] approved_decision: AutoSellRiskDecision | None = None for candidate in candidates: decision = risk_engine.evaluate( risk_context := AutoSellRiskContext( mode=settings.mode, enabled=settings.live_auto_seller_enabled, candidate=candidate, metadata=metadata_by_symbol[candidate.symbol], account=account, daily_attempts=daily_attempts, has_recent_symbol_attempt=storage.has_recent_auto_sell_attempt(candidate.symbol, recent_since_ms), has_open_sell_order=_has_open_sell_order(open_orders_by_symbol[candidate.symbol]), quantity=quantity_by_symbol[candidate.symbol], ) ) audit_logger.log( "auto_sell.risk", params={"symbol": candidate.symbol}, formula={ "spread_bps": "(ask - bid) / ((ask + bid) / 2) * 10000", "rounded_quantity": "round_down(base_free, step_size)", "notional": "quantity * bid", }, result={ "approved": decision.approved, "reason": decision.reason, "score": candidate.score, "quantity": risk_context.quantity, "notional": risk_context.quantity * candidate.reference_price, "daily_attempts": risk_context.daily_attempts, "daily_limit": settings.auto_sell_daily_limit, "spread_bps": candidate.spread_bps, "max_spread_bps": settings.auto_sell_max_spread_bps, }, ) if decision.approved: cost_decision = evaluate_sell_cost_guard( storage, candidate.symbol, quantity=quantity_by_symbol[candidate.symbol], bid_price=candidate.reference_price, now_ms=now_ms, require_profit=settings.auto_sell_require_profit, min_hold_seconds=settings.auto_sell_min_hold_seconds, ) if not cost_decision.approved: decision = AutoSellRiskDecision(False, cost_decision.reason, candidate) audit_logger.log( "auto_sell.cost_guard", params={"symbol": candidate.symbol}, formula={"estimated_proceeds": "quantity * bid", "matched_cost": "FIFO open lot cost"}, result={ "approved": False, "reason": cost_decision.reason, "quantity": quantity_by_symbol[candidate.symbol], "bid_price": candidate.reference_price, "estimated_proceeds": cost_decision.estimated_proceeds, "matched_cost": cost_decision.matched_cost, "oldest_opened_at_ms": cost_decision.oldest_opened_at_ms, "min_hold_seconds": settings.auto_sell_min_hold_seconds, }, ) rejected.append(decision) if decision.reason not in SKIPPABLE_CANDIDATE_REJECTION_REASONS: break continue approved_decision = decision break rejected.append(decision) if decision.reason not in SKIPPABLE_CANDIDATE_REJECTION_REASONS: break except Exception: run_id = storage.record_auto_sell_run( symbol=None, status="rejected", score=Decimal("0"), reason="pre-sell data error", quantity=Decimal("0"), created_at_ms=now_ms, ) return AutoSellExecutionResult(run_id, "rejected", "", "pre-sell data error") if approved_decision is None: decision = rejected[0] candidate = decision.candidate run_id = storage.record_auto_sell_run( symbol=candidate.symbol, status="rejected", score=candidate.score, reason=decision.reason, quantity=quantity_by_symbol[candidate.symbol], created_at_ms=now_ms, ) return AutoSellExecutionResult(run_id, "rejected", candidate.symbol, decision.reason) candidate = approved_decision.candidate reservation = storage.reserve_auto_sell_attempt( symbol=candidate.symbol, score=candidate.score, reason=", ".join(candidate.reasons), quantity=quantity_by_symbol[candidate.symbol], daily_limit=settings.auto_sell_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_sell_run( symbol=candidate.symbol, status="rejected", score=candidate.score, reason=reservation.reason, quantity=quantity_by_symbol[candidate.symbol], created_at_ms=now_ms, ) return AutoSellExecutionResult(run_id, "rejected", candidate.symbol, reservation.reason) return AutoSellExecutionEngine(exchange, storage, audit_logger).execute( candidate, metadata_by_symbol[candidate.symbol], quantity_by_symbol[candidate.symbol], run_id=reservation.run_id, )