Signed-off-by: TIANHE <TIANHE@GMAIL.COM>
This commit is contained in:
TIANHE
2025-12-29 04:45:59 +08:00
parent d5e987b8ee
commit 278b88ec71
26 changed files with 3220 additions and 317 deletions
+19 -3
View File
@@ -107,7 +107,7 @@ def summary():
cur = db.cursor()
cur.execute(
"""
SELECT id, strategy_name, strategy_type, status, initial_capital
SELECT id, strategy_name, strategy_type, status, initial_capital, trading_config
FROM qd_strategies_trading
"""
)
@@ -116,7 +116,23 @@ def summary():
running = [s for s in strategies if (s.get("status") or "").strip().lower() == "running"]
indicator_strategy_count = len([s for s in running if (s.get("strategy_type") or "") == "IndicatorStrategy"])
ai_strategy_count = max(0, len(running) - indicator_strategy_count)
# "AI strategies" in dashboard card: count strategies that enabled AI analysis/filtering.
# This aligns with the UI toggle `enable_ai_filter` in trading_config.
def _truthy(v: Any) -> bool:
if v is True:
return True
if isinstance(v, (int, float)) and float(v) == 1:
return True
if isinstance(v, str) and v.strip().lower() in ("1", "true", "yes", "y", "on"):
return True
return False
ai_enabled_strategy_count = 0
for s in strategies:
tc = _safe_json_loads(s.get("trading_config"), {}) or {}
if isinstance(tc, dict) and _truthy(tc.get("enable_ai_filter")):
ai_enabled_strategy_count += 1
# Positions (best-effort)
with get_db_connection() as db:
@@ -212,7 +228,7 @@ def summary():
"code": 1,
"msg": "success",
"data": {
"ai_strategy_count": int(ai_strategy_count),
"ai_strategy_count": int(ai_enabled_strategy_count),
"indicator_strategy_count": int(indicator_strategy_count),
"total_equity": float(total_equity),
"total_pnl": float(total_pnl),
+1 -1
View File
@@ -116,7 +116,7 @@ def get_trades():
cur = db.cursor()
cur.execute(
"""
SELECT id, strategy_id, symbol, type, price, amount, value, commission, profit, created_at
SELECT id, strategy_id, symbol, type, price, amount, value, commission, commission_ccy, profit, created_at
FROM qd_strategy_trades
WHERE strategy_id = ?
ORDER BY id DESC
@@ -254,6 +254,82 @@ class BinanceFuturesClient(BaseRestClient):
"""
return self._signed_request("GET", "/fapi/v2/account", params={})
def get_user_trades(self, *, symbol: str, order_id: str = "", limit: int = 100) -> Any:
"""
Fetch user trades (fills).
Endpoint: GET /fapi/v1/userTrades
Note: Binance order endpoints do NOT include commissions; commissions live on fills.
"""
sym = to_binance_futures_symbol(symbol)
if not sym:
return []
params: Dict[str, Any] = {"symbol": sym}
if order_id:
# Binance expects numeric orderId; keep as string and let server validate.
params["orderId"] = str(order_id)
try:
lim = int(limit or 100)
except Exception:
lim = 100
lim = max(1, min(1000, lim))
params["limit"] = lim
data = self._signed_request("GET", "/fapi/v1/userTrades", params=params)
return data
def get_fee_for_order(self, *, symbol: str, order_id: str) -> Tuple[float, str]:
"""
Best-effort: sum commissions from fills for a specific order.
Returns: (total_fee, fee_ccy)
"""
try:
trades = self.get_user_trades(symbol=symbol, order_id=str(order_id or ""), limit=200)
except Exception:
trades = []
if not isinstance(trades, list):
return 0.0, ""
total_fee = 0.0
fee_ccy = ""
for t in trades:
if not isinstance(t, dict):
continue
try:
fee = float(t.get("commission") or 0.0)
except Exception:
fee = 0.0
ccy = str(t.get("commissionAsset") or "").strip()
if fee != 0.0:
total_fee += abs(float(fee))
if (not fee_ccy) and ccy:
fee_ccy = ccy
return float(total_fee), str(fee_ccy or "")
def set_leverage(self, *, symbol: str, leverage: float) -> Dict[str, Any]:
"""
Set futures leverage for a symbol (USDT-M).
Endpoint: POST /fapi/v1/leverage
Notes:
- Binance applies leverage per symbol.
- If leverage is not set, the exchange may keep a default (often 1x),
which will change the actual margin used for a given notional.
"""
sym = to_binance_futures_symbol(symbol)
if not sym:
raise LiveTradingError(f"Invalid symbol: {symbol}")
try:
lev = int(float(leverage or 1.0))
except Exception:
lev = 1
if lev < 1:
lev = 1
if lev > 125:
lev = 125
return self._signed_request("POST", "/fapi/v1/leverage", params={"symbol": sym, "leverage": lev})
def get_dual_side_position(self) -> Optional[bool]:
"""
Best-effort read of position mode:
@@ -322,6 +322,55 @@ class BinanceSpotClient(BaseRestClient):
"""
return self._signed_request("GET", "/api/v3/account", params={})
def get_my_trades(self, *, symbol: str, order_id: str = "", limit: int = 100) -> Any:
"""
Fetch spot trade fills.
Endpoint: GET /api/v3/myTrades
"""
sym = to_binance_futures_symbol(symbol)
if not sym:
return []
params: Dict[str, Any] = {"symbol": sym}
if order_id:
params["orderId"] = str(order_id)
try:
lim = int(limit or 100)
except Exception:
lim = 100
lim = max(1, min(1000, lim))
params["limit"] = lim
data = self._signed_request("GET", "/api/v3/myTrades", params=params)
return data
def get_fee_for_order(self, *, symbol: str, order_id: str) -> Tuple[float, str]:
"""
Best-effort: sum commissions from fills for a specific spot order.
Returns: (total_fee, fee_ccy)
"""
try:
trades = self.get_my_trades(symbol=symbol, order_id=str(order_id or ""), limit=200)
except Exception:
trades = []
if not isinstance(trades, list):
return 0.0, ""
total_fee = 0.0
fee_ccy = ""
for t in trades:
if not isinstance(t, dict):
continue
try:
fee = float(t.get("commission") or 0.0)
except Exception:
fee = 0.0
ccy = str(t.get("commissionAsset") or "").strip()
if fee != 0.0:
total_fee += abs(float(fee))
if (not fee_ccy) and ccy:
fee_ccy = ccy
return float(total_fee), str(fee_ccy or "")
def cancel_order(self, *, symbol: str, order_id: str = "", client_order_id: str = "") -> Dict[str, Any]:
sym = to_binance_futures_symbol(symbol)
params: Dict[str, Any] = {"symbol": sym}
@@ -0,0 +1,256 @@
"""
Bitfinex (direct REST) client (v2, exchange spot).
Auth headers:
- bfx-apikey
- bfx-nonce
- bfx-signature = hex(hmac_sha384(secret, "/api/v2" + path + nonce + body))
Notes:
- This client targets "exchange" (spot) order types: EXCHANGE MARKET / EXCHANGE LIMIT.
- Derivatives/perps are not fully implemented here; only best-effort spot execution.
"""
from __future__ import annotations
import hashlib
import hmac
import time
from typing import Any, Dict, Optional
from app.services.live_trading.base import BaseRestClient, LiveOrderResult, LiveTradingError
from app.services.live_trading.symbols import to_bitfinex_spot_symbol
from app.services.live_trading.symbols import to_bitfinex_perp_symbol
class BitfinexClient(BaseRestClient):
def __init__(self, *, api_key: str, secret_key: str, base_url: str = "https://api.bitfinex.com", timeout_sec: float = 15.0):
super().__init__(base_url=base_url, timeout_sec=timeout_sec)
self.api_key = (api_key or "").strip()
self.secret_key = (secret_key or "").strip()
if not self.api_key or not self.secret_key:
raise LiveTradingError("Missing Bitfinex api_key/secret_key")
def _nonce(self) -> str:
# Use ms; Bitfinex accepts monotonic increasing nonces.
return str(int(time.time() * 1000))
def _sign(self, path: str, nonce: str, body_str: str) -> str:
payload = f"/api/v2{path}{nonce}{body_str}"
return hmac.new(self.secret_key.encode("utf-8"), payload.encode("utf-8"), hashlib.sha384).hexdigest()
def _headers(self, nonce: str, sign: str) -> Dict[str, str]:
return {"bfx-apikey": self.api_key, "bfx-nonce": nonce, "bfx-signature": sign, "content-type": "application/json"}
def _signed_request(self, method: str, path: str, *, json_body: Optional[Dict[str, Any]] = None) -> Any:
m = str(method or "POST").upper()
nonce = self._nonce()
body_str = self._json_dumps(json_body) if json_body is not None else ""
sign = self._sign(path, nonce, body_str)
code, data, text = self._request(m, path, params=None, data=body_str if body_str else None, headers=self._headers(nonce, sign))
if code >= 400:
raise LiveTradingError(f"Bitfinex HTTP {code}: {text[:500]}")
return data
def _public_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None) -> Any:
code, data, text = self._request(method, path, params=params, headers=None, json_body=None, data=None)
if code >= 400:
raise LiveTradingError(f"Bitfinex HTTP {code}: {text[:500]}")
return data
def ping(self) -> bool:
try:
d = self._public_request("GET", "/v2/platform/status")
return isinstance(d, list) and d and int(d[0]) == 1
except Exception:
return False
def get_wallets(self) -> Any:
"""
Private endpoint to validate credentials (best-effort).
"""
return self._signed_request("POST", "/v2/auth/r/wallets", json_body={})
class BitfinexDerivativesClient(BitfinexClient):
"""
Bitfinex derivatives/perpetual client (best-effort).
Differences vs spot:
- Symbol uses tBASEF0:QUOTEF0 (e.g. tBTCF0:USTF0)
- Order type typically uses MARKET/LIMIT (not EXCHANGE MARKET/LIMIT)
"""
def place_market_order(self, *, symbol: str, side: str, size: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
if qty <= 0:
raise LiveTradingError("Invalid size")
sym = to_bitfinex_perp_symbol(symbol)
amt = qty if sd == "buy" else -qty
body: Dict[str, Any] = {"type": "MARKET", "symbol": sym, "amount": str(amt)}
if client_order_id:
try:
cid = int("".join([c for c in str(client_order_id) if c.isdigit()])[:18] or "0")
if cid > 0:
body["cid"] = cid
except Exception:
pass
raw = self._signed_request("POST", "/v2/auth/w/order/submit", json_body=body)
oid = ""
try:
if isinstance(raw, list) and len(raw) >= 4 and isinstance(raw[3], list) and raw[3]:
order = raw[3][0]
if isinstance(order, list) and order:
oid = str(order[0])
except Exception:
oid = ""
return LiveOrderResult(exchange_id="bitfinex", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw={"raw": raw})
def place_limit_order(self, *, symbol: str, side: str, size: float, price: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
px = float(price or 0.0)
if qty <= 0 or px <= 0:
raise LiveTradingError("Invalid size/price")
sym = to_bitfinex_perp_symbol(symbol)
amt = qty if sd == "buy" else -qty
body: Dict[str, Any] = {"type": "LIMIT", "symbol": sym, "amount": str(amt), "price": str(px)}
if client_order_id:
try:
cid = int("".join([c for c in str(client_order_id) if c.isdigit()])[:18] or "0")
if cid > 0:
body["cid"] = cid
except Exception:
pass
raw = self._signed_request("POST", "/v2/auth/w/order/submit", json_body=body)
oid = ""
try:
if isinstance(raw, list) and len(raw) >= 4 and isinstance(raw[3], list) and raw[3]:
order = raw[3][0]
if isinstance(order, list) and order:
oid = str(order[0])
except Exception:
oid = ""
return LiveOrderResult(exchange_id="bitfinex", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw={"raw": raw})
def get_positions(self) -> Any:
return self._signed_request("POST", "/v2/auth/r/positions", json_body={})
def place_market_order(self, *, symbol: str, side: str, size: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
if qty <= 0:
raise LiveTradingError("Invalid size")
sym = to_bitfinex_spot_symbol(symbol)
amt = qty if sd == "buy" else -qty
body: Dict[str, Any] = {"type": "EXCHANGE MARKET", "symbol": sym, "amount": str(amt)}
if client_order_id:
# Bitfinex uses numeric cid; best-effort digits only
try:
cid = int("".join([c for c in str(client_order_id) if c.isdigit()])[:18] or "0")
if cid > 0:
body["cid"] = cid
except Exception:
pass
raw = self._signed_request("POST", "/v2/auth/w/order/submit", json_body=body)
oid = ""
try:
# Response is usually [.., [order_fields]]
if isinstance(raw, list) and len(raw) >= 4 and isinstance(raw[3], list) and raw[3]:
order = raw[3][0]
if isinstance(order, list) and order:
oid = str(order[0])
except Exception:
oid = ""
return LiveOrderResult(exchange_id="bitfinex", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw={"raw": raw})
def place_limit_order(self, *, symbol: str, side: str, size: float, price: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
px = float(price or 0.0)
if qty <= 0 or px <= 0:
raise LiveTradingError("Invalid size/price")
sym = to_bitfinex_spot_symbol(symbol)
amt = qty if sd == "buy" else -qty
body: Dict[str, Any] = {"type": "EXCHANGE LIMIT", "symbol": sym, "amount": str(amt), "price": str(px)}
if client_order_id:
try:
cid = int("".join([c for c in str(client_order_id) if c.isdigit()])[:18] or "0")
if cid > 0:
body["cid"] = cid
except Exception:
pass
raw = self._signed_request("POST", "/v2/auth/w/order/submit", json_body=body)
oid = ""
try:
if isinstance(raw, list) and len(raw) >= 4 and isinstance(raw[3], list) and raw[3]:
order = raw[3][0]
if isinstance(order, list) and order:
oid = str(order[0])
except Exception:
oid = ""
return LiveOrderResult(exchange_id="bitfinex", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw={"raw": raw})
def cancel_order(self, *, order_id: str = "", client_order_id: str = "") -> Any:
if order_id:
try:
oid = int(float(order_id))
except Exception:
oid = 0
if oid <= 0:
raise LiveTradingError("Bitfinex cancel_order invalid order_id")
return self._signed_request("POST", "/v2/auth/w/order/cancel", json_body={"id": oid})
# Best-effort cancel by cid is possible via /auth/w/order/cancel/multi, but not implemented.
if client_order_id:
raise LiveTradingError("Bitfinex cancel by client_order_id is not implemented (requires cid date)")
raise LiveTradingError("Bitfinex cancel_order requires order_id")
def get_order(self, *, order_id: str) -> Any:
try:
oid = int(float(order_id))
except Exception:
oid = 0
if oid <= 0:
raise LiveTradingError("Bitfinex get_order invalid order_id")
# Bitfinex v2 order status endpoint
return self._signed_request("POST", f"/v2/auth/r/order/{oid}")
def wait_for_fill(self, *, order_id: str, max_wait_sec: float = 10.0, poll_interval_sec: float = 0.5) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Any = None
while True:
try:
last = self.get_order(order_id=str(order_id))
except Exception:
last = last or []
filled = 0.0
avg_price = 0.0
status = ""
# best-effort parsing from array fields
try:
if isinstance(last, list) and len(last) >= 15:
status = str(last[13] or "")
amount_remaining = float(last[6] or 0.0)
amount_orig = float(last[7] or 0.0)
filled = abs(amount_orig - amount_remaining)
avg_price = float(last[14] or 0.0)
except Exception:
pass
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if isinstance(status, str) and ("EXECUTED" in status.upper() or "CANCELED" in status.upper()):
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
@@ -316,6 +316,8 @@ class BitgetSpotClient(BaseRestClient):
fills = data if isinstance(data, list) else []
total_base = 0.0
total_quote = 0.0
total_fee = 0.0
fee_ccy = ""
if isinstance(fills, list):
for f in fills:
try:
@@ -324,10 +326,33 @@ class BitgetSpotClient(BaseRestClient):
if sz > 0 and px > 0:
total_base += sz
total_quote += sz * px
# Best-effort fee extraction (fields vary by endpoint/version).
fee_v = f.get("fee")
if fee_v is None:
fee_v = f.get("fillFee")
if fee_v is None:
fee_v = f.get("tradeFee")
try:
fee = float(fee_v or 0.0)
except Exception:
fee = 0.0
ccy = str(f.get("feeCoin") or f.get("feeCcy") or f.get("fillFeeCoin") or f.get("fillFeeCcy") or "").strip()
if fee != 0.0:
total_fee += abs(float(fee))
if (not fee_ccy) and ccy:
fee_ccy = ccy
except Exception:
continue
if total_base > 0 and total_quote > 0:
return {"filled": total_base, "avg_price": total_quote / total_base, "state": state, "order": last_order, "fills": last_fills}
return {
"filled": total_base,
"avg_price": total_quote / total_base,
"fee": float(total_fee),
"fee_ccy": str(fee_ccy or ""),
"state": state,
"order": last_order,
"fills": last_fills
}
except Exception:
pass
@@ -0,0 +1,350 @@
"""
Bybit (direct REST) client for spot / linear perpetual orders (v5).
Signing (v5):
- X-BAPI-SIGN = hex(hmac_sha256(secret, timestamp + api_key + recv_window + payload))
- payload:
- GET: query string (sorted, urlencoded)
- POST: raw body string
"""
from __future__ import annotations
import hashlib
import hmac
import time
from decimal import Decimal, ROUND_DOWN
from typing import Any, Dict, Optional, Tuple
from urllib.parse import urlencode
from app.services.live_trading.base import BaseRestClient, LiveOrderResult, LiveTradingError
from app.services.live_trading.symbols import to_bybit_symbol
class BybitClient(BaseRestClient):
def __init__(
self,
*,
api_key: str,
secret_key: str,
base_url: str = "https://api.bybit.com",
timeout_sec: float = 15.0,
category: str = "linear", # "linear" (USDT perpetual) or "spot"
recv_window_ms: int = 5000,
):
super().__init__(base_url=base_url, timeout_sec=timeout_sec)
self.api_key = (api_key or "").strip()
self.secret_key = (secret_key or "").strip()
self.category = (category or "linear").strip().lower()
if self.category not in ("linear", "spot"):
self.category = "linear"
try:
self.recv_window_ms = int(recv_window_ms or 5000)
except Exception:
self.recv_window_ms = 5000
if self.recv_window_ms <= 0:
self.recv_window_ms = 5000
if not self.api_key or not self.secret_key:
raise LiveTradingError("Missing Bybit api_key/secret_key")
# Best-effort cache for linear instrument metadata (qty step, min qty, etc.)
# Key: f"{category}:{symbol}" -> (fetched_at_ts, info_dict)
self._inst_cache: Dict[str, Tuple[float, Dict[str, Any]]] = {}
self._inst_cache_ttl_sec = 300.0
@staticmethod
def _to_dec(x: Any) -> Decimal:
try:
return Decimal(str(x))
except Exception:
return Decimal("0")
@staticmethod
def _dec_str(d: Decimal) -> str:
try:
return format(d, "f")
except Exception:
return str(d)
@staticmethod
def _floor_to_step(value: Decimal, step: Decimal) -> Decimal:
if step is None:
return value
if value <= 0:
return Decimal("0")
try:
st = Decimal(step)
except Exception:
st = Decimal("0")
if st <= 0:
return value
try:
n = (value / st).to_integral_value(rounding=ROUND_DOWN)
return n * st
except Exception:
return Decimal("0")
def _sign(self, prehash: str) -> str:
return hmac.new(self.secret_key.encode("utf-8"), prehash.encode("utf-8"), hashlib.sha256).hexdigest()
def _headers(self, ts_ms: str, sign: str) -> Dict[str, str]:
return {
"X-BAPI-API-KEY": self.api_key,
"X-BAPI-SIGN": sign,
"X-BAPI-TIMESTAMP": ts_ms,
"X-BAPI-RECV-WINDOW": str(self.recv_window_ms),
"X-BAPI-SIGN-TYPE": "2",
"Content-Type": "application/json",
}
def _signed_request(
self,
method: str,
path: str,
*,
params: Optional[Dict[str, Any]] = None,
json_body: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
m = str(method or "GET").upper()
ts_ms = str(int(time.time() * 1000))
body_str = self._json_dumps(json_body) if json_body is not None else ""
qs = ""
if params:
norm = {str(k): "" if v is None else str(v) for k, v in dict(params).items()}
qs = urlencode(sorted(norm.items()), doseq=True)
payload = qs if m == "GET" else body_str
prehash = f"{ts_ms}{self.api_key}{self.recv_window_ms}{payload}"
sign = self._sign(prehash)
code, data, text = self._request(
m,
path,
params=params if (m == "GET" and params) else (params or None),
data=body_str if body_str else None,
headers=self._headers(ts_ms, sign),
)
if code >= 400:
raise LiveTradingError(f"Bybit HTTP {code}: {text[:500]}")
if isinstance(data, dict):
rc = data.get("retCode")
if rc not in (0, "0", None, ""):
raise LiveTradingError(f"Bybit error: {data}")
return data if isinstance(data, dict) else {"raw": data}
def _public_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
code, data, text = self._request(method, path, params=params, headers=None, json_body=None, data=None)
if code >= 400:
raise LiveTradingError(f"Bybit HTTP {code}: {text[:500]}")
if isinstance(data, dict):
rc = data.get("retCode")
if rc not in (0, "0", None, ""):
raise LiveTradingError(f"Bybit error: {data}")
return data if isinstance(data, dict) else {"raw": data}
def ping(self) -> bool:
try:
data = self._public_request("GET", "/v5/market/time")
return isinstance(data, dict) and (data.get("retCode") in (0, "0", None, ""))
except Exception:
return False
def get_wallet_balance(self, *, account_type: str = "UNIFIED") -> Dict[str, Any]:
return self._signed_request("GET", "/v5/account/wallet-balance", params={"accountType": str(account_type or "UNIFIED")})
def get_instrument_info(self, *, category: str, symbol: str) -> Dict[str, Any]:
cat = str(category or self.category or "linear").strip().lower()
sym = to_bybit_symbol(symbol)
if not sym:
return {}
key = f"{cat}:{sym}"
now = time.time()
cached = self._inst_cache.get(key)
if cached:
ts, obj = cached
if obj and (now - float(ts or 0.0)) <= float(self._inst_cache_ttl_sec or 300.0):
return obj
raw = self._public_request("GET", "/v5/market/instruments-info", params={"category": cat, "symbol": sym})
lst = (((raw.get("result") or {}).get("list")) if isinstance(raw, dict) else None) or []
first: Dict[str, Any] = lst[0] if isinstance(lst, list) and lst else {}
if isinstance(first, dict) and first:
self._inst_cache[key] = (now, first)
return first if isinstance(first, dict) else {}
def _normalize_qty(self, *, symbol: str, qty: float) -> Decimal:
q = self._to_dec(qty)
if q <= 0:
return Decimal("0")
sym = to_bybit_symbol(symbol)
try:
info = self.get_instrument_info(category=self.category, symbol=sym) or {}
except Exception:
info = {}
lot = (info.get("lotSizeFilter") if isinstance(info, dict) else None) or {}
step = self._to_dec((lot or {}).get("qtyStep") or "0")
mn = self._to_dec((lot or {}).get("minOrderQty") or "0")
if step > 0:
q = self._floor_to_step(q, step)
if mn > 0 and q < mn:
return Decimal("0")
return q
def place_market_order(
self,
*,
symbol: str,
side: str,
qty: float,
reduce_only: bool = False,
client_order_id: Optional[str] = None,
) -> LiveOrderResult:
sym = to_bybit_symbol(symbol)
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
q_req = float(qty or 0.0)
q_dec = self._normalize_qty(symbol=symbol, qty=q_req)
if float(q_dec or 0) <= 0:
raise LiveTradingError(f"Invalid qty (below step/min): requested={q_req}")
body: Dict[str, Any] = {
"category": self.category,
"symbol": sym,
"side": "Buy" if sd == "buy" else "Sell",
"orderType": "Market",
"qty": self._dec_str(q_dec),
"timeInForce": "GTC",
}
if reduce_only and self.category == "linear":
body["reduceOnly"] = True
if client_order_id:
body["orderLinkId"] = str(client_order_id)
raw = self._signed_request("POST", "/v5/order/create", json_body=body)
res = (raw.get("result") or {}) if isinstance(raw, dict) else {}
oid = str(res.get("orderId") or res.get("orderLinkId") or "")
return LiveOrderResult(exchange_id="bybit", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw)
def place_limit_order(
self,
*,
symbol: str,
side: str,
qty: float,
price: float,
reduce_only: bool = False,
client_order_id: Optional[str] = None,
) -> LiveOrderResult:
sym = to_bybit_symbol(symbol)
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
q_req = float(qty or 0.0)
px = float(price or 0.0)
if q_req <= 0 or px <= 0:
raise LiveTradingError("Invalid qty/price")
q_dec = self._normalize_qty(symbol=symbol, qty=q_req)
if float(q_dec or 0) <= 0:
raise LiveTradingError(f"Invalid qty (below step/min): requested={q_req}")
body: Dict[str, Any] = {
"category": self.category,
"symbol": sym,
"side": "Buy" if sd == "buy" else "Sell",
"orderType": "Limit",
"qty": self._dec_str(q_dec),
"price": str(px),
"timeInForce": "GTC",
}
if reduce_only and self.category == "linear":
body["reduceOnly"] = True
if client_order_id:
body["orderLinkId"] = str(client_order_id)
raw = self._signed_request("POST", "/v5/order/create", json_body=body)
res = (raw.get("result") or {}) if isinstance(raw, dict) else {}
oid = str(res.get("orderId") or res.get("orderLinkId") or "")
return LiveOrderResult(exchange_id="bybit", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw)
def cancel_order(self, *, symbol: str, order_id: str = "", client_order_id: str = "") -> Dict[str, Any]:
sym = to_bybit_symbol(symbol)
body: Dict[str, Any] = {"category": self.category, "symbol": sym}
if order_id:
body["orderId"] = str(order_id)
elif client_order_id:
body["orderLinkId"] = str(client_order_id)
else:
raise LiveTradingError("Bybit cancel_order requires order_id or client_order_id")
return self._signed_request("POST", "/v5/order/cancel", json_body=body)
def get_order(self, *, symbol: str, order_id: str = "", client_order_id: str = "") -> Dict[str, Any]:
sym = to_bybit_symbol(symbol)
params: Dict[str, Any] = {"category": self.category, "symbol": sym}
if order_id:
params["orderId"] = str(order_id)
elif client_order_id:
params["orderLinkId"] = str(client_order_id)
else:
raise LiveTradingError("Bybit get_order requires order_id or client_order_id")
raw = self._signed_request("GET", "/v5/order/realtime", params=params)
lst = (((raw.get("result") or {}).get("list")) if isinstance(raw, dict) else None) or []
first: Dict[str, Any] = lst[0] if isinstance(lst, list) and lst else {}
return first if isinstance(first, dict) else {}
def wait_for_fill(
self,
*,
symbol: str,
order_id: str = "",
client_order_id: str = "",
max_wait_sec: float = 3.0,
poll_interval_sec: float = 0.5,
) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Dict[str, Any] = {}
while True:
try:
last = self.get_order(symbol=symbol, order_id=str(order_id or ""), client_order_id=str(client_order_id or ""))
except Exception:
last = last or {}
status = str(last.get("orderStatus") or last.get("order_status") or "")
try:
filled = float(last.get("cumExecQty") or 0.0)
except Exception:
filled = 0.0
avg_price = 0.0
try:
avg_price = float(last.get("avgPrice") or 0.0)
except Exception:
avg_price = 0.0
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if status.lower() in ("filled", "cancelled", "canceled", "rejected"):
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
def get_positions(self) -> Dict[str, Any]:
if self.category != "linear":
raise LiveTradingError("Bybit positions are only supported for linear category in this client")
return self._signed_request("GET", "/v5/position/list", params={"category": "linear"})
def set_leverage(self, *, symbol: str, leverage: float) -> bool:
if self.category != "linear":
return False
sym = to_bybit_symbol(symbol)
try:
lv = int(float(leverage or 1.0))
except Exception:
lv = 1
if lv < 1:
lv = 1
# Bybit leverage caps vary per symbol; keep best-effort.
body = {"category": "linear", "symbol": sym, "buyLeverage": str(lv), "sellLeverage": str(lv)}
try:
resp = self._signed_request("POST", "/v5/position/set-leverage", json_body=body)
ok = isinstance(resp, dict) and (resp.get("retCode") in (0, "0", None, ""))
return bool(ok)
except Exception:
return False
@@ -0,0 +1,191 @@
"""
Coinbase Exchange (legacy, direct REST) client.
Auth headers:
- CB-ACCESS-KEY
- CB-ACCESS-SIGN = base64(hmac_sha256(base64_decode(secret), timestamp + method + request_path + body))
- CB-ACCESS-TIMESTAMP (seconds)
- CB-ACCESS-PASSPHRASE
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import time
from typing import Any, Dict, Optional
from app.services.live_trading.base import BaseRestClient, LiveOrderResult, LiveTradingError
from app.services.live_trading.symbols import to_coinbase_product_id
class CoinbaseExchangeClient(BaseRestClient):
def __init__(
self,
*,
api_key: str,
secret_key: str,
passphrase: str,
base_url: str = "https://api.exchange.coinbase.com",
timeout_sec: float = 15.0,
):
super().__init__(base_url=base_url, timeout_sec=timeout_sec)
self.api_key = (api_key or "").strip()
self.secret_key = (secret_key or "").strip()
self.passphrase = (passphrase or "").strip()
if not self.api_key or not self.secret_key or not self.passphrase:
raise LiveTradingError("Missing CoinbaseExchange api_key/secret_key/passphrase")
try:
self._secret_bytes = base64.b64decode(self.secret_key)
except Exception as e:
raise LiveTradingError(f"Invalid CoinbaseExchange secret_key (base64 decode failed): {e}")
def _sign(self, message: str) -> str:
mac = hmac.new(self._secret_bytes, message.encode("utf-8"), hashlib.sha256).digest()
return base64.b64encode(mac).decode("utf-8")
def _headers(self, ts: str, sign: str) -> Dict[str, str]:
return {
"CB-ACCESS-KEY": self.api_key,
"CB-ACCESS-SIGN": sign,
"CB-ACCESS-TIMESTAMP": ts,
"CB-ACCESS-PASSPHRASE": self.passphrase,
"Content-Type": "application/json",
}
def _signed_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Dict[str, Any]] = None) -> Any:
m = str(method or "GET").upper()
ts = str(int(time.time()))
body_str = self._json_dumps(json_body) if json_body is not None else ""
# Coinbase expects request_path to include query string for signature when GET params exist.
# We keep signature aligned with actual request params by relying on requests to encode params,
# but include them in the prehash in a stable order.
signed_path = path
if params:
# stable ordering
items = []
for k in sorted(params.keys()):
v = params.get(k)
if v is None:
continue
items.append(f"{k}={v}")
if items:
signed_path = f"{path}?{'&'.join(items)}"
prehash = f"{ts}{m}{signed_path}{body_str}"
sign = self._sign(prehash)
code, data, text = self._request(m, path, params=params, data=body_str if body_str else None, headers=self._headers(ts, sign))
if code >= 400:
raise LiveTradingError(f"CoinbaseExchange HTTP {code}: {text[:500]}")
return data
def _public_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None) -> Any:
code, data, text = self._request(method, path, params=params, headers=None, json_body=None, data=None)
if code >= 400:
raise LiveTradingError(f"CoinbaseExchange HTTP {code}: {text[:500]}")
return data
def ping(self) -> bool:
try:
_ = self._public_request("GET", "/time")
return True
except Exception:
return False
def get_accounts(self) -> Any:
return self._signed_request("GET", "/accounts")
def place_market_order(self, *, symbol: str, side: str, size: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
if qty <= 0:
raise LiveTradingError("Invalid size")
body: Dict[str, Any] = {
"product_id": to_coinbase_product_id(symbol),
"side": sd,
"type": "market",
"size": str(qty),
}
if client_order_id:
body["client_oid"] = str(client_order_id)
raw = self._signed_request("POST", "/orders", json_body=body)
oid = str(raw.get("id") or raw.get("order_id") or raw.get("client_oid") or "")
return LiveOrderResult(exchange_id="coinbaseexchange", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def place_limit_order(self, *, symbol: str, side: str, size: float, price: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
px = float(price or 0.0)
if qty <= 0 or px <= 0:
raise LiveTradingError("Invalid size/price")
body: Dict[str, Any] = {
"product_id": to_coinbase_product_id(symbol),
"side": sd,
"type": "limit",
"price": str(px),
"size": str(qty),
"time_in_force": "GTC",
}
if client_order_id:
body["client_oid"] = str(client_order_id)
raw = self._signed_request("POST", "/orders", json_body=body)
oid = str(raw.get("id") or raw.get("order_id") or raw.get("client_oid") or "")
return LiveOrderResult(exchange_id="coinbaseexchange", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def cancel_order(self, *, order_id: str = "", client_order_id: str = "") -> Any:
if order_id:
return self._signed_request("DELETE", f"/orders/{str(order_id)}")
if client_order_id:
return self._signed_request("DELETE", f"/orders/client:{str(client_order_id)}")
raise LiveTradingError("CoinbaseExchange cancel_order requires order_id or client_order_id")
def get_order(self, *, order_id: str = "", client_order_id: str = "") -> Any:
if order_id:
return self._signed_request("GET", f"/orders/{str(order_id)}")
if client_order_id:
return self._signed_request("GET", f"/orders/client:{str(client_order_id)}")
raise LiveTradingError("CoinbaseExchange get_order requires order_id or client_order_id")
def wait_for_fill(
self,
*,
order_id: str = "",
client_order_id: str = "",
max_wait_sec: float = 10.0,
poll_interval_sec: float = 0.5,
) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Dict[str, Any] = {}
while True:
try:
resp = self.get_order(order_id=str(order_id or ""), client_order_id=str(client_order_id or ""))
last = resp if isinstance(resp, dict) else {"raw": resp}
except Exception:
last = last or {}
status = str(last.get("status") or "")
filled = 0.0
avg_price = 0.0
try:
filled = float(last.get("filled_size") or 0.0)
except Exception:
filled = 0.0
try:
executed_value = float(last.get("executed_value") or 0.0)
if filled > 0 and executed_value > 0:
avg_price = executed_value / filled
except Exception:
avg_price = 0.0
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if status.lower() in ("done", "rejected", "canceled", "cancelled"):
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
@@ -12,6 +12,14 @@ from app.services.live_trading.binance_spot import BinanceSpotClient
from app.services.live_trading.okx import OkxClient
from app.services.live_trading.bitget import BitgetMixClient
from app.services.live_trading.bitget_spot import BitgetSpotClient
from app.services.live_trading.bybit import BybitClient
from app.services.live_trading.coinbase_exchange import CoinbaseExchangeClient
from app.services.live_trading.kraken import KrakenClient
from app.services.live_trading.kraken_futures import KrakenFuturesClient
from app.services.live_trading.kucoin import KucoinSpotClient
from app.services.live_trading.kucoin import KucoinFuturesClient
from app.services.live_trading.gate import GateSpotClient, GateUsdtFuturesClient
from app.services.live_trading.bitfinex import BitfinexClient, BitfinexDerivativesClient
def _signal_to_sides(signal_type: str) -> Tuple[str, str, bool]:
@@ -108,6 +116,33 @@ def place_order_from_signal(
size=qty,
client_order_id=client_order_id,
)
if isinstance(client, BybitClient):
return client.place_market_order(
symbol=symbol,
side=side,
qty=qty,
reduce_only=reduce_only,
client_order_id=client_order_id,
)
if isinstance(client, CoinbaseExchangeClient):
return client.place_market_order(symbol=symbol, side=side, size=qty, client_order_id=client_order_id)
if isinstance(client, KrakenClient):
return client.place_market_order(symbol=symbol, side=side, size=qty, client_order_id=client_order_id)
if isinstance(client, KucoinSpotClient):
# KuCoin market BUY often requires quote funds; this simplified path does not convert.
return client.place_market_order(symbol=symbol, side=side, size=qty, client_order_id=client_order_id, quote_size=False)
if isinstance(client, KucoinFuturesClient):
return client.place_market_order(symbol=symbol, side=side, size=qty, reduce_only=reduce_only, client_order_id=client_order_id)
if isinstance(client, GateSpotClient):
return client.place_market_order(symbol=symbol, side=side, size=qty, client_order_id=client_order_id)
if isinstance(client, GateUsdtFuturesClient):
return client.place_market_order(symbol=symbol, side=side, size=qty, reduce_only=reduce_only, client_order_id=client_order_id)
if isinstance(client, BitfinexClient):
return client.place_market_order(symbol=symbol, side=side, size=qty, client_order_id=client_order_id)
if isinstance(client, BitfinexDerivativesClient):
return client.place_market_order(symbol=symbol, side=side, size=qty, client_order_id=client_order_id)
if isinstance(client, KrakenFuturesClient):
return client.place_market_order(symbol=symbol, side=side, size=qty, reduce_only=reduce_only, client_order_id=client_order_id)
raise LiveTradingError(f"Unsupported client type: {type(client)}")
@@ -12,6 +12,13 @@ from app.services.live_trading.binance_spot import BinanceSpotClient
from app.services.live_trading.okx import OkxClient
from app.services.live_trading.bitget import BitgetMixClient
from app.services.live_trading.bitget_spot import BitgetSpotClient
from app.services.live_trading.bybit import BybitClient
from app.services.live_trading.coinbase_exchange import CoinbaseExchangeClient
from app.services.live_trading.kraken import KrakenClient
from app.services.live_trading.kraken_futures import KrakenFuturesClient
from app.services.live_trading.kucoin import KucoinSpotClient, KucoinFuturesClient
from app.services.live_trading.gate import GateSpotClient, GateUsdtFuturesClient
from app.services.live_trading.bitfinex import BitfinexClient, BitfinexDerivativesClient
def _get(cfg: Dict[str, Any], *keys: str) -> str:
@@ -54,6 +61,46 @@ def create_client(exchange_config: Dict[str, Any], *, market_type: str = "swap")
return BitgetSpotClient(api_key=api_key, secret_key=secret_key, passphrase=passphrase, base_url=base_url, channel_api_code=channel_api_code)
return BitgetMixClient(api_key=api_key, secret_key=secret_key, passphrase=passphrase, base_url=base_url)
if exchange_id == "bybit":
base_url = _get(exchange_config, "base_url", "baseUrl") or "https://api.bybit.com"
category = "spot" if mt == "spot" else "linear"
recv_window_ms = int(exchange_config.get("recv_window_ms") or exchange_config.get("recvWindow") or 5000)
return BybitClient(api_key=api_key, secret_key=secret_key, base_url=base_url, category=category, recv_window_ms=recv_window_ms)
if exchange_id in ("coinbaseexchange", "coinbase_exchange"):
base_url = _get(exchange_config, "base_url", "baseUrl") or "https://api.exchange.coinbase.com"
if mt != "spot":
raise LiveTradingError("CoinbaseExchange only supports spot market_type in this project")
return CoinbaseExchangeClient(api_key=api_key, secret_key=secret_key, passphrase=passphrase, base_url=base_url)
if exchange_id == "kraken":
base_url = _get(exchange_config, "base_url", "baseUrl") or "https://api.kraken.com"
if mt == "spot":
return KrakenClient(api_key=api_key, secret_key=secret_key, base_url=base_url)
# Futures/perp
fut_url = _get(exchange_config, "futures_base_url", "futuresBaseUrl") or "https://futures.kraken.com"
return KrakenFuturesClient(api_key=api_key, secret_key=secret_key, base_url=fut_url)
if exchange_id == "kucoin":
base_url = _get(exchange_config, "base_url", "baseUrl") or "https://api.kucoin.com"
if mt == "spot":
return KucoinSpotClient(api_key=api_key, secret_key=secret_key, passphrase=passphrase, base_url=base_url)
fut_url = _get(exchange_config, "futures_base_url", "futuresBaseUrl") or "https://api-futures.kucoin.com"
return KucoinFuturesClient(api_key=api_key, secret_key=secret_key, passphrase=passphrase, base_url=fut_url)
if exchange_id == "gate":
base_url = _get(exchange_config, "base_url", "baseUrl") or "https://api.gateio.ws"
if mt == "spot":
return GateSpotClient(api_key=api_key, secret_key=secret_key, base_url=base_url)
# Default to USDT futures for swap
return GateUsdtFuturesClient(api_key=api_key, secret_key=secret_key, base_url=base_url)
if exchange_id == "bitfinex":
base_url = _get(exchange_config, "base_url", "baseUrl") or "https://api.bitfinex.com"
if mt == "spot":
return BitfinexClient(api_key=api_key, secret_key=secret_key, base_url=base_url)
return BitfinexDerivativesClient(api_key=api_key, secret_key=secret_key, base_url=base_url)
raise LiveTradingError(f"Unsupported exchange_id: {exchange_id}")
@@ -0,0 +1,342 @@
"""
Gate.io (direct REST) clients:
- Spot: /api/v4/spot/*
- Futures USDT: /api/v4/futures/usdt/*
Signing (apiv4):
SIGN = hex(hmac_sha512(secret, method + "\\n" + url + "\\n" + query + "\\n" + body + "\\n" + timestamp))
Headers:
- KEY: api key
- Timestamp: unix seconds
- SIGN: signature hex
"""
from __future__ import annotations
import hashlib
import hmac
import time
from decimal import Decimal, ROUND_DOWN
from typing import Any, Dict, Optional, Tuple
from urllib.parse import urlencode
from app.services.live_trading.base import BaseRestClient, LiveOrderResult, LiveTradingError
from app.services.live_trading.symbols import to_gate_currency_pair
class _GateBase(BaseRestClient):
def __init__(self, *, api_key: str, secret_key: str, base_url: str = "https://api.gateio.ws", timeout_sec: float = 15.0):
super().__init__(base_url=base_url, timeout_sec=timeout_sec)
self.api_key = (api_key or "").strip()
self.secret_key = (secret_key or "").strip()
if not self.api_key or not self.secret_key:
raise LiveTradingError("Missing Gate api_key/secret_key")
def _sign(self, *, method: str, url: str, query_string: str, body_str: str, ts: str) -> str:
msg = f"{method.upper()}\n{url}\n{query_string}\n{body_str}\n{ts}"
return hmac.new(self.secret_key.encode("utf-8"), msg.encode("utf-8"), hashlib.sha512).hexdigest()
def _headers(self, ts: str, sign: str) -> Dict[str, str]:
return {"KEY": self.api_key, "Timestamp": ts, "SIGN": sign, "Content-Type": "application/json"}
def _signed_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Dict[str, Any]] = None) -> Any:
m = str(method or "GET").upper()
ts = str(int(time.time()))
body_str = self._json_dumps(json_body) if json_body is not None else ""
qs = ""
if params:
norm = {str(k): "" if v is None else str(v) for k, v in dict(params).items()}
qs = urlencode(sorted(norm.items()), doseq=True)
sign = self._sign(method=m, url=path, query_string=qs, body_str=body_str, ts=ts)
code, data, text = self._request(m, path, params=params, data=body_str if body_str else None, headers=self._headers(ts, sign))
if code >= 400:
raise LiveTradingError(f"Gate HTTP {code}: {text[:500]}")
return data
def _public_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None) -> Any:
code, data, text = self._request(method, path, params=params, headers=None, json_body=None, data=None)
if code >= 400:
raise LiveTradingError(f"Gate HTTP {code}: {text[:500]}")
return data
class GateSpotClient(_GateBase):
def ping(self) -> bool:
try:
_ = self._public_request("GET", "/api/v4/spot/time")
return True
except Exception:
return False
def get_accounts(self) -> Any:
return self._signed_request("GET", "/api/v4/spot/accounts")
def place_limit_order(self, *, symbol: str, side: str, size: float, price: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
px = float(price or 0.0)
if qty <= 0 or px <= 0:
raise LiveTradingError("Invalid size/price")
body: Dict[str, Any] = {
"currency_pair": to_gate_currency_pair(symbol),
"side": sd,
"type": "limit",
"amount": str(qty),
"price": str(px),
"time_in_force": "gtc",
}
if client_order_id:
body["text"] = str(client_order_id)
raw = self._signed_request("POST", "/api/v4/spot/orders", json_body=body)
oid = str(raw.get("id") or "") if isinstance(raw, dict) else ""
return LiveOrderResult(exchange_id="gate", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def place_market_order(self, *, symbol: str, side: str, size: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
if qty <= 0:
raise LiveTradingError("Invalid size")
body: Dict[str, Any] = {
"currency_pair": to_gate_currency_pair(symbol),
"side": sd,
"type": "market",
"amount": str(qty),
}
if client_order_id:
body["text"] = str(client_order_id)
raw = self._signed_request("POST", "/api/v4/spot/orders", json_body=body)
oid = str(raw.get("id") or "") if isinstance(raw, dict) else ""
return LiveOrderResult(exchange_id="gate", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def cancel_order(self, *, order_id: str) -> Any:
if not order_id:
raise LiveTradingError("Gate spot cancel_order requires order_id")
return self._signed_request("DELETE", f"/api/v4/spot/orders/{str(order_id)}")
def get_order(self, *, order_id: str) -> Any:
if not order_id:
raise LiveTradingError("Gate spot get_order requires order_id")
return self._signed_request("GET", f"/api/v4/spot/orders/{str(order_id)}")
def wait_for_fill(self, *, order_id: str, max_wait_sec: float = 10.0, poll_interval_sec: float = 0.5) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Dict[str, Any] = {}
while True:
try:
resp = self.get_order(order_id=str(order_id))
last = resp if isinstance(resp, dict) else {"raw": resp}
except Exception:
last = last or {}
status = str(last.get("status") or "")
filled = 0.0
avg_price = 0.0
try:
filled = float(last.get("filled_amount") or 0.0)
except Exception:
filled = 0.0
try:
filled_total = float(last.get("filled_total") or 0.0)
if filled > 0 and filled_total > 0:
avg_price = filled_total / filled
except Exception:
avg_price = 0.0
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if status.lower() in ("closed", "cancelled", "canceled"):
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
class GateUsdtFuturesClient(_GateBase):
def __init__(self, *, api_key: str, secret_key: str, base_url: str = "https://api.gateio.ws", timeout_sec: float = 15.0):
super().__init__(api_key=api_key, secret_key=secret_key, base_url=base_url, timeout_sec=timeout_sec)
# Best-effort cache for contract metadata to convert base qty -> contracts.
self._contract_cache: Dict[str, Tuple[float, Dict[str, Any]]] = {}
self._contract_cache_ttl_sec = 300.0
@staticmethod
def _to_dec(x: Any) -> Decimal:
try:
return Decimal(str(x))
except Exception:
return Decimal("0")
@staticmethod
def _floor(value: Decimal) -> Decimal:
try:
return value.to_integral_value(rounding=ROUND_DOWN)
except Exception:
return Decimal("0")
def ping(self) -> bool:
try:
_ = self._public_request("GET", "/api/v4/futures/usdt/time")
return True
except Exception:
return False
def get_contract(self, *, contract: str) -> Dict[str, Any]:
c = str(contract or "").strip()
if not c:
return {}
now = time.time()
cached = self._contract_cache.get(c)
if cached:
ts, obj = cached
if obj and (now - float(ts or 0.0)) <= float(self._contract_cache_ttl_sec or 300.0):
return obj
raw = self._public_request("GET", f"/api/v4/futures/usdt/contracts/{c}")
obj = raw if isinstance(raw, dict) else {}
if obj:
self._contract_cache[c] = (now, obj)
return obj
def _base_to_contracts(self, *, contract: str, base_size: float) -> int:
req = self._to_dec(base_size)
if req <= 0:
return 0
meta: Dict[str, Any] = {}
try:
meta = self.get_contract(contract=contract) or {}
except Exception:
meta = {}
qm = self._to_dec(meta.get("quanto_multiplier") or meta.get("quantoMultiplier") or meta.get("contract_size") or meta.get("contractSize") or "0")
if qm <= 0:
# Fallback: 1 contract ~= 1 base unit (best-effort)
qm = Decimal("1")
contracts = req / qm
return int(self._floor(contracts))
def get_accounts(self) -> Any:
return self._signed_request("GET", "/api/v4/futures/usdt/accounts")
def get_positions(self) -> Any:
return self._signed_request("GET", "/api/v4/futures/usdt/positions")
def set_leverage(self, *, contract: str, leverage: float) -> bool:
c = str(contract or "").strip()
if not c:
return False
try:
lv = int(float(leverage or 1.0))
except Exception:
lv = 1
if lv < 1:
lv = 1
try:
_ = self._signed_request("POST", f"/api/v4/futures/usdt/positions/{c}/leverage", json_body={"leverage": str(lv)})
return True
except Exception:
return False
def place_market_order(
self,
*,
symbol: str,
side: str,
size: float,
reduce_only: bool = False,
client_order_id: Optional[str] = None,
) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
contract = to_gate_currency_pair(symbol)
csz = self._base_to_contracts(contract=contract, base_size=float(size or 0.0))
if csz <= 0:
raise LiveTradingError("Invalid size (converted contracts <= 0)")
signed_size = int(csz) if sd == "buy" else -int(csz)
body: Dict[str, Any] = {"contract": contract, "size": signed_size, "price": "0", "tif": "ioc"}
if reduce_only:
body["reduce_only"] = True
if client_order_id:
body["text"] = str(client_order_id)
raw = self._signed_request("POST", "/api/v4/futures/usdt/orders", json_body=body)
oid = str(raw.get("id") or "") if isinstance(raw, dict) else ""
return LiveOrderResult(exchange_id="gate", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def place_limit_order(
self,
*,
symbol: str,
side: str,
size: float,
price: float,
reduce_only: bool = False,
client_order_id: Optional[str] = None,
) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
contract = to_gate_currency_pair(symbol)
csz = self._base_to_contracts(contract=contract, base_size=float(size or 0.0))
if csz <= 0:
raise LiveTradingError("Invalid size (converted contracts <= 0)")
px = float(price or 0.0)
if px <= 0:
raise LiveTradingError("Invalid price")
signed_size = int(csz) if sd == "buy" else -int(csz)
body: Dict[str, Any] = {"contract": contract, "size": signed_size, "price": str(px), "tif": "gtc"}
if reduce_only:
body["reduce_only"] = True
if client_order_id:
body["text"] = str(client_order_id)
raw = self._signed_request("POST", "/api/v4/futures/usdt/orders", json_body=body)
oid = str(raw.get("id") or "") if isinstance(raw, dict) else ""
return LiveOrderResult(exchange_id="gate", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def cancel_order(self, *, order_id: str) -> Any:
if not order_id:
raise LiveTradingError("Gate futures cancel_order requires order_id")
return self._signed_request("DELETE", f"/api/v4/futures/usdt/orders/{str(order_id)}")
def get_order(self, *, order_id: str) -> Any:
if not order_id:
raise LiveTradingError("Gate futures get_order requires order_id")
return self._signed_request("GET", f"/api/v4/futures/usdt/orders/{str(order_id)}")
def wait_for_fill(self, *, order_id: str, contract: str, max_wait_sec: float = 3.0, poll_interval_sec: float = 0.5) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Dict[str, Any] = {}
qm = Decimal("1")
try:
meta = self.get_contract(contract=str(contract)) or {}
qm = self._to_dec(meta.get("quanto_multiplier") or meta.get("contract_size") or "1")
if qm <= 0:
qm = Decimal("1")
except Exception:
qm = Decimal("1")
while True:
try:
resp = self.get_order(order_id=str(order_id))
last = resp if isinstance(resp, dict) else {"raw": resp}
except Exception:
last = last or {}
status = str(last.get("status") or "")
filled = 0.0
avg_price = 0.0
try:
# Gate futures often returns "filled_size" in contracts.
filled_ct = abs(float(last.get("filled_size") or last.get("filledSize") or 0.0))
filled = float(Decimal(str(filled_ct)) * qm)
except Exception:
filled = 0.0
try:
avg_price = float(last.get("fill_price") or last.get("fillPrice") or last.get("price") or 0.0)
except Exception:
avg_price = 0.0
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if str(status).lower() in ("finished", "cancelled", "canceled"):
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
@@ -0,0 +1,175 @@
"""
Kraken (direct REST) client (spot).
Auth:
- API-Key: api key string
- API-Sign: base64(hmac_sha512(base64_decode(secret), uri_path + sha256(nonce + postdata)))
Notes:
- Kraken spot uses asset pairs like XBTUSDT; we do best-effort normalization.
- This client is spot-only in this project (no futures).
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import time
from typing import Any, Dict, Optional
from urllib.parse import urlencode
from app.services.live_trading.base import BaseRestClient, LiveOrderResult, LiveTradingError
from app.services.live_trading.symbols import to_kraken_pair
class KrakenClient(BaseRestClient):
def __init__(self, *, api_key: str, secret_key: str, base_url: str = "https://api.kraken.com", timeout_sec: float = 15.0):
super().__init__(base_url=base_url, timeout_sec=timeout_sec)
self.api_key = (api_key or "").strip()
self.secret_key = (secret_key or "").strip()
if not self.api_key or not self.secret_key:
raise LiveTradingError("Missing Kraken api_key/secret_key")
try:
self._secret_bytes = base64.b64decode(self.secret_key)
except Exception as e:
raise LiveTradingError(f"Invalid Kraken secret_key (base64 decode failed): {e}")
def ping(self) -> bool:
try:
code, data, _ = self._request("GET", "/0/public/Time")
return code == 200 and isinstance(data, dict) and (data.get("error") in ([], None, ""))
except Exception:
return False
def get_balance(self) -> Dict[str, Any]:
"""
Private balance endpoint (best-effort credential validation).
"""
return self._signed_request("POST", "/0/private/Balance", data={})
def _sign(self, *, urlpath: str, nonce: str, postdata: str) -> str:
sha = hashlib.sha256((nonce + postdata).encode("utf-8")).digest()
mac = hmac.new(self._secret_bytes, urlpath.encode("utf-8") + sha, hashlib.sha512).digest()
return base64.b64encode(mac).decode("utf-8")
def _signed_request(self, method: str, path: str, *, data: Dict[str, Any]) -> Dict[str, Any]:
m = str(method or "POST").upper()
if m != "POST":
raise LiveTradingError("Kraken private endpoints in this client use POST")
nonce = str(int(time.time() * 1000))
body = dict(data or {})
body["nonce"] = nonce
postdata = urlencode(body, doseq=True)
sign = self._sign(urlpath=path, nonce=nonce, postdata=postdata)
headers = {"API-Key": self.api_key, "API-Sign": sign, "Content-Type": "application/x-www-form-urlencoded"}
code, resp, text = self._request("POST", path, params=None, json_body=None, data=postdata, headers=headers)
if code >= 400:
raise LiveTradingError(f"Kraken HTTP {code}: {text[:500]}")
if isinstance(resp, dict):
errs = resp.get("error")
if isinstance(errs, list) and errs:
raise LiveTradingError(f"Kraken error: {errs}")
return resp if isinstance(resp, dict) else {"raw": resp}
def add_order(
self,
*,
pair: str,
side: str,
ordertype: str,
volume: float,
price: float = 0.0,
client_order_id: str = "",
) -> Dict[str, Any]:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
ot = (ordertype or "").strip().lower()
if ot not in ("market", "limit"):
raise LiveTradingError(f"Invalid ordertype: {ordertype}")
vol = float(volume or 0.0)
if vol <= 0:
raise LiveTradingError("Invalid volume")
body: Dict[str, Any] = {"pair": str(pair), "type": sd, "ordertype": ot, "volume": str(vol)}
if ot == "limit":
px = float(price or 0.0)
if px <= 0:
raise LiveTradingError("Invalid limit price")
body["price"] = str(px)
# Best-effort userref (integer). Only digits allowed. Keep short.
if client_order_id:
try:
body["userref"] = int("".join([c for c in str(client_order_id) if c.isdigit()])[:9] or "0")
except Exception:
pass
return self._signed_request("POST", "/0/private/AddOrder", data=body)
def place_market_order(self, *, symbol: str, side: str, size: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
pair = to_kraken_pair(symbol)
raw = self.add_order(pair=pair, side=side, ordertype="market", volume=float(size or 0.0), client_order_id=str(client_order_id or ""))
txid = ""
try:
tx = ((raw.get("result") or {}).get("txid")) if isinstance(raw, dict) else None
if isinstance(tx, list) and tx:
txid = str(tx[0])
except Exception:
txid = ""
return LiveOrderResult(exchange_id="kraken", exchange_order_id=txid, filled=0.0, avg_price=0.0, raw=raw)
def place_limit_order(self, *, symbol: str, side: str, size: float, price: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
pair = to_kraken_pair(symbol)
raw = self.add_order(pair=pair, side=side, ordertype="limit", volume=float(size or 0.0), price=float(price or 0.0), client_order_id=str(client_order_id or ""))
txid = ""
try:
tx = ((raw.get("result") or {}).get("txid")) if isinstance(raw, dict) else None
if isinstance(tx, list) and tx:
txid = str(tx[0])
except Exception:
txid = ""
return LiveOrderResult(exchange_id="kraken", exchange_order_id=txid, filled=0.0, avg_price=0.0, raw=raw)
def cancel_order(self, *, order_id: str) -> Dict[str, Any]:
if not order_id:
raise LiveTradingError("Kraken cancel_order requires order_id")
return self._signed_request("POST", "/0/private/CancelOrder", data={"txid": str(order_id)})
def get_order(self, *, order_id: str) -> Dict[str, Any]:
if not order_id:
raise LiveTradingError("Kraken get_order requires order_id")
resp = self._signed_request("POST", "/0/private/QueryOrders", data={"txid": str(order_id)})
res = (resp.get("result") or {}) if isinstance(resp, dict) else {}
od = (res.get(str(order_id)) if isinstance(res, dict) else None) or {}
return od if isinstance(od, dict) else {}
def wait_for_fill(self, *, order_id: str, max_wait_sec: float = 10.0, poll_interval_sec: float = 0.5) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Dict[str, Any] = {}
while True:
try:
last = self.get_order(order_id=str(order_id))
except Exception:
last = last or {}
status = str(last.get("status") or "")
filled = 0.0
avg_price = 0.0
try:
filled = float(last.get("vol_exec") or 0.0)
except Exception:
filled = 0.0
# Kraken provides "cost" in quote currency. avg = cost / filled.
try:
cost = float(last.get("cost") or 0.0)
if filled > 0 and cost > 0:
avg_price = cost / filled
except Exception:
avg_price = 0.0
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if status.lower() in ("closed", "canceled", "cancelled", "expired"):
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
@@ -0,0 +1,205 @@
"""
Kraken Futures (direct REST) client.
Kraken Futures (formerly CryptoFacilities) uses a different API than Kraken spot.
Base URL example: https://futures.kraken.com
API prefix: /derivatives/api/v3
Auth (best-effort):
- APIKey: <api key>
- Nonce: <milliseconds>
- Authent: base64(hmac_sha256(secret, nonce + postdata + endpoint_path))
IMPORTANT:
- Instruments are exchange-specific (e.g. PF_XBTUSD, PI_XBTUSD). This project will pass through
those symbols if you choose them in UI, or best-effort map BTC/USDT -> PF_XBTUSD.
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import time
from typing import Any, Dict, Optional
from urllib.parse import urlencode
from app.services.live_trading.base import BaseRestClient, LiveOrderResult, LiveTradingError
from app.services.live_trading.symbols import to_kraken_futures_symbol
class KrakenFuturesClient(BaseRestClient):
def __init__(self, *, api_key: str, secret_key: str, base_url: str = "https://futures.kraken.com", timeout_sec: float = 15.0):
super().__init__(base_url=base_url, timeout_sec=timeout_sec)
self.api_key = (api_key or "").strip()
self.secret_key = (secret_key or "").strip()
if not self.api_key or not self.secret_key:
raise LiveTradingError("Missing KrakenFutures api_key/secret_key")
def _b64_hmac_sha256(self, msg: str) -> str:
mac = hmac.new(self.secret_key.encode("utf-8"), msg.encode("utf-8"), hashlib.sha256).digest()
return base64.b64encode(mac).decode("utf-8")
def _headers(self, nonce: str, authent: str) -> Dict[str, str]:
return {"APIKey": self.api_key, "Nonce": nonce, "Authent": authent, "Content-Type": "application/x-www-form-urlencoded"}
def _signed_request(self, method: str, path: str, *, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
m = str(method or "POST").upper()
# Kraken Futures private endpoints often use POST.
nonce = str(int(time.time() * 1000))
body = dict(data or {})
postdata = urlencode(body, doseq=True) if body else ""
# Sign with endpoint path (not including domain)
prehash = f"{nonce}{postdata}{path}"
authent = self._b64_hmac_sha256(prehash)
code, resp, text = self._request(m, path, params=None, json_body=None, data=postdata if postdata else None, headers=self._headers(nonce, authent))
if code >= 400:
raise LiveTradingError(f"KrakenFutures HTTP {code}: {text[:500]}")
if isinstance(resp, dict):
# Futures API often uses "result":"success"/"error" or "errors"
if str(resp.get("result") or "").lower() == "error" or resp.get("errors"):
raise LiveTradingError(f"KrakenFutures error: {resp}")
return resp if isinstance(resp, dict) else {"raw": resp}
def _public_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
code, resp, text = self._request(method, path, params=params, headers=None, json_body=None, data=None)
if code >= 400:
raise LiveTradingError(f"KrakenFutures HTTP {code}: {text[:500]}")
return resp if isinstance(resp, dict) else {"raw": resp}
def ping(self) -> bool:
try:
_ = self._public_request("GET", "/derivatives/api/v3/tickers")
return True
except Exception:
return False
def get_accounts(self) -> Dict[str, Any]:
# Best-effort private endpoint (varies by account type)
return self._signed_request("GET", "/derivatives/api/v3/accounts")
def get_open_positions(self) -> Dict[str, Any]:
return self._signed_request("GET", "/derivatives/api/v3/openpositions")
def place_market_order(
self,
*,
symbol: str,
side: str,
size: float,
reduce_only: bool = False,
client_order_id: Optional[str] = None,
) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
if qty <= 0:
raise LiveTradingError("Invalid size")
instr = to_kraken_futures_symbol(symbol)
body: Dict[str, Any] = {
"orderType": "mkt",
"symbol": str(instr),
"side": sd,
# Kraken Futures uses "size" in contracts; we treat incoming size as "contracts" for now.
"size": str(qty),
}
if reduce_only:
body["reduceOnly"] = "true"
if client_order_id:
body["cliOrdId"] = str(client_order_id)[:32]
raw = self._signed_request("POST", "/derivatives/api/v3/sendorder", data=body)
oid = str((raw.get("sendStatus") or {}).get("order_id") or (raw.get("order_id") or "")) if isinstance(raw, dict) else ""
return LiveOrderResult(exchange_id="kraken", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw)
def place_limit_order(
self,
*,
symbol: str,
side: str,
size: float,
price: float,
reduce_only: bool = False,
post_only: bool = False,
client_order_id: Optional[str] = None,
) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
px = float(price or 0.0)
if qty <= 0 or px <= 0:
raise LiveTradingError("Invalid size/price")
instr = to_kraken_futures_symbol(symbol)
body: Dict[str, Any] = {
"orderType": "lmt",
"symbol": str(instr),
"side": sd,
"size": str(qty),
"limitPrice": str(px),
}
if reduce_only:
body["reduceOnly"] = "true"
if post_only:
body["postOnly"] = "true"
if client_order_id:
body["cliOrdId"] = str(client_order_id)[:32]
raw = self._signed_request("POST", "/derivatives/api/v3/sendorder", data=body)
oid = str((raw.get("sendStatus") or {}).get("order_id") or (raw.get("order_id") or "")) if isinstance(raw, dict) else ""
return LiveOrderResult(exchange_id="kraken", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw)
def cancel_order(self, *, order_id: str = "", client_order_id: str = "") -> Dict[str, Any]:
body: Dict[str, Any] = {}
if order_id:
body["order_id"] = str(order_id)
elif client_order_id:
body["cliOrdId"] = str(client_order_id)
else:
raise LiveTradingError("KrakenFutures cancel_order requires order_id or client_order_id")
return self._signed_request("POST", "/derivatives/api/v3/cancelorder", data=body)
def get_order(self, *, order_id: str = "", client_order_id: str = "") -> Dict[str, Any]:
params: Dict[str, Any] = {}
if order_id:
params["order_id"] = str(order_id)
elif client_order_id:
params["cliOrdId"] = str(client_order_id)
else:
raise LiveTradingError("KrakenFutures get_order requires order_id or client_order_id")
return self._signed_request("GET", "/derivatives/api/v3/order", data=params)
def wait_for_fill(
self,
*,
order_id: str = "",
client_order_id: str = "",
max_wait_sec: float = 3.0,
poll_interval_sec: float = 0.5,
) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Dict[str, Any] = {}
while True:
try:
last = self.get_order(order_id=str(order_id or ""), client_order_id=str(client_order_id or ""))
except Exception:
last = last or {}
status = str(last.get("status") or last.get("orderStatus") or "")
filled = 0.0
avg_price = 0.0
try:
filled = float(last.get("filledSize") or last.get("filled_size") or 0.0)
except Exception:
filled = 0.0
try:
avg_price = float(last.get("avgFillPrice") or last.get("avg_fill_price") or 0.0)
except Exception:
avg_price = 0.0
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if status.lower() in ("filled", "cancelled", "canceled", "rejected"):
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
@@ -0,0 +1,508 @@
"""
KuCoin (direct REST) client (spot).
Signing (v2):
- KC-API-SIGN = base64(hmac_sha256(secret, timestamp + method + requestPathWithQuery + body))
- KC-API-PASSPHRASE = base64(hmac_sha256(secret, passphrase))
- KC-API-KEY-VERSION: 2
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import time
from decimal import Decimal, ROUND_DOWN
from typing import Any, Dict, Optional, Tuple
from urllib.parse import urlencode
from app.services.live_trading.base import BaseRestClient, LiveOrderResult, LiveTradingError
from app.services.live_trading.symbols import to_kucoin_symbol
class KucoinSpotClient(BaseRestClient):
def __init__(
self,
*,
api_key: str,
secret_key: str,
passphrase: str,
base_url: str = "https://api.kucoin.com",
timeout_sec: float = 15.0,
):
super().__init__(base_url=base_url, timeout_sec=timeout_sec)
self.api_key = (api_key or "").strip()
self.secret_key = (secret_key or "").strip()
self.passphrase = (passphrase or "").strip()
if not self.api_key or not self.secret_key or not self.passphrase:
raise LiveTradingError("Missing KuCoin api_key/secret_key/passphrase")
def _b64_hmac_sha256(self, key: str, msg: str) -> str:
mac = hmac.new(key.encode("utf-8"), msg.encode("utf-8"), hashlib.sha256).digest()
return base64.b64encode(mac).decode("utf-8")
def _headers(self, ts_ms: str, sign: str) -> Dict[str, str]:
# passphrase must be signed (v2)
p = self._b64_hmac_sha256(self.secret_key, self.passphrase)
return {
"KC-API-KEY": self.api_key,
"KC-API-SIGN": sign,
"KC-API-TIMESTAMP": ts_ms,
"KC-API-PASSPHRASE": p,
"KC-API-KEY-VERSION": "2",
"Content-Type": "application/json",
}
def _signed_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Dict[str, Any]] = None) -> Any:
m = str(method or "GET").upper()
ts_ms = str(int(time.time() * 1000))
body_str = self._json_dumps(json_body) if json_body is not None else ""
qs = ""
if params:
norm = {str(k): "" if v is None else str(v) for k, v in dict(params).items()}
qs = urlencode(sorted(norm.items()), doseq=True)
signed_path = f"{path}?{qs}" if qs else path
prehash = f"{ts_ms}{m}{signed_path}{body_str}"
sign = self._b64_hmac_sha256(self.secret_key, prehash)
code, data, text = self._request(m, path, params=params, data=body_str if body_str else None, headers=self._headers(ts_ms, sign))
if code >= 400:
raise LiveTradingError(f"KuCoin HTTP {code}: {text[:500]}")
return data
def _public_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None) -> Any:
code, data, text = self._request(method, path, params=params, headers=None, json_body=None, data=None)
if code >= 400:
raise LiveTradingError(f"KuCoin HTTP {code}: {text[:500]}")
return data
def ping(self) -> bool:
try:
d = self._public_request("GET", "/api/v1/timestamp")
return isinstance(d, dict) and str(d.get("code") or "") in ("200000", "0", "")
except Exception:
return False
def get_accounts(self) -> Any:
return self._signed_request("GET", "/api/v1/accounts")
def place_limit_order(self, *, symbol: str, side: str, size: float, price: float, client_order_id: Optional[str] = None) -> LiveOrderResult:
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
px = float(price or 0.0)
if qty <= 0 or px <= 0:
raise LiveTradingError("Invalid size/price")
body: Dict[str, Any] = {
"clientOid": str(client_order_id or str(int(time.time() * 1000))),
"side": sd,
"symbol": to_kucoin_symbol(symbol),
"type": "limit",
"price": str(px),
"size": str(qty),
"timeInForce": "GTC",
}
raw = self._signed_request("POST", "/api/v1/orders", json_body=body)
oid = ""
if isinstance(raw, dict):
d = raw.get("data")
if isinstance(d, dict):
oid = str(d.get("orderId") or "")
elif isinstance(d, str):
oid = str(d)
return LiveOrderResult(exchange_id="kucoin", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def place_market_order(
self,
*,
symbol: str,
side: str,
size: float,
client_order_id: Optional[str] = None,
quote_size: bool = False,
) -> LiveOrderResult:
"""
KuCoin market order:
- sell: use size (base quantity)
- buy: typically use funds (quote quantity). Set quote_size=True to treat `size` as funds.
"""
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
qty = float(size or 0.0)
if qty <= 0:
raise LiveTradingError("Invalid size")
body: Dict[str, Any] = {
"clientOid": str(client_order_id or str(int(time.time() * 1000))),
"side": sd,
"symbol": to_kucoin_symbol(symbol),
"type": "market",
}
if sd == "buy" and quote_size:
body["funds"] = str(qty)
else:
body["size"] = str(qty)
raw = self._signed_request("POST", "/api/v1/orders", json_body=body)
oid = ""
if isinstance(raw, dict):
d = raw.get("data")
if isinstance(d, dict):
oid = str(d.get("orderId") or "")
elif isinstance(d, str):
oid = str(d)
return LiveOrderResult(exchange_id="kucoin", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def cancel_order(self, *, order_id: str = "", client_order_id: str = "") -> Any:
if order_id:
return self._signed_request("DELETE", f"/api/v1/orders/{str(order_id)}")
if client_order_id:
return self._signed_request("DELETE", f"/api/v1/order/client-order/{str(client_order_id)}")
raise LiveTradingError("KuCoin cancel_order requires order_id or client_order_id")
def get_order(self, *, order_id: str = "", client_order_id: str = "") -> Any:
if order_id:
return self._signed_request("GET", f"/api/v1/orders/{str(order_id)}")
if client_order_id:
return self._signed_request("GET", f"/api/v1/order/client-order/{str(client_order_id)}")
raise LiveTradingError("KuCoin get_order requires order_id or client_order_id")
def get_fills(self, *, order_id: str) -> Any:
return self._signed_request("GET", "/api/v1/fills", params={"orderId": str(order_id)})
def wait_for_fill(self, *, order_id: str, max_wait_sec: float = 10.0, poll_interval_sec: float = 0.5) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Dict[str, Any] = {}
while True:
try:
resp = self.get_order(order_id=str(order_id))
last = resp if isinstance(resp, dict) else {"raw": resp}
except Exception:
last = last or {}
data = last.get("data") if isinstance(last, dict) else None
od = data if isinstance(data, dict) else {}
status = str(od.get("isActive") if od else "")
filled = 0.0
avg_price = 0.0
fee = 0.0
fee_ccy = ""
try:
filled = float(od.get("dealSize") or 0.0)
except Exception:
filled = 0.0
try:
funds = float(od.get("dealFunds") or 0.0)
if filled > 0 and funds > 0:
avg_price = funds / filled
except Exception:
avg_price = 0.0
try:
fee = abs(float(od.get("fee") or 0.0))
except Exception:
fee = 0.0
fee_ccy = str(od.get("feeCurrency") or "").strip()
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "fee": fee, "fee_ccy": fee_ccy, "status": status, "order": last}
# If order is inactive, consider it terminal
try:
is_active = bool(od.get("isActive"))
except Exception:
is_active = False
if not is_active:
return {"filled": filled, "avg_price": avg_price, "fee": fee, "fee_ccy": fee_ccy, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "fee": fee, "fee_ccy": fee_ccy, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
class KucoinFuturesClient(BaseRestClient):
"""
KuCoin Futures (USDT perpetual) direct REST client.
Notes:
- Base URL typically: https://api-futures.kucoin.com
- Auth headers/signing are the same KC-API-* style as spot (v2 passphrase signing),
but endpoints and symbol formats differ.
- Futures order size is typically in contracts; we convert from "base qty" best-effort.
"""
def __init__(
self,
*,
api_key: str,
secret_key: str,
passphrase: str,
base_url: str = "https://api-futures.kucoin.com",
timeout_sec: float = 15.0,
):
super().__init__(base_url=base_url, timeout_sec=timeout_sec)
self.api_key = (api_key or "").strip()
self.secret_key = (secret_key or "").strip()
self.passphrase = (passphrase or "").strip()
if not self.api_key or not self.secret_key or not self.passphrase:
raise LiveTradingError("Missing KuCoin Futures api_key/secret_key/passphrase")
# Best-effort contract cache: symbol -> (ts, contract_dict)
self._contract_cache: Dict[str, Tuple[float, Dict[str, Any]]] = {}
self._contract_cache_ttl_sec = 300.0
def _b64_hmac_sha256(self, key: str, msg: str) -> str:
mac = hmac.new(key.encode("utf-8"), msg.encode("utf-8"), hashlib.sha256).digest()
return base64.b64encode(mac).decode("utf-8")
def _headers(self, ts_ms: str, sign: str) -> Dict[str, str]:
p = self._b64_hmac_sha256(self.secret_key, self.passphrase)
return {
"KC-API-KEY": self.api_key,
"KC-API-SIGN": sign,
"KC-API-TIMESTAMP": ts_ms,
"KC-API-PASSPHRASE": p,
"KC-API-KEY-VERSION": "2",
"Content-Type": "application/json",
}
def _signed_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Dict[str, Any]] = None) -> Any:
m = str(method or "GET").upper()
ts_ms = str(int(time.time() * 1000))
body_str = self._json_dumps(json_body) if json_body is not None else ""
qs = ""
if params:
norm = {str(k): "" if v is None else str(v) for k, v in dict(params).items()}
qs = urlencode(sorted(norm.items()), doseq=True)
signed_path = f"{path}?{qs}" if qs else path
prehash = f"{ts_ms}{m}{signed_path}{body_str}"
sign = self._b64_hmac_sha256(self.secret_key, prehash)
code, data, text = self._request(m, path, params=params, data=body_str if body_str else None, headers=self._headers(ts_ms, sign))
if code >= 400:
raise LiveTradingError(f"KuCoinFutures HTTP {code}: {text[:500]}")
return data
def _public_request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None) -> Any:
code, data, text = self._request(method, path, params=params, headers=None, json_body=None, data=None)
if code >= 400:
raise LiveTradingError(f"KuCoinFutures HTTP {code}: {text[:500]}")
return data
def ping(self) -> bool:
try:
d = self._public_request("GET", "/api/v1/timestamp")
return isinstance(d, dict) and str(d.get("code") or "") in ("200000", "0", "")
except Exception:
return False
def get_contract(self, *, symbol: str) -> Dict[str, Any]:
from app.services.live_trading.symbols import to_kucoin_futures_symbol
sym = to_kucoin_futures_symbol(symbol)
if not sym:
return {}
now = time.time()
cached = self._contract_cache.get(sym)
if cached:
ts, obj = cached
if obj and (now - float(ts or 0.0)) <= float(self._contract_cache_ttl_sec or 300.0):
return obj
# KuCoin futures active contracts list
raw = self._public_request("GET", "/api/v1/contracts/active")
data = (raw.get("data") if isinstance(raw, dict) else None) or []
found: Dict[str, Any] = {}
if isinstance(data, list):
for it in data:
if not isinstance(it, dict):
continue
if str(it.get("symbol") or "").upper() == sym.upper():
found = it
break
if found:
self._contract_cache[sym] = (now, found)
return found
def _base_to_contracts(self, *, symbol: str, base_size: float) -> int:
"""
Convert base-asset qty -> contracts best-effort using multiplier.
"""
from app.services.live_trading.symbols import to_kucoin_futures_symbol
req = Decimal(str(base_size or 0.0))
if req <= 0:
return 0
sym = to_kucoin_futures_symbol(symbol)
meta: Dict[str, Any] = {}
try:
meta = self.get_contract(symbol=sym) or {}
except Exception:
meta = {}
# multiplier is base per contract for many KuCoin perps (best-effort)
mult = Decimal(str(meta.get("multiplier") or meta.get("lotSize") or "0"))
if mult <= 0:
mult = Decimal("1")
ct = (req / mult).to_integral_value(rounding=ROUND_DOWN)
try:
return int(ct)
except Exception:
return 0
def get_accounts(self) -> Any:
# Futures account overview
return self._signed_request("GET", "/api/v1/account-overview", params={"currency": "USDT"})
def get_positions(self) -> Any:
return self._signed_request("GET", "/api/v1/positions")
def set_leverage(self, *, symbol: str, leverage: float) -> bool:
from app.services.live_trading.symbols import to_kucoin_futures_symbol
sym = to_kucoin_futures_symbol(symbol)
try:
lv = int(float(leverage or 1.0))
except Exception:
lv = 1
if lv < 1:
lv = 1
body = {"symbol": sym, "leverage": str(lv)}
try:
_ = self._signed_request("POST", "/api/v1/position/leverage", json_body=body)
return True
except Exception:
return False
def place_market_order(
self,
*,
symbol: str,
side: str,
size: float,
reduce_only: bool = False,
client_order_id: Optional[str] = None,
) -> LiveOrderResult:
from app.services.live_trading.symbols import to_kucoin_futures_symbol
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
sym = to_kucoin_futures_symbol(symbol)
qty_ct = self._base_to_contracts(symbol=sym, base_size=float(size or 0.0))
if qty_ct <= 0:
raise LiveTradingError("Invalid size (converted contracts <= 0)")
body: Dict[str, Any] = {
"clientOid": str(client_order_id or str(int(time.time() * 1000))),
"side": sd,
"symbol": sym,
"type": "market",
"size": qty_ct,
}
if reduce_only:
body["reduceOnly"] = True
raw = self._signed_request("POST", "/api/v1/orders", json_body=body)
oid = ""
if isinstance(raw, dict):
d = raw.get("data")
if isinstance(d, dict):
oid = str(d.get("orderId") or "")
elif isinstance(d, str):
oid = str(d)
return LiveOrderResult(exchange_id="kucoin", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def place_limit_order(
self,
*,
symbol: str,
side: str,
size: float,
price: float,
reduce_only: bool = False,
post_only: bool = False,
client_order_id: Optional[str] = None,
) -> LiveOrderResult:
from app.services.live_trading.symbols import to_kucoin_futures_symbol
sd = (side or "").strip().lower()
if sd not in ("buy", "sell"):
raise LiveTradingError(f"Invalid side: {side}")
sym = to_kucoin_futures_symbol(symbol)
px = float(price or 0.0)
if px <= 0:
raise LiveTradingError("Invalid price")
qty_ct = self._base_to_contracts(symbol=sym, base_size=float(size or 0.0))
if qty_ct <= 0:
raise LiveTradingError("Invalid size (converted contracts <= 0)")
body: Dict[str, Any] = {
"clientOid": str(client_order_id or str(int(time.time() * 1000))),
"side": sd,
"symbol": sym,
"type": "limit",
"price": str(px),
"size": qty_ct,
}
if reduce_only:
body["reduceOnly"] = True
if post_only:
body["postOnly"] = True
raw = self._signed_request("POST", "/api/v1/orders", json_body=body)
oid = ""
if isinstance(raw, dict):
d = raw.get("data")
if isinstance(d, dict):
oid = str(d.get("orderId") or "")
elif isinstance(d, str):
oid = str(d)
return LiveOrderResult(exchange_id="kucoin", exchange_order_id=oid, filled=0.0, avg_price=0.0, raw=raw if isinstance(raw, dict) else {"raw": raw})
def cancel_order(self, *, order_id: str = "", client_order_id: str = "") -> Any:
if order_id:
return self._signed_request("DELETE", f"/api/v1/orders/{str(order_id)}")
if client_order_id:
return self._signed_request("DELETE", f"/api/v1/orders/client-order/{str(client_order_id)}")
raise LiveTradingError("KuCoinFutures cancel_order requires order_id or client_order_id")
def get_order(self, *, order_id: str = "", client_order_id: str = "") -> Any:
if order_id:
return self._signed_request("GET", f"/api/v1/orders/{str(order_id)}")
if client_order_id:
return self._signed_request("GET", f"/api/v1/orders/byClientOid", params={"clientOid": str(client_order_id)})
raise LiveTradingError("KuCoinFutures get_order requires order_id or client_order_id")
def wait_for_fill(self, *, order_id: str, max_wait_sec: float = 3.0, poll_interval_sec: float = 0.5) -> Dict[str, Any]:
end_ts = time.time() + float(max_wait_sec or 0.0)
last: Dict[str, Any] = {}
while True:
try:
resp = self.get_order(order_id=str(order_id))
last = resp if isinstance(resp, dict) else {"raw": resp}
except Exception:
last = last or {}
od = (last.get("data") if isinstance(last, dict) else None) or {}
status = str(od.get("status") or "")
filled = 0.0
avg_price = 0.0
try:
# dealSize is in contracts; convert back to base using multiplier best-effort.
deal_ct = float(od.get("dealSize") or 0.0)
except Exception:
deal_ct = 0.0
try:
deal_value = float(od.get("dealValue") or 0.0)
except Exception:
deal_value = 0.0
# Best-effort: infer avg price from dealValue / (deal contracts * multiplier)
mult = 1.0
try:
sym = str(od.get("symbol") or "")
meta = self.get_contract(symbol=sym) or {}
mult = float(meta.get("multiplier") or meta.get("lotSize") or 1.0)
if mult <= 0:
mult = 1.0
except Exception:
mult = 1.0
filled = abs(float(deal_ct or 0.0)) * float(mult)
if filled > 0 and deal_value > 0:
avg_price = float(deal_value) / float(filled)
if filled > 0 and avg_price > 0:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if status.lower() in ("done", "canceled", "cancelled", "filled"):
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
if time.time() >= end_ts:
return {"filled": filled, "avg_price": avg_price, "status": status, "order": last}
time.sleep(float(poll_interval_sec or 0.5))
@@ -10,7 +10,7 @@ We convert them into exchange-specific identifiers.
from __future__ import annotations
from typing import Tuple
from typing import Dict, Tuple
def _split_base_quote(symbol: str) -> Tuple[str, str]:
@@ -53,3 +53,135 @@ def to_bitget_um_symbol(symbol: str) -> str:
return f"{base}{quote}"
_KRAKEN_BASE_MAP: Dict[str, str] = {
# Common spot naming differences
"BTC": "XBT",
}
_BITFINEX_QUOTE_MAP: Dict[str, str] = {
# Bitfinex uses "UST" for Tether USDt
"USDT": "UST",
}
_KUCOIN_FUTURES_BASE_MAP: Dict[str, str] = {
# KuCoin futures uses XBT for BTC on many contracts
"BTC": "XBT",
}
def to_bybit_symbol(symbol: str) -> str:
"""
Bybit symbol format (v5): typically concatenated, e.g. BTCUSDT.
"""
return to_binance_futures_symbol(symbol)
def to_coinbase_product_id(symbol: str) -> str:
"""
Coinbase Exchange product id format: BASE-QUOTE, e.g. BTC-USDT.
"""
base, quote = _split_base_quote(symbol)
if not base or not quote:
return symbol
return f"{base}-{quote}"
def to_kraken_pair(symbol: str) -> str:
"""
Kraken spot pair format is exchange-specific (e.g. XBTUSDT).
We use a best-effort mapping for common assets; callers can override by passing
already-normalized Kraken pair strings.
"""
base, quote = _split_base_quote(symbol)
if not base or not quote:
return symbol
b = _KRAKEN_BASE_MAP.get(base, base)
return f"{b}{quote}"
def to_kucoin_symbol(symbol: str) -> str:
"""
KuCoin spot symbol format: BASE-QUOTE, e.g. BTC-USDT.
"""
base, quote = _split_base_quote(symbol)
if not base or not quote:
return symbol
return f"{base}-{quote}"
def to_kucoin_futures_symbol(symbol: str) -> str:
"""
KuCoin Futures (USDT perpetual) symbol is exchange-specific, common examples:
- XBTUSDTM, ETHUSDTM
We provide a best-effort mapping: BASEQUOTE + "M".
If caller already provides an exchange-native symbol (no '/'), we return as-is.
"""
s = (symbol or "").strip()
if "/" not in s:
return s
base, quote = _split_base_quote(symbol)
if not base or not quote:
return s
b = _KUCOIN_FUTURES_BASE_MAP.get(base, base)
return f"{b}{quote}M"
def to_kraken_futures_symbol(symbol: str) -> str:
"""
Kraken Futures instruments are exchange-specific (e.g. PF_XBTUSD, PI_XBTUSD).
This helper is best-effort:
- If caller already passes an exchange-native instrument (contains '_' or starts with PF_/PI_), return as-is.
- Otherwise, map BTC->XBT and assume USD quote for futures (most Kraken Futures perps are USD margined).
"""
s = (symbol or "").strip()
if not s:
return s
up = s.upper()
if "_" in up or up.startswith("PF_") or up.startswith("PI_"):
return s
base, quote = _split_base_quote(symbol)
if not base:
return s
b = _KRAKEN_BASE_MAP.get(base, base)
q = "USD"
# Keep USDT as USD best-effort (platform-dependent)
if quote and quote.upper() == "USD":
q = "USD"
return f"PF_{b}{q}"
def to_gate_currency_pair(symbol: str) -> str:
"""
Gate spot/futures currency_pair/contract format: BASE_QUOTE, e.g. BTC_USDT.
"""
base, quote = _split_base_quote(symbol)
if not base or not quote:
return symbol
return f"{base}_{quote}"
def to_bitfinex_spot_symbol(symbol: str) -> str:
"""
Bitfinex spot trading symbol format: tBASEQUOTE, e.g. tBTCUST.
"""
base, quote = _split_base_quote(symbol)
if not base or not quote:
s = str(symbol or "").strip()
return s if s.startswith("t") else f"t{s}"
q = _BITFINEX_QUOTE_MAP.get(quote, quote)
return f"t{base}{q}"
def to_bitfinex_perp_symbol(symbol: str) -> str:
"""
Bitfinex derivatives perpetual naming (best-effort): tBASEF0:QUOTEF0, e.g. tBTCF0:USTF0.
"""
base, quote = _split_base_quote(symbol)
if not base or not quote:
s = str(symbol or "").strip()
return s if s.startswith("t") else f"t{s}"
q = _BITFINEX_QUOTE_MAP.get(quote, quote)
return f"t{base}F0:{q}F0"
@@ -25,7 +25,17 @@ from app.services.live_trading.binance_spot import BinanceSpotClient
from app.services.live_trading.okx import OkxClient
from app.services.live_trading.bitget import BitgetMixClient
from app.services.live_trading.bitget_spot import BitgetSpotClient
from app.services.live_trading.bybit import BybitClient
from app.services.live_trading.coinbase_exchange import CoinbaseExchangeClient
from app.services.live_trading.kraken import KrakenClient
from app.services.live_trading.kraken_futures import KrakenFuturesClient
from app.services.live_trading.kucoin import KucoinSpotClient
from app.services.live_trading.kucoin import KucoinFuturesClient
from app.services.live_trading.gate import GateSpotClient, GateUsdtFuturesClient
from app.services.live_trading.bitfinex import BitfinexClient
from app.services.live_trading.bitfinex import BitfinexDerivativesClient
from app.services.live_trading.symbols import to_okx_swap_inst_id
from app.services.live_trading.symbols import to_gate_currency_pair
from app.utils.db import get_db_connection
from app.utils.logger import get_logger
@@ -225,6 +235,116 @@ class PendingOrderWorker:
side = "long" if hold_side == "long" else "short"
exch_size.setdefault(hb_sym, {"long": 0.0, "short": 0.0})[side] = abs(float(total))
elif isinstance(client, BybitClient) and market_type == "swap":
# Bybit linear positions
resp = client.get_positions()
lst = (((resp.get("result") or {}).get("list")) if isinstance(resp, dict) else None) or []
if isinstance(lst, list):
for p in lst:
if not isinstance(p, dict):
continue
sym = str(p.get("symbol") or "").strip().upper()
side0 = str(p.get("side") or "").strip().lower() # Buy/Sell
try:
sz = float(p.get("size") or 0.0)
except Exception:
sz = 0.0
if not sym or abs(sz) <= 0:
continue
hb_sym = sym
if hb_sym.endswith("USDT") and len(hb_sym) > 4 and "/" not in hb_sym:
hb_sym = f"{hb_sym[:-4]}/USDT"
side = "long" if side0 == "buy" else ("short" if side0 == "sell" else ("long" if sz > 0 else "short"))
exch_size.setdefault(hb_sym, {"long": 0.0, "short": 0.0})[side] = abs(float(sz))
elif isinstance(client, GateUsdtFuturesClient) and market_type == "swap":
resp = client.get_positions()
items = resp if isinstance(resp, list) else []
if isinstance(items, list):
for p in items:
if not isinstance(p, dict):
continue
contract = str(p.get("contract") or "").strip()
try:
sz_ct = float(p.get("size") or 0.0) # contracts, signed
except Exception:
sz_ct = 0.0
if not contract or abs(sz_ct) <= 0:
continue
hb_sym = contract.replace("_", "/")
side = "long" if sz_ct > 0 else "short"
# Convert contracts -> base using quanto_multiplier.
qty_base = abs(sz_ct)
try:
meta = client.get_contract(contract=contract) or {}
qm = float(meta.get("quanto_multiplier") or meta.get("contract_size") or 0.0)
if qm > 0:
qty_base = qty_base * qm
except Exception:
pass
exch_size.setdefault(hb_sym, {"long": 0.0, "short": 0.0})[side] = float(qty_base)
elif isinstance(client, KucoinFuturesClient) and market_type == "swap":
resp = client.get_positions()
data = (resp.get("data") if isinstance(resp, dict) else None) or []
if isinstance(data, list):
for p in data:
if not isinstance(p, dict):
continue
sym = str(p.get("symbol") or "").strip()
try:
qty_ct = float(p.get("currentQty") or p.get("quantity") or 0.0)
except Exception:
qty_ct = 0.0
if not sym or abs(qty_ct) <= 0:
continue
side = "long" if qty_ct > 0 else "short"
# Convert contracts -> base using multiplier.
qty_base = abs(qty_ct)
try:
meta = client.get_contract(symbol=sym) or {}
mult = float(meta.get("multiplier") or meta.get("lotSize") or 0.0)
if mult > 0:
qty_base = qty_base * mult
except Exception:
pass
exch_size.setdefault(sym, {"long": 0.0, "short": 0.0})[side] = float(qty_base)
elif isinstance(client, KrakenFuturesClient) and market_type == "swap":
resp = client.get_open_positions()
positions = (resp.get("openPositions") if isinstance(resp, dict) else None) or (resp.get("open_positions") if isinstance(resp, dict) else None) or []
if isinstance(positions, list):
for p in positions:
if not isinstance(p, dict):
continue
sym = str(p.get("symbol") or p.get("instrument") or "").strip()
try:
sz = float(p.get("size") or p.get("positionSize") or 0.0)
except Exception:
sz = 0.0
if not sym or abs(sz) <= 0:
continue
side = "long" if sz > 0 else "short"
exch_size.setdefault(sym, {"long": 0.0, "short": 0.0})[side] = abs(float(sz))
elif isinstance(client, BitfinexDerivativesClient) and market_type == "swap":
resp = client.get_positions()
items = resp if isinstance(resp, list) else []
if isinstance(items, list):
for p in items:
# Bitfinex positions are arrays; best-effort parse:
# [symbol, status, amount, base_price, ...]
try:
if isinstance(p, list) and len(p) >= 3:
sym = str(p[0] or "")
amt = float(p[2] or 0.0)
if not sym or abs(amt) <= 0:
continue
side = "long" if amt > 0 else "short"
exch_size.setdefault(sym, {"long": 0.0, "short": 0.0})[side] = abs(float(amt))
except Exception:
continue
else:
# Spot reconciliation is optional; skip for now (keeps self-check low-risk).
logger.debug(f"position sync: skip unsupported market/client: sid={sid}, cfg={safe_cfg}, market_type={market_type}, client={type(client)}")
@@ -642,12 +762,38 @@ class PendingOrderWorker:
if leverage <= 0:
leverage = 1.0
# Collect raw exchange interactions / intermediate states for debugging & persistence.
phases: Dict[str, Any] = {}
# Ensure ref price exists (used by maker pricing, fallbacks, and local DB snapshots).
if ref_price <= 0:
try:
if isinstance(client, BinanceFuturesClient):
ref_price = float(client.get_mark_price(symbol=str(symbol)) or 0.0)
except Exception:
pass
# Binance Futures leverage is per-symbol on the exchange side.
# If we do not set it, Binance may keep default 1x and the user will observe
# margin ~= notional (i.e., "margin = invested * leverage" when we sized using leverage).
if isinstance(client, BinanceFuturesClient) and market_type == "swap":
try:
client.set_leverage(symbol=str(symbol), leverage=float(leverage or 1.0))
phases["set_leverage"] = {"exchange": "binance", "symbol": str(symbol), "leverage": float(leverage or 1.0)}
except Exception as e:
# Safer default: do NOT place orders with an unintended leverage.
err = f"binance_set_leverage_failed:{e}"
logger.warning(f"live leverage set failed: pending_id={order_id}, strategy_id={strategy_id}, cfg={safe_cfg}, err={e}")
self._mark_failed(order_id=order_id, error=err)
_console_print(f"[worker] order rejected: strategy_id={strategy_id} pending_id={order_id} {err}")
_notify_live_best_effort(status="failed", error=err, amount_hint=amount, price_hint=ref_price)
return
# Accumulate fills across phases
total_base = 0.0
total_quote = 0.0
total_fee = 0.0
fee_ccy = ""
phases: Dict[str, Any] = {}
def _apply_fill(filled_qty: float, avg_px: float) -> None:
nonlocal total_base, total_quote
@@ -668,6 +814,23 @@ class PendingOrderWorker:
if (not fee_ccy) and ccy:
fee_ccy = str(ccy or "")
def _fetch_fee_best_effort(*, order_id0: str, client_order_id0: str) -> Tuple[float, str]:
"""
Some exchanges (notably Binance) do not expose commissions on order endpoints.
We fetch fills and sum commissions best-effort.
"""
oid = str(order_id0 or "").strip()
if not oid:
return 0.0, ""
try:
if isinstance(client, BinanceFuturesClient):
return client.get_fee_for_order(symbol=str(symbol), order_id=oid)
if isinstance(client, BinanceSpotClient):
return client.get_fee_for_order(symbol=str(symbol), order_id=oid)
except Exception:
return 0.0, ""
return 0.0, ""
def _current_avg() -> float:
return float(total_quote / total_base) if total_base > 0 else 0.0
@@ -770,6 +933,104 @@ class PendingOrderWorker:
price=limit_price,
client_order_id=limit_client_oid,
)
elif isinstance(client, BybitClient):
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
qty=remaining,
price=limit_price,
reduce_only=reduce_only,
client_order_id=limit_client_oid,
)
elif isinstance(client, CoinbaseExchangeClient):
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
client_order_id=limit_client_oid,
)
elif isinstance(client, KrakenClient):
# Kraken is spot-only and returns txid as order id.
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
client_order_id=limit_client_oid,
)
elif isinstance(client, KrakenFuturesClient):
# Kraken Futures expects instrument symbols; size is treated as contracts in this client.
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
reduce_only=reduce_only,
post_only=(order_mode in ("maker", "maker_then_market", "limit_first", "limit")),
client_order_id=limit_client_oid,
)
elif isinstance(client, KucoinSpotClient):
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
client_order_id=limit_client_oid,
)
elif isinstance(client, KucoinFuturesClient):
try:
if market_type == "swap":
client.set_leverage(symbol=str(symbol), leverage=leverage)
except Exception:
pass
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
reduce_only=reduce_only,
post_only=(order_mode in ("maker", "maker_then_market", "limit_first", "limit")),
client_order_id=limit_client_oid,
)
elif isinstance(client, GateSpotClient):
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
client_order_id=limit_client_oid,
)
elif isinstance(client, GateUsdtFuturesClient):
# Best-effort set leverage before futures order
try:
client.set_leverage(contract=to_gate_currency_pair(str(symbol)), leverage=leverage)
except Exception:
pass
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
reduce_only=reduce_only,
client_order_id=limit_client_oid,
)
elif isinstance(client, BitfinexClient):
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
client_order_id=limit_client_oid,
)
elif isinstance(client, BitfinexDerivativesClient):
res1 = client.place_limit_order(
symbol=str(symbol),
side=side,
size=remaining,
price=limit_price,
client_order_id=limit_client_oid,
)
else:
raise LiveTradingError(f"Unsupported client type: {type(client)}")
@@ -781,10 +1042,14 @@ class PendingOrderWorker:
q = client.wait_for_fill(symbol=str(symbol), order_id=limit_order_id, client_order_id=limit_client_oid, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
fee_v, fee_c = _fetch_fee_best_effort(order_id0=limit_order_id, client_order_id0=limit_client_oid)
_apply_fee(float(fee_v or 0.0), str(fee_c or ""))
elif isinstance(client, BinanceSpotClient):
q = client.wait_for_fill(symbol=str(symbol), order_id=limit_order_id, client_order_id=limit_client_oid, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
fee_v, fee_c = _fetch_fee_best_effort(order_id0=limit_order_id, client_order_id0=limit_client_oid)
_apply_fee(float(fee_v or 0.0), str(fee_c or ""))
elif isinstance(client, OkxClient):
q = client.wait_for_fill(symbol=str(symbol), ord_id=limit_order_id, cl_ord_id=limit_client_oid, market_type=market_type, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
@@ -800,6 +1065,48 @@ class PendingOrderWorker:
q = client.wait_for_fill(symbol=str(symbol), order_id=limit_order_id, client_order_id=limit_client_oid, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
_apply_fee(float(q.get("fee") or 0.0), str(q.get("fee_ccy") or ""))
elif isinstance(client, BybitClient):
q = client.wait_for_fill(symbol=str(symbol), order_id=limit_order_id, client_order_id=limit_client_oid, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
elif isinstance(client, CoinbaseExchangeClient):
q = client.wait_for_fill(order_id=limit_order_id, client_order_id=limit_client_oid, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
elif isinstance(client, KrakenClient):
q = client.wait_for_fill(order_id=limit_order_id, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
elif isinstance(client, KrakenFuturesClient):
q = client.wait_for_fill(order_id=limit_order_id, client_order_id=limit_client_oid, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
elif isinstance(client, KucoinSpotClient):
q = client.wait_for_fill(order_id=limit_order_id, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
_apply_fee(float(q.get("fee") or 0.0), str(q.get("fee_ccy") or ""))
elif isinstance(client, KucoinFuturesClient):
q = client.wait_for_fill(order_id=limit_order_id, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
elif isinstance(client, GateSpotClient):
q = client.wait_for_fill(order_id=limit_order_id, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
elif isinstance(client, GateUsdtFuturesClient):
q = client.wait_for_fill(order_id=limit_order_id, contract=to_gate_currency_pair(str(symbol)), max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
elif isinstance(client, BitfinexClient):
q = client.wait_for_fill(order_id=limit_order_id, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
elif isinstance(client, BitfinexDerivativesClient):
q = client.wait_for_fill(order_id=limit_order_id, max_wait_sec=maker_wait_sec)
phases["limit_query"] = q
_apply_fill(float(q.get("filled") or 0.0), float(q.get("avg_price") or 0.0))
remaining = max(0.0, float(amount or 0.0) - total_base)
@@ -842,6 +1149,26 @@ class PendingOrderWorker:
phases["limit_cancel"] = client.cancel_order(symbol=str(symbol), product_type=product_type, margin_coin=margin_coin, order_id=limit_order_id, client_oid=limit_client_oid)
elif isinstance(client, BitgetSpotClient):
phases["limit_cancel"] = client.cancel_order(symbol=str(symbol), client_order_id=limit_client_oid)
elif isinstance(client, BybitClient):
phases["limit_cancel"] = client.cancel_order(symbol=str(symbol), order_id=limit_order_id, client_order_id=limit_client_oid)
elif isinstance(client, CoinbaseExchangeClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id, client_order_id=limit_client_oid)
elif isinstance(client, KrakenClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id)
elif isinstance(client, KrakenFuturesClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id, client_order_id=limit_client_oid)
elif isinstance(client, KucoinSpotClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id, client_order_id=limit_client_oid)
elif isinstance(client, KucoinFuturesClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id, client_order_id=limit_client_oid)
elif isinstance(client, GateSpotClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id)
elif isinstance(client, GateUsdtFuturesClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id)
elif isinstance(client, BitfinexClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id, client_order_id=limit_client_oid)
elif isinstance(client, BitfinexDerivativesClient):
phases["limit_cancel"] = client.cancel_order(order_id=limit_order_id, client_order_id=limit_client_oid)
except Exception:
pass
except LiveTradingError as e:
@@ -930,6 +1257,95 @@ class PendingOrderWorker:
size=mkt_size,
client_order_id=market_client_oid,
)
elif isinstance(client, BybitClient):
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
qty=remaining,
reduce_only=reduce_only,
client_order_id=market_client_oid,
)
elif isinstance(client, CoinbaseExchangeClient):
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=remaining,
client_order_id=market_client_oid,
)
elif isinstance(client, KrakenClient):
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=remaining,
client_order_id=market_client_oid,
)
elif isinstance(client, KrakenFuturesClient):
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=remaining,
reduce_only=reduce_only,
client_order_id=market_client_oid,
)
elif isinstance(client, KucoinSpotClient):
# KuCoin market BUY expects quote funds; convert base->quote using ref_price.
if side == "buy" and ref_price > 0:
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=float(remaining) * float(ref_price),
quote_size=True,
client_order_id=market_client_oid,
)
else:
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=remaining,
quote_size=False,
client_order_id=market_client_oid,
)
elif isinstance(client, KucoinFuturesClient):
try:
if market_type == "swap":
client.set_leverage(symbol=str(symbol), leverage=leverage)
except Exception:
pass
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=remaining,
reduce_only=reduce_only,
client_order_id=market_client_oid,
)
elif isinstance(client, GateSpotClient):
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=remaining,
client_order_id=market_client_oid,
)
elif isinstance(client, GateUsdtFuturesClient):
try:
client.set_leverage(contract=to_gate_currency_pair(str(symbol)), leverage=leverage)
except Exception:
pass
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=remaining,
reduce_only=reduce_only,
client_order_id=market_client_oid,
)
elif isinstance(client, BitfinexClient):
res2 = client.place_market_order(
symbol=str(symbol),
side=side,
size=remaining,
client_order_id=market_client_oid,
)
elif isinstance(client, BitfinexDerivativesClient):
res2 = client.place_market_order(symbol=str(symbol), side=side, size=remaining, client_order_id=market_client_oid)
else:
raise LiveTradingError(f"Unsupported client type: {type(client)}")
@@ -941,10 +1357,14 @@ class PendingOrderWorker:
q2 = client.wait_for_fill(symbol=str(symbol), order_id=market_order_id, client_order_id=market_client_oid, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
fee_v, fee_c = _fetch_fee_best_effort(order_id0=market_order_id, client_order_id0=market_client_oid)
_apply_fee(float(fee_v or 0.0), str(fee_c or ""))
elif isinstance(client, BinanceSpotClient):
q2 = client.wait_for_fill(symbol=str(symbol), order_id=market_order_id, client_order_id=market_client_oid, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
fee_v, fee_c = _fetch_fee_best_effort(order_id0=market_order_id, client_order_id0=market_client_oid)
_apply_fee(float(fee_v or 0.0), str(fee_c or ""))
elif isinstance(client, OkxClient):
# OKX fills endpoint may lag shortly after execution; wait a bit longer to capture fee.
q2 = client.wait_for_fill(symbol=str(symbol), ord_id=market_order_id, cl_ord_id=market_client_oid, market_type=market_type, max_wait_sec=12.0)
@@ -961,6 +1381,48 @@ class PendingOrderWorker:
q2 = client.wait_for_fill(symbol=str(symbol), order_id=market_order_id, client_order_id=market_client_oid, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
_apply_fee(float(q2.get("fee") or 0.0), str(q2.get("fee_ccy") or ""))
elif isinstance(client, BybitClient):
q2 = client.wait_for_fill(symbol=str(symbol), order_id=market_order_id, client_order_id=market_client_oid, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
elif isinstance(client, CoinbaseExchangeClient):
q2 = client.wait_for_fill(order_id=market_order_id, client_order_id=market_client_oid, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
elif isinstance(client, KrakenClient):
q2 = client.wait_for_fill(order_id=market_order_id, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
elif isinstance(client, KrakenFuturesClient):
q2 = client.wait_for_fill(order_id=market_order_id, client_order_id=market_client_oid, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
elif isinstance(client, KucoinSpotClient):
q2 = client.wait_for_fill(order_id=market_order_id, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
_apply_fee(float(q2.get("fee") or 0.0), str(q2.get("fee_ccy") or ""))
elif isinstance(client, KucoinFuturesClient):
q2 = client.wait_for_fill(order_id=market_order_id, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
elif isinstance(client, GateSpotClient):
q2 = client.wait_for_fill(order_id=market_order_id, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
elif isinstance(client, GateUsdtFuturesClient):
q2 = client.wait_for_fill(order_id=market_order_id, contract=to_gate_currency_pair(str(symbol)), max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
elif isinstance(client, BitfinexClient):
q2 = client.wait_for_fill(order_id=market_order_id, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
elif isinstance(client, BitfinexDerivativesClient):
q2 = client.wait_for_fill(order_id=market_order_id, max_wait_sec=3.0)
phases["market_query"] = q2
_apply_fill(float(q2.get("filled") or 0.0), float(q2.get("avg_price") or 0.0))
except LiveTradingError as e:
logger.warning(f"live market phase failed: pending_id={order_id}, strategy_id={strategy_id}, cfg={safe_cfg}, err={e}")
phases["market_error"] = str(e)
@@ -1015,6 +1477,10 @@ class PendingOrderWorker:
# Record trade + update local position snapshot (best-effort).
try:
if filled > 0 and avg_price > 0:
logger.info(
f"live record begin: pending_id={order_id} strategy_id={strategy_id} symbol={symbol} "
f"signal={signal_type} filled={filled} avg_price={avg_price} fee={total_fee} fee_ccy={fee_ccy}"
)
profit, _pos = apply_fill_to_local_position(
strategy_id=strategy_id,
symbol=str(symbol),
@@ -1037,6 +1503,7 @@ class PendingOrderWorker:
commission_ccy=str(fee_ccy or "").strip().upper(),
profit=profit,
)
logger.info(f"live record done: pending_id={order_id} strategy_id={strategy_id} symbol={symbol} signal={signal_type}")
except Exception as e:
logger.warning(f"record_trade/update_position failed: pending_id={order_id}, err={e}")
@@ -18,6 +18,8 @@ notification_config = {
from __future__ import annotations
import html
import hmac
import hashlib
import json
import os
import smtplib
@@ -186,6 +188,13 @@ class SignalNotifier:
ok, err = self._notify_webhook(
url=url,
payload=payload,
headers_override=(targets.get("webhook_headers") or targets.get("webhookHeaders") or None),
token_override=(targets.get("webhook_token") or targets.get("webhookToken") or None),
signing_secret_override=(
targets.get("webhook_signing_secret")
or targets.get("webhookSigningSecret")
or None
),
)
elif c == "discord":
url = (targets.get("discord") or "").strip()
@@ -228,6 +237,11 @@ class SignalNotifier:
ok, err = False, str(e)
results[c] = {"ok": bool(ok), "error": (err or "")}
if not ok and c in ("webhook", "discord"):
# Keep logs high-signal and avoid leaking full URLs (webhook URLs contain secrets).
logger.info(
f"notify failed: channel={c} strategy_id={strategy_id} symbol={symbol} signal={signal_type} err={err}"
)
return results
@@ -462,16 +476,93 @@ class SignalNotifier:
logger.warning(f"browser notify persist failed: {e}")
return False, str(e)
def _notify_webhook(self, *, url: str, payload: Dict[str, Any]) -> Tuple[bool, str]:
def _notify_webhook(
self,
*,
url: str,
payload: Dict[str, Any],
headers_override: Any = None,
token_override: Any = None,
signing_secret_override: Any = None,
) -> Tuple[bool, str]:
"""
Generic webhook delivery.
Supports (best-effort):
- per-strategy headers: notification_config.targets.webhook_headers (dict or JSON string)
- per-strategy bearer token: notification_config.targets.webhook_token
- global bearer token: SIGNAL_WEBHOOK_TOKEN
- optional signing secret: notification_config.targets.webhook_signing_secret or env SIGNAL_WEBHOOK_SIGNING_SECRET
Adds headers:
- X-QD-Timestamp: unix seconds
- X-QD-Signature: hex(HMAC_SHA256("{ts}.{body}", secret))
- retry once on 429/5xx
"""
if not url:
return False, "missing_webhook_url"
headers = {"Content-Type": "application/json"}
if self.webhook_token:
headers["Authorization"] = f"Bearer {self.webhook_token}"
if not (str(url).startswith("http://") or str(url).startswith("https://")):
return False, "invalid_webhook_url"
headers: Dict[str, str] = {
"Content-Type": "application/json",
"User-Agent": "QuantDinger/1.0 (+https://www.quantdinger.com)",
}
# Per-strategy header overrides (optional)
wh = headers_override
if isinstance(wh, str) and wh.strip():
try:
obj = json.loads(wh)
wh = obj if isinstance(obj, dict) else None
except Exception:
wh = None
if isinstance(wh, dict):
for k, v in wh.items():
kk = str(k or "").strip()
if not kk:
continue
headers[kk] = str(v if v is not None else "")
# Auth (per-strategy token first, fallback to global token)
tok = str(token_override or "").strip()
if not tok:
tok = self.webhook_token
if tok and "Authorization" not in headers:
headers["Authorization"] = f"Bearer {tok}"
# Optional signing secret (per-strategy override, else env)
signing_secret = str(signing_secret_override or "").strip() or (os.getenv("SIGNAL_WEBHOOK_SIGNING_SECRET") or "").strip()
if signing_secret:
try:
ts = str(int(time.time()))
body = json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
sig_base = (ts + ".").encode("utf-8") + body
sig = hmac.new(signing_secret.encode("utf-8"), sig_base, hashlib.sha256).hexdigest()
headers["X-QD-Timestamp"] = ts
headers["X-QD-Signature"] = sig
# Send raw bytes so signature matches what we sign.
def _post_once(timeout: float) -> requests.Response:
return requests.post(url, data=body, headers=headers, timeout=timeout)
except Exception as e:
return False, f"webhook_signing_failed:{e}"
else:
def _post_once(timeout: float) -> requests.Response:
return requests.post(url, json=payload, headers=headers, timeout=timeout)
# Post with minimal retry on 429/5xx
try:
resp = requests.post(url, json=payload, headers=headers, timeout=self.timeout_sec)
resp = _post_once(self.timeout_sec)
if 200 <= resp.status_code < 300:
return True, ""
if resp.status_code in (429, 500, 502, 503, 504):
try:
time.sleep(1.0)
except Exception:
pass
resp2 = _post_once(self.timeout_sec)
if 200 <= resp2.status_code < 300:
return True, ""
return False, f"http_{resp2.status_code}:{(resp2.text or '')[:300]}"
return False, f"http_{resp.status_code}:{(resp.text or '')[:300]}"
except Exception as e:
return False, str(e)
@@ -479,6 +570,8 @@ class SignalNotifier:
def _notify_discord(self, *, url: str, payload: Dict[str, Any], fallback_text: str) -> Tuple[bool, str]:
if not url:
return False, "missing_discord_webhook_url"
if not (str(url).startswith("http://") or str(url).startswith("https://")):
return False, "invalid_discord_webhook_url"
strategy = (payload or {}).get("strategy") or {}
instrument = (payload or {}).get("instrument") or {}
@@ -508,15 +601,41 @@ class SignalNotifier:
embed["timestamp"] = str(payload.get("timestamp_iso") or "")
if trace.get("pending_order_id"):
embed["footer"] = {"text": f"pending_order_id={int(trace.get('pending_order_id'))}"}
headers = {
"Content-Type": "application/json",
"User-Agent": "QuantDinger/1.0 (+https://www.quantdinger.com)",
}
def _post(payload_json: Dict[str, Any]) -> requests.Response:
return requests.post(url, json=payload_json, headers=headers, timeout=self.timeout_sec)
try:
resp = requests.post(url, json={"content": "", "embeds": [embed]}, timeout=self.timeout_sec)
resp = _post({"content": "", "embeds": [embed]})
if 200 <= resp.status_code < 300:
return True, ""
# Fallback: try plain text.
# Rate limit: retry once if Discord asks us to.
if resp.status_code == 429:
try:
data = resp.json() if resp is not None else {}
retry_after = float((data or {}).get("retry_after") or 1.0)
time.sleep(min(max(retry_after, 0.5), 3.0))
except Exception:
try:
time.sleep(1.0)
except Exception:
pass
resp_retry = _post({"content": "", "embeds": [embed]})
if 200 <= resp_retry.status_code < 300:
return True, ""
resp = resp_retry
# Fallback: plain text (some servers reject embeds)
try:
resp2 = requests.post(url, json={"content": str(fallback_text or "")[:1900]}, timeout=self.timeout_sec)
resp2 = _post({"content": str(fallback_text or "")[:1900]})
if 200 <= resp2.status_code < 300:
return True, ""
# If fallback also fails, return the original error (more useful than fallback sometimes).
except Exception:
pass
return False, f"http_{resp.status_code}:{(resp.text or '')[:300]}"
+191 -2
View File
@@ -60,13 +60,174 @@ class StrategyService:
获取交易所交易对列表 (无需API Key)
"""
try:
import ccxt
exchange_id = exchange_config.get('exchange_id', '')
proxies = exchange_config.get('proxies')
if not exchange_id:
return {'success': False, 'message': '请选择交易所', 'symbols': []}
# For these exchanges, prefer direct REST (no ccxt), aligned with local live-trading design.
ex = str(exchange_id or "").strip().lower()
if ex in ("bybit", "coinbaseexchange", "coinbase_exchange", "kraken", "kucoin", "gate", "bitfinex"):
import requests
def _req_json(url: str) -> Any:
r = requests.get(url, timeout=15, proxies=proxies)
r.raise_for_status()
return r.json()
symbols: List[str] = []
market_type = str(exchange_config.get("market_type") or exchange_config.get("defaultType") or "spot").strip().lower()
if market_type in ("futures", "future", "perp", "perpetual"):
market_type = "swap"
if ex == "bybit":
base = str(exchange_config.get("base_url") or exchange_config.get("baseUrl") or "https://api.bybit.com").rstrip("/")
cat = "spot" if market_type == "spot" else "linear"
j = _req_json(f"{base}/v5/market/instruments-info?category={cat}")
lst = (((j.get("result") or {}).get("list")) if isinstance(j, dict) else None) or []
if isinstance(lst, list):
for it in lst:
if not isinstance(it, dict):
continue
sym = str(it.get("symbol") or "")
status = str(it.get("status") or "").lower()
if not sym or (status and status not in ("trading", "tradable", "online")):
continue
if sym.endswith("USDT") and len(sym) > 4:
symbols.append(f"{sym[:-4]}/USDT")
symbols = sorted(list(set(symbols)))
return {'success': True, 'message': f'获取成功,共 {len(symbols)} 个交易对', 'symbols': symbols}
if ex in ("coinbaseexchange", "coinbase_exchange"):
base = str(exchange_config.get("base_url") or exchange_config.get("baseUrl") or "https://api.exchange.coinbase.com").rstrip("/")
j = _req_json(f"{base}/products")
if isinstance(j, list):
for it in j:
if not isinstance(it, dict):
continue
if str(it.get("status") or "").lower() not in ("online", ""):
continue
base_ccy = str(it.get("base_currency") or "").upper()
quote_ccy = str(it.get("quote_currency") or "").upper()
if quote_ccy == "USDT" and base_ccy:
symbols.append(f"{base_ccy}/USDT")
symbols = sorted(list(set(symbols)))
return {'success': True, 'message': f'获取成功,共 {len(symbols)} 个交易对', 'symbols': symbols}
if ex == "kraken":
if market_type == "spot":
j = _req_json("https://api.kraken.com/0/public/AssetPairs")
res = (j.get("result") if isinstance(j, dict) else None) or {}
if isinstance(res, dict):
for _k, v in res.items():
if not isinstance(v, dict):
continue
wsname = str(v.get("wsname") or "")
if not wsname or "/" not in wsname:
continue
base_ccy, quote_ccy = wsname.split("/", 1)
if str(quote_ccy).upper() == "USDT":
symbols.append(f"{str(base_ccy).upper()}/USDT")
else:
base = str(exchange_config.get("futures_base_url") or exchange_config.get("futuresBaseUrl") or "https://futures.kraken.com").rstrip("/")
j = _req_json(f"{base}/derivatives/api/v3/instruments")
instruments = j.get("instruments") if isinstance(j, dict) else None
if isinstance(instruments, list):
for it in instruments:
if not isinstance(it, dict):
continue
sym = str(it.get("symbol") or "")
typ = str(it.get("type") or "").lower()
if sym and ("perpetual" in typ or typ.startswith("pf") or sym.startswith("PF_")):
symbols.append(sym)
symbols = sorted(list(set(symbols)))
return {'success': True, 'message': f'获取成功,共 {len(symbols)} 个交易对', 'symbols': symbols}
if ex == "kucoin":
if market_type == "spot":
base = str(exchange_config.get("base_url") or exchange_config.get("baseUrl") or "https://api.kucoin.com").rstrip("/")
j = _req_json(f"{base}/api/v1/symbols")
data = (j.get("data") if isinstance(j, dict) else None) or []
if isinstance(data, list):
for it in data:
if not isinstance(it, dict):
continue
if not bool(it.get("enableTrading", True)):
continue
if str(it.get("quoteCurrency") or "").upper() != "USDT":
continue
b = str(it.get("baseCurrency") or "").upper()
if b:
symbols.append(f"{b}/USDT")
else:
base = str(exchange_config.get("futures_base_url") or exchange_config.get("futuresBaseUrl") or "https://api-futures.kucoin.com").rstrip("/")
j = _req_json(f"{base}/api/v1/contracts/active")
data = (j.get("data") if isinstance(j, dict) else None) or []
if isinstance(data, list):
for it in data:
if not isinstance(it, dict):
continue
sym = str(it.get("symbol") or "")
if not sym or not sym.upper().endswith("USDTM"):
continue
base_ccy = sym[:-5].upper()
if base_ccy == "XBT":
base_ccy = "BTC"
if base_ccy:
symbols.append(f"{base_ccy}/USDT")
symbols = sorted(list(set(symbols)))
return {'success': True, 'message': f'获取成功,共 {len(symbols)} 个交易对', 'symbols': symbols}
if ex == "gate":
base = str(exchange_config.get("base_url") or exchange_config.get("baseUrl") or "https://api.gateio.ws").rstrip("/")
if market_type == "spot":
j = _req_json(f"{base}/api/v4/spot/currency_pairs")
if isinstance(j, list):
for it in j:
if not isinstance(it, dict):
continue
if str(it.get("trade_status") or "").lower() not in ("tradable", "trading", ""):
continue
base_ccy = str(it.get("base") or "").upper()
quote_ccy = str(it.get("quote") or "").upper()
if quote_ccy == "USDT" and base_ccy:
symbols.append(f"{base_ccy}/USDT")
else:
j = _req_json(f"{base}/api/v4/futures/usdt/contracts")
if isinstance(j, list):
for it in j:
if not isinstance(it, dict):
continue
name = str(it.get("name") or it.get("contract") or "")
if name and name.upper().endswith("_USDT"):
symbols.append(name.replace("_", "/"))
symbols = sorted(list(set(symbols)))
return {'success': True, 'message': f'获取成功,共 {len(symbols)} 个交易对', 'symbols': symbols}
if ex == "bitfinex":
j = _req_json("https://api-pub.bitfinex.com/v2/conf/pub:list:pair:exchange") if market_type == "spot" else _req_json(
"https://api-pub.bitfinex.com/v2/conf/pub:list:pair:futures"
)
pairs = []
if isinstance(j, list) and j and isinstance(j[0], list):
pairs = j[0]
for p in pairs:
s = str(p or "").upper()
if not s:
continue
if market_type != "spot":
symbols.append(s)
continue
# Focus USDT (Bitfinex uses UST)
if s.endswith("UST") and len(s) > 3:
symbols.append(f"{s[:-3]}/USDT")
elif s.endswith("USDT") and len(s) > 4:
symbols.append(f"{s[:-4]}/USDT")
symbols = sorted(list(set(symbols)))
return {'success': True, 'message': f'获取成功,共 {len(symbols)} 个交易对', 'symbols': symbols}
return {'success': True, 'message': '获取成功', 'symbols': symbols}
import ccxt
# 创建交易所实例 (public only)
exchange_class = getattr(ccxt, exchange_id, None)
@@ -112,6 +273,14 @@ class StrategyService:
from app.services.live_trading.binance_spot import BinanceSpotClient
from app.services.live_trading.okx import OkxClient
from app.services.live_trading.bitget import BitgetMixClient
from app.services.live_trading.bybit import BybitClient
from app.services.live_trading.coinbase_exchange import CoinbaseExchangeClient
from app.services.live_trading.kraken import KrakenClient
from app.services.live_trading.kraken_futures import KrakenFuturesClient
from app.services.live_trading.kucoin import KucoinSpotClient
from app.services.live_trading.kucoin import KucoinFuturesClient
from app.services.live_trading.gate import GateSpotClient, GateUsdtFuturesClient
from app.services.live_trading.bitfinex import BitfinexClient, BitfinexDerivativesClient
resolved = resolve_exchange_config(exchange_config or {})
safe_cfg = safe_exchange_config_for_log(resolved)
@@ -160,6 +329,26 @@ class StrategyService:
elif isinstance(client, BitgetMixClient):
product_type = str(resolved.get("product_type") or resolved.get("productType") or "USDT-FUTURES")
priv_data = client.get_accounts(product_type=product_type)
elif isinstance(client, BybitClient):
priv_data = client.get_wallet_balance()
elif isinstance(client, CoinbaseExchangeClient):
priv_data = client.get_accounts()
elif isinstance(client, KrakenClient):
priv_data = client.get_balance()
elif isinstance(client, KrakenFuturesClient):
priv_data = client.get_accounts()
elif isinstance(client, KucoinSpotClient):
priv_data = client.get_accounts()
elif isinstance(client, KucoinFuturesClient):
priv_data = client.get_accounts()
elif isinstance(client, GateSpotClient):
priv_data = client.get_accounts()
elif isinstance(client, GateUsdtFuturesClient):
priv_data = client.get_accounts()
elif isinstance(client, BitfinexClient):
priv_data = client.get_wallets()
elif isinstance(client, BitfinexDerivativesClient):
priv_data = client.get_wallets()
except Exception as e:
msg = str(e)
# Add actionable hints for the most common Binance auth error.
@@ -1889,7 +1889,9 @@ class TradingExecutor:
if market_type == 'spot':
amount = available_capital * position_ratio / current_price
else:
amount = (initial_capital * position_ratio * leverage) / current_price
# Futures sizing: treat available_capital as margin budget.
# Notional = margin * leverage, so base quantity = (margin * leverage) / price.
amount = (available_capital * position_ratio * leverage) / current_price
# Reduce sizing: position_size is treated as a reduce ratio (close X% of current position).
if sig in ("reduce_long", "reduce_short"):