dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复中控

Browse Source
This commit is contained in:
dekun
2026-05-25 08:02:06 +08:00
parent e89708726f
commit 5f4f33cc10
7 changed files with 154 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
manual_trading_hub/hub.py
+83 -48
View File
@@ -43,7 +43,11 @@ HUB_BRIDGE_TOKEN = (os.getenv("HUB_BRIDGE_TOKEN") or os.getenv("CONTROL_TOKEN")
_trust_raw = (os.getenv("HUB_TRUST_LAN", "true") or "").strip().lower()
HUB_TRUST_LAN = _trust_raw not in ("0", "false", "no", "off")
DIR = Path(__file__).resolve().parent
HUB_BUILD = "20260525-mobile"
HUB_BUILD = "20260525-perf"
HUB_AGENT_TIMEOUT = float(os.getenv("HUB_AGENT_TIMEOUT", "8"))
HUB_FLASK_TIMEOUT = float(os.getenv("HUB_FLASK_TIMEOUT", "10"))
_board_key_prices_raw = (os.getenv("HUB_BOARD_KEY_PRICES", "true") or "").strip().lower()
HUB_BOARD_KEY_PRICES = _board_key_prices_raw in ("1", "true", "yes", "on")
def _is_local(host: str | None) -> bool:
@@ -261,7 +265,7 @@ def api_settings_meta():
async def _fetch_agent_status(client: httpx.AsyncClient, ex: dict) -> dict:
url = f"{ex['agent_url'].rstrip('/')}/status"
try:
r = await client.get(url, headers=_agent_headers(), timeout=12.0)
r = await client.get(url, headers=_agent_headers(), timeout=HUB_AGENT_TIMEOUT)
body = r.json() if r.content else {}
return {
"id": ex["id"],
@@ -317,7 +321,7 @@ async def _fetch_flask_json(
return None
try:
if method == "GET":
r = await client.get(f"{base}{path}", headers=_hub_headers(), timeout=15.0)
r = await client.get(f"{base}{path}", headers=_hub_headers(), timeout=HUB_FLASK_TIMEOUT)
else:
r = await client.post(f"{base}{path}", headers=_hub_headers(), data=data, timeout=120.0)
if r.status_code >= 400:
@@ -330,55 +334,86 @@ async def _fetch_flask_json(
return {"ok": False, "error": str(e)}
def _flask_error_from_hub_mon(hub_mon: dict | None) -> str | None:
if not isinstance(hub_mon, dict) or hub_mon.get("ok") is not False:
return None
st = hub_mon.get("status")
if st == 404:
return (
"HTTP 404：该 Flask 未注册 /api/hub/*（hub_bridge 未加载）。"
"请在仓库根目录 git pull 后 pm2 restart crypto_binance crypto_gate crypto_gate_bot，"
"并查看启动日志是否含 [hub_bridge] ImportError"
)
return (
hub_mon.get("msg")
or hub_mon.get("error")
or (f"HTTP {st}" if st else None)
or (str(hub_mon.get("text") or "")[:120] or None)
)
async def _fetch_exchange_flask_bundle(
client: httpx.AsyncClient, ex: dict
) -> tuple[dict | None, dict | None, list | None]:
"""单所 Flask：monitor / meta /（可选）price_snapshot 并行拉取。"""
caps = ex.get("capabilities") or []
tasks = [
_fetch_flask_json(client, ex, "/api/hub/monitor"),
_fetch_flask_json(client, ex, "/api/hub/meta"),
]
want_prices = HUB_BOARD_KEY_PRICES and "key" in caps
if want_prices:
tasks.append(_fetch_flask_json(client, ex, "/api/price_snapshot"))
results = await asyncio.gather(*tasks)
hub_mon = results[0]
meta = results[1]
key_prices = None
if want_prices and len(results) > 2:
snap = results[2]
if isinstance(snap, dict):
key_prices = snap.get("key_prices")
return hub_mon, meta, key_prices
async def _assemble_board_row(
client: httpx.AsyncClient, ex: dict, agent_row: dict
) -> dict:
hub_mon, meta, key_prices = await _fetch_exchange_flask_bundle(client, ex)
flask_ok = isinstance(hub_mon, dict) and hub_mon.get("ok") is not False
raw_review = (ex.get("review_url") or "").strip()
review_link = browser_url(raw_review) if raw_review else default_review_url(
ex.get("flask_url")
)
return {
**agent_row,
"flask_url": ex.get("flask_url") or "",
"flask_url_browser": browser_url(ex.get("flask_url")),
"review_url": review_link,
"hub_monitor": hub_mon,
"flask_ok": flask_ok,
"flask_error": _flask_error_from_hub_mon(hub_mon if isinstance(hub_mon, dict) else None),
"meta": (meta or {}).get("meta") if isinstance(meta, dict) else meta,
"key_prices": key_prices,
}
@app.get("/api/monitor/board")
async def api_monitor_board():
exchanges = enabled_exchanges()
async with httpx.AsyncClient() as client:
agent_rows = await asyncio.gather(*[_fetch_agent_status(client, ex) for ex in exchanges])
out = []
for ex, agent_row in zip(exchanges, agent_rows):
hub_mon = await _fetch_flask_json(client, ex, "/api/hub/monitor")
meta = await _fetch_flask_json(client, ex, "/api/hub/meta")
key_prices = None
if "key" in (ex.get("capabilities") or []):
snap = await _fetch_flask_json(client, ex, "/api/price_snapshot")
if isinstance(snap, dict):
key_prices = snap.get("key_prices")
flask_ok = isinstance(hub_mon, dict) and hub_mon.get("ok") is not False
flask_err = None
if isinstance(hub_mon, dict) and hub_mon.get("ok") is False:
st = hub_mon.get("status")
if st == 404:
flask_err = (
"HTTP 404：该 Flask 未注册 /api/hub/*（hub_bridge 未加载）。"
"请在仓库根目录 git pull 后 pm2 restart crypto_binance crypto_gate crypto_gate_bot，"
"并查看启动日志是否含 [hub_bridge] ImportError"
)
else:
flask_err = (
hub_mon.get("msg")
or hub_mon.get("error")
or (f"HTTP {st}" if st else None)
or (str(hub_mon.get("text") or "")[:120] or None)
)
raw_review = (ex.get("review_url") or "").strip()
review_link = browser_url(raw_review) if raw_review else default_review_url(
ex.get("flask_url")
)
out.append(
{
**agent_row,
"flask_url": ex.get("flask_url") or "",
"flask_url_browser": browser_url(ex.get("flask_url")),
"review_url": review_link,
"hub_monitor": hub_mon,
"flask_ok": flask_ok,
"flask_error": flask_err,
"meta": (meta or {}).get("meta") if isinstance(meta, dict) else meta,
"key_prices": key_prices,
}
)
return {"rows": out, "updated_at": __import__("datetime").datetime.now().isoformat(timespec="seconds")}
agent_rows = await asyncio.gather(
*[_fetch_agent_status(client, ex) for ex in exchanges]
)
out = await asyncio.gather(
*[
_assemble_board_row(client, ex, agent_row)
for ex, agent_row in zip(exchanges, agent_rows)
]
)
return {
"rows": list(out),
"updated_at": __import__("datetime").datetime.now().isoformat(timespec="seconds"),
}
class CloseAllBody(BaseModel):