fix(trend): surface DCA block reasons and ensure gate_bot poll thread

Log poll exceptions, diagnose live-trading and mark-price blocks on the trend page, start background monitors on app import, and add /api/trend_poll_status for debugging. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-07 17:46:26 +08:00
parent 0760873d9d
commit f5b4513ddb
3 changed files with 213 additions and 7 deletions
@@ -5270,11 +5270,27 @@ def background_task():

                check_roll_monitors(_roll_cfg)
            check_order_monitors()
-        except:
-            pass
+        except Exception as e:
+            print(f"[background_task] error: {e}", flush=True)
        time.sleep(MONITOR_POLL_SECONDS)


+_BG_MONITORS_LOCK = threading.Lock()
+_BG_MONITORS_STARTED = False
+
+
+def _ensure_background_monitors_started():
+    global _BG_MONITORS_STARTED
+    with _BG_MONITORS_LOCK:
+        if _BG_MONITORS_STARTED:
+            return
+        _BG_MONITORS_STARTED = True
+        threading.Thread(
+            target=background_task, daemon=True, name="gate-bot-monitors"
+        ).start()
+        print("[startup] background monitor thread started", flush=True)
+
+
 # ====================== 登录路由 ======================
@app.route("/login", methods=["GET", "POST"])
 def login():
@@ -5400,12 +5416,26 @@ def render_main_page(page="trade"):
        "SELECT * FROM trend_pullback_plans WHERE status='active' ORDER BY id DESC"
    ).fetchall()
    trend_plans = []
+    trend_dca_probes = []
+    _trend_cfg = app.extensions.get("strategy_trend_cfg")
    for r in trend_plans_raw:
        try:
-            trend_plans.append(enrich_active_trend_plan_row(r))
+            enriched = enrich_active_trend_plan_row(r)
+            trend_plans.append(enriched)
        except Exception as e:
            print(f"[render_main_page] enrich trend plan: {e}")
-            trend_plans.append(row_to_dict(r))
+            enriched = row_to_dict(r)
+            trend_plans.append(enriched)
+        if _trend_cfg and page in ("strategy", "strategy_trend", "strategy_roll"):
+            try:
+                from strategy_trend_register import summarize_trend_dca_probe
+
+                probe = summarize_trend_dca_probe(_trend_cfg, r)
+                trend_dca_probes.append(probe)
+                if isinstance(enriched, dict):
+                    enriched["dca_probe"] = probe
+            except Exception as e:
+                print(f"[render_main_page] trend dca probe: {e}")
    preview_snapshots = []
    if page == "records":
        try:
@@ -5511,6 +5541,8 @@ def render_main_page(page="trade"):
        manual_min_planned_rr=MANUAL_MIN_PLANNED_RR,
        can_trade=can_trade,
        trend_plans=trend_plans,
+        trend_dca_probes=trend_dca_probes,
+        live_trading_enabled=LIVE_TRADING_ENABLED,
        preview_snapshots=preview_snapshots,
        exchange_sync_from_label=(EXCHANGE_POSITION_SYNC_FROM_BJ or "最近90天"),
        trend_pullback_dca_legs=TREND_PULLBACK_DCA_LEGS,
@@ -8017,6 +8049,35 @@ def strategy_trend_page():
    return redirect(f"/strategy?{qs}" if qs else "/strategy")


+@app.route("/api/trend_poll_status")
+@login_required
+def api_trend_poll_status():
+    from strategy_trend_register import (
+        build_trend_config,
+        get_trend_poll_state,
+        summarize_trend_dca_probe,
+    )
+
+    cfg = app.extensions.get("strategy_trend_cfg") or build_trend_config(
+        sys.modules[__name__]
+    )
+    conn = get_db()
+    probes = []
+    for r in conn.execute(
+        "SELECT * FROM trend_pullback_plans WHERE status='active' ORDER BY id DESC"
+    ).fetchall():
+        probes.append(summarize_trend_dca_probe(cfg, r))
+    conn.close()
+    return jsonify(
+        {
+            "poll": get_trend_poll_state(),
+            "probes": probes,
+            "live_trading_enabled": LIVE_TRADING_ENABLED,
+            "monitor_poll_seconds": MONITOR_POLL_SECONDS,
+        }
+    )
+
+
@app.route("/strategy/roll")
@login_required
 def strategy_roll_page():
@@ -8035,9 +8096,9 @@ install_strategy_trading(
 app.extensions["strategy_trend_cfg"] = build_trend_config(sys.modules[__name__])

 _purge_key_monitors_if_full_margin()
+_ensure_background_monitors_started()


 # 启动
 if __name__ == "__main__":
-    threading.Thread(target=background_task, daemon=True).start()
    app.run(host=HOST, port=PORT, debug=DEBUG)
@@ -7,6 +7,19 @@
        ② <strong>确认执行</strong>：市价首仓 50% + 挂交易所止损；首仓后可<strong>手动保本</strong>（默认均价+{{ trend_manual_breakeven_offset_pct }}%）；剩余 50% 在止损与补仓区间之间共 {{ trend_pullback_dca_legs }} 档（做多为<strong>上沿</strong>、做空为<strong>下沿</strong>；程序可能因最小张数自动减档）市价补仓；<strong>止盈由程序监控</strong>。<br>
        确认执行时若当前可用余额与预览快照相对偏差 &gt; <strong>{{ trend_preview_max_drift_pct }}%</strong> 会拒绝并要求重新预览。
    </div>
+    {% if trend_dca_probes %}
+    {% for p in trend_dca_probes %}
+    {% if p.trigger_reached and p.block_reason %}
+    <div class="rule-tip" style="margin-bottom:10px;border-color:#a55;background:#2a1818;color:#ffb4b4">
+        <strong>计划 #{{ p.plan_id }}</strong> 标记价 {{ p.mark_price }} 已触达补仓触发价 {{ p.next_trigger }}，但未自动补仓：
+        {{ p.block_reason }}。
+        {% if not live_trading_enabled %}
+        请在 <code>crypto_monitor_gate_bot/.env</code> 设置 <code>LIVE_TRADING_ENABLED=true</code> 后重启 PM2 进程 <strong>crypto_gate_bot</strong>（不是 manual-agent-gate-bot）。
+        {% endif %}
+    </div>
+    {% endif %}
+    {% endfor %}
+    {% endif %}
    <form id="trend-pullback-form" action="{{ url_for('preview_trend_pullback') }}" method="post" class="form-row">
        <input name="symbol" placeholder="BTC 或 ETH/USDT" required>
        <select name="direction" id="trend-direction" required>
@@ -41,6 +41,109 @@ MONITOR_TYPE_TREND = MONITOR_TYPE_TREND_PULLBACK
 _TREND_FLAT_STREAK: dict[int, int] = {}
 TREND_FLAT_CONFIRM_POLLS = max(1, int(os.getenv("TREND_FLAT_CONFIRM_POLLS", "5")))
 TREND_OPEN_GRACE_SEC = max(0, int(os.getenv("TREND_OPEN_GRACE_SEC", "180")))
+_TREND_LIVE_SKIP_LOG_TS = 0.0
+_TREND_POLL_STATE: dict[str, Any] = {
+    "updated_at": None,
+    "live_ok": True,
+    "live_reason": "",
+    "plans": {},
+}
+
+
+def get_trend_poll_state() -> dict:
+    return dict(_TREND_POLL_STATE or {})
+
+
+def _log_trend_live_skip(reason: str) -> None:
+    global _TREND_LIVE_SKIP_LOG_TS
+    now = time.time()
+    if now - _TREND_LIVE_SKIP_LOG_TS < 60:
+        return
+    _TREND_LIVE_SKIP_LOG_TS = now
+    print(f"[trend_pullback] poll skipped (live not ready): {reason}", flush=True)
+
+
+def _set_trend_poll_plan(plan_id: int, info: dict) -> None:
+    plans = dict(_TREND_POLL_STATE.get("plans") or {})
+    plans[str(plan_id)] = info
+    _TREND_POLL_STATE["plans"] = plans
+
+
+def summarize_trend_dca_probe(cfg: dict, row) -> dict:
+    """诊断单计划为何未补仓（供页面 / API）。"""
+    m = _m(cfg)
+    d = _row(cfg, row)
+    plan_id = int(d.get("id") or 0)
+    sym = d.get("symbol") or ""
+    direction = (d.get("direction") or "long").lower()
+    ex_sym = d.get("exchange_symbol") or m.normalize_exchange_symbol(sym)
+    out: dict[str, Any] = {
+        "plan_id": plan_id,
+        "symbol": sym,
+        "mark_price": None,
+        "next_trigger": None,
+        "trigger_reached": False,
+        "legs_done": int(d.get("legs_done") or 0),
+        "first_order_done": int(d.get("first_order_done") or 0),
+        "block_reason": None,
+    }
+    try:
+        legs_done = int(d.get("legs_done") or 0)
+        grid = json.loads(d.get("grid_prices_json") or "[]")
+        if not isinstance(grid, list):
+            grid = []
+        leg_amounts = json.loads(d.get("leg_amounts_json") or "[]")
+        if not isinstance(leg_amounts, list):
+            leg_amounts = []
+    except Exception:
+        grid = []
+        leg_amounts = []
+        legs_done = 0
+    pf = _trend_poll_price(m, sym, ex_sym, direction)
+    out["mark_price"] = pf
+    ok_live, live_reason = m.ensure_exchange_live_ready()
+    out["live_ok"] = ok_live
+    if not ok_live:
+        out["block_reason"] = live_reason or "实盘未就绪"
+    if not int(d.get("first_order_done") or 0):
+        out["block_reason"] = out["block_reason"] or "首仓未完成"
+        return out
+    if legs_done >= len(grid) or legs_done >= len(leg_amounts):
+        out["block_reason"] = out["block_reason"] or "补仓档已全部完成或无 grid"
+        return out
+    try:
+        level = float(grid[legs_done])
+    except (TypeError, ValueError, IndexError):
+        out["block_reason"] = out["block_reason"] or "无效补仓触发价"
+        return out
+    out["next_trigger"] = level
+    if pf is None:
+        out["block_reason"] = out["block_reason"] or "无法读取标记价"
+        return out
+    reached = trend_dca_level_reached(direction, float(pf), level)
+    out["trigger_reached"] = reached
+    if reached and not ok_live:
+        out["block_reason"] = live_reason or "LIVE_TRADING_ENABLED=false"
+    elif reached and ok_live:
+        pos = m.get_live_position_contracts(ex_sym, direction)
+        try:
+            local_open = float(d.get("order_amount_open") or 0)
+        except (TypeError, ValueError):
+            local_open = 0.0
+        if pos is None and local_open > 0:
+            pos = local_open
+        if pos is None:
+            out["block_reason"] = "无法读取交易所持仓"
+        elif float(pos) <= 0:
+            out["block_reason"] = "交易所无持仓"
+        else:
+            out["block_reason"] = (
+                "标记价已触达，轮询应自动下单；若仍未补请确认 PM2 进程 crypto_gate_bot "
+                "（非 manual-agent-gate-bot）在运行，并查看 pm2 logs crypto_gate_bot"
+            )
+    elif not reached:
+        out["block_reason"] = f"标记价 {pf} 未触达下一档 {level}"
+    return out


 def trend_add_zone_label(direction: str) -> str:
@@ -747,8 +850,24 @@ def _should_finalize_trend_flat(row, pos, plan_id: int, m) -> bool:

 def check_trend_pullback_plans(cfg: dict) -> None:
    m = _m(cfg)
-    ok_live, _ = m.ensure_exchange_live_ready()
+    ok_live, live_reason = m.ensure_exchange_live_ready()
+    _TREND_POLL_STATE["updated_at"] = time.time()
+    _TREND_POLL_STATE["live_ok"] = ok_live
+    _TREND_POLL_STATE["live_reason"] = live_reason or ""
    if not ok_live:
+        _log_trend_live_skip(live_reason or "unknown")
+        conn = cfg["get_db"]()
+        try:
+            for row in conn.execute(
+                "SELECT * FROM trend_pullback_plans WHERE status='active'"
+            ).fetchall():
+                probe = summarize_trend_dca_probe(cfg, row)
+                if probe.get("trigger_reached"):
+                    _set_trend_poll_plan(int(row["id"]), probe)
+        except Exception as e:
+            print(f"[trend_pullback] live-skip probe error: {e}", flush=True)
+        finally:
+            conn.close()
        return
    conn = cfg["get_db"]()
    rows = conn.execute(
@@ -867,7 +986,20 @@ def check_trend_pullback_plans(cfg: dict) -> None:
                "UPDATE trend_pullback_plans SET last_mark_price=? WHERE id=?",
                (pf, row["id"]),
            )
-        except Exception:
+            probe = summarize_trend_dca_probe(cfg, row)
+            probe["last_poll_mark"] = pf
+            _set_trend_poll_plan(plan_id, probe)
+            if probe.get("trigger_reached") and probe.get("block_reason"):
+                print(
+                    f"[trend_pullback] dca blocked plan={plan_id} sym={sym} "
+                    f"mark={pf} next={probe.get('next_trigger')} reason={probe.get('block_reason')}",
+                    flush=True,
+                )
+        except Exception as e:
+            print(
+                f"[trend_pullback] poll error plan={row['id'] if row else '?'}: {e}",
+                flush=True,
+            )
            continue
    conn.commit()
    conn.close()