crypto_monitor/lib/hub/hub_bridge.py

"""
各 crypto_monitor_* 注册 /api/hub/* JSON 接口,供 manual_trading_hub 调用.
实例末尾:app.config["HUB_CTX"] = {...}; register_hub_routes(app)
"""

from __future__ import annotations

import json
import time
from functools import wraps

from flask import (
    current_app,
    flash,
    get_flashed_messages,
    jsonify,
    redirect,
    request,
    session,
)

from lib.hub.hub_auth import request_allowed
from lib.hub.hub_sso import (
    mint_hub_embed_bootstrap,
    safe_next_path,
    verify_hub_embed_bootstrap,
    verify_hub_sso_token,
)


def _merge_query_into_path(path: str, **params: str) -> str:
    from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

    split = urlsplit(path or "/")
    q = list(parse_qsl(split.query, keep_blank_values=True))
    keys = {k for k, _ in q}
    for k, v in params.items():
        if not v or k in keys:
            continue
        q.append((k, str(v)))
    return urlunsplit((split.scheme, split.netloc, split.path, urlencode(q), split.fragment))


def install_instance_theme_static(app) -> None:
    """仓库 lib/common/static 下 instance_theme.* 等供三所页面共用."""
    import os

    from flask import Response, send_file

    from lib.paths import common_static_dir

    repo_static = common_static_dir()
    assets = {
        "instance_theme.js": "application/javascript; charset=utf-8",
        "instance_theme_early.css": "text/css; charset=utf-8",
        "instance_theme.css": "text/css; charset=utf-8",
        "account_risk_badge.css": "text/css; charset=utf-8",
        "account_risk_badge.js": "application/javascript; charset=utf-8",
        "instance_ui.js": "application/javascript; charset=utf-8",
        "instance_records_mobile.js": "application/javascript; charset=utf-8",
        "records_review_page.js": "application/javascript; charset=utf-8",
        "ai_review_render.js": "application/javascript; charset=utf-8",
        "form_submit_guard.js": "application/javascript; charset=utf-8",
        "autofill_guard.js": "application/javascript; charset=utf-8",
        "key_monitor_form.js": "application/javascript; charset=utf-8",
        "time_close_ui.js": "application/javascript; charset=utf-8",
        "manual_order_rr_preview.js": "application/javascript; charset=utf-8",
        "symbol_live_price.js": "application/javascript; charset=utf-8",
        "journal_upload_slots.js": "application/javascript; charset=utf-8",
        "journal_form_save.js": "application/javascript; charset=utf-8",
        "strategy_roll.js": "application/javascript; charset=utf-8",
        "instance_page.css": "text/css; charset=utf-8",
        "instance_embed.js": "application/javascript; charset=utf-8",
        "instance_mobile_nav.js": "application/javascript; charset=utf-8",
        "instance_stats.js": "application/javascript; charset=utf-8",
        "instance_live.js": "application/javascript; charset=utf-8",
        "instance_settings_prefs.js": "application/javascript; charset=utf-8",
        "instance_dashboard.js": "application/javascript; charset=utf-8",
        "account_ledger.js": "application/javascript; charset=utf-8",
        "options_expiry_countdown.js": "application/javascript; charset=utf-8",
        "options_panel.js": "application/javascript; charset=utf-8",
        "order_entry_model.js": "application/javascript; charset=utf-8",
        "focus_chart_page.js": "application/javascript; charset=utf-8",
        "focus_chart_page.css": "text/css; charset=utf-8",
        "trade_stats_calendar.js": "application/javascript; charset=utf-8",
        "trade_stats_calendar.css": "text/css; charset=utf-8",
    }

    for name, mime in assets.items():
        path = os.path.join(repo_static, name)

        def _view(p=path, m=mime):
            if not os.path.isfile(p):
                return Response("not found", status=404, mimetype="text/plain; charset=utf-8")
            return send_file(p, mimetype=m)

        app.add_url_rule(
            f"/static/{name}",
            endpoint=f"repo_static_{name.replace('.', '_')}",
            view_func=_view,
        )


def register_trade_stats_calendar_route(
    app,
    *,
    login_required_fn,
    load_pnls_fn,
    row_matches_segment_fn,
    reset_hour: int,
    get_db_fn=None,
):
    """三所统计分析页:按月返回各交易日盈亏/笔数."""
    from flask import jsonify, request

    from lib.trade.trade_stats_calendar_lib import build_trade_stats_calendar

    @app.route("/api/stats/calendar")
    @login_required_fn
    def api_stats_calendar():
        year = request.args.get("year", type=int)
        month = request.args.get("month", type=int)
        segment = (request.args.get("segment") or "all").strip() or "all"
        if not year or not month:
            from datetime import datetime

            now = datetime.now()
            year = year or now.year
            month = month or now.month
        get_db = get_db_fn or (app.config.get("HUB_CTX") or {}).get("get_db")
        if not get_db:
            return jsonify({"ok": False, "msg": "未配置数据库"}), 500
        conn = get_db()
        try:
            pnls = load_pnls_fn(conn)
        finally:
            conn.close()
        try:
            payload = build_trade_stats_calendar(
                pnls,
                year,
                month,
                segment,
                row_matches_segment_fn,
                reset_hour=int(reset_hour),
            )
        except ValueError as exc:
            return jsonify({"ok": False, "msg": str(exc)}), 400
        return jsonify({"ok": True, **payload})


def _hub_auth_required(f):
    @wraps(f)
    def wrapped(*args, **kwargs):
        from flask import current_app as cap

        auth_disabled = bool(cap.config.get("HUB_AUTH_DISABLED"))
        if not request_allowed(bool(session.get("logged_in")), auth_disabled):
            return jsonify({"ok": False, "msg": "未授权(登录或 HUB_BRIDGE_TOKEN)"}), 401
        return f(*args, **kwargs)

    return wrapped


def _ctx():
    return current_app.config.get("HUB_CTX") or {}


def _row_to_dict(row):
    fn = _ctx().get("row_to_dict")
    if fn and row is not None:
        return fn(row)
    return dict(row) if row is not None else {}


def build_hub_monitor_payload(
    *,
    keys,
    orders,
    trends,
    rolls,
    hedges=None,
    enrich=None,
    risk_status=None,
) -> dict:
    """合并 enrich 增量字段;enrich 只返回 trends 等局部时不得丢掉 keys/orders."""
    payload = {
        "ok": True,
        "keys": keys,
        "orders": orders,
        "trends": trends,
        "rolls": rolls,
        "hedges": hedges if isinstance(hedges, list) else [],
        "key_prices": [],
    }
    if isinstance(risk_status, dict):
        payload["risk_status"] = risk_status
    if callable(enrich):
        extra = enrich(keys=keys, orders=orders, trends=trends, rolls=rolls)
        if isinstance(extra, dict):
            payload.update(extra)
            # enrich 可能不返回 hedges,保留本地组装的对冲列表.
            if "hedges" not in extra:
                payload["hedges"] = hedges if isinstance(hedges, list) else []
    return payload


_FAIL_HINTS = (
    "失败",
    "错误",
    "拒绝",
    "无效",
    "缺少",
    "无法",
    "过期",
    "未达",
    "不能为空",
    "已有",
    "不允许",
    "异常",
)


def _invoke_view(view_name: str, path: str, form=None) -> dict:
    views = _ctx().get("views") or {}
    view = views.get(view_name)
    if not view:
        return {"ok": False, "messages": [f"未配置视图 {view_name}"]}
    data = form if form is not None else request.form
    if hasattr(data, "items") and not isinstance(data, dict):
        data = {k: v for k, v in data.items()}
    with current_app.test_request_context(path, method="POST", data=data):
        session["logged_in"] = True
        try:
            view()
        except Exception as e:
            return {"ok": False, "messages": [str(e)]}
        try:
            msgs = [str(x) for x in get_flashed_messages()]
        except Exception as e:
            return {"ok": False, "messages": [f"读取提示信息失败: {e}"]}
        ok = True
        for m in msgs:
            if any(k in m for k in _FAIL_HINTS):
                ok = False
                break
        return {"ok": ok, "messages": msgs}


def _invoke_view_get(view_name: str, path: str) -> dict:
    views = _ctx().get("views") or {}
    view = views.get(view_name)
    if not view:
        return {"ok": False, "messages": [f"未配置视图 {view_name}"]}
    with current_app.test_request_context(path, method="GET"):
        session["logged_in"] = True
        try:
            view()
        except Exception as e:
            return {"ok": False, "messages": [str(e)]}
        try:
            msgs = [str(x) for x in get_flashed_messages()]
        except Exception as e:
            return {"ok": False, "messages": [f"读取提示信息失败: {e}"]}
        ok = True
        for m in msgs:
            if any(k in m for k in _FAIL_HINTS):
                ok = False
                break
        return {"ok": ok, "messages": msgs}


def _hub_json(view_name: str, path: str, form=None):
    try:
        return jsonify(_invoke_view(view_name, path, form=form))
    except Exception as e:
        return jsonify({"ok": False, "messages": [str(e)]})


def _embed_login_dest(next_path: str) -> str:
    """embed=1 时把 /trade 等映射到 /embed?tab=…"""
    ht = (request.args.get("hub_theme") or "").strip().lower()
    hub_theme = ht if ht in ("light", "dark") else None
    if request.args.get("embed", "").strip().lower() in ("1", "true", "yes", "on"):
        from lib.instance.instance_embed_lib import rewrite_embed_dest

        return rewrite_embed_dest(next_path, hub_theme=hub_theme)
    if hub_theme:
        return _merge_query_into_path(next_path, hub_theme=hub_theme)
    return next_path


def install_on_app(
    app,
    *,
    exchange: str,
    capabilities: list,
    has_trend: bool,
    get_db,
    row_to_dict,
    meta_fn,
    views: dict,
    ohlcv_fn=None,
    account_fn=None,
    volume_rank_fn=None,
    market_fn=None,
    reconcile_hub_flat_fn=None,
    risk_status_fn=None,
    user_close_fn=None,
    render_main_page_fn=None,
    login_required_fn=None,
):
    app.config["HUB_CTX"] = {
        "exchange": exchange,
        "capabilities": list(capabilities),
        "has_trend": bool(has_trend),
        "get_db": get_db,
        "row_to_dict": row_to_dict,
        "meta_fn": meta_fn,
        "account_fn": account_fn,
        "views": views,
        "ohlcv_fn": ohlcv_fn,
        "volume_rank_fn": volume_rank_fn,
        "market_fn": market_fn,
        "reconcile_hub_flat_fn": reconcile_hub_flat_fn,
        "risk_status_fn": risk_status_fn,
        "user_close_fn": user_close_fn,
    }
    install_hub_embed_headers(app)
    configure_hub_embed_session(app)
    install_instance_theme_static(app)
    register_hub_routes(app)
    if render_main_page_fn and login_required_fn:
        from lib.instance.instance_embed_lib import attach_embed_templates, register_embed_routes
        from lib.paths import REPO_ROOT

        attach_embed_templates(app, str(REPO_ROOT))
        register_embed_routes(app, login_required_fn, render_main_page_fn)


def configure_hub_embed_session(app):
    """HTTPS iframe 内嵌须 SameSite=None + Secure;hub-sso / hub-embed-auth 自动启用."""
    import os

    allowed = (os.getenv("APP_ALLOW_HUB_EMBED") or "true").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    )
    if not allowed:
        return

    secure_env = (os.getenv("APP_COOKIE_SECURE") or "auto").strip().lower()
    if secure_env in ("1", "true", "yes", "on"):
        app.config.update(
            SESSION_COOKIE_SECURE=True,
            SESSION_COOKIE_SAMESITE="None",
            SESSION_COOKIE_HTTPONLY=True,
        )
        return

    @app.before_request
    def _hub_embed_session_cookie():
        if request.path not in ("/hub-sso", "/hub-embed-auth"):
            return
        embed = (request.args.get("embed") or "").strip().lower() in (
            "1",
            "true",
            "yes",
            "on",
        )
        in_iframe = (request.headers.get("Sec-Fetch-Dest") or "").lower() == "iframe"
        if not embed and not in_iframe:
            return
        if not request.is_secure:
            return
        app.config["SESSION_COOKIE_SECURE"] = True
        app.config["SESSION_COOKIE_SAMESITE"] = "None"
        app.config["SESSION_COOKIE_HTTPONLY"] = True


def _sso_wants_embed_auth() -> bool:
    embed = (request.args.get("embed") or "").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    )
    in_iframe = (request.headers.get("Sec-Fetch-Dest") or "").lower() == "iframe"
    return bool(embed or in_iframe)


def install_hub_embed_headers(app):
    """允许复盘中控 iframe 内嵌打开本实例(须与 hub 的 HUB_EMBED_ORIGINS 或域名一致)."""
    import os

    allowed = (os.getenv("APP_ALLOW_HUB_EMBED") or "true").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    )
    if not allowed:
        return
    origins = (
        (os.getenv("HUB_EMBED_PARENT_ORIGINS") or os.getenv("HUB_EMBED_ORIGINS") or "*")
        .strip()
    )

    @app.after_request
    def _hub_embed_frame_headers(response):
        if origins == "*":
            response.headers["Content-Security-Policy"] = "frame-ancestors *"
        else:
            response.headers["Content-Security-Policy"] = (
                f"frame-ancestors 'self' {origins}"
            )
        return response


def register_hub_routes(app):
    auth_disabled = False
    try:
        import os

        auth_disabled = os.getenv("APP_AUTH_DISABLED", "false").lower() in (
            "1",
            "true",
            "yes",
            "on",
        )
    except Exception:
        pass
    app.config.setdefault("HUB_AUTH_DISABLED", auth_disabled)

    @app.route("/api/hub/ping")
    @_hub_auth_required
    def api_hub_ping():
        c = _ctx()
        return jsonify(
            {
                "ok": True,
                "exchange": c.get("exchange"),
                "capabilities": c.get("capabilities") or [],
            }
        )

    @app.route("/api/hub/meta")
    @_hub_auth_required
    def api_hub_meta():
        c = _ctx()
        meta_fn = c.get("meta_fn")
        meta = meta_fn() if callable(meta_fn) else {}
        return jsonify({"ok": True, "meta": meta})

    @app.route("/api/hub/account")
    @_hub_auth_required
    def api_hub_account():
        """中控 AI:资金账户 / 交易账户余额(无需浏览器登录)."""
        fn = _ctx().get("account_fn")
        if not callable(fn):
            return jsonify({"ok": False, "msg": "未配置 account_fn"}), 501
        try:
            data = fn()
            if not isinstance(data, dict):
                data = {}
            return jsonify({"ok": True, **data})
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500

    @app.route("/api/hub/options/snapshot")
    @_hub_auth_required
    def api_hub_options_snapshot():
        """中控监控:期权持仓 / 资金 / 本地统计(只读)."""
        fn = _ctx().get("options_snapshot_fn")
        if not callable(fn):
            return jsonify({"ok": True, "enabled": False})
        try:
            data = fn()
            if not isinstance(data, dict):
                data = {"ok": False, "enabled": True, "msg": "invalid snapshot"}
            return jsonify(data)
        except Exception as e:
            return jsonify({"ok": False, "enabled": True, "msg": str(e)}), 500

    @app.route("/api/account_risk_status")
    @_hub_auth_required
    def api_account_risk_status():
        c = _ctx()
        get_db = c.get("get_db")
        risk_fn = c.get("risk_status_fn")
        if not callable(get_db) or not callable(risk_fn):
            return jsonify({"ok": False, "msg": "未配置风控"}), 501
        conn = get_db()
        try:
            payload = risk_fn(conn)
            return jsonify({"ok": True, **(payload if isinstance(payload, dict) else {})})
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500
        finally:
            conn.close()

    @app.route("/api/hub/account-risk/user-close", methods=["POST"])
    @_hub_auth_required
    def api_hub_account_risk_user_close():
        """中控/实例:登记用户主动平仓(计入冷静期与日冻结)."""
        c = _ctx()
        get_db = c.get("get_db")
        user_close_fn = c.get("user_close_fn")
        if not callable(get_db) or not callable(user_close_fn):
            return jsonify({"ok": False, "msg": "未配置 user_close_fn"}), 501
        body = request.get_json(silent=True) or {}
        source = (body.get("source") or request.form.get("source") or "").strip()
        try:
            count = max(0, int(body.get("count") if body.get("count") is not None else 1))
        except (TypeError, ValueError):
            count = 1
        trade_record_id = body.get("trade_record_id")
        closed_at_ms = body.get("closed_at_ms")
        if count <= 0:
            return jsonify({"ok": True, "skipped": True, "count": 0})
        conn = get_db()
        try:
            user_close_fn(
                conn,
                source=source,
                count=count,
                trade_record_id=trade_record_id,
                closed_at_ms=closed_at_ms,
            )
            conn.commit()
            return jsonify({"ok": True, "count": count, "source": source})
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500
        finally:
            conn.close()

    @app.route("/api/hub/monitor")
    @_hub_auth_required
    def api_hub_monitor():
        c = _ctx()
        get_db = c.get("get_db")
        if not get_db:
            return jsonify({"ok": False, "msg": "HUB_CTX 缺少 get_db"}), 500
        conn = get_db()
        keys = []
        for row in conn.execute("SELECT * FROM key_monitors ORDER BY id DESC").fetchall():
            keys.append(_row_to_dict(row))
        orders = []
        for row in conn.execute(
            "SELECT * FROM order_monitors WHERE status='active' ORDER BY id DESC"
        ).fetchall():
            od = _row_to_dict(row)
            try:
                from lib.strategy.strategy_trade_labels import apply_order_monitor_source_labels

                od = apply_order_monitor_source_labels(od)
            except Exception:
                pass
            try:
                from lib.trade.entry_model_lib import enrich_entry_model_display

                enrich_entry_model_display(od)
            except Exception:
                pass
            orders.append(od)
        trends = []
        if c.get("has_trend"):
            for row in conn.execute(
                "SELECT * FROM trend_pullback_plans WHERE status='active' ORDER BY id DESC"
            ).fetchall():
                trends.append(_row_to_dict(row))
        rolls = []
        try:
            for row in conn.execute(
                """SELECT g.* FROM roll_groups g
                   INNER JOIN order_monitors m ON m.id = g.order_monitor_id AND m.status='active'
                   WHERE g.status='active' ORDER BY g.id DESC"""
            ).fetchall():
                rolls.append(_row_to_dict(row))
        except Exception:
            pass
        hedges = []
        try:
            from lib.hedge_plan.hedge_plan_db import attach_legs_to_plans, list_plans

            hedge_rows: list = []
            for st in ("opening", "active", "partial"):
                hedge_rows.extend(list_plans(conn, status=st, limit=80))
            hedge_rows.sort(key=lambda row: int(row.get("id") or 0), reverse=True)
            hedges = attach_legs_to_plans(conn, hedge_rows)
        except Exception:
            hedges = []
        risk_status = None
        risk_fn = c.get("risk_status_fn")
        if callable(risk_fn):
            try:
                risk_status = risk_fn(conn)
            except Exception:
                risk_status = None
        conn.close()
        enrich = c.get("enrich_monitor")
        if callable(enrich):
            try:
                return jsonify(
                    build_hub_monitor_payload(
                        keys=keys,
                        orders=orders,
                        trends=trends,
                        rolls=rolls,
                        hedges=hedges,
                        enrich=enrich,
                        risk_status=risk_status,
                    )
                )
            except Exception as e:
                return jsonify({"ok": False, "msg": str(e)}), 500
        return jsonify(
            build_hub_monitor_payload(
                keys=keys,
                orders=orders,
                trends=trends,
                rolls=rolls,
                hedges=hedges,
                risk_status=risk_status,
            )
        )

    @app.route("/api/hub/trades/archive")
    @_hub_auth_required
    def api_hub_trades_archive():
        """中控币种档案:近 N 天已平仓记录."""
        from lib.hub.hub_trades_lib import fetch_trades_for_archive, summarize_trades

        c = _ctx()
        get_db = c.get("get_db")
        if not get_db:
            return jsonify({"ok": False, "msg": "HUB_CTX 缺少 get_db"}), 500
        try:
            days = int(request.args.get("days") or "365")
        except ValueError:
            days = 365
        try:
            limit = int(request.args.get("limit") or "2000")
        except ValueError:
            limit = 2000
        try:
            import os

            reset_hour = int(os.getenv("TRADING_DAY_RESET_HOUR", "8") or "8")
        except ValueError:
            reset_hour = 8
        conn = get_db()
        try:
            trades = fetch_trades_for_archive(
                conn,
                exchange_key=str(c.get("exchange") or ""),
                days=days,
                row_to_dict_fn=c.get("row_to_dict"),
                reset_hour=reset_hour,
                limit=limit,
            )
        finally:
            conn.close()
        stats = summarize_trades(trades)
        return jsonify(
            {
                "ok": True,
                "days": max(1, min(days, 3650)),
                "trading_day_reset_hour": reset_hour,
                "trades": trades,
                "stats": stats,
            }
        )

    @app.route("/api/hub/options/review/archive")
    @_hub_auth_required
    def api_hub_options_review_archive():
        """中控期权档案:近 N 天已平仓复盘记录(默认排除对冲腿)."""
        from datetime import datetime, timedelta
        from zoneinfo import ZoneInfo

        from flask import current_app

        from lib.options.options_review_lib import (
            compute_review_stats,
            ensure_local_review_synced,
            list_review_trades,
        )

        c = _ctx()
        get_db = c.get("get_db")
        if not get_db:
            return jsonify({"ok": False, "msg": "HUB_CTX 缺少 get_db"}), 500
        try:
            days = int(request.args.get("days") or "365")
        except ValueError:
            days = 365
        days = max(1, min(days, 3650))
        try:
            limit = int(request.args.get("limit") or "2000")
        except ValueError:
            limit = 2000
        limit = max(1, min(limit, 5000))
        include_hedge_legs = str(request.args.get("include_hedge_legs") or "").strip() in (
            "1",
            "true",
            "yes",
        )
        tz = ZoneInfo("Asia/Shanghai")
        closed_from = (datetime.now(tz) - timedelta(days=days)).strftime("%Y-%m-%d")
        cfg = (current_app.extensions or {}).get("options_cfg") or {}
        ex = cfg.get("exchange_options")
        conn = get_db()
        try:
            ensure_local_review_synced(conn, ex=ex, backfill_exchange_pnl=bool(ex))
            trades = list_review_trades(
                conn,
                include_hedge_legs=include_hedge_legs,
                closed_from=closed_from,
                limit=limit,
                offset=0,
            )
            stats = compute_review_stats(
                conn,
                include_hedge_legs=include_hedge_legs,
                closed_from=closed_from,
            )
        finally:
            conn.close()
        return jsonify(
            {
                "ok": True,
                "days": days,
                "limit": limit,
                "product": "options",
                "trades": trades,
                "stats": stats,
            }
        )

    @app.route("/api/hub/trades/today")
    @_hub_auth_required
    def api_hub_trades_today():
        """中控 AI:当日已平仓记录(按实例交易日)."""
        from lib.hub.hub_trades_lib import (
            current_trading_day,
            fetch_trades_for_trading_day,
            summarize_trades,
        )
        from lib.trade.daily_open_limit_lib import count_opens_for_trading_day

        c = _ctx()
        get_db = c.get("get_db")
        if not get_db:
            return jsonify({"ok": False, "msg": "HUB_CTX 缺少 get_db"}), 500
        day_arg = (request.args.get("trading_day") or request.args.get("date") or "").strip()[:10]
        try:
            import os

            reset_hour = int(os.getenv("TRADING_DAY_RESET_HOUR", "8") or "8")
        except ValueError:
            reset_hour = 8
        trading_day = day_arg or current_trading_day(reset_hour=reset_hour)
        conn = get_db()
        try:
            trades = fetch_trades_for_trading_day(
                conn,
                trading_day,
                row_to_dict_fn=c.get("row_to_dict"),
                reset_hour=reset_hour,
            )
            opens_today = count_opens_for_trading_day(conn, trading_day)
        finally:
            conn.close()
        stats = summarize_trades(trades)
        return jsonify(
            {
                "ok": True,
                "trading_day": trading_day,
                "trading_day_reset_hour": reset_hour,
                "opens_today": opens_today,
                "trades": trades,
                "stats": stats,
            }
        )

    @app.route("/api/hub/volume-rank")
    @_hub_auth_required
    def api_hub_volume_rank():
        fn = _ctx().get("volume_rank_fn")
        if not callable(fn):
            return jsonify({"ok": False, "msg": "该实例未配置成交量排名接口"}), 501
        top_raw = (request.args.get("top") or "").strip()
        top_n = 20
        if top_raw.isdigit():
            top_n = int(top_raw)
        try:
            result = fn(top_n=top_n)
            if isinstance(result, dict):
                return jsonify(result)
            return jsonify({"ok": False, "msg": "成交量排名返回格式无效"}), 500
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500

    @app.route("/api/hub/market")
    @_hub_auth_required
    def api_hub_market():
        fn = _ctx().get("market_fn")
        if not callable(fn):
            return jsonify({"ok": False, "msg": "该实例未配置合约信息接口"}), 501
        base = (request.args.get("base") or request.args.get("symbol") or "").strip()
        try:
            result = fn(base=base)
            if isinstance(result, dict):
                return jsonify(result)
            return jsonify({"ok": False, "msg": "合约信息返回格式无效"}), 500
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500

    @app.route("/api/hub/ohlcv")
    @_hub_auth_required
    def api_hub_ohlcv():
        fn = _ctx().get("ohlcv_fn")
        if not callable(fn):
            return jsonify({"ok": False, "msg": "该实例未配置 OHLCV 接口"}), 501
        symbol = (request.args.get("symbol") or "").strip()
        timeframe = (request.args.get("timeframe") or "5m").strip()
        since_raw = (request.args.get("since_ms") or "").strip()
        limit_raw = (request.args.get("limit") or "").strip()
        since_ms = None
        if since_raw.isdigit():
            since_ms = int(since_raw)
        limit = 500
        if limit_raw.isdigit():
            limit = int(limit_raw)
        try:
            result = fn(symbol=symbol, timeframe=timeframe, since_ms=since_ms, limit=limit)
            if isinstance(result, dict):
                return jsonify(result)
            return jsonify({"ok": False, "msg": "OHLCV 返回格式无效"}), 500
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500

    @app.route("/api/hub/add_order", methods=["POST"])
    @_hub_auth_required
    def api_hub_add_order():
        return _hub_json("add_order", "/add_order")

    @app.route("/api/hub/add_key", methods=["POST"])
    @_hub_auth_required
    def api_hub_add_key():
        return _hub_json("add_key", "/add_key")

    @app.route("/api/hub/trend/preview", methods=["POST"])
    @_hub_auth_required
    def api_hub_trend_preview():
        if not _ctx().get("has_trend"):
            return jsonify({"ok": False, "msg": "该实例无趋势回调"}), 400
        data = _invoke_view("preview_trend_pullback", "/trade")
        pid = _latest_preview_id()
        preview = _fetch_preview(pid) if pid else None
        return jsonify(
            {
                "ok": bool(data.get("ok")),
                "messages": data.get("messages") or [],
                "preview_id": pid,
                "preview": preview,
            }
        )

    @app.route("/api/hub/trend/execute", methods=["POST"])
    @_hub_auth_required
    def api_hub_trend_execute():
        if not _ctx().get("has_trend"):
            return jsonify({"ok": False, "msg": "该实例无趋势回调"}), 400
        pid = (request.form.get("preview_id") or "").strip()
        if not pid:
            body = request.get_json(silent=True) or {}
            pid = str(body.get("preview_id") or "").strip()
        form = {"preview_id": pid} if pid else {}
        return jsonify(_invoke_view("execute_trend_pullback", "/trade", form=form))

    @app.route("/api/hub/trend/preview/<pid>")
    @_hub_auth_required
    def api_hub_trend_preview_get(pid):
        if not _ctx().get("has_trend"):
            return jsonify({"ok": False, "msg": "该实例无趋势回调"}), 400
        preview = _fetch_preview(pid)
        if not preview:
            return jsonify({"ok": False, "msg": "预览不存在或已过期"}), 404
        return jsonify({"ok": True, "preview": preview})

    @app.route("/api/hub/trend/stop/<int:pid>", methods=["POST"])
    @_hub_auth_required
    def api_hub_trend_stop(pid):
        if not _ctx().get("has_trend"):
            return jsonify({"ok": False, "msg": "该实例无趋势回调"}), 400
        return jsonify(_invoke_view_get("stop_trend_pullback", f"/stop_trend_pullback/{pid}"))

    @app.route("/api/hub/order/sync-tpsl", methods=["POST"])
    @_hub_auth_required
    def api_hub_order_sync_tpsl():
        """中控 agent 已挂 TP/SL 后:同步 order_monitors 计划价,避免刷新仍显示旧止损止盈."""
        body = request.get_json(silent=True) or {}
        symbol = (body.get("symbol") or request.form.get("symbol") or "").strip()
        side = (
            body.get("side")
            or body.get("direction")
            or request.form.get("side")
            or ""
        ).strip().lower()
        if not symbol:
            return jsonify({"ok": False, "msg": "symbol 不能为空"}), 400
        if side not in ("long", "short"):
            return jsonify({"ok": False, "msg": "side 须为 long 或 short"}), 400
        try:
            sl = float(body.get("stop_loss"))
            tp = float(body.get("take_profit"))
        except (TypeError, ValueError):
            return jsonify({"ok": False, "msg": "stop_loss / take_profit 须为数字"}), 400
        get_db = _ctx().get("get_db")
        if not callable(get_db):
            return jsonify({"ok": False, "msg": "HUB_CTX 缺少 get_db"}), 500
        from lib.hub.hub_symbol_lib import symbols_match
        from lib.hub.hub_order_sync_lib import sync_active_monitor_tpsl_prices

        conn = get_db()
        try:
            out = sync_active_monitor_tpsl_prices(
                conn, symbol, side, sl, tp, symbols_match=symbols_match
            )
            if out.get("ok"):
                conn.commit()
            return jsonify(out)
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500
        finally:
            conn.close()

    @app.route("/api/hub/order/sync-flat", methods=["POST"])
    @_hub_auth_required
    def api_hub_order_sync_flat():
        """中控市价全平后:同步 order_monitors 并读 Gate 平仓历史写交易记录."""
        fn = _ctx().get("reconcile_hub_flat_fn")
        if not callable(fn):
            return jsonify({"ok": False, "msg": "该实例未配置 order sync-flat"}), 400
        body = request.get_json(silent=True) or {}
        symbol = (body.get("symbol") or request.form.get("symbol") or "").strip()
        side = (
            body.get("side")
            or body.get("direction")
            or request.form.get("side")
            or ""
        ).strip().lower()
        if not symbol:
            return jsonify({"ok": False, "msg": "symbol 不能为空"}), 400
        if side not in ("long", "short"):
            return jsonify({"ok": False, "msg": "side 须为 long 或 short"}), 400
        get_db = _ctx().get("get_db")
        if not callable(get_db):
            return jsonify({"ok": False, "msg": "HUB_CTX 缺少 get_db"}), 500
        conn = get_db()
        try:
            out = fn(conn, symbol, side)
            if not isinstance(out, dict):
                out = {"ok": True, "synced": int(out or 0)}
            conn.commit()
            return jsonify(out)
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500
        finally:
            conn.close()

    @app.route("/api/hub/trend/sync-flat", methods=["POST"])
    @_hub_auth_required
    def api_hub_trend_sync_flat():
        """中控市价全平后:结束仍 active 的同币种同向趋势计划."""
        if not _ctx().get("has_trend"):
            return jsonify({"ok": False, "msg": "该实例无趋势回调"}), 400
        body = request.get_json(silent=True) or {}
        symbol = (body.get("symbol") or request.form.get("symbol") or "").strip()
        side = (
            body.get("side")
            or body.get("direction")
            or request.form.get("side")
            or ""
        ).strip().lower()
        if not symbol:
            return jsonify({"ok": False, "msg": "symbol 不能为空"}), 400
        if side not in ("long", "short"):
            return jsonify({"ok": False, "msg": "side 须为 long 或 short"}), 400
        cfg = current_app.extensions.get("strategy_trend_cfg")
        get_db = _ctx().get("get_db")
        if not cfg or not callable(get_db):
            return jsonify({"ok": False, "msg": "趋势配置未就绪"}), 500
        from lib.strategy.strategy_trend_register import sync_trend_plans_after_external_close

        conn = get_db()
        try:
            return jsonify(sync_trend_plans_after_external_close(cfg, conn, symbol, side))
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500
        finally:
            conn.close()

    @app.route("/api/hub/roll/sync-flat", methods=["POST"])
    @_hub_auth_required
    def api_hub_roll_sync_flat():
        """中控/实例手动平仓后:取消滚仓 pending 并关闭 active 滚仓组."""
        body = request.get_json(silent=True) or {}
        symbol = (body.get("symbol") or request.form.get("symbol") or "").strip()
        side = (
            body.get("side")
            or body.get("direction")
            or request.form.get("side")
            or ""
        ).strip().lower()
        if not symbol:
            return jsonify({"ok": False, "msg": "symbol 不能为空"}), 400
        if side not in ("long", "short"):
            return jsonify({"ok": False, "msg": "side 须为 long 或 short"}), 400
        cfg = current_app.extensions.get("strategy_roll_cfg")
        get_db = _ctx().get("get_db")
        if not cfg or not callable(get_db):
            return jsonify({"ok": False, "msg": "滚仓配置未就绪"}), 500
        from lib.strategy.strategy_register import roll_sync_after_external_close

        conn = get_db()
        try:
            out = roll_sync_after_external_close(cfg, conn, symbol, side)
            conn.commit()
            return jsonify(out)
        except Exception as e:
            return jsonify({"ok": False, "msg": str(e)}), 500
        finally:
            conn.close()

    @app.route("/api/hub/trend/breakeven/<int:pid>", methods=["POST"])
    @_hub_auth_required
    def api_hub_trend_breakeven(pid):
        if not _ctx().get("has_trend"):
            return jsonify({"ok": False, "msg": "该实例无趋势回调"}), 400
        body = request.get_json(silent=True) or {}
        raw = (request.form.get("breakeven_offset_pct") or body.get("breakeven_offset_pct") or "").strip()
        form = {}
        if raw != "":
            form["breakeven_offset_pct"] = raw
        return jsonify(
            _invoke_view(
                "trend_pullback_breakeven",
                f"/trend_pullback_breakeven/{pid}",
                form=form,
            )
        )

    @app.route("/hub-sso")
    def hub_sso_login():
        """中控签发的临时链接:写入 session 后跳转,直链访问仍走 /login."""
        from urllib.parse import urlencode

        auth_disabled = bool(current_app.config.get("HUB_AUTH_DISABLED"))
        next_arg = request.args.get("next")
        if auth_disabled:
            session["logged_in"] = True
            return redirect(safe_next_path(next_arg))
        ex = str((_ctx().get("exchange") or "")).strip().lower()
        token = (request.args.get("token") or "").strip()
        ok, next_path, err = verify_hub_sso_token(token, ex)
        if ok:
            embed_on = request.args.get("embed", "").strip().lower() in (
                "1",
                "true",
                "yes",
                "on",
            )
            dest_next = _embed_login_dest(next_path) if embed_on else next_path
            if not embed_on:
                ht = (request.args.get("hub_theme") or "").strip().lower()
                if ht in ("light", "dark"):
                    dest_next = _merge_query_into_path(next_path, hub_theme=ht)
            if embed_on and _sso_wants_embed_auth() and request.is_secure:
                boot = mint_hub_embed_bootstrap(ex, dest_next)
                if boot:
                    from urllib.parse import urlencode as _ue

                    qdict = {"t": boot, "next": dest_next, "embed": "1"}
                    ht0 = (request.args.get("hub_theme") or "").strip().lower()
                    if ht0 in ("light", "dark"):
                        qdict["hub_theme"] = ht0
                    return redirect(f"/hub-embed-auth?{_ue(qdict)}")
            session["logged_in"] = True
            session.modified = True
            return redirect(dest_next)
        hint = err or "校验失败"
        flash(
            f"中控 SSO 未生效({hint})."
            "请确认中控与实例 .env 中 HUB_BRIDGE_TOKEN 一致,"
            f"且中控设置里该账户 key 为「{ex}」."
            "经本地导航 iframe 打开时,实例须 HTTPS 且可设 APP_COOKIE_SECURE=true."
        )
        return redirect("/login")

    @app.route("/hub-embed-auth")
    def hub_embed_auth_login():
        """LocalNav 等 iframe 内嵌:单独写入 SameSite=None 会话后跳转."""
        auth_disabled = bool(current_app.config.get("HUB_AUTH_DISABLED"))
        next_arg = request.args.get("next")
        if auth_disabled:
            session["logged_in"] = True
            return redirect(safe_next_path(next_arg))
        ex = str((_ctx().get("exchange") or "")).strip().lower()
        boot = (request.args.get("t") or "").strip()
        ok, next_path, err = verify_hub_embed_bootstrap(boot, ex)
        if ok:
            session["logged_in"] = True
            session.modified = True
            return redirect(_embed_login_dest(next_path))
        hint = err or "校验失败"
        flash(f"iframe 登录未生效({hint}).可点本地导航工具栏「实例免密」重试.")
        return redirect("/login")


def _latest_preview_id():
    get_db = _ctx().get("get_db")
    if not get_db:
        return None
    conn = get_db()
    row = conn.execute(
        "SELECT id FROM trend_pullback_previews ORDER BY created_at DESC LIMIT 1"
    ).fetchone()
    conn.close()
    return row["id"] if row else None


def _fetch_preview(pid):
    get_db = _ctx().get("get_db")
    if not get_db or not pid:
        return None
    conn = get_db()
    row = conn.execute(
        "SELECT * FROM trend_pullback_previews WHERE id=?", (pid,)
    ).fetchone()
    conn.close()
    if not row:
        return None
    d = _row_to_dict(row)
    now_ms = int(time.time() * 1000)
    d["expires_in_sec"] = max(0, int((int(d.get("expires_at_ms") or 0) - now_ms) / 1000))
    try:
        from lib.strategy.strategy_trend_lib import build_trend_preview_level_rows

        enriched, level_rows = build_trend_preview_level_rows(d)
        for key in (
            "preview_target_rr",
            "preview_first_take_profit",
            "preview_unified_stop_loss",
            "preview_risk_amount_u",
            "preview_first_profit_u",
            "preview_take_profit_price",
        ):
            if key in enriched:
                d[key] = enriched[key]
        d["preview_level_rows"] = level_rows
        d["grid_levels"] = [
            {
                "i": row.get("i"),
                "label": row.get("label"),
                "price": row.get("price"),
                "contracts": row.get("contracts"),
                "cum_contracts": row.get("cum_contracts"),
                "avg_entry": row.get("avg_entry"),
                "take_profit_price": row.get("take_profit_price"),
                "profit_u": row.get("profit_u"),
                "risk_u": row.get("risk_u"),
                "rr": row.get("rr"),
                "stop_loss_price": row.get("stop_loss_price"),
                "take_profit": row.get("profit_u"),
                "stop_loss": row.get("risk_u"),
            }
            for row in level_rows
        ]
    except Exception:
        d["grid_levels"] = []
        d["preview_level_rows"] = []
    return d