Coverage for adrs/oms/position.py: 99%
74 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1import logging
2import asyncio
3import time
5from decimal import Decimal
6from datetime import datetime, timezone
8from cybotrade import Symbol
9from cybotrade.models import Position, OrderSide
11from adrs.oms.config import ConfigManager
12from adrs.oms.ops.order_pool import OpenOrdersSnapshot
13from adrs.oms.rate_limit.exchange_limit_profiles import Endpoints
14from adrs.oms.rate_limit.rate_limiter import RateLimiter
16logger = logging.getLogger(__name__)
18Positions = dict[Symbol, Position] # symbol -> Position
20# How stale an exchange-position read may be before update_exchange() goes back
21# to the REST API. Positions are only ever as fresh as the last poll anyway, so
22# this changes nothing for the placement tick (which runs far slower than this),
23# and collapses a burst of websocket-driven refreshes into a single request.
24POSITION_REFRESH_TTL_SEC = 1.0
26# How stale the REST anchor may be before the order-sizing path stops trusting
27# the streamed position and forces a read.
28#
29# ACCOUNT_UPDATE is event-driven, so silence is ambiguous -- a quiet account and a
30# dead socket look identical -- and unlike the price feed there is no busy stream
31# on the user-data socket to use as a heartbeat: its ping is every 3 minutes and
32# the listenKey keepalive every 30. So the 60s REST reconcile is the liveness
33# mechanism, and this is the window it is trusted for. Must exceed that 60s
34# cadence, or the sizing path forces a read on most ticks and the saving is lost.
35POSITION_ANCHOR_MAX_AGE_SEC = 90.0
38class PositionManager:
39 def __init__(self, config: ConfigManager, rate_limiter: RateLimiter):
40 # The OrderPlacementManager will update these states
41 self.exchange: Positions = {}
42 self.pending: Positions = {}
44 self.desired: Positions = {}
45 self.config = config
46 self.rate_limiter = rate_limiter
47 self.delta_lock = asyncio.Lock()
48 # Serialises exchange-position refreshes so concurrent callers share one
49 # REST call instead of each spending GET_POSITION weight
50 self._refresh_lock = asyncio.Lock()
51 self._exchange_refreshed_at: float | None = None # monotonic, None = never
53 async def delta_calculation(
54 self, snapshot: OpenOrdersSnapshot
55 ) -> dict[Symbol, Decimal]:
56 """
57 Pending and Exchange values would be updated from REST
58 Would fall back to WS local copy if rate limits has been reached
60 Case 1:
61 desire: +3
62 exchange: -1
63 pending: +2
64 delta = 3 - (-1 + 2) = 2
66 Case 2:
67 desire: -3
68 exchange: +1
69 pending: -2
70 delta = -3 - (1 - 2) = -2
71 """
72 # Prevents conditions where orders being replaced and calculation runs in the middle of it
73 async with self.delta_lock:
74 self.update_pending(snapshot)
75 # Trust the streamed position while the REST anchor is fresh. The
76 # anchor (set only by a real read, refreshed every 60s by
77 # on_aegis_update) is the liveness proof; once it ages out this
78 # forces a read exactly as it always did.
79 await self.update_exchange(max_age_sec=POSITION_ANCHOR_MAX_AGE_SEC)
80 deltas = {}
81 for symbol, position in self.desired.items():
82 exchange_pos = self.exchange[symbol]
83 pending_pos = self.pending[symbol]
84 delta = position.quantity - (
85 exchange_pos.quantity + pending_pos.quantity
86 )
87 deltas[symbol] = delta
89 return deltas
91 def compute_base_quantity(self, price: Decimal, weightage: Decimal) -> Decimal:
92 """
93 The quantity being return can be -ve/+ve depending on the side.
94 If the position is sell, it would be negative and positive otherwise.
95 """
96 initial_balance, leverage = (
97 self.config.config.initial_balance,
98 self.config.config.leverage,
99 )
100 return (initial_balance * leverage * weightage) / price
102 def _positions_fresh_within(self, max_age_sec: float) -> bool:
103 if self._exchange_refreshed_at is None or max_age_sec <= 0:
104 return False
105 return (time.monotonic() - self._exchange_refreshed_at) < max_age_sec
107 def invalidate_exchange_anchor(self) -> None:
108 """
109 Forget the REST anchor, forcing the next delta_calculation to read.
111 Call this on a websocket (re)connect. cybotrade emits Authenticated
112 on every connect, including reconnects, and everything that happened
113 during the gap -- ORDER_TRADE_UPDATE as well as ACCOUNT_UPDATE -- is
114 lost with no other signal that it happened. The anchor being fresh
115 would otherwise make delta_calculation keep trusting a position that
116 predates the gap for up to POSITION_ANCHOR_MAX_AGE_SEC.
117 """
118 self._exchange_refreshed_at = None
120 async def update_exchange(self, max_age_sec: float = POSITION_REFRESH_TTL_SEC):
121 """
122 Get the latest positions available from the exchange.
124 Skips the call when the last successful read is younger than
125 `max_age_sec`, and makes concurrent callers share one request rather than
126 each spending GET_POSITION weight. A burst of websocket order updates
127 used to fire one /fapi/v2/positionRisk per event — on Binance that is
128 weight 5 apiece, against a budget metered per IP and shared with every
129 co-tenant on the shard. Pass max_age_sec=0 to force a read.
131 A failed read leaves the timestamp alone, so the next caller retries
132 rather than trusting positions that were never fetched.
133 """
134 if self._positions_fresh_within(max_age_sec):
135 return
136 async with self._refresh_lock:
137 # Re-check: whoever held the lock may have just refreshed, in which
138 # case every caller queued behind them can use that result.
139 if self._positions_fresh_within(max_age_sec):
140 return
141 # Captured before rate_limiter.reserve(), not just before
142 # get_positions(): reserve.__aenter__ is itself an await and can
143 # block for minutes on a rate-limit cooldown, so most of the
144 # suspension window would otherwise go unguarded.
145 before = dict(self.exchange) # cheap: a handful of symbols
146 endpoint = Endpoints.GET_POSITION
147 try:
148 async with self.rate_limiter.reserve(endpoint=endpoint):
149 exchange_positions = await self.config.exchange.get_positions()
150 for position in exchange_positions:
151 sym = position.symbol
152 # This await is a real suspension point on the shared
153 # loop: an ACCOUNT_UPDATE frame can be applied to
154 # self.exchange while we were awaiting REST. That
155 # stream value is absolute and newer, so don't clobber
156 # it with the now-stale REST snapshot. Relies on every
157 # writer REBINDING the Position object rather than
158 # mutating it in place -- see apply_stream_positions --
159 # so identity comparison alone detects a fresher write.
160 if self.exchange.get(sym) is not before.get(sym):
161 continue
162 self.exchange[sym] = position
163 # The read did happen, so liveness is genuinely proved here,
164 # and any symbol we kept instead of overwriting is the fresher
165 # of the two values anyway.
166 self._exchange_refreshed_at = time.monotonic()
167 except Exception as e:
168 logger.warning(f"Failed to update exchange due to {e}")
170 def apply_stream_positions(self, positions: list[Position]) -> None:
171 """
172 Overwrite the exchange position for each symbol in a stream frame.
174 Absolute, never accumulated: this replaces the incremental fill
175 accounting, which drifted by construction. Applying the same frame twice
176 is a no-op, which is the property that makes the replacement worth making.
178 Frames are partial -- they carry what changed -- so symbols absent from
179 `positions` keep their previous value rather than being zeroed. A position
180 that has closed arrives as quantity 0 and is applied, not skipped: skipping
181 it would leave the OMS sizing orders against a position it no longer holds.
183 Deliberately does NOT stamp the REST anchor. Only a real REST read may do
184 that, because the anchor is what proves liveness -- if the stream stamped
185 it, a dead REST path would look healthy forever.
186 """
187 for position in positions:
188 self.exchange[position.symbol] = position
190 def update_pending(self, snapshot: OpenOrdersSnapshot):
191 """
192 Derive pending positions from an open-orders snapshot.
194 Symbols absent from the snapshot had a failed fetch; their previous
195 pending state is kept rather than zeroed.
196 """
197 for symbol, pending_orders in snapshot.orders.items():
198 quantity = Decimal("0")
199 for order in pending_orders:
200 quantity += (
201 order.remain_size
202 if order.side == OrderSide.BUY
203 else -order.remain_size
204 )
205 entry_price = (
206 Decimal("0") if len(pending_orders) == 0 else pending_orders[-1].price
207 )
208 self.pending[symbol] = Position(
209 symbol=symbol,
210 quantity=quantity,
211 entry_price=entry_price,
212 updated_time=datetime.now(timezone.utc),
213 )