修改持仓显示，读取历史仓位

crypto_monitor_gate/app.py

@@ -137,6 +137,9 @@ KEY_CONFIRM_BAR = int(os.getenv("KEY_CONFIRM_BAR", "-1"))
 KEY_SIZING_USE_ZERO_POSITION_SNAPSHOT = os.getenv("KEY_SIZING_USE_ZERO_POSITION_SNAPSHOT", "true").lower() == "true"
 ORDER_MONITOR_TYPE_MANUAL = "下单监控"
 ORDER_MONITOR_TYPE_KEY_AUTO = "关键位监控"
+EXCHANGE_POSITION_SYNC_FROM_BJ = (os.getenv("EXCHANGE_POSITION_SYNC_FROM_BJ") or "").strip()
+EXCHANGE_POSITION_HISTORY_LIMIT = max(50, min(1000, int(os.getenv("EXCHANGE_POSITION_HISTORY_LIMIT", "200"))))
+_LAST_EXCHANGE_PNL_SYNC_AT = 0.0
 
 KEY_MONITOR_AUTO_TYPES = frozenset({"箱体突破", "收敛突破"})
 KEY_MONITOR_ALERT_ONLY_TYPES = frozenset({"关键阻力位", "关键支撑位"})
@@ -1241,6 +1244,21 @@ def init_db():
         c.execute("ALTER TABLE trading_sessions ADD COLUMN key_sizing_capital_snapshot REAL")
     except Exception:
         pass
+    try:
+        c.execute("ALTER TABLE order_monitors ADD COLUMN key_signal_type TEXT")
+    except Exception:
+        pass
+    for ddl in (
+        "ALTER TABLE trade_records ADD COLUMN key_signal_type TEXT",
+        "ALTER TABLE trade_records ADD COLUMN exchange_realized_pnl REAL",
+        "ALTER TABLE trade_records ADD COLUMN exchange_opened_at TEXT",
+        "ALTER TABLE trade_records ADD COLUMN exchange_closed_at TEXT",
+        "ALTER TABLE trade_records ADD COLUMN exchange_sync_key TEXT",
+    ):
+        try:
+            c.execute(ddl)
+        except Exception:
+            pass
 
     c.execute(
         """CREATE TABLE IF NOT EXISTS key_monitor_history
@@ -1673,6 +1691,25 @@ def to_effective_trade_dict(row):
     item["effective_hold_seconds"] = get_effective_trade_field(row, "reviewed_hold_seconds", "hold_seconds", item.get("hold_seconds"))
     er_eff = get_effective_trade_field(row, "reviewed_entry_reason", "entry_reason", item.get("entry_reason"))
     item["effective_entry_reason"] = (str(er_eff).strip() if er_eff is not None else "") or ""
+    reviewed_pnl = get_effective_trade_field(row, "reviewed_pnl_amount", "pnl_amount", None)
+    has_reviewed_pnl = reviewed_pnl is not None and str(reviewed_pnl).strip() != ""
+    ex_pnl = item.get("exchange_realized_pnl")
+    if not has_reviewed_pnl and ex_pnl is not None and str(ex_pnl).strip() != "":
+        try:
+            item["effective_pnl_amount"] = round(float(ex_pnl), 2)
+            item["display_pnl_source"] = "exchange"
+            ex_open = (str(item.get("exchange_opened_at") or "").strip() or None)
+            ex_close = (str(item.get("exchange_closed_at") or "").strip() or None)
+            if ex_open:
+                item["effective_opened_at"] = ex_open
+            if ex_close:
+                item["effective_closed_at"] = ex_close
+        except (TypeError, ValueError):
+            item["display_pnl_source"] = "local"
+    elif has_reviewed_pnl:
+        item["display_pnl_source"] = "reviewed"
+    else:
+        item["display_pnl_source"] = "local"
     return item
 
 
@@ -1930,16 +1967,20 @@ def insert_trade_record(
     closed_at=None,
     closed_at_ms=None,
     exchange_trade_id=None,
+    key_signal_type=None,
 ):
     hold_minutes = calc_hold_minutes(hold_seconds)
     open_ts = opened_at or app_now_str()
     close_ts = closed_at or app_now_str()
     open_ts_ms = _to_ms_with_fallback(opened_at_ms, open_ts)
     close_ts_ms = _to_ms_with_fallback(closed_at_ms, close_ts)
+    kst = (key_signal_type or "").strip()
+    if kst not in KEY_MONITOR_AUTO_TYPES:
+        kst = None
     conn.execute(
-        "INSERT INTO trade_records (symbol,monitor_type,direction,trigger_price,stop_loss,initial_stop_loss,take_profit,margin_capital,leverage,pnl_amount,hold_seconds,trade_style,risk_amount,planned_rr,actual_rr,hold_minutes,opened_at,opened_at_ms,closed_at,closed_at_ms,result,miss_reason,exchange_trade_id) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        "INSERT INTO trade_records (symbol,monitor_type,key_signal_type,direction,trigger_price,stop_loss,initial_stop_loss,take_profit,margin_capital,leverage,pnl_amount,hold_seconds,trade_style,risk_amount,planned_rr,actual_rr,hold_minutes,opened_at,opened_at_ms,closed_at,closed_at_ms,result,miss_reason,exchange_trade_id) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
         (
-            symbol, monitor_type, direction, trigger_price, stop_loss, initial_stop_loss, take_profit,
+            symbol, monitor_type, kst, direction, trigger_price, stop_loss, initial_stop_loss, take_profit,
             margin_capital, leverage, pnl_amount, hold_seconds,
             trade_style, risk_amount, planned_rr, actual_rr, hold_minutes,
             open_ts, open_ts_ms, close_ts, close_ts_ms, result, miss_reason, exchange_trade_id
@@ -2011,6 +2052,33 @@ def ensure_exchange_live_ready():
     return True, ""
 
 
+def order_row_monitor_type(row):
+    if row is None:
+        return ORDER_MONITOR_TYPE_MANUAL
+    try:
+        keys = row.keys() if hasattr(row, "keys") else []
+    except Exception:
+        keys = []
+    if "monitor_type" in keys:
+        mt = (row["monitor_type"] or "").strip()
+        if mt:
+            return mt
+    return ORDER_MONITOR_TYPE_MANUAL
+
+
+def order_row_key_signal_type(row):
+    if row is None:
+        return None
+    try:
+        keys = row.keys() if hasattr(row, "keys") else []
+    except Exception:
+        keys = []
+    if "key_signal_type" not in keys:
+        return None
+    kst = (row["key_signal_type"] or "").strip()
+    return kst if kst in KEY_MONITOR_AUTO_TYPES else None
+
+
 def exchange_private_api_configured():
     """仅表示已配置密钥；与是否允许下单（LIVE_TRADING_ENABLED）无关，用于只读拉仓等。"""
     return bool(GATE_API_KEY and GATE_API_SECRET)
@@ -3475,7 +3543,8 @@ def reconcile_external_closes(conn, days=None):
         insert_trade_record(
             conn,
             symbol=r["symbol"],
-            monitor_type="下单监控",
+            monitor_type=order_row_monitor_type(r),
+            key_signal_type=order_row_key_signal_type(r),
             direction=r["direction"],
             trigger_price=r["trigger_price"],
             stop_loss=r["stop_loss"],
@@ -3779,7 +3848,7 @@ def _key_plan_auto_sl_tp(direction, upper, lower, checks, outside_pct):
     return E, sl_raw, tp_raw, H
 
 
-def _market_open_for_key_monitor(conn, symbol, direction, exchange_symbol, stop_loss, take_profit):
+def _market_open_for_key_monitor(conn, symbol, direction, exchange_symbol, stop_loss, take_profit, key_signal_type=None):
     """
     与手动「实盘下单」对齐的市价开仓与 order_monitors 写入。
     返回 (ok: bool, err_msg: Optional[str], detail: Optional[dict])
@@ -3898,8 +3967,8 @@ def _market_open_for_key_monitor(conn, symbol, direction, exchange_symbol, stop_
         "(symbol, exchange_symbol, direction, trigger_price, stop_loss, initial_stop_loss, take_profit, "
         "margin_capital, leverage, trade_style, risk_percent, risk_amount, "
         "breakeven_rr_trigger, breakeven_offset_pct, breakeven_step_r, breakeven_armed, breakeven_price, breakeven_enabled, "
-        "notional_value, position_ratio, base_amount, order_amount, exchange_order_id, opened_at, opened_at_ms, session_date, monitor_type) "
-        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        "notional_value, position_ratio, base_amount, order_amount, exchange_order_id, opened_at, opened_at_ms, session_date, monitor_type, key_signal_type) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
         (
             symbol,
             exchange_symbol,
@@ -3928,6 +3997,7 @@ def _market_open_for_key_monitor(conn, symbol, direction, exchange_symbol, stop_
             opened_at_ms,
             trading_day,
             ORDER_MONITOR_TYPE_KEY_AUTO,
+            (key_signal_type if key_signal_type in KEY_MONITOR_AUTO_TYPES else None),
         ),
     )
     new_order_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
@@ -4053,8 +4123,9 @@ def check_key_monitors():
             _finalize_key_monitor_one_shot(conn, r, rr_msg, "rr_insufficient")
             continue
 
+        key_sig = typ if typ in KEY_MONITOR_AUTO_TYPES else None
         ok_trade, trade_err, det = _market_open_for_key_monitor(
-            conn, sym, direction, exchange_symbol, sl_raw, tp_raw,
+            conn, sym, direction, exchange_symbol, sl_raw, tp_raw, key_signal_type=key_sig,
         )
         planned_rr_txt = (
             format_wechat_scalar_2dp(planned_rr) if planned_rr is not None else "-"
@@ -4301,7 +4372,8 @@ def check_order_monitors():
                     insert_trade_record(
                         conn,
                         symbol=sym,
-                        monitor_type="下单监控",
+                        monitor_type=order_row_monitor_type(r),
+                        key_signal_type=order_row_key_signal_type(r),
                         direction=direction,
                         trigger_price=trigger_price,
                         stop_loss=stop_loss,
@@ -4371,7 +4443,8 @@ def check_order_monitors():
             insert_trade_record(
                 conn,
                 symbol=sym,
-                monitor_type="下单监控",
+                monitor_type=order_row_monitor_type(r),
+                key_signal_type=order_row_key_signal_type(r),
                 direction=direction,
                 trigger_price=trigger_price,
                 stop_loss=stop_loss,
@@ -4437,7 +4510,8 @@ def force_close_before_reset():
         insert_trade_record(
             conn,
             symbol=r["symbol"],
-            monitor_type="下单监控",
+            monitor_type=order_row_monitor_type(r),
+            key_signal_type=order_row_key_signal_type(r),
             direction=direction,
             trigger_price=trigger_price,
             stop_loss=r["stop_loss"],
@@ -4567,6 +4641,216 @@ def api_sync_positions():
     return jsonify({"ok": True, "days": days, "synced": int(synced)})
 
 
+def _coerce_ts_ms(val):
+    if val is None or val == "":
+        return None
+    try:
+        v = float(val)
+    except (TypeError, ValueError):
+        return None
+    if v > 1e12:
+        return int(v)
+    if v > 1e9:
+        return int(v)
+    return int(v * 1000.0)
+
+
+def _unified_symbol_for_match(symbol_str):
+    if not symbol_str:
+        return ""
+    s = str(symbol_str).strip()
+    if not s:
+        return ""
+    try:
+        return normalize_exchange_symbol(s).split(":")[0]
+    except Exception:
+        x = s.upper().replace(" ", "")
+        if "/" in x:
+            return x.split(":")[0]
+        if x.endswith("USDT") and len(x) > 4:
+            return f"{x[:-4]}/USDT"
+        return x
+
+
+def exchange_position_sync_since_ms():
+    s = EXCHANGE_POSITION_SYNC_FROM_BJ
+    if s:
+        for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d", 10)):
+            try:
+                chunk = s[:ln] if len(s) >= ln else s[:10]
+                dt = datetime.strptime(chunk, fmt)
+                aware = dt.replace(tzinfo=APP_TZ)
+                return int(aware.timestamp() * 1000)
+            except Exception:
+                continue
+    dt0 = app_now() - timedelta(days=90)
+    try:
+        aware0 = datetime(dt0.year, dt0.month, dt0.day, 0, 0, 0, tzinfo=APP_TZ)
+    except Exception:
+        aware0 = datetime.now(APP_TZ)
+    return int(aware0.timestamp() * 1000)
+
+
+def _normalize_gate_position_history_entry(p):
+    if not p or not isinstance(p, dict):
+        return None
+    info = p.get("info") or {}
+    sym = p.get("symbol") or ""
+    side = (p.get("side") or "").strip().lower()
+    if side not in ("long", "short"):
+        sz = info.get("accum_size") if info.get("accum_size") is not None else info.get("size")
+        try:
+            szf = float(sz)
+            if szf > 0:
+                side = "long"
+            elif szf < 0:
+                side = "short"
+        except (TypeError, ValueError):
+            side = ""
+    rp = p.get("realizedPnl")
+    if rp is None:
+        rp = info.get("pnl")
+    try:
+        rp_f = float(rp) if rp is not None and str(rp).strip() != "" else None
+    except (TypeError, ValueError):
+        rp_f = None
+    close_ms = _coerce_ts_ms(p.get("lastUpdateTimestamp"))
+    if close_ms is None:
+        close_ms = _coerce_ts_ms(info.get("time"))
+    open_ms = _coerce_ts_ms(p.get("timestamp"))
+    if open_ms is None:
+        open_ms = _coerce_ts_ms(info.get("first_open_time"))
+    c_raw = str(info.get("contract") or "").strip()
+    t_raw = info.get("time")
+    sync_key = f"{c_raw}|{t_raw}|{side}"
+    return {
+        "symbol_u": _unified_symbol_for_match(sym),
+        "side": side,
+        "close_ms": close_ms,
+        "open_ms": open_ms,
+        "pnl": rp_f,
+        "sync_key": sync_key,
+    }
+
+
+def fetch_gate_positions_close_history():
+    if not exchange_private_api_configured():
+        return []
+    ensure_markets_loaded()
+    since_ms = exchange_position_sync_since_ms()
+    try:
+        rows = exchange.fetch_positions_history(
+            None,
+            since=int(since_ms),
+            limit=int(EXCHANGE_POSITION_HISTORY_LIMIT),
+            params={"settle": "usdt"},
+        )
+    except Exception:
+        try:
+            rows = exchange.fetch_positions_history(
+                None,
+                since=int(since_ms),
+                limit=int(EXCHANGE_POSITION_HISTORY_LIMIT),
+                params={},
+            )
+        except Exception:
+            return []
+    out = []
+    for p in rows or []:
+        h = _normalize_gate_position_history_entry(p)
+        if h and h["close_ms"] and h["side"] in ("long", "short") and h["symbol_u"]:
+            out.append(h)
+    return out
+
+
+def sync_trade_records_from_exchange(conn):
+    """为未同步的 trade_records 回填 Gate 平仓历史中的已实现盈亏。"""
+    global _LAST_EXCHANGE_PNL_SYNC_AT
+    if not exchange_private_api_configured():
+        return
+    now = time.time()
+    if now - _LAST_EXCHANGE_PNL_SYNC_AT < 25.0:
+        return
+    try:
+        hist = fetch_gate_positions_close_history()
+    except Exception:
+        return
+    if not hist:
+        _LAST_EXCHANGE_PNL_SYNC_AT = now
+        return
+    candidates = conn.execute(
+        """
+        SELECT id, symbol, direction, closed_at, opened_at, opened_at_ms
+        FROM trade_records
+        WHERE (exchange_sync_key IS NULL OR TRIM(exchange_sync_key) = '')
+        ORDER BY id DESC
+        LIMIT 120
+        """
+    ).fetchall()
+    if not candidates:
+        _LAST_EXCHANGE_PNL_SYNC_AT = now
+        return
+    used = set()
+    for tr in candidates:
+        close_ms_trade = _to_ms_with_fallback(
+            tr["closed_at_ms"] if "closed_at_ms" in tr.keys() else None, tr["closed_at"]
+        ) or opened_at_str_to_ms(tr["closed_at"])
+        open_ms_trade = _to_ms_with_fallback(
+            tr["opened_at_ms"] if "opened_at_ms" in tr.keys() else None, tr["opened_at"]
+        ) or opened_at_str_to_ms(tr["opened_at"])
+        if close_ms_trade is None:
+            continue
+        best = None
+        best_d = None
+        for h in hist:
+            sk = h["sync_key"]
+            if not sk or sk in used:
+                continue
+            if h["symbol_u"] != _unified_symbol_for_match(tr["symbol"]):
+                continue
+            if h["side"] != (tr["direction"] or "long").strip().lower():
+                continue
+            cm = h["close_ms"]
+            if cm is None:
+                continue
+            if open_ms_trade is not None:
+                if cm < open_ms_trade - 15 * 60 * 1000:
+                    continue
+                if cm > open_ms_trade + 15 * 86400 * 1000:
+                    continue
+            else:
+                if abs(cm - close_ms_trade) > 3 * 86400 * 1000:
+                    continue
+            d = abs(cm - close_ms_trade)
+            if best_d is None or d < best_d:
+                best_d = d
+                best = h
+        if best is None or best_d is None or best_d > 25 * 60 * 1000:
+            continue
+        sk = best["sync_key"]
+        if sk in used:
+            continue
+        eo = ms_to_app_local_str(best["open_ms"]) if best.get("open_ms") else None
+        ec = ms_to_app_local_str(best["close_ms"]) if best.get("close_ms") else None
+        pnl_val = best.get("pnl")
+        if pnl_val is None:
+            pnl_val = 0.0
+        conn.execute(
+            """
+            UPDATE trade_records
+            SET exchange_realized_pnl = ?, exchange_opened_at = ?, exchange_closed_at = ?, exchange_sync_key = ?
+            WHERE id = ?
+            """,
+            (float(pnl_val), eo, ec, sk, int(tr["id"])),
+        )
+        used.add(sk)
+    _LAST_EXCHANGE_PNL_SYNC_AT = now
+    try:
+        conn.commit()
+    except Exception:
+        pass
+
+
 # ====================== 主页面 ======================
 def render_main_page(page="trade"):
     now = app_now()
@@ -4586,6 +4870,11 @@ def render_main_page(page="trade"):
     order_list = []
     for o in raw_order_list:
         order_list.append(enrich_order_item(row_to_dict(o), current_capital))
+    if exchange_private_api_configured():
+        try:
+            sync_trade_records_from_exchange(conn)
+        except Exception:
+            pass
     raw_records = conn.execute("SELECT * FROM trade_records ORDER BY id DESC").fetchall()
     records = [to_effective_trade_dict(r) for r in raw_records]
     total = len(records)
@@ -5885,7 +6174,8 @@ def del_order(id):
             insert_trade_record(
                 conn,
                 symbol=row["symbol"],
-                monitor_type="下单监控",
+                monitor_type=order_row_monitor_type(row),
+                key_signal_type=order_row_key_signal_type(row),
                 direction=row["direction"],
                 trigger_price=row["trigger_price"],
                 stop_loss=row["stop_loss"],