"""中控 Bridge:只读快照 + 受限写(买一全平) + SSO 登录入口.""" from __future__ import annotations from typing import Any from flask import Flask, jsonify, redirect, request, session from urllib.parse import quote from lib.hub.hub_bridge_auth import ( frame_ancestors_header_value, hub_bridge_token, hub_bridge_token_ok, make_embed_session_token, verify_sso_token, ) def _safe_next_path(raw: str, default: str = "/embed") -> str: p = (raw or "").strip() or default if not p.startswith("/"): p = "/" + p if "://" in p or p.startswith("//"): return default return p def _enrich_hub_protocol(snap: dict[str, Any], *, app_module: Any) -> dict[str, Any]: """对齐中控设计文档 version=1 字段,同时保留原 snapshot 键.""" out = dict(snap or {}) out["version"] = 1 mode = "live" try: from lib.sim.mode_lib import is_sim_mode get_db = getattr(app_module, "get_db", None) if callable(get_db) and is_sim_mode(get_db): mode = "sim" except Exception: mode = "live" label = "" try: from lib.sim.mode_lib import exchange_mode_label get_db = getattr(app_module, "get_db", None) if callable(get_db): label = exchange_mode_label(get_db=get_db, exchange_key="okx") or "" except Exception: label = "" if not label: label = "okx_sim" if mode == "sim" else "okx_live" positions = out.get("positions") if isinstance(out.get("positions"), list) else [] slim_pos = [] premium_total_usdt = 0.0 has_premium = False for p in positions: if not isinstance(p, dict): continue preview = p.get("close_preview") if isinstance(p.get("close_preview"), dict) else {} # 仅在实例已附带 close_preview 时判定残档;缺字段=未同步,勿假装「无买一/残档」 preview_present = bool(preview) and ( "bid_invalid" in preview or "levels" in preview or "covered_sheets" in preview or "liquidity_ok" in preview ) levels = preview.get("levels") if isinstance(preview.get("levels"), list) else [] top = levels[0] if levels and isinstance(levels[0], dict) else {} bid_px = top.get("px") if bid_px is None: bid_px = preview.get("raw_bid_px") bid_sz = top.get("available_sheets") try: uncovered = int(float(preview.get("uncovered_sheets") or 0)) if preview_present else 0 except (TypeError, ValueError): uncovered = 0 try: covered = int(float(preview.get("covered_sheets") or 0)) if preview_present else 0 except (TypeError, ValueError): covered = 0 bid_invalid = bool(preview.get("bid_invalid")) if preview_present else False bid_invalid_reason = preview.get("bid_invalid_reason") if preview_present else None # 残档/无效买一,或买一深度盖不住持仓 → 流动性不足;未同步时不标黄残档 if not preview_present: liq_ok = None else: liq_ok = (not bid_invalid) and uncovered <= 0 and ( preview.get("liquidity_ok") is not False ) if not levels and not bid_invalid and covered <= 0: # 已拉盘口但无可用买一档 liq_ok = False bid_invalid = True if not bid_invalid_reason: bid_invalid_reason = "无买一" pnl_usdt = p.get("display_pnl_usdt") if pnl_usdt is None: try: from lib.options.options_positions_lib import display_pnl_usdt_from_option_row pnl_usdt = display_pnl_usdt_from_option_row(p) except Exception: pnl_usdt = p.get("net_pnl_usdt") premium_paid = p.get("premium_paid") premium_usdt = None try: from lib.options.options_positions_lib import premium_paid_usdt_from_option_row premium_usdt = premium_paid_usdt_from_option_row(p) except Exception: premium_usdt = None if premium_usdt is not None: has_premium = True premium_total_usdt += float(premium_usdt) contract_leverage = p.get("contract_leverage") if contract_leverage is None: try: from lib.options.options_pricing_lib import equivalent_contract_leverage contract_leverage = equivalent_contract_leverage( index_px=p.get("idx_px") or p.get("index_px"), eth_amount=p.get("eth_amount"), total_premium=premium_paid, margin_mode=p.get("margin_mode"), ) except Exception: contract_leverage = None slim_pos.append( { "inst_id": p.get("inst_id"), "opt_type": p.get("opt_type"), "sheets": p.get("sheets") or p.get("pos"), "pnl_usdt": pnl_usdt, "premium_paid": premium_paid, "premium_paid_usdt": round(float(premium_usdt), 4) if premium_usdt is not None else None, "premium_ccy": p.get("premium_ccy"), "contract_leverage": contract_leverage, "mark": p.get("mark") or p.get("mark_px"), "underlying": p.get("underlying"), "source_label": p.get("source_label"), "target_index": p.get("target_index"), "profit_exit_enabled": p.get("profit_exit_enabled"), "profit_exit_mult": p.get("profit_exit_mult"), "profit_exit_state": p.get("profit_exit_state"), "target_monitor_text": p.get("target_monitor_text"), "bid_px": bid_px, "bid_sz": bid_sz, "covered_sheets": covered if preview_present else None, "uncovered_sheets": uncovered if preview_present else None, "bid_invalid": bid_invalid, "bid_invalid_reason": bid_invalid_reason, "liquidity_ok": liq_ok, } ) out["instance"] = { "label": label, "mode": mode, "options_enabled": bool(out.get("enabled")), "account_label": out.get("account_label") or "", "margin_mode": out.get("options_margin_mode") or "coin", "underly": out.get("options_underly") or "ETH", } funding_usdt = out.get("funding_usdt") trading_usdt = out.get("trading_usdt") funding_usdc = out.get("funding_usdc") trading_usdc = out.get("trading_usdc") out["funds"] = { "funding_usdt": funding_usdt, "trading_usdt": trading_usdt, "funding_usdc": funding_usdc, "trading_usdc": trading_usdc, } def _f(v: Any) -> float | None: if v is None or v == "": return None try: return float(v) except (TypeError, ValueError): return None # 账户可用余额:优先 USDT 资+交;都空则回退 USDC(≈U) cash_parts = [_f(trading_usdt), _f(funding_usdt)] if all(x is None for x in cash_parts): cash_parts = [_f(trading_usdc), _f(funding_usdc)] cash_known = [x for x in cash_parts if x is not None] cash_usdt = round(sum(cash_known), 4) if cash_known else None premium_usdt = round(premium_total_usdt, 4) if has_premium else (0.0 if not slim_pos else None) # 占比分母 = 当前余额 + 已付权利金(权利金已从余额扣出,需加回才是完整账户资金) equity_usdt = None if cash_usdt is not None or premium_usdt is not None: equity_usdt = round(float(cash_usdt or 0.0) + float(premium_usdt or 0.0), 4) ratio_pct = None if premium_usdt is not None and equity_usdt is not None and equity_usdt > 1e-9: ratio_pct = round(float(premium_usdt) / float(equity_usdt) * 100.0, 2) out["pnl"] = { "unrealized_usdt": out.get("upl_total_usdc"), } initial_cash_usdt = None try: import os if mode == "sim": initial_cash_usdt = float(os.getenv("SIM_INITIAL_EQUITY_USDT") or 10000) else: for key in ( "LIVE_INITIAL_EQUITY_USDT", "OPTIONS_INITIAL_EQUITY_USDT", "HUB_INITIAL_CASH_USDT", ): raw = (os.getenv(key) or "").strip() if not raw: continue initial_cash_usdt = float(raw) break if initial_cash_usdt is not None and initial_cash_usdt <= 0: initial_cash_usdt = None elif initial_cash_usdt is not None: initial_cash_usdt = round(float(initial_cash_usdt), 4) except (TypeError, ValueError): initial_cash_usdt = None out["premium"] = { "paid_usdt": premium_usdt, "cash_usdt": cash_usdt, "equity_usdt": equity_usdt, "ratio_pct": ratio_pct, "initial_cash_usdt": initial_cash_usdt, } out["positions_brief"] = slim_pos out["cond_order_locked"] = bool(out.get("cond_order_locked")) active_cond = out.get("cond_orders_active") out["cond_orders_active"] = active_cond if isinstance(active_cond, list) else [] try: from lib.options.options_time_lib import options_app_now_str out["ts"] = options_app_now_str() except Exception: from datetime import datetime out["ts"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") quotes = {"tickers": [], "updated_at": out.get("ts") or ""} try: from lib.market.nav_spot_tickers_lib import build_nav_spot_tickers ex = getattr(app_module, "exchange", None) reset_hour = int(getattr(app_module, "TRADING_DAY_RESET_HOUR", 8) or 8) tickers = build_nav_spot_tickers(ex, reset_hour=reset_hour) if ex is not None else [] quotes = { "tickers": tickers if isinstance(tickers, list) else [], "updated_at": out.get("ts") or "", } except Exception: pass out["quotes"] = quotes return out def register_hub_bridge_routes(app: Flask, app_module: Any) -> None: @app.route("/api/hub/options/snapshot") def api_hub_options_snapshot(): if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401 hub_ctx = app.config.get("HUB_CTX") or {} fn = hub_ctx.get("options_snapshot_fn") if not callable(fn): return jsonify( {"ok": False, "msg": "期权快照未注册", "version": 1, "enabled": False} ), 503 try: snap = fn() or {} except Exception as e: return jsonify({"ok": False, "msg": str(e), "version": 1, "enabled": True}), 500 return jsonify(_enrich_hub_protocol(snap, app_module=app_module)) @app.route("/api/hub/options/pnl_curve") def api_hub_options_pnl_curve(): """中控汇总用:累计已实现盈亏曲线(划转/提现不影响).""" if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401 hub_ctx = app.config.get("HUB_CTX") or {} cfg = hub_ctx.get("options_cfg") if not isinstance(cfg, dict) or not callable(cfg.get("get_db")): return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503 range_key = (request.args.get("range") or request.args.get("window") or "7d").strip() conn = cfg["get_db"]() try: from lib.instance.instance_cash_sample_lib import ( query_realized_pnl_series_with_db, ) out = query_realized_pnl_series_with_db(conn, range_key=range_key) out["version"] = 1 return jsonify(out) except Exception as e: return jsonify({"ok": False, "msg": str(e) or "曲线失败", "version": 1}), 500 finally: conn.close() @app.route("/api/hub/options/stats") def api_hub_options_stats(): """中控汇总用:窗口化已实现盈亏统计(不含浮盈进净盈亏主指标以外的展示字段仍附带).""" if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401 hub_ctx = app.config.get("HUB_CTX") or {} cfg = hub_ctx.get("options_cfg") if not isinstance(cfg, dict): return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503 from lib.options.options_margin_mode_lib import is_coin_margin_mode from lib.options.options_positions_lib import sum_options_net_pnl_usdc from lib.options.options_stats_lib import ( compute_options_stats_for_window, history_pnl_to_usdt, load_live_exchange_stats_history, load_local_options_stats_history, normalize_stats_window, ) window = normalize_stats_window(request.args.get("window") or "7d") ex = cfg.get("exchange_options") if ex is None: return jsonify({"ok": False, "msg": "交易所未就绪", "version": 1}), 503 try: raw_live = cfg["fetch_option_positions"](ex) if raw_live is None: return jsonify({"ok": False, "msg": "获取期权持仓失败", "version": 1}) use_sim = False try: from lib.sim.mode_lib import is_sim_mode use_sim = is_sim_mode(cfg["get_db"]) except Exception: use_sim = False if use_sim: conn = cfg["get_db"]() try: history = load_local_options_stats_history(conn, ex) try: conn.commit() except Exception: pass finally: conn.close() else: # 实盘:先网后短事务,禁止握着 DB 锁拉行情(否则中控整机超时) history = load_live_exchange_stats_history(cfg, ex) stats = compute_options_stats_for_window(history_pnl_to_usdt(history, ex), window) open_float = sum_options_net_pnl_usdc(cfg, ex, raw_live) net_realized = stats.get("net_realized_pnl") try: net_f = float(net_realized) if net_realized is not None else 0.0 except (TypeError, ValueError): net_f = 0.0 total_pnl = None if open_float is not None: total_pnl = round(net_f + float(open_float), 4) elif stats.get("total_closed"): total_pnl = round(net_f, 4) pnl_unit = "U" if is_coin_margin_mode() else "USDC" return jsonify( { "ok": True, "version": 1, "pnl_unit": pnl_unit, "open_float_pnl": open_float, "total_pnl": total_pnl, **stats, } ) except Exception as e: return jsonify({"ok": False, "msg": str(e) or "统计失败", "version": 1}), 500 @app.route("/api/hub/options/close", methods=["POST"]) def api_hub_options_close(): """中控下发全平指令:客户端本地买一限价执行;无效买一拒绝.""" if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401 data = request.get_json(silent=True) or {} if data.get("market"): return jsonify({"ok": False, "msg": "已禁用市价平仓,仅支持买一限价", "version": 1}) # 总览约定只全平,忽略/拒绝自定义张数 if data.get("sheets") is not None and str(data.get("sheets")).strip() != "": return jsonify({"ok": False, "msg": "中控平仓仅支持全平,勿传 sheets", "version": 1}) inst_id = (data.get("inst_id") or "").strip() if not inst_id: return jsonify({"ok": False, "msg": "缺少 inst_id", "version": 1}) hub_ctx = app.config.get("HUB_CTX") or {} fn = hub_ctx.get("options_close_fn") if not callable(fn): return jsonify({"ok": False, "msg": "平仓接口未注册", "version": 1}), 503 try: result = fn(inst_id) or {} except Exception as e: return jsonify({"ok": False, "msg": str(e) or "平仓失败", "version": 1}), 500 if not isinstance(result, dict): return jsonify({"ok": False, "msg": "平仓返回异常", "version": 1}), 500 out = dict(result) out["version"] = 1 return jsonify(out) @app.route("/api/hub/options/cond_orders", methods=["GET"]) def api_hub_options_cond_orders_list(): if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401 hub_ctx = app.config.get("HUB_CTX") or {} cfg = hub_ctx.get("options_cfg") if not isinstance(cfg, dict): return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503 from lib.options.options_cond_order_lib import list_cond_orders from lib.options.options_db import init_options_tables conn = cfg["get_db"]() try: init_options_tables(conn) orders = list_cond_orders(conn, active_only=False) return jsonify({"ok": True, "version": 1, "orders": orders}) finally: conn.close() @app.route("/api/hub/options/cond_orders", methods=["POST"]) def api_hub_options_cond_orders_create(): if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401 hub_ctx = app.config.get("HUB_CTX") or {} cfg = hub_ctx.get("options_cfg") if not isinstance(cfg, dict): return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503 from lib.options.options_cond_order_lib import ( create_cond_order, machine_has_option_position, ) from lib.options.options_db import init_options_tables from lib.options.options_position_limit_lib import option_position_limit_block_msg ex = cfg.get("exchange_options") if machine_has_option_position(cfg, ex): return jsonify({"ok": False, "msg": "本机已有期权仓,拒绝条件单", "version": 1}) limit_msg = option_position_limit_block_msg( ex, opening_inst_id="", fetch_positions=cfg.get("fetch_option_positions"), ) if limit_msg: return jsonify({"ok": False, "msg": limit_msg, "version": 1}) data = request.get_json(silent=True) or {} conn = cfg["get_db"]() try: init_options_tables(conn) out = create_cond_order( conn, data, send_wechat=cfg.get("send_wechat"), account_label=str(cfg.get("account_label") or ""), ) out["version"] = 1 return jsonify(out) finally: conn.close() @app.route("/api/hub/options/cond_orders//cancel", methods=["POST"]) def api_hub_options_cond_orders_cancel(hub_order_id: str): if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401 hub_ctx = app.config.get("HUB_CTX") or {} cfg = hub_ctx.get("options_cfg") if not isinstance(cfg, dict): return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503 from lib.options.options_cond_order_lib import cancel_cond_order from lib.options.options_db import init_options_tables conn = cfg["get_db"]() try: init_options_tables(conn) out = cancel_cond_order( conn, hub_order_id, send_wechat=cfg.get("send_wechat"), account_label=str(cfg.get("account_label") or ""), ) out["version"] = 1 return jsonify(out) finally: conn.close() @app.route("/api/hub/options/cond_orders//exit", methods=["POST"]) def api_hub_options_cond_orders_exit(hub_order_id: str): if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401 hub_ctx = app.config.get("HUB_CTX") or {} cfg = hub_ctx.get("options_cfg") if not isinstance(cfg, dict): return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503 from lib.options.options_cond_order_lib import update_cond_order_exit from lib.options.options_db import init_options_tables data = request.get_json(silent=True) or {} conn = cfg["get_db"]() try: init_options_tables(conn) out = update_cond_order_exit( conn, hub_order_id, target_index=data.get("target_index"), profit_exit_enabled=data.get("profit_exit_enabled"), profit_exit_mult=data.get("profit_exit_mult"), ) out["version"] = 1 return jsonify(out) finally: conn.close() @app.route("/api/hub/version", methods=["GET"]) def hub_version(): """中控探测客户端当前 git 版本/在线.""" if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权"}), 401 from lib.instance.instance_update_lib import read_git_version ver = read_git_version() ver["online"] = True mode = "live" try: from lib.sim.mode_lib import exchange_mode_label, is_sim_mode get_db = getattr(app_module, "get_db", None) if callable(get_db) and is_sim_mode(get_db): mode = "sim" ver["mode"] = mode if callable(get_db): ver["mode_label"] = exchange_mode_label(get_db=get_db, exchange_key="okx") or "" except Exception: ver["mode"] = mode return jsonify(ver) @app.route("/api/hub/update", methods=["POST"]) def hub_update(): """中控远程快速更新:git pull + pm2 restart(异步).""" if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权"}), 401 from lib.instance.instance_update_lib import schedule_pull_and_restart out = schedule_pull_and_restart(delay_seconds=1.5) return jsonify(out), (200 if out.get("ok") else 400) @app.route("/api/hub/backup/list", methods=["GET"]) def hub_backup_list(): if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权"}), 401 from lib.instance.backup_lib import backup_dir, backup_keep_days, list_backups return jsonify( { "ok": True, "items": list_backups(), "backup_dir": backup_dir(), "keep_days": backup_keep_days(), "auto_hour_bj": 3, "includes": [".env", "database", "options_journal"], } ) @app.route("/api/hub/backup/download/", methods=["GET"]) def hub_backup_download(name: str): if not hub_bridge_token_ok(request.headers): return jsonify({"ok": False, "msg": "未授权"}), 401 import os from flask import send_file from lib.instance.backup_lib import backup_path path = 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("/hub_sso") def hub_sso_login(): token = (request.args.get("token") or "").strip() next_path = _safe_next_path(request.args.get("next") or "/embed") if not verify_sso_token(token): return ( "SSO 失败" "

中控登录链无效或已过期,请从中控重新打开.

", 403, ) session["logged_in"] = True session.permanent = True session.modified = True # iframe 跨站时常丢 Cookie:附带可放入 sessionStorage 的 embed_sso embed_tok = make_embed_session_token(hub_bridge_token(), ttl_sec=604800) joiner = "&" if "?" in next_path else "?" return redirect(f"{next_path}{joiner}embed_sso={quote(embed_tok, safe='')}") @app.after_request def _hub_frame_ancestors(resp): ancestors = frame_ancestors_header_value() if ancestors: resp.headers["Content-Security-Policy"] = f"frame-ancestors {ancestors}" resp.headers.pop("X-Frame-Options", None) return resp