from __future__ import annotations import asyncio import json import subprocess import tempfile import time from dataclasses import dataclass, replace from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any, Callable from . import auto_buy, auto_sell from .audit_log import AuditLogger from .auto_buy import AutoBuyCandidate, AutoBuyExecutionEngine, AutoBuyRiskContext, AutoBuyRiskEngine, AutoBuyRiskSettings from .auto_sell import AutoSellCandidate, AutoSellExecutionEngine, AutoSellRiskContext, AutoSellRiskEngine, AutoSellRiskSettings from .config import Settings from .exchange import BinanceSpotClient from .market_data import sample_realtime_market from .models import AccountSnapshot, Balance, Candle, SymbolMetadata, TradingMode from .spot_cost_basis import evaluate_sell_cost_guard from .storage import Storage AI_TRADE_ACTIONS = {"BUY", "SELL", "HOLD"} BPS = Decimal("10000") AI_TRADER_INDICATOR_WARMUP_CANDLES = 50 BINANCE_KLINE_LIMIT_MAX = 1000 _AI_TRADE_EXECUTED_STATUSES = {"protected", "rollback_completed", "rollback_failed", "sold", "sell_failed"} _AI_TRADE_FAILED_STATUSES = {"rollback_completed", "rollback_failed", "sell_failed"} _PROTECTIVE_STOP_CLIENT_ORDER_PREFIX = "bqab-stop-" _PROTECTIVE_TAKE_PROFIT_CLIENT_ORDER_PREFIX = "bqab-tp-" _LOG_TEXT_TAIL_BYTES = 2048 _AI_REASON_TRANSLATIONS = { "live mode required": "需要 live 模式", "live ai trader disabled": "AI实盘交易开关未开启", "pre-ai-trade data error": "AI交易前置数据获取失败", "ai confidence below minimum": "AI置信度低于最小阈值", "model symbol unavailable": "模型返回的交易对不可用", "invalid model response": "模型响应无效", "codex exec timed out": "codex exec 超时", "codex exec failed": "codex exec 执行失败", "approved": "已通过风控", "live auto buyer disabled": "AI实盘交易开关未开启", "live auto seller disabled": "AI实盘交易开关未开启", "invalid risk input": "风控输入无效", "daily auto-buy limit reached": "已达到每日自动买入次数上限", "daily auto-sell limit reached": "已达到每日自动卖出次数上限", "symbol cooldown active": "交易对冷却中", "metadata symbol mismatch": "交易规则与交易对不匹配", "spread above max": "买卖价差超过上限", "quote amount below min notional": "买入金额低于最小下单金额", "insufficient quote balance": "USDT余额不足", "existing base balance": "已持有基础资产", "24h rally above max": "24小时涨幅过高", "missing exchange filters": "交易所过滤规则缺失", "open sell order exists": "已有卖出挂单", "protective stop cancel failed": "保护性止损单取消失败", "no free balance": "无可用持仓", "quantity below min quantity": "卖出数量低于最小下单数量", "notional below min notional": "名义金额低于最小下单金额", "quantity exceeds free balance": "卖出数量超过可用持仓", "protective oco accepted": "保护性OCO止盈止损单已接受", "protective oco calculation failed": "保护性OCO价格计算失败", "protective oco failed; rollback completed": "保护性OCO失败，回滚卖出已完成", "protective oco failed; rollback failed": "保护性OCO失败，回滚卖出失败", "protective oco calculation failed; rollback completed": "保护性OCO价格计算失败，回滚卖出已完成", "protective oco calculation failed; rollback failed": "保护性OCO价格计算失败，回滚卖出失败", "protective stop accepted": "保护性止损单已接受", "buy submission failed": "买入提交失败", "invalid buy response": "买入响应无效", "invalid fill quantities": "成交数量无效", "protective stop calculation failed": "保护性止损价格计算失败", "protective stop failed; rollback completed": "保护性止损失败，回滚卖出已完成", "protective stop failed; rollback failed": "保护性止损失败，回滚卖出失败", "invalid fill quantities; rollback completed": "成交数量无效，回滚卖出已完成", "invalid fill quantities; rollback failed": "成交数量无效，回滚卖出失败", "protective stop calculation failed; rollback completed": "保护性止损价格计算失败，回滚卖出已完成", "protective stop calculation failed; rollback failed": "保护性止损价格计算失败，回滚卖出失败", "market sell filled": "市价卖出已成交", "sell submission failed": "卖出提交失败", "invalid sell response": "卖出响应无效", } def is_ai_trade_executed_status(status: str) -> bool: return status in _AI_TRADE_EXECUTED_STATUSES @dataclass(frozen=True) class AiTradePrediction: symbol: str action: str confidence: Decimal reason: str def __post_init__(self) -> None: object.__setattr__(self, "symbol", self.symbol.upper()) object.__setattr__(self, "action", self.action.upper()) @classmethod def hold(cls, reason: str, symbol: str = "") -> "AiTradePrediction": return cls(symbol=symbol, action="HOLD", confidence=Decimal("0"), reason=reason) @classmethod def from_json(cls, text: str, allowed_symbols: tuple[str, ...]) -> "AiTradePrediction": try: payload = json.loads(text) except json.JSONDecodeError: return cls.hold("invalid model response") if not isinstance(payload, dict): return cls.hold("invalid model response") symbol = str(payload.get("symbol", "")).strip().upper() action = str(payload.get("action", "")).strip().upper() reason = str(payload.get("reason", "")).strip() try: confidence = Decimal(str(payload.get("confidence", ""))) except (InvalidOperation, ValueError): return cls.hold("invalid model response") allowed = tuple(item.upper() for item in allowed_symbols) if action not in AI_TRADE_ACTIONS: return cls.hold("invalid model response") if action in {"BUY", "SELL"} and symbol not in allowed: return cls.hold("invalid model response") if action == "HOLD": symbol = symbol if symbol in allowed else "" if not confidence.is_finite() or confidence < 0 or confidence > 1: return cls.hold("invalid model response") if not reason: return cls.hold("invalid model response") return cls(symbol=symbol, action=action, confidence=confidence, reason=reason) @classmethod def from_json_list(cls, text: str, allowed_symbols: tuple[str, ...]) -> list["AiTradePrediction"]: allowed = tuple(item.upper() for item in allowed_symbols) try: payload = json.loads(text) except json.JSONDecodeError: return cls.holds_for_symbols(allowed, "模型响应无效") if not isinstance(payload, dict): return cls.holds_for_symbols(allowed, "模型响应无效") predictions_payload = payload.get("predictions") if not isinstance(predictions_payload, list): return cls.holds_for_symbols(allowed, "模型响应无效") by_symbol: dict[str, AiTradePrediction] = {} invalid_symbols: set[str] = set() for item in predictions_payload: symbol = _payload_symbol(item, allowed) prediction = cls.from_payload(item, allowed, require_symbol=True) if prediction is None: if symbol: invalid_symbols.add(symbol) continue by_symbol.setdefault(prediction.symbol, prediction) predictions: list[AiTradePrediction] = [] for symbol in allowed: if symbol in by_symbol: predictions.append(by_symbol[symbol]) elif symbol in invalid_symbols: predictions.append(cls.hold("模型返回的该交易对预测无效", symbol)) else: predictions.append(cls.hold("模型未返回该交易对预测", symbol)) return predictions @classmethod def from_payload( cls, payload: object, allowed_symbols: tuple[str, ...], *, require_symbol: bool, ) -> "AiTradePrediction | None": if not isinstance(payload, dict): return None allowed = tuple(item.upper() for item in allowed_symbols) symbol = str(payload.get("symbol", "")).strip().upper() action = str(payload.get("action", "")).strip().upper() reason = str(payload.get("reason", "")).strip() try: confidence = Decimal(str(payload.get("confidence", ""))) except (InvalidOperation, ValueError): return None if action not in AI_TRADE_ACTIONS: return None if require_symbol and symbol not in allowed: return None if action in {"BUY", "SELL"} and symbol not in allowed: return None if action == "HOLD" and symbol not in allowed: symbol = "" if not confidence.is_finite() or confidence < 0 or confidence > 1: return None if not reason: return None return cls(symbol=symbol, action=action, confidence=confidence, reason=reason) @classmethod def holds_for_symbols(cls, allowed_symbols: tuple[str, ...], reason: str) -> list["AiTradePrediction"]: return [cls.hold(reason, symbol) for symbol in allowed_symbols] class CodexExecPredictor: def __init__( self, model: str | None = None, codex_command: str = "codex", timeout_seconds: int = 60, cwd: str | Path | None = None, log_dir: str | Path | None = None, log_max_bytes: int = 20 * 1024 * 1024, log_backup_count: int = 5, run_command: Callable[..., subprocess.CompletedProcess] = subprocess.run, ) -> None: self.model = model self.codex_command = codex_command self.timeout_seconds = timeout_seconds self.cwd = Path(cwd) if cwd is not None else Path.cwd() self.log_dir = Path(log_dir) if log_dir is not None else self.cwd / "logs" self.log_max_bytes = log_max_bytes self.log_backup_count = log_backup_count self.run_command = run_command self.audit_logger = AuditLogger(self.log_dir) def predict(self, context: dict[str, Any], allowed_symbols: tuple[str, ...]) -> list[AiTradePrediction]: schema_path = self._write_schema(allowed_symbols) command = [self.codex_command, "exec", "--output-schema", str(schema_path)] if self.model: command.extend(["--model", self.model]) prompt = _build_prompt(context) started_at = time.perf_counter() try: completed = self.run_command( command, input=prompt, cwd=self.cwd, timeout=self.timeout_seconds, capture_output=True, text=True, encoding="utf-8", errors="replace", ) except subprocess.TimeoutExpired: duration_ms = _elapsed_ms(started_at) self._write_call_log( context=context, allowed_symbols=allowed_symbols, prompt=prompt, status="timeout", duration_ms=duration_ms, returncode=None, stdout="", stderr="", error="codex exec timed out", ) self._write_audit_summary( allowed_symbols=allowed_symbols, status="timeout", duration_ms=duration_ms, returncode=None, stdout="", stderr="", error="codex exec timed out", ) return AiTradePrediction.holds_for_symbols(allowed_symbols, "codex exec 超时") except OSError as exc: duration_ms = _elapsed_ms(started_at) self._write_call_log( context=context, allowed_symbols=allowed_symbols, prompt=prompt, status="error", duration_ms=duration_ms, returncode=None, stdout="", stderr="", error=f"codex exec failed: {type(exc).__name__}", ) self._write_audit_summary( allowed_symbols=allowed_symbols, status="error", duration_ms=duration_ms, returncode=None, stdout="", stderr="", error=f"codex exec failed: {type(exc).__name__}", ) return AiTradePrediction.holds_for_symbols(allowed_symbols, "codex exec 执行失败") finally: try: schema_path.unlink(missing_ok=True) except OSError: pass status = "completed" if completed.returncode == 0 else "failed" duration_ms = _elapsed_ms(started_at) self._write_call_log( context=context, allowed_symbols=allowed_symbols, prompt=prompt, status=status, duration_ms=duration_ms, returncode=completed.returncode, stdout=completed.stdout, stderr=completed.stderr, error="", ) self._write_audit_summary( allowed_symbols=allowed_symbols, status=status, duration_ms=duration_ms, returncode=completed.returncode, stdout=completed.stdout, stderr=completed.stderr, error="", ) if completed.returncode != 0: return AiTradePrediction.holds_for_symbols(allowed_symbols, "codex exec 执行失败") return AiTradePrediction.from_json_list(completed.stdout.strip(), allowed_symbols) def _write_audit_summary( self, *, allowed_symbols: tuple[str, ...], status: str, duration_ms: int, returncode: int | None, stdout: str, stderr: str, error: str, ) -> None: try: prediction_count = 0 if stdout.strip(): payload = json.loads(stdout) predictions = payload.get("predictions") if isinstance(payload, dict) else None prediction_count = len(predictions) if isinstance(predictions, list) else 0 except json.JSONDecodeError: prediction_count = 0 self.audit_logger.log( "ai_trader.llm_call", params={ "model": self.model or "-", "allowed_symbols": list(allowed_symbols), "timeout_seconds": self.timeout_seconds, }, result={ "status": status, "returncode": returncode, "duration_ms": duration_ms, "prediction_count": prediction_count, "stdout_bytes": len(stdout.encode("utf-8", errors="replace")), "stderr_bytes": len(stderr.encode("utf-8", errors="replace")), }, error=error or None, level="ERROR" if status in {"failed", "timeout", "error"} else "INFO", ) def _write_call_log( self, *, context: dict[str, Any], allowed_symbols: tuple[str, ...], prompt: str, status: str, duration_ms: int, returncode: int | None, stdout: str, stderr: str, error: str, ) -> None: self.log_dir.mkdir(parents=True, exist_ok=True) log_path = self.log_dir / "ai_trader_llm.log" model = self.model or "-" return_code = "-" if returncode is None else str(returncode) lines = [ "=" * 80, "Codex Exec AI Trader Call", f"Time: {datetime.now(timezone.utc).isoformat()}", f"Model: {model}", f"Allowed symbols: {', '.join(allowed_symbols)}", f"Timeout seconds: {self.timeout_seconds}", f"Status: {status}", f"Return code: {return_code}", f"Duration ms: {duration_ms}", ] if error: lines.extend(["", "Error:", error]) lines.extend( [ "", "Context:", json.dumps(context, ensure_ascii=False, sort_keys=True, default=str, indent=2), "", "Prompt:", prompt, "", "Stdout:", _readable_log_text(stdout), "", "Stderr:", _readable_log_text(stderr), "", ] ) log_entry = "\n".join(lines) + "\n" self._rotate_call_log_if_needed(log_path, log_entry) with log_path.open("a", encoding="utf-8", errors="replace") as handle: handle.write(log_entry) def _rotate_call_log_if_needed(self, log_path: Path, log_entry: str) -> None: if self.log_max_bytes <= 0 or self.log_backup_count <= 0: return entry_size = len(log_entry.encode("utf-8", errors="replace")) if not log_path.exists() or log_path.stat().st_size + entry_size <= self.log_max_bytes: return oldest = log_path.with_name(f"{log_path.name}.{self.log_backup_count}") if oldest.exists(): oldest.unlink() for index in range(self.log_backup_count - 1, 0, -1): source = log_path.with_name(f"{log_path.name}.{index}") if source.exists(): source.replace(log_path.with_name(f"{log_path.name}.{index + 1}")) log_path.replace(log_path.with_name(f"{log_path.name}.1")) def _write_schema(self, allowed_symbols: tuple[str, ...]) -> Path: handle = tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".schema.json", delete=False) with handle: json.dump(_prediction_schema(allowed_symbols), handle, sort_keys=True) return Path(handle.name) def _prediction_schema(allowed_symbols: tuple[str, ...] = ()) -> dict[str, Any]: symbols = [symbol.upper() for symbol in allowed_symbols] symbol_schema: dict[str, Any] = {"type": "string"} if symbols: symbol_schema["enum"] = symbols predictions_schema: dict[str, Any] = { "type": "array", "minItems": len(symbols) if symbols else 1, "items": { "type": "object", "additionalProperties": False, "required": ["symbol", "action", "confidence", "reason"], "properties": { "symbol": symbol_schema, "action": {"type": "string", "enum": ["BUY", "SELL", "HOLD"]}, "confidence": {"type": "string", "pattern": r"^(0(\.\d{1,4})?|1(\.0{1,4})?)$"}, "reason": {"type": "string", "minLength": 1}, }, }, } if symbols: predictions_schema["maxItems"] = len(symbols) return { "type": "object", "additionalProperties": False, "required": ["predictions"], "properties": { "predictions": predictions_schema, }, } def _build_prompt(context: dict[str, Any]) -> str: return "\n".join( [ "You are a market-direction classifier for a local Binance Spot trading program.", "Return one JSON object only, matching the supplied schema.", "The object must contain a predictions array with exactly one item for every symbol in context.symbols.", "Do not place trades, call tools, edit files, or include markdown.", "Use market[].summary first; raw timeframe arrays are only supporting evidence.", "Use market[].execution to consider current spot balances, estimated position notional, and open sell order counts.", "Visible timeframe arrays contain the recent display window; indicators may use additional warm-up candles described by indicator_metadata.", "Choose BUY when price trend and momentum are bullish, volume or order-flow confirms demand, and higher timeframes do not strongly disagree.", "Choose SELL when price trend and momentum are bearish, volume or order-flow confirms supply, and higher timeframes do not strongly disagree.", "Choose HOLD when evidence is mixed, weak, stale, or only supported by one isolated signal.", "Use confidence from 0.00 to 1.00.", "Use confidence at or above 0.70 when at least three independent evidence groups support the same direction and there is no strong contradiction from higher timeframes.", "Reserve 0.85 or above for broad agreement across trend, momentum, volume, 24h context, and order-flow evidence.", "When realtime.status is unavailable, aggregate_trades.status is unavailable, or market data is stale, lower confidence unless other evidence is very strong.", "Base the direction mainly on trend, momentum, volatility, volume confirmation, breakout or breakdown, and 24h context.", "Use liquidity, order-book imbalance, and aggregate trades only as secondary confirmation of market pressure.", "Every reason must be written in Chinese, and reason must describe evidence in natural Chinese.", "Do not write raw JSON field names in reason; say things like 1小时RSI偏高、4小时MACD转弱、成交量高于均值、盘口买卖力量失衡、或24小时涨跌幅较大。", "Market context:", json.dumps(context, sort_keys=True, default=str), ] ) def _payload_symbol(payload: object, allowed_symbols: tuple[str, ...]) -> str: if not isinstance(payload, dict): return "" symbol = str(payload.get("symbol", "")).strip().upper() return symbol if symbol in allowed_symbols else "" @dataclass(frozen=True) class AiTradeExecutionResult: run_id: int | None status: str symbol: str action: str reason: str confidence: Decimal | None = None def __post_init__(self) -> None: object.__setattr__(self, "symbol", self.symbol.upper()) object.__setattr__(self, "action", self.action.upper()) if self.confidence is not None and not isinstance(self.confidence, Decimal): object.__setattr__(self, "confidence", Decimal(str(self.confidence))) @dataclass(frozen=True) class AiTradeBatchExecutionResult: status: str reason: str max_actions: int prediction_count: int executed_count: int results: tuple[AiTradeExecutionResult, ...] @dataclass(frozen=True) class _SymbolContext: symbol: str metadata: SymbolMetadata bid: Decimal ask: Decimal spread_bps: Decimal best_bid_qty: Decimal best_ask_qty: Decimal candles_by_interval: dict[str, list[Candle]] timeframe_display_limits: dict[str, int] open_orders: list[dict] depth: dict[str, Any] ticker_24hr: dict[str, Any] avg_price: dict[str, Any] aggregate_trades: list[dict[str, Any]] aggregate_trades_available: bool = True def run_live_ai_trade_once( settings: Settings, exchange_client=None, predictor: CodexExecPredictor | None = None, now_ms: int | None = None, ) -> AiTradeExecutionResult: 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: reason = "需要 live 模式" run_id = _record_ai_trade_rejection(storage, reason, now_ms) return AiTradeExecutionResult(run_id, "rejected", "", "HOLD", reason) if not settings.live_ai_trader_enabled: reason = "AI实盘交易开关未开启" run_id = _record_ai_trade_rejection(storage, reason, now_ms) return AiTradeExecutionResult(run_id, "rejected", "", "HOLD", reason) 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, ) predictor = predictor if predictor is not None else CodexExecPredictor( model=settings.ai_trader_model, codex_command=settings.ai_trader_codex_command, timeout_seconds=settings.ai_trader_codex_timeout_seconds, cwd=Path.cwd(), ) try: account = exchange.account() contexts = _load_symbol_contexts( exchange, settings.ai_trader_symbols, settings.ai_trader_klines, settings.ai_trader_agg_trades_limit, ) context_by_symbol = {context.symbol: context for context in contexts} realtime = _load_realtime_context(settings) model_context = _model_context(settings, contexts, account, realtime, now_ms) audit_logger.log( "ai_trader.context", params={"symbols": list(settings.ai_trader_symbols)}, result=_audit_context(model_context), ) raw_predictions = predictor.predict(model_context, settings.ai_trader_symbols) except Exception: reason = "AI交易前置数据获取失败" run_id = _record_ai_trade_rejection(storage, reason, now_ms) return AiTradeExecutionResult(run_id, "rejected", "", "HOLD", reason) predictions = _prediction_list(raw_predictions, settings.ai_trader_symbols) audit_logger.log("ai_trader.predictions", result=_audit_predictions(predictions)) prediction_rows = _record_ai_trade_predictions(storage, predictions, now_ms) candidates = _actionable_prediction_rows(prediction_rows, settings.ai_trader_symbols) first_rejection: AiTradeExecutionResult | None = None for prediction, ai_run_id in candidates: if prediction.confidence < settings.ai_trader_min_confidence: reason = _ai_trade_reason( prediction, ( "AI置信度低于最小阈值" f"（当前 {prediction.confidence}，最小 {settings.ai_trader_min_confidence}）" ), ) storage.update_ai_trade_run_status( ai_run_id, status="rejected", reason=reason, raw=_prediction_raw(prediction), ) audit_logger.log( "ai_trader.risk", params={"action": prediction.action, "symbol": prediction.symbol, "ai_run_id": ai_run_id}, formula=_audit_risk_formula(), result={ "approved": False, "reason": "ai confidence below minimum", "confidence": prediction.confidence, "min_confidence": settings.ai_trader_min_confidence, }, ) first_rejection = first_rejection or AiTradeExecutionResult( ai_run_id, "rejected", prediction.symbol, prediction.action, reason, prediction.confidence, ) continue symbol_context = context_by_symbol.get(prediction.symbol) if symbol_context is None: reason = _ai_trade_reason(prediction, "模型返回的交易对不可用") storage.update_ai_trade_run_status( ai_run_id, status="rejected", reason=reason, raw=_prediction_raw(prediction), ) audit_logger.log( "ai_trader.risk", params={"action": prediction.action, "symbol": prediction.symbol, "ai_run_id": ai_run_id}, formula=_audit_risk_formula(), result={"approved": False, "reason": "model symbol unavailable"}, ) first_rejection = first_rejection or AiTradeExecutionResult( ai_run_id, "rejected", prediction.symbol, prediction.action, reason, prediction.confidence, ) continue if prediction.action == "BUY": result = _execute_ai_buy( settings, storage, exchange, account, symbol_context, prediction, ai_run_id, now_ms, audit_logger, ) else: result = _execute_ai_sell( settings, storage, exchange, account, symbol_context, prediction, ai_run_id, now_ms, audit_logger, ) if result.status in _AI_TRADE_EXECUTED_STATUSES: _skip_remaining_ai_predictions(storage, prediction_rows, executed_run_id=ai_run_id) return result first_rejection = first_rejection or result if first_rejection is not None: return first_rejection _, first_run_id = prediction_rows[0] return AiTradeExecutionResult(first_run_id, "no_action", "", "HOLD", "所有交易对均为 HOLD") def run_live_ai_trade_batch_once( settings: Settings, exchange_client=None, predictor: CodexExecPredictor | None = None, now_ms: int | None = None, ) -> AiTradeBatchExecutionResult: storage = Storage(settings.database_path) storage.initialize() now_ms = now_ms if now_ms is not None else int(time.time() * 1000) max_actions = settings.ai_trader_max_actions_per_run audit_logger = AuditLogger(_audit_log_dir(settings)) if settings.mode is not TradingMode.LIVE: reason = "需要 live 模式" run_id = _record_ai_trade_rejection(storage, reason, now_ms) result = AiTradeExecutionResult(run_id, "rejected", "", "HOLD", reason) return AiTradeBatchExecutionResult("rejected", reason, max_actions, 0, 0, (result,)) if not settings.live_ai_trader_enabled: reason = "AI实盘交易开关未开启" run_id = _record_ai_trade_rejection(storage, reason, now_ms) result = AiTradeExecutionResult(run_id, "rejected", "", "HOLD", reason) return AiTradeBatchExecutionResult("rejected", reason, max_actions, 0, 0, (result,)) 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, ) predictor = predictor if predictor is not None else CodexExecPredictor( model=settings.ai_trader_model, codex_command=settings.ai_trader_codex_command, timeout_seconds=settings.ai_trader_codex_timeout_seconds, cwd=Path.cwd(), ) try: account = exchange.account() contexts = _load_symbol_contexts( exchange, settings.ai_trader_symbols, settings.ai_trader_klines, settings.ai_trader_agg_trades_limit, ) realtime = _load_realtime_context(settings) model_context = _model_context(settings, contexts, account, realtime, now_ms) audit_logger.log( "ai_trader.context", params={"symbols": list(settings.ai_trader_symbols)}, result=_audit_context(model_context), ) raw_predictions = predictor.predict(model_context, settings.ai_trader_symbols) except Exception: reason = "AI交易前置数据获取失败" run_id = _record_ai_trade_rejection(storage, reason, now_ms) result = AiTradeExecutionResult(run_id, "rejected", "", "HOLD", reason) return AiTradeBatchExecutionResult("rejected", reason, max_actions, 0, 0, (result,)) predictions = _prediction_list(raw_predictions, settings.ai_trader_symbols) audit_logger.log("ai_trader.predictions", result=_audit_predictions(predictions)) prediction_rows = _record_ai_trade_predictions(storage, predictions, now_ms) candidates = _actionable_prediction_rows(prediction_rows, settings.ai_trader_symbols) results: list[AiTradeExecutionResult] = [] processed_run_ids: set[int] = set() executed_count = 0 for prediction, ai_run_id in candidates: if executed_count >= max_actions: _skip_batch_limit_ai_predictions(storage, prediction_rows, processed_run_ids=processed_run_ids) break processed_run_ids.add(ai_run_id) if prediction.confidence < settings.ai_trader_min_confidence: reason = _ai_trade_reason( prediction, ( "AI置信度低于最小阈值" f"（当前 {prediction.confidence}，最小 {settings.ai_trader_min_confidence}）" ), ) storage.update_ai_trade_run_status( ai_run_id, status="rejected", reason=reason, raw=_prediction_raw(prediction), ) audit_logger.log( "ai_trader.risk", params={"action": prediction.action, "symbol": prediction.symbol, "ai_run_id": ai_run_id}, formula=_audit_risk_formula(), result={ "approved": False, "reason": "ai confidence below minimum", "confidence": prediction.confidence, "min_confidence": settings.ai_trader_min_confidence, }, ) results.append( AiTradeExecutionResult( ai_run_id, "rejected", prediction.symbol, prediction.action, reason, prediction.confidence, ) ) continue account = exchange.account() symbol_context = _load_symbol_context( exchange, prediction.symbol, settings.ai_trader_klines, settings.ai_trader_agg_trades_limit, ) if prediction.action == "BUY": result = _execute_ai_buy( settings, storage, exchange, account, symbol_context, prediction, ai_run_id, now_ms, audit_logger, ) else: result = _execute_ai_sell( settings, storage, exchange, account, symbol_context, prediction, ai_run_id, now_ms, audit_logger, ) results.append(result) if result.status in _AI_TRADE_EXECUTED_STATUSES: executed_count += 1 status, reason = _ai_trade_batch_status(results, executed_count) results.extend(_hold_prediction_results(prediction_rows)) return AiTradeBatchExecutionResult( status, reason, max_actions, len(prediction_rows), executed_count, tuple(results), ) def _hold_prediction_results(prediction_rows: list[tuple[AiTradePrediction, int]]) -> list[AiTradeExecutionResult]: return [ AiTradeExecutionResult(run_id, "no_action", prediction.symbol, prediction.action, prediction.reason, prediction.confidence) for prediction, run_id in prediction_rows if prediction.action == "HOLD" ] def _ai_trade_batch_status(results: list[AiTradeExecutionResult], executed_count: int) -> tuple[str, str]: if any(result.status in _AI_TRADE_FAILED_STATUSES for result in results): return "failed", "AI批量交易存在失败" if executed_count: return "completed", "AI批量交易完成" if results: return "rejected", "所有可执行AI预测均被拒绝" return "no_action", "所有交易对均为 HOLD" def _load_symbol_contexts( exchange, symbols: tuple[str, ...], kline_specs: tuple[tuple[str, int], ...], agg_trades_limit: int = 500, ) -> list[_SymbolContext]: return [_load_symbol_context(exchange, symbol, kline_specs, agg_trades_limit) for symbol in symbols] def _audit_log_dir(settings: Settings) -> Path: return Path(settings.database_path).parent / "logs" def _audit_predictions(predictions: list[AiTradePrediction]) -> list[dict[str, str]]: return [_prediction_raw(prediction) for prediction in predictions] def _audit_context(context: dict[str, Any]) -> dict[str, Any]: market_summary = [] for item in context.get("market", []): if not isinstance(item, dict): continue market_summary.append( { "symbol": item.get("symbol"), "bid": item.get("bid"), "ask": item.get("ask"), "spread_bps": item.get("spread_bps"), "timeframes": { interval: { "candle_count": len(features.get("closes", [])) if isinstance(features, dict) else 0, "latest_close": ( features.get("closes", [None])[-1] if isinstance(features, dict) and features.get("closes") else None ), } for interval, features in (item.get("timeframes") or {}).items() }, "liquidity": item.get("liquidity"), "summary": item.get("summary"), "realtime_status": (item.get("realtime") or {}).get("status") if isinstance(item.get("realtime"), dict) else None, } ) return { "symbols": context.get("symbols", []), "market": market_summary, } def _audit_risk_formula() -> dict[str, str]: return { "spread_bps": "(ask - bid) / ((ask + bid) / 2) * 10000", "price_change_percent_24h": "ticker_24hr.priceChangePercent", "buy_base_value": "(base_free + base_locked) * ask", "sell_notional": "rounded_quantity * bid", } def _load_symbol_context( exchange, symbol: str, kline_specs: tuple[tuple[str, int], ...], agg_trades_limit: int = 500, ) -> _SymbolContext: normalized_symbol = symbol.upper() metadata = exchange.exchange_info(normalized_symbol) ticker = exchange.symbol_order_book_ticker(normalized_symbol) bid = Decimal(str(ticker["bidPrice"])) ask = Decimal(str(ticker["askPrice"])) best_bid_qty = Decimal(str(ticker.get("bidQty", "0"))) best_ask_qty = Decimal(str(ticker.get("askQty", "0"))) midpoint = (bid + ask) / Decimal("2") spread_bps = ((ask - bid) / midpoint) * BPS depth = exchange.depth(normalized_symbol, limit=20) ticker_24hr = exchange.ticker_24hr(normalized_symbol) avg_price = exchange.avg_price(normalized_symbol) try: aggregate_trades = exchange.aggregate_trades(normalized_symbol, limit=agg_trades_limit) aggregate_trades_available = True except Exception: aggregate_trades = [] aggregate_trades_available = False timeframe_display_limits = {interval: limit for interval, limit in kline_specs} candles_by_interval = { interval: [ _kline_to_candle(normalized_symbol, row) for row in exchange.klines( normalized_symbol, interval=interval, limit=_indicator_calculation_limit(limit), ) ] for interval, limit in kline_specs } open_orders = exchange.open_orders(normalized_symbol) return _SymbolContext( symbol=normalized_symbol, metadata=metadata, bid=bid, ask=ask, spread_bps=spread_bps, best_bid_qty=best_bid_qty, best_ask_qty=best_ask_qty, candles_by_interval=candles_by_interval, timeframe_display_limits=timeframe_display_limits, open_orders=open_orders, depth=depth, ticker_24hr=ticker_24hr, avg_price=avg_price, aggregate_trades=aggregate_trades, aggregate_trades_available=aggregate_trades_available, ) def _indicator_calculation_limit(display_limit: int) -> int: return min(display_limit + AI_TRADER_INDICATOR_WARMUP_CANDLES, BINANCE_KLINE_LIMIT_MAX) def _is_sell_order(order: dict) -> bool: return str(order.get("side", "")).upper() == "SELL" def _is_bot_protective_stop_order(order: dict) -> bool: if not _is_sell_order(order): return False order_type = str(order.get("type", "")).upper() client_order_id = str(order.get("clientOrderId", "")) if order_type == "STOP_LOSS_LIMIT" and client_order_id.startswith(_PROTECTIVE_STOP_CLIENT_ORDER_PREFIX): return True return order_type == "LIMIT_MAKER" and client_order_id.startswith(_PROTECTIVE_TAKE_PROFIT_CLIENT_ORDER_PREFIX) def _non_protective_sell_orders(open_orders: list[dict]) -> list[dict]: return [order for order in open_orders if _is_sell_order(order) and not _is_bot_protective_stop_order(order)] def _bot_protective_stop_orders(open_orders: list[dict]) -> list[dict]: return [order for order in open_orders if _is_bot_protective_stop_order(order)] def _account_with_unlocked_protective_stops( account: AccountSnapshot, metadata: SymbolMetadata, ) -> AccountSnapshot: balance = account.balances.get(metadata.base_asset) if balance is None or balance.locked <= 0: return account balances = dict(account.balances) balances[metadata.base_asset] = Balance( metadata.base_asset, balance.free + balance.locked, Decimal("0"), ) return AccountSnapshot(balances) def _empty_realtime_features(sample_seconds: int, status: str) -> dict[str, Any]: return { "status": status, "sample_seconds": sample_seconds, "bookticker_updates": 0, "depth_updates": 0, "agg_trade_updates": 0, "spread_min_bps": None, "spread_max_bps": None, "spread_avg_bps": None, "best_bid_qty_avg": None, "best_ask_qty_avg": None, "depth_bid_notional_avg": None, "depth_ask_notional_avg": None, "book_imbalance_avg": None, "agg_trade_buy_quote_volume": "0", "agg_trade_sell_quote_volume": "0", "agg_trade_buy_sell_ratio": None, "large_trade_count": 0, "large_trade_quote_sum": "0", } def _load_realtime_context(settings: Settings) -> dict[str, dict[str, Any]]: if not settings.ai_trader_ws_enabled: return { symbol: _empty_realtime_features(settings.ai_trader_ws_sample_seconds, "disabled") for symbol in settings.ai_trader_symbols } try: return asyncio.run( sample_realtime_market( settings.ai_trader_symbols, stream_base_url=settings.stream_base_url, sample_seconds=settings.ai_trader_ws_sample_seconds, ) ) except Exception: return { symbol: _empty_realtime_features(settings.ai_trader_ws_sample_seconds, "unavailable") for symbol in settings.ai_trader_symbols } def _elapsed_ms(started_at: float) -> int: return int((time.perf_counter() - started_at) * 1000) def _readable_log_text(value: str) -> str: if not value: return "-" value = _truncate_log_text(value) stripped = value.strip() try: payload = json.loads(stripped) except json.JSONDecodeError: return value return json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str, indent=2) def _truncate_log_text(value: str, max_bytes: int = _LOG_TEXT_TAIL_BYTES) -> str: encoded = value.encode("utf-8", errors="replace") if len(encoded) <= max_bytes: return value tail = encoded[-max_bytes:].decode("utf-8", errors="replace") return f"[truncated {len(encoded) - max_bytes} bytes; showing last {max_bytes} bytes]\n{tail}" def _decimal_text(value: Decimal | None) -> str | None: if value is None: return None return format(value.normalize(), "f") def _decimal_from_payload(payload: dict[str, Any], field: str) -> Decimal: try: value = Decimal(str(payload[field])) except (KeyError, InvalidOperation, ValueError): return Decimal("NaN") return value def _average(values: list[Decimal]) -> Decimal | None: if not values: return None return sum(values, Decimal("0")) / Decimal(len(values)) def _sma(values: list[Decimal], period: int) -> Decimal | None: if len(values) < period: return None return _average(values[-period:]) def _ema_series(values: list[Decimal], period: int) -> list[Decimal]: if not values: return [] multiplier = Decimal("2") / Decimal(period + 1) ema_values = [values[0]] for value in values[1:]: ema_values.append((value - ema_values[-1]) * multiplier + ema_values[-1]) return ema_values def _ema(values: list[Decimal], period: int) -> Decimal | None: if len(values) < period: return None return _ema_series(values, period)[-1] def _rsi(values: list[Decimal], period: int = 14) -> Decimal | None: if len(values) <= period: return None changes = [current - previous for previous, current in zip(values[:-1], values[1:])] gains: list[Decimal] = [] losses: list[Decimal] = [] for change in changes: gains.append(max(change, Decimal("0"))) losses.append(abs(min(change, Decimal("0")))) average_gain = _average(gains[:period]) average_loss = _average(losses[:period]) if average_gain is None or average_loss is None: return None for gain, loss in zip(gains[period:], losses[period:]): average_gain = ((average_gain * Decimal(period - 1)) + gain) / Decimal(period) average_loss = ((average_loss * Decimal(period - 1)) + loss) / Decimal(period) if average_loss == 0: return Decimal("100") relative_strength = average_gain / average_loss return Decimal("100") - (Decimal("100") / (Decimal("1") + relative_strength)) def _macd(values: list[Decimal]) -> tuple[Decimal | None, Decimal | None, Decimal | None]: if len(values) < 26: return None, None, None ema_12 = _ema_series(values, 12) ema_26 = _ema_series(values, 26) macd_values = [short - long for short, long in zip(ema_12[-len(ema_26) :], ema_26)] current_macd = macd_values[-1] if len(macd_values) < 9: return current_macd, None, None signal = _ema_series(macd_values, 9)[-1] return current_macd, signal, current_macd - signal def _atr(candles: list[Candle], period: int = 14) -> Decimal | None: if len(candles) <= period: return None true_ranges: list[Decimal] = [] for previous, current in zip(candles[:-1], candles[1:]): true_ranges.append( max( current.high - current.low, abs(current.high - previous.close), abs(current.low - previous.close), ) ) atr = _average(true_ranges[:period]) if atr is None: return None for true_range in true_ranges[period:]: atr = ((atr * Decimal(period - 1)) + true_range) / Decimal(period) return atr def _ratio(numerator: Decimal, denominator: Decimal | None) -> Decimal | None: if denominator is None or denominator == 0: return None return numerator / denominator def _timeframe_features(candles: list[Candle], display_limit: int | None = None) -> dict[str, Any]: visible_candles = candles[-display_limit:] if display_limit is not None else candles closes = [candle.close for candle in candles] volumes = [candle.volume for candle in candles] quote_volumes = [candle.quote_volume for candle in candles] macd_value, macd_signal, macd_histogram = _macd(closes) previous_volumes = volumes[-21:-1] if len(volumes) > 1 else [] previous_quote_volumes = quote_volumes[-21:-1] if len(quote_volumes) > 1 else [] latest = candles[-1] if candles else None recent = candles[-20:] previous_recent = candles[-21:-1] if len(candles) > 1 else [] high_20 = max((candle.high for candle in recent), default=None) low_20 = min((candle.low for candle in recent), default=None) previous_high_20 = max((candle.high for candle in previous_recent), default=None) previous_low_20 = min((candle.low for candle in previous_recent), default=None) average_volume_20 = _average(previous_volumes) average_quote_volume_20 = _average(previous_quote_volumes) latest_volume = latest.volume if latest else Decimal("0") latest_quote_volume = latest.quote_volume if latest else Decimal("0") return { "opens": [_decimal_text(candle.open) for candle in visible_candles], "highs": [_decimal_text(candle.high) for candle in visible_candles], "lows": [_decimal_text(candle.low) for candle in visible_candles], "closes": [_decimal_text(candle.close) for candle in visible_candles], "volumes": [_decimal_text(candle.volume) for candle in visible_candles], "quote_volumes": [_decimal_text(candle.quote_volume) for candle in visible_candles], "trade_counts": [candle.trade_count for candle in visible_candles], "taker_buy_base_volumes": [_decimal_text(candle.taker_buy_base_volume) for candle in visible_candles], "taker_buy_quote_volumes": [_decimal_text(candle.taker_buy_quote_volume) for candle in visible_candles], "indicators": { "sma_7": _decimal_text(_sma(closes, 7)), "sma_20": _decimal_text(_sma(closes, 20)), "ema_12": _decimal_text(_ema(closes, 12)), "ema_26": _decimal_text(_ema(closes, 26)), "rsi_14": _decimal_text(_rsi(closes, 14)), "macd": _decimal_text(macd_value), "macd_signal": _decimal_text(macd_signal), "macd_histogram": _decimal_text(macd_histogram), "atr_14": _decimal_text(_atr(candles, 14)), }, "volume": { "latest_volume": _decimal_text(latest_volume), "average_volume_20": _decimal_text(average_volume_20), "volume_ratio_20": _decimal_text(_ratio(latest_volume, average_volume_20)), "latest_quote_volume": _decimal_text(latest_quote_volume), "average_quote_volume_20": _decimal_text(average_quote_volume_20), "quote_volume_ratio_20": _decimal_text(_ratio(latest_quote_volume, average_quote_volume_20)), }, "range": { "high_20": _decimal_text(high_20), "low_20": _decimal_text(low_20), "distance_from_high_20_bps": _decimal_text( ((latest.close - high_20) / high_20) * BPS if latest and high_20 and high_20 != 0 else None ), "distance_from_low_20_bps": _decimal_text( ((latest.close - low_20) / low_20) * BPS if latest and low_20 and low_20 != 0 else None ), "breakout_up_20": bool(latest and previous_high_20 is not None and latest.close > previous_high_20), "breakdown_down_20": bool(latest and previous_low_20 is not None and latest.close < previous_low_20), }, "indicator_metadata": { "display_candle_count": len(visible_candles), "calculation_candle_count": len(candles), "warmup_candle_count": max(len(candles) - len(visible_candles), 0), "warmup_requested": AI_TRADER_INDICATOR_WARMUP_CANDLES if display_limit is not None else 0, "rsi_method": "wilder", "atr_method": "wilder", "ema_warmup_applied": len(candles) > len(visible_candles), }, } def _sum_depth(rows: object) -> tuple[Decimal, Decimal]: if not isinstance(rows, list): return Decimal("0"), Decimal("0") quantity_total = Decimal("0") notional_total = Decimal("0") for row in rows: if not isinstance(row, list) or len(row) < 2: continue price = Decimal(str(row[0])) quantity = Decimal(str(row[1])) quantity_total += quantity notional_total += price * quantity return quantity_total, notional_total def _spread_rating(spread_bps: Decimal) -> str: if spread_bps <= Decimal("2"): return "tight" if spread_bps <= Decimal("5"): return "normal" if spread_bps <= Decimal("10"): return "wide" return "very_wide" def _liquidity_rating(depth_bid_notional: Decimal, depth_ask_notional: Decimal) -> str: smaller_side = min(depth_bid_notional, depth_ask_notional) if smaller_side >= Decimal("100000"): return "excellent" if smaller_side >= Decimal("25000"): return "good" if smaller_side >= Decimal("5000"): return "fair" return "poor" def _ticker_24h_features(payload: dict[str, Any]) -> dict[str, Any]: return { "price_change": str(payload.get("priceChange", "0")), "price_change_percent": str(payload.get("priceChangePercent", "0")), "weighted_avg_price": str(payload.get("weightedAvgPrice", "0")), "high_price": str(payload.get("highPrice", "0")), "low_price": str(payload.get("lowPrice", "0")), "volume": str(payload.get("volume", "0")), "quote_volume": str(payload.get("quoteVolume", "0")), "trade_count": int(payload.get("count", 0)), } def _average_price_features(payload: dict[str, Any]) -> dict[str, Any]: return { "mins": int(payload.get("mins", 0)), "price": str(payload.get("price", "0")), } def _aggregate_trade_features( trades: list[dict[str, Any]], now_ms: int | None = None, *, available: bool = True, ) -> dict[str, Any]: normalized = [trade for trade in trades if isinstance(trade, dict)] if not normalized: return { "status": "ok" if available else "unavailable", "trade_count": 0, "first_trade_id": None, "last_trade_id": None, "first_trade_time": None, "last_trade_time": None, "last_trade_age_ms": None, "base_volume": "0", "quote_volume": "0", "buy_base_volume": "0", "buy_quote_volume": "0", "sell_base_volume": "0", "sell_quote_volume": "0", "taker_buy_sell_ratio": None, "vwap": None, "large_trade_count": 0, "large_trade_quote_sum": "0", "largest_trade_quote": "0", } quote_values: list[Decimal] = [] base_volume = Decimal("0") quote_volume = Decimal("0") buy_base_volume = Decimal("0") buy_quote_volume = Decimal("0") sell_base_volume = Decimal("0") sell_quote_volume = Decimal("0") first_trade_id = normalized[0].get("a") last_trade_id = normalized[-1].get("a") first_trade_time = normalized[0].get("T") last_trade_time = normalized[-1].get("T") for trade in normalized: price = Decimal(str(trade.get("p", "0"))) quantity = Decimal(str(trade.get("q", "0"))) notional = price * quantity quote_values.append(notional) base_volume += quantity quote_volume += notional if bool(trade.get("m", False)): sell_base_volume += quantity sell_quote_volume += notional else: buy_base_volume += quantity buy_quote_volume += notional largest_trade_quote = max(quote_values, default=Decimal("0")) large_threshold = max(largest_trade_quote * Decimal("0.8"), Decimal("100")) large_trades = [value for value in quote_values if value >= large_threshold] last_age = None if now_ms is not None and isinstance(last_trade_time, int): last_age = max(now_ms - last_trade_time, 0) return { "status": "ok", "trade_count": len(normalized), "first_trade_id": first_trade_id, "last_trade_id": last_trade_id, "first_trade_time": first_trade_time, "last_trade_time": last_trade_time, "last_trade_age_ms": last_age, "base_volume": _decimal_text(base_volume), "quote_volume": _decimal_text(quote_volume), "buy_base_volume": _decimal_text(buy_base_volume), "buy_quote_volume": _decimal_text(buy_quote_volume), "sell_base_volume": _decimal_text(sell_base_volume), "sell_quote_volume": _decimal_text(sell_quote_volume), "taker_buy_sell_ratio": _decimal_text(_ratio(buy_quote_volume, sell_quote_volume)), "vwap": _decimal_text(_ratio(quote_volume, base_volume)), "large_trade_count": len(large_trades), "large_trade_quote_sum": _decimal_text(sum(large_trades, Decimal("0"))), "largest_trade_quote": _decimal_text(largest_trade_quote), } def _liquidity_features(context: _SymbolContext) -> dict[str, Any]: depth_bid_qty, depth_bid_notional = _sum_depth(context.depth.get("bids", [])) depth_ask_qty, depth_ask_notional = _sum_depth(context.depth.get("asks", [])) depth_total_notional = depth_bid_notional + depth_ask_notional depth_imbalance = ( (depth_bid_notional - depth_ask_notional) / depth_total_notional if depth_total_notional else None ) best_bid_notional = context.bid * context.best_bid_qty best_ask_notional = context.ask * context.best_ask_qty return { "best_bid_qty": _decimal_text(context.best_bid_qty), "best_ask_qty": _decimal_text(context.best_ask_qty), "best_bid_notional": _decimal_text(best_bid_notional), "best_ask_notional": _decimal_text(best_ask_notional), "depth_limit": 20, "depth_bid_qty": _decimal_text(depth_bid_qty), "depth_ask_qty": _decimal_text(depth_ask_qty), "depth_bid_notional": _decimal_text(depth_bid_notional), "depth_ask_notional": _decimal_text(depth_ask_notional), "depth_imbalance": _decimal_text(depth_imbalance), "spread_rating": _spread_rating(context.spread_bps), "liquidity_rating": _liquidity_rating(depth_bid_notional, depth_ask_notional), } def _timeframe_summary(features: dict[str, Any]) -> dict[str, Any]: closes = features.get("closes") if isinstance(features.get("closes"), list) else [] indicators = features.get("indicators") if isinstance(features.get("indicators"), dict) else {} volume = features.get("volume") if isinstance(features.get("volume"), dict) else {} range_features = features.get("range") if isinstance(features.get("range"), dict) else {} return { "latest_close": closes[-1] if closes else None, "previous_close": closes[-2] if len(closes) > 1 else None, "sma_7": indicators.get("sma_7"), "sma_20": indicators.get("sma_20"), "ema_12": indicators.get("ema_12"), "ema_26": indicators.get("ema_26"), "rsi_14": indicators.get("rsi_14"), "macd": indicators.get("macd"), "macd_signal": indicators.get("macd_signal"), "macd_histogram": indicators.get("macd_histogram"), "atr_14": indicators.get("atr_14"), "volume_ratio_20": volume.get("volume_ratio_20"), "quote_volume_ratio_20": volume.get("quote_volume_ratio_20"), "distance_from_high_20_bps": range_features.get("distance_from_high_20_bps"), "distance_from_low_20_bps": range_features.get("distance_from_low_20_bps"), "breakout_up_20": range_features.get("breakout_up_20"), "breakdown_down_20": range_features.get("breakdown_down_20"), } def _market_summary( *, context: _SymbolContext, timeframes: dict[str, dict[str, Any]], ticker_24h: dict[str, Any], aggregate_trades: dict[str, Any], realtime: dict[str, Any], liquidity: dict[str, Any], ) -> dict[str, Any]: return { "price": { "bid": str(context.bid), "ask": str(context.ask), "spread_bps": str(context.spread_bps.quantize(Decimal("0.0001"))), "spread_rating": liquidity.get("spread_rating"), }, "ticker_24h": { "price_change_percent": ticker_24h.get("price_change_percent"), "high_price": ticker_24h.get("high_price"), "low_price": ticker_24h.get("low_price"), "quote_volume": ticker_24h.get("quote_volume"), "trade_count": ticker_24h.get("trade_count"), }, "timeframes": { interval: _timeframe_summary(features) for interval, features in timeframes.items() }, "liquidity": { "liquidity_rating": liquidity.get("liquidity_rating"), "depth_imbalance": liquidity.get("depth_imbalance"), "depth_bid_notional": liquidity.get("depth_bid_notional"), "depth_ask_notional": liquidity.get("depth_ask_notional"), }, "aggregate_trades": { "status": aggregate_trades.get("status"), "last_trade_age_ms": aggregate_trades.get("last_trade_age_ms"), "taker_buy_sell_ratio": aggregate_trades.get("taker_buy_sell_ratio"), "large_trade_count": aggregate_trades.get("large_trade_count"), "largest_trade_quote": aggregate_trades.get("largest_trade_quote"), }, "realtime": { "status": realtime.get("status"), "spread_avg_bps": realtime.get("spread_avg_bps"), "book_imbalance_avg": realtime.get("book_imbalance_avg"), "agg_trade_buy_sell_ratio": realtime.get("agg_trade_buy_sell_ratio"), "large_trade_count": realtime.get("large_trade_count"), }, } def _balance_for(account: AccountSnapshot, asset: str) -> Balance: return account.balances.get(asset.upper(), Balance(asset, Decimal("0"), Decimal("0"))) def _execution_context(settings: Settings, context: _SymbolContext, account: AccountSnapshot) -> dict[str, Any]: base_balance = _balance_for(account, context.metadata.base_asset) quote_balance = _balance_for(account, context.metadata.quote_asset) open_sell_orders = sum(1 for order in context.open_orders if _is_sell_order(order)) protective_stop_orders = _bot_protective_stop_orders(context.open_orders) non_protective_sell_orders = _non_protective_sell_orders(context.open_orders) base_free = base_balance.free base_total = base_balance.free + base_balance.locked base_notional_bid = base_free * context.bid base_notional_ask = base_total * context.ask return { "quote_asset": context.metadata.quote_asset, "base_asset": context.metadata.base_asset, "quote_free": _decimal_text(quote_balance.free), "base_free": _decimal_text(base_balance.free), "base_locked": _decimal_text(base_balance.locked), "base_notional_bid": _decimal_text(base_notional_bid), "base_notional_ask": _decimal_text(base_notional_ask), "min_notional": _decimal_text(context.metadata.min_notional), "min_qty": _decimal_text(context.metadata.min_qty), "auto_buy_quote_amount": _decimal_text(settings.auto_buy_quote_amount), "auto_buy_existing_base_value_limit": _decimal_text(settings.auto_buy_existing_base_value_limit), "max_buy_spread_bps": _decimal_text(settings.auto_buy_max_spread_bps), "max_sell_spread_bps": _decimal_text(settings.auto_sell_max_spread_bps), "open_sell_orders": open_sell_orders, "protective_stop_orders": len(protective_stop_orders), "non_protective_sell_orders": len(non_protective_sell_orders), } def _model_context( settings: Settings, contexts: list[_SymbolContext], account: AccountSnapshot, realtime: dict[str, dict[str, Any]] | None = None, now_ms: int | None = None, ) -> dict[str, Any]: symbols = settings.ai_trader_symbols market = [] for context in contexts: timeframes = {} for interval, candles in context.candles_by_interval.items(): timeframes[interval] = _timeframe_features(candles, context.timeframe_display_limits.get(interval)) ticker_24h = _ticker_24h_features(context.ticker_24hr) average_price = _average_price_features(context.avg_price) aggregate_trades = _aggregate_trade_features( context.aggregate_trades, now_ms, available=context.aggregate_trades_available, ) symbol_realtime = (realtime or {}).get(context.symbol, _empty_realtime_features(0, "unavailable")) liquidity = _liquidity_features(context) market.append( { "symbol": context.symbol, "bid": str(context.bid), "ask": str(context.ask), "spread_bps": str(context.spread_bps.quantize(Decimal("0.0001"))), "best_bid_qty": str(context.best_bid_qty), "best_ask_qty": str(context.best_ask_qty), "timeframes": timeframes, "summary": _market_summary( context=context, timeframes=timeframes, ticker_24h=ticker_24h, aggregate_trades=aggregate_trades, realtime=symbol_realtime, liquidity=liquidity, ), "ticker_24h": ticker_24h, "average_price": average_price, "aggregate_trades": aggregate_trades, "realtime": symbol_realtime, "liquidity": liquidity, "execution": _execution_context(settings, context, account), } ) return {"symbols": list(symbols), "market": market} def _prediction_list(raw_predictions: object, allowed_symbols: tuple[str, ...]) -> list[AiTradePrediction]: if isinstance(raw_predictions, AiTradePrediction): return _normalize_predictions([raw_predictions], allowed_symbols) if isinstance(raw_predictions, list) and all(isinstance(item, AiTradePrediction) for item in raw_predictions): return _normalize_predictions(raw_predictions, allowed_symbols) return AiTradePrediction.holds_for_symbols(tuple(item.upper() for item in allowed_symbols), "模型响应无效") def _normalize_predictions( predictions: list[AiTradePrediction], allowed_symbols: tuple[str, ...], ) -> list[AiTradePrediction]: allowed = tuple(item.upper() for item in allowed_symbols) by_symbol = { prediction.symbol: prediction for prediction in predictions if prediction.symbol in allowed } normalized: list[AiTradePrediction] = [] for symbol in allowed: prediction = by_symbol.get(symbol) normalized.append(prediction if prediction is not None else AiTradePrediction.hold("模型未返回该交易对预测", symbol)) return normalized def _record_ai_trade_predictions( storage: Storage, predictions: list[AiTradePrediction], created_at_ms: int, ) -> list[tuple[AiTradePrediction, int]]: prediction_rows: list[tuple[AiTradePrediction, int]] = [] for prediction in predictions: run_id = storage.record_ai_trade_run( symbol=prediction.symbol, action=prediction.action, confidence=prediction.confidence, status="no_action" if prediction.action == "HOLD" else "pending", reason=prediction.reason, raw=_prediction_raw(prediction), created_at_ms=created_at_ms, ) prediction_rows.append((prediction, run_id)) return prediction_rows def _actionable_prediction_rows( prediction_rows: list[tuple[AiTradePrediction, int]], symbols: tuple[str, ...], ) -> list[tuple[AiTradePrediction, int]]: symbol_order = {symbol.upper(): index for index, symbol in enumerate(symbols)} return sorted( ( (prediction, run_id) for prediction, run_id in prediction_rows if prediction.action in {"BUY", "SELL"} ), key=lambda item: (-item[0].confidence, symbol_order.get(item[0].symbol, len(symbol_order))), ) def _skip_remaining_ai_predictions( storage: Storage, prediction_rows: list[tuple[AiTradePrediction, int]], *, executed_run_id: int, ) -> None: for prediction, run_id in prediction_rows: if run_id == executed_run_id or prediction.action == "HOLD": continue storage.update_ai_trade_run_status( run_id, status="skipped", reason="已有更高优先级AI预测执行", raw=_prediction_raw(prediction), ) def _skip_batch_limit_ai_predictions( storage: Storage, prediction_rows: list[tuple[AiTradePrediction, int]], *, processed_run_ids: set[int], ) -> None: for prediction, run_id in prediction_rows: if run_id in processed_run_ids or prediction.action == "HOLD": continue storage.update_ai_trade_run_status( run_id, status="skipped", reason="已达到本次AI批量交易次数上限", raw=_prediction_raw(prediction), ) def _execute_ai_buy( settings: Settings, storage: Storage, exchange, account: AccountSnapshot, symbol_context: _SymbolContext, prediction: AiTradePrediction, ai_run_id: int, now_ms: int, audit_logger: AuditLogger, ) -> AiTradeExecutionResult: candidate = AutoBuyCandidate( symbol=prediction.symbol, score=prediction.confidence, reference_price=symbol_context.ask, spread_bps=symbol_context.spread_bps, reasons=(f"AI: {prediction.reason}",), ) day_start_ms = _local_day_start_ms(now_ms) recent_since_ms = now_ms - (settings.auto_buy_cooldown_seconds * 1000) risk_settings = 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, ) risk_context = AutoBuyRiskContext( mode=settings.mode, enabled=settings.live_ai_trader_enabled, candidate=candidate, metadata=symbol_context.metadata, 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=_decimal_from_payload(symbol_context.ticker_24hr, "priceChangePercent"), ) decision = AutoBuyRiskEngine(risk_settings).evaluate(risk_context) audit_logger.log( "ai_trader.risk", params={"action": "BUY", "symbol": candidate.symbol, "ai_run_id": ai_run_id}, formula=_audit_risk_formula(), result={ "approved": decision.approved, "reason": decision.reason, "confidence": prediction.confidence, "quote_amount": settings.auto_buy_quote_amount, "existing_base_value_limit": risk_settings.existing_base_value_limit, "daily_attempts": risk_context.daily_attempts, "daily_limit": risk_settings.daily_limit, "spread_bps": candidate.spread_bps, "max_spread_bps": risk_settings.max_spread_bps, "price_change_percent_24h": risk_context.price_change_percent_24h, "max_24h_rally_percent": risk_settings.max_24h_rally_percent, }, ) if not decision.approved: reason = _ai_trade_reason( prediction, _ai_buy_risk_reason(decision.reason, risk_context, risk_settings, settings, symbol_context), ) run_id = storage.record_auto_buy_run( symbol=candidate.symbol, status="rejected", score=candidate.score, reason=reason, quote_amount=settings.auto_buy_quote_amount, created_at_ms=now_ms, ) storage.update_ai_trade_run_status( ai_run_id, status="rejected", reason=reason, raw=_prediction_raw(prediction), auto_buy_run_id=run_id, ) result = AiTradeExecutionResult(ai_run_id, "rejected", candidate.symbol, "BUY", reason, prediction.confidence) _audit_execution_result(audit_logger, result) return result 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: reason = _ai_trade_reason(prediction, _ai_reason(reservation.reason)) run_id = storage.record_auto_buy_run( symbol=candidate.symbol, status="rejected", score=candidate.score, reason=reason, quote_amount=settings.auto_buy_quote_amount, created_at_ms=now_ms, ) storage.update_ai_trade_run_status( ai_run_id, status="rejected", reason=reason, raw=_prediction_raw(prediction), auto_buy_run_id=run_id, ) result = AiTradeExecutionResult(ai_run_id, "rejected", candidate.symbol, "BUY", reason, prediction.confidence) _audit_execution_result(audit_logger, result) return result result = 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, take_profit_percent=settings.auto_buy_take_profit_percent, ).execute(candidate, symbol_context.metadata, settings.auto_buy_quote_amount, run_id=reservation.run_id) reason = f"AI: {prediction.reason}; {_ai_reason(result.reason)}" storage.update_auto_buy_run_status(result.run_id, result.status, reason) storage.update_ai_trade_run_status( ai_run_id, status=result.status, reason=reason, raw=_prediction_raw(prediction), auto_buy_run_id=result.run_id, ) ai_result = AiTradeExecutionResult(ai_run_id, result.status, result.symbol, "BUY", reason, prediction.confidence) _audit_execution_result(audit_logger, ai_result) return ai_result def _execute_ai_sell( settings: Settings, storage: Storage, exchange, account: AccountSnapshot, symbol_context: _SymbolContext, prediction: AiTradePrediction, ai_run_id: int, now_ms: int, audit_logger: AuditLogger, ) -> AiTradeExecutionResult: def reject(reason: str, rejected_quantity: Decimal) -> AiTradeExecutionResult: run_id = storage.record_auto_sell_run( symbol=candidate.symbol, status="rejected", score=candidate.score, reason=reason, quantity=rejected_quantity, created_at_ms=now_ms, ) storage.update_ai_trade_run_status( ai_run_id, status="rejected", reason=reason, raw=_prediction_raw(prediction), auto_sell_run_id=run_id, ) result = AiTradeExecutionResult(ai_run_id, "rejected", candidate.symbol, "SELL", reason, prediction.confidence) _audit_execution_result(audit_logger, result) return result protective_stop_orders = _bot_protective_stop_orders(symbol_context.open_orders) non_protective_sell_orders = _non_protective_sell_orders(symbol_context.open_orders) should_cancel_protective_stops = bool(protective_stop_orders) and not non_protective_sell_orders effective_open_orders = ( [order for order in symbol_context.open_orders if order not in protective_stop_orders] if should_cancel_protective_stops else symbol_context.open_orders ) effective_account = ( _account_with_unlocked_protective_stops(account, symbol_context.metadata) if should_cancel_protective_stops else account ) quantity = auto_sell._round_down( effective_account.free(symbol_context.metadata.base_asset), symbol_context.metadata.step_size, ) candidate = AutoSellCandidate( symbol=prediction.symbol, score=prediction.confidence, reference_price=symbol_context.bid, spread_bps=symbol_context.spread_bps, reasons=(f"AI: {prediction.reason}",), ) day_start_ms = _local_day_start_ms(now_ms) recent_since_ms = now_ms - (settings.auto_sell_cooldown_seconds * 1000) risk_settings = AutoSellRiskSettings( daily_limit=settings.auto_sell_daily_limit, max_spread_bps=settings.auto_sell_max_spread_bps, ) risk_context = AutoSellRiskContext( mode=settings.mode, enabled=settings.live_ai_trader_enabled, candidate=candidate, metadata=symbol_context.metadata, account=effective_account, daily_attempts=storage.count_auto_sell_attempts_since(day_start_ms), has_recent_symbol_attempt=storage.has_recent_auto_sell_attempt(candidate.symbol, recent_since_ms), has_open_sell_order=auto_sell._has_open_sell_order(effective_open_orders), quantity=quantity, ) decision = AutoSellRiskEngine(risk_settings).evaluate(risk_context) audit_logger.log( "ai_trader.risk", params={"action": "SELL", "symbol": candidate.symbol, "ai_run_id": ai_run_id}, formula=_audit_risk_formula(), result={ "approved": decision.approved, "reason": decision.reason, "confidence": prediction.confidence, "quantity": quantity, "daily_attempts": risk_context.daily_attempts, "daily_limit": risk_settings.daily_limit, "spread_bps": candidate.spread_bps, "max_spread_bps": risk_settings.max_spread_bps, }, ) if not decision.approved: reason = _ai_trade_reason( prediction, _ai_sell_risk_reason(decision.reason, risk_context, risk_settings, settings, symbol_context), ) return reject(reason, quantity) if should_cancel_protective_stops: for order in protective_stop_orders: client_order_id = str(order.get("clientOrderId", "")) try: exchange.cancel_order(candidate.symbol, client_order_id) except Exception: reason = _ai_trade_reason(prediction, _ai_reason("protective stop cancel failed")) return reject(reason, quantity) account = exchange.account() quantity = auto_sell._round_down( account.free(symbol_context.metadata.base_asset), symbol_context.metadata.step_size, ) symbol_context = replace(symbol_context, open_orders=effective_open_orders) risk_context = AutoSellRiskContext( mode=settings.mode, enabled=settings.live_ai_trader_enabled, candidate=candidate, metadata=symbol_context.metadata, account=account, daily_attempts=risk_context.daily_attempts, has_recent_symbol_attempt=risk_context.has_recent_symbol_attempt, has_open_sell_order=auto_sell._has_open_sell_order(symbol_context.open_orders), quantity=quantity, ) decision = AutoSellRiskEngine(risk_settings).evaluate(risk_context) if not decision.approved: reason = _ai_trade_reason( prediction, _ai_sell_risk_reason(decision.reason, risk_context, risk_settings, settings, symbol_context), ) return reject(reason, quantity) cost_decision = evaluate_sell_cost_guard( storage, candidate.symbol, quantity=quantity, 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, ) audit_logger.log( "ai_trader.cost_guard", params={"action": "SELL", "symbol": candidate.symbol, "ai_run_id": ai_run_id}, formula={"estimated_proceeds": "quantity * bid", "matched_cost": "FIFO open lot cost"}, result={ "approved": cost_decision.approved, "reason": cost_decision.reason, "quantity": quantity, "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, }, ) if not cost_decision.approved: reason = _ai_trade_reason(prediction, _ai_reason(cost_decision.reason)) return reject(reason, quantity) reservation = storage.reserve_auto_sell_attempt( symbol=candidate.symbol, score=candidate.score, reason=", ".join(candidate.reasons), quantity=quantity, 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: reason = _ai_trade_reason(prediction, _ai_reason(reservation.reason)) run_id = storage.record_auto_sell_run( symbol=candidate.symbol, status="rejected", score=candidate.score, reason=reason, quantity=quantity, created_at_ms=now_ms, ) storage.update_ai_trade_run_status( ai_run_id, status="rejected", reason=reason, raw=_prediction_raw(prediction), auto_sell_run_id=run_id, ) result = AiTradeExecutionResult(ai_run_id, "rejected", candidate.symbol, "SELL", reason, prediction.confidence) _audit_execution_result(audit_logger, result) return result result = AutoSellExecutionEngine(exchange, storage).execute( candidate, symbol_context.metadata, quantity, run_id=reservation.run_id, ) reason = f"AI: {prediction.reason}; {_ai_reason(result.reason)}" storage.update_auto_sell_run_status(result.run_id, result.status, reason) storage.update_ai_trade_run_status( ai_run_id, status=result.status, reason=reason, raw=_prediction_raw(prediction), auto_sell_run_id=result.run_id, ) ai_result = AiTradeExecutionResult(ai_run_id, result.status, result.symbol, "SELL", reason, prediction.confidence) _audit_execution_result(audit_logger, ai_result) return ai_result def _audit_execution_result(audit_logger: AuditLogger, result: AiTradeExecutionResult) -> None: audit_logger.log( "ai_trader.execution", params={"action": result.action, "symbol": result.symbol, "ai_run_id": result.run_id}, result={ "status": result.status, "reason": result.reason, "confidence": result.confidence, }, ) def _record_ai_trade_rejection(storage: Storage, reason: str, created_at_ms: int) -> int: return storage.record_ai_trade_run( symbol=None, action="HOLD", confidence=Decimal("0"), status="rejected", reason=reason, raw={"reason": reason}, created_at_ms=created_at_ms, ) def _prediction_raw(prediction: AiTradePrediction) -> dict[str, str]: return { "symbol": prediction.symbol, "action": prediction.action, "confidence": str(prediction.confidence), "reason": prediction.reason, } def _ai_reason(reason: str) -> str: return _AI_REASON_TRANSLATIONS.get(reason, reason) def _ai_buy_risk_reason( reason: str, context: AutoBuyRiskContext, risk_settings: AutoBuyRiskSettings, settings: Settings, symbol_context: _SymbolContext, ) -> str: base = _ai_reason(reason) candidate = context.candidate metadata = context.metadata if reason == "daily auto-buy limit reached": return f"{base}（当前 {context.daily_attempts}，上限 {risk_settings.daily_limit}）" if reason == "symbol cooldown active": return f"{base}（冷却 {settings.auto_buy_cooldown_seconds} 秒）" if reason == "spread above max": return f"{base}（当前 {_decimal_text(candidate.spread_bps)}，最大 {_decimal_text(risk_settings.max_spread_bps)}）" if reason == "24h rally above max": return ( f"{base}（当前 {context.price_change_percent_24h}%，" f"上限 {_decimal_text(risk_settings.max_24h_rally_percent)}%）" ) if reason == "quote amount below min notional": return f"{base}（当前 {_decimal_text(context.quote_amount)}，最小 {_decimal_text(metadata.min_notional)}）" if reason == "insufficient quote balance": quote_free = context.account.free(metadata.quote_asset) required = context.quote_amount + risk_settings.quote_reserve return f"{base}（可用 {_decimal_text(quote_free)}，需要 {_decimal_text(required)}）" if reason == "existing base balance": base_free = context.account.free(metadata.base_asset) base_locked = context.account.balances.get(metadata.base_asset) locked = base_locked.locked if base_locked is not None else Decimal("0") base_value = (base_free + locked) * candidate.reference_price existing_base_value_limit = risk_settings.existing_base_value_limit or context.quote_amount return f"{base}（当前价值 {_decimal_text(base_value)}，限制 {_decimal_text(existing_base_value_limit)}）" if reason == "missing exchange filters": return ( f"{base}（minQty {_decimal_text(metadata.min_qty)}，" f"stepSize {_decimal_text(metadata.step_size)}，tickSize {_decimal_text(metadata.tick_size)}）" ) return base def _ai_sell_risk_reason( reason: str, context: AutoSellRiskContext, risk_settings: AutoSellRiskSettings, settings: Settings, symbol_context: _SymbolContext, ) -> str: base = _ai_reason(reason) candidate = context.candidate metadata = context.metadata if reason == "daily auto-sell limit reached": return f"{base}（当前 {context.daily_attempts}，上限 {risk_settings.daily_limit}）" if reason == "symbol cooldown active": return f"{base}（冷却 {settings.auto_sell_cooldown_seconds} 秒）" if reason == "spread above max": return f"{base}（当前 {_decimal_text(candidate.spread_bps)}，最大 {_decimal_text(risk_settings.max_spread_bps)}）" if reason == "open sell order exists": open_sell_count = sum(1 for order in symbol_context.open_orders if str(order.get("side", "")).upper() == "SELL") return f"{base}（当前 {open_sell_count}）" if reason == "missing exchange filters": return ( f"{base}（minQty {_decimal_text(metadata.min_qty)}，" f"stepSize {_decimal_text(metadata.step_size)}，minNotional {_decimal_text(metadata.min_notional)}）" ) if reason == "no free balance": base_free = context.account.free(metadata.base_asset) return f"{base}（可用 {_decimal_text(base_free)}）" if reason == "quantity below min quantity": return f"{base}（当前 {_decimal_text(context.quantity)}，最小 {_decimal_text(metadata.min_qty)}）" if reason == "notional below min notional": notional = context.quantity * candidate.reference_price return f"{base}（当前 {_decimal_text(notional)}，最小 {_decimal_text(metadata.min_notional)}）" if reason == "quantity exceeds free balance": base_free = context.account.free(metadata.base_asset) return f"{base}（当前 {_decimal_text(context.quantity)}，可用 {_decimal_text(base_free)}）" return base def _ai_trade_reason(prediction: AiTradePrediction, rejection_reason: str) -> str: return f"AI: {prediction.reason}; 风控: {rejection_reason}" def _optional_decimal(row: list, index: int) -> Decimal: if len(row) <= index: return Decimal("0") return Decimal(str(row[index])) def _optional_int(row: list, index: int) -> int: if len(row) <= index: return 0 return int(row[index]) 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])), quote_volume=_optional_decimal(row, 7), trade_count=_optional_int(row, 8), taker_buy_base_volume=_optional_decimal(row, 9), taker_buy_quote_volume=_optional_decimal(row, 10), ) 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)