+"""Reconcile exchange-side closes that bypass Quick Trade routes."""\n\nfrom __future__ import annotations\n\nimport json\nfrom typing import Any\n\nfrom app.services.quick_trade.credentials import build_exchange_config, create_exchange_client\nfrom app.services.quick_trade.symbols import is_supported_crypto_exchange\nfrom app.utils.db import get_db_connection\nfrom app.utils.logger import get_logger\nfrom app.utils.request_guard import cache_key, guarded_cached\n\n\nlogger = get_logger(__name__)\n\n\ndef _sync_gate_native_protection_fills(\n *, user_id: int, credential_id: int, symbol: str, market_type: str\n) -> int:\n """Import Gate-native TP/SL fills that bypass the Quick Trade order route."""\n if market_type != "swap":\n return 0\n try:\n config = build_exchange_config(\n credential_id,\n user_id,\n {"market_type": market_type},\n )\n exchange_id = str(config.get("exchange_id") or "").strip().lower()\n if exchange_id not in ("gate", "gateio"):\n return 0\n client = create_exchange_client(config, market_type=market_type)\n from app.services.quick_trade.position_activity import gate_native_protection_fills\n\n payload = guarded_cached(\n cache_key(\n "quick_trade_gate_native_protection_fills",\n user_id,\n credential_id,\n market_type,\n symbol,\n ),\n lambda: gate_native_protection_fills(client, symbol, limit=100),\n ttl_sec=2,\n stale_ttl_sec=15,\n timeout_sec=4,\n namespace="quick_trade_gate_native_protection_fills",\n max_concurrent=4,\n )\n fills = payload.get("fills") if isinstance(payload, dict) else []\n if not fills:\n return 0\n\n inserted = 0\n with get_db_connection() as db:\n cur = db.cursor()\n for fill in fills:\n order_id = str(fill.get("exchange_order_id") or "").strip()\n if not order_id:\n continue\n close_side = str(fill.get("close_side") or "").strip().lower()\n entry_side = "buy" if close_side == "long" else "sell"\n cur.execute(\n """\n SELECT leverage, avg_fill_price\n FROM qd_quick_trades\n WHERE user_id = %s AND credential_id = %s\n AND UPPER(symbol) = UPPER(%s) AND market_type = %s\n AND side = %s AND status = 'filled'\n AND created_at <= %s\n ORDER BY created_at DESC\n LIMIT 1\n """,\n (\n user_id,\n credential_id,\n symbol,\n market_type,\n entry_side,\n fill.get("created_at"),\n ),\n )\n entry = cur.fetchone() or {}\n quantity = float(fill.get("filled_amount") or 0)\n avg_price = float(fill.get("avg_fill_price") or 0)\n commission = float(fill.get("commission") or 0)\n metadata = {\n "requested_base_qty": quantity,\n "exchange_status": "filled",\n "is_close": True,\n "close_side": close_side,\n "close_scope": "exchange_protection",\n "entry_price": float(entry.get("avg_fill_price") or 0),\n "fees_by_ccy": {"USDT": commission} if commission > 0 else {},\n "fee_status": "actual" if commission > 0 else "actual_zero",\n "native_protection_fill": True,\n }\n raw_result = {\n "gate_auto_order": {\n "text": str(fill.get("client_order_id") or ""),\n "exchange_fill_ids": list(fill.get("exchange_fill_ids") or []),\n },\n "_quick_trade": metadata,\n }\n cur.execute(\n """\n INSERT INTO qd_quick_trades\n (user_id, credential_id, exchange_id, symbol, side, order_type,\n amount, price, leverage, market_type, tp_price, sl_price,\n status, exchange_order_id, filled_amount, avg_fill_price,\n commission, commission_ccy, commission_quote,\n client_order_id, error_msg, source, raw_result, created_at)\n SELECT %s, %s, %s, %s, %s, 'market',\n %s, 0, %s, %s, 0, 0,\n 'filled', %s, %s, %s,\n %s, 'USDT', %s,\n %s, '', 'quick_trade', %s, %s\n WHERE NOT EXISTS (\n SELECT 1 FROM qd_quick_trades\n WHERE user_id = %s AND credential_id = %s\n AND market_type = %s AND exchange_order_id = %s\n )\n RETURNING id\n """,\n (\n user_id,\n credential_id,\n exchange_id,\n symbol,\n str(fill.get("side") or ""),\n quantity * avg_price,\n int(entry.get("leverage") or 1),\n market_type,\n order_id,\n quantity,\n avg_price,\n commission,\n commission,\n str(fill.get("client_order_id") or "")[:100],\n json.dumps(raw_result, ensure_ascii=False),\n fill.get("created_at"),\n user_id,\n credential_id,\n market_type,\n order_id,\n ),\n )\n if cur.fetchone():\n inserted += 1\n db.commit()\n cur.close()\n return inserted\n except Exception as exc:\n logger.info("Gate native protection fill sync deferred: %s", exc)\n return 0\n\n\ndef _json_safe_close_snapshot(snapshot: Any) -> Any:\n if not isinstance(snapshot, dict):\n return snapshot\n result = dict(snapshot)\n closed_at = result.get("closed_at")\n if hasattr(closed_at, "isoformat"):\n result["closed_at"] = closed_at.isoformat()\n components = []\n for value in result.get("components") or []:\n row = dict(value) if isinstance(value, dict) else {}\n timestamp = row.get("closed_at")\n if hasattr(timestamp, "isoformat"):\n row["closed_at"] = timestamp.isoformat()\n components.append(row)\n result["components"] = components\n return result\n\n\ndef sync_exchange_close_fills(\n *, user_id: int, credential_id: int, symbol: str, market_type: str\n) -> int:\n """Import exchange-side closes that bypassed the Quick Trade order route."""\n market = str(market_type or "swap").strip().lower()\n try:\n config = build_exchange_config(credential_id, user_id, {"market_type": market})\n exchange_id = str(config.get("exchange_id") or "").strip().lower()\n if exchange_id in ("gate", "gateio") and market == "swap":\n return _sync_gate_native_protection_fills(\n user_id=user_id,\n credential_id=credential_id,\n symbol=symbol,\n market_type=market,\n )\n if exchange_id != "alpaca" and not is_supported_crypto_exchange(exchange_id):\n return 0\n\n with get_db_connection() as db:\n cur = db.cursor()\n cur.execute(\n """\n SELECT id, side, amount, leverage, status, exchange_order_id,\n filled_amount, avg_fill_price, commission, commission_quote,\n raw_result, created_at\n FROM qd_quick_trades\n WHERE user_id = %s AND credential_id = %s\n AND UPPER(symbol) = UPPER(%s) AND market_type = %s\n ORDER BY created_at ASC, id ASC\n """,\n (user_id, credential_id, symbol, market),\n )\n rows = cur.fetchall() or []\n cur.close()\n if not rows:\n return 0\n\n from app.services.live_trading.exchange_close_history import resolve_exchange_external_close\n from app.services.quick_trade.position_activity import tracked_quick_trade_legs\n\n tracked = tracked_quick_trade_legs(rows, market)\n legs = tracked.get("legs") or []\n recorded_ids = set(tracked.get("recorded_ids") or set())\n if not legs:\n return 0\n client = create_exchange_client(config, market_type=market)\n inserted = 0\n for leg in legs:\n quantity = float(leg.get("quantity") or 0)\n opened_at = leg.get("opened_at")\n if quantity <= 0 or not opened_at:\n continue\n\n def _load_close():\n return _json_safe_close_snapshot(\n resolve_exchange_external_close(\n client,\n symbol=symbol,\n side=str(leg.get("side") or "long"),\n target_qty=quantity,\n opened_after=opened_at,\n market_type=market,\n is_recorded=lambda value: str(value or "") in recorded_ids,\n )\n )\n\n snapshot = guarded_cached(\n cache_key(\n "quick_trade_exchange_close_fills",\n user_id,\n credential_id,\n market,\n symbol,\n leg.get("side"),\n quantity,\n ),\n _load_close,\n ttl_sec=2,\n stale_ttl_sec=15,\n timeout_sec=5,\n namespace="quick_trade_exchange_close_fills",\n max_concurrent=4,\n cache_if=lambda value: bool(value),\n )\n if not isinstance(snapshot, dict):\n continue\n components = [row for row in snapshot.get("components") or [] if isinstance(row, dict)]\n fill_ids = [str(row.get("exchange_fill_id") or "") for row in components if row.get("exchange_fill_id")]\n order_id = str(snapshot.get("exchange_order_id") or "").strip()\n if not order_id and fill_ids:\n order_id = f"fill-{exchange_id}-{fill_ids[-1]}"[:120]\n if not order_id:\n continue\n avg_price = float(snapshot.get("price") or 0)\n commission = abs(float(snapshot.get("fee") or 0))\n fee_ccy = str(snapshot.get("fee_ccy") or ("USD" if exchange_id == "alpaca" else "USDT")).upper()\n commission_quote = commission if fee_ccy in {"USD", "USDT", "USDC"} else None\n close_side = str(leg.get("side") or "").lower()\n side = "sell" if close_side == "long" else "buy"\n metadata = {\n "requested_base_qty": quantity,\n "exchange_status": "filled",\n "is_close": True,\n "close_side": close_side,\n "close_scope": "exchange_protection",\n "entry_price": float(leg.get("entry_price") or 0),\n "fees_by_ccy": {fee_ccy: commission} if commission > 0 else {},\n "fee_status": "actual" if commission > 0 else "actual_zero",\n "native_protection_fill": True,\n "close_reason": str(snapshot.get("close_reason") or "exchange_native_close"),\n }\n raw_result = {\n "exchange_close": {\n "exchange_fill_ids": fill_ids,\n "components": components,\n "source": str(snapshot.get("source") or "fills_history"),\n },\n "_quick_trade": metadata,\n }\n with get_db_connection() as db:\n cur = db.cursor()\n cur.execute(\n """\n INSERT INTO qd_quick_trades\n (user_id, credential_id, exchange_id, symbol, side, order_type,\n amount, price, leverage, market_type, tp_price, sl_price,\n status, exchange_order_id, filled_amount, avg_fill_price,\n commission, commission_ccy, commission_quote,\n client_order_id, error_msg, source, raw_result, created_at)\n SELECT %s, %s, %s, %s, %s, 'market',\n %s, 0, %s, %s, 0, 0,\n 'filled', %s, %s, %s,\n %s, %s, %s,\n %s, '', 'quick_trade', %s, %s\n WHERE NOT EXISTS (\n SELECT 1 FROM qd_quick_trades\n WHERE user_id = %s AND credential_id = %s\n AND market_type = %s AND exchange_order_id = %s\n )\n RETURNING id\n """,\n (\n user_id,\n credential_id,\n exchange_id,\n symbol,\n side,\n quantity * avg_price,\n int(leg.get("leverage") or 1),\n market,\n order_id,\n quantity,\n avg_price,\n commission,\n fee_ccy,\n commission_quote,\n str(snapshot.get("client_order_id") or "")[:100],\n json.dumps(raw_result, ensure_ascii=False),\n snapshot.get("closed_at"),\n user_id,\n credential_id,\n market,\n order_id,\n ),\n )\n if cur.fetchone():\n inserted += 1\n recorded_ids.add(order_id)\n recorded_ids.update(fill_ids)\n db.commit()\n cur.close()\n return inserted\n except Exception as exc:\n logger.info("Exchange native protection fill sync deferred: %s", exc)\n return 0\n\n
0 commit comments