"""OKX 期权模块:Flask 路由注册.""" from __future__ import annotations import os import threading from typing import Any from flask import Flask, jsonify, redirect, request, url_for from jinja2 import ChoiceLoader, FileSystemLoader from lib.options.options_db import init_options_tables from lib.options.options_monitor_lib import options_monitor_loop from lib.options.options_pricing_lib import ( calc_order_size, ct_mult_from_meta, min_sz_from_meta, premium_per_sheet, total_premium, ) from lib.exchange.okx_options_lib import _pos_side_from_position, _safe_float, td_mode_for_option_buy def _env_bool(key: str, default: bool = False) -> bool: raw = (os.getenv(key) or "").strip().lower() if not raw: return default return raw in ("1", "true", "yes", "on") def _env_float(key: str, default: float) -> float: try: return float(os.getenv(key, str(default))) except (TypeError, ValueError): return default def attach_options_templates(app: Flask, repo_root: str) -> None: tpl_dir = os.path.join(repo_root, "lib", "options", "templates") if not os.path.isdir(tpl_dir): return existing = app.jinja_loader loaders = [FileSystemLoader(tpl_dir)] if existing is not None: if isinstance(existing, ChoiceLoader): loaders = list(existing.loaders) + loaders else: loaders.insert(0, existing) app.jinja_loader = ChoiceLoader(loaders) def install_options_trading(app: Flask, repo_root: str, app_module: Any) -> None: enabled = _env_bool("OKX_OPTIONS_ENABLED", False) attach_options_templates(app, repo_root) cfg = _build_cfg(app_module) app.extensions["options_cfg"] = cfg register_options_routes(app, cfg) _register_options_hub_bridge(app, cfg) if enabled: _start_monitor_thread(app, cfg) def _register_options_hub_bridge(app: Flask, cfg: dict[str, Any]) -> None: from lib.options.options_hub_lib import build_options_hub_snapshot def snapshot_fn(): return build_options_hub_snapshot(cfg) hub_ctx = dict(app.config.get("HUB_CTX") or {}) hub_ctx["options_snapshot_fn"] = snapshot_fn app.config["HUB_CTX"] = hub_ctx def _build_cfg(app_module: Any) -> dict[str, Any]: from lib.exchange.okx_options_lib import ( build_option_chain, estimate_usdt_to_usdc, execute_convert, fetch_option_positions, fetch_options_balances, format_position_row, options_api_ready, place_option_limit_order, place_option_market_order, quote_option_contract, spot_market_swap_usdt_usdc, transfer_ccy, transfer_main_sub_account, ) return { "enabled": _env_bool("OKX_OPTIONS_ENABLED", False), "sub_account_name": (os.getenv("OKX_SUB_ACCOUNT_NAME") or "").strip(), "get_db": app_module.get_db, "login_required": app_module.login_required, "exchange_options": getattr(app_module, "exchange_options", None), "send_wechat": app_module.send_wechat_msg, "render_main_page": app_module.render_main_page, "trade_budget": _env_float("OKX_OPTIONS_TRADE_BUDGET_USDC", 10.0), "budget_buffer": _env_float("OKX_OPTIONS_BUDGET_BUFFER", 0.95), "default_underly": (os.getenv("OKX_OPTIONS_DEFAULT_UNDERLY") or "ETH").strip().upper(), "max_dte_days": _env_float("OKX_OPTIONS_MAX_DTE_DAYS", 2.0), "chain_max_dte_days": _env_float("OKX_OPTIONS_CHAIN_MAX_DTE_DAYS", 14.0), "itm_max_dist": _env_float("OKX_OPTIONS_ITM_MAX_DIST_USD", 30.0), "td_mode": (os.getenv("OKX_OPTIONS_TD_MODE") or "isolated").strip(), "allow_market_close": _env_bool("OKX_OPTIONS_ALLOW_MARKET_CLOSE", False), "profit_ratio": _env_float("OKX_OPTIONS_PROFIT_ALERT_RATIO", 1.0), "poll_seconds": _env_float("OKX_OPTIONS_POLL_SECONDS", 15.0), "account_label": (os.getenv("OKX_OPTIONS_ACCOUNT_LABEL") or "OKX期权").strip(), "build_option_chain": build_option_chain, "quote_option_contract": quote_option_contract, "place_option_limit_order": place_option_limit_order, "place_option_market_order": place_option_market_order, "fetch_option_positions": fetch_option_positions, "fetch_options_balances": fetch_options_balances, "format_position_row": format_position_row, "estimate_usdt_to_usdc": estimate_usdt_to_usdc, "execute_convert": execute_convert, "transfer_ccy": transfer_ccy, "spot_market_swap_usdt_usdc": spot_market_swap_usdt_usdc, "transfer_main_sub_account": transfer_main_sub_account, "options_api_ready": options_api_ready, } def _require_options_ex(cfg: dict[str, Any]): if not cfg.get("enabled"): return None, "期权模块未启用,请在 .env 设置 OKX_OPTIONS_ENABLED=true 并重启 PM2" ex = cfg.get("exchange_options") ok, reason = cfg["options_api_ready"](ex) if not ok: return None, reason or "期权 API 未配置" return ex, "" def _budget_full_usdc(cfg: dict[str, Any], ex: Any) -> tuple[float | None, str]: """交易账户 USDC 可用余额(由 calc_order_size 再乘 budget_buffer 留余量).""" from lib.exchange.okx_options_lib import fetch_options_trading_usdc raw = fetch_options_trading_usdc(ex) if raw is None or float(raw) <= 0: return None, "交易账户 USDC 可用余额不足" return float(raw), "" def _sync_options_trades(cfg: dict[str, Any]) -> None: ex = cfg.get("exchange_options") if ex is None: return from lib.exchange.okx_options_lib import fetch_option_position_history from lib.options.options_monitor_lib import reconcile_live_open_trades, sync_open_options_trades raw = cfg["fetch_option_positions"](ex) live_ids = {str(p.get("instId") or "") for p in raw if str(p.get("instId") or "")} def _hist(inst_id: str): return fetch_option_position_history(ex, inst_id) conn = cfg["get_db"]() try: init_options_tables(conn) reconcile_live_open_trades(conn, live_inst_ids=live_ids) sync_open_options_trades(conn, live_inst_ids=live_ids, fetch_history_fn=_hist) conn.commit() finally: conn.close() def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None: lr = cfg["login_required"] @app.route("/api/options/balances") @lr def api_options_balances(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) bal = cfg["fetch_options_balances"](ex) return jsonify({"ok": True, **bal, "trade_budget": cfg["trade_budget"]}) @app.route("/api/options/chain") @lr def api_options_chain(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) u = (request.args.get("underlying") or cfg["default_underly"]).upper() chain = cfg["build_option_chain"]( ex, u, max_dte_days=cfg["chain_max_dte_days"], itm_only=False, itm_max_dist_usd=cfg["itm_max_dist"], ) return jsonify({"ok": True, **chain, "chain_max_dte_days": cfg["chain_max_dte_days"]}) @app.route("/api/options/quote") @lr def api_options_quote(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) inst_id = (request.args.get("inst_id") or "").strip() if not inst_id: return jsonify({"ok": False, "msg": "缺少 inst_id"}) q = cfg["quote_option_contract"](ex, inst_id) if not q.get("ok"): return jsonify(q) ask = q.get("ask") ct_mult = q.get("ct_mult") or 0.01 min_sz = q.get("min_sz") or 1 mode = (request.args.get("mode") or "budget_full").strip() budget = cfg["trade_budget"] budget_cap = cfg["trade_budget"] available_usdc = None if mode == "budget_full": budget, budget_err = _budget_full_usdc(cfg, ex) if budget is None: return jsonify({"ok": False, "msg": budget_err}) budget_cap = budget from lib.exchange.okx_options_lib import fetch_options_trading_usdc available_usdc = fetch_options_trading_usdc(ex) eth_amount = None sheet_count = None try: if request.args.get("eth_amount"): eth_amount = float(request.args.get("eth_amount")) except (TypeError, ValueError): pass try: if request.args.get("sheets"): sheet_count = int(request.args.get("sheets")) except (TypeError, ValueError): pass if ask is None or ask <= 0: return jsonify({**q, "ok": False, "msg": "暂无卖一价"}) sizing = calc_order_size( quote_per_unit=float(ask), ct_mult=float(ct_mult), min_sz=int(min_sz), budget_usdc=budget if mode == "budget_full" else None, budget_buffer=cfg["budget_buffer"], eth_amount=eth_amount if mode == "eth_amount" else None, sheets=sheet_count if mode == "sheets" else None, budget_cap=budget_cap if mode in ("budget_full", "sheets", "eth_amount") else None, ) return jsonify( { **q, "quote_per_unit": ask, "premium_per_sheet": premium_per_sheet(float(ask), float(ct_mult)), "sizing": sizing, "available_usdc": available_usdc, "budget_full_usdc": budget if mode == "budget_full" else None, } ) @app.route("/api/options/open", methods=["POST"]) @lr def api_options_open(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) data = request.get_json(silent=True) or {} inst_id = (data.get("inst_id") or "").strip() mode = (data.get("mode") or "budget_full").strip() signal_note = (data.get("signal_note") or "").strip() if not inst_id: return jsonify({"ok": False, "msg": "缺少 inst_id"}) q = cfg["quote_option_contract"](ex, inst_id) if not q.get("ok"): return jsonify(q) ask = q.get("ask") if ask is None or ask <= 0: return jsonify({"ok": False, "msg": "暂无卖一价,无法买入"}) ct_mult = float(q.get("ct_mult") or 0.01) min_sz = int(q.get("min_sz") or 1) eth_amount = None sheet_count = None if mode == "eth_amount": try: eth_amount = float(data.get("eth_amount")) except (TypeError, ValueError): return jsonify({"ok": False, "msg": "ETH 数量无效"}) elif mode == "sheets": try: sheet_count = int(data.get("sheets")) except (TypeError, ValueError): return jsonify({"ok": False, "msg": "张数无效"}) budget = cfg["trade_budget"] budget_cap = cfg["trade_budget"] if mode == "budget_full": budget, budget_err = _budget_full_usdc(cfg, ex) if budget is None: return jsonify({"ok": False, "msg": budget_err}) budget_cap = budget sizing = calc_order_size( quote_per_unit=float(ask), ct_mult=ct_mult, min_sz=min_sz, budget_usdc=budget if mode == "budget_full" else None, budget_buffer=cfg["budget_buffer"], eth_amount=eth_amount, sheets=sheet_count, budget_cap=budget_cap if mode in ("budget_full", "sheets", "eth_amount") else None, ) if not sizing.get("ok"): return jsonify({"ok": False, "msg": sizing.get("msg") or "张数计算失败", "sizing": sizing}) sheets = int(sizing["sheets"]) tick_sz = q.get("tick_sz") order = cfg["place_option_limit_order"]( ex, inst_id=inst_id, side="buy", sheets=sheets, price=float(ask), td_mode=td_mode_for_option_buy(cfg["td_mode"]), tick_sz=tick_sz, ) if not order.get("ok"): return jsonify(order) conn = cfg["get_db"]() try: init_options_tables(conn) meta = q.get("meta") or {} u = str(meta.get("uly") or inst_id).split("-")[0] conn.execute( """ INSERT INTO options_trades (inst_id, underlying, opt_type, strike, exp_time, sheets, eth_amount, open_quote, premium_paid, status, signal_note, exchange_ord_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?) """, ( inst_id, u, meta.get("optType"), q.get("strike"), str(q.get("exp_time") or ""), sheets, sizing["eth_amount"], float(ask), sizing["total_premium"], signal_note, (order.get("data") or {}).get("ordId"), ), ) conn.commit() finally: conn.close() return jsonify({"ok": True, "order": order, "sizing": sizing}) @app.route("/api/options/positions") @lr def api_options_positions(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) _sync_options_trades(cfg) raw = cfg["fetch_option_positions"](ex) rows = [cfg["format_position_row"](p) for p in raw] conn = cfg["get_db"]() try: for row in rows: inst = row.get("inst_id") if not inst: continue rec = conn.execute( """ SELECT premium_paid FROM options_trades WHERE inst_id = ? AND status = 'open' ORDER BY id DESC LIMIT 1 """, (inst,), ).fetchone() if rec and rec["premium_paid"] is not None: row["premium_paid"] = round(float(rec["premium_paid"]), 4) finally: conn.close() return jsonify({"ok": True, "positions": rows}) @app.route("/api/options/close", methods=["POST"]) @lr def api_options_close(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) data = request.get_json(silent=True) or {} inst_id = (data.get("inst_id") or "").strip() use_market = bool(data.get("market")) and cfg["allow_market_close"] if not inst_id: return jsonify({"ok": False, "msg": "缺少 inst_id"}) sheets = data.get("sheets") q = cfg["quote_option_contract"](ex, inst_id) bid = q.get("bid") if not use_market and (bid is None or bid <= 0): return jsonify({"ok": False, "msg": "暂无买一价,无法限价平仓"}) raw_positions = cfg["fetch_option_positions"](ex) pos = next((p for p in raw_positions if str(p.get("instId")) == inst_id), None) if not pos: return jsonify({"ok": False, "msg": "未找到持仓"}) avail = _safe_float(pos.get("availPos")) if avail is None or avail <= 0: avail = abs(_safe_float(pos.get("pos")) or 0) close_sheets = int(sheets) if sheets else int(avail) if close_sheets < 1: return jsonify({"ok": False, "msg": "可平张数不足"}) td_mode = str(pos.get("mgnMode") or cfg["td_mode"]) pos_side = _pos_side_from_position(pos) or "net" tick_sz = q.get("tick_sz") if use_market: order = cfg["place_option_market_order"]( ex, inst_id=inst_id, side="sell", sheets=close_sheets, td_mode=td_mode, reduce_only=True, pos_side=pos_side, ) if not order.get("ok"): return jsonify(order) else: close_px = float(bid) order = cfg["place_option_limit_order"]( ex, inst_id=inst_id, side="sell", sheets=close_sheets, price=close_px, td_mode=td_mode, tick_sz=tick_sz, reduce_only=True, pos_side=pos_side, ) if not order.get("ok"): return jsonify(order) bid = order.get("px", close_px) prem_recv = total_premium(float(bid or 0), close_sheets * float(q.get("ct_mult") or 0.01)) conn = cfg["get_db"]() try: init_options_tables(conn) row = conn.execute( "SELECT id, premium_paid FROM options_trades WHERE inst_id = ? AND status = 'open' ORDER BY id DESC LIMIT 1", (inst_id,), ).fetchone() if row: paid = float(row["premium_paid"] or 0) pnl = prem_recv - paid conn.execute( """ UPDATE options_trades SET status = 'closed', close_quote = ?, premium_received = ?, realized_pnl = ?, close_ord_id = ?, closed_at = CURRENT_TIMESTAMP WHERE id = ? """, ( bid, prem_recv, pnl, (order.get("data") or {}).get("ordId"), int(row["id"]), ), ) conn.commit() finally: conn.close() return jsonify({"ok": True, "order": order, "bid": bid, "sheets": close_sheets}) @app.route("/api/options/convert/quote", methods=["POST"]) @lr def api_options_convert_quote(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) data = request.get_json(silent=True) or {} try: amount = float(data.get("amount")) except (TypeError, ValueError): return jsonify({"ok": False, "msg": "数量无效"}) return jsonify(cfg["estimate_usdt_to_usdc"](ex, amount)) @app.route("/api/options/convert/execute", methods=["POST"]) @lr def api_options_convert_execute(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) data = request.get_json(silent=True) or {} quote_id = (data.get("quote_id") or "").strip() result = cfg["execute_convert"](ex, quote_id) if result.get("ok"): conn = cfg["get_db"]() try: init_options_tables(conn) conn.execute( """ INSERT INTO options_convert_log (from_ccy, to_ccy, rfq_sz, received_sz, quote_id, status, message) VALUES ('USDT', 'USDC', ?, ?, ?, 'ok', '') """, ( data.get("rfq_sz"), (result.get("data") or {}).get("baseSz"), quote_id, ), ) conn.commit() finally: conn.close() return jsonify(result) @app.route("/api/options/transfer", methods=["POST"]) @lr def api_options_transfer(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) data = request.get_json(silent=True) or {} ccy = (data.get("ccy") or "USDC").upper() from_acct = (data.get("from") or "funding").strip() to_acct = (data.get("to") or "trading").strip() try: amount = float(data.get("amount")) except (TypeError, ValueError): return jsonify({"ok": False, "msg": "数量无效"}) result = cfg["transfer_ccy"](ex, ccy, amount, from_acct, to_acct) if result.get("ok"): conn = cfg["get_db"]() try: init_options_tables(conn) conn.execute( """ INSERT INTO options_transfer_log (ccy, amount, from_account, to_account, status, message) VALUES (?, ?, ?, ?, 'ok', '') """, (ccy, amount, from_acct, to_acct), ) conn.commit() finally: conn.close() return jsonify(result) @app.route("/api/options/spot/swap", methods=["POST"]) @lr def api_options_spot_swap(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) data = request.get_json(silent=True) or {} direction = (data.get("direction") or "usdt_to_usdc").strip() try: amount = float(data.get("amount")) except (TypeError, ValueError): return jsonify({"ok": False, "msg": "数量无效"}) return jsonify(cfg["spot_market_swap_usdt_usdc"](ex, direction=direction, amount=amount)) @app.route("/api/options/cross-transfer", methods=["POST"]) @lr def api_options_cross_transfer(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) data = request.get_json(silent=True) or {} ccy = (data.get("ccy") or "USDT").upper() direction = (data.get("direction") or "sub_to_main").strip() from_account = (data.get("from_account") or data.get("account") or "funding").strip() to_account = (data.get("to_account") or data.get("account") or "funding").strip() try: amount = float(data.get("amount")) except (TypeError, ValueError): return jsonify({"ok": False, "msg": "数量无效"}) main_to_sub = direction == "main_to_sub" result = cfg["transfer_main_sub_account"]( ex, ccy=ccy, amount=amount, sub_acct=cfg.get("sub_account_name") or "", main_to_sub=main_to_sub, from_account=from_account, to_account=to_account, ) if result.get("ok"): conn = cfg["get_db"]() try: init_options_tables(conn) conn.execute( """ INSERT INTO options_transfer_log (ccy, amount, from_account, to_account, status, message) VALUES (?, ?, ?, ?, 'ok', ?) """, ( ccy, amount, ("main" if main_to_sub else "sub") + ":" + from_account, ("sub" if main_to_sub else "main") + ":" + to_account, "cross", ), ) conn.commit() finally: conn.close() return jsonify(result) @app.route("/api/options/history") @lr def api_options_history(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) _sync_options_trades(cfg) conn = cfg["get_db"]() try: init_options_tables(conn) rows = conn.execute( """ SELECT id, inst_id, underlying, opt_type, strike, sheets, eth_amount, open_quote, premium_paid, close_quote, premium_received, realized_pnl, status, signal_note, created_at, closed_at FROM options_trades ORDER BY id DESC LIMIT 200 """ ).fetchall() items = [dict(r) for r in rows] finally: conn.close() return jsonify({"ok": True, "history": items}) @app.route("/api/options/stats") @lr def api_options_stats(): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) from lib.instance.instance_embed_context_lib import profit_loss_ratio_from_averages conn = cfg["get_db"]() try: init_options_tables(conn) rows = conn.execute( """ SELECT realized_pnl FROM options_trades WHERE status = 'closed' AND realized_pnl IS NOT NULL """ ).fetchall() finally: conn.close() wins: list[float] = [] losses: list[float] = [] for row in rows: pnl = float(row["realized_pnl"]) if pnl > 0: wins.append(pnl) elif pnl < 0: losses.append(pnl) total_closed = len(wins) + len(losses) win_rate = round(len(wins) / total_closed * 100, 2) if total_closed else 0 avg_win = sum(wins) / len(wins) if wins else None avg_loss = sum(losses) / len(losses) if losses else None total_profit = round(sum(wins), 4) if wins else 0.0 total_loss = round(abs(sum(losses)), 4) if losses else 0.0 return jsonify( { "ok": True, "total_closed": total_closed, "win_rate": win_rate, "profit_loss_ratio": profit_loss_ratio_from_averages(avg_win, avg_loss), "total_profit": total_profit, "total_loss": total_loss, } ) @app.route("/api/options/history/", methods=["DELETE"]) @lr def api_options_history_delete(trade_id: int): ex, err = _require_options_ex(cfg) if ex is None: return jsonify({"ok": False, "msg": err}) conn = cfg["get_db"]() try: init_options_tables(conn) row = conn.execute( "SELECT id, status FROM options_trades WHERE id = ?", (trade_id,), ).fetchone() if not row: return jsonify({"ok": False, "msg": "记录不存在"}) conn.execute("DELETE FROM options_trades WHERE id = ?", (trade_id,)) conn.commit() finally: conn.close() return jsonify({"ok": True}) def _start_monitor_thread(app: Flask, cfg: dict[str, Any]) -> None: if app.extensions.get("options_monitor_started"): return app.extensions["options_monitor_started"] = True def _bid(inst_id: str) -> float | None: ex = cfg.get("exchange_options") if ex is None: return None try: q = cfg["quote_option_contract"](ex, inst_id) return q.get("bid") except Exception: return None def _positions(): ex = cfg.get("exchange_options") if ex is None: return [] raw = cfg["fetch_option_positions"](ex) return [cfg["format_position_row"](p) for p in raw] def _sync(conn): from lib.exchange.okx_options_lib import fetch_option_position_history from lib.options.options_monitor_lib import reconcile_live_open_trades, sync_open_options_trades ex = cfg.get("exchange_options") if ex is None: return 0 raw = cfg["fetch_option_positions"](ex) live_ids = {str(p.get("instId") or "") for p in raw if str(p.get("instId") or "")} reconcile_live_open_trades(conn, live_inst_ids=live_ids) return sync_open_options_trades( conn, live_inst_ids=live_ids, fetch_history_fn=lambda inst_id: fetch_option_position_history(ex, inst_id), ) t = threading.Thread( target=options_monitor_loop, kwargs={ "enabled": True, "poll_seconds": cfg["poll_seconds"], "get_db": cfg["get_db"], "fetch_positions": _positions, "ticker_bid_fn": _bid, "send_wechat": cfg["send_wechat"], "account_label": cfg["account_label"], "profit_ratio": cfg["profit_ratio"], "sync_trades_fn": _sync, }, daemon=True, name="options-monitor", ) t.start()