修复币安交易记录

crypto_monitor_binance/app.py

@@ -5480,32 +5480,25 @@ def api_sync_exchange_pnl():
         fetch_binance_closed_positions_history(symbols=symbols, force_refresh=True)
     except Exception:
         pass
-    sync_trade_records_from_exchange(conn, force=True)
-    matched = 0
-    for r in rows:
-        tr = conn.execute(
-            "SELECT exchange_realized_pnl, exchange_sync_key FROM trade_records WHERE id=?",
-            (int(r["id"]),),
-        ).fetchone()
-        if tr and tr["exchange_realized_pnl"] is not None:
-            matched += 1
+    updated = sync_trade_records_from_exchange(conn, force=True)
     conn.commit()
     conn.close()
     tc = _BINANCE_CLOSED_POS_CACHE.get("trade_counts") or {}
     trade_n = sum(int(v) for v in tc.values())
     pos_n = len(_BINANCE_CLOSED_POS_CACHE.get("hist") or [])
     err = _BINANCE_CLOSED_POS_CACHE.get("last_err") or {}
-    msg = f"成交 {trade_n} 笔，重建仓位 {pos_n} 条，已写入 {matched}/{len(rows)} 条记录"
     if trade_n == 0:
-        msg += "（未拉到成交，请检查 API 权限/代理/EXCHANGE_POSITION_SYNC_FROM_BJ）"
-    if pos_n == 0 and trade_n > 0:
-        msg += "（有成交但未识别出完整平仓，请核对持仓模式 BINANCE_POSITION_MODE）"
+        msg = f"本次未拉到成交（0 笔），未更新记录。请检查 API 权限/代理；同步起点 EXCHANGE_POSITION_SYNC_FROM_BJ 会按 7 天分段拉取。"
+    else:
+        msg = f"成交 {trade_n} 笔，重建仓位 {pos_n} 条，本次更新 {updated}/{len(rows)} 条"
+        if pos_n == 0:
+            msg += "（有成交但未识别完整平仓，请核对 BINANCE_POSITION_MODE=hedge）"
     if err:
-        msg += f" 错误: {err}"
+        msg += f" 接口错误: {err}"
     return jsonify(
         {
-            "ok": True,
-            "synced": matched,
+            "ok": trade_n > 0 and (updated > 0 or pos_n > 0),
+            "synced": updated,
             "candidates": len(rows),
             "positions": pos_n,
             "trade_counts": tc,
@@ -5541,45 +5534,62 @@ def _unified_symbol_for_match(symbol_str):
     return s
 
 
-def _fetch_my_trades_paginated(exchange_symbol, since_ms, until_ms=None, max_pages=40):
-    """分页拉取成交（Binance userTrades）。"""
+def _fetch_my_trades_paginated(exchange_symbol, since_ms, until_ms=None, max_pages=120):
+    """
+    分页拉取 U 本位成交。Binance 限制：startTime~endTime 窗口最长 7 天，需分段请求。
+    """
     global _BINANCE_CLOSED_POS_CACHE
     if not (BINANCE_API_KEY and BINANCE_API_SECRET):
         _BINANCE_CLOSED_POS_CACHE["last_err"][exchange_symbol] = "未配置 API Key"
         return []
     ensure_markets_loaded()
+    until_ms = int(until_ms) if until_ms else int(time.time() * 1000)
+    start_ms = int(since_ms) if since_ms else until_ms - 7 * 24 * 60 * 60 * 1000
+    if start_ms >= until_ms:
+        start_ms = until_ms - 7 * 24 * 60 * 60 * 1000
+    week_ms = 7 * 24 * 60 * 60 * 1000 - 5000
     out = []
-    since = int(since_ms) if since_ms else None
-    until_ms = int(until_ms) if until_ms else None
+    seen_ids = set()
     last_err = None
-    for page in range(max_pages):
+    window_start = start_ms
+    pages = 0
+    while window_start < until_ms and pages < max_pages:
+        window_end = min(window_start + week_ms, until_ms)
         try:
-            batch = exchange.fetch_my_trades(exchange_symbol, since=since, limit=1000)
+            batch = exchange.fetch_my_trades(
+                exchange_symbol,
+                since=window_start,
+                limit=1000,
+                params={"endTime": window_end},
+            )
         except Exception as e:
             last_err = str(e)
-            if page == 0 and since:
+            if pages == 0:
                 try:
                     batch = exchange.fetch_my_trades(exchange_symbol, limit=1000)
+                    window_start = until_ms
                 except Exception as e2:
                     last_err = str(e2)
                     break
             else:
                 break
-        if not batch:
-            break
-        last_ts = None
+        pages += 1
+        if batch:
             for t in batch:
-            ts = _coerce_ts_ms(t.get("timestamp"))
-            if until_ms and ts and ts > until_ms:
+                tid = t.get("id")
+                if tid is not None and tid in seen_ids:
                     continue
+                ts = _coerce_ts_ms(t.get("timestamp"))
+                if ts and ts < start_ms:
+                    continue
+                if ts and ts > until_ms:
+                    continue
+                if tid is not None:
+                    seen_ids.add(tid)
                 out.append(t)
-            if ts:
-                last_ts = ts
-        if len(batch) < 1000:
-            break
-        if last_ts is None:
-            break
-        since = last_ts + 1
+        window_start = window_end + 1
+        if not batch and window_start < until_ms:
+            continue
     if last_err:
         _BINANCE_CLOSED_POS_CACHE["last_err"][exchange_symbol] = last_err
     return out
@@ -5856,13 +5866,13 @@ def sync_trade_record_exchange_pnl(conn, record_id, commit=True, force=False):
 
 
 def sync_trade_records_from_exchange(conn, force=False):
-    """为 trade_records 回填盈亏：成交重建已平仓位 + 时间匹配（对齐 App 仓位历史）。"""
+    """为 trade_records 回填盈亏：成交重建已平仓位 + 时间匹配（对齐 App 仓位历史）。返回本次更新条数。"""
     global _LAST_EXCHANGE_PNL_SYNC_AT, _BINANCE_CLOSED_POS_CACHE
     if not exchange_private_api_configured():
-        return
+        return 0
     now = time.time()
     if not force and now - _LAST_EXCHANGE_PNL_SYNC_AT < 25.0:
-        return
+        return 0
     if force:
         _BINANCE_CLOSED_POS_CACHE["at"] = 0.0
     if force:
@@ -5881,7 +5891,7 @@ def sync_trade_records_from_exchange(conn, force=False):
         ).fetchall()
     if not candidates:
         _LAST_EXCHANGE_PNL_SYNC_AT = now
-        return
+        return 0
     symbols = list({tr["symbol"] for tr in candidates if tr["symbol"]})
     try:
         hist = fetch_binance_closed_positions_history(symbols=symbols, force_refresh=force)
@@ -5893,8 +5903,13 @@ def sync_trade_records_from_exchange(conn, force=False):
             sk0 = (tr["exchange_sync_key"] if "exchange_sync_key" in tr.keys() else None) or ""
             if sk0 and str(sk0).startswith("pos|"):
                 used.add(str(sk0).strip())
-    matched = 0
+    updated = 0
     for tr in candidates:
+        rid = int(tr["id"])
+        before = conn.execute(
+            "SELECT exchange_sync_key, exchange_realized_pnl FROM trade_records WHERE id=?",
+            (rid,),
+        ).fetchone()
         pos, _ = match_trade_record_to_position(
             tr,
             hist,
@@ -5902,24 +5917,29 @@ def sync_trade_records_from_exchange(conn, force=False):
             unified_symbol_fn=_unified_symbol_for_match,
             to_ms_fn=_to_ms_with_fallback,
         )
-        if not pos:
-            continue
+        if pos:
             sk = pos.get("sync_key")
-        if not sk or sk in used:
-            continue
-        if _apply_closed_position_to_trade_record(conn, int(tr["id"]), pos):
+            if sk and sk not in used:
+                if _apply_closed_position_to_trade_record(conn, rid, pos):
                     used.add(sk)
-            matched += 1
-    for tr in candidates:
-        sk0 = (tr["exchange_sync_key"] if "exchange_sync_key" in tr.keys() else None) or ""
+                    updated += 1
+                    continue
+        sk0 = (before["exchange_sync_key"] if before else None) or ""
         if sk0 and str(sk0).startswith("pos|"):
             continue
-        sync_trade_record_exchange_pnl(conn, int(tr["id"]), commit=False, force=force)
+        if sync_trade_record_exchange_pnl(conn, rid, commit=False, force=force):
+            after = conn.execute(
+                "SELECT exchange_sync_key FROM trade_records WHERE id=?", (rid,)
+            ).fetchone()
+            sk1 = (after["exchange_sync_key"] if after else None) or ""
+            if sk1 != sk0 or (before and before["exchange_realized_pnl"] is None):
+                updated += 1
     _LAST_EXCHANGE_PNL_SYNC_AT = now
     try:
         conn.commit()
     except Exception:
         pass
+    return updated
 
 
 # ====================== 主页面 ======================