Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- def get_new_private_trades(self) -> List[Dict[str, Any]]:
- """
- Returns new USER trades since last call.
- On first call, returns [] (but sets internal cursor).
- On error, returns [].
- Format (price is execAmt/execQty):
- {
- 'price': float,
- 'quantity': float,
- 'side': 'buy' or 'sell',
- 'timestamp': int # seconds from matchTime
- }
- """
- page_num = 1
- page_size = 100
- max_pages = 20
- trades_raw: List[Dict[str, Any]] = []
- i = 0
- try:
- while True:
- params = {
- "symbol": self._symbol_code.lower(),
- "pageNum": page_num,
- "pageSize": page_size,
- }
- data = self._private_get("/trade/match/accountMatches", params=params)
- print('data part', i, len(data), data)
- i += 1
- page = data or []
- if not page:
- break
- trades_raw.extend(page)
- if len(page) < page_size:
- break
- if self._last_private_trade_id is not None:
- try:
- min_id_in_page = min(
- int(t.get("tradeId", 0)) for t in page if t.get("tradeId") is not None
- )
- except Exception:
- min_id_in_page = None
- if min_id_in_page is not None and min_id_in_page <= self._last_private_trade_id:
- break
- page_num += 1
- if page_num > max_pages:
- break
- except Exception as e:
- self.logger.error(
- f"Failed to fetch private trades for {self._symbol_code}: {e}"
- )
- return []
- if not trades_raw:
- return []
- self.logger.info('get_new_private_trades::trades_raw:', len(trades_raw))
- ids = [int(t.get("tradeId", 0)) for t in trades_raw if t.get("tradeId") is not None]
- if not ids:
- return []
- max_id = max(ids)
- # First call: set cursor, return no trades
- if self._last_private_trade_id is None:
- self._last_private_trade_id = max_id
- return []
- new_trades: List[Dict[str, Any]] = []
- for t in trades_raw:
- try:
- tid = int(t.get("tradeId"))
- except Exception:
- continue
- if tid <= self._last_private_trade_id:
- continue
- try:
- qty = float(t.get("execQty", 0))
- amt = float(t.get("execAmt", 0))
- price = amt / qty if qty > 0 else 0.0
- except Exception:
- qty = 0.0
- price = 0.0
- side_val = int(t.get("side", 0))
- side = "buy" if side_val == 1 else "sell"
- ts = int(t.get("matchTime", 0)) # seconds
- new_trades.append(
- {
- "price": price,
- "quantity": qty,
- "side": side,
- "timestamp": ts,
- }
- )
- if new_trades:
- self._last_private_trade_id = max(max_id, self._last_private_trade_id)
- return new_trades
Advertisement
Add Comment
Please, Sign In to add comment