修复okx 趋势回调

crypto_monitor_okx/app.py

@@ -2672,7 +2672,36 @@ def _position_matches_wanted_contract(exchange_symbol, position):
     if not position:
         return False
     sym = position.get("symbol")
-    return sym == exchange_symbol
+    if sym == exchange_symbol:
+        return True
+    try:
+        return normalize_okx_symbol(sym or "") == normalize_okx_symbol(exchange_symbol or "")
+    except Exception:
+        return False
+
+
+def _fetch_okx_swap_position_rows(exchange_symbol=None):
+    """OKX 单合约 fetch_positions([sym]) 常返回空；与 /api/prices 一致拉全量 SWAP 再本地匹配。"""
+    ensure_markets_loaded()
+    rows = None
+    for fetcher in (
+        lambda: exchange.fetch_positions(None, {"instType": OKX_POSITION_INST_TYPE}),
+        lambda: exchange.fetch_positions(),
+    ):
+        try:
+            rows = fetcher() or []
+            break
+        except Exception:
+            continue
+    if rows is None:
+        return None
+    if not exchange_symbol:
+        return rows
+    out = []
+    for p in rows:
+        if _position_matches_wanted_contract(exchange_symbol, p):
+            out.append(p)
+    return out
 
 
 def _select_live_position_row(rows, exchange_symbol, direction, relax_hedge=False):
@@ -2975,35 +3004,15 @@ def is_no_position_error(err_msg):
 
 
 def get_live_position_contracts(exchange_symbol, direction):
-    ensure_markets_loaded()
+    rows = _fetch_okx_swap_position_rows(exchange_symbol)
-    try:
+    if rows is None:
-        rows = exchange.fetch_positions([exchange_symbol], params={"instType": OKX_POSITION_INST_TYPE})
-    except Exception:
         return None
-    total = 0.0
+    if not rows:
-    for p in rows:
+        return 0.0
-        if p.get("symbol") != exchange_symbol:
+    prow = _select_live_position_row(rows, exchange_symbol, direction)
-            continue
+    if not prow:
-        info = p.get("info", {}) or {}
+        return 0.0
-        side = (p.get("side") or info.get("posSide") or "").lower()
+    return _position_row_effective_contracts(prow)
-        contracts = p.get("contracts")
-        if contracts is None:
-            raw_pos = info.get("pos")
-            try:
-                contracts = abs(float(raw_pos)) if raw_pos is not None else 0.0
-            except Exception:
-                contracts = 0.0
-        try:
-            contracts = float(contracts)
-        except Exception:
-            contracts = 0.0
-        if contracts <= 0:
-            continue
-        if OKX_POS_MODE == "hedge":
-            if side and side != direction:
-                continue
-        total += contracts
-    return total
 
 
 def opened_at_str_to_ms(opened_at_str):

strategy_trend_register.py

@@ -32,6 +32,11 @@ from strategy_trade_labels import (
 
 MONITOR_TYPE_TREND = MONITOR_TYPE_TREND_PULLBACK
 
+# 趋势回调：交易所报空仓需连续 N 次轮询确认，避免 OKX 等 API 瞬时误判立即结束计划
+_TREND_FLAT_STREAK: dict[int, int] = {}
+TREND_FLAT_CONFIRM_POLLS = max(1, int(os.getenv("TREND_FLAT_CONFIRM_POLLS", "3")))
+TREND_OPEN_GRACE_SEC = max(0, int(os.getenv("TREND_OPEN_GRACE_SEC", "90")))
+
 
 def trend_add_zone_label(direction: str) -> str:
     return "补仓下沿" if (direction or "long").strip().lower() == "short" else "补仓上沿"
@@ -402,6 +407,36 @@ def _finalize_plan(cfg: dict, conn, row, result_label: str, exit_price: float) -
     conn.commit()
 
 
+def _trend_plan_open_age_sec(row, m) -> float:
+    opened_ms = None
+    try:
+        if "opened_at_ms" in row.keys() and row["opened_at_ms"]:
+            opened_ms = int(row["opened_at_ms"])
+    except Exception:
+        opened_ms = None
+    to_ms = getattr(m, "_to_ms_with_fallback", None)
+    if callable(to_ms):
+        opened_ms = to_ms(opened_ms, row["opened_at"] if "opened_at" in row.keys() else None)
+    if not opened_ms:
+        return 999999.0
+    return max(0.0, (time.time() * 1000 - opened_ms) / 1000.0)
+
+
+def _should_finalize_trend_flat(row, pos, plan_id: int, m) -> bool:
+    """首仓后交易所报无仓：需过开仓宽限期 + 连续空仓轮询，避免误判止损。"""
+    if pos is None or float(pos) > 0:
+        _TREND_FLAT_STREAK.pop(plan_id, None)
+        return False
+    if not int(row["first_order_done"] or 0):
+        return False
+    if _trend_plan_open_age_sec(row, m) < TREND_OPEN_GRACE_SEC:
+        _TREND_FLAT_STREAK.pop(plan_id, None)
+        return False
+    streak = int(_TREND_FLAT_STREAK.get(plan_id, 0)) + 1
+    _TREND_FLAT_STREAK[plan_id] = streak
+    return streak >= TREND_FLAT_CONFIRM_POLLS
+
+
 def check_trend_pullback_plans(cfg: dict) -> None:
     m = _m(cfg)
     ok_live, _ = m.ensure_exchange_live_ready()
@@ -413,6 +448,7 @@ def check_trend_pullback_plans(cfg: dict) -> None:
     ).fetchall()
     for row in rows:
         try:
+            plan_id = int(row["id"])
             sym = row["symbol"]
             direction = (row["direction"] or "long").lower()
             ex_sym = row["exchange_symbol"] or m.normalize_exchange_symbol(sym)
@@ -447,9 +483,11 @@ def check_trend_pullback_plans(cfg: dict) -> None:
                         continue
                     exit_p = pf
                 _finalize_plan(cfg, conn, row, "止盈", exit_p)
+                _TREND_FLAT_STREAK.pop(plan_id, None)
                 continue
-            if pos <= 0 and int(row["first_order_done"] or 0):
+            if _should_finalize_trend_flat(row, pos, plan_id, m):
                 _finalize_plan(cfg, conn, row, "止损", pf)
+                _TREND_FLAT_STREAK.pop(plan_id, None)
                 continue
             if int(row["first_order_done"] or 0) and legs_done < len(grid) and legs_done < len(leg_amounts):
                 level = float(grid[legs_done])