dekun
291 lines
11 KiB
Python
291 lines
11 KiB
Python
"""拉取各 crypto_okx 客户端快照;以及受限写指令(买一全平/条件单)."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from lib.instances_lib import get_instance, list_instances
|
|
|
|
|
|
def _timeout() -> float:
|
|
try:
|
|
return max(2.0, float(os.getenv("HUB_PULL_TIMEOUT_SEC") or 5))
|
|
except (TypeError, ValueError):
|
|
return 5.0
|
|
|
|
|
|
def _close_timeout() -> float:
|
|
try:
|
|
return max(8.0, float(os.getenv("HUB_CLOSE_TIMEOUT_SEC") or 45))
|
|
except (TypeError, ValueError):
|
|
return 45.0
|
|
|
|
|
|
def _headers(token: str) -> dict[str, str]:
|
|
return {
|
|
"X-Hub-Token": token,
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def fetch_one_snapshot(inst: dict[str, Any]) -> dict[str, Any]:
|
|
base = (inst.get("base_url") or "").rstrip("/")
|
|
token = (inst.get("bridge_token") or "").strip()
|
|
out: dict[str, Any] = {
|
|
"id": inst.get("id"),
|
|
"name": inst.get("name"),
|
|
"base_url": base,
|
|
"embed_path": inst.get("embed_path") or "/embed",
|
|
"tags": inst.get("tags") or "",
|
|
"online": False,
|
|
"error": "",
|
|
"snapshot": None,
|
|
"pulled_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
if not base or not token:
|
|
out["error"] = "缺少 URL 或 Token"
|
|
return out
|
|
url = f"{base}/api/hub/options/snapshot"
|
|
try:
|
|
r = requests.get(
|
|
url,
|
|
headers={"X-Hub-Token": token, "Accept": "application/json"},
|
|
timeout=_timeout(),
|
|
)
|
|
data = {}
|
|
try:
|
|
data = r.json()
|
|
except Exception:
|
|
data = {}
|
|
if r.status_code != 200 or not isinstance(data, dict) or not data.get("ok"):
|
|
out["error"] = (data.get("msg") if isinstance(data, dict) else None) or f"HTTP {r.status_code}"
|
|
out["snapshot"] = data if isinstance(data, dict) else None
|
|
return out
|
|
out["online"] = True
|
|
out["snapshot"] = data
|
|
return out
|
|
except requests.Timeout:
|
|
out["error"] = "超时"
|
|
return out
|
|
except Exception as e:
|
|
out["error"] = str(e) or "拉取失败"
|
|
return out
|
|
|
|
|
|
def post_close_full(inst: dict[str, Any], *, inst_id: str) -> dict[str, Any]:
|
|
"""向客户端下发买一全平指令;下单在客户端本地执行."""
|
|
base = (inst.get("base_url") or "").rstrip("/")
|
|
token = (inst.get("bridge_token") or "").strip()
|
|
iid = (inst_id or "").strip()
|
|
if not base or not token:
|
|
return {"ok": False, "msg": "缺少 URL 或 Token"}
|
|
if not iid:
|
|
return {"ok": False, "msg": "缺少 inst_id"}
|
|
url = f"{base}/api/hub/options/close"
|
|
try:
|
|
r = requests.post(
|
|
url,
|
|
headers=_headers(token),
|
|
json={"inst_id": iid},
|
|
timeout=_close_timeout(),
|
|
)
|
|
try:
|
|
data = r.json()
|
|
except Exception:
|
|
data = {}
|
|
if isinstance(data, dict) and data:
|
|
if r.status_code >= 400 and "ok" not in data:
|
|
data = dict(data)
|
|
data["ok"] = False
|
|
data.setdefault("msg", f"HTTP {r.status_code}")
|
|
return data
|
|
return {"ok": False, "msg": f"HTTP {r.status_code}"}
|
|
except requests.Timeout:
|
|
return {"ok": False, "msg": "客户端超时(下单可能仍在进行,请刷新确认)"}
|
|
except Exception as e:
|
|
return {"ok": False, "msg": str(e) or "下发失败"}
|
|
|
|
|
|
def post_cond_order(inst: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
|
|
base = (inst.get("base_url") or "").rstrip("/")
|
|
token = (inst.get("bridge_token") or "").strip()
|
|
if not base or not token:
|
|
return {"ok": False, "msg": "缺少 URL 或 Token", "instance_id": inst.get("id")}
|
|
url = f"{base}/api/hub/options/cond_orders"
|
|
try:
|
|
r = requests.post(url, headers=_headers(token), json=payload, timeout=_close_timeout())
|
|
data = r.json() if r.content else {}
|
|
if not isinstance(data, dict):
|
|
data = {"ok": False, "msg": f"HTTP {r.status_code}"}
|
|
data["instance_id"] = inst.get("id")
|
|
data["instance_name"] = inst.get("name")
|
|
return data
|
|
except requests.Timeout:
|
|
return {
|
|
"ok": False,
|
|
"msg": "客户端超时",
|
|
"instance_id": inst.get("id"),
|
|
"instance_name": inst.get("name"),
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"ok": False,
|
|
"msg": str(e) or "下发失败",
|
|
"instance_id": inst.get("id"),
|
|
"instance_name": inst.get("name"),
|
|
}
|
|
|
|
|
|
def post_cond_cancel(inst: dict[str, Any], hub_order_id: str) -> dict[str, Any]:
|
|
base = (inst.get("base_url") or "").rstrip("/")
|
|
token = (inst.get("bridge_token") or "").strip()
|
|
oid = (hub_order_id or "").strip()
|
|
if not base or not token or not oid:
|
|
return {"ok": False, "msg": "参数不完整"}
|
|
url = f"{base}/api/hub/options/cond_orders/{oid}/cancel"
|
|
try:
|
|
r = requests.post(url, headers=_headers(token), json={}, timeout=_timeout())
|
|
data = r.json() if r.content else {}
|
|
return data if isinstance(data, dict) else {"ok": False, "msg": f"HTTP {r.status_code}"}
|
|
except Exception as e:
|
|
return {"ok": False, "msg": str(e) or "取消失败"}
|
|
|
|
|
|
def post_cond_exit(inst: dict[str, Any], hub_order_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
base = (inst.get("base_url") or "").rstrip("/")
|
|
token = (inst.get("bridge_token") or "").strip()
|
|
oid = (hub_order_id or "").strip()
|
|
if not base or not token or not oid:
|
|
return {"ok": False, "msg": "参数不完整"}
|
|
url = f"{base}/api/hub/options/cond_orders/{oid}/exit"
|
|
try:
|
|
r = requests.post(url, headers=_headers(token), json=payload or {}, timeout=_timeout())
|
|
data = r.json() if r.content else {}
|
|
return data if isinstance(data, dict) else {"ok": False, "msg": f"HTTP {r.status_code}"}
|
|
except Exception as e:
|
|
return {"ok": False, "msg": str(e) or "更新失败"}
|
|
|
|
|
|
def fetch_cond_orders(inst: dict[str, Any]) -> dict[str, Any]:
|
|
base = (inst.get("base_url") or "").rstrip("/")
|
|
token = (inst.get("bridge_token") or "").strip()
|
|
if not base or not token:
|
|
return {"ok": False, "msg": "缺少 URL 或 Token", "orders": []}
|
|
url = f"{base}/api/hub/options/cond_orders"
|
|
try:
|
|
r = requests.get(
|
|
url,
|
|
headers={"X-Hub-Token": token, "Accept": "application/json"},
|
|
timeout=_timeout(),
|
|
)
|
|
data = r.json() if r.content else {}
|
|
if isinstance(data, dict):
|
|
data["instance_id"] = inst.get("id")
|
|
data["instance_name"] = inst.get("name")
|
|
return data
|
|
return {"ok": False, "msg": f"HTTP {r.status_code}", "orders": []}
|
|
except Exception as e:
|
|
return {"ok": False, "msg": str(e) or "拉取失败", "orders": [], "instance_id": inst.get("id")}
|
|
|
|
|
|
def pull_all_enabled() -> list[dict[str, Any]]:
|
|
instances = list_instances(enabled_only=True)
|
|
if not instances:
|
|
return []
|
|
results: list[dict[str, Any]] = []
|
|
with ThreadPoolExecutor(max_workers=min(8, len(instances))) as pool:
|
|
futs = {pool.submit(fetch_one_snapshot, inst): inst["id"] for inst in instances}
|
|
by_id: dict[Any, dict[str, Any]] = {}
|
|
for fut in as_completed(futs):
|
|
row = fut.result()
|
|
by_id[row.get("id")] = row
|
|
for inst in instances:
|
|
results.append(by_id.get(inst["id"]) or fetch_one_snapshot(inst))
|
|
return results
|
|
|
|
|
|
def fetch_one_stats(inst: dict[str, Any], *, window: str = "7d") -> dict[str, Any]:
|
|
"""拉取单机窗口化期权统计."""
|
|
base = (inst.get("base_url") or "").rstrip("/")
|
|
token = (inst.get("bridge_token") or "").strip()
|
|
out: dict[str, Any] = {
|
|
"id": inst.get("id"),
|
|
"name": inst.get("name"),
|
|
"online": False,
|
|
"error": "",
|
|
"stats": None,
|
|
}
|
|
if not base or not token:
|
|
out["error"] = "缺少 URL 或 Token"
|
|
return out
|
|
url = f"{base}/api/hub/options/stats"
|
|
try:
|
|
r = requests.get(
|
|
url,
|
|
headers={"X-Hub-Token": token, "Accept": "application/json"},
|
|
params={"window": window},
|
|
timeout=max(_timeout(), 15.0),
|
|
)
|
|
data = {}
|
|
try:
|
|
data = r.json()
|
|
except Exception:
|
|
data = {}
|
|
if r.status_code != 200 or not isinstance(data, dict) or not data.get("ok"):
|
|
out["error"] = (data.get("msg") if isinstance(data, dict) else None) or f"HTTP {r.status_code}"
|
|
out["stats"] = data if isinstance(data, dict) else None
|
|
return out
|
|
out["online"] = True
|
|
out["stats"] = data
|
|
return out
|
|
except requests.Timeout:
|
|
out["error"] = "超时"
|
|
return out
|
|
except Exception as e:
|
|
out["error"] = str(e) or "拉取失败"
|
|
return out
|
|
|
|
|
|
def pull_all_stats(window: str = "7d") -> list[dict[str, Any]]:
|
|
instances = list_instances(enabled_only=True)
|
|
if not instances:
|
|
return []
|
|
results: list[dict[str, Any]] = []
|
|
with ThreadPoolExecutor(max_workers=min(8, len(instances))) as pool:
|
|
futs = {
|
|
pool.submit(fetch_one_stats, inst, window=window): inst["id"] for inst in instances
|
|
}
|
|
by_id: dict[Any, dict[str, Any]] = {}
|
|
for fut in as_completed(futs):
|
|
row = fut.result()
|
|
by_id[row.get("id")] = row
|
|
for inst in instances:
|
|
results.append(by_id.get(inst["id"]) or fetch_one_stats(inst, window=window))
|
|
return results
|
|
|
|
|
|
def dispatch_cond_order_to_ids(instance_ids: list[int], payload: dict[str, Any]) -> list[dict[str, Any]]:
|
|
results = []
|
|
for iid in instance_ids:
|
|
inst = get_instance(int(iid))
|
|
if not inst:
|
|
results.append({"ok": False, "msg": "实例不存在", "instance_id": iid})
|
|
continue
|
|
if not inst.get("enabled"):
|
|
results.append(
|
|
{
|
|
"ok": False,
|
|
"msg": "实例未启用",
|
|
"instance_id": iid,
|
|
"instance_name": inst.get("name"),
|
|
}
|
|
)
|
|
continue
|
|
results.append(post_cond_order(inst, payload))
|
|
return results
|