修复币安交易记录

crypto_monitor_binance/.env.example

@@ -75,7 +75,7 @@ BINANCE_TRIGGER_WORKING_TYPE=CONTRACT_PRICE
 # EXCHANGE_DISPLAY_NAME=Binance
 # 企业微信推送里展示的账户备注
 # BINANCE_ACCOUNT_LABEL=binance实盘账户
-# 盈亏同步口径：false=与 App「仓位历史-实现盈亏」一致（不含资金费）；true=含资金费等完整流水
+# 盈亏同步：false=按仓位历史口径（已实现盈亏+手续费，不含资金费）；true=含资金费
 # BINANCE_PNL_INCLUDE_FUNDING=false
 
 # =============================================================================

crypto_monitor_binance/app.py

@@ -162,8 +162,9 @@ ORDER_MONITOR_TYPE_KEY_AUTO = "关键位监控"
 KEY_MONITOR_AUTO_TYPES = frozenset({"箱体突破", "收敛突破"})
 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"))))
-# 与币安 App「仓位历史-实现盈亏」对齐：仅已实现盈亏 + 手续费（不含资金费）
-BINANCE_APP_PNL_INCOME_TYPES = frozenset({"REALIZED_PNL", "COMMISSION"})
+# 与币安 App「仓位历史-实现盈亏」对齐：默认仅 REALIZED_PNL（手续费另计；避免与 COMMISSION 重复扣）
+BINANCE_APP_PNL_INCOME_TYPES = frozenset({"REALIZED_PNL"})
+BINANCE_APP_PNL_INCOME_WITH_FEE = frozenset({"REALIZED_PNL", "COMMISSION"})
 BINANCE_NET_INCOME_TYPES = frozenset(
     {"REALIZED_PNL", "COMMISSION", "FUNDING_FEE", "INSURANCE_CLEAR", "INTERNAL_AUTO_CLOSE"}
 )
@@ -2068,19 +2069,47 @@ def get_plan_notional_usdt(row_or_dict):
 
 
 def _trade_ids_from_fills(trades):
+    """仅使用 Binance 原始 tradeId（与 income 流水一致），不用 ccxt 的 id。"""
     ids = set()
     for t in trades or []:
         info = t.get("info") if isinstance(t.get("info"), dict) else {}
-        for src in (t, info):
-            if not isinstance(src, dict):
-                continue
-            for k in ("tradeId", "trade_id", "id"):
-                v = src.get(k)
-                if v is not None and str(v).strip() != "":
-                    ids.add(str(v).strip())
+        for k in ("tradeId", "trade_id"):
+            v = info.get(k)
+            if v is not None and str(v).strip() != "":
+                ids.add(str(v).strip())
     return ids
 
 
+def _cluster_closing_trades_near_close(trades, closed_ms, spread_ms=8 * 60 * 1000):
+    """只保留平仓时刻附近的一簇减仓成交，避免把相邻其它仓位算进来。"""
+    if not trades:
+        return []
+    if closed_ms is None:
+        return list(trades)
+    try:
+        closed_ms = int(closed_ms)
+    except (TypeError, ValueError):
+        return list(trades)
+    scored = []
+    for t in trades:
+        ts = _coerce_ts_ms(t.get("timestamp"))
+        if ts is None:
+            continue
+        scored.append((abs(ts - closed_ms), t))
+    if not scored:
+        return list(trades)
+    scored.sort(key=lambda x: x[0])
+    anchor_ts = _coerce_ts_ms(scored[0][1].get("timestamp"))
+    if anchor_ts is None:
+        return [scored[0][1]]
+    return [
+        t
+        for t in trades
+        if _coerce_ts_ms(t.get("timestamp")) is not None
+        and abs(_coerce_ts_ms(t.get("timestamp")) - anchor_ts) <= spread_ms
+    ]
+
+
 def _income_entry_trade_id(entry):
     if not isinstance(entry, dict):
         return ""
@@ -2094,47 +2123,61 @@ def _income_entry_trade_id(entry):
 
 
 def calc_binance_realized_pnl_from_trades(trades):
-    """
-    按 Binance 成交回报汇总：sum(realizedPnl) + 手续费（与 App「仓位历史-实现盈亏」一致）。
-    """
+    """仅汇总成交回报中的 realizedPnl（勿再扣 commission，避免与 income 重复）。"""
     if not trades:
         return None
     total = 0.0
     has = False
     for t in trades:
         info = t.get("info") if isinstance(t.get("info"), dict) else {}
-        for src in (info, t):
-            if not isinstance(src, dict):
-                continue
-            for key in ("realizedPnl", "realized_pnl"):
-                v = src.get(key)
-                if v is None or str(v).strip() == "":
-                    continue
-                try:
-                    total += float(v)
-                    has = True
-                except (TypeError, ValueError):
-                    pass
-        fee = t.get("fee")
-        if isinstance(fee, dict) and fee.get("cost") is not None:
-            try:
-                total += float(fee["cost"])
-                has = True
-            except (TypeError, ValueError):
-                pass
-        comm = info.get("commission")
-        if comm is not None and str(comm).strip() != "":
-            try:
-                c = float(comm)
-                total -= c if c > 0 else -abs(c)
-                has = True
-            except (TypeError, ValueError):
-                pass
+        v = info.get("realizedPnl")
+        if v is None or str(v).strip() == "":
+            v = t.get("realizedPnl") or t.get("realized_pnl")
+        if v is None or str(v).strip() == "":
+            continue
+        try:
+            total += float(v)
+            has = True
+        except (TypeError, ValueError):
+            pass
     if not has:
         return None
     return round(total, FUNDS_DECIMALS)
 
 
+def _sum_binance_income(entries, income_types, trade_ids=None):
+    net = 0.0
+    first_t = None
+    last_t = None
+    strict = bool(trade_ids)
+    for e in entries:
+        it = (e.get("incomeType") or e.get("income_type") or "").strip()
+        if it not in income_types:
+            continue
+        if strict:
+            if it in ("REALIZED_PNL", "COMMISSION"):
+                tid = _income_entry_trade_id(e)
+                if not tid or tid not in trade_ids:
+                    continue
+            else:
+                continue
+        elif trade_ids and it in ("REALIZED_PNL", "COMMISSION"):
+            tid = _income_entry_trade_id(e)
+            if tid and tid not in trade_ids:
+                continue
+        try:
+            net += float(e.get("income") or 0)
+        except (TypeError, ValueError):
+            pass
+        t = _coerce_ts_ms(e.get("time"))
+        if t:
+            first_t = t if first_t is None else min(first_t, t)
+            last_t = t if last_t is None else max(last_t, t)
+    if first_t is None:
+        return None, None, None
+    return round(net, FUNDS_DECIMALS), first_t, last_t
+
+
 def calc_pnl_from_closing_trades(direction, entry_price, trades, exchange_symbol=None):
     """按减仓成交数量×价差汇总盈亏（不含资金费；比单点标记价更接近交易所）。"""
     try:
@@ -2208,6 +2251,8 @@ def resolve_trade_pnl_amount(
             opened_at_ms=open_ms,
             closed_at_ms=close_ms,
         )
+        if closing_trades and close_ms:
+            closing_trades = _cluster_closing_trades_near_close(closing_trades, int(close_ms))
     if closing_trades:
         wexit = calc_weighted_exit_price(closing_trades)
         if wexit and (exit_price is None or float(exit_price or 0) <= 0):
@@ -3613,9 +3658,7 @@ def fetch_closing_fills_for_record(exchange_symbol, direction, opened_at_str, cl
     since_ms = _to_ms_with_fallback(opened_at_ms, opened_at_str)
     close_side = "sell" if direction == "long" else "buy"
     closed_ms = _to_ms_with_fallback(closed_at_ms, closed_at_str) if (closed_at_str or closed_at_ms is not None) else None
-    # 历史记录回填给一点缓冲，兼容成交落在记录时间附近的情况
-    if closed_ms is not None:
-        closed_ms += 6 * 60 * 60 * 1000
+    close_upper_ms = (int(closed_ms) + 15 * 60 * 1000) if closed_ms is not None else None
     candidates = []
     all_side_candidates = []
     try:
@@ -3639,7 +3682,7 @@ def fetch_closing_fills_for_record(exchange_symbol, direction, opened_at_str, cl
             continue
         if since_ms and ts < since_ms:
             continue
-        if closed_ms and ts > closed_ms:
+        if close_upper_ms and ts > close_upper_ms:
             continue
         info = t.get("info") or {}
         if not isinstance(info, dict):
@@ -3649,10 +3692,6 @@ def fetch_closing_fills_for_record(exchange_symbol, direction, opened_at_str, cl
             if pos_side in ("long", "short") and pos_side != direction:
                 continue
         all_side_candidates.append(t)
-        if since_ms and ts < since_ms:
-            continue
-        if closed_ms and ts > closed_ms:
-            continue
         candidates.append(t)
     candidates.sort(key=lambda x: x.get("timestamp") or 0)
     if candidates:
@@ -3663,25 +3702,21 @@ def fetch_closing_fills_for_record(exchange_symbol, direction, opened_at_str, cl
     if not all_side_candidates:
         return []
     if not closed_ms:
-        return all_side_candidates[-20:]
+        return all_side_candidates[-5:]
     near = []
     for t in all_side_candidates:
-        ts = t.get("timestamp")
+        ts = _coerce_ts_ms(t.get("timestamp"))
         if ts is None:
             continue
-        try:
-            delta = abs(int(ts) - int(closed_ms))
-        except Exception:
-            continue
-        # 放宽到前后 7 天
-        if delta <= 7 * 24 * 60 * 60 * 1000:
+        delta = abs(ts - int(closed_ms))
+        if delta <= 45 * 60 * 1000:
             near.append((delta, t))
     if near:
         near.sort(key=lambda x: x[0])
-        picked = [x[1] for x in near[:20]]
+        picked = [x[1] for x in near[:12]]
         picked.sort(key=lambda x: x.get("timestamp") or 0)
-        return picked
-    return all_side_candidates[-20:]
+        return _cluster_closing_trades_near_close(picked, int(closed_ms))
+    return _cluster_closing_trades_near_close(all_side_candidates[-5:], int(closed_ms))
 
 
 def calc_weighted_exit_price(trades):
@@ -5482,66 +5517,58 @@ def fetch_binance_net_pnl_for_trade(
 ):
     if open_ms is None or close_ms is None or close_ms < open_ms:
         return None, None, None, None
-    trade_ids = _trade_ids_from_fills(closing_trades) if closing_trades else None
     if closing_trades:
-        trade_pnl = calc_binance_realized_pnl_from_trades(closing_trades)
-        if trade_pnl is not None:
-            first_t = None
-            last_t = None
-            for t in closing_trades:
-                ts = _coerce_ts_ms(t.get("timestamp"))
-                if ts:
-                    first_t = ts if first_t is None else min(first_t, ts)
-                    last_t = ts if last_t is None else max(last_t, ts)
-            ensure_markets_loaded()
-            market = exchange.market(exchange_symbol)
-            cid = market.get("id") or exchange_symbol
-            sync_key = f"trades|{cid}|{direction}|{open_ms}|{close_ms}|{trade_pnl}"
-            eo = ms_to_app_local_str(first_t) if first_t else None
-            ec = ms_to_app_local_str(last_t) if last_t else None
-            return trade_pnl, sync_key, eo, ec
-    income_types = (
-        BINANCE_NET_INCOME_TYPES
-        if BINANCE_PNL_INCLUDE_FUNDING
-        else BINANCE_APP_PNL_INCOME_TYPES
-    )
-    buffer_ms = 90 * 1000 if trade_ids else 5 * 60 * 1000
+        closing_trades = _cluster_closing_trades_near_close(closing_trades, int(close_ms))
+    trade_ids = _trade_ids_from_fills(closing_trades) if closing_trades else None
+    buffer_ms = 3 * 60 * 1000 if trade_ids else 5 * 60 * 1000
     entries = _fetch_binance_income_entries(
         exchange_symbol, max(0, int(open_ms) - buffer_ms), int(close_ms) + buffer_ms
     )
-    if not entries:
-        return None, None, None, None
-    net = 0.0
-    first_t = None
-    last_t = None
-    for e in entries:
-        it = (e.get("incomeType") or e.get("income_type") or "").strip()
-        if it not in income_types:
-            continue
-        if trade_ids and it in ("REALIZED_PNL", "COMMISSION"):
-            tid = _income_entry_trade_id(e)
-            if tid and tid not in trade_ids:
-                continue
-        if trade_ids and it == "FUNDING_FEE":
-            continue
-        try:
-            net += float(e.get("income") or 0)
-        except (TypeError, ValueError):
-            pass
-        t = _coerce_ts_ms(e.get("time"))
-        if t:
-            first_t = t if first_t is None else min(first_t, t)
-            last_t = t if last_t is None else max(last_t, t)
-    if first_t is None:
-        return None, None, None, None
-    net = round(net, FUNDS_DECIMALS)
     ensure_markets_loaded()
     market = exchange.market(exchange_symbol)
     cid = market.get("id") or exchange_symbol
-    sync_key = f"income|{cid}|{direction}|{open_ms}|{close_ms}|{net}"
-    eo = ms_to_app_local_str(first_t)
-    ec = ms_to_app_local_str(last_t)
-    return net, sync_key, eo, ec
+
+    def _pack(net, first_t, last_t, prefix):
+        if net is None:
+            return None
+        sk = f"{prefix}|{cid}|{direction}|{open_ms}|{close_ms}|{net}"
+        eo = ms_to_app_local_str(first_t) if first_t else None
+        ec = ms_to_app_local_str(last_t) if last_t else None
+        return net, sk, eo, ec
+
+    if entries and trade_ids:
+        net, ft, lt = _sum_binance_income(entries, BINANCE_APP_PNL_INCOME_WITH_FEE, trade_ids)
+        out = _pack(net, ft, lt, "income_net")
+        if out:
+            return out
+        net, ft, lt = _sum_binance_income(entries, BINANCE_APP_PNL_INCOME_TYPES, trade_ids)
+        out = _pack(net, ft, lt, "income_rp")
+        if out:
+            return out
+
+    if closing_trades:
+        trade_pnl = calc_binance_realized_pnl_from_trades(closing_trades)
+        if trade_pnl is not None:
+            fts = [_coerce_ts_ms(t.get("timestamp")) for t in closing_trades]
+            fts = [x for x in fts if x]
+            ft = min(fts) if fts else None
+            lt = max(fts) if fts else None
+            out = _pack(trade_pnl, ft, lt, "trades_rp")
+            if out:
+                return out
+
+    if entries:
+        loose_types = (
+            BINANCE_NET_INCOME_TYPES
+            if BINANCE_PNL_INCLUDE_FUNDING
+            else BINANCE_APP_PNL_INCOME_WITH_FEE
+        )
+        net, ft, lt = _sum_binance_income(entries, loose_types, trade_ids if trade_ids else None)
+        out = _pack(net, ft, lt, "income")
+        if out:
+            return out
+
+    return None, None, None, None
 
 
 def sync_trade_record_exchange_pnl(conn, record_id, commit=True, force=False):