"""crypto_okx_hub — 期权多机中控.""" from __future__ import annotations import os import sys import threading import time from functools import wraps from pathlib import Path from dotenv import load_dotenv from flask import ( Flask, Response, flash, jsonify, redirect, render_template, request, send_file, session, url_for, ) ROOT = Path(__file__).resolve().parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) ENV_PATH = ROOT / ".env" load_dotenv(ENV_PATH) from lib.backup_lib import ( backup_dir as hub_backup_dir, backup_keep_days as hub_backup_keep_days, backup_path as hub_backup_path, create_backup as create_hub_backup, list_backups as list_hub_backups, restore_backup as restore_hub_backup, save_uploaded_backup as save_hub_uploaded_backup, start_backup_scheduler as start_hub_backup_scheduler, ) from lib.bridge_client_lib import ( dispatch_cond_order_to_ids, fetch_backup_file, fetch_backup_list, fetch_client_version, fetch_cond_orders, post_client_update, post_close_full, post_cond_cancel, post_cond_exit, pull_all_enabled, ) from lib.cash_sample_lib import record_cash_samples_from_overview from lib.db_lib import db_path as hub_db_path, init_db from lib.env_lib import apply_env_updates, env_get from lib.health_watch_lib import inspect_overview_rows from lib.instances_lib import ( create_instance, delete_instance, get_instance, list_instances, update_instance, ) from lib.sso_lib import build_embed_open_url from lib.stats_agg_lib import build_stats_curve, build_stats_summary, normalize_window from lib.wechat_lib import ( hub_alert_cooldown_sec, hub_health_fail_threshold, hub_wechat_enabled, hub_wechat_webhook, send_hub_wechat, ) APP_HOST = (os.getenv("APP_HOST") or "0.0.0.0").strip() or "0.0.0.0" APP_PORT = int(os.getenv("APP_PORT") or 5100) SECRET = (os.getenv("FLASK_SECRET_KEY") or "hub-dev-secret").strip() # 可变登录态:系统设置页可改,并立刻生效 _auth = { "username": (os.getenv("HUB_USERNAME") or "admin").strip() or "admin", "password": (os.getenv("HUB_PASSWORD") or "admin123").strip() or "admin123", } app = Flask(__name__) app.secret_key = SECRET app.config["TEMPLATES_AUTO_RELOAD"] = True init_db() def hub_username() -> str: return _auth["username"] def hub_password() -> str: return _auth["password"] def set_hub_credentials(username: str, password: str) -> None: u = (username or "").strip() or "admin" p = password if password is not None else "" _auth["username"] = u _auth["password"] = p os.environ["HUB_USERNAME"] = u os.environ["HUB_PASSWORD"] = p def login_required(f): @wraps(f) def wrapped(*args, **kwargs): if session.get("logged_in"): return f(*args, **kwargs) if request.path.startswith("/api/"): return jsonify({"ok": False, "msg": "未登录"}), 401 return redirect(url_for("login", next=request.path)) return wrapped @app.route("/login", methods=["GET", "POST"]) def login(): if request.method == "POST": u = (request.form.get("username") or "").strip() p = request.form.get("password") or "" if u == hub_username() and p == hub_password(): session["logged_in"] = True session.permanent = True nxt = (request.args.get("next") or request.form.get("next") or "/").strip() or "/" if not nxt.startswith("/"): nxt = "/" return redirect(nxt) flash("用户名或密码错误") if session.get("logged_in"): return redirect("/") return render_template("login.html") @app.route("/logout") def logout(): session.clear() return redirect(url_for("login")) @app.route("/") @login_required def dashboard(): return render_template("dashboard.html") @app.route("/stats") @login_required def stats_page(): return render_template("stats.html") @app.route("/api/stats/summary") @login_required def api_stats_summary(): window = normalize_window(request.args.get("window") or "7d") try: return jsonify(build_stats_summary(window)) except Exception as e: return jsonify({"ok": False, "msg": str(e) or "统计失败"}), 500 @app.route("/api/stats/curve") @login_required def api_stats_curve(): range_key = normalize_window(request.args.get("range") or request.args.get("window") or "7d") try: return jsonify(build_stats_curve(range_key)) except Exception as e: return jsonify({"ok": False, "msg": str(e) or "曲线失败"}), 500 @app.route("/instances", methods=["GET", "POST"]) @login_required def instances_page(): if request.method == "POST": action = (request.form.get("action") or "create").strip() if action == "create": _, err = create_instance( name=request.form.get("name") or "", base_url=request.form.get("base_url") or "", bridge_token=request.form.get("bridge_token") or "", embed_path=request.form.get("embed_path") or "/embed", tags=request.form.get("tags") or "", enabled=bool(request.form.get("enabled")), ) flash(err or "已添加") elif action == "update": iid = int(request.form.get("id") or 0) _, err = update_instance( iid, name=request.form.get("name") or "", base_url=request.form.get("base_url") or "", bridge_token=request.form.get("bridge_token") or "", embed_path=request.form.get("embed_path") or "/embed", tags=request.form.get("tags") or "", enabled=bool(request.form.get("enabled")), ) flash(err or "已保存") elif action == "delete": iid = int(request.form.get("id") or 0) ok = delete_instance(iid) flash("已删除" if ok else "删除失败") return redirect(url_for("instances_page")) return render_template("instances.html", instances=list_instances()) def _wechat_settings_view() -> dict: enabled_raw = (env_get(ENV_PATH, "HUB_WECHAT_ENABLED") or os.getenv("HUB_WECHAT_ENABLED") or "").strip().lower() if enabled_raw in ("0", "false", "no", "off"): enabled = False elif enabled_raw in ("1", "true", "yes", "on"): enabled = True else: enabled = bool(hub_wechat_webhook()) return { "enabled": enabled, "webhook": hub_wechat_webhook() or env_get(ENV_PATH, "HUB_WECHAT_WEBHOOK", ""), "cooldown_sec": hub_alert_cooldown_sec(), "fail_threshold": hub_health_fail_threshold(), } def _apply_runtime_env(updates: dict[str, str]) -> None: for key, val in updates.items(): os.environ[key] = str(val) def _handle_account_settings() -> None: current_pw = request.form.get("current_password") or "" new_username = (request.form.get("new_username") or "").strip() new_password = request.form.get("new_password") or "" confirm = request.form.get("confirm_password") or "" if current_pw != hub_password(): flash("当前密码错误") return if len(new_username) < 2 or len(new_username) > 64: flash("用户名长度需为 2–64") return if any(ch.isspace() for ch in new_username): flash("用户名不能包含空格") return updates: dict[str, str] = {"HUB_USERNAME": new_username} final_password = hub_password() if new_password or confirm: if len(new_password) < 6: flash("新密码至少 6 位") return if new_password != confirm: flash("两次输入的新密码不一致") return updates["HUB_PASSWORD"] = new_password final_password = new_password try: apply_env_updates(ENV_PATH, updates) except OSError as exc: flash(f"写入 .env 失败: {exc}") return set_hub_credentials(new_username, final_password) flash("账号已更新,立即生效") def _handle_wechat_settings(*, send_test: bool = False) -> None: enabled = "1" if (request.form.get("wechat_enabled") or "") in ("1", "on", "true") else "0" webhook = (request.form.get("wechat_webhook") or "").strip() try: cooldown = max(60, int(float(request.form.get("alert_cooldown_sec") or 600))) except (TypeError, ValueError): flash("告警冷却须为数字(秒)") return try: threshold = max(1, int(float(request.form.get("health_fail_threshold") or 2))) except (TypeError, ValueError): flash("连续失败阈值须为正整数") return updates = { "HUB_WECHAT_ENABLED": "true" if enabled == "1" else "false", "HUB_WECHAT_WEBHOOK": webhook, "HUB_ALERT_COOLDOWN_SEC": str(cooldown), "HUB_HEALTH_FAIL_THRESHOLD": str(threshold), } try: apply_env_updates(ENV_PATH, updates) except OSError as exc: flash(f"写入 .env 失败: {exc}") return _apply_runtime_env(updates) if send_test: result = send_hub_wechat("中控企业微信测试消息\n配置已保存并立即生效") if result.get("ok"): flash("告警配置已保存,测试消息已发送") else: flash(f"告警配置已保存,但测试发送失败: {result.get('msg') or '未知错误'}") return flash("企业微信告警配置已保存,立即生效") @app.route("/settings", methods=["GET", "POST"]) @login_required def settings_page(): if request.method == "POST": action = (request.form.get("form_action") or "account").strip() return_tab = (request.form.get("return_tab") or "").strip() if action == "wechat": _handle_wechat_settings(send_test=False) return_tab = return_tab or "wechat" elif action == "wechat_test": _handle_wechat_settings(send_test=True) return_tab = return_tab or "wechat" else: _handle_account_settings() return_tab = return_tab or "account" if return_tab not in ("account", "wechat", "backup"): return_tab = "account" return redirect(url_for("settings_page", tab=return_tab)) return render_template( "settings.html", current_username=hub_username(), wechat=_wechat_settings_view(), ) @app.route("/api/settings/backup/hub/list", methods=["GET"]) @login_required def api_hub_backup_list(): return jsonify( { "ok": True, "items": list_hub_backups(), "backup_dir": hub_backup_dir(), "keep_days": hub_backup_keep_days(), "auto_hour_bj": 3, "includes": [".env", "database"], } ) def _resolved_hub_db_path() -> str: p = hub_db_path() if os.path.isabs(p): return p return str(ROOT / p) @app.route("/api/settings/backup/hub/create", methods=["POST"]) @login_required def api_hub_backup_create(): try: result = create_hub_backup( root_dir=str(ROOT), env_path=str(ENV_PATH), db_path=_resolved_hub_db_path(), label="manual", username=hub_username(), password=hub_password(), ) return jsonify(result) except Exception as exc: return jsonify({"ok": False, "msg": str(exc) or "备份失败"}), 500 @app.route("/api/settings/backup/hub/download/", methods=["GET"]) @login_required def api_hub_backup_download(name: str): path = hub_backup_path(name) if not path or not os.path.isfile(path): return jsonify({"ok": False, "msg": "备份不存在"}), 404 return send_file( path, mimetype="application/gzip", as_attachment=True, download_name=os.path.basename(path), ) @app.route("/api/settings/backup/hub/restore", methods=["POST"]) @login_required def api_hub_backup_restore(): data = request.get_json(silent=True) or {} name = str(data.get("name") or "").strip() if not name: return jsonify({"ok": False, "msg": "缺少备份文件名"}), 400 result = restore_hub_backup( name, root_dir=str(ROOT), env_path=str(ENV_PATH), db_path=_resolved_hub_db_path(), ) if result.get("ok"): u = str(result.get("hub_username") or "").strip() p = result.get("hub_password") if u or p is not None: set_hub_credentials(u or hub_username(), "" if p is None else str(p)) result.pop("hub_password", None) return jsonify(result), (200 if result.get("ok") else 400) @app.route("/api/settings/backup/hub/upload", methods=["POST"]) @login_required def api_hub_backup_upload(): fs = request.files.get("file") if fs is None or not getattr(fs, "filename", None): return jsonify({"ok": False, "msg": "请选择备份文件"}), 400 result = save_hub_uploaded_backup(fs) return jsonify(result), (200 if result.get("ok") else 400) @app.route("/api/settings/backup/instances", methods=["GET"]) @login_required def api_backup_instances(): items = [ { "id": i["id"], "name": i.get("name") or f"#{i['id']}", "enabled": bool(i.get("enabled")), } for i in list_instances(enabled_only=False) ] return jsonify({"ok": True, "items": items}) @app.route("/api/settings/backup/client//list", methods=["GET"]) @login_required def api_client_backup_list(iid: int): inst = get_instance(iid) if not inst: return jsonify({"ok": False, "msg": "实例不存在", "items": []}), 404 result = fetch_backup_list(inst) code = 200 if result.get("ok") else 502 return jsonify(result), code @app.route("/api/settings/backup/client//download/", methods=["GET"]) @login_required def api_client_backup_download(iid: int, name: str): inst = get_instance(iid) if not inst: return jsonify({"ok": False, "msg": "实例不存在"}), 404 ok, content, meta = fetch_backup_file(inst, name) if not ok or content is None: return jsonify({"ok": False, "msg": meta or "下载失败"}), 502 return Response( content, mimetype="application/gzip", headers={ "Content-Disposition": f'attachment; filename="{os.path.basename(meta)}"' }, ) @app.route("/desk") @login_required def desk(): instances = list_instances(enabled_only=True) iid = request.args.get("id") # 未指定客户端时默认打开第一个 if (not iid or not str(iid).isdigit()) and instances: return redirect(url_for("desk", id=instances[0]["id"])) inst = None open_url = "" if iid and str(iid).isdigit(): inst = get_instance(int(iid)) if inst: open_url = build_embed_open_url( inst["base_url"], bridge_token=inst["bridge_token"], embed_path=inst.get("embed_path") or "/embed", ttl_sec=3600, ) return render_template( "desk.html", instances=instances, current=inst, open_url=open_url, ) @app.route("/api/overview") @login_required def api_overview(): rows = pull_all_enabled() return jsonify({"ok": True, "instances": rows}) def _health_pull_interval_sec() -> float: try: return max(5.0, float(os.getenv("HUB_PULL_INTERVAL_SEC") or 10)) except (TypeError, ValueError): return 10.0 def _health_watch_loop() -> None: """后台巡检:即使无人打开总览也会告警;并采样余额曲线.""" while True: try: rows = pull_all_enabled() inspect_overview_rows(rows) record_cash_samples_from_overview(rows) except Exception: pass time.sleep(_health_pull_interval_sec()) def start_health_watch() -> None: t = threading.Thread(target=_health_watch_loop, name="hub-health-watch", daemon=True) t.start() start_health_watch() try: start_hub_backup_scheduler( root_dir=str(ROOT), env_path=str(ENV_PATH), db_path_fn=_resolved_hub_db_path, hour=3, get_credentials=lambda: (hub_username(), hub_password()), ) except Exception as _hub_bak_err: print(f"[backup] scheduler: {_hub_bak_err}") @app.route("/api/instances//open_url") @login_required def api_open_url(iid: int): inst = get_instance(iid) if not inst: return jsonify({"ok": False, "msg": "不存在"}), 404 url = build_embed_open_url( inst["base_url"], bridge_token=inst["bridge_token"], embed_path=inst.get("embed_path") or "/embed", ttl_sec=3600, ) return jsonify({"ok": True, "url": url, "base_url": inst["base_url"]}) @app.route("/api/settings/update/hub/version", methods=["GET"]) @login_required def api_hub_self_version(): from lib.update_lib import read_git_version return jsonify(read_git_version()) @app.route("/api/settings/update/hub", methods=["POST"]) @login_required def api_hub_self_update(): """系统设置:中控本机快速更新.""" from lib.update_lib import schedule_pull_and_restart try: # 同步 pull + 延迟重启;timeout 留给 git pull result = schedule_pull_and_restart(delay_seconds=1.2) except Exception as e: result = {"ok": False, "msg": f"触发更新异常: {e}", "deferred": False} return jsonify(result), (200 if result.get("ok") else 400) @app.route("/api/settings/update/clients", methods=["GET"]) @login_required def api_update_clients(): items = [ { "id": i["id"], "name": i.get("name") or f"#{i['id']}", "base_url": i.get("base_url") or "", "enabled": bool(i.get("enabled")), } for i in list_instances(enabled_only=False) ] return jsonify({"ok": True, "items": items}) @app.route("/api/instances//version", methods=["GET"]) @login_required def api_instance_version(iid: int): inst = get_instance(iid) if not inst: return jsonify({"ok": False, "msg": "实例不存在", "online": False}), 404 result = fetch_client_version(inst) code = 200 if result.get("online") else 502 return jsonify(result), code @app.route("/api/instances//update", methods=["POST"]) @login_required def api_instance_update(iid: int): """系统设置远程更新客户端:转发 Bridge /api/hub/update.""" inst = get_instance(iid) if not inst: return jsonify({"ok": False, "msg": "实例不存在"}), 404 if not inst.get("enabled"): return jsonify({"ok": False, "msg": "实例未启用"}), 400 result = post_client_update(inst) status = 200 if not result.get("ok"): msg = str(result.get("msg") or "") if msg.startswith("HTTP ") or "超时" in msg or "下发失败" in msg or "缺少 URL" in msg: status = 502 else: status = 400 return jsonify(result), status @app.route("/api/instances//options/close", methods=["POST"]) @login_required def api_instance_options_close(iid: int): """总览下发全平:中控只转发,客户端本地买一限价执行.""" inst = get_instance(iid) if not inst: return jsonify({"ok": False, "msg": "实例不存在"}), 404 if not inst.get("enabled"): return jsonify({"ok": False, "msg": "实例未启用"}), 400 data = request.get_json(silent=True) or {} opt_inst = (data.get("inst_id") or "").strip() if not opt_inst: return jsonify({"ok": False, "msg": "缺少 inst_id"}) result = post_close_full(inst, inst_id=opt_inst) status = 200 if not result.get("ok"): # 客户端业务拒绝(残档等)仍 200 带 ok:false;仅传输层失败用 502 msg = str(result.get("msg") or "") if msg.startswith("HTTP ") or "超时" in msg or "下发失败" in msg or "缺少 URL" in msg: status = 502 return jsonify(result), status @app.route("/cond_orders") @login_required def cond_orders_page(): return render_template("cond_orders.html") @app.route("/api/cond_orders/candidates") @login_required def api_cond_order_candidates(): """可选客户端:在线且无持仓、未被条件单锁定.""" rows = pull_all_enabled() out = [] for row in rows: snap = row.get("snapshot") if isinstance(row.get("snapshot"), dict) else {} brief = snap.get("positions_brief") or snap.get("positions") or [] has_pos = isinstance(brief, list) and len(brief) > 0 locked = bool(snap.get("cond_order_locked")) selectable = bool(row.get("online")) and not has_pos and not locked inst = snap.get("instance") if isinstance(snap.get("instance"), dict) else {} mode = str(inst.get("mode") or "").strip().lower() if mode not in ("sim", "live"): mode = "" out.append( { "id": row.get("id"), "name": row.get("name"), "online": bool(row.get("online")), "has_position": has_pos, "cond_order_locked": locked, "selectable": selectable, "mode": mode, "error": row.get("error") or "", } ) return jsonify({"ok": True, "instances": out}) @app.route("/api/cond_orders", methods=["GET"]) @login_required def api_cond_orders_list(): instances = list_instances(enabled_only=True) all_orders = [] for inst in instances: data = fetch_cond_orders(inst) for o in data.get("orders") or []: if not isinstance(o, dict): continue item = dict(o) item["instance_id"] = inst.get("id") item["instance_name"] = inst.get("name") all_orders.append(item) return jsonify({"ok": True, "orders": all_orders}) @app.route("/api/cond_orders", methods=["POST"]) @login_required def api_cond_orders_create(): import uuid data = request.get_json(silent=True) or {} ids_raw = data.get("instance_ids") or [] try: instance_ids = [int(x) for x in ids_raw] except (TypeError, ValueError): return jsonify({"ok": False, "msg": "instance_ids 无效"}) if not instance_ids: return jsonify({"ok": False, "msg": "请选择客户端"}) # 再次按快照过滤有仓/锁定 rows = {r.get("id"): r for r in pull_all_enabled()} allowed = [] rejected = [] for iid in instance_ids: row = rows.get(iid) if not row or not row.get("online"): rejected.append({"instance_id": iid, "msg": "离线或不可用"}) continue snap = row.get("snapshot") if isinstance(row.get("snapshot"), dict) else {} brief = snap.get("positions_brief") or [] if brief: rejected.append({"instance_id": iid, "msg": "已有持仓不可选"}) continue if snap.get("cond_order_locked"): rejected.append({"instance_id": iid, "msg": "已有条件单占用"}) continue allowed.append(iid) if not allowed: return jsonify({"ok": False, "msg": "没有可下发的客户端", "rejected": rejected}) hub_order_id = (data.get("hub_order_id") or "").strip() or str(uuid.uuid4()) opt_type = (data.get("opt_type") or "P").strip().upper() if opt_type not in ("P", "C"): opt_type = "P" # 左侧单:看跌上穿、看涨下穿 trigger_side = "gte" if opt_type == "P" else "lte" payload = { "hub_order_id": hub_order_id, "underlying": (data.get("underlying") or "ETH").strip().upper(), "opt_type": opt_type, "trigger_side": trigger_side, "trigger_px": data.get("trigger_px"), "deviation_pct": data.get("deviation_pct"), "min_expiry_hours": data.get("min_expiry_hours"), "exit": { "target_index": data.get("target_index"), "profit_exit_enabled": bool(data.get("profit_exit_enabled")), "profit_exit_mult": data.get("profit_exit_mult"), }, } results = dispatch_cond_order_to_ids(allowed, payload) ok_n = sum(1 for r in results if r.get("ok")) return jsonify( { "ok": ok_n > 0, "msg": f"成功 {ok_n}/{len(results)}", "hub_order_id": hub_order_id, "results": results, "rejected": rejected, } ) @app.route("/api/cond_orders//cancel", methods=["POST"]) @login_required def api_cond_orders_cancel(hub_order_id: str): data = request.get_json(silent=True) or {} iid = data.get("instance_id") if iid is None: return jsonify({"ok": False, "msg": "缺少 instance_id"}) inst = get_instance(int(iid)) if not inst: return jsonify({"ok": False, "msg": "实例不存在"}), 404 return jsonify(post_cond_cancel(inst, hub_order_id)) @app.route("/api/cond_orders//exit", methods=["POST"]) @login_required def api_cond_orders_exit(hub_order_id: str): data = request.get_json(silent=True) or {} iid = data.get("instance_id") if iid is None: return jsonify({"ok": False, "msg": "缺少 instance_id"}) inst = get_instance(int(iid)) if not inst: return jsonify({"ok": False, "msg": "实例不存在"}), 404 payload = { "target_index": data.get("target_index"), "profit_exit_enabled": data.get("profit_exit_enabled"), "profit_exit_mult": data.get("profit_exit_mult"), } return jsonify(post_cond_exit(inst, hub_order_id, payload)) @app.route("/manifest.webmanifest") def pwa_manifest(): path = ROOT / "static" / "icons" / "manifest.webmanifest" return send_file(path, mimetype="application/manifest+json; charset=utf-8") @app.route("/sw.js") def pwa_service_worker(): path = ROOT / "static" / "sw.js" resp = send_file(path, mimetype="application/javascript; charset=utf-8") resp.headers["Cache-Control"] = "no-cache" resp.headers["Service-Worker-Allowed"] = "/" return resp if __name__ == "__main__": # PM2 下禁止 debug reloader,否则会 Bad file descriptor 崩循环 _debug = (os.getenv("APP_DEBUG") or "").strip().lower() in ("1", "true", "yes", "on") app.run( host=APP_HOST, port=APP_PORT, debug=_debug, use_reloader=False, threaded=True, )