dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

删除同步仓位

Browse Source
This commit is contained in:
dekun
2026-05-19 18:49:55 +08:00
parent fe9cff4870
commit 90cb9abf31
5 changed files with 4 additions and 775 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crypto_monitor_binance/app.py
+1 -459
View File
@@ -44,10 +44,6 @@ from fib_key_monitor_lib import (
key_signal_type_for_trade_record,
stored_key_signal_type,
)
from binance_closed_positions_lib import (
match_trade_record_to_position,
rebuild_closed_positions_from_trades,
)
from history_window_lib import (
PRESET_CUSTOM,
PRESET_UTC_LAST24H,
@@ -164,8 +160,6 @@ KEY_SIZING_USE_ZERO_POSITION_SNAPSHOT = os.getenv("KEY_SIZING_USE_ZERO_POSITION_
ORDER_MONITOR_TYPE_MANUAL = "下单监控"
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「仓位历史-实现盈亏」对齐：默认仅 REALIZED_PNL（手续费另计；避免与 COMMISSION 重复扣）
BINANCE_APP_PNL_INCOME_TYPES = frozenset({"REALIZED_PNL"})
BINANCE_APP_PNL_INCOME_WITH_FEE = frozenset({"REALIZED_PNL", "COMMISSION"})
@@ -177,10 +171,6 @@ BINANCE_PNL_INCLUDE_FUNDING = os.getenv("BINANCE_PNL_INCLUDE_FUNDING", "false").
"true",
"yes",
)
# 与币安 App 仓位历史对齐目标误差（USDT）
BINANCE_PNL_MATCH_TOLERANCE = max(0.01, float(os.getenv("BINANCE_PNL_MATCH_TOLERANCE", "0.05")))
_LAST_EXCHANGE_PNL_SYNC_AT = 0.0
_BINANCE_CLOSED_POS_CACHE = {"at": 0.0, "hist": [], "trade_counts": {}, "last_err": {}}
KEY_MONITOR_ALERT_ONLY_TYPES = frozenset({"关键阻力位", "关键支撑位"})
AUTO_TRANSFER_ENABLED = os.getenv("AUTO_TRANSFER_ENABLED", "false").lower() == "true"
AUTO_TRANSFER_AMOUNT = float(os.getenv("AUTO_TRANSFER_AMOUNT", "30"))
@@ -2268,39 +2258,6 @@ def resolve_trade_pnl_amount(
if last_ts and not closed_at_str:
closed_at_str = ms_to_app_local_str(int(last_ts))
close_ms = int(last_ts)
if close_ms and open_ms:
try:
hist = fetch_binance_closed_positions_history(symbols=[sym])
fake = {
"symbol": sym,
"direction": direction,
"opened_at": opened_at_str or (row["opened_at"] if hasattr(row, "keys") else ""),
"closed_at": closed_at_str,
"opened_at_ms": open_ms,
"closed_at_ms": close_ms,
}
pos, _ = match_trade_record_to_position(
fake,
hist,
set(),
unified_symbol_fn=_unified_symbol_for_match,
to_ms_fn=_to_ms_with_fallback,
)
if pos and pos.get("pnl") is not None:
eo = ms_to_app_local_str(pos["open_ms"]) if pos.get("open_ms") else None
ec = ms_to_app_local_str(pos["close_ms"]) if pos.get("close_ms") else None
ep = pos.get("exit_price")
if ep and (exit_price is None or float(exit_price or 0) <= 0):
exit_price = float(ep)
return (
float(pos["pnl"]),
exit_price,
eo,
ec,
pos.get("sync_key"),
)
except Exception:
pass
net, sync_key, eo, ec = fetch_binance_net_pnl_for_trade(
ex_sym, direction, open_ms, close_ms, closing_trades=closing_trades
)
@@ -2468,13 +2425,7 @@ def insert_trade_record(
open_ts, open_ts_ms, close_ts, close_ts_ms, result, miss_reason, exchange_trade_id, er or None
)
)
record_id = int(cur.lastrowid or 0)
if record_id and close_ts and result != "错过" and exchange_private_api_configured():
try:
sync_trade_record_exchange_pnl(conn, record_id, commit=False)
except Exception:
pass
return record_id
return int(cur.lastrowid or 0)
def calc_duration_text(open_str, close_str):
@@ -5435,78 +5386,6 @@ def api_sync_positions():
return jsonify({"ok": True, "days": days, "synced": int(synced)})
@app.route("/api/sync_exchange_pnl", methods=["POST"])
@login_required
def api_sync_exchange_pnl():
"""立即为近期未同步记录回填 Binance 净盈亏（含手续费）。"""
if not exchange_private_api_configured():
return jsonify({"ok": False, "msg": "未配置 Binance API，无法同步"}), 400
payload = request.get_json(silent=True) or {}
limit = 120
force_all = str(payload.get("force", "")).lower() in ("1", "true", "yes")
try:
if payload.get("limit") is not None:
limit = max(1, min(500, int(payload.get("limit"))))
except (TypeError, ValueError):
pass
global _LAST_EXCHANGE_PNL_SYNC_AT, _BINANCE_CLOSED_POS_CACHE
_LAST_EXCHANGE_PNL_SYNC_AT = 0.0
_BINANCE_CLOSED_POS_CACHE["at"] = 0.0
conn = get_db()
if force_all:
rows = conn.execute(
"SELECT id FROM trade_records ORDER BY id DESC LIMIT ?",
(limit,),
).fetchall()
else:
rows = conn.execute(
"""
SELECT id FROM trade_records
WHERE (exchange_sync_key IS NULL OR TRIM(exchange_sync_key) = '')
ORDER BY id DESC
LIMIT ?
""",
(limit,),
).fetchall()
sym_rows = conn.execute(
f"""
SELECT DISTINCT symbol FROM trade_records
WHERE id IN ({",".join("?" * len(rows)) if rows else "NULL"})
""",
tuple(int(r["id"]) for r in rows),
).fetchall() if rows else []
symbols = [sr["symbol"] for sr in sym_rows if sr["symbol"]]
try:
fetch_binance_closed_positions_history(symbols=symbols, force_refresh=True)
except Exception:
pass
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 {}
if trade_n == 0:
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}"
return jsonify(
{
"ok": trade_n > 0 and (updated > 0 or pos_n > 0),
"synced": updated,
"candidates": len(rows),
"positions": pos_n,
"trade_counts": tc,
"msg": msg,
}
)
def _coerce_ts_ms(val):
if val is None or val == "":
return None
@@ -5521,179 +5400,6 @@ def _coerce_ts_ms(val):
return int(v * 1000.0)
def _unified_symbol_for_match(symbol_str):
s = (symbol_str or "").strip().upper()
if not s:
return ""
if ":" in s:
s = s.split(":")[0]
if "_" in s and "/" not in s:
s = s.replace("_", "/")
if s.endswith("USDT") and "/" not in s and len(s) > 4:
s = f"{s[:-4]}/USDT"
return s
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 = []
seen_ids = set()
last_err = None
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=window_start,
limit=1000,
params={"endTime": window_end},
)
except Exception as e:
last_err = str(e)
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
pages += 1
if batch:
for t in batch:
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)
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
def fetch_binance_closed_positions_history(symbols=None, force_refresh=False):
"""
从成交重建已平仓位（对齐 App 仓位历史实现盈亏，不含资金费）。
symbols: 可选 symbol 列表（如 NEAR/USDT），为空则仅返回缓存。
"""
global _BINANCE_CLOSED_POS_CACHE
now = time.time()
if (
not force_refresh
and _BINANCE_CLOSED_POS_CACHE["hist"]
and now - float(_BINANCE_CLOSED_POS_CACHE["at"] or 0) < 25.0
and not symbols
):
return list(_BINANCE_CLOSED_POS_CACHE["hist"])
if not exchange_private_api_configured():
return []
sym_list = []
for s in symbols or []:
try:
sym_list.append(normalize_exchange_symbol(s))
except Exception:
continue
if not sym_list:
return list(_BINANCE_CLOSED_POS_CACHE["hist"] or [])
since_ms = exchange_position_sync_since_ms()
until_ms = int(time.time() * 1000) + 120_000
trades_by_symbol = {}
trade_counts = {}
for ex_sym in sym_list:
trades = _fetch_my_trades_paginated(ex_sym, since_ms, until_ms)
trades_by_symbol[ex_sym] = trades
trade_counts[ex_sym] = len(trades)
def _contract_size(ex_sym):
try:
ensure_markets_loaded()
return float(exchange.market(ex_sym).get("contractSize") or 1)
except Exception:
return 1.0
hist = rebuild_closed_positions_from_trades(
trades_by_symbol,
unified_symbol_fn=_unified_symbol_for_match,
position_mode=BINANCE_POSITION_MODE,
contract_size_fn=_contract_size,
since_ms=since_ms,
)
prev = list(_BINANCE_CLOSED_POS_CACHE["hist"] or [])
if prev and not force_refresh:
keys = {h.get("sync_key") for h in prev if h.get("sync_key")}
for h in hist:
sk = h.get("sync_key")
if sk and sk not in keys:
prev.append(h)
hist = sorted(prev, key=lambda x: int(x.get("close_ms") or 0), reverse=True)
_BINANCE_CLOSED_POS_CACHE["at"] = now
_BINANCE_CLOSED_POS_CACHE["hist"] = hist
_BINANCE_CLOSED_POS_CACHE["trade_counts"] = trade_counts
return hist
def _apply_closed_position_to_trade_record(conn, trade_id, pos):
pnl_val = pos.get("pnl")
if pnl_val is None:
return False
sk = pos.get("sync_key") or ""
eo = ms_to_app_local_str(pos["open_ms"]) if pos.get("open_ms") else None
ec = ms_to_app_local_str(pos["close_ms"]) if pos.get("close_ms") else None
conn.execute(
"""
UPDATE trade_records
SET exchange_realized_pnl = ?, exchange_opened_at = ?, exchange_closed_at = ?,
exchange_sync_key = ?, pnl_amount = ?
WHERE id = ?
""",
(float(pnl_val), eo, ec, sk, float(pnl_val), int(trade_id)),
)
return True
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 _fetch_binance_income_entries(exchange_symbol, start_ms, end_ms):
if not hasattr(exchange, "fapiPrivateGetIncome"):
return []
@@ -5783,165 +5489,6 @@ def fetch_binance_net_pnl_for_trade(
return None, None, None, None
def sync_trade_record_exchange_pnl(conn, record_id, commit=True, force=False):
"""单条 trade_records：优先按成交重建的已平仓位匹配（对齐 App 仓位历史）。"""
if not exchange_private_api_configured():
return False
tr = conn.execute("SELECT * FROM trade_records WHERE id=?", (int(record_id),)).fetchone()
if not tr:
return False
sk_existing = (tr["exchange_sync_key"] if "exchange_sync_key" in tr.keys() else None) or ""
if not force and str(sk_existing).strip().startswith("pos|"):
return tr["exchange_realized_pnl"] is not None
if not force and str(sk_existing).strip() and not str(sk_existing).startswith("pos|"):
pass
open_ms = _to_ms_with_fallback(
tr["opened_at_ms"] if "opened_at_ms" in tr.keys() else None, tr["opened_at"]
)
close_ms = _to_ms_with_fallback(
tr["closed_at_ms"] if "closed_at_ms" in tr.keys() else None, tr["closed_at"]
)
if open_ms is None or close_ms is None:
return False
try:
hist = fetch_binance_closed_positions_history(
symbols=[tr["symbol"]], force_refresh=force
)
except Exception:
hist = []
used = set()
if force:
used = set()
else:
rows = conn.execute(
"SELECT exchange_sync_key FROM trade_records WHERE exchange_sync_key LIKE 'pos|%'"
).fetchall()
for r in rows:
sk = (r["exchange_sync_key"] if "exchange_sync_key" in r.keys() else None) or ""
if sk:
used.add(str(sk).strip())
pos, _ = match_trade_record_to_position(
tr,
hist,
used,
unified_symbol_fn=_unified_symbol_for_match,
to_ms_fn=_to_ms_with_fallback,
)
if pos:
if _apply_closed_position_to_trade_record(conn, int(record_id), pos):
if commit:
try:
conn.commit()
except Exception:
pass
return True
direction = (tr["direction"] or "long").strip().lower()
try:
ex_sym = normalize_exchange_symbol(tr["symbol"])
except Exception:
return False
closing_trades = fetch_closing_fills_for_record(
ex_sym, direction, tr["opened_at"], tr["closed_at"], opened_at_ms=open_ms, closed_at_ms=close_ms
)
net, sync_key, eo, ec = fetch_binance_net_pnl_for_trade(
ex_sym, direction, open_ms, close_ms, closing_trades=closing_trades
)
if net is None or not sync_key:
return False
conn.execute(
"""
UPDATE trade_records
SET exchange_realized_pnl = ?, exchange_opened_at = ?, exchange_closed_at = ?,
exchange_sync_key = ?, pnl_amount = ?
WHERE id = ?
""",
(float(net), eo, ec, sync_key, float(net), int(record_id)),
)
if commit:
try:
conn.commit()
except Exception:
pass
return True
def sync_trade_records_from_exchange(conn, force=False):
"""为 trade_records 回填盈亏：成交重建已平仓位 + 时间匹配（对齐 App 仓位历史）。返回本次更新条数。"""
global _LAST_EXCHANGE_PNL_SYNC_AT, _BINANCE_CLOSED_POS_CACHE
if not exchange_private_api_configured():
return 0
now = time.time()
if not force and now - _LAST_EXCHANGE_PNL_SYNC_AT < 25.0:
return 0
if force:
_BINANCE_CLOSED_POS_CACHE["at"] = 0.0
if force:
candidates = conn.execute(
"SELECT * FROM trade_records ORDER BY id DESC LIMIT 200"
).fetchall()
else:
candidates = conn.execute(
"""
SELECT * FROM trade_records
WHERE (exchange_sync_key IS NULL OR TRIM(exchange_sync_key) = ''
OR exchange_sync_key NOT LIKE 'pos|%')
ORDER BY id DESC
LIMIT 200
"""
).fetchall()
if not candidates:
_LAST_EXCHANGE_PNL_SYNC_AT = now
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)
except Exception:
hist = []
used = set()
if not force:
for tr in candidates:
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())
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,
used,
unified_symbol_fn=_unified_symbol_for_match,
to_ms_fn=_to_ms_with_fallback,
)
if pos:
sk = pos.get("sync_key")
if sk and sk not in used:
if _apply_closed_position_to_trade_record(conn, rid, pos):
used.add(sk)
updated += 1
continue
sk0 = (before["exchange_sync_key"] if before else None) or ""
if sk0 and str(sk0).startswith("pos|"):
continue
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
# ====================== 主页面 ======================
def render_main_page(page="trade"):
now = app_now()
@@ -5966,11 +5513,6 @@ 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 WHERE COALESCE(closed_at, created_at, opened_at) >= ? "
"AND COALESCE(closed_at, created_at, opened_at) <= ? ORDER BY id DESC LIMIT 1000",