dekun
121 lines
3.5 KiB
Python
121 lines
3.5 KiB
Python
"""中控交易所配置（hub_settings.json）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
DIR = Path(__file__).resolve().parent
|
|
SETTINGS_PATH = DIR / "hub_settings.json"
|
|
|
|
DEFAULT_DISPLAY = {
|
|
"show_account_pnl": True,
|
|
"show_nav_funds": True,
|
|
"show_nav_dashboard": True,
|
|
"show_nav_plan": True,
|
|
"show_nav_archive": True,
|
|
"show_nav_ai": True,
|
|
"show_nav_calculator": True,
|
|
}
|
|
|
|
DEFAULT_EXCHANGES = [
|
|
{
|
|
"id": "0",
|
|
"key": "binance",
|
|
"name": "币安 · crypto_monitor_binance",
|
|
"agent_url": "http://127.0.0.1:15200",
|
|
"flask_url": "http://127.0.0.1:5001",
|
|
"review_url": "http://127.0.0.1:5001/records",
|
|
"enabled": True,
|
|
"capabilities": ["key", "trend"],
|
|
},
|
|
{
|
|
"id": "1",
|
|
"key": "okx",
|
|
"name": "OKX · crypto_monitor_okx",
|
|
"agent_url": "http://127.0.0.1:15201",
|
|
"flask_url": "http://127.0.0.1:5004",
|
|
"review_url": "http://127.0.0.1:5004/records",
|
|
"enabled": True,
|
|
"capabilities": ["key", "trend"],
|
|
},
|
|
{
|
|
"id": "2",
|
|
"key": "gate",
|
|
"name": "Gate训练 · crypto_monitor_gate",
|
|
"agent_url": "http://127.0.0.1:15202",
|
|
"flask_url": "http://127.0.0.1:5000",
|
|
"review_url": "http://127.0.0.1:5000/records",
|
|
"enabled": True,
|
|
"capabilities": ["key", "trend"],
|
|
},
|
|
{
|
|
"id": "3",
|
|
"key": "gate_bot",
|
|
"name": "Gate趋势 · crypto_monitor_gate_bot",
|
|
"agent_url": "http://127.0.0.1:15203",
|
|
"flask_url": "http://127.0.0.1:5002",
|
|
"review_url": "http://127.0.0.1:5002/records",
|
|
"enabled": True,
|
|
"capabilities": ["key", "trend"],
|
|
},
|
|
]
|
|
|
|
|
|
def _ids_from_csv(raw: str | None) -> set[str]:
|
|
if not raw or not str(raw).strip():
|
|
return set()
|
|
return {x.strip() for x in str(raw).split(",") if x.strip()}
|
|
|
|
|
|
def env_force_disabled_ids() -> set[str]:
|
|
# 未设置时默认不强制关闭任何账户；要用旧行为可设 HUB_DISABLED_IDS=1
|
|
raw = (os.getenv("HUB_DISABLED_IDS") or "").strip()
|
|
return _ids_from_csv(raw)
|
|
|
|
|
|
def normalize_display_prefs(raw: dict | None) -> dict:
|
|
out = dict(DEFAULT_DISPLAY)
|
|
if isinstance(raw, dict):
|
|
for key in DEFAULT_DISPLAY:
|
|
if key in raw:
|
|
out[key] = bool(raw.get(key))
|
|
return out
|
|
|
|
|
|
def load_settings() -> dict:
|
|
data = {
|
|
"exchanges": [dict(x) for x in DEFAULT_EXCHANGES],
|
|
"version": 1,
|
|
"display": dict(DEFAULT_DISPLAY),
|
|
}
|
|
if SETTINGS_PATH.is_file():
|
|
try:
|
|
loaded = json.loads(SETTINGS_PATH.read_text(encoding="utf-8"))
|
|
if isinstance(loaded, dict) and isinstance(loaded.get("exchanges"), list):
|
|
data = loaded
|
|
except Exception:
|
|
pass
|
|
data["display"] = normalize_display_prefs(data.get("display"))
|
|
force_off = env_force_disabled_ids()
|
|
for ex in data.get("exchanges") or []:
|
|
if str(ex.get("id")) in force_off:
|
|
ex["enabled"] = False
|
|
ex["env_disabled"] = True
|
|
else:
|
|
ex.setdefault("env_disabled", False)
|
|
return data
|
|
|
|
|
|
def save_settings(data: dict) -> None:
|
|
SETTINGS_PATH.write_text(
|
|
json.dumps(data, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def enabled_exchanges(data: dict | None = None) -> list[dict]:
|
|
data = data or load_settings()
|
|
return [x for x in data.get("exchanges") or [] if x.get("enabled")]
|