fix: 交易安全审计修复 — 补偿平仓、中控同步、滚仓/趋势防护

Co-authored-by: Cursor <cursoragent@cursor.com>

crypto_monitor_binance/app.py

@@ -130,6 +130,9 @@ from lib.key_monitor.trigger_entry_key_monitor_lib import (
     TRIGGER_ENTRY_VALIDITY_HOURS,
     check_trigger_entry_intent_limit,
     count_pending_trigger_entries,
+    acquire_trigger_entry_exec_lock,
+    is_trigger_entry_in_flight_row,
+    release_trigger_entry_exec_lock,
     is_breakout_trigger_entry_key_monitor_type,
     is_trigger_entry_expired,
     is_trigger_entry_key_monitor_type,
@@ -3469,6 +3472,40 @@ def ensure_markets_loaded(force=False):
         MARKETS_LOADED = True
 
 
+def _abort_market_open_after_tpsl_failure(exchange_symbol, direction, order, planned_amount):
+    from lib.trade.compensating_close_lib import run_compensating_close
+
+    def _close():
+        ensure_markets_loaded()
+        try:
+            cancel_binance_futures_open_orders(exchange_symbol)
+        except Exception:
+            pass
+        live = get_live_position_contracts(exchange_symbol, direction)
+        amt = live if live is not None and live > 0 else _filled_amount_for_tpsl(order, planned_amount)
+        if amt is None or float(amt) <= 0:
+            return
+        side = "sell" if direction == "long" else "buy"
+        try:
+            amount = float(exchange.amount_to_precision(exchange_symbol, float(amt)))
+        except Exception:
+            amount = float(amt)
+        last_err = None
+        for params in _binance_market_close_param_candidates(direction):
+            try:
+                exchange.create_order(exchange_symbol, "market", side, amount, None, params)
+                return
+            except Exception as e:
+                last_err = e
+                if _is_binance_close_param_retryable(str(e)):
+                    continue
+                raise
+        if last_err:
+            raise last_err
+
+    run_compensating_close(_close, log_prefix="binance_compensating_close")
+
+
 def place_exchange_order(exchange_symbol, direction, amount, leverage, stop_loss=None, take_profit=None):
     ensure_markets_loaded()
     mm = "cross" if BINANCE_MARGIN_MODE in ("cross", "cross_margin") else "isolated"
@@ -3487,8 +3524,10 @@ def place_exchange_order(exchange_symbol, direction, amount, leverage, stop_loss
             _binance_place_tp_sl_orders(exchange_symbol, direction, pos_amt, stop_loss, take_profit)
             order["tpsl_attached"] = True
         except RuntimeError:
+            _abort_market_open_after_tpsl_failure(exchange_symbol, direction, order, amount)
             raise
         except Exception as e:
+            _abort_market_open_after_tpsl_failure(exchange_symbol, direction, order, amount)
             raise RuntimeError(f"交易所未接受条件止盈/止损委托，已拒绝开仓：{str(e)}") from e
     return order
 
@@ -4495,6 +4534,72 @@ def resolve_synced_flat_close(row, opened_at_str, opened_at_ms=None):
     )
 
 
+def _finalize_hub_flat_monitor_binance(conn, r, *, result, pnl_amount, closed_at, miss_reason):
+    opened_at = get_opened_at_value(r)
+    closed_at_dt = parse_dt_for_trading_day(closed_at) or app_now()
+    hold_seconds = calc_hold_seconds(opened_at, closed_at_dt)
+    session_date = r["session_date"] or get_trading_day(closed_at_dt)
+    update_session_capital(conn, session_date, pnl_amount)
+    insert_trade_record(
+        conn,
+        symbol=r["symbol"],
+        monitor_type=trade_record_monitor_type(conn, r),
+        trend_plan_id=trend_plan_id_from_monitor_row(r),
+        key_signal_type=order_row_key_signal_type(r),
+        direction=r["direction"],
+        trigger_price=r["trigger_price"],
+        stop_loss=r["stop_loss"],
+        initial_stop_loss=r["initial_stop_loss"] or r["stop_loss"],
+        take_profit=r["take_profit"],
+        margin_capital=r["margin_capital"],
+        leverage=r["leverage"],
+        pnl_amount=pnl_amount,
+        hold_seconds=hold_seconds,
+        trade_style=r["trade_style"],
+        risk_amount=r["risk_amount"],
+        planned_rr=calc_rr_ratio(
+            r["direction"],
+            r["trigger_price"],
+            r["initial_stop_loss"] or r["stop_loss"],
+            r["take_profit"],
+        ),
+        actual_rr=calc_actual_rr(pnl_amount, r["risk_amount"]),
+        result=result,
+        miss_reason=handoff_trade_miss_reason(miss_reason, r),
+        opened_at=opened_at,
+        closed_at=closed_at,
+    )
+    conn.execute("UPDATE order_monitors SET status='stopped' WHERE id=?", (r["id"],))
+    clear_key_sizing_snapshot_if_flat(conn, r["session_date"] or get_trading_day())
+
+
+def reconcile_hub_external_close(conn, symbol, direction):
+    from lib.hub.hub_reconcile_flat_lib import reconcile_hub_external_close_impl
+    from lib.hub.hub_symbol_lib import symbols_match
+
+    global _RECONCILE_FLAT_STREAK
+
+    return reconcile_hub_external_close_impl(
+        conn,
+        symbol,
+        direction,
+        exchange_configured=exchange_private_api_configured,
+        not_configured_msg="未配置 BINANCE_API_KEY / BINANCE_API_SECRET",
+        symbols_match=symbols_match,
+        get_opened_at_value=get_opened_at_value,
+        resolve_monitor_exchange_symbol=resolve_monitor_exchange_symbol,
+        get_live_position_contracts=get_live_position_contracts,
+        cancel_conditional_orders=cancel_binance_futures_open_orders,
+        resolve_synced_flat_close=resolve_synced_flat_close,
+        finalize_stopped_monitor=_finalize_hub_flat_monitor_binance,
+        sync_trade_records=None,
+        reconcile_flat_streak=_RECONCILE_FLAT_STREAK,
+        to_ms_with_fallback=_to_ms_with_fallback,
+        prefer_manual_resolve=False,
+        order_row_monitor_type=order_row_monitor_type,
+    )
+
+
 def reconcile_external_closes(conn, days=None):
     global _RECONCILE_FLAT_STREAK
     if not exchange_private_api_configured():
@@ -5777,7 +5882,7 @@ def _market_open_for_trigger_entry(
 
 
 def _execute_trigger_entry_cross(conn, row):
-    """标记价触达计划入场：先删监控行防重复触发，再市价开仓。"""
+    """标记价触达计划入场：加锁防重复触发，成交成功后再删监控行。"""
     symbol = row["symbol"]
     direction = (row["direction"] or "long").lower()
     ex_sym = normalize_exchange_symbol(symbol)
@@ -5788,7 +5893,8 @@ def _execute_trigger_entry_cross(conn, row):
     tc_en, tc_h, _ = time_close_settings_from_row(row)
 
     kid = int(row["id"])
-    conn.execute("DELETE FROM key_monitors WHERE id=?", (kid,))
+    if not acquire_trigger_entry_exec_lock(conn, kid):
+        return False, "触价开仓进行中"
     conn.commit()
 
     try:
@@ -5806,6 +5912,8 @@ def _execute_trigger_entry_cross(conn, row):
             time_close_hours=tc_h,
         )
     except Exception as e:
+        release_trigger_entry_exec_lock(conn, kid)
+        conn.commit()
         fail_msg = friendly_exchange_error(e)
         send_wechat_msg(
             f"# ❌ {symbol} 触价开仓异常\n"
@@ -5817,6 +5925,8 @@ def _execute_trigger_entry_cross(conn, row):
         return False, fail_msg
 
     if ok and det:
+        conn.execute("DELETE FROM key_monitors WHERE id=?", (kid,))
+        conn.commit()
         rr_txt = format_wechat_scalar_2dp(det.get("planned_rr_fill")) if det.get("planned_rr_fill") is not None else "-"
         msg = (
             f"# ✅ {symbol} 触价开仓成交\n"
@@ -5833,6 +5943,8 @@ def _execute_trigger_entry_cross(conn, row):
         send_wechat_msg(msg)
         insert_key_monitor_history(conn, row, 0, msg, TRIGGER_ENTRY_CLOSE_FILLED)
         return True, None
+    release_trigger_entry_exec_lock(conn, kid)
+    conn.commit()
     fail_msg = err or "触价触发后开仓失败"
     send_wechat_msg(
         f"# ❌ {symbol} 触价开仓失败\n"
@@ -5860,6 +5972,8 @@ def check_trigger_entry_key_monitors():
         sl = float(_sqlite_row_val(r, "fib_stop_loss") or 0)
         tp = float(_sqlite_row_val(r, "fib_take_profit") or 0)
         kid = int(r["id"])
+        if is_trigger_entry_in_flight_row(r):
+            continue
         if entry <= 0 or sl <= 0 or tp <= 0:
             _finalize_key_monitor_one_shot(conn, r, "触价计划价位无效", "fib_plan_invalid")
             continue
@@ -6388,7 +6502,7 @@ def check_order_monitors():
                             new_sl = round_price_to_exchange(ex_sym, new_sl)
                             tp_ex = float(take_profit or 0)
                             ok_live, _live_reason = ensure_exchange_live_ready()
-                            synced_ex = not ok_live
+                            synced_ex = False
                             if ok_live and tp_ex > 0:
                                 try:
                                     replace_active_monitor_tpsl_on_exchange(r, new_sl, tp_ex)
@@ -6818,8 +6932,8 @@ def background_task():
                 from lib.strategy.strategy_trend_register import check_trend_pullback_plans
 
                 check_trend_pullback_plans(cfg)
-        except:
-            pass
+        except Exception as e:
+            print(f"[monitor_loop] {e}", flush=True)
         time.sleep(MONITOR_POLL_SECONDS)
 
 
@@ -9667,6 +9781,7 @@ try:
         ohlcv_fn=_hub_fetch_ohlcv,
         volume_rank_fn=_hub_fetch_volume_rank,
         market_fn=_hub_fetch_market,
+        reconcile_hub_flat_fn=reconcile_hub_external_close,
         risk_status_fn=hub_account_risk_status,
         user_close_fn=hub_user_initiated_close,
         render_main_page_fn=render_main_page,

crypto_monitor_gate/app.py

@@ -131,6 +131,9 @@ from lib.key_monitor.trigger_entry_key_monitor_lib import (
     TRIGGER_ENTRY_VALIDITY_HOURS,
     check_trigger_entry_intent_limit,
     count_pending_trigger_entries,
+    acquire_trigger_entry_exec_lock,
+    is_trigger_entry_in_flight_row,
+    release_trigger_entry_exec_lock,
     is_breakout_trigger_entry_key_monitor_type,
     is_trigger_entry_expired,
     is_trigger_entry_key_monitor_type,
@@ -3135,7 +3138,7 @@ def _gate_place_tp_sl_orders_position_price_orders(exchange_symbol, direction, s
             try:
                 exchange.privateFuturesPostSettlePriceOrders(_payload(tp_s, tp_rule))
             except Exception:
-                cancel_gate_swap_trigger_orders(exchange_symbol)
+                # 保留已挂止损，仅放弃本次 TP；上层可补偿平仓或重试
                 raise
             return
         except Exception as e:
@@ -3252,6 +3255,27 @@ def ensure_markets_loaded(force=False):
         MARKETS_LOADED = True
 
 
+def _abort_market_open_after_tpsl_failure(exchange_symbol, direction, order, planned_amount):
+    """TP/SL 挂失败时市价平掉刚开的仓并撤残留条件单。"""
+    from lib.trade.compensating_close_lib import run_compensating_close
+
+    def _close():
+        ensure_markets_loaded()
+        try:
+            cancel_gate_swap_trigger_orders(exchange_symbol)
+        except Exception:
+            pass
+        live = get_live_position_contracts(exchange_symbol, direction)
+        amt = live if live is not None and live > 0 else _gate_contracts_amount_for_tpsl(order, planned_amount)
+        if amt is None or float(amt) <= 0:
+            return
+        side = "sell" if direction == "long" else "buy"
+        params = build_gate_order_params(direction, reduce_only=True)
+        exchange.create_order(exchange_symbol, "market", side, float(amt), None, params)
+
+    run_compensating_close(_close, log_prefix="gate_compensating_close")
+
+
 def place_exchange_order(exchange_symbol, direction, amount, leverage, stop_loss=None, take_profit=None):
     ensure_markets_loaded()
     exchange.set_leverage(leverage, exchange_symbol)
@@ -3265,22 +3289,48 @@ def place_exchange_order(exchange_symbol, direction, amount, leverage, stop_loss
             _gate_place_tp_sl_orders(exchange_symbol, direction, contracts_amt, stop_loss, take_profit)
             order["tpsl_attached"] = True
         except RuntimeError:
+            _abort_market_open_after_tpsl_failure(exchange_symbol, direction, order, amount)
             raise
         except Exception as e:
+            _abort_market_open_after_tpsl_failure(exchange_symbol, direction, order, amount)
             raise RuntimeError(f"交易所未接受条件止盈/止损委托，已拒绝开仓：{str(e)}") from e
     return order
 
 
 def close_exchange_order(order_row):
+    """
+    市价全平。数量优先取交易所当前持仓张数，避免仅用入库 order_amount 导致平不干净。
+    """
     ensure_markets_loaded()
     exchange_symbol = order_row["exchange_symbol"] or normalize_exchange_symbol(order_row["symbol"])
-    amount = float(order_row["order_amount"] or 0)
-    if amount <= 0:
-        raise ValueError("平仓失败：缺少有效下单数量")
     direction = order_row["direction"]
+    db_amt = float(order_row["order_amount"] or 0)
     side = "sell" if direction == "long" else "buy"
-    params = build_gate_order_params(direction, reduce_only=True)
-    return exchange.create_order(exchange_symbol, "market", side, amount, None, params)
+    last_resp = None
+    for _ in range(3):
+        live = get_live_position_contracts(exchange_symbol, direction)
+        if live is not None and live > 0:
+            raw_amt = live
+        else:
+            raw_amt = db_amt
+        if raw_amt <= 0:
+            if last_resp is not None:
+                return last_resp
+            raise ValueError("平仓失败：缺少有效下单数量")
+        try:
+            amount = float(exchange.amount_to_precision(exchange_symbol, raw_amt))
+        except Exception:
+            amount = float(raw_amt)
+        if amount <= 0:
+            if last_resp is not None:
+                return last_resp
+            raise ValueError("平仓失败：数量经精度舍入后为 0")
+        params = build_gate_order_params(direction, reduce_only=True)
+        last_resp = exchange.create_order(exchange_symbol, "market", side, amount, None, params)
+        live_after = get_live_position_contracts(exchange_symbol, direction)
+        if live_after is None or live_after <= 0:
+            return last_resp
+    return last_resp
 
 
 def _gate_swap_trigger_order_params():
@@ -4113,89 +4163,71 @@ def resolve_synced_flat_close(row, opened_at_str, opened_at_ms=None, *, prefer_m
     )
 
 
+def _finalize_hub_flat_monitor(conn, r, *, result, pnl_amount, closed_at, miss_reason):
+    opened_at = get_opened_at_value(r)
+    closed_at_dt = parse_dt_for_trading_day(closed_at) or app_now()
+    hold_seconds = calc_hold_seconds(opened_at, closed_at_dt)
+    session_date = r["session_date"] or get_trading_day(closed_at_dt)
+    update_session_capital(conn, session_date, pnl_amount)
+    insert_trade_record(
+        conn,
+        symbol=r["symbol"],
+        monitor_type=trade_record_monitor_type(conn, r),
+        trend_plan_id=trend_plan_id_from_monitor_row(r),
+        key_signal_type=order_row_key_signal_type(r),
+        direction=r["direction"],
+        trigger_price=r["trigger_price"],
+        stop_loss=r["stop_loss"],
+        initial_stop_loss=r["initial_stop_loss"] or r["stop_loss"],
+        take_profit=r["take_profit"],
+        margin_capital=margin_capital_for_trade_record(r),
+        leverage=r["leverage"],
+        pnl_amount=pnl_amount,
+        hold_seconds=hold_seconds,
+        trade_style=r["trade_style"],
+        risk_amount=r["risk_amount"],
+        planned_rr=calc_rr_ratio(
+            r["direction"],
+            r["trigger_price"],
+            r["initial_stop_loss"] or r["stop_loss"],
+            r["take_profit"],
+        ),
+        actual_rr=calc_actual_rr(pnl_amount, r["risk_amount"]),
+        result=result,
+        miss_reason=handoff_trade_miss_reason(miss_reason, r),
+        opened_at=opened_at,
+        closed_at=closed_at,
+    )
+    conn.execute("UPDATE order_monitors SET status='stopped' WHERE id=?", (r["id"],))
+    clear_key_sizing_snapshot_if_flat(conn, r["session_date"] or get_trading_day())
+
+
 def reconcile_hub_external_close(conn, symbol, direction):
     """中控市价全平后：立即同步匹配 order_monitor，并读 Gate 平仓历史。"""
-    if not exchange_private_api_configured():
-        return {"ok": False, "msg": "未配置 GATE_API_KEY / GATE_API_SECRET", "synced": 0}
-    from lib.exchange.gate_position_history_lib import unified_symbol_for_match
+    from lib.hub.hub_reconcile_flat_lib import reconcile_hub_external_close_impl
+    from lib.hub.hub_symbol_lib import symbols_match
 
-    sym_u = unified_symbol_for_match(symbol)
-    dir_l = (direction or "").strip().lower()
-    if dir_l not in ("long", "short"):
-        return {"ok": False, "msg": "side 须为 long 或 short", "synced": 0}
-    synced = 0
-    rows = conn.execute(
-        "SELECT * FROM order_monitors WHERE status IN ('active', 'error')"
-    ).fetchall()
-    for r in rows:
-        if unified_symbol_for_match(r["symbol"]) != sym_u:
-            continue
-        if (r["direction"] or "").strip().lower() != dir_l:
-            continue
-        oid = int(r["id"])
-        if r["status"] == "error":
-            opened_at_chk = get_opened_at_value(r)
-            existing = conn.execute(
-                "SELECT id FROM trade_records WHERE symbol=? AND opened_at=? AND monitor_type=? LIMIT 1",
-                (r["symbol"], opened_at_chk, order_row_monitor_type(r)),
-            ).fetchone()
-            if existing:
-                conn.execute("UPDATE order_monitors SET status='stopped' WHERE id=?", (oid,))
-                synced += 1
-                continue
-        exchange_symbol = resolve_monitor_exchange_symbol(r)
-        live_contracts = get_live_position_contracts(exchange_symbol, r["direction"])
-        if live_contracts is None:
-            continue
-        if live_contracts > 0:
-            time.sleep(0.6)
-            live_contracts = get_live_position_contracts(exchange_symbol, r["direction"])
-        if live_contracts is None or live_contracts > 0:
-            continue
-        global _RECONCILE_FLAT_STREAK
-        _RECONCILE_FLAT_STREAK.pop(oid, None)
-        cancel_gate_swap_trigger_orders(exchange_symbol)
-        opened_at = get_opened_at_value(r)
-        opened_at_ms = _to_ms_with_fallback(r["opened_at_ms"] if "opened_at_ms" in r.keys() else None, opened_at)
-        result, pnl_amount, closed_at, miss_reason = resolve_synced_flat_close(
-            r, opened_at, opened_at_ms=opened_at_ms, prefer_manual=True
-        )
-        closed_at_dt = parse_dt_for_trading_day(closed_at) or app_now()
-        hold_seconds = calc_hold_seconds(opened_at, closed_at_dt)
-        session_date = r["session_date"] or get_trading_day(closed_at_dt)
-        update_session_capital(conn, session_date, pnl_amount)
-        insert_trade_record(
-            conn,
-            symbol=r["symbol"],
-            monitor_type=trade_record_monitor_type(conn, r),
-            trend_plan_id=trend_plan_id_from_monitor_row(r),
-            key_signal_type=order_row_key_signal_type(r),
-            direction=r["direction"],
-            trigger_price=r["trigger_price"],
-            stop_loss=r["stop_loss"],
-            initial_stop_loss=r["initial_stop_loss"] or r["stop_loss"],
-            take_profit=r["take_profit"],
-            margin_capital=margin_capital_for_trade_record(r),
-            leverage=r["leverage"],
-            pnl_amount=pnl_amount,
-            hold_seconds=hold_seconds,
-            trade_style=r["trade_style"],
-            risk_amount=r["risk_amount"],
-            planned_rr=calc_rr_ratio(r["direction"], r["trigger_price"], r["initial_stop_loss"] or r["stop_loss"], r["take_profit"]),
-            actual_rr=calc_actual_rr(pnl_amount, r["risk_amount"]),
-            result=result,
-            miss_reason=handoff_trade_miss_reason(miss_reason, r),
-            opened_at=opened_at,
-            closed_at=closed_at,
-        )
-        conn.execute("UPDATE order_monitors SET status='stopped' WHERE id=?", (r["id"],))
-        clear_key_sizing_snapshot_if_flat(conn, r["session_date"] or get_trading_day())
-        synced += 1
-    try:
-        sync_trade_records_from_exchange(conn, force=True)
-    except Exception:
-        pass
-    return {"ok": True, "synced": synced}
+    global _RECONCILE_FLAT_STREAK
+
+    return reconcile_hub_external_close_impl(
+        conn,
+        symbol,
+        direction,
+        exchange_configured=exchange_private_api_configured,
+        not_configured_msg="未配置 GATE_API_KEY / GATE_API_SECRET",
+        symbols_match=symbols_match,
+        get_opened_at_value=get_opened_at_value,
+        resolve_monitor_exchange_symbol=resolve_monitor_exchange_symbol,
+        get_live_position_contracts=get_live_position_contracts,
+        cancel_conditional_orders=cancel_gate_swap_trigger_orders,
+        resolve_synced_flat_close=resolve_synced_flat_close,
+        finalize_stopped_monitor=_finalize_hub_flat_monitor,
+        sync_trade_records=sync_trade_records_from_exchange,
+        reconcile_flat_streak=_RECONCILE_FLAT_STREAK,
+        to_ms_with_fallback=_to_ms_with_fallback,
+        prefer_manual_resolve=True,
+        order_row_monitor_type=order_row_monitor_type,
+    )
 
 
 def reconcile_external_closes(conn, days=None):
@@ -5491,7 +5523,7 @@ def _market_open_for_trigger_entry(
 
 
 def _execute_trigger_entry_cross(conn, row):
-    """标记价触达计划入场：先删监控行防重复触发，再市价开仓。"""
+    """标记价触达计划入场：加锁防重复触发，成交成功后再删监控行。"""
     symbol = row["symbol"]
     direction = (row["direction"] or "long").lower()
     ex_sym = normalize_exchange_symbol(symbol)
@@ -5502,7 +5534,8 @@ def _execute_trigger_entry_cross(conn, row):
     tc_en, tc_h, _ = time_close_settings_from_row(row)
 
     kid = int(row["id"])
-    conn.execute("DELETE FROM key_monitors WHERE id=?", (kid,))
+    if not acquire_trigger_entry_exec_lock(conn, kid):
+        return False, "触价开仓进行中"
     conn.commit()
 
     try:
@@ -5520,6 +5553,8 @@ def _execute_trigger_entry_cross(conn, row):
             time_close_hours=tc_h,
         )
     except Exception as e:
+        release_trigger_entry_exec_lock(conn, kid)
+        conn.commit()
         fail_msg = friendly_exchange_error(e)
         send_wechat_msg(
             f"# ❌ {symbol} 触价开仓异常\n"
@@ -5531,6 +5566,8 @@ def _execute_trigger_entry_cross(conn, row):
         return False, fail_msg
 
     if ok and det:
+        conn.execute("DELETE FROM key_monitors WHERE id=?", (kid,))
+        conn.commit()
         rr_txt = format_wechat_scalar_2dp(det.get("planned_rr_fill")) if det.get("planned_rr_fill") is not None else "-"
         msg = (
             f"# ✅ {symbol} 触价开仓成交\n"
@@ -5547,6 +5584,8 @@ def _execute_trigger_entry_cross(conn, row):
         send_wechat_msg(msg)
         insert_key_monitor_history(conn, row, 0, msg, TRIGGER_ENTRY_CLOSE_FILLED)
         return True, None
+    release_trigger_entry_exec_lock(conn, kid)
+    conn.commit()
     fail_msg = err or "触价触发后开仓失败"
     send_wechat_msg(
         f"# ❌ {symbol} 触价开仓失败\n"
@@ -5574,6 +5613,8 @@ def check_trigger_entry_key_monitors():
         sl = float(_sqlite_row_val(r, "fib_stop_loss") or 0)
         tp = float(_sqlite_row_val(r, "fib_take_profit") or 0)
         kid = int(r["id"])
+        if is_trigger_entry_in_flight_row(r):
+            continue
         if entry <= 0 or sl <= 0 or tp <= 0:
             _finalize_key_monitor_one_shot(conn, r, "触价计划价位无效", "fib_plan_invalid")
             continue
@@ -6117,7 +6158,7 @@ def check_order_monitors():
                             new_sl = round_price_to_exchange(ex_sym, new_sl)
                             tp_ex = float(take_profit or 0)
                             ok_live, _live_reason = ensure_exchange_live_ready()
-                            synced_ex = not ok_live
+                            synced_ex = False
                             if ok_live and tp_ex > 0:
                                 try:
                                     replace_active_monitor_tpsl_on_exchange(r, new_sl, tp_ex)
@@ -6526,8 +6567,8 @@ def background_task():
                 from lib.strategy.strategy_trend_register import check_trend_pullback_plans
 
                 check_trend_pullback_plans(cfg)
-        except:
-            pass
+        except Exception as e:
+            print(f"[monitor_loop] {e}", flush=True)
         time.sleep(MONITOR_POLL_SECONDS)
 
 

crypto_monitor_okx/app.py

@@ -131,6 +131,9 @@ from lib.key_monitor.trigger_entry_key_monitor_lib import (
     TRIGGER_ENTRY_VALIDITY_HOURS,
     check_trigger_entry_intent_limit,
     count_pending_trigger_entries,
+    acquire_trigger_entry_exec_lock,
+    is_trigger_entry_in_flight_row,
+    release_trigger_entry_exec_lock,
     is_breakout_trigger_entry_key_monitor_type,
     is_trigger_entry_expired,
     is_trigger_entry_key_monitor_type,
@@ -2754,15 +2757,39 @@ def place_exchange_order(exchange_symbol, direction, amount, leverage, stop_loss
 
 
 def close_exchange_order(order_row):
+    """
+    市价全平。数量优先取交易所当前持仓张数，避免仅用入库 order_amount 导致平不干净。
+    """
     ensure_markets_loaded()
     exchange_symbol = order_row["exchange_symbol"] or normalize_okx_symbol(order_row["symbol"])
-    amount = float(order_row["order_amount"] or 0)
-    if amount <= 0:
-        raise ValueError("平仓失败：缺少有效下单数量")
     direction = order_row["direction"]
+    db_amt = float(order_row["order_amount"] or 0)
     side = "sell" if direction == "long" else "buy"
-    params = build_okx_order_params(direction, reduce_only=True)
-    return exchange.create_order(exchange_symbol, "market", side, amount, None, params)
+    last_resp = None
+    for _ in range(3):
+        live = get_live_position_contracts(exchange_symbol, direction)
+        if live is not None and live > 0:
+            raw_amt = live
+        else:
+            raw_amt = db_amt
+        if raw_amt <= 0:
+            if last_resp is not None:
+                return last_resp
+            raise ValueError("平仓失败：缺少有效下单数量")
+        try:
+            amount = float(exchange.amount_to_precision(exchange_symbol, raw_amt))
+        except Exception:
+            amount = float(raw_amt)
+        if amount <= 0:
+            if last_resp is not None:
+                return last_resp
+            raise ValueError("平仓失败：数量经精度舍入后为 0")
+        params = build_okx_order_params(direction, reduce_only=True)
+        last_resp = exchange.create_order(exchange_symbol, "market", side, amount, None, params)
+        live_after = get_live_position_contracts(exchange_symbol, direction)
+        if live_after is None or live_after <= 0:
+            return last_resp
+    return last_resp
 
 
 def cancel_okx_swap_open_orders(exchange_symbol):
@@ -3557,6 +3584,71 @@ def resolve_synced_flat_close(row, opened_at_str, opened_at_ms=None):
     )
 
 
+def _finalize_hub_flat_monitor_okx(conn, r, *, result, pnl_amount, closed_at, miss_reason):
+    opened_at = get_opened_at_value(r)
+    closed_at_dt = parse_dt_for_trading_day(closed_at) or app_now()
+    hold_seconds = calc_hold_seconds(opened_at, closed_at_dt)
+    session_date = r["session_date"] or get_trading_day(closed_at_dt)
+    update_session_capital(conn, session_date, pnl_amount)
+    insert_trade_record(
+        conn,
+        symbol=r["symbol"],
+        monitor_type=trade_record_monitor_type(conn, r),
+        trend_plan_id=trend_plan_id_from_monitor_row(r),
+        key_signal_type=order_row_key_signal_type(r),
+        direction=r["direction"],
+        trigger_price=r["trigger_price"],
+        stop_loss=r["stop_loss"],
+        initial_stop_loss=r["initial_stop_loss"] or r["stop_loss"],
+        take_profit=r["take_profit"],
+        margin_capital=r["margin_capital"],
+        leverage=r["leverage"],
+        pnl_amount=pnl_amount,
+        hold_seconds=hold_seconds,
+        trade_style=r["trade_style"],
+        risk_amount=r["risk_amount"],
+        planned_rr=calc_rr_ratio(
+            r["direction"],
+            r["trigger_price"],
+            r["initial_stop_loss"] or r["stop_loss"],
+            r["take_profit"],
+        ),
+        actual_rr=calc_actual_rr(pnl_amount, r["risk_amount"]),
+        result=result,
+        miss_reason=handoff_trade_miss_reason(miss_reason, r),
+        opened_at=opened_at,
+        closed_at=closed_at,
+    )
+    conn.execute("UPDATE order_monitors SET status='stopped' WHERE id=?", (r["id"],))
+
+
+def reconcile_hub_external_close(conn, symbol, direction):
+    from lib.hub.hub_reconcile_flat_lib import reconcile_hub_external_close_impl
+    from lib.hub.hub_symbol_lib import symbols_match
+
+    global _RECONCILE_FLAT_STREAK
+
+    return reconcile_hub_external_close_impl(
+        conn,
+        symbol,
+        direction,
+        exchange_configured=exchange_private_api_configured,
+        not_configured_msg="未配置 OKX_API_KEY / OKX_API_SECRET",
+        symbols_match=symbols_match,
+        get_opened_at_value=get_opened_at_value,
+        resolve_monitor_exchange_symbol=resolve_monitor_exchange_symbol,
+        get_live_position_contracts=get_live_position_contracts,
+        cancel_conditional_orders=cancel_okx_swap_open_orders,
+        resolve_synced_flat_close=resolve_synced_flat_close,
+        finalize_stopped_monitor=_finalize_hub_flat_monitor_okx,
+        sync_trade_records=sync_trade_records_from_exchange,
+        reconcile_flat_streak=_RECONCILE_FLAT_STREAK,
+        to_ms_with_fallback=_to_ms_with_fallback,
+        prefer_manual_resolve=False,
+        order_row_monitor_type=order_row_monitor_type,
+    )
+
+
 def reconcile_external_closes(conn, days=None):
     global _RECONCILE_FLAT_STREAK
     if not exchange_private_api_configured():
@@ -5006,7 +5098,7 @@ def _market_open_for_trigger_entry(
 
 
 def _execute_trigger_entry_cross(conn, row):
-    """标记价触达计划入场：先删监控行防重复触发，再市价开仓。"""
+    """标记价触达计划入场：加锁防重复触发，成交成功后再删监控行。"""
     symbol = row["symbol"]
     direction = (row["direction"] or "long").lower()
     ex_sym = normalize_exchange_symbol(symbol)
@@ -5017,7 +5109,8 @@ def _execute_trigger_entry_cross(conn, row):
     tc_en, tc_h, _ = time_close_settings_from_row(row)
 
     kid = int(row["id"])
-    conn.execute("DELETE FROM key_monitors WHERE id=?", (kid,))
+    if not acquire_trigger_entry_exec_lock(conn, kid):
+        return False, "触价开仓进行中"
     conn.commit()
 
     try:
@@ -5035,6 +5128,8 @@ def _execute_trigger_entry_cross(conn, row):
             time_close_hours=tc_h,
         )
     except Exception as e:
+        release_trigger_entry_exec_lock(conn, kid)
+        conn.commit()
         fail_msg = friendly_exchange_error(e)
         send_wechat_msg(
             f"# ❌ {symbol} 触价开仓异常\n"
@@ -5046,6 +5141,8 @@ def _execute_trigger_entry_cross(conn, row):
         return False, fail_msg
 
     if ok and det:
+        conn.execute("DELETE FROM key_monitors WHERE id=?", (kid,))
+        conn.commit()
         rr_txt = format_wechat_scalar_2dp(det.get("planned_rr_fill")) if det.get("planned_rr_fill") is not None else "-"
         msg = (
             f"# ✅ {symbol} 触价开仓成交\n"
@@ -5062,6 +5159,8 @@ def _execute_trigger_entry_cross(conn, row):
         send_wechat_msg(msg)
         insert_key_monitor_history(conn, row, 0, msg, TRIGGER_ENTRY_CLOSE_FILLED)
         return True, None
+    release_trigger_entry_exec_lock(conn, kid)
+    conn.commit()
     fail_msg = err or "触价触发后开仓失败"
     send_wechat_msg(
         f"# ❌ {symbol} 触价开仓失败\n"
@@ -5089,6 +5188,8 @@ def check_trigger_entry_key_monitors():
         sl = float(_sqlite_row_val(r, "fib_stop_loss") or 0)
         tp = float(_sqlite_row_val(r, "fib_take_profit") or 0)
         kid = int(r["id"])
+        if is_trigger_entry_in_flight_row(r):
+            continue
         if entry <= 0 or sl <= 0 or tp <= 0:
             _finalize_key_monitor_one_shot(conn, r, "触价计划价位无效", "fib_plan_invalid")
             continue
@@ -5856,7 +5957,7 @@ def check_order_monitors():
                             new_sl = round_price_to_exchange(ex_sym, new_sl)
                             tp_ex = float(take_profit or 0)
                             ok_live, _live_reason = ensure_okx_live_ready()
-                            synced_ex = not ok_live
+                            synced_ex = False
                             last_ex_sync = float(_BREAKEVEN_LAST_EX_SYNC.get(pid, 0))
                             interval_ok = (
                                 time.time() - last_ex_sync
@@ -6265,8 +6366,8 @@ def background_task():
                 from lib.strategy.strategy_trend_register import check_trend_pullback_plans
 
                 check_trend_pullback_plans(cfg)
-        except:
-            pass
+        except Exception as e:
+            print(f"[monitor_loop] {e}", flush=True)
         time.sleep(MONITOR_POLL_SECONDS)
 
 
@@ -9051,6 +9152,7 @@ try:
         ohlcv_fn=_hub_fetch_ohlcv,
         volume_rank_fn=_hub_fetch_volume_rank,
         market_fn=_hub_fetch_market,
+        reconcile_hub_flat_fn=reconcile_hub_external_close,
         risk_status_fn=hub_account_risk_status,
         user_close_fn=hub_user_initiated_close,
         render_main_page_fn=render_main_page,

lib/hub/hub_reconcile_flat_lib.py

@@ -0,0 +1,95 @@
+"""Hub 中控市价全平后立即同步 order_monitors（三所共用）。"""
+from __future__ import annotations
+
+import time
+from typing import Any, Callable
+
+
+def reconcile_hub_external_close_impl(
+    conn,
+    symbol: str,
+    direction: str,
+    *,
+    exchange_configured: Callable[[], bool],
+    not_configured_msg: str,
+    symbols_match: Callable[[str, str], bool],
+    get_opened_at_value: Callable[[Any], str],
+    resolve_monitor_exchange_symbol: Callable[[Any], str],
+    get_live_position_contracts: Callable[[str, str], float | None],
+    cancel_conditional_orders: Callable[[str], None],
+    resolve_synced_flat_close: Callable[..., tuple],
+    finalize_stopped_monitor: Callable[..., None],
+    sync_trade_records: Callable[..., None] | None = None,
+    reconcile_flat_streak: dict | None = None,
+    to_ms_with_fallback: Callable[..., int | None] | None = None,
+    prefer_manual_resolve: bool = False,
+    order_row_monitor_type: Callable[[Any], str] | None = None,
+) -> dict[str, Any]:
+    if not exchange_configured():
+        return {"ok": False, "msg": not_configured_msg, "synced": 0}
+    sym_req = (symbol or "").strip()
+    dir_l = (direction or "").strip().lower()
+    if dir_l not in ("long", "short"):
+        return {"ok": False, "msg": "side 须为 long 或 short", "synced": 0}
+    synced = 0
+    streak = reconcile_flat_streak if reconcile_flat_streak is not None else {}
+    rows = conn.execute(
+        "SELECT * FROM order_monitors WHERE status IN ('active', 'error')"
+    ).fetchall()
+    for r in rows:
+        if not symbols_match(str(r["symbol"] or ""), sym_req):
+            continue
+        if (r["direction"] or "").strip().lower() != dir_l:
+            continue
+        oid = int(r["id"])
+        if r["status"] == "error":
+            opened_at_chk = get_opened_at_value(r)
+            mtype = order_row_monitor_type(r) if order_row_monitor_type else r["monitor_type"]
+            existing = conn.execute(
+                "SELECT id FROM trade_records WHERE symbol=? AND opened_at=? AND monitor_type=? LIMIT 1",
+                (r["symbol"], opened_at_chk, mtype),
+            ).fetchone()
+            if existing:
+                conn.execute("UPDATE order_monitors SET status='stopped' WHERE id=?", (oid,))
+                synced += 1
+                continue
+        exchange_symbol = resolve_monitor_exchange_symbol(r)
+        live_contracts = get_live_position_contracts(exchange_symbol, r["direction"])
+        if live_contracts is None:
+            continue
+        if live_contracts > 0:
+            time.sleep(0.6)
+            live_contracts = get_live_position_contracts(exchange_symbol, r["direction"])
+        if live_contracts is None or live_contracts > 0:
+            continue
+        streak.pop(oid, None)
+        cancel_conditional_orders(exchange_symbol)
+        opened_at = get_opened_at_value(r)
+        opened_at_ms = None
+        if to_ms_with_fallback is not None:
+            keys = r.keys() if hasattr(r, "keys") else ()
+            opened_at_ms = to_ms_with_fallback(
+                r["opened_at_ms"] if "opened_at_ms" in keys else None,
+                opened_at,
+            )
+        resolve_kw = {"opened_at_ms": opened_at_ms}
+        if prefer_manual_resolve:
+            resolve_kw["prefer_manual"] = True
+        result, pnl_amount, closed_at, miss_reason = resolve_synced_flat_close(
+            r, opened_at, **resolve_kw
+        )
+        finalize_stopped_monitor(
+            conn,
+            r,
+            result=result,
+            pnl_amount=pnl_amount,
+            closed_at=closed_at,
+            miss_reason=miss_reason,
+        )
+        synced += 1
+    if sync_trade_records is not None:
+        try:
+            sync_trade_records(conn, force=True)
+        except Exception:
+            pass
+    return {"ok": True, "synced": synced}

lib/key_monitor/trigger_entry_key_monitor_lib.py

@@ -37,6 +37,34 @@ KEY_ENTRY_REASON_CALLBACK = "关键位回调触价开仓"
 KEY_ENTRY_REASON_BREAKOUT = "关键位突破触价开仓"
 KEY_ENTRY_REASON_TRIGGER_LEGACY = "关键位触价开仓"
 
+TRIGGER_ENTRY_IN_FLIGHT_OID = "__trigger_entry_in_flight__"
+
+
+def is_trigger_entry_in_flight_row(row: Any) -> bool:
+    if row is None:
+        return False
+    try:
+        v = row["fib_limit_order_id"]
+    except (KeyError, IndexError, TypeError):
+        v = getattr(row, "fib_limit_order_id", None)
+    return (v or "").strip() == TRIGGER_ENTRY_IN_FLIGHT_OID
+
+
+def acquire_trigger_entry_exec_lock(conn: Any, monitor_id: int) -> bool:
+    cur = conn.execute(
+        "UPDATE key_monitors SET fib_limit_order_id=? WHERE id=? "
+        "AND (fib_limit_order_id IS NULL OR fib_limit_order_id='')",
+        (TRIGGER_ENTRY_IN_FLIGHT_OID, int(monitor_id)),
+    )
+    return int(cur.rowcount or 0) == 1
+
+
+def release_trigger_entry_exec_lock(conn: Any, monitor_id: int) -> None:
+    conn.execute(
+        "UPDATE key_monitors SET fib_limit_order_id=NULL WHERE id=? AND fib_limit_order_id=?",
+        (int(monitor_id), TRIGGER_ENTRY_IN_FLIGHT_OID),
+    )
+
 
 def normalize_trigger_entry_monitor_type(monitor_type: Optional[str]) -> str:
     mt = (monitor_type or "").strip()

lib/strategy/strategy_roll_monitor_lib.py

@@ -40,7 +40,8 @@ def check_roll_monitors(cfg: dict[str, Any]) -> None:
         _reconcile_roll_groups(conn, cfg)
         _check_pending_roll_legs(conn, cfg)
         conn.commit()
-    except Exception:
+    except Exception as e:
+        print(f"[roll_monitor] {e}", flush=True)
         try:
             conn.rollback()
         except Exception:
@@ -408,7 +409,19 @@ def _execute_pending_roll_leg(
         return
 
     oid = str(order.get("id") or "") if isinstance(order, dict) else ""
-    cfg["replace_tpsl"](ex_sym, direction, sl, tp0, mon)
+    try:
+        cfg["replace_tpsl"](ex_sym, direction, sl, tp0, mon)
+    except Exception as tpsl_err:
+        fe = cfg.get("friendly_error")
+        msg = fe(tpsl_err) if callable(fe) else str(tpsl_err)
+        conn.execute(
+            """UPDATE roll_legs SET status='error', exchange_order_id=?, fill_price=?, amount=?
+               WHERE id=? AND status='pending'""",
+            (oid, fill, float(amount), leg_id),
+        )
+        _notify_roll_fail(cfg, group, leg, mark, f"加仓成交但止盈止损更新失败: {msg}")
+        return
+
     conn.execute(
         """UPDATE roll_legs SET status='filled', fill_price=?, amount=?, exchange_order_id=?,
            new_stop_loss=? WHERE id=? AND status='pending'""",

lib/strategy/strategy_trend_register.py

@@ -244,7 +244,7 @@ def _row(cfg, row) -> dict:
     return cfg["row_to_dict"](row)
 
 
-def precheck_trend_start(cfg: dict, conn) -> tuple[bool, str]:
+def precheck_trend_start(cfg: dict, conn, *, symbol: str = "", direction: str = "long") -> tuple[bool, str]:
     m = _m(cfg)
     mode = getattr(m, "POSITION_SIZING_MODE", None) or "risk"
     try:
@@ -255,9 +255,41 @@ def precheck_trend_start(cfg: dict, conn) -> tuple[bool, str]:
             return False, src_msg
     except Exception:
         pass
-    now = m.app_now()
-    if not m.trading_day_reset_allows_new_open(now):
-        return False, f"北京时间 {cfg['reset_hour']}:00 前不允许持仓"
+    sym = (symbol or "").strip()
+    dir_l = (direction or "long").strip().lower()
+    if sym and dir_l in ("long", "short") and hasattr(m, "precheck_risk"):
+        ok_risk, risk_msg = m.precheck_risk(conn, sym, dir_l)
+        if not ok_risk:
+            return False, risk_msg
+    else:
+        now = m.app_now()
+        if not m.trading_day_reset_allows_new_open(now):
+            return False, f"北京时间 {cfg['reset_hour']}:00 前不允许持仓"
+        from lib.trade.account_risk_lib import account_risk_blocks_trading, position_limit_reached
+
+        ok_risk, risk_reason = account_risk_blocks_trading(
+            conn,
+            trading_day=m.get_trading_day(now),
+            now=now,
+            fmt_local_ms=getattr(m, "ms_to_app_local_str", lambda _x: ""),
+        )
+        if not ok_risk:
+            return False, risk_reason
+        reached, active_count, mx = position_limit_reached(
+            conn, max_active_positions=cfg["max_active_positions"]
+        )
+        if reached:
+            return False, f"已达最大持仓数（{active_count}/{mx}）"
+        from lib.trade.daily_open_limit_lib import check_daily_open_hard_limit
+
+        ok_daily, daily_reason, _opens = check_daily_open_hard_limit(
+            conn,
+            m.get_trading_day(now),
+            getattr(m, "DAILY_OPEN_HARD_LIMIT", 0),
+            cfg["reset_hour"],
+        )
+        if not ok_daily:
+            return False, daily_reason
     active = m.get_active_position_count(conn)
     if active >= cfg["max_active_positions"]:
         return (
@@ -1604,22 +1636,27 @@ def register_trend_routes(app: Flask, cfg: dict) -> None:
     def preview_trend_pullback():
         conn = get_db()
         init_strategy_tables(conn)
-        okp, msg = precheck_trend_start(cfg, conn)
-        if not okp:
-            conn.close()
-            flash(msg)
-            return _redirect_trend()
         m = _m(cfg)
-        ok_live, reason = m.ensure_exchange_live_ready()
-        if not ok_live:
-            conn.close()
-            flash(reason)
-            return _redirect_trend()
         payload, err = parse_trend_plan(cfg, request.form)
         if err:
             conn.close()
             flash(err)
             return _redirect_trend()
+        okp, msg = precheck_trend_start(
+            cfg,
+            conn,
+            symbol=str(payload.get("symbol") or ""),
+            direction=str(payload.get("direction") or "long"),
+        )
+        if not okp:
+            conn.close()
+            flash(msg)
+            return _redirect_trend()
+        ok_live, reason = m.ensure_exchange_live_ready()
+        if not ok_live:
+            conn.close()
+            flash(reason)
+            return _redirect_trend()
         pid = str(uuid.uuid4())
         exp_ms = int(time.time() * 1000) + cfg["preview_ttl"] * 1000
         created = m.app_now_str()
@@ -1678,7 +1715,12 @@ def register_trend_routes(app: Flask, cfg: dict) -> None:
             conn.close()
             flash("预览已过期或不存在，请重新生成预览")
             return _redirect_trend()
-        okp, msg = precheck_trend_start(cfg, conn)
+        okp, msg = precheck_trend_start(
+            cfg,
+            conn,
+            symbol=str(pr["symbol"] or ""),
+            direction=str(pr["direction"] or "long"),
+        )
         if not okp:
             conn.close()
             flash(msg)
@@ -1718,7 +1760,18 @@ def register_trend_routes(app: Flask, cfg: dict) -> None:
                 exchange_symbol, direction, first_amt, leverage, stop_loss=None, take_profit=None
             )
             fill1 = m.resolve_order_entry_price(o1, exchange_symbol, live_price)
-            trend_refresh_stop_only(cfg, exchange_symbol, direction, stop_loss)
+            try:
+                trend_refresh_stop_only(cfg, exchange_symbol, direction, stop_loss)
+            except Exception as sl_err:
+                from lib.strategy.strategy_trend_exchange import cancel_symbol_orders, trend_market_close
+
+                try:
+                    pos_qty = m.get_live_position_contracts(exchange_symbol, direction) or first_amt
+                    trend_market_close(cfg, exchange_symbol, direction, float(pos_qty), leverage)
+                    cancel_symbol_orders(cfg, exchange_symbol)
+                except Exception as close_err:
+                    print(f"[trend_start] compensating close failed: {close_err}", flush=True)
+                raise sl_err
         except Exception as e:
             conn.close()
             fe = getattr(m, "friendly_exchange_error", lambda x, **k: str(x))

lib/trade/compensating_close_lib.py

@@ -0,0 +1,16 @@
+"""开仓后挂 TP/SL 失败时的补偿平仓（避免裸仓）。"""
+from __future__ import annotations
+
+from typing import Callable
+
+
+def log_compensating_close_error(prefix: str, exc: BaseException) -> None:
+    print(f"[{prefix}] {exc}", flush=True)
+
+
+def run_compensating_close(close_fn: Callable[[], None], *, log_prefix: str = "compensating_close") -> None:
+    """执行补偿平仓；二次失败只打日志，不掩盖原始异常。"""
+    try:
+        close_fn()
+    except Exception as e:
+        log_compensating_close_error(log_prefix, e)

manual_trading_hub/agent.py

@@ -868,7 +868,7 @@ def emergency_close_position(
     for p in raw:
         if not isinstance(p, dict):
             continue
-        if (p.get("symbol") or "").strip() != sym:
+        if not symbols_match(sym, (p.get("symbol") or "").strip()):
             continue
         c = _position_contracts(p)
         if abs(c) < 1e-12:

manual_trading_hub/exchange_orders.py

@@ -453,8 +453,11 @@ def cancel_orders_for_symbol(
         try:
             cancel_order(ex, exchange_kind, symbol, o["id"], o.get("channel") or "regular")
             n += 1
-        except Exception:
-            pass
+        except Exception as e:
+            print(
+                f"[cancel_orders_for_symbol] {exchange_kind} {symbol} id={o.get('id')}: {e}",
+                flush=True,
+            )
     return n
 
 
@@ -655,7 +658,7 @@ def _gate_place_tp_sl_position(
             try:
                 ex.privateFuturesPostSettlePriceOrders(_payload(tp_s, tp_rule))
             except Exception:
-                cancel_orders_for_symbol(ex, "gate", symbol, scope="conditional")
+                # 保留已挂止损，仅放弃本次 TP
                 raise
             return
         except Exception as e:

manual_trading_hub/hub.py

@@ -1535,6 +1535,50 @@ async def _notify_instance_user_close(
     )
 
 
+async def _sync_flask_after_position_close(
+    client: httpx.AsyncClient,
+    ex: dict,
+    *,
+    symbol: str,
+    side: str,
+) -> dict:
+    """中控/agent 平仓后同步 Flask order_monitors、趋势与滚仓状态。"""
+    sym = (symbol or "").strip()
+    side_l = (side or "").strip().lower()
+    out: dict = {}
+    if not sym or side_l not in ("long", "short"):
+        return out
+    order_sync = await _fetch_flask_json(
+        client,
+        ex,
+        "/api/hub/order/sync-flat",
+        method="POST",
+        json_body={"symbol": sym, "side": side_l},
+    )
+    if isinstance(order_sync, dict):
+        out["order_sync"] = order_sync
+    if "trend" in (ex.get("capabilities") or []):
+        sync_parsed = await _fetch_flask_json(
+            client,
+            ex,
+            "/api/hub/trend/sync-flat",
+            method="POST",
+            json_body={"symbol": sym, "side": side_l},
+        )
+        if isinstance(sync_parsed, dict):
+            out["trend_sync"] = sync_parsed
+    roll_sync = await _fetch_flask_json(
+        client,
+        ex,
+        "/api/hub/roll/sync-flat",
+        method="POST",
+        json_body={"symbol": sym, "side": side_l},
+    )
+    if isinstance(roll_sync, dict):
+        out["roll_sync"] = roll_sync
+    return out
+
+
 def _flask_error_from_hub_mon(hub_mon: dict | None) -> str | None:
     if not isinstance(hub_mon, dict) or hub_mon.get("ok") is not False:
         return None
@@ -2346,37 +2390,11 @@ async def api_close_position(exchange_id: str, body: ClosePositionBody):
             "ok": bool(isinstance(payload, dict) and payload.get("ok")),
         }
         if out.get("ok"):
-            ex_key = (ex.get("key") or "").strip().lower()
             async with httpx.AsyncClient() as flask_client:
-                if ex_key == "gate":
-                    order_sync = await _fetch_flask_json(
-                        flask_client,
-                        ex,
-                        "/api/hub/order/sync-flat",
-                        method="POST",
-                        json_body={"symbol": sym, "side": side},
-                    )
-                    if isinstance(order_sync, dict):
-                        out["order_sync"] = order_sync
-                if "trend" in (ex.get("capabilities") or []):
-                    sync_parsed = await _fetch_flask_json(
-                        flask_client,
-                        ex,
-                        "/api/hub/trend/sync-flat",
-                        method="POST",
-                        json_body={"symbol": sym, "side": side},
-                    )
-                    if isinstance(sync_parsed, dict):
-                        out["trend_sync"] = sync_parsed
-                roll_sync = await _fetch_flask_json(
-                    flask_client,
-                    ex,
-                    "/api/hub/roll/sync-flat",
-                    method="POST",
-                    json_body={"symbol": sym, "side": side},
+                sync_bundle = await _sync_flask_after_position_close(
+                    flask_client, ex, symbol=sym, side=side
                 )
-                if isinstance(roll_sync, dict):
-                    out["roll_sync"] = roll_sync
+                out.update(sync_bundle)
                 risk_sync = await _notify_instance_user_close(flask_client, ex, count=1)
                 if isinstance(risk_sync, dict):
                     out["risk_sync"] = risk_sync
@@ -2414,6 +2432,14 @@ async def api_place_tpsl(exchange_id: str, body: PlaceTpslBody):
             "ok": bool(isinstance(payload, dict) and payload.get("ok")),
         }
         if out.get("ok") and (ex.get("flask_url") or "").strip():
+            placed = payload.get("placed") if isinstance(payload, dict) else None
+            sl_sync = body.stop_loss
+            tp_sync = body.take_profit
+            if isinstance(placed, dict):
+                if placed.get("stop_loss") is not None:
+                    sl_sync = placed["stop_loss"]
+                if placed.get("take_profit") is not None:
+                    tp_sync = placed["take_profit"]
             async with httpx.AsyncClient() as flask_client:
                 sync_parsed = await _fetch_flask_json(
                     flask_client,
@@ -2423,8 +2449,8 @@ async def api_place_tpsl(exchange_id: str, body: PlaceTpslBody):
                     json_body={
                         "symbol": body.symbol,
                         "side": body.side,
-                        "stop_loss": body.stop_loss,
-                        "take_profit": body.take_profit,
+                        "stop_loss": sl_sync,
+                        "take_profit": tp_sync,
                     },
                 )
                 if isinstance(sync_parsed, dict):
@@ -2451,9 +2477,19 @@ async def api_close_exchange(exchange_id: str):
             closed = body.get("closed") or []
             n = len(closed) if isinstance(closed, list) else 0
             if n > 0:
-                risk_sync = await _notify_instance_user_close(client, ex, count=n)
-                if isinstance(risk_sync, dict):
-                    out["risk_sync"] = risk_sync
+                async with httpx.AsyncClient() as flask_client:
+                    for item in closed:
+                        if not isinstance(item, dict):
+                            continue
+                        sym_i = (item.get("symbol") or "").strip()
+                        side_i = (item.get("side") or "").strip().lower()
+                        if sym_i and side_i in ("long", "short"):
+                            await _sync_flask_after_position_close(
+                                flask_client, ex, symbol=sym_i, side=side_i
+                            )
+                    risk_sync = await _notify_instance_user_close(flask_client, ex, count=n)
+                    if isinstance(risk_sync, dict):
+                        out["risk_sync"] = risk_sync
         _schedule_board_refresh()
         return out
 
@@ -2478,6 +2514,17 @@ async def api_close_all(body: CloseAllBody | None = Body(default=None)):
                     closed = payload.get("closed") or []
                     n = len(closed) if isinstance(closed, list) else 0
                     if n > 0:
+                        for item in closed:
+                            if not isinstance(item, dict):
+                                continue
+                            sym_i = (item.get("symbol") or "").strip()
+                            side_i = (item.get("side") or "").strip().lower()
+                            if sym_i and side_i in ("long", "short"):
+                                sync_bundle = await _sync_flask_after_position_close(
+                                    client, ex, symbol=sym_i, side=side_i
+                                )
+                                if sync_bundle:
+                                    row["flask_sync"] = sync_bundle
                         risk_sync = await _notify_instance_user_close(client, ex, count=n)
                         if isinstance(risk_sync, dict):
                             row["risk_sync"] = risk_sync