Rebuild position history from Binance fills
Sync account fills into a new trading_fills table every 60s and reconstruct closed positions from them, so positions opened outside the program are included and external closes get an exit price. Positions link to plans by order id. The history drawer gains symbol, close-date, outcome and source filters, a summary bar and pagination.
This commit is contained in:
eezd committed
1 parent
a2e51a3011
commit
010d89e0f5
10 files changed
+589
-140
No files matched your search
@@ -23,6 +23,7 @@ from alphapulse.core.db import SessionLocal
|
||||
from alphapulse.market.binance import BinanceClient
|
||||
from alphapulse.models import TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol
|
||||
from alphapulse.trading.client import OWN_CLIENT_ID, DemoClient, ExchangeError, mark_price, symbol_rules
|
||||
from alphapulse.trading.fills import sync_fills
|
||||
from alphapulse.trading.plan import (
|
||||
MAX_HOLD_HOURS, TAKE_PROFIT_STEPS, OrderSpec, SymbolRules, condition_met, desired_orders, diff_orders, round_to_step,
|
||||
)
|
||||
@@ -31,6 +32,8 @@ from alphapulse.trading.schedule import request_analysis
|
||||
logger = logging.getLogger("alphapulse.trading.executor")
|
||||
|
||||
EXECUTOR_SECONDS = 2
|
||||
# 成交记录只用于历史仓位，不需要跟对账一样频繁
|
||||
FILLS_SYNC_SECONDS = 60
|
||||
ACTIVE_ORDER_STATUSES = ("submitting", "new", "partially_filled")
|
||||
LIVE_PLAN_STATUSES = ("pending", "open", "closing")
|
||||
# 刚下的单可能短时间查不到（C1 实测约 0.25s）
|
||||
@@ -78,6 +81,7 @@ class Executor:
|
||||
self.kline_cache: dict[tuple[str, str], tuple[int, Decimal]] = {}
|
||||
# 已提示过的无计划持仓：symbol -> 持仓均价，同一笔持仓只提示一次
|
||||
self.unmanaged_noticed: dict[str, Decimal] = {}
|
||||
self.fills_due = 0.0
|
||||
|
||||
# ---------- 基础 ----------
|
||||
|
||||
@@ -537,7 +541,7 @@ class Executor:
|
||||
account = await session.get(TradingAccount, 1)
|
||||
symbols = list((await session.scalars(select(TradingSymbol))).all())
|
||||
client = await self.ensure_client(account)
|
||||
if client is None or not symbols:
|
||||
if client is None:
|
||||
return
|
||||
positions = {}
|
||||
for p in await client.request("GET", "/fapi/v3/positionRisk"):
|
||||
@@ -547,6 +551,17 @@ class Executor:
|
||||
side="long" if amount > 0 else "short", qty=abs(amount), entry=Decimal(p["entryPrice"]),
|
||||
break_even=Decimal(p["breakEvenPrice"]), mark=Decimal(p["markPrice"]),
|
||||
)
|
||||
# 交易列表为空时也要同步：手动成交同样计入历史仓位。
|
||||
# 用独立会话：失败回滚会让会话里的对象全部过期，不能影响下面的对账
|
||||
if time.monotonic() >= self.fills_due:
|
||||
self.fills_due = time.monotonic() + FILLS_SYNC_SECONDS
|
||||
try:
|
||||
async with SessionLocal() as fills_session:
|
||||
fetched = await sync_fills(fills_session, client, {s.symbol for s in symbols} | set(positions))
|
||||
if fetched:
|
||||
logger.info("同步成交记录 %d 条", fetched)
|
||||
except Exception:
|
||||
logger.warning("同步成交记录失败，%ds 后重试", FILLS_SYNC_SECONDS, exc_info=True)
|
||||
for symbol in symbols:
|
||||
try:
|
||||
await self.reconcile(session, client, symbol, account.running, positions.get(symbol.symbol))
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
"""同步币安成交记录：历史仓位由它还原（见 positions.py），包括程序外的手动成交"""
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.dialects.postgresql import insert
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from alphapulse.models import TradingFill
|
||||
from alphapulse.trading.client import DemoClient
|
||||
|
||||
PAGE = 1000
|
||||
# 币安只保留近 3 个月的成交和资金流水
|
||||
HISTORY_DAYS = 90
|
||||
# 流水时间只精确到秒：往前多取一段，已有的成交按 trade id 去重
|
||||
OVERLAP = timedelta(minutes=1)
|
||||
|
||||
|
||||
def to_ms(dt: datetime) -> int:
|
||||
return int(dt.timestamp() * 1000)
|
||||
|
||||
|
||||
async def missing_starts(session: AsyncSession, client: DemoClient) -> dict[str, int]:
|
||||
"""用手续费流水（每笔成交一条，不限币种）发现新成交：返回每个币库里还没有的最小 trade id"""
|
||||
latest = await session.scalar(select(func.max(TradingFill.time)))
|
||||
start = to_ms(latest - OVERLAP if latest else datetime.now(UTC) - timedelta(days=HISTORY_DAYS))
|
||||
found: dict[str, set[int]] = {}
|
||||
while True:
|
||||
rows = await client.request("GET", "/fapi/v1/income", {"incomeType": "COMMISSION", "startTime": start, "limit": PAGE})
|
||||
for row in rows:
|
||||
found.setdefault(row["symbol"], set()).add(int(row["tradeId"]))
|
||||
if len(rows) < PAGE:
|
||||
break
|
||||
if rows[-1]["time"] == start:
|
||||
raise RuntimeError("同一秒内的手续费流水超过一页，无法按时间翻页")
|
||||
start = rows[-1]["time"]
|
||||
starts = {}
|
||||
for symbol, ids in found.items():
|
||||
known = set(await session.scalars(select(TradingFill.trade_id).where(TradingFill.symbol == symbol, TradingFill.trade_id.in_(ids))))
|
||||
if ids - known:
|
||||
starts[symbol] = min(ids - known)
|
||||
return starts
|
||||
|
||||
|
||||
async def sync_fills(session: AsyncSession, client: DemoClient, symbols: set[str]) -> int:
|
||||
"""把新成交写入 trading_fills，返回拉到的成交数。symbols 为交易列表和当前持仓的币：
|
||||
每轮按库里最大 trade id 往后查一次，兜底手续费为 0、没有流水的成交"""
|
||||
starts = await missing_starts(session, client)
|
||||
for symbol in symbols:
|
||||
last = await session.scalar(select(func.max(TradingFill.trade_id)).where(TradingFill.symbol == symbol))
|
||||
if last is not None:
|
||||
starts[symbol] = min(starts.get(symbol, last + 1), last + 1)
|
||||
fetched = 0
|
||||
for symbol, from_id in starts.items():
|
||||
while True:
|
||||
trades = await client.request("GET", "/fapi/v1/userTrades", {"symbol": symbol, "fromId": from_id, "limit": PAGE})
|
||||
if trades:
|
||||
await session.execute(insert(TradingFill).values([{
|
||||
"symbol": symbol, "trade_id": t["id"], "order_id": t["orderId"], "side": t["side"],
|
||||
"price": Decimal(t["price"]), "qty": Decimal(t["qty"]), "realized_pnl": Decimal(t["realizedPnl"]),
|
||||
"commission": Decimal(t["commission"]), "commission_asset": t["commissionAsset"],
|
||||
"time": datetime.fromtimestamp(t["time"] / 1000, UTC),
|
||||
} for t in trades]).on_conflict_do_nothing())
|
||||
fetched += len(trades)
|
||||
if len(trades) < PAGE:
|
||||
break
|
||||
from_id = trades[-1]["id"] + 1
|
||||
await session.commit()
|
||||
return fetched
|
||||
@@ -0,0 +1,85 @@
|
||||
"""由成交记录还原历史仓位。纯计算，不依赖数据库，便于测试"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from typing import Protocol
|
||||
|
||||
|
||||
class Fill(Protocol):
|
||||
"""一笔成交；models.TradingFill 满足此接口"""
|
||||
|
||||
symbol: str
|
||||
order_id: int
|
||||
side: str
|
||||
price: Decimal
|
||||
qty: Decimal
|
||||
realized_pnl: Decimal
|
||||
commission: Decimal
|
||||
commission_asset: str
|
||||
time: datetime
|
||||
|
||||
|
||||
@dataclass
|
||||
class ClosedPosition:
|
||||
symbol: str
|
||||
# long / short
|
||||
direction: str
|
||||
opened_at: datetime
|
||||
closed_at: datetime | None = None
|
||||
# 开仓（含加仓）成交量合计，及开仓、平仓方向的成交额
|
||||
quantity: Decimal = Decimal(0)
|
||||
open_value: Decimal = Decimal(0)
|
||||
close_qty: Decimal = Decimal(0)
|
||||
close_value: Decimal = Decimal(0)
|
||||
realized_pnl: Decimal = Decimal(0)
|
||||
# 只计 USDT 计价的手续费，与 Executor.realized 口径一致
|
||||
commission: Decimal = Decimal(0)
|
||||
# 参与这笔仓位的订单，用于关联程序的计划
|
||||
order_ids: set[int] = field(default_factory=set)
|
||||
# 带符号的持仓量：多为正、空为负
|
||||
position: Decimal = Decimal(0)
|
||||
|
||||
@property
|
||||
def entry_price(self) -> Decimal:
|
||||
return self.open_value / self.quantity
|
||||
|
||||
@property
|
||||
def exit_price(self) -> Decimal:
|
||||
return self.close_value / self.close_qty
|
||||
|
||||
@property
|
||||
def net(self) -> Decimal:
|
||||
return self.realized_pnl - self.commission
|
||||
|
||||
|
||||
def build_positions(fills: list[Fill]) -> list[ClosedPosition]:
|
||||
"""按币、按成交顺序累计带符号的持仓量（账户为单向持仓）：从 0 开始到回到 0 为一笔仓位。
|
||||
一笔成交跨过 0（反手）时拆成平仓和新开仓两段，realizedPnl 归平仓段，手续费按数量拆分。
|
||||
还没回到 0 的仓位不算历史。fills 须按 (symbol, time, trade_id) 排序"""
|
||||
closed: list[ClosedPosition] = []
|
||||
current: dict[str, ClosedPosition] = {}
|
||||
for fill in fills:
|
||||
fee = fill.commission if fill.commission_asset == "USDT" else Decimal(0)
|
||||
remaining = fill.qty if fill.side == "BUY" else -fill.qty
|
||||
while remaining:
|
||||
pos = current.get(fill.symbol)
|
||||
if pos is None:
|
||||
pos = current[fill.symbol] = ClosedPosition(fill.symbol, "long" if remaining > 0 else "short", fill.time)
|
||||
if pos.position == 0 or (pos.position > 0) == (remaining > 0):
|
||||
part = remaining
|
||||
pos.quantity += abs(part)
|
||||
pos.open_value += abs(part) * fill.price
|
||||
else:
|
||||
part = remaining if abs(remaining) <= abs(pos.position) else -pos.position
|
||||
pos.close_qty += abs(part)
|
||||
pos.close_value += abs(part) * fill.price
|
||||
pos.realized_pnl += fill.realized_pnl
|
||||
pos.position += part
|
||||
pos.commission += fee * abs(part) / fill.qty
|
||||
pos.order_ids.add(fill.order_id)
|
||||
remaining -= part
|
||||
if pos.position == 0:
|
||||
pos.closed_at = fill.time
|
||||
closed.append(current.pop(fill.symbol))
|
||||
return closed
|
||||
Reference in new issue
Block a user