diff --git a/crypto_monitor_binance/app.py b/crypto_monitor_binance/app.py index 79133a6..fb1da1f 100644 --- a/crypto_monitor_binance/app.py +++ b/crypto_monitor_binance/app.py @@ -9619,6 +9619,10 @@ register_instance_dashboard_routes( hedge_enabled=False, ) +from lib.account_ledger.account_ledger_register import install_account_ledger + +install_account_ledger(app, _REPO_ROOT, app_module=sys.modules[__name__], exchange_key="binance") + @app.route("/api/journals") @login_required diff --git a/crypto_monitor_gate/app.py b/crypto_monitor_gate/app.py index fba3dd0..6c611ed 100644 --- a/crypto_monitor_gate/app.py +++ b/crypto_monitor_gate/app.py @@ -9461,6 +9461,10 @@ register_instance_dashboard_routes( hedge_enabled=False, ) +from lib.account_ledger.account_ledger_register import install_account_ledger + +install_account_ledger(app, _REPO_ROOT, app_module=sys.modules[__name__], exchange_key="gate") + @app.route("/api/journals") @login_required diff --git a/crypto_monitor_okx/app.py b/crypto_monitor_okx/app.py index b1d8bca..7a609be 100644 --- a/crypto_monitor_okx/app.py +++ b/crypto_monitor_okx/app.py @@ -9200,6 +9200,10 @@ register_instance_dashboard_routes( hedge_enabled=hedge_module_enabled, ) +from lib.account_ledger.account_ledger_register import install_account_ledger + +install_account_ledger(app, _REPO_ROOT, app_module=sys.modules[__name__], exchange_key="okx") + @app.route("/api/journals") @login_required diff --git a/docs/审计修复报告-账户流水-2026-08-10.md b/docs/审计修复报告-账户流水-2026-08-10.md new file mode 100644 index 0000000..cbc86cf --- /dev/null +++ b/docs/审计修复报告-账户流水-2026-08-10.md @@ -0,0 +1,26 @@ +# 审计修复报告：账户流水（2026-08-10） + +## 范围 + +新增「账户流水」功能：`lib/account_ledger/*`、`lib/exchange/*_ledger_lib.py`、三所 `app.py` 安装、导航显示开关、前端 SSE 页。 + +## 结论 + +**可上线。** 认证、SQL、SSE 载荷范围、单实例隔离与现有实例模式一致。发现 1 项中危并已在同批修复。 + +## 发现与处理 + +| 级别 | 问题 | 处理 | +|------|------|------| +| 中 | `POST /api/account_ledger/refresh` 可把 `start_ms` 拉到极早，触发大量交易所分页请求；无冷却 | 同步窗口强制 `LOOKBACK_DAYS` 下限；手动同步默认 30s 冷却 | +| 低 | 导航关闭仍可直连 URL（与数据看板相同，仅 UI 隐藏） | 保持与现有 display pref 一致；embed `tab_allowed` 仍 403 | +| 信息 | 交易所异常文案写入 `last_error` 展示 | 可接受；未记录密钥 | + +## 验证 + +- `python -m unittest tests.test_account_ledger_normalize` 通过 +- 三所仅增加 `install_account_ledger`，不改动开仓/风控主路径 + +## 使用提醒 + +默认导航关闭；需在系统设置打开「账户流水」。数据来自交易所，首次打开可能需等待一轮后台同步或点「立即同步」。 diff --git a/docs/账户流水.md b/docs/账户流水.md new file mode 100644 index 0000000..b734fd1 --- /dev/null +++ b/docs/账户流水.md @@ -0,0 +1,71 @@ +# 账户流水（三所统一） + +从**交易所 API**拉取资金账户与交易账户账单，在实例内展示。 +不使用程序本地 `transfer_logs` 作为主数据源。 + +## 能力概览 + +| 项 | 说明 | +|----|------| +| 导航 | 「账户流水」Tab；默认关闭，在 **系统设置 → 导航显示** 打开 | +| Tab | **资金账户** / **交易账户** | +| 分页 | 每页 10 条，时间倒序 | +| 时间窗 | 跟随顶栏 UTC **预设**（与列表 `list_window` 一致） | +| 同步 | 后台约 **120s** 拉一次交易所；完成后 **SSE** 推版本，前端自动刷新 | +| 币种 | USDT；OKX 另含 **USDC** | +| 三所 | Binance / OKX / Gate 同一套 UI 与路由 | + +## 使用 + +1. 系统设置 → 导航显示 → 勾选「账户流水」→ 保存 +2. 顶栏选预设时间并点「应用」 +3. 打开「账户流水」，切换资金/交易 Tab；可点「立即同步」 + +## API + +| 路由 | 说明 | +|------|------| +| `GET /api/account_ledger?account=funding\|trading&page=1` | 按当前 session 时间窗分页查询缓存 | +| `GET /api/account_ledger/stream` | SSE，`event: ledger`，载荷含 `ledger_version` | +| `POST /api/account_ledger/refresh` | 手动触发同步（有冷却，默认 30s） | + +均需登录（与实例其他 API 相同）。 + +## 交易所数据源 + +| 所 | 资金账户 | 交易账户 | +|----|----------|----------| +| Gate | spot `account_book`（USDT） | USDT 永续 `account_book` | +| OKX | `asset/bills`（USDT+USDC） | `account/bills` + `bills-archive`（USDT+USDC） | +| Binance | 充提 + `fetch_transfers`（USDT） | U 本位 `fapi` income（USDT） | + +后台默认回看 **90 天**（`ACCOUNT_LEDGER_LOOKBACK_DAYS`），写入本地 SQLite 缓存后再按顶栏时间窗过滤展示。 +「全部 / 近 6 月」等超出回看窗口的部分，仅能看到缓存内数据。 + +## 环境变量（可选） + +| 变量 | 默认 | 说明 | +|------|------|------| +| `ACCOUNT_LEDGER_POLL_SEC` | `120` | 后台轮询秒数 | +| `ACCOUNT_LEDGER_LOOKBACK_DAYS` | `90` | 拉取与手动同步上限天数 | +| `ACCOUNT_LEDGER_MANUAL_COOLDOWN_SEC` | `30` | 手动同步冷却 | +| `ACCOUNT_LEDGER_SSE_HEARTBEAT_SEC` | `25` | SSE 心跳 | + +## 代码位置 + +``` +lib/account_ledger/ # DB / 同步 / SSE / 注册 / 面板 +lib/exchange/*_ledger_lib.py # 三所拉取适配 +lib/common/static/account_ledger.js +``` + +三所 `app.py` 调用：`install_account_ledger(..., exchange_key=...)`。 + +## 审计摘要（2026-08-10） + +- 路由均 `@login_required`；SSE 仅推版本号，不含账单正文 +- SQL 参数化；`account` 白名单；币种服务端固定 +- 前端表格字段 `escapeHtml` +- **已修复**：手动同步强制套用 lookback 上限 + 冷却，避免滥用刷交易所 API + +详见同目录旁注或 PR 说明；安全复查子代理结论：修复后无未关闭的中高危项。 diff --git a/lib/account_ledger/__init__.py b/lib/account_ledger/__init__.py new file mode 100644 index 0000000..226daa3 --- /dev/null +++ b/lib/account_ledger/__init__.py @@ -0,0 +1 @@ +"""实例账户流水(交易所资金/交易账户账单).""" diff --git a/lib/account_ledger/account_ledger_db.py b/lib/account_ledger/account_ledger_db.py new file mode 100644 index 0000000..6d029a5 --- /dev/null +++ b/lib/account_ledger/account_ledger_db.py @@ -0,0 +1,186 @@ +"""账户流水 SQLite 缓存.""" +from __future__ import annotations + +import time +from typing import Any, Optional + +from lib.account_ledger.account_ledger_normalize import PAGE_SIZE, VALID_ACCOUNTS + + +def ensure_account_ledger_tables(conn) -> None: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS account_ledger_entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account TEXT NOT NULL, + ccy TEXT NOT NULL, + amount REAL NOT NULL, + balance_after REAL, + kind TEXT, + raw_type TEXT, + symbol TEXT, + ref_id TEXT NOT NULL, + ts_ms INTEGER NOT NULL, + note TEXT, + synced_at REAL, + UNIQUE(account, ref_id, ccy, ts_ms) + ) + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_account_ledger_acc_ts " + "ON account_ledger_entries(account, ts_ms DESC)" + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS account_ledger_meta ( + key TEXT PRIMARY KEY, + value TEXT + ) + """ + ) + conn.commit() + + +def meta_get(conn, key: str, default: str = "") -> str: + row = conn.execute( + "SELECT value FROM account_ledger_meta WHERE key=?", (key,) + ).fetchone() + if not row: + return default + try: + return str(row[0] if not hasattr(row, "keys") else row["value"]) + except Exception: + return default + + +def meta_set(conn, key: str, value: str) -> None: + conn.execute( + "INSERT INTO account_ledger_meta(key, value) VALUES(?, ?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (key, str(value)), + ) + + +def upsert_entries(conn, rows: list[dict[str, Any]]) -> int: + if not rows: + return 0 + now = time.time() + n = 0 + for r in rows: + try: + conn.execute( + """ + INSERT INTO account_ledger_entries( + account, ccy, amount, balance_after, kind, raw_type, + symbol, ref_id, ts_ms, note, synced_at + ) VALUES (?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(account, ref_id, ccy, ts_ms) DO UPDATE SET + amount=excluded.amount, + balance_after=excluded.balance_after, + kind=excluded.kind, + raw_type=excluded.raw_type, + symbol=excluded.symbol, + note=excluded.note, + synced_at=excluded.synced_at + """, + ( + r["account"], + r["ccy"], + float(r["amount"]), + r.get("balance_after"), + r.get("kind") or "other", + r.get("raw_type") or "", + r.get("symbol") or "", + r["ref_id"], + int(r["ts_ms"]), + r.get("note") or "", + now, + ), + ) + n += 1 + except Exception: + continue + conn.commit() + return n + + +def query_entries( + conn, + *, + account: str, + start_ms: int, + end_ms: int, + page: int = 1, + page_size: int = PAGE_SIZE, + currencies: Optional[list[str]] = None, +) -> dict[str, Any]: + acc = (account or "").strip().lower() + if acc not in VALID_ACCOUNTS: + return {"items": [], "total": 0, "page": 1, "page_size": page_size, "pages": 0} + page = max(1, int(page or 1)) + page_size = max(1, min(50, int(page_size or PAGE_SIZE))) + start_ms = int(start_ms) + end_ms = int(end_ms) + params: list[Any] = [acc, start_ms, end_ms] + ccy_sql = "" + if currencies: + ccy_list = [c.strip().upper() for c in currencies if c and str(c).strip()] + if ccy_list: + placeholders = ",".join("?" for _ in ccy_list) + ccy_sql = f" AND ccy IN ({placeholders})" + params.extend(ccy_list) + total = conn.execute( + f"SELECT COUNT(*) FROM account_ledger_entries " + f"WHERE account=? AND ts_ms>=? AND ts_ms<=?{ccy_sql}", + params, + ).fetchone()[0] + total = int(total or 0) + pages = (total + page_size - 1) // page_size if total else 0 + if pages and page > pages: + page = pages + offset = (page - 1) * page_size + rows = conn.execute( + f""" + SELECT account, ccy, amount, balance_after, kind, raw_type, symbol, + ref_id, ts_ms, note + FROM account_ledger_entries + WHERE account=? AND ts_ms>=? AND ts_ms<=?{ccy_sql} + ORDER BY ts_ms DESC, id DESC + LIMIT ? OFFSET ? + """, + params + [page_size, offset], + ).fetchall() + items = [] + for r in rows: + if hasattr(r, "keys"): + d = {k: r[k] for k in r.keys()} + else: + d = { + "account": r[0], + "ccy": r[1], + "amount": r[2], + "balance_after": r[3], + "kind": r[4], + "raw_type": r[5], + "symbol": r[6], + "ref_id": r[7], + "ts_ms": r[8], + "note": r[9], + } + from lib.account_ledger.account_ledger_normalize import kind_label_zh + + d["kind_label"] = kind_label_zh(d.get("kind") or "") + items.append(d) + return { + "items": items, + "total": total, + "page": page, + "page_size": page_size, + "pages": pages, + } + + +def prune_older_than(conn, min_ts_ms: int) -> None: + conn.execute("DELETE FROM account_ledger_entries WHERE ts_ms < ?", (int(min_ts_ms),)) + conn.commit() diff --git a/lib/account_ledger/account_ledger_normalize.py b/lib/account_ledger/account_ledger_normalize.py new file mode 100644 index 0000000..d08372c --- /dev/null +++ b/lib/account_ledger/account_ledger_normalize.py @@ -0,0 +1,192 @@ +"""账户流水:交易所原始记录 → 统一行模型.""" +from __future__ import annotations + +from typing import Any, Optional + + +ACCOUNT_FUNDING = "funding" +ACCOUNT_TRADING = "trading" +VALID_ACCOUNTS = frozenset({ACCOUNT_FUNDING, ACCOUNT_TRADING}) + +PAGE_SIZE = 10 + + +def _safe_float(v: Any) -> Optional[float]: + if v is None or v == "": + return None + try: + return float(v) + except (TypeError, ValueError): + return None + + +def _safe_int(v: Any) -> Optional[int]: + if v is None or v == "": + return None + try: + n = float(v) + if n > 1e12: + return int(n) + if n > 1e9: + return int(n) + return int(n) + except (TypeError, ValueError): + return None + + +def _ts_ms(v: Any) -> Optional[int]: + if v is None or v == "": + return None + try: + n = float(v) + except (TypeError, ValueError): + return None + if n > 1e12: + return int(n) + if n > 1e10: + return int(n) + return int(n * 1000.0) + + +def kind_from_raw(raw_type: str, amount: Optional[float] = None) -> str: + t = (raw_type or "").strip().lower() + if not t: + return "other" + if "deposit" in t or t in ("1", "funding_deposit"): + return "deposit" + if "withdraw" in t or "withdrawal" in t: + return "withdraw" + if "transfer" in t or "dnw" in t or t in ("2", "18", "19"): + if amount is not None and amount < 0: + return "transfer_out" + if amount is not None and amount > 0: + return "transfer_in" + return "transfer" + if "funding" in t and "fee" in t: + return "funding_fee" + if t in ("funding_fee", "fundingfee", "8"): + return "funding_fee" + if "commission" in t or "fee" in t or t in ("commission", "5", "fee"): + return "commission" + if "realiz" in t or "pnl" in t or t in ("realized_pnl", "realizedpnl", "3"): + return "realized_pnl" + if "liqui" in t: + return "liquidate" + return "other" + + +def kind_label_zh(kind: str) -> str: + return { + "deposit": "充值", + "withdraw": "提现", + "transfer": "划转", + "transfer_in": "划入", + "transfer_out": "划出", + "realized_pnl": "已实现盈亏", + "funding_fee": "资金费", + "commission": "手续费", + "liquidate": "强平", + "other": "其他", + }.get((kind or "").strip().lower(), "其他") + + +def make_ref_id(*parts: Any) -> str: + bits = [] + for p in parts: + if p is None: + continue + s = str(p).strip() + if s: + bits.append(s) + return "|".join(bits) if bits else "" + + +def normalize_row( + *, + account: str, + ccy: str, + amount: Any, + ts_ms: Any, + ref_id: str, + raw_type: str = "", + balance_after: Any = None, + symbol: str = "", + note: str = "", + kind: str = "", +) -> Optional[dict[str, Any]]: + acc = (account or "").strip().lower() + if acc not in VALID_ACCOUNTS: + return None + ccy_u = (ccy or "").strip().upper() + if not ccy_u: + return None + amt = _safe_float(amount) + if amt is None: + return None + ts = _ts_ms(ts_ms) + if ts is None or ts <= 0: + return None + rid = (ref_id or "").strip() or make_ref_id(acc, ccy_u, ts, amt, raw_type) + k = (kind or "").strip().lower() or kind_from_raw(raw_type, amt) + bal = _safe_float(balance_after) + return { + "account": acc, + "ccy": ccy_u, + "amount": amt, + "balance_after": bal, + "kind": k, + "kind_label": kind_label_zh(k), + "raw_type": (raw_type or "").strip()[:120], + "symbol": (symbol or "").strip()[:80], + "ref_id": rid[:200], + "ts_ms": int(ts), + "note": (note or "").strip()[:240], + } + + +def from_ccxt_ledger_entry(entry: dict[str, Any], *, account: str) -> Optional[dict[str, Any]]: + if not isinstance(entry, dict): + return None + info = entry.get("info") if isinstance(entry.get("info"), dict) else {} + amount = entry.get("amount") + if amount is None: + amount = entry.get("change") + if amount is None: + amount = info.get("balChg") or info.get("change") or info.get("income") or info.get("amount") + ts = entry.get("timestamp") or entry.get("datetime") + if ts is None: + ts = info.get("time") or info.get("uTime") or info.get("ts") or info.get("create_time") or info.get("createDate") + ccy = entry.get("currency") or info.get("ccy") or info.get("asset") or info.get("currency") or "USDT" + raw_type = ( + entry.get("type") + or entry.get("status") + or info.get("type") + or info.get("incomeType") + or info.get("change_type") + or info.get("subType") + or "" + ) + if isinstance(raw_type, (int, float)): + raw_type = str(raw_type) + balance_after = entry.get("balance") or info.get("bal") or info.get("balance") + symbol = entry.get("symbol") or info.get("instId") or info.get("symbol") or info.get("contract") or "" + ref = ( + entry.get("id") + or info.get("billId") + or info.get("tranId") + or info.get("id") + or info.get("trade_id") + or "" + ) + note = entry.get("description") or info.get("info") or info.get("text") or "" + return normalize_row( + account=account, + ccy=str(ccy), + amount=amount, + ts_ms=ts, + ref_id=str(ref) if ref != "" else make_ref_id(account, ccy, ts, amount, raw_type), + raw_type=str(raw_type), + balance_after=balance_after, + symbol=str(symbol or ""), + note=str(note or ""), + ) diff --git a/lib/account_ledger/account_ledger_register.py b/lib/account_ledger/account_ledger_register.py new file mode 100644 index 0000000..997d336 --- /dev/null +++ b/lib/account_ledger/account_ledger_register.py @@ -0,0 +1,208 @@ +"""三所统一:账户流水路由 + 后台同步安装.""" +from __future__ import annotations + +import os +from typing import Any, Callable + +from flask import Flask, Response, jsonify, request, session, stream_with_context +from jinja2 import ChoiceLoader, FileSystemLoader + +from lib.account_ledger.account_ledger_db import ensure_account_ledger_tables, query_entries +from lib.account_ledger.account_ledger_normalize import ( + ACCOUNT_FUNDING, + ACCOUNT_TRADING, + PAGE_SIZE, + VALID_ACCOUNTS, +) +from lib.account_ledger.account_ledger_sync import account_ledger_store +from lib.common.history_window_lib import resolve_list_window + + +def attach_account_ledger_templates(app: Flask, repo_root: str) -> None: + tpl_dir = os.path.join(repo_root, "lib", "account_ledger", "templates") + if not os.path.isdir(tpl_dir): + return + existing = app.jinja_loader + loaders = [FileSystemLoader(tpl_dir)] + if existing is not None: + if isinstance(existing, ChoiceLoader): + loaders = list(existing.loaders) + loaders + else: + loaders.insert(0, existing) + app.jinja_loader = ChoiceLoader(loaders) + + +def _build_fetch_fn(exchange_key: str, app_module: Any) -> Callable: + ex_key = (exchange_key or "").strip().lower() + exchange = getattr(app_module, "exchange", None) + ensure_markets = getattr(app_module, "ensure_markets_loaded", None) + + def _fetch(*, start_ms: int, end_ms: int): + if exchange is None: + return [], ["exchange missing"] + if ex_key == "okx": + from lib.exchange.okx_ledger_lib import fetch_okx_account_ledger + + return fetch_okx_account_ledger( + exchange, + start_ms=start_ms, + end_ms=end_ms, + ensure_markets=ensure_markets, + ) + if ex_key == "binance": + from lib.exchange.binance_ledger_lib import fetch_binance_account_ledger + + return fetch_binance_account_ledger( + exchange, + start_ms=start_ms, + end_ms=end_ms, + ensure_markets=ensure_markets, + ) + from lib.exchange.gate_ledger_lib import fetch_gate_account_ledger + + return fetch_gate_account_ledger( + exchange, + start_ms=start_ms, + end_ms=end_ms, + ensure_markets=ensure_markets, + ) + + return _fetch + + +def _currencies_for_exchange(exchange_key: str) -> list[str]: + if (exchange_key or "").strip().lower() == "okx": + return ["USDT", "USDC"] + return ["USDT"] + + +def install_account_ledger( + app: Flask, + repo_root: str, + app_module: Any, + *, + exchange_key: str = "", +) -> None: + ex = (exchange_key or "").strip().lower() + if not ex: + mod_name = getattr(app_module, "__name__", "") or "" + if "okx" in mod_name.lower(): + ex = "okx" + elif "binance" in mod_name.lower(): + ex = "binance" + else: + ex = "gate" + exchange_key = ex + + attach_account_ledger_templates(app, repo_root) + get_db = app_module.get_db + login_required = app_module.login_required + + # 初始化表 + try: + conn = get_db() + try: + ensure_account_ledger_tables(conn) + finally: + conn.close() + except Exception: + pass + + account_ledger_store.configure( + get_db=get_db, + fetch_fn=_build_fetch_fn(exchange_key, app_module), + exchange_key=str(exchange_key), + ) + account_ledger_store.start() + app.extensions["account_ledger_exchange"] = str(exchange_key).lower() + + def _list_window(): + resolve = getattr(app_module, "_list_window_from_request", None) + if callable(resolve): + return resolve() + return resolve_list_window(request.args, session) + + @app.route("/api/account_ledger") + @login_required + def api_account_ledger(): + account = (request.args.get("account") or ACCOUNT_FUNDING).strip().lower() + if account not in VALID_ACCOUNTS: + account = ACCOUNT_FUNDING + try: + page = int(request.args.get("page") or 1) + except Exception: + page = 1 + win = _list_window() + start_ms = int(win.get("start_ms") or 0) + end_ms = int(win.get("end_ms") or 0) + ccys = _currencies_for_exchange(app.extensions.get("account_ledger_exchange") or "") + conn = get_db() + try: + ensure_account_ledger_tables(conn) + data = query_entries( + conn, + account=account, + start_ms=start_ms, + end_ms=end_ms, + page=page, + page_size=PAGE_SIZE, + currencies=ccys, + ) + finally: + conn.close() + st = account_ledger_store.status_dict() + return jsonify( + { + "ok": True, + "account": account, + "window": { + "preset": win.get("preset"), + "label": win.get("label"), + "start_ms": start_ms, + "end_ms": end_ms, + }, + "currencies": ccys, + **data, + **st, + } + ) + + @app.route("/api/account_ledger/stream") + @login_required + def api_account_ledger_stream(): + return Response( + stream_with_context(account_ledger_store.iter_sse()), + mimetype="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) + + @app.route("/api/account_ledger/refresh", methods=["POST"]) + @login_required + def api_account_ledger_refresh(): + win = _list_window() + body = request.get_json(silent=True) or {} + start_ms = body.get("start_ms", win.get("start_ms")) + end_ms = body.get("end_ms", win.get("end_ms")) + try: + start_i = int(start_ms) if start_ms is not None else None + end_i = int(end_ms) if end_ms is not None else None + except Exception: + start_i, end_i = None, None + result = account_ledger_store.sync_once( + reason="manual", start_ms=start_i, end_ms=end_i + ) + return jsonify(result) + + @app.route("/account_ledger") + @login_required + def account_ledger_page(): + from lib.instance.instance_embed_lib import redirect_to_embed_shell_if_enabled + + redir = redirect_to_embed_shell_if_enabled("account_ledger") + if redir is not None: + return redir + return app_module.render_main_page("account_ledger") diff --git a/lib/account_ledger/account_ledger_sync.py b/lib/account_ledger/account_ledger_sync.py new file mode 100644 index 0000000..75a1f75 --- /dev/null +++ b/lib/account_ledger/account_ledger_sync.py @@ -0,0 +1,252 @@ +"""账户流水:后台定时拉取交易所 + SSE 版本推送.""" +from __future__ import annotations + +import json +import os +import queue +import threading +import time +from collections.abc import Callable, Iterator +from datetime import datetime, timezone +from typing import Any, Optional + +from lib.account_ledger.account_ledger_db import ( + ensure_account_ledger_tables, + meta_get, + meta_set, + prune_older_than, + upsert_entries, +) + +ACCOUNT_LEDGER_POLL_SEC = float(os.getenv("ACCOUNT_LEDGER_POLL_SEC", "120")) +ACCOUNT_LEDGER_LOOKBACK_DAYS = int(os.getenv("ACCOUNT_LEDGER_LOOKBACK_DAYS", "90")) +ACCOUNT_LEDGER_SSE_HEARTBEAT_SEC = float(os.getenv("ACCOUNT_LEDGER_SSE_HEARTBEAT_SEC", "25")) + + +class AccountLedgerStore: + def __init__(self) -> None: + self._lock = threading.Lock() + self.version = 0 + self._subscribers: list[queue.Queue[str | None]] = [] + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._syncing = False + self._get_db: Optional[Callable] = None + self._fetch_fn: Optional[Callable[..., tuple[list[dict[str, Any]], list[str]]]] = None + self._exchange_key = "" + self.last_sync_at: Optional[float] = None + self.last_error: str = "" + self.last_upserted: int = 0 + self._last_manual_at: float = 0.0 + self._manual_cooldown_sec = float(os.getenv("ACCOUNT_LEDGER_MANUAL_COOLDOWN_SEC", "30")) + + def configure( + self, + *, + get_db: Callable, + fetch_fn: Callable[..., tuple[list[dict[str, Any]], list[str]]], + exchange_key: str, + ) -> None: + self._get_db = get_db + self._fetch_fn = fetch_fn + self._exchange_key = (exchange_key or "").strip().lower() + + def start(self) -> None: + if self._thread and self._thread.is_alive(): + return + if not self._get_db or not self._fetch_fn: + return + self._stop.clear() + self._thread = threading.Thread( + target=self._loop, daemon=True, name=f"account-ledger-{self._exchange_key or 'x'}" + ) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + self._broadcast(close=True) + + def lookback_bounds_ms(self, start_ms: Optional[int] = None, end_ms: Optional[int] = None) -> tuple[int, int]: + now = datetime.now(timezone.utc) + end = int(end_ms) if end_ms is not None else int(now.timestamp() * 1000) + floor = int(end - ACCOUNT_LEDGER_LOOKBACK_DAYS * 86400 * 1000) + if start_ms is not None: + start = max(int(start_ms), floor) + else: + start = floor + if start > end: + start, end = end, start + return start, end + + def sync_once( + self, + *, + reason: str = "poll", + start_ms: Optional[int] = None, + end_ms: Optional[int] = None, + ) -> dict[str, Any]: + if not self._get_db or not self._fetch_fn: + return {"ok": False, "msg": "未配置"} + with self._lock: + if self._syncing: + return {"ok": True, "busy": True, "ledger_version": self.version} + if reason == "manual": + gap = time.time() - self._last_manual_at + if gap < self._manual_cooldown_sec: + wait = int(self._manual_cooldown_sec - gap) + 1 + return { + "ok": False, + "msg": f"同步过于频繁,请 {wait}s 后再试", + "ledger_version": self.version, + } + self._syncing = True + try: + start, end = self.lookback_bounds_ms(start_ms, end_ms) + rows, errors = self._fetch_fn(start_ms=start, end_ms=end) + conn = self._get_db() + try: + ensure_account_ledger_tables(conn) + n = upsert_entries(conn, rows or []) + # 保留略宽于 lookback 的缓存 + prune_ms = int( + (datetime.now(timezone.utc).timestamp() - (ACCOUNT_LEDGER_LOOKBACK_DAYS + 7) * 86400) + * 1000 + ) + prune_older_than(conn, prune_ms) + self.last_sync_at = time.time() + self.last_upserted = n + self.last_error = "; ".join(errors[:3]) if errors else "" + meta_set(conn, "last_sync_at", str(self.last_sync_at)) + meta_set(conn, "last_error", self.last_error) + meta_set(conn, "last_upserted", str(n)) + conn.commit() + finally: + try: + conn.close() + except Exception: + pass + if reason == "manual": + self._last_manual_at = time.time() + ver = self.bump(reason) + return { + "ok": True, + "ledger_version": ver, + "upserted": n, + "errors": errors, + "start_ms": start, + "end_ms": end, + } + except Exception as e: + self.last_error = str(e) + try: + conn = self._get_db() + try: + ensure_account_ledger_tables(conn) + meta_set(conn, "last_error", self.last_error) + conn.commit() + finally: + conn.close() + except Exception: + pass + return {"ok": False, "msg": str(e), "ledger_version": self.version} + finally: + with self._lock: + self._syncing = False + + def bump(self, reason: str = "poll") -> int: + with self._lock: + self.version += 1 + ver = self.version + payload = json.dumps( + {"ledger_version": ver, "reason": reason, "exchange": self._exchange_key}, + ensure_ascii=False, + ) + self._broadcast(payload) + return ver + + def status_dict(self) -> dict[str, Any]: + last_at = self.last_sync_at + if last_at is None and self._get_db: + try: + conn = self._get_db() + try: + ensure_account_ledger_tables(conn) + raw = meta_get(conn, "last_sync_at", "") + if raw: + last_at = float(raw) + self.last_error = meta_get(conn, "last_error", self.last_error) + finally: + conn.close() + except Exception: + pass + return { + "ledger_version": self.version, + "poll_sec": ACCOUNT_LEDGER_POLL_SEC, + "lookback_days": ACCOUNT_LEDGER_LOOKBACK_DAYS, + "last_sync_at": last_at, + "last_error": self.last_error, + "last_upserted": self.last_upserted, + "exchange": self._exchange_key, + } + + def _loop(self) -> None: + # 启动后稍等再拉,避免和启动高峰撞车 + if self._stop.wait(3): + return + while not self._stop.is_set(): + try: + self.sync_once(reason="poll") + except Exception: + pass + if self._stop.wait(ACCOUNT_LEDGER_POLL_SEC): + break + + def _broadcast(self, event: str | None = None, *, close: bool = False) -> None: + with self._lock: + subs = list(self._subscribers) + dead: list[queue.Queue[str | None]] = [] + for q in subs: + try: + q.put_nowait(None if close else event) + except Exception: + dead.append(q) + if dead: + with self._lock: + for q in dead: + if q in self._subscribers: + self._subscribers.remove(q) + + def _subscribe(self) -> queue.Queue[str | None]: + q: queue.Queue[str | None] = queue.Queue(maxsize=16) + with self._lock: + self._subscribers.append(q) + return q + + def _unsubscribe(self, q: queue.Queue[str | None]) -> None: + with self._lock: + if q in self._subscribers: + self._subscribers.remove(q) + + def iter_sse(self) -> Iterator[str]: + q = self._subscribe() + try: + yield f"event: ledger\ndata: {json.dumps({'ledger_version': self.version, 'reason': 'hello'}, ensure_ascii=False)}\n\n" + last_hb = time.time() + while not self._stop.is_set(): + try: + item = q.get(timeout=1.0) + except queue.Empty: + item = "timeout" + if item is None: + break + if item != "timeout": + yield f"event: ledger\ndata: {item}\n\n" + last_hb = time.time() + elif time.time() - last_hb >= ACCOUNT_LEDGER_SSE_HEARTBEAT_SEC: + yield ": heartbeat\n\n" + last_hb = time.time() + finally: + self._unsubscribe(q) + + +account_ledger_store = AccountLedgerStore() diff --git a/lib/account_ledger/templates/account_ledger_panel.html b/lib/account_ledger/templates/account_ledger_panel.html new file mode 100644 index 0000000..3a5fba0 --- /dev/null +++ b/lib/account_ledger/templates/account_ledger_panel.html @@ -0,0 +1,72 @@ +{# 账户流水:资金/交易 Tab · 交易所账单 · SSE #} +
+
+
+

账户流水

+ +
+ +
+
+ + +
+

+
+ + + + + + + + + + + + + + + +
+
+ + — + +
+
+ diff --git a/lib/common/static/account_ledger.js b/lib/common/static/account_ledger.js new file mode 100644 index 0000000..a7ce569 --- /dev/null +++ b/lib/common/static/account_ledger.js @@ -0,0 +1,337 @@ +/** + * 账户流水:资金/交易 Tab · 分页 10 · SSE 自动刷新 · 时间窗跟随顶栏预设. + */ +(function (global) { + const PAGE_SIZE = 10; + let account = "funding"; + let page = 1; + let pages = 0; + let localVersion = 0; + let es = null; + let reconnectTimer = null; + let loading = false; + let booted = false; + + function root() { + const active = document.querySelector('.embed-tab-pane.is-active-pane [data-account-ledger="1"]'); + if (active) return active; + return document.getElementById("account-ledger-root"); + } + + function $(id) { + const r = root(); + return (r && r.querySelector("#" + id)) || document.getElementById(id); + } + + function escapeHtml(s) { + return String(s == null ? "" : s) + .replace(/&/g, "&") + .replace(//g, ">") + .replace(/"/g, """); + } + + function listWindowQs() { + if (typeof global.listWindowQueryString === "function") { + const q = global.listWindowQueryString(); + return q ? (q.charAt(0) === "?" ? q.slice(1) : q) : ""; + } + try { + return new URLSearchParams(location.search).toString(); + } catch (_) { + return ""; + } + } + + function fmtBj(ms) { + const n = Number(ms); + if (!Number.isFinite(n) || n <= 0) return "—"; + try { + const d = new Date(n); + const parts = new Intl.DateTimeFormat("zh-CN", { + timeZone: "Asia/Shanghai", + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + hour12: false, + }).formatToParts(d); + const get = (t) => (parts.find((p) => p.type === t) || {}).value || ""; + return ( + get("year") + + "-" + + get("month") + + "-" + + get("day") + + " " + + get("hour") + + ":" + + get("minute") + + ":" + + get("second") + ); + } catch (_) { + return "—"; + } + } + + function fmtAmt(v) { + const n = Number(v); + if (!Number.isFinite(n)) return "—"; + const cls = n > 0 ? "account-ledger-amt-pos" : n < 0 ? "account-ledger-amt-neg" : ""; + const sign = n > 0 ? "+" : ""; + return '' + sign + n.toFixed(6).replace(/\.?0+$/, "") + ""; + } + + function fmtBal(v) { + if (v == null || v === "") return "—"; + const n = Number(v); + if (!Number.isFinite(n)) return "—"; + return n.toFixed(6).replace(/\.?0+$/, ""); + } + + function setStatus(msg, isErr) { + const el = $("account-ledger-status"); + if (!el) return; + el.textContent = msg || ""; + el.style.color = isErr ? "#f07178" : ""; + } + + function setSyncLabel(data) { + const el = $("account-ledger-sync"); + if (!el) return; + const ts = data && data.last_sync_at; + if (!ts) { + el.textContent = "尚未同步"; + return; + } + el.textContent = "同步 " + fmtBj(Number(ts) * 1000); + } + + function renderRows(items) { + const tbody = $("account-ledger-tbody"); + if (!tbody) return; + if (!items || !items.length) { + tbody.innerHTML = '当前时间窗暂无流水'; + return; + } + tbody.innerHTML = items + .map(function (it) { + const note = [it.symbol, it.note, it.raw_type].filter(Boolean).join(" · "); + return ( + "" + + "" + + escapeHtml(fmtBj(it.ts_ms)) + + "" + + "" + + escapeHtml(it.ccy || "") + + "" + + "" + + escapeHtml(it.kind_label || it.kind || "") + + "" + + "" + + fmtAmt(it.amount) + + "" + + "" + + escapeHtml(fmtBal(it.balance_after)) + + "" + + "" + + escapeHtml(note || "—") + + "" + + "" + ); + }) + .join(""); + } + + function renderPager(data) { + pages = Number(data.pages || 0); + page = Number(data.page || 1); + const info = $("account-ledger-page-info"); + const prev = $("account-ledger-prev"); + const next = $("account-ledger-next"); + if (info) { + info.textContent = + "第 " + page + " / " + (pages || 1) + " 页 · 共 " + (data.total || 0) + " 条 · 每页 " + PAGE_SIZE; + } + if (prev) prev.disabled = page <= 1; + if (next) next.disabled = !pages || page >= pages; + } + + async function loadList(opts) { + const r = root(); + if (!r) return; + if (loading) return; + loading = true; + const force = opts && opts.force; + try { + if (!force) setStatus("加载中…"); + const qs = new URLSearchParams(listWindowQs()); + qs.set("account", account); + qs.set("page", String(page)); + const res = await fetch("/api/account_ledger?" + qs.toString(), { + credentials: "same-origin", + }); + const data = await res.json().catch(function () { + return {}; + }); + if (!res.ok || data.ok === false) { + throw new Error(data.msg || res.statusText || "加载失败"); + } + if (data.ledger_version != null) localVersion = Number(data.ledger_version) || localVersion; + renderRows(data.items || []); + renderPager(data); + setSyncLabel(data); + const winLabel = (data.window && data.window.label) || ""; + const err = data.last_error ? " · 同步提示: " + data.last_error : ""; + setStatus( + (winLabel ? "时间窗 " + winLabel + " · " : "") + + (account === "trading" ? "交易账户" : "资金账户") + + err, + !!data.last_error + ); + } catch (e) { + setStatus(e.message || String(e), true); + } finally { + loading = false; + } + } + + async function refreshNow() { + setStatus("正在从交易所同步…"); + try { + const res = await fetch("/api/account_ledger/refresh", { + method: "POST", + credentials: "same-origin", + headers: { "Content-Type": "application/json" }, + body: "{}", + }); + const data = await res.json().catch(function () { + return {}; + }); + if (!res.ok || data.ok === false) { + throw new Error(data.msg || "同步失败"); + } + await loadList({ force: true }); + } catch (e) { + setStatus(e.message || String(e), true); + } + } + + function bindUi() { + const r = root(); + if (!r || r.getAttribute("data-ledger-bound") === "1") return; + r.setAttribute("data-ledger-bound", "1"); + r.querySelectorAll(".account-ledger-tab").forEach(function (btn) { + btn.addEventListener("click", function () { + const acc = btn.getAttribute("data-ledger-account") || "funding"; + if (acc === account) return; + account = acc; + page = 1; + r.querySelectorAll(".account-ledger-tab").forEach(function (b) { + const on = b.getAttribute("data-ledger-account") === account; + b.classList.toggle("active", on); + b.setAttribute("aria-selected", on ? "true" : "false"); + }); + loadList(); + }); + }); + const prev = $("account-ledger-prev"); + const next = $("account-ledger-next"); + const ref = $("account-ledger-refresh"); + if (prev) + prev.addEventListener("click", function () { + if (page > 1) { + page -= 1; + loadList(); + } + }); + if (next) + next.addEventListener("click", function () { + if (!pages || page < pages) { + page += 1; + loadList(); + } + }); + if (ref) ref.addEventListener("click", refreshNow); + } + + function connectSse() { + if (es) { + try { + es.close(); + } catch (_) {} + es = null; + } + if (typeof EventSource === "undefined") return; + try { + es = new EventSource("/api/account_ledger/stream"); + es.addEventListener("ledger", function (ev) { + let data = {}; + try { + data = JSON.parse(ev.data || "{}"); + } catch (_) {} + const ver = Number(data.ledger_version || 0); + if (ver && ver !== localVersion) { + localVersion = ver; + loadList({ force: true }); + } + }); + es.onerror = function () { + try { + es.close(); + } catch (_) {} + es = null; + if (reconnectTimer) clearTimeout(reconnectTimer); + reconnectTimer = setTimeout(connectSse, 5000); + }; + } catch (_) {} + } + + function boot() { + const r = root(); + if (!r) return; + bindUi(); + if (!booted) { + booted = true; + connectSse(); + } + loadList(); + } + + function onTabActivated(tab) { + if (tab !== "account_ledger") return; + boot(); + } + + global.AccountLedgerPage = { + boot: boot, + onTabActivated: onTabActivated, + reload: function () { + page = 1; + loadList(); + }, + }; + + document.addEventListener("DOMContentLoaded", function () { + const page = + (document.body && document.body.getAttribute("data-page")) || + (document.body && document.body.getAttribute("data-initial-tab")) || + ""; + if (page === "account_ledger" || root()) { + // embed 延后到 tab 激活;独立页直接 boot + if (!document.body || document.body.getAttribute("data-embed-shell") !== "1") { + boot(); + } else if (page === "account_ledger") { + boot(); + } + } + }); + + document.addEventListener("instance-embed-tab-activated", function (ev) { + const tab = ev && ev.detail && ev.detail.tab; + onTabActivated(tab); + }); +})(window); diff --git a/lib/common/static/instance_embed.js b/lib/common/static/instance_embed.js index 641763c..be8b93b 100644 --- a/lib/common/static/instance_embed.js +++ b/lib/common/static/instance_embed.js @@ -5,6 +5,7 @@ (function (global) { const TAB_PATH = { dashboard: "/dashboard", + account_ledger: "/account_ledger", key_monitor: "/key_monitor", trade: "/trade", strategy: "/strategy", @@ -114,6 +115,9 @@ if (tab === "dashboard" && global.InstanceDashboard && typeof global.InstanceDashboard.init === "function") { global.InstanceDashboard.init(!!revisit); } + if (tab === "account_ledger" && global.AccountLedgerPage && typeof global.AccountLedgerPage.boot === "function") { + global.AccountLedgerPage.boot(); + } if (!revisit && tab === "strategy" && typeof global.initStrategyRollForm === "function") { global.initStrategyRollForm(); } diff --git a/lib/common/static/instance_settings_prefs.js b/lib/common/static/instance_settings_prefs.js index e9c7ac3..bd61802 100644 --- a/lib/common/static/instance_settings_prefs.js +++ b/lib/common/static/instance_settings_prefs.js @@ -20,7 +20,11 @@ } /** 默认关闭的导航开关:缺失时按 false,不能用 !== false */ - const NAV_DEFAULT_OFF = { show_nav_dashboard: true, show_nav_system_guide: true }; + const NAV_DEFAULT_OFF = { + show_nav_dashboard: true, + show_nav_account_ledger: true, + show_nav_system_guide: true, + }; function navPrefShow(display, key) { if (!key) return true; @@ -31,6 +35,7 @@ function applyDisplayToNav(display) { const map = { dashboard: "show_nav_dashboard", + account_ledger: "show_nav_account_ledger", key_monitor: "show_nav_key_monitor", trade: "show_nav_trade", strategy: "show_nav_strategy", @@ -66,6 +71,7 @@ const d = DISPLAY(); const map = { dashboard: "show_nav_dashboard", + account_ledger: "show_nav_account_ledger", key_monitor: "show_nav_key_monitor", trade: "show_nav_trade", strategy: "show_nav_strategy", diff --git a/lib/exchange/binance_ledger_lib.py b/lib/exchange/binance_ledger_lib.py new file mode 100644 index 0000000..795b53a --- /dev/null +++ b/lib/exchange/binance_ledger_lib.py @@ -0,0 +1,187 @@ +"""Binance:交易账户 futures income;资金账户 deposits/withdrawals/transfers.USDT.""" +from __future__ import annotations + +from typing import Any, Callable, Optional + +from lib.account_ledger.account_ledger_normalize import ( + ACCOUNT_FUNDING, + ACCOUNT_TRADING, + from_ccxt_ledger_entry, + kind_from_raw, + make_ref_id, + normalize_row, +) + + +def _paginate_income(exchange, *, start_ms: int, end_ms: int, max_pages: int = 15) -> list[dict]: + out: list[dict] = [] + cursor = int(start_ms) + end = int(end_ms) + for _ in range(max_pages): + try: + if hasattr(exchange, "fapiPrivateGetIncome"): + batch = exchange.fapiPrivateGetIncome( + {"startTime": cursor, "endTime": end, "limit": 1000} + ) + else: + batch = exchange.fetch_ledger( + "USDT", cursor, 1000, {"type": "swap", "until": end} + ) + # already unified + return batch or [] + except Exception: + break + if not batch: + break + out.extend(batch) + if len(batch) < 1000: + break + last_t = batch[-1].get("time") or batch[-1].get("timestamp") + try: + last_i = int(float(last_t)) + except Exception: + break + if last_i >= end: + break + cursor = last_i + 1 + return out + + +def _income_to_row(raw: dict) -> Optional[dict[str, Any]]: + if not isinstance(raw, dict): + return None + # raw fapi income + if "income" in raw or "incomeType" in raw: + amt = raw.get("income") + ts = raw.get("time") + ccy = raw.get("asset") or "USDT" + raw_type = str(raw.get("incomeType") or "") + ref = str(raw.get("tranId") or raw.get("tradeId") or "") + return normalize_row( + account=ACCOUNT_TRADING, + ccy=str(ccy), + amount=amt, + ts_ms=ts, + ref_id=ref or make_ref_id("trading", ccy, ts, amt, raw_type), + raw_type=raw_type, + symbol=str(raw.get("symbol") or ""), + note=str(raw.get("info") or ""), + kind=kind_from_raw(raw_type, float(amt) if amt is not None else None), + ) + return from_ccxt_ledger_entry(raw, account=ACCOUNT_TRADING) + + +def _dep_wd_to_row(entry: dict, *, kind: str) -> Optional[dict[str, Any]]: + if not isinstance(entry, dict): + return None + info = entry.get("info") if isinstance(entry.get("info"), dict) else {} + amount = entry.get("amount") + ts = entry.get("timestamp") or info.get("insertTime") or info.get("applyTime") + ccy = entry.get("currency") or info.get("coin") or "USDT" + status = entry.get("status") or info.get("status") or "" + ref = str(entry.get("id") or info.get("txId") or info.get("id") or "") + amt = amount + try: + af = float(amount) + if kind == "withdraw" and af > 0: + af = -af + amt = af + except Exception: + pass + return normalize_row( + account=ACCOUNT_FUNDING, + ccy=str(ccy), + amount=amt, + ts_ms=ts, + ref_id=ref or make_ref_id("funding", kind, ccy, ts, amount), + raw_type=kind, + note=str(status), + kind=kind, + ) + + +def _transfer_to_row(entry: dict) -> Optional[dict[str, Any]]: + if not isinstance(entry, dict): + return None + info = entry.get("info") if isinstance(entry.get("info"), dict) else {} + amount = entry.get("amount") + ts = entry.get("timestamp") or info.get("timestamp") + ccy = entry.get("currency") or info.get("asset") or "USDT" + ref = str(entry.get("id") or info.get("tranId") or info.get("id") or "") + frm = str(entry.get("fromAccount") or info.get("from") or "") + to = str(entry.get("toAccount") or info.get("to") or "") + try: + amt = float(amount) + except Exception: + return None + # 资金侧视角:从资金转出为负,转入为正(粗分) + note = f"{frm}->{to}".strip("->") + raw_type = "transfer" + return normalize_row( + account=ACCOUNT_FUNDING, + ccy=str(ccy), + amount=amt, + ts_ms=ts, + ref_id=ref or make_ref_id("funding", "transfer", ccy, ts, amt), + raw_type=raw_type, + note=note, + kind=kind_from_raw("transfer", amt), + ) + + +def fetch_binance_account_ledger( + exchange, + *, + start_ms: int, + end_ms: int, + ensure_markets: Optional[Callable[[], None]] = None, +) -> tuple[list[dict[str, Any]], list[str]]: + errors: list[str] = [] + rows: list[dict[str, Any]] = [] + if ensure_markets: + try: + ensure_markets() + except Exception as e: + errors.append(f"markets:{e}") + + # 交易账户 + try: + raw = _paginate_income(exchange, start_ms=start_ms, end_ms=end_ms) + for e in raw: + n = _income_to_row(e) + if n and n["ccy"] == "USDT": + rows.append(n) + except Exception as e: + errors.append(f"trading:{e}") + + # 资金账户:充提 + 划转 + for label, fn, kind in ( + ("deposits", "fetch_deposits", "deposit"), + ("withdrawals", "fetch_withdrawals", "withdraw"), + ): + try: + meth = getattr(exchange, fn, None) + if not callable(meth): + continue + batch = meth("USDT", int(start_ms), 1000, {"until": int(end_ms)}) or [] + for e in batch: + n = _dep_wd_to_row(e, kind=kind) + if n: + rows.append(n) + except Exception as e: + errors.append(f"{label}:{e}") + + try: + if hasattr(exchange, "fetch_transfers"): + batch = ( + exchange.fetch_transfers("USDT", int(start_ms), 1000, {"until": int(end_ms)}) + or [] + ) + for e in batch: + n = _transfer_to_row(e) + if n: + rows.append(n) + except Exception as e: + errors.append(f"transfers:{e}") + + return rows, errors diff --git a/lib/exchange/gate_ledger_lib.py b/lib/exchange/gate_ledger_lib.py new file mode 100644 index 0000000..9df63a9 --- /dev/null +++ b/lib/exchange/gate_ledger_lib.py @@ -0,0 +1,212 @@ +"""Gate:资金账户(spot account_book) + 交易账户(futures account_book),USDT.""" +from __future__ import annotations + +from typing import Any, Callable, Optional + +from lib.account_ledger.account_ledger_normalize import ( + ACCOUNT_FUNDING, + ACCOUNT_TRADING, + from_ccxt_ledger_entry, + kind_from_raw, + make_ref_id, + normalize_row, +) + + +def _sec(ms: int) -> int: + return max(0, int(int(ms) // 1000)) + + +def _paginate_spot_book(exchange, *, start_ms: int, end_ms: int, max_pages: int = 10) -> list[dict]: + out: list[dict] = [] + # Gate spot account_book: from/to 为秒 + cursor = _sec(start_ms) + end = _sec(end_ms) + for _ in range(max_pages): + try: + batch = exchange.privateSpotGetAccountBook( + { + "currency": "USDT", + "from": cursor, + "to": end, + "limit": 100, + } + ) + except Exception: + break + if not batch: + break + if isinstance(batch, dict): + batch = batch.get("data") or batch.get("result") or [] + if not isinstance(batch, list) or not batch: + break + out.extend(batch) + if len(batch) < 100: + break + last_t = batch[-1].get("time") or batch[-1].get("create_time") + try: + last_i = int(float(last_t)) + except Exception: + break + # spot 返回秒 + if last_i > 1e12: + last_i = last_i // 1000 + if last_i >= end: + break + cursor = last_i + 1 + return out + + +def _paginate_swap_book(exchange, *, start_ms: int, end_ms: int, max_pages: int = 10) -> list[dict]: + out: list[dict] = [] + cursor = _sec(start_ms) + end = _sec(end_ms) + for _ in range(max_pages): + try: + batch = exchange.privateFuturesGetSettleAccountBook( + { + "settle": "usdt", + "from": cursor, + "to": end, + "limit": 100, + } + ) + except Exception: + break + if not batch: + break + if isinstance(batch, dict): + batch = batch.get("data") or batch.get("result") or [] + if not isinstance(batch, list) or not batch: + break + out.extend(batch) + if len(batch) < 100: + break + last_t = batch[-1].get("time") + try: + last_i = int(float(last_t)) + except Exception: + break + if last_i > 1e12: + last_i = last_i // 1000 + if last_i >= end: + break + cursor = last_i + 1 + return out + + +def _spot_row(raw: dict) -> Optional[dict[str, Any]]: + if not isinstance(raw, dict): + return None + amt = raw.get("change") + ts = raw.get("time") or raw.get("create_time") + # 秒 → 毫秒 + try: + t = float(ts) + if t < 1e12: + t = t * 1000.0 + ts = t + except Exception: + pass + raw_type = str(raw.get("type") or raw.get("change_type") or "") + bal = raw.get("balance") + ref = str(raw.get("id") or raw.get("txid") or "") + return normalize_row( + account=ACCOUNT_FUNDING, + ccy="USDT", + amount=amt, + ts_ms=ts, + ref_id=ref or make_ref_id("funding", raw_type, ts, amt), + raw_type=raw_type, + balance_after=bal, + note=str(raw.get("text") or ""), + kind=kind_from_raw(raw_type, float(amt) if amt is not None else None), + ) + + +def _swap_row(raw: dict) -> Optional[dict[str, Any]]: + if not isinstance(raw, dict): + return None + # futures account_book: change, balance, type, text, time, contract... + amt = raw.get("change") + ts = raw.get("time") + try: + t = float(ts) + if t < 1e12: + t = t * 1000.0 + ts = t + except Exception: + pass + raw_type = str(raw.get("type") or "") + bal = raw.get("balance") + ref = str(raw.get("id") or "") + return normalize_row( + account=ACCOUNT_TRADING, + ccy="USDT", + amount=amt, + ts_ms=ts, + ref_id=ref or make_ref_id("trading", raw_type, ts, amt, raw.get("contract")), + raw_type=raw_type, + balance_after=bal, + symbol=str(raw.get("contract") or ""), + note=str(raw.get("text") or ""), + kind=kind_from_raw(raw_type, float(amt) if amt is not None else None), + ) + + +def fetch_gate_account_ledger( + exchange, + *, + start_ms: int, + end_ms: int, + ensure_markets: Optional[Callable[[], None]] = None, +) -> tuple[list[dict[str, Any]], list[str]]: + errors: list[str] = [] + rows: list[dict[str, Any]] = [] + if ensure_markets: + try: + ensure_markets() + except Exception as e: + errors.append(f"markets:{e}") + + try: + for e in _paginate_spot_book(exchange, start_ms=start_ms, end_ms=end_ms): + n = _spot_row(e) + if n: + rows.append(n) + except Exception as e: + errors.append(f"funding:{e}") + # 回退 ccxt fetch_ledger + try: + batch = exchange.fetch_ledger( + "USDT", int(start_ms), 100, {"type": "spot", "until": int(end_ms)} + ) or [] + for e in batch: + n = from_ccxt_ledger_entry(e, account=ACCOUNT_FUNDING) + if n: + rows.append(n) + except Exception as e2: + errors.append(f"funding_fallback:{e2}") + + try: + for e in _paginate_swap_book(exchange, start_ms=start_ms, end_ms=end_ms): + n = _swap_row(e) + if n: + rows.append(n) + except Exception as e: + errors.append(f"trading:{e}") + try: + batch = exchange.fetch_ledger( + "USDT", + int(start_ms), + 100, + {"type": "swap", "settle": "usdt", "until": int(end_ms)}, + ) or [] + for e in batch: + n = from_ccxt_ledger_entry(e, account=ACCOUNT_TRADING) + if n: + rows.append(n) + except Exception as e2: + errors.append(f"trading_fallback:{e2}") + + return rows, errors diff --git a/lib/exchange/okx_ledger_lib.py b/lib/exchange/okx_ledger_lib.py new file mode 100644 index 0000000..e4f7cb8 --- /dev/null +++ b/lib/exchange/okx_ledger_lib.py @@ -0,0 +1,99 @@ +"""OKX:资金账户 asset bills + 交易账户 account bills;USDT + USDC.""" +from __future__ import annotations + +from typing import Any, Callable, Optional + +from lib.account_ledger.account_ledger_normalize import ( + ACCOUNT_FUNDING, + ACCOUNT_TRADING, + from_ccxt_ledger_entry, +) + +OKX_LEDGER_CCYS = ("USDT", "USDC") + + +def _fetch_one( + exchange, + *, + code: str, + since: int, + until: int, + method: str, + max_pages: int = 10, +) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + after = None + for _ in range(max_pages): + params: dict[str, Any] = {"method": method, "until": int(until)} + if after: + params["after"] = after + try: + batch = exchange.fetch_ledger(code, int(since), 100, params) or [] + except Exception: + # archive / bills 窗口差异:失败则停 + break + if not batch: + break + out.extend(batch) + if len(batch) < 100: + break + # OKX 翻页用 billId + last = batch[-1] + info = last.get("info") if isinstance(last.get("info"), dict) else {} + bid = last.get("id") or info.get("billId") + if not bid: + break + after = str(bid) + return out + + +def fetch_okx_account_ledger( + exchange, + *, + start_ms: int, + end_ms: int, + ensure_markets: Optional[Callable[[], None]] = None, +) -> tuple[list[dict[str, Any]], list[str]]: + errors: list[str] = [] + rows: list[dict[str, Any]] = [] + if ensure_markets: + try: + ensure_markets() + except Exception as e: + errors.append(f"markets:{e}") + + for ccy in OKX_LEDGER_CCYS: + # 资金账户 + try: + raw = _fetch_one( + exchange, + code=ccy, + since=start_ms, + until=end_ms, + method="privateGetAssetBills", + ) + for e in raw: + n = from_ccxt_ledger_entry(e, account=ACCOUNT_FUNDING) + if n: + rows.append(n) + except Exception as e: + errors.append(f"funding:{ccy}:{e}") + + # 交易账户:近 3 月 archive + 近 7 日 bills(去重靠 upsert) + for method in ("privateGetAccountBillsArchive", "privateGetAccountBills"): + try: + raw = _fetch_one( + exchange, + code=ccy, + since=start_ms, + until=end_ms, + method=method, + ) + for e in raw: + n = from_ccxt_ledger_entry(e, account=ACCOUNT_TRADING) + if n: + rows.append(n) + except Exception as e: + errors.append(f"trading:{ccy}:{method}:{e}") + + return rows, errors diff --git a/lib/hub/hub_bridge.py b/lib/hub/hub_bridge.py index 10601f8..2c3ffa4 100644 --- a/lib/hub/hub_bridge.py +++ b/lib/hub/hub_bridge.py @@ -76,6 +76,7 @@ def install_instance_theme_static(app) -> None: "instance_live.js": "application/javascript; charset=utf-8", "instance_settings_prefs.js": "application/javascript; charset=utf-8", "instance_dashboard.js": "application/javascript; charset=utf-8", + "account_ledger.js": "application/javascript; charset=utf-8", "options_expiry_countdown.js": "application/javascript; charset=utf-8", "options_panel.js": "application/javascript; charset=utf-8", "order_entry_model.js": "application/javascript; charset=utf-8", diff --git a/lib/instance/instance_display_prefs_lib.py b/lib/instance/instance_display_prefs_lib.py index 1601edb..e1df362 100644 --- a/lib/instance/instance_display_prefs_lib.py +++ b/lib/instance/instance_display_prefs_lib.py @@ -9,6 +9,7 @@ DISPLAY_RUNTIME_PREFIX = "display." DEFAULT_INSTANCE_DISPLAY: dict[str, bool] = { "show_nav_dashboard": False, + "show_nav_account_ledger": False, "show_nav_key_monitor": True, "show_nav_trade": True, "show_nav_strategy": True, @@ -30,6 +31,7 @@ DEFAULT_INSTANCE_DISPLAY: dict[str, bool] = { DISPLAY_LABELS: dict[str, str] = { "show_nav_dashboard": "数据看板", + "show_nav_account_ledger": "账户流水", "show_nav_key_monitor": "关键位监控", "show_nav_trade": "实盘下单", "show_nav_strategy": "策略交易", @@ -51,6 +53,7 @@ DISPLAY_LABELS: dict[str, str] = { NAV_TAB_ALLOWED: dict[str, str] = { "dashboard": "show_nav_dashboard", + "account_ledger": "show_nav_account_ledger", "key_monitor": "show_nav_key_monitor", "trade": "show_nav_trade", "strategy": "show_nav_strategy", @@ -116,6 +119,7 @@ def tab_allowed(tab: str, display: Optional[dict[str, bool]] = None) -> bool: def display_meta_for_ui() -> list[dict[str, Any]]: nav_keys = [ "show_nav_dashboard", + "show_nav_account_ledger", "show_nav_key_monitor", "show_nav_trade", "show_nav_strategy", diff --git a/lib/instance/instance_embed_lib.py b/lib/instance/instance_embed_lib.py index 9fa673d..f0b8e45 100644 --- a/lib/instance/instance_embed_lib.py +++ b/lib/instance/instance_embed_lib.py @@ -12,6 +12,7 @@ from jinja2 import ChoiceLoader, FileSystemLoader EMBED_TABS: tuple[str, ...] = ( "dashboard", + "account_ledger", "key_monitor", "trade", "strategy", @@ -31,6 +32,7 @@ PATH_TO_EMBED_TAB: dict[str, str] = { "/": "trade", "/trade": "trade", "/dashboard": "dashboard", + "/account_ledger": "account_ledger", "/key_monitor": "key_monitor", "/strategy": "strategy", "/strategy/trend": "strategy", diff --git a/lib/instance/templates/display_prefs_panel.html b/lib/instance/templates/display_prefs_panel.html index 38a744b..edcbecc 100644 --- a/lib/instance/templates/display_prefs_panel.html +++ b/lib/instance/templates/display_prefs_panel.html @@ -10,7 +10,7 @@
{% for item in group.entries %} {% endfor %} diff --git a/lib/instance/templates/embed_page_fragment.html b/lib/instance/templates/embed_page_fragment.html index e49a528..e64fd96 100644 --- a/lib/instance/templates/embed_page_fragment.html +++ b/lib/instance/templates/embed_page_fragment.html @@ -119,6 +119,8 @@
{% if page == 'dashboard' %} {% include 'dashboard_panel.html' %} + {% elif page == 'account_ledger' %} + {% include 'account_ledger_panel.html' %} {% elif page == 'key_monitor' %} {% include 'key_monitor_panel.html' %} {% elif page == 'trade' %} diff --git a/lib/instance/templates/embed_shell.html b/lib/instance/templates/embed_shell.html index 9be09b6..0accef1 100644 --- a/lib/instance/templates/embed_shell.html +++ b/lib/instance/templates/embed_shell.html @@ -32,6 +32,7 @@
+ {% if not intraday_discipline and display.show_nav_strategy %} @@ -197,6 +198,8 @@
{% if page == 'dashboard' %} {% include 'dashboard_panel.html' %} + {% elif page == 'account_ledger' %} + {% include 'account_ledger_panel.html' %} {% elif page == 'key_monitor' %} {% include 'key_monitor_panel.html' %} {% elif page == 'trade' %} @@ -2028,6 +2031,7 @@ setInterval(refreshPriceSnapshotConditional, {{ price_refresh_seconds * 1000 }}) + - + \ No newline at end of file diff --git a/tests/test_account_ledger_normalize.py b/tests/test_account_ledger_normalize.py new file mode 100644 index 0000000..9b8657a --- /dev/null +++ b/tests/test_account_ledger_normalize.py @@ -0,0 +1,101 @@ +"""账户流水归一化与分页查询.""" +from __future__ import annotations + +import sqlite3 +import unittest + +from lib.account_ledger.account_ledger_db import ( + ensure_account_ledger_tables, + query_entries, + upsert_entries, +) +from lib.account_ledger.account_ledger_normalize import ( + ACCOUNT_FUNDING, + ACCOUNT_TRADING, + PAGE_SIZE, + from_ccxt_ledger_entry, + kind_from_raw, + normalize_row, +) + + +class TestAccountLedgerNormalize(unittest.TestCase): + def test_kind_from_raw(self): + self.assertEqual(kind_from_raw("REALIZED_PNL", 1.0), "realized_pnl") + self.assertEqual(kind_from_raw("FUNDING_FEE", -0.1), "funding_fee") + self.assertEqual(kind_from_raw("TRANSFER", 5.0), "transfer_in") + self.assertEqual(kind_from_raw("TRANSFER", -5.0), "transfer_out") + + def test_normalize_row_and_ccxt(self): + row = normalize_row( + account=ACCOUNT_TRADING, + ccy="usdt", + amount="-1.25", + ts_ms=1_700_000_000_000, + ref_id="abc", + raw_type="COMMISSION", + ) + self.assertIsNotNone(row) + assert row is not None + self.assertEqual(row["ccy"], "USDT") + self.assertEqual(row["kind"], "commission") + self.assertEqual(row["account"], "trading") + + entry = { + "id": "9", + "currency": "USDC", + "amount": 2.5, + "timestamp": 1_700_000_000_000, + "type": "transfer", + "info": {"billId": "9", "bal": "10"}, + } + n = from_ccxt_ledger_entry(entry, account=ACCOUNT_FUNDING) + self.assertIsNotNone(n) + assert n is not None + self.assertEqual(n["ccy"], "USDC") + self.assertEqual(n["balance_after"], 10.0) + + def test_upsert_and_paginate(self): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + ensure_account_ledger_tables(conn) + rows = [] + for i in range(25): + rows.append( + normalize_row( + account=ACCOUNT_FUNDING, + ccy="USDT", + amount=1.0 + i, + ts_ms=1_700_000_000_000 + i * 1000, + ref_id=f"r{i}", + raw_type="deposit", + ) + ) + self.assertEqual(upsert_entries(conn, [r for r in rows if r]), 25) + page1 = query_entries( + conn, + account=ACCOUNT_FUNDING, + start_ms=0, + end_ms=9_000_000_000_000, + page=1, + page_size=PAGE_SIZE, + ) + self.assertEqual(page1["total"], 25) + self.assertEqual(len(page1["items"]), 10) + self.assertEqual(page1["pages"], 3) + # 倒序:最新在前 + self.assertGreater(page1["items"][0]["ts_ms"], page1["items"][1]["ts_ms"]) + page3 = query_entries( + conn, + account=ACCOUNT_FUNDING, + start_ms=0, + end_ms=9_000_000_000_000, + page=3, + page_size=PAGE_SIZE, + ) + self.assertEqual(len(page3["items"]), 5) + conn.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_instance_embed_lib.py b/tests/test_instance_embed_lib.py index 0af29a6..d08470d 100644 --- a/tests/test_instance_embed_lib.py +++ b/tests/test_instance_embed_lib.py @@ -32,10 +32,12 @@ def test_embed_tabs_cover_main_nav(): assert "risk_policy" in EMBED_TABS assert "system_guide" in EMBED_TABS assert "settings" in EMBED_TABS + assert "account_ledger" in EMBED_TABS assert path_to_embed_tab("/env_config") == "env_config" assert path_to_embed_tab("/risk_policy") == "risk_policy" assert path_to_embed_tab("/system_guide") == "system_guide" assert path_to_embed_tab("/settings") == "settings" + assert path_to_embed_tab("/account_ledger") == "account_ledger" def test_embed_context_extras_unified_ui_flags():