from __future__ import annotations import json import re 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 .audit_log import AuditLogger from .config import Settings from .futures_exchange import BinanceUsdmFuturesClient, parse_position from .futures_execution import FuturesExecutionEngine from .futures_models import FuturesAction, FuturesExecutionResult, FuturesPosition, FuturesPrediction from .futures_risk import FuturesRiskContext, FuturesRiskEngine, FuturesRiskSettings from .models import TradingMode from .storage import Storage FUTURES_ACTIONS = {item.value for item in FuturesAction} FUTURES_CLOSE_PERCENTS = {25, 50, 100} FUTURES_DAILY_PNL_INCOME_TYPES = {"REALIZED_PNL", "COMMISSION", "FUNDING_FEE"} FUTURES_PREFLIGHT_CONTEXT_FAILED_REASON = "futures ai preflight/context failed" FUTURES_EXECUTION_FAILED_REASON = "futures ai execution failed" FUTURES_ACTIVE_ALGO_ORDER_STATUSES = {"", "NEW", "PARTIALLY_FILLED", "PENDING", "ACCEPTED"} FUTURES_EXECUTED_STATUSES = { "protected", "open_unknown", "rollback_completed", "rollback_failed", "closed", "partially_closed", "close_unknown", "close_protection_failed", "dry_run", } FUTURES_FAILED_STATUSES = { "open_unknown", "rollback_completed", "rollback_failed", "close_failed", "close_unknown", "close_protection_failed", } _FUTURES_REASON_TRANSLATIONS = { "approved": "已通过风控", "confidence below minimum": "AI置信度低于最小阈值", "metadata symbol mismatch": "交易规则与交易对不匹配", "hold signal": "保持观望信号", "no position to close": "没有可关闭的持仓", "invalid close percent": "关闭仓位比例无效", "daily net loss limit reached": "每日净亏损达到上限", "same-direction position exists": "已有同方向持仓", "opposite position requires close first": "反向持仓需要先平仓", "total margin above max": "总逐仓保证金超过上限", "notional below min notional": "名义金额低于最小下单金额", "missing protection prices": "缺少保护价格", "invalid risk input": "风控输入无效", "invalid stop or take-profit direction": "止损或止盈方向无效", "stop distance outside limits": "止损距离超出限制", "take-profit distance outside limits": "止盈距离超出限制", "estimated liquidation buffer below minimum": "预估强平缓冲低于下限", "estimated stop loss above margin risk budget": "预估止损超过保证金风险预算", "trend filter blocked": "4h趋势过滤拦截", "reverse cooldown active": "反向开仓冷却中", "stop distance below volatility floor": "止损距离低于波动率底线", "futures execution not wired": "合约执行路径未启用", "max actions reached": "已达到本次合约AI交易次数上限", FUTURES_EXECUTION_FAILED_REASON: "合约AI执行失败", "unsupported futures open action": "不支持的合约开仓动作", "invalid reference price": "参考价格无效", "invalid execution quantity": "执行数量无效", "dry run": "模拟执行", "open order status unknown": "开仓订单状态未知", "protection failed; rollback failed": "保护单失败，回滚失败", "protection failed; rollback completed": "保护单失败，回滚已完成", "protected": "保护单已创建", "unsupported futures close action": "不支持的合约平仓动作", "invalid close quantity": "平仓数量无效", "protection cancel failed": "保护单取消失败", "close order status unknown": "平仓订单状态未知", "closed": "已平仓", "close completed; remaining protection missing": "平仓完成，剩余保护单缺失", "close completed; remaining protection failed": "平仓完成，剩余保护单创建失败", "partially closed": "已部分平仓", } _LOG_TEXT_TAIL_BYTES = 2048 _SENSITIVE_ERROR_PATTERNS = ( re.compile(r"signedUrl=[^\s,&]+", re.IGNORECASE), re.compile(r"signature=[^\s,&]+", re.IGNORECASE), re.compile(r"api[_-]?key=[^\s,&]+", re.IGNORECASE), ) _FUTURES_TREND_THRESHOLD_PERCENT = Decimal("0.15") _FUTURES_TREND_EMA_THRESHOLD_PERCENT = Decimal("0.25") _FUTURES_TREND_EMA_FAST_PERIOD = 8 _FUTURES_TREND_EMA_SLOW_PERIOD = 21 @dataclass(frozen=True) class _TrendEvidence: interval: str prev_close: Decimal | None last_close: Decimal | None delta_percent: Decimal | None direction: str @dataclass(frozen=True) class _EmaTrendEvidence: fast_period: int slow_period: int fast: Decimal | None slow: Decimal | None delta_percent: Decimal | None direction: str @dataclass(frozen=True) class _TrendFilterDecision: blocked: bool details: dict[str, Any] @dataclass(frozen=True) class FuturesAiBatchResult: status: str reason: str max_actions: int prediction_count: int executed_count: int daily_realized_pnl: Decimal total_isolated_margin: Decimal results: tuple[FuturesExecutionResult, ...] @dataclass(frozen=True) class FuturesAiPrediction(FuturesPrediction): @classmethod def hold(cls, reason: str, symbol: str = "") -> "FuturesAiPrediction": return cls(symbol=symbol, action=FuturesAction.HOLD, confidence=Decimal("0"), reason=reason) @classmethod def from_json_list(cls, text: str, allowed_symbols: tuple[str, ...]) -> list["FuturesAiPrediction"]: allowed = tuple(item.upper() for item in allowed_symbols) try: payload = json.loads(text) except json.JSONDecodeError: return [cls.hold("模型响应无效", symbol) for symbol in allowed] if not isinstance(payload, dict) or not isinstance(payload.get("predictions"), list): return [cls.hold("模型响应无效", symbol) for symbol in allowed] by_symbol: dict[str, FuturesAiPrediction] = {} invalid_symbols: set[str] = set() for item in payload["predictions"]: symbol = str(item.get("symbol", "")).strip().upper() if isinstance(item, dict) else "" prediction = cls.from_payload(item, allowed) if prediction is None: if symbol in allowed: invalid_symbols.add(symbol) continue by_symbol.setdefault(prediction.symbol, prediction) result: list[FuturesAiPrediction] = [] for symbol in allowed: if symbol in by_symbol: result.append(by_symbol[symbol]) elif symbol in invalid_symbols: result.append(cls.hold("模型返回的该交易对预测无效", symbol)) else: result.append(cls.hold("模型未返回该交易对预测", symbol)) return result @classmethod def from_payload(cls, payload: object, allowed_symbols: tuple[str, ...]) -> "FuturesAiPrediction | None": if not isinstance(payload, dict): return None symbol = str(payload.get("symbol", "")).strip().upper() if symbol not in allowed_symbols: return None action_text = str(payload.get("action", "")).strip().upper() if action_text not in FUTURES_ACTIONS: return None try: confidence = Decimal(str(payload.get("confidence", ""))) except (InvalidOperation, ValueError): return None if not confidence.is_finite() or confidence < 0 or confidence > 1: return None reason_value = payload.get("reason") if not isinstance(reason_value, str): return None reason = reason_value.strip() if not reason: return None action = FuturesAction(action_text) leverage = _optional_int(payload.get("leverage")) stop_loss_price = _optional_decimal(payload.get("stop_loss_price")) take_profit_price = _optional_decimal(payload.get("take_profit_price")) close_percent = _optional_exact_int(payload.get("close_percent")) if action in {FuturesAction.OPEN_LONG, FuturesAction.OPEN_SHORT}: if stop_loss_price is None or take_profit_price is None: return None if action is FuturesAction.CLOSE and close_percent not in FUTURES_CLOSE_PERCENTS: return None return cls( symbol=symbol, action=action, confidence=confidence, reason=reason, leverage=leverage, stop_loss_price=stop_loss_price, take_profit_price=take_profit_price, close_percent=close_percent, ) class StaticNoopFuturesPredictor: def predict(self, context: dict[str, Any], allowed_symbols: tuple[str, ...]) -> list[FuturesAiPrediction]: return [FuturesAiPrediction.hold("未配置 futures 预测器", symbol) for symbol in allowed_symbols] class CodexExecFuturesPredictor: 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[FuturesAiPrediction]: 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_futures_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 _hold_predictions(allowed_symbols, "codex exec 超时") except OSError as exc: duration_ms = _elapsed_ms(started_at) error = f"codex exec failed: {type(exc).__name__}" self._write_call_log( context=context, allowed_symbols=allowed_symbols, prompt=prompt, status="error", duration_ms=duration_ms, returncode=None, stdout="", stderr="", error=error, ) self._write_audit_summary( allowed_symbols=allowed_symbols, status="error", duration_ms=duration_ms, returncode=None, stdout="", stderr="", error=error, ) return _hold_predictions(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 _hold_predictions(allowed_symbols, "codex exec 执行失败") return FuturesAiPrediction.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( "futures_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 / "futures_ai_trader_llm.log" model = self.model or "-" return_code = "-" if returncode is None else str(returncode) lines = [ "=" * 80, "Codex Exec Futures 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 run_live_futures_ai_trade_batch_once( settings: Settings, exchange_client=None, predictor=None, now_ms: int | None = None, ) -> FuturesAiBatchResult: 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.futures_ai_max_actions_per_run if settings.mode is not TradingMode.LIVE: return _futures_batch_rejection(storage, "live mode required", max_actions, now_ms) if not settings.live_futures_ai_trader_enabled: return _futures_batch_rejection(storage, "live futures ai trader disabled", max_actions, now_ms) if not settings.live_futures_confirm_production: return _futures_batch_rejection( storage, "live futures production confirmation disabled", max_actions, now_ms, ) preflight_stage = "create_exchange" try: exchange = exchange_client if exchange_client is not None else BinanceUsdmFuturesClient( api_key=settings.api_key, api_secret=settings.api_secret.get_secret_value(), base_url=settings.futures_rest_base_url, base_urls=settings.futures_rest_base_urls, recv_window=settings.recv_window, proxy_url=settings.proxy_url, ) preflight_stage = "position_mode" if exchange.current_position_mode(): return _futures_batch_rejection(storage, "hedge mode active", max_actions, now_ms) allowed_symbols = tuple(settings.futures_ai_symbols) position_rows: dict[str, list[dict]] = {} for symbol in allowed_symbols: preflight_stage = f"position_risk:{symbol}" position_rows[symbol] = list(exchange.position_risk(symbol)) preflight_stage = "position_risk:account" account_position_rows = _account_position_rows(exchange) preflight_stage = "account" account_snapshot = exchange.account() preflight_stage = "income" daily_realized_pnl = _daily_realized_pnl(exchange, now_ms) total_isolated_margin = _total_isolated_margin(account_position_rows) preflight_stage = "model_context" context = _model_context( settings, exchange, position_rows, account_snapshot, daily_realized_pnl=daily_realized_pnl, total_isolated_margin=total_isolated_margin, now_ms=now_ms, ) predictor = predictor if predictor is not None else CodexExecFuturesPredictor( model=settings.ai_trader_model, codex_command=settings.ai_trader_codex_command, timeout_seconds=settings.ai_trader_codex_timeout_seconds, ) preflight_stage = "predictor" raw_predictions = predictor.predict(context, allowed_symbols) except Exception as exc: return _futures_batch_rejection( storage, FUTURES_PREFLIGHT_CONTEXT_FAILED_REASON, max_actions, now_ms, raw=_preflight_failure_raw(preflight_stage, exc), ) predictions = _prediction_list(raw_predictions, allowed_symbols) prediction_rows = _record_futures_ai_trade_predictions(storage, predictions, now_ms) results: list[FuturesExecutionResult] = [] for prediction, run_id in prediction_rows: raw = _prediction_raw(prediction) if prediction.action is FuturesAction.HOLD: storage.update_futures_ai_trade_run_status(run_id, "no_action", prediction.reason, raw=raw) results.append( FuturesExecutionResult( run_id=run_id, status="no_action", symbol=prediction.symbol, action=prediction.action.value, reason=prediction.reason, confidence=prediction.confidence, raw=raw, ) ) continue if prediction.action not in {FuturesAction.OPEN_LONG, FuturesAction.OPEN_SHORT, FuturesAction.CLOSE}: reason = _futures_result_reason(prediction, "执行", "futures execution not wired") storage.update_futures_ai_trade_run_status(run_id, "skipped", reason, raw=raw) results.append( FuturesExecutionResult( run_id=run_id, status="skipped", symbol=prediction.symbol, action=prediction.action.value, reason=reason, confidence=prediction.confidence, leverage=prediction.leverage, margin_amount=settings.futures_margin_amount, close_percent=prediction.close_percent, raw=raw, ) ) actionable = [ (prediction, run_id) for prediction, run_id in prediction_rows if prediction.action in {FuturesAction.OPEN_LONG, FuturesAction.OPEN_SHORT, FuturesAction.CLOSE} ] symbol_order = {symbol: index for index, symbol in enumerate(settings.futures_ai_symbols)} actionable.sort(key=lambda item: (-item[0].confidence, symbol_order.get(item[0].symbol, 9999))) executed_count = 0 risk_settings = _risk_settings(settings) for prediction, run_id in actionable: raw = _prediction_raw(prediction) if executed_count >= settings.futures_ai_max_actions_per_run: reason = _futures_result_reason(prediction, "执行", "max actions reached") storage.update_futures_ai_trade_run_status(run_id, "skipped", reason, raw=raw) results.append( FuturesExecutionResult( run_id, "skipped", prediction.symbol, prediction.action.value, reason, prediction.confidence, raw=raw, ) ) continue execution_stage = "exchange_info" try: execution_stage = "exchange_info" metadata = exchange.exchange_info(prediction.symbol) execution_stage = "position_risk" position = parse_position(exchange.position_risk(prediction.symbol)[0]) execution_stage = "mark_price" mark = Decimal(str(exchange.mark_price(prediction.symbol)["markPrice"])) execution_stage = "account_position_risk" account_position_rows = _account_position_rows(exchange) total_isolated_margin = _total_isolated_margin(account_position_rows) execution_stage = "income" daily_realized_pnl = _daily_realized_pnl(exchange, int(time.time() * 1000)) context = FuturesRiskContext( prediction=prediction, metadata=metadata, position=position, mark_price=mark, total_isolated_margin=total_isolated_margin, daily_realized_pnl=daily_realized_pnl, hedge_mode=False, ) execution_stage = "strategy" strategy_rejection = _strategy_rejection(settings, storage, prediction, context, exchange, now_ms) if strategy_rejection is not None: stage, rejection_reason, rejection_details = strategy_rejection reason = _futures_result_reason( prediction, stage, rejection_reason, _rejection_detail_text(rejection_details), ) if rejection_details: raw["risk_rejection"] = rejection_details storage.update_futures_ai_trade_run_status(run_id, "rejected", reason, raw=raw) results.append( FuturesExecutionResult( run_id, "rejected", prediction.symbol, prediction.action.value, reason, prediction.confidence, raw=raw, ) ) continue execution_stage = "risk" decision = FuturesRiskEngine(risk_settings).evaluate(context) if not decision.approved: rejection_details, detail_text = _risk_rejection_details( decision.reason, context, risk_settings, ) reason = _futures_result_reason(prediction, "风控", decision.reason, detail_text) if rejection_details: raw["risk_rejection"] = rejection_details storage.update_futures_ai_trade_run_status(run_id, "rejected", reason, raw=raw) results.append( FuturesExecutionResult( run_id, "rejected", prediction.symbol, prediction.action.value, reason, prediction.confidence, raw=raw, ) ) continue engine = FuturesExecutionEngine(settings.mode, exchange, storage) if prediction.action is FuturesAction.CLOSE: execution_stage = "existing_protection" stop_loss_price, take_profit_price, protection_ids = _existing_protection_prices( exchange.all_algo_orders(prediction.symbol) ) execution_stage = "execute_close" execution_result = engine.execute_close( run_id, prediction, metadata, position_quantity=position.quantity, protection_client_algo_ids=protection_ids, existing_stop_loss_price=stop_loss_price, existing_take_profit_price=take_profit_price, ) else: execution_stage = "execute_open" execution_result = engine.execute_open( run_id, decision, metadata, reference_price=mark, current_margin_type=position.margin_type, ) except Exception as exc: reason = _futures_result_reason(prediction, "执行", FUTURES_EXECUTION_FAILED_REASON) raw = _prediction_raw_with_execution_error(prediction, execution_stage, exc) storage.update_futures_ai_trade_run_status(run_id, "rejected", reason, raw=raw) results.append( FuturesExecutionResult( run_id, "rejected", prediction.symbol, prediction.action.value, reason, prediction.confidence, raw=raw, ) ) continue result_reason = _futures_result_reason(prediction, "执行", execution_result.reason) execution_result = replace(execution_result, reason=result_reason, raw=raw) storage.update_futures_ai_trade_run_status(run_id, execution_result.status, result_reason, raw=raw) if execution_result.status in FUTURES_EXECUTED_STATUSES: _record_position_snapshot(storage, exchange, run_id, prediction.symbol, now_ms) results.append(execution_result) if execution_result.status in FUTURES_EXECUTED_STATUSES: executed_count += 1 status = _futures_batch_status(results) return FuturesAiBatchResult( status, _futures_batch_reason(status), settings.futures_ai_max_actions_per_run, len(predictions), executed_count, daily_realized_pnl, total_isolated_margin, tuple(results), ) def _risk_settings(settings: Settings) -> FuturesRiskSettings: return FuturesRiskSettings( margin_amount=settings.futures_margin_amount, default_leverage=settings.futures_default_leverage, max_leverage=settings.futures_max_leverage, max_total_margin=settings.futures_max_total_margin, min_confidence=settings.futures_ai_min_confidence, daily_realized_loss_limit=settings.futures_daily_realized_loss_limit, max_margin_loss_percent=settings.futures_max_margin_loss_percent, min_liquidation_buffer_percent=settings.futures_min_liquidation_buffer_percent, min_stop_distance_percent=settings.futures_min_stop_distance_percent, max_stop_distance_percent=settings.futures_max_stop_distance_percent, min_take_profit_distance_percent=settings.futures_min_take_profit_distance_percent, max_take_profit_distance_percent=settings.futures_max_take_profit_distance_percent, ) def _risk_rejection_details( reason: str, context: FuturesRiskContext, settings: FuturesRiskSettings, ) -> tuple[dict[str, Any] | None, str | None]: prediction = context.prediction base: dict[str, Any] = {"reason": reason} if reason == "confidence below minimum": return ( { **base, "confidence": _decimal_text(prediction.confidence), "min_confidence": _decimal_text(settings.min_confidence), }, f"当前 {_decimal_text(prediction.confidence)}，最小 {_decimal_text(settings.min_confidence)}", ) if reason == "daily net loss limit reached": return ( { **base, "daily_net_pnl": _decimal_text(context.daily_realized_pnl), "daily_net_loss_limit": _decimal_text(settings.daily_realized_loss_limit), }, ( f"当前 {_decimal_text(context.daily_realized_pnl)}，" f"亏损上限 {_decimal_text(settings.daily_realized_loss_limit)}" ), ) if reason == "metadata symbol mismatch": return ( { **base, "prediction_symbol": prediction.symbol, "metadata_symbol": context.metadata.symbol, }, f"预测 {prediction.symbol}，交易规则 {context.metadata.symbol}", ) if reason == "no position to close": return ( { **base, "position_side": context.position.side.value, "position_quantity": _decimal_text(context.position.quantity), }, f"当前持仓 {context.position.side.value}，数量 {_decimal_text(context.position.quantity)}", ) if reason == "invalid close percent": return ( { **base, "close_percent": prediction.close_percent, "allowed_close_percents": [25, 50, 100], }, f"当前 {prediction.close_percent}，允许 25/50/100", ) if reason in {"same-direction position exists", "opposite position requires close first"}: return ( { **base, "position_side": context.position.side.value, "position_quantity": _decimal_text(context.position.quantity), "action": prediction.action.value, }, ( f"当前持仓 {context.position.side.value}，" f"数量 {_decimal_text(context.position.quantity)}，动作 {prediction.action.value}" ), ) leverage = _resolved_risk_leverage(prediction, settings) if reason == "total margin above max": projected_margin = context.total_isolated_margin + settings.margin_amount return ( { **base, "total_isolated_margin": _decimal_text(context.total_isolated_margin), "margin_amount": _decimal_text(settings.margin_amount), "projected_total_isolated_margin": _decimal_text(projected_margin), "max_total_margin": _decimal_text(settings.max_total_margin), }, ( f"当前 {_decimal_text(context.total_isolated_margin)}，" f"新增 {_decimal_text(settings.margin_amount)}，" f"预计 {_decimal_text(projected_margin)}，" f"最大 {_decimal_text(settings.max_total_margin)}" ), ) if reason == "notional below min notional" and leverage is not None: notional = settings.margin_amount * Decimal(leverage) return ( { **base, "margin_amount": _decimal_text(settings.margin_amount), "leverage": leverage, "notional": _decimal_text(notional), "min_notional": _decimal_text(context.metadata.min_notional), }, f"当前 {_decimal_text(notional)}，最小 {_decimal_text(context.metadata.min_notional)}", ) if reason == "missing protection prices": return ( { **base, "stop_loss_price": _decimal_text(prediction.stop_loss_price), "take_profit_price": _decimal_text(prediction.take_profit_price), }, ( f"止损 {_decimal_text(prediction.stop_loss_price) or '-'}，" f"止盈 {_decimal_text(prediction.take_profit_price) or '-'}" ), ) if reason == "invalid stop or take-profit direction": return ( { **base, "mark_price": _decimal_text(context.mark_price), "stop_loss_price": _decimal_text(prediction.stop_loss_price), "take_profit_price": _decimal_text(prediction.take_profit_price), }, ( f"标记价 {_decimal_text(context.mark_price)}，" f"止损 {_decimal_text(prediction.stop_loss_price)}，" f"止盈 {_decimal_text(prediction.take_profit_price)}" ), ) if reason == "stop distance outside limits" and prediction.stop_loss_price is not None: distance = _distance_percent(context.mark_price, prediction.stop_loss_price) return ( { **base, "mark_price": _decimal_text(context.mark_price), "stop_loss_price": _decimal_text(prediction.stop_loss_price), "stop_distance_percent": _decimal_text(distance), "min_stop_distance_percent": _decimal_text(settings.min_stop_distance_percent), "max_stop_distance_percent": _decimal_text(settings.max_stop_distance_percent), }, ( f"当前 {_decimal_text(distance)}%，" f"最小 {_decimal_text(settings.min_stop_distance_percent)}%，" f"最大 {_decimal_text(settings.max_stop_distance_percent)}%" ), ) if reason == "take-profit distance outside limits" and prediction.take_profit_price is not None: distance = _distance_percent(context.mark_price, prediction.take_profit_price) return ( { **base, "mark_price": _decimal_text(context.mark_price), "take_profit_price": _decimal_text(prediction.take_profit_price), "take_profit_distance_percent": _decimal_text(distance), "min_take_profit_distance_percent": _decimal_text(settings.min_take_profit_distance_percent), "max_take_profit_distance_percent": _decimal_text(settings.max_take_profit_distance_percent), }, ( f"当前 {_decimal_text(distance)}%，" f"最小 {_decimal_text(settings.min_take_profit_distance_percent)}%，" f"最大 {_decimal_text(settings.max_take_profit_distance_percent)}%" ), ) if reason == "estimated liquidation buffer below minimum" and leverage is not None: buffer_percent = Decimal("100") / Decimal(leverage) return ( { **base, "leverage": leverage, "estimated_liquidation_buffer_percent": _decimal_text(buffer_percent), "min_liquidation_buffer_percent": _decimal_text(settings.min_liquidation_buffer_percent), }, ( f"当前 {_decimal_text(buffer_percent)}%，" f"最小 {_decimal_text(settings.min_liquidation_buffer_percent)}%" ), ) if ( reason == "estimated stop loss above margin risk budget" and leverage is not None and prediction.stop_loss_price is not None ): notional = settings.margin_amount * Decimal(leverage) stop_distance_percent = _distance_percent(context.mark_price, prediction.stop_loss_price) stop_loss_amount = notional * stop_distance_percent / Decimal("100") max_loss_amount = settings.margin_amount * settings.max_margin_loss_percent / Decimal("100") return ( { **base, "margin_amount": _decimal_text(settings.margin_amount), "leverage": leverage, "notional": _decimal_text(notional), "stop_distance_percent": _decimal_text(stop_distance_percent), "estimated_stop_loss_amount": _decimal_text(stop_loss_amount), "max_loss_amount": _decimal_text(max_loss_amount), "max_margin_loss_percent": _decimal_text(settings.max_margin_loss_percent), }, ( f"当前 {_decimal_text(stop_loss_amount)}，" f"最大 {_decimal_text(max_loss_amount)}" ), ) if reason == "hedge mode active": return ( { **base, "hedge_mode": context.hedge_mode, "required_position_mode": "one-way", }, "当前 hedge mode，要求 one-way", ) return None, None def _resolved_risk_leverage(prediction: FuturesPrediction, settings: FuturesRiskSettings) -> int | None: leverage = settings.default_leverage if prediction.leverage is None else prediction.leverage if not isinstance(leverage, int) or isinstance(leverage, bool) or leverage <= 0: return None if ( not isinstance(settings.max_leverage, int) or isinstance(settings.max_leverage, bool) or settings.max_leverage <= 0 ): return None return min(leverage, settings.max_leverage) def _distance_percent(reference: Decimal, price: Decimal) -> Decimal: return abs(reference - price) / reference * Decimal("100") def _strategy_rejection( settings: Settings, storage: Storage, prediction: FuturesPrediction, context: FuturesRiskContext, exchange, now_ms: int, ) -> tuple[str, str, dict[str, Any] | None] | None: if prediction.action not in {FuturesAction.OPEN_LONG, FuturesAction.OPEN_SHORT}: return None if settings.futures_reverse_cooldown_seconds > 0: since_ms = now_ms - settings.futures_reverse_cooldown_seconds * 1000 if storage.has_recent_futures_close(prediction.symbol, since_ms): return ( "策略", "reverse cooldown active", { "reason": "reverse cooldown active", "cooldown_seconds": settings.futures_reverse_cooldown_seconds, "since_ms": since_ms, "current_recent_close": True, }, ) klines_4h = _safe_klines(exchange, prediction.symbol, "4h", 30) if settings.futures_trend_filter_enabled: klines_1d = _safe_klines(exchange, prediction.symbol, "1d", 30) trend_decision = _trend_filter_decision(prediction.action, klines_4h, klines_1d, now_ms) if trend_decision.blocked: return "策略", "trend filter blocked", trend_decision.details atr = _atr_from_klines(klines_4h, period=14) if ( atr is not None and prediction.stop_loss_price is not None and settings.futures_min_stop_atr_multiple > 0 ): stop_distance = abs(context.mark_price - prediction.stop_loss_price) min_stop_distance_from_atr = atr * settings.futures_min_stop_atr_multiple if stop_distance < min_stop_distance_from_atr: return ( "风控", "stop distance below volatility floor", { "reason": "stop distance below volatility floor", "mark_price": _decimal_text(context.mark_price), "stop_loss_price": _decimal_text(prediction.stop_loss_price), "stop_distance": _decimal_text(stop_distance), "atr_4h": _decimal_text(atr), "min_stop_atr_multiple": _decimal_text(settings.futures_min_stop_atr_multiple), "min_stop_distance_from_atr": _decimal_text(min_stop_distance_from_atr), }, ) return None def _rejection_detail_text(details: dict[str, Any] | None) -> str | None: if not details: return None reason = details.get("reason") if reason == "reverse cooldown active": return f"当前冷却窗口内有平仓，冷却 {details.get('cooldown_seconds')} 秒" if reason == "trend filter blocked": primary = details.get("primary") if isinstance(primary, dict): interval = primary.get("interval") or "4h" direction = primary.get("direction") or "-" delta_percent = primary.get("delta_percent") or "-" threshold_percent = details.get("threshold_percent") or "-" return f"{interval}方向 {direction}，变化 {delta_percent}%，阈值 {threshold_percent}%" return None if reason == "stop distance below volatility floor": stop_distance = details.get("stop_distance") minimum = details.get("min_stop_distance_from_atr") if stop_distance is not None and minimum is not None: return f"当前 {stop_distance}，最小 {minimum}" return None def _safe_klines(exchange, symbol: str, interval: str, limit: int) -> list: rows = exchange.klines(symbol, interval, limit) return list(rows) if isinstance(rows, list) else [] def _trend_filter_decision( action: FuturesAction, rows_4h: list, rows_1d: list, now_ms: int, ) -> _TrendFilterDecision: primary = _close_trend_evidence("4h", rows_4h, now_ms) confirmation = _close_trend_evidence("1d", rows_1d, now_ms) ema = _ema_trend_evidence(rows_4h, now_ms) details = { "reason": "trend filter blocked", "threshold_percent": _decimal_text(_FUTURES_TREND_THRESHOLD_PERCENT), "action": action.value, "primary": _trend_evidence_raw(primary), "confirmation": _trend_evidence_raw(confirmation), "ema": _ema_trend_evidence_raw(ema), } if primary.direction == "neutral": return _TrendFilterDecision(False, details) if not _direction_opposes_action(primary.direction, action): return _TrendFilterDecision(False, details) if _direction_opposes_action(confirmation.direction, action) or _ema_strongly_opposes_action(ema, action): return _TrendFilterDecision(True, details) return _TrendFilterDecision(False, details) def _close_trend_evidence(interval: str, rows: list, now_ms: int) -> _TrendEvidence: closes = _closed_kline_closes(rows, now_ms) if len(closes) < 2: return _TrendEvidence(interval, None, None, None, "neutral") prev_close = closes[-2] last_close = closes[-1] if prev_close <= 0: return _TrendEvidence(interval, prev_close, last_close, None, "neutral") delta_percent = (last_close - prev_close) / prev_close * Decimal("100") return _TrendEvidence( interval, prev_close, last_close, delta_percent, _direction_from_delta(delta_percent), ) def _ema_trend_evidence(rows: list, now_ms: int) -> _EmaTrendEvidence: closes = _closed_kline_closes(rows, now_ms) fast = _ema(closes, _FUTURES_TREND_EMA_FAST_PERIOD) slow = _ema(closes, _FUTURES_TREND_EMA_SLOW_PERIOD) delta_percent = None direction = "neutral" if fast is not None and slow is not None and slow > 0: delta_percent = (fast - slow) / slow * Decimal("100") direction = _direction_from_delta(delta_percent, threshold=_FUTURES_TREND_EMA_THRESHOLD_PERCENT) return _EmaTrendEvidence( _FUTURES_TREND_EMA_FAST_PERIOD, _FUTURES_TREND_EMA_SLOW_PERIOD, fast, slow, delta_percent, direction, ) def _closed_kline_closes(rows: list, now_ms: int) -> list[Decimal]: closes: list[Decimal] = [] for row in rows: if not isinstance(row, (list, tuple)) or len(row) < 5: continue if len(row) > 6: close_time = _optional_int(row[6]) if close_time is not None and close_time > now_ms: continue close = _optional_decimal(row[4]) if close is not None: closes.append(close) return closes def _direction_from_delta( delta_percent: Decimal | None, *, threshold: Decimal = _FUTURES_TREND_THRESHOLD_PERCENT, ) -> str: if delta_percent is None: return "neutral" if delta_percent >= threshold: return "up" if delta_percent <= -threshold: return "down" return "neutral" def _direction_opposes_action(direction: str, action: FuturesAction) -> bool: return ( (action is FuturesAction.OPEN_LONG and direction == "down") or (action is FuturesAction.OPEN_SHORT and direction == "up") ) def _ema_strongly_opposes_action(evidence: _EmaTrendEvidence, action: FuturesAction) -> bool: return _direction_opposes_action(evidence.direction, action) def _ema(values: list[Decimal], period: int) -> Decimal | None: if period <= 0 or len(values) < period: return None ema = sum(values[:period], Decimal("0")) / Decimal(period) multiplier = Decimal("2") / Decimal(period + 1) for value in values[period:]: ema = (value - ema) * multiplier + ema return ema def _trend_evidence_raw(evidence: _TrendEvidence) -> dict[str, Any]: return { "interval": evidence.interval, "prev_close": _decimal_text(evidence.prev_close), "last_close": _decimal_text(evidence.last_close), "delta_percent": _decimal_text(evidence.delta_percent), "direction": evidence.direction, } def _ema_trend_evidence_raw(evidence: _EmaTrendEvidence) -> dict[str, Any]: return { "fast_period": evidence.fast_period, "slow_period": evidence.slow_period, "fast": _decimal_text(evidence.fast), "slow": _decimal_text(evidence.slow), "delta_percent": _decimal_text(evidence.delta_percent), "direction": evidence.direction, "threshold_percent": _decimal_text(_FUTURES_TREND_EMA_THRESHOLD_PERCENT), } def _kline_closes(rows: list) -> list[Decimal]: closes: list[Decimal] = [] for row in rows: if not isinstance(row, (list, tuple)) or len(row) < 5: continue close = _optional_decimal(row[4]) if close is not None: closes.append(close) return closes def _atr_from_klines(rows: list, period: int = 14) -> Decimal | None: candles: list[tuple[Decimal, Decimal, Decimal]] = [] for row in rows: if not isinstance(row, (list, tuple)) or len(row) < 5: continue high = _optional_decimal(row[2]) low = _optional_decimal(row[3]) close = _optional_decimal(row[4]) if high is None or low is None or close is None: continue candles.append((high, low, close)) if len(candles) <= period: return None true_ranges: list[Decimal] = [] for previous, current in zip(candles[:-1], candles[1:]): previous_close = previous[2] high, low, _close = current true_ranges.append(max(high - low, abs(high - previous_close), abs(low - previous_close))) atr = sum(true_ranges[:period], Decimal("0")) / Decimal(period) for true_range in true_ranges[period:]: atr = ((atr * Decimal(period - 1)) + true_range) / Decimal(period) return atr def _record_position_snapshot( storage: Storage, exchange, run_id: int | None, symbol: str, created_at_ms: int, ) -> None: if run_id is None: return try: rows = exchange.position_risk(symbol) if not rows: return position = parse_position(rows[0]) storage.record_futures_position_snapshot( run_id, position.symbol, position.side.value, position.quantity, position.entry_price, position.mark_price, position.liquidation_price, position.isolated_margin, position.unrealized_pnl, created_at_ms=created_at_ms, ) except Exception: return def _existing_protection_prices(open_algo_orders: list[dict]) -> tuple[Decimal | None, Decimal | None, tuple[str, ...]]: stop_loss_price = None take_profit_price = None client_algo_ids: list[str] = [] for order in open_algo_orders: if not isinstance(order, dict): continue status = str(order.get("algoStatus", order.get("status", ""))).upper() if status not in FUTURES_ACTIVE_ALGO_ORDER_STATUSES: continue client_algo_id = str(order.get("clientAlgoId", "")) if not client_algo_id.startswith("bqf-"): continue order_type = str(order.get("orderType", order.get("type", ""))) trigger_price = _optional_decimal(order.get("triggerPrice", order.get("price", ""))) client_algo_ids.append(client_algo_id) if trigger_price is None or trigger_price <= 0: continue if order_type == "STOP_MARKET": stop_loss_price = trigger_price if order_type == "TAKE_PROFIT_MARKET": take_profit_price = trigger_price return stop_loss_price, take_profit_price, tuple(client_algo_ids) def _futures_batch_status(results: list[FuturesExecutionResult]) -> str: if any(item.status in FUTURES_FAILED_STATUSES for item in results): return "failed" if any(item.status not in {"no_action", "rejected", "skipped"} for item in results): return "completed" return "no_action" def _futures_batch_reason(status: str) -> str: if status == "failed": return "AI合约批量交易失败" if status == "no_action": return "无合约动作" return "AI合约批量交易完成" def _futures_result_reason( prediction: FuturesPrediction, stage: str, reason: str, detail_text: str | None = None, ) -> str: translated = _futures_reason(reason) if detail_text: translated = f"{translated}（{detail_text}）" return f"AI: {prediction.reason}; {stage}: {translated}" def _futures_reason(reason: str) -> str: return _FUTURES_REASON_TRANSLATIONS.get(reason, reason) def _preflight_failure_raw(stage: str, error: Exception) -> dict[str, Any]: payload: dict[str, Any] = { "reason": FUTURES_PREFLIGHT_CONTEXT_FAILED_REASON, "stage": stage, "error": { "type": type(error).__name__, "message": _sanitize_error_message(str(error)), }, } response = getattr(error, "response", None) status_code = getattr(response, "status_code", None) if status_code is not None: payload["error"]["status_code"] = status_code return payload def _sanitize_error_message(message: str) -> str: sanitized = message for pattern in _SENSITIVE_ERROR_PATTERNS: sanitized = pattern.sub("***REDACTED***", sanitized) return sanitized def _prediction_schema(allowed_symbols: tuple[str, ...]) -> dict[str, Any]: symbols = [symbol.upper() for symbol in allowed_symbols] return { "type": "object", "additionalProperties": False, "required": ["predictions"], "properties": { "predictions": { "type": "array", "minItems": len(symbols), "maxItems": len(symbols), "items": { "type": "object", "additionalProperties": False, "required": [ "symbol", "action", "confidence", "leverage", "stop_loss_price", "take_profit_price", "close_percent", "reason", ], "properties": { "symbol": {"type": "string", "enum": symbols}, "action": {"type": "string", "enum": ["OPEN_LONG", "OPEN_SHORT", "CLOSE", "HOLD"]}, "confidence": {"type": "string", "pattern": r"^(0(\.\d{1,4})?|1(\.0{1,4})?)$"}, "leverage": {"type": ["integer", "null"], "minimum": 1}, "stop_loss_price": {"type": ["string", "null"]}, "take_profit_price": {"type": ["string", "null"]}, "close_percent": {"type": ["integer", "null"], "enum": [25, 50, 100, None]}, "reason": {"type": "string", "minLength": 1}, }, }, } }, } def _build_futures_prompt(context: dict[str, Any]) -> str: return "\n".join( [ "You are a futures signal generator for a local Binance USD-M futures 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.", "Choose exactly one action per symbol: OPEN_LONG, OPEN_SHORT, CLOSE, or HOLD.", "Do not place trades, call tools, edit files, or include markdown.", "Local risk and execution checks remain authoritative; your job is only to produce candidate signals.", "Use market[].position to decide whether CLOSE is appropriate for an existing position.", "Use OPEN_LONG only when trend, momentum, liquidity, order book, and derivatives context support upside.", "Use OPEN_SHORT only when trend, momentum, liquidity, order book, and derivatives context support downside.", "Choose CLOSE when the existing position risk has increased or the original thesis is invalidated.", "Choose HOLD when evidence is mixed, weak, stale, or a new position would rely on one isolated signal.", "For OPEN_LONG and OPEN_SHORT, include stop_loss_price and take_profit_price as strings.", "For OPEN_LONG and OPEN_SHORT, include leverage only when a value below or equal to context.risk.max_leverage is justified.", "For CLOSE, include close_percent as one of 25, 50, or 100.", "Use null for leverage, stop_loss_price, take_profit_price, or close_percent when the field does not apply.", "Use confidence from 0.00 to 1.00.", "Use confidence at or above context.risk.min_confidence only when multiple independent evidence groups agree.", "For OPEN_LONG and OPEN_SHORT, use market[].local_risk to set protection prices that satisfy local risk.", "Stop-loss distance must be at least market[].local_risk.effective_min_stop_distance and no more than market[].local_risk.max_stop_distance_price when those fields are present.", "The effective minimum combines the fixed percent floor and the 4h ATR volatility floor.", "Every reason must be written in Chinese, and reason must describe the evidence in natural Chinese.", "Market context:", json.dumps(context, ensure_ascii=False, sort_keys=True, default=str), ] ) def _optional_decimal(value: object) -> Decimal | None: if value is None or value == "": return None try: parsed = Decimal(str(value)) except (InvalidOperation, ValueError): return None if not parsed.is_finite(): return None return parsed def _optional_int(value: object) -> int | None: if value is None or value == "": return None try: return int(value) except (TypeError, ValueError): return None def _optional_exact_int(value: object) -> int | None: if value is None: return None if isinstance(value, bool): return None if isinstance(value, int): return value if isinstance(value, str): text = value.strip() if text.isdecimal(): return int(text) if not text: return None return None def _futures_batch_rejection( storage: Storage, reason: str, max_actions: int, created_at_ms: int, raw: dict[str, Any] | None = None, ) -> FuturesAiBatchResult: raw_payload = raw or {"reason": reason} run_id = storage.record_futures_ai_trade_run( symbol=None, action=FuturesAction.HOLD.value, confidence=Decimal("0"), status="rejected", reason=reason, raw=raw_payload, created_at_ms=created_at_ms, ) result = FuturesExecutionResult( run_id=run_id, status="rejected", symbol="", action=FuturesAction.HOLD.value, reason=reason, confidence=Decimal("0"), raw=raw_payload, ) return FuturesAiBatchResult( status="rejected", reason=reason, max_actions=max_actions, prediction_count=0, executed_count=0, daily_realized_pnl=Decimal("0"), total_isolated_margin=Decimal("0"), results=(result,), ) def _position_rows_by_symbol(exchange, symbols: tuple[str, ...]) -> dict[str, list[dict]]: return {symbol: list(exchange.position_risk(symbol)) for symbol in symbols} def _account_position_rows(exchange) -> list[dict]: return list(exchange.position_risk()) def _daily_realized_pnl(exchange, now_ms: int) -> Decimal: total = Decimal("0") for row in exchange.income(_local_day_start_ms(now_ms), now_ms): if str(row.get("incomeType", "")).upper() not in FUTURES_DAILY_PNL_INCOME_TYPES: continue total += _decimal_or_zero(row.get("income")) return total def _total_isolated_margin(position_rows: list[dict]) -> Decimal: total = Decimal("0") for row in position_rows: total += parse_position(row).isolated_margin return total def _model_context( settings: Settings, exchange, position_rows: dict[str, list[dict]], account_snapshot: dict, *, daily_realized_pnl: Decimal, total_isolated_margin: Decimal, now_ms: int, ) -> dict[str, Any]: market = [] for symbol in settings.futures_ai_symbols: mark = exchange.mark_price(symbol) depth = exchange.depth(symbol, limit=20) ticker = exchange.ticker_24hr(symbol) open_interest = exchange.open_interest(symbol) open_interest_history = exchange.open_interest_hist(symbol, period="15m", limit=12) long_short = exchange.global_long_short_ratio(symbol, period="5m", limit=1) taker_long_short = exchange.taker_long_short_ratio(symbol, period="5m", limit=1) funding_history = exchange.funding_rate_history(symbol, limit=8) leverage_bracket = exchange.leverage_bracket(symbol) metadata = exchange.exchange_info(symbol) klines = _klines_by_interval(exchange, symbol, settings.ai_trader_klines) market.append( { "symbol": symbol, "metadata": _metadata_features(metadata), "mark_price": str(mark.get("markPrice", "")), "funding": { "last_funding_rate": str(mark.get("lastFundingRate", "")), "next_funding_time": mark.get("nextFundingTime"), "history": funding_history, }, "open_interest": str(open_interest.get("openInterest", "")), "open_interest_history": open_interest_history, "leverage_bracket": _leverage_bracket_features(leverage_bracket, symbol), "long_short_ratio": _latest_ratio(long_short, "longShortRatio"), "taker_long_short_ratio": _latest_ratio(taker_long_short, "buySellRatio"), "order_book": _order_book_features(depth), "ticker_24h": ticker, "klines": klines, "local_risk": _local_risk_features(settings, mark, klines), "aggregate_trades": exchange.aggregate_trades(symbol, limit=settings.ai_trader_agg_trades_limit), "position": _position_features(position_rows.get(symbol, [])), "algo_orders": exchange.all_algo_orders(symbol), } ) return { "symbols": list(settings.futures_ai_symbols), "created_at_ms": now_ms, "account": { "daily_realized_pnl": str(daily_realized_pnl), "daily_net_pnl": str(daily_realized_pnl), "total_isolated_margin": str(total_isolated_margin), "max_total_margin": str(settings.futures_max_total_margin), **_account_features(account_snapshot), }, "risk": { "margin_amount": str(settings.futures_margin_amount), "default_leverage": settings.futures_default_leverage, "max_leverage": settings.futures_max_leverage, "min_confidence": str(settings.futures_ai_min_confidence), "min_stop_distance_percent": str(settings.futures_min_stop_distance_percent), "max_stop_distance_percent": str(settings.futures_max_stop_distance_percent), "min_stop_atr_multiple": str(settings.futures_min_stop_atr_multiple), }, "market": market, } def _metadata_features(metadata) -> dict[str, str]: return { "symbol": metadata.symbol, "base_asset": metadata.base_asset, "quote_asset": metadata.quote_asset, "min_qty": str(metadata.min_qty), "step_size": str(metadata.step_size), "tick_size": str(metadata.tick_size), "min_notional": str(metadata.min_notional), } def _local_risk_features(settings: Settings, mark: dict, klines: dict[str, list]) -> dict[str, str]: mark_price = _optional_decimal(mark.get("markPrice", "")) if mark_price is None: return { "atr_4h": "", "min_stop_distance_from_atr": "", "min_stop_distance_price": "", "effective_min_stop_distance": "", "max_stop_distance_price": "", } min_stop_distance_price = mark_price * settings.futures_min_stop_distance_percent / Decimal("100") max_stop_distance_price = mark_price * settings.futures_max_stop_distance_percent / Decimal("100") atr = _atr_from_klines(klines.get("4h", []), period=14) min_stop_distance_from_atr = ( atr * settings.futures_min_stop_atr_multiple if atr is not None and settings.futures_min_stop_atr_multiple > 0 else None ) effective_min_stop_distance = min_stop_distance_price if min_stop_distance_from_atr is not None: effective_min_stop_distance = max(effective_min_stop_distance, min_stop_distance_from_atr) return { "atr_4h": _decimal_text(atr), "min_stop_distance_from_atr": _decimal_text(min_stop_distance_from_atr), "min_stop_distance_price": _decimal_text(min_stop_distance_price), "effective_min_stop_distance": _decimal_text(effective_min_stop_distance), "max_stop_distance_price": _decimal_text(max_stop_distance_price), } def _account_features(payload: dict) -> dict[str, str]: return { "available_balance": str(payload.get("availableBalance", "")), "total_wallet_balance": str(payload.get("totalWalletBalance", "")), "total_maint_margin": str(payload.get("totalMaintMargin", "")), "total_position_initial_margin": str(payload.get("totalPositionInitialMargin", "")), "total_open_order_initial_margin": str(payload.get("totalOpenOrderInitialMargin", "")), "total_unrealized_pnl": str(payload.get("totalUnrealizedProfit", "")), } def _leverage_bracket_features(payload: list[dict], symbol: str) -> dict[str, Any]: for row in payload: if not isinstance(row, dict): continue if str(row.get("symbol", "")).upper() == symbol.upper(): brackets = row.get("brackets") return {"brackets": brackets if isinstance(brackets, list) else []} return {"brackets": []} def _order_book_features(depth: dict) -> dict[str, str]: bids = depth.get("bids") if isinstance(depth, dict) else None asks = depth.get("asks") if isinstance(depth, dict) else None best_bid = _level_decimal(bids, 0) best_ask = _level_decimal(asks, 0) bid_quantity = _level_decimal(bids, 1) ask_quantity = _level_decimal(asks, 1) spread = best_ask - best_bid if best_bid is not None and best_ask is not None else None features = { "best_bid": _decimal_text(best_bid), "best_ask": _decimal_text(best_ask), "best_bid_quantity": _decimal_text(bid_quantity), "best_ask_quantity": _decimal_text(ask_quantity), "spread": _decimal_text(spread), } for depth_limit in (3, 5, 10, 20): bid_total = _quantity_total(bids, depth_limit) ask_total = _quantity_total(asks, depth_limit) features[f"bid_quantity_top_{depth_limit}"] = _decimal_text(bid_total) features[f"ask_quantity_top_{depth_limit}"] = _decimal_text(ask_total) features[f"quantity_imbalance_top_{depth_limit}"] = _decimal_text(_quantity_imbalance(bid_total, ask_total)) return features def _quantity_total(levels: object, limit: int) -> Decimal | None: if not isinstance(levels, list): return None total = Decimal("0") counted = 0 for level in levels[:limit]: if not isinstance(level, list) or len(level) < 2: continue quantity = _optional_decimal(level[1]) if quantity is None: continue total += quantity counted += 1 return total if counted else None def _quantity_imbalance(bid_total: Decimal | None, ask_total: Decimal | None) -> Decimal | None: if bid_total is None or ask_total is None: return None total = bid_total + ask_total if total == 0: return None return (bid_total - ask_total) / total def _level_decimal(levels: object, index: int) -> Decimal | None: if not isinstance(levels, list) or not levels: return None first = levels[0] if not isinstance(first, list) or len(first) <= index: return None return _optional_decimal(first[index]) def _klines_by_interval(exchange, symbol: str, intervals: tuple[tuple[str, int], ...]) -> dict[str, list]: return {interval: exchange.klines(symbol, interval, limit) for interval, limit in intervals} def _latest_ratio(rows: object, key: str) -> str: if not isinstance(rows, list) or not rows: return "" row = rows[-1] if not isinstance(row, dict): return "" return str(row.get(key, "")) def _position_features(rows: list[dict]) -> dict[str, str]: if not rows: return {} return _single_position_features(parse_position(rows[0])) def _single_position_features(position: FuturesPosition) -> dict[str, str]: return { "symbol": position.symbol, "quantity": str(position.quantity), "side": position.side.value, "entry_price": str(position.entry_price), "mark_price": str(position.mark_price), "liquidation_price": str(position.liquidation_price), "isolated_margin": str(position.isolated_margin), "unrealized_pnl": str(position.unrealized_pnl), "leverage": str(position.leverage), "margin_type": position.margin_type, } def _prediction_list(raw_predictions: object, allowed_symbols: tuple[str, ...]) -> list[FuturesAiPrediction]: if isinstance(raw_predictions, FuturesAiPrediction): return _normalize_predictions([raw_predictions], allowed_symbols) if isinstance(raw_predictions, (list, tuple)) and all(isinstance(item, FuturesAiPrediction) for item in raw_predictions): return _normalize_predictions(list(raw_predictions), allowed_symbols) return [FuturesAiPrediction.hold("模型响应无效", symbol) for symbol in allowed_symbols] def _normalize_predictions( predictions: list[FuturesAiPrediction], allowed_symbols: tuple[str, ...], ) -> list[FuturesAiPrediction]: allowed = tuple(symbol.upper() for symbol in allowed_symbols) by_symbol: dict[str, FuturesAiPrediction] = {} invalid_symbols: set[str] = set() for prediction in predictions: if prediction.symbol not in allowed: continue validated = _validated_prediction(prediction, allowed) if validated is None: invalid_symbols.add(prediction.symbol) continue by_symbol.setdefault(validated.symbol, validated) result: list[FuturesAiPrediction] = [] for symbol in allowed: if symbol in by_symbol: result.append(by_symbol[symbol]) elif symbol in invalid_symbols: result.append(FuturesAiPrediction.hold("模型返回的该交易对预测无效", symbol)) else: result.append(FuturesAiPrediction.hold("模型未返回该交易对预测", symbol)) return result def _validated_prediction( prediction: FuturesAiPrediction, allowed_symbols: tuple[str, ...], ) -> FuturesAiPrediction | None: action = prediction.action.value if isinstance(prediction.action, FuturesAction) else prediction.action return FuturesAiPrediction.from_payload( { "symbol": prediction.symbol, "action": action, "confidence": prediction.confidence, "leverage": prediction.leverage, "stop_loss_price": prediction.stop_loss_price, "take_profit_price": prediction.take_profit_price, "close_percent": prediction.close_percent, "reason": prediction.reason, }, allowed_symbols, ) def _record_futures_ai_trade_predictions( storage: Storage, predictions: list[FuturesAiPrediction], created_at_ms: int, ) -> list[tuple[FuturesAiPrediction, int]]: rows: list[tuple[FuturesAiPrediction, int]] = [] for prediction in predictions: run_id = storage.record_futures_ai_trade_run( symbol=prediction.symbol, action=prediction.action.value, confidence=prediction.confidence, status="predicted", reason=prediction.reason, raw=_prediction_raw(prediction), leverage=prediction.leverage, margin_amount=None, stop_loss_price=prediction.stop_loss_price, take_profit_price=prediction.take_profit_price, close_percent=prediction.close_percent, created_at_ms=created_at_ms, ) rows.append((prediction, run_id)) return rows def _prediction_raw(prediction: FuturesAiPrediction) -> dict[str, str | int | None]: return { "symbol": prediction.symbol, "action": prediction.action.value, "confidence": str(prediction.confidence), "leverage": prediction.leverage, "stop_loss_price": _decimal_text(prediction.stop_loss_price), "take_profit_price": _decimal_text(prediction.take_profit_price), "close_percent": prediction.close_percent, "reason": prediction.reason, } def _prediction_raw_with_execution_error( prediction: FuturesAiPrediction, stage: str, error: Exception, ) -> dict[str, Any]: raw = _prediction_raw(prediction) raw["execution_error"] = _execution_error_raw(stage, error) return raw def _execution_error_raw(stage: str, error: Exception) -> dict[str, Any]: payload: dict[str, Any] = { "stage": stage, "error": { "type": type(error).__name__, "message": _sanitize_error_message(str(error)), }, } response = getattr(error, "response", None) status_code = getattr(response, "status_code", None) if status_code is not None: payload["error"]["status_code"] = status_code return payload def _batch_status(results: list[FuturesExecutionResult]) -> str: if not results: return "no_action" if any(result.status not in {"no_action", "skipped"} for result in results): return "completed" if any(result.status == "skipped" for result in results): return "skipped" return "no_action" def _decimal_or_zero(value: object) -> Decimal: parsed = _optional_decimal(value) return parsed if parsed is not None else Decimal("0") def _decimal_text(value: Decimal | None) -> str: return "" if value is None else str(value) 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 _hold_predictions(allowed_symbols: tuple[str, ...], reason: str) -> list[FuturesAiPrediction]: return [FuturesAiPrediction.hold(reason, symbol) for symbol in allowed_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}"