From 0d9447c9d98cd85ea2112871f74eef6f47ddd739 Mon Sep 17 00:00:00 2001 From: cardosofede Date: Fri, 11 Sep 2026 17:04:52 +0300 Subject: [PATCH 1/2] fix(trading): page order history by a real keyset cursor, not the same "0:" forever POST /trading/orders/search built each order's cursor from `timestamp` and `client_order_id`, but order rows carry `created_at` and `order_id`, so every row got the cursor "0:" and sorted as a tie. The page after "0:" was the list minus its first row -- most of the previous page again -- and it handed "0:" back, so a client that followed next_cursor looped forever over the same rows. An agent paging its order history with limit=3 read records 2-3 again on every call and never reached the end. Behind that sat a second cap. Each account was read as its newest `limit * 2` rows at offset 0, then merged and sliced in memory, so even a correct cursor could never reach an order older than that window, while has_more said the history had ended. Multi-value connector_names/trading_pairs were filtered in memory after the cap, shrinking pages further, and total_count counted the fetched window rather than the matching orders. Pagination now happens in the database, keyset on (created_at, client_order_id): - OrderRepository.get_orders takes the account/connector/pair filters as lists and a `before` cursor, orders by created_at desc then client_order_id desc, and cuts strictly after the cursor, so orders sharing a created_at are neither skipped nor served twice. count_orders applies the same filters for a real total_count. - The route runs one query across every account with limit + 1 rows; the extra row is how has_more is known. next_cursor is "|", split back on the first "|", which an ISO timestamp never contains. - A cursor the route did not hand out (including the old "0:") is refused with a 400. An unknown cursor used to restart at page one, which a client walking the history cannot tell from genuinely older orders. TradingHistoryService.get_orders becomes search_orders, returning the page and the total from one session; the route was its only caller. routers/trading.py also clears the flake8 findings it already carried on main -- unused imports, inline `import logging` in except blocks (the module logger was already there), untyped `connector_service = Depends(...)` parameters, over-long lines -- because the pre-commit flake8 hook lints the whole file and refuses any commit that touches it. Formatting only; no behaviour change outside the order search route. test/test_order_history_cursor.py drives the real route, service and repository against a real SQL table and walks the cursor as a client does. Against the old route every walk looped on "0:"; now each order comes back exactly once, newest first, at every page size. --- database/repositories/order_repository.py | 82 +++++++--- routers/trading.py | 158 ++++++++++-------- services/trading_history_service.py | 47 ++++-- test/test_order_history_cursor.py | 190 ++++++++++++++++++++++ 4 files changed, 369 insertions(+), 108 deletions(-) create mode 100644 test/test_order_history_cursor.py diff --git a/database/repositories/order_repository.py b/database/repositories/order_repository.py index 49b38afa..11fa67b9 100644 --- a/database/repositories/order_repository.py +++ b/database/repositories/order_repository.py @@ -1,9 +1,9 @@ import logging from datetime import datetime from decimal import Decimal -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Tuple -from sqlalchemy import func, select +from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from database.models import Order @@ -104,23 +104,23 @@ async def update_order_fill(self, client_order_id: str, filled_amount: Decimal, await self.session.flush() return order - async def get_orders(self, account_name: Optional[str] = None, - connector_name: Optional[str] = None, - trading_pair: Optional[str] = None, - status: Optional[str] = None, - start_time: Optional[int] = None, - end_time: Optional[int] = None, - limit: int = 100, offset: int = 0) -> List[Order]: - """Get orders with filtering and pagination.""" - query = select(Order) + @staticmethod + def _filter_orders(query, account_names: Optional[List[str]] = None, + connector_names: Optional[List[str]] = None, + trading_pairs: Optional[List[str]] = None, + status: Optional[str] = None, + start_time: Optional[int] = None, + end_time: Optional[int] = None): + """Apply the order-history filters shared by `get_orders` and `count_orders`. - # Apply filters - if account_name: - query = query.where(Order.account_name == account_name) - if connector_name: - query = query.where(Order.connector_name == connector_name) - if trading_pair: - query = query.where(Order.trading_pair == trading_pair) + `account_names=None` means every account; an empty list matches none. + """ + if account_names is not None: + query = query.where(Order.account_name.in_(account_names)) + if connector_names: + query = query.where(Order.connector_name.in_(connector_names)) + if trading_pairs: + query = query.where(Order.trading_pair.in_(trading_pairs)) if status: query = query.where(Order.status == status) if start_time: @@ -129,14 +129,54 @@ async def get_orders(self, account_name: Optional[str] = None, if end_time: end_dt = datetime.fromtimestamp(end_time / 1000) query = query.where(Order.created_at <= end_dt) + return query - # Apply ordering and pagination - query = query.order_by(Order.created_at.desc()) - query = query.limit(limit).offset(offset) + async def get_orders(self, account_names: Optional[List[str]] = None, + connector_names: Optional[List[str]] = None, + trading_pairs: Optional[List[str]] = None, + status: Optional[str] = None, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + limit: int = 100, + before: Optional[Tuple[datetime, str]] = None) -> List[Order]: + """Get orders newest first, keyset-paginated on (created_at, client_order_id). + + `before` is the (created_at, client_order_id) of the last order already served; + only orders strictly older than it come back. created_at alone is not unique -- + orders placed in the same microsecond share it -- so client_order_id breaks the + tie in both the ordering and the cut, and no order is skipped or served twice. + """ + query = self._filter_orders( + select(Order), account_names, connector_names, trading_pairs, status, start_time, end_time + ) + if before is not None: + created_at, client_order_id = before + query = query.where( + or_( + Order.created_at < created_at, + and_(Order.created_at == created_at, Order.client_order_id < client_order_id), + ) + ) + + query = query.order_by(Order.created_at.desc(), Order.client_order_id.desc()).limit(limit) result = await self.session.execute(query) return result.scalars().all() + async def count_orders(self, account_names: Optional[List[str]] = None, + connector_names: Optional[List[str]] = None, + trading_pairs: Optional[List[str]] = None, + status: Optional[str] = None, + start_time: Optional[int] = None, + end_time: Optional[int] = None) -> int: + """Count the orders `get_orders` would page through with the same filters.""" + query = self._filter_orders( + select(func.count()).select_from(Order), + account_names, connector_names, trading_pairs, status, start_time, end_time, + ) + result = await self.session.execute(query) + return result.scalar_one() + async def get_active_orders(self, account_name: Optional[str] = None, connector_name: Optional[str] = None, trading_pair: Optional[str] = None, diff --git a/routers/trading.py b/routers/trading.py index 4b6df289..57c16887 100644 --- a/routers/trading.py +++ b/routers/trading.py @@ -1,10 +1,10 @@ import logging import math -from typing import Dict, List, Optional +from datetime import datetime +from typing import Dict, Tuple from fastapi import APIRouter, Depends, HTTPException from hummingbot.core.data_type.common import OrderType, PositionAction, PositionMode, TradeType -from pydantic import BaseModel from starlette import status from deps import get_accounts_service, get_connector_service, get_trading_history_service @@ -22,6 +22,7 @@ from models.pagination import paginate_by_cursor from services.accounts_service import AccountsService from services.trading_history_service import TradingHistoryService +from services.unified_connector_service import UnifiedConnectorService # Create module-specific logger logger = logging.getLogger(__name__) @@ -120,7 +121,7 @@ async def cancel_order( async def get_positions( filter_request: PositionFilterRequest, accounts_service: AccountsService = Depends(get_accounts_service), - connector_service = Depends(get_connector_service) + connector_service: UnifiedConnectorService = Depends(get_connector_service) ): """ Get current positions across all or filtered perpetual connectors. @@ -164,8 +165,6 @@ async def get_positions( all_positions.extend(positions) except Exception as e: # Log error but continue with other connectors - import logging - logger.warning(f"Failed to get positions for {account_name}/{connector_name}: {e}") # Sort by cursor_id and apply cursor-based pagination @@ -179,7 +178,7 @@ async def get_positions( @router.post("/orders/active", response_model=PaginatedResponse) async def get_active_orders( filter_request: ActiveOrderFilterRequest, - connector_service = Depends(get_connector_service) + connector_service: UnifiedConnectorService = Depends(get_connector_service) ): """ Get active (in-flight) orders across all or filtered accounts and connectors. @@ -231,8 +230,6 @@ async def get_active_orders( except Exception as e: # Log error but continue with other connectors - import logging - logger.warning(f"Failed to get active orders for {account_name}/{connector_name}: {e}") # Sort by cursor_id and apply cursor-based pagination @@ -247,75 +244,86 @@ async def get_active_orders( async def get_orders( filter_request: OrderFilterRequest, trading_history_service: TradingHistoryService = Depends(get_trading_history_service), - connector_service = Depends(get_connector_service) + connector_service: UnifiedConnectorService = Depends(get_connector_service) ): """ Get historical order data across all or filtered accounts from the database/registry. + Orders come newest first. `pagination.next_cursor` is the cursor of the page's last + order; pass it back as `cursor` to get the orders older than it. It is `null` on the + last page. A cursor this route did not hand out is refused with a 400. + Args: filter_request: JSON payload with filtering criteria Returns: Paginated response with historical order data and pagination metadata """ - try: - all_orders = [] + before = _parse_order_cursor(filter_request.cursor) if filter_request.cursor else None - # Determine which accounts to query + try: if filter_request.account_names: accounts_to_check = filter_request.account_names else: - # Get all accounts - all_connectors = connector_service.get_all_trading_connectors() - accounts_to_check = list(all_connectors.keys()) + accounts_to_check = list(connector_service.get_all_trading_connectors().keys()) + + # One query over every account, cut at the cursor inside the database, so the page + # is exactly the `limit` newest orders older than the cursor. The extra row fetched + # past `limit` is how has_more is known without a second query. + result = await trading_history_service.search_orders( + account_names=accounts_to_check, + connector_names=filter_request.connector_names, + trading_pairs=filter_request.trading_pairs, + status=filter_request.status, + start_time=filter_request.start_time, + end_time=filter_request.end_time, + limit=filter_request.limit + 1, + before=before, + ) + orders = result["orders"] + page = orders[: filter_request.limit] + has_more = len(orders) > filter_request.limit + + return PaginatedResponse( + data=page, + pagination={ + "limit": filter_request.limit, + "has_more": has_more, + "next_cursor": _order_cursor(page[-1]) if has_more else None, + "total_count": result["total_count"], + }, + ) + except Exception as e: + raise HTTPException(status_code=500, detail=f"Error fetching orders: {str(e)}") - # Collect orders from all specified accounts - for account_name in accounts_to_check: - try: - orders = await trading_history_service.get_orders( - account_name=account_name, - connector_name=( - filter_request.connector_names[0] - if filter_request.connector_names and len(filter_request.connector_names) == 1 - else None - ), - trading_pair=( - filter_request.trading_pairs[0] - if filter_request.trading_pairs and len(filter_request.trading_pairs) == 1 - else None - ), - status=filter_request.status, - start_time=filter_request.start_time, - end_time=filter_request.end_time, - limit=filter_request.limit * 2, # Get more for filtering - offset=0, - ) - # Add cursor-friendly identifier to each order - for order in orders: - order["_cursor_id"] = f"{order.get('timestamp', 0)}:{order.get('client_order_id', '')}" - all_orders.extend(orders) - except Exception as e: - # Log error but continue with other accounts - import logging - logger.warning(f"Failed to get orders for {account_name}: {e}") +# An order's cursor is "|". An ISO timestamp +# never contains "|", so the first "|" splits it back whatever the order id holds. +_ORDER_CURSOR_SEPARATOR = "|" - # Apply filters for multiple values - if filter_request.connector_names and len(filter_request.connector_names) > 1: - all_orders = [order for order in all_orders if order.get("connector_name") in filter_request.connector_names] - if filter_request.trading_pairs and len(filter_request.trading_pairs) > 1: - all_orders = [order for order in all_orders if order.get("trading_pair") in filter_request.trading_pairs] - # Sort by timestamp (most recent first) then cursor_id, and apply cursor-based pagination - return paginate_by_cursor( - all_orders, - filter_request.cursor, - filter_request.limit, - sort_key=lambda x: (x.get("timestamp", 0), x.get("_cursor_id", "")), - reverse=True, +def _order_cursor(order: Dict) -> str: + """The keyset cursor after `order`, built from the fields an order row really carries.""" + return f"{order['created_at']}{_ORDER_CURSOR_SEPARATOR}{order['order_id']}" + + +def _parse_order_cursor(cursor: str) -> Tuple[datetime, str]: + """Split a cursor from `_order_cursor` back into (created_at, client_order_id). + + Anything else is refused rather than read as "start over": an unrecognised cursor + used to serve page one again, which a client walking the history cannot tell from + genuinely older orders. + """ + created_at, _, client_order_id = cursor.partition(_ORDER_CURSOR_SEPARATOR) + try: + if not client_order_id: + raise ValueError("no client order id") + return datetime.fromisoformat(created_at), client_order_id + except ValueError: + raise HTTPException( + status_code=400, + detail=f"Invalid cursor {cursor!r}: pass back the next_cursor of a previous /trading/orders/search page", ) - except Exception as e: - raise HTTPException(status_code=500, detail=f"Error fetching orders: {str(e)}") # Trade History @@ -323,7 +331,7 @@ async def get_orders( async def get_trades( filter_request: TradeFilterRequest, trading_history_service: TradingHistoryService = Depends(get_trading_history_service), - connector_service = Depends(get_connector_service) + connector_service: UnifiedConnectorService = Depends(get_connector_service) ): """ Get trade history across all or filtered accounts with complex filtering. @@ -376,8 +384,6 @@ async def get_trades( all_trades.extend(trades) except Exception as e: # Log error but continue with other accounts - import logging - logger.warning(f"Failed to get trades for {account_name}: {e}") # Apply filters for multiple values @@ -482,7 +488,8 @@ async def set_leverage( Dictionary with success status and message Raises: - HTTPException: 400 for invalid parameters or non-perpetual connector, 404 for account/connector not found, 500 for execution errors + HTTPException: 400 for invalid parameters or non-perpetual connector, 404 for account/connector + not found, 500 for execution errors """ try: result = await accounts_service.set_leverage( @@ -499,7 +506,7 @@ async def set_leverage( async def get_funding_payments( filter_request: FundingPaymentFilterRequest, trading_history_service: TradingHistoryService = Depends(get_trading_history_service), - connector_service = Depends(get_connector_service) + connector_service: UnifiedConnectorService = Depends(get_connector_service) ): """ Get funding payment history across all or filtered perpetual connectors. @@ -545,13 +552,12 @@ async def get_funding_payments( # Add cursor-friendly identifier to each payment for payment in payments: payment["_cursor_id"] = ( - f"{account_name}:{connector_name}:{payment.get('timestamp', '')}:{payment.get('trading_pair', '')}" + f"{account_name}:{connector_name}:" + f"{payment.get('timestamp', '')}:{payment.get('trading_pair', '')}" ) all_funding_payments.extend(payments) except Exception as e: # Log error but continue with other connectors - import logging - logger.warning(f"Failed to get funding payments for {account_name}/{connector_name}: {e}") # Sort by timestamp (most recent first) then cursor_id, and apply cursor-based pagination @@ -617,9 +623,23 @@ def _standardize_in_flight_order_response(order, account_name: str, connector_na "amount": float(order.amount) if order.amount and not math.isnan(float(order.amount)) else 0, "price": float(order.price) if order.price and not math.isnan(float(order.price)) else None, "status": status, - "filled_amount": float(getattr(order, "executed_amount_base", 0) or 0) if not math.isnan(float(getattr(order, "executed_amount_base", 0) or 0)) else 0, - "average_fill_price": float(getattr(order, "last_executed_price", 0)) if getattr(order, "last_executed_price", None) and not math.isnan(float(getattr(order, "last_executed_price", 0))) else None, - "fee_paid": float(getattr(order, "cumulative_fee_paid_quote", 0)) if getattr(order, "cumulative_fee_paid_quote", None) and not math.isnan(float(getattr(order, "cumulative_fee_paid_quote", 0))) else None, + "filled_amount": ( + float(getattr(order, "executed_amount_base", 0) or 0) + if not math.isnan(float(getattr(order, "executed_amount_base", 0) or 0)) + else 0 + ), + "average_fill_price": ( + float(getattr(order, "last_executed_price", 0)) + if getattr(order, "last_executed_price", None) + and not math.isnan(float(getattr(order, "last_executed_price", 0))) + else None + ), + "fee_paid": ( + float(getattr(order, "cumulative_fee_paid_quote", 0)) + if getattr(order, "cumulative_fee_paid_quote", None) + and not math.isnan(float(getattr(order, "cumulative_fee_paid_quote", 0))) + else None + ), "fee_currency": None, # InFlightOrder doesn't store fee currency directly "created_at": created_at, "updated_at": updated_at, diff --git a/services/trading_history_service.py b/services/trading_history_service.py index 96f07d09..dbcf056c 100644 --- a/services/trading_history_service.py +++ b/services/trading_history_service.py @@ -8,7 +8,8 @@ helper (``_run_in_repo``). """ import logging -from typing import Dict, List, Optional +from datetime import datetime +from typing import Dict, List, Optional, Tuple from database import AsyncDatabaseManager, FundingRepository, OrderRepository, TradeRepository @@ -53,25 +54,35 @@ async def _run_in_repo(self, repo_cls, fn, default, error_message): logger.error(f"{error_message}: {e}") return default(e) if callable(default) else default - async def get_orders(self, account_name: Optional[str] = None, connector_name: Optional[str] = None, - trading_pair: Optional[str] = None, status: Optional[str] = None, - start_time: Optional[int] = None, end_time: Optional[int] = None, - limit: int = 100, offset: int = 0) -> List[Dict]: - """Get order history using OrderRepository.""" + async def search_orders(self, account_names: Optional[List[str]] = None, + connector_names: Optional[List[str]] = None, + trading_pairs: Optional[List[str]] = None, status: Optional[str] = None, + start_time: Optional[int] = None, end_time: Optional[int] = None, + limit: int = 100, before: Optional[Tuple[datetime, str]] = None) -> Dict: + """One page of order history plus the count of every order matching the filters. + + Returns ``{"orders": [...], "total_count": n}``. See ``OrderRepository.get_orders`` + for the ``before`` keyset cursor. + """ + filters = dict( + account_names=account_names, + connector_names=connector_names, + trading_pairs=trading_pairs, + status=status, + start_time=start_time, + end_time=end_time, + ) + async def _fn(order_repo): - orders = await order_repo.get_orders( - account_name=account_name, - connector_name=connector_name, - trading_pair=trading_pair, - status=status, - start_time=start_time, - end_time=end_time, - limit=limit, - offset=offset - ) - return [order_repo.to_dict(order) for order in orders] + orders = await order_repo.get_orders(**filters, limit=limit, before=before) + return { + "orders": [order_repo.to_dict(order) for order in orders], + "total_count": await order_repo.count_orders(**filters), + } - return await self._run_in_repo(OrderRepository, _fn, [], "Error getting orders") + return await self._run_in_repo( + OrderRepository, _fn, {"orders": [], "total_count": 0}, "Error getting orders" + ) async def get_active_orders_history(self, account_name: Optional[str] = None, connector_name: Optional[str] = None, trading_pair: Optional[str] = None) -> List[Dict]: diff --git a/test/test_order_history_cursor.py b/test/test_order_history_cursor.py new file mode 100644 index 00000000..cb48256e --- /dev/null +++ b/test/test_order_history_cursor.py @@ -0,0 +1,190 @@ +"""POST /trading/orders/search must page through order history, not round one page. + +The route built each order's cursor from `timestamp` and `client_order_id`, but order +rows carry `created_at` and `order_id`, so every row got the cursor "0:" and sorted as +a tie. The page after "0:" was the list minus its first row -- most of the previous +page again -- and it handed "0:" back, so a client that followed the cursor looped +forever over the same rows. Behind that sat a second cap: each account was read as its +newest `limit * 2` rows at offset 0, so even a correct cursor could never reach an +order older than that window, while `has_more` said the history had ended. + +These tests drive the real route, service and repository against a real SQL table +(sync SQLite behind an async shim: the environment has no aiosqlite) and walk the +cursor the way a client does, asserting on every order id that comes back. + +Run with: pytest test/test_order_history_cursor.py -v +""" + +from contextlib import asynccontextmanager +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import Session +from sqlalchemy.pool import StaticPool + +from database.models import Order +from deps import get_connector_service, get_trading_history_service +from routers import trading +from services.trading_history_service import TradingHistoryService + +BASE = datetime(2026, 9, 11, 10, 0, 0, 123456, tzinfo=timezone.utc) +MAX_PAGES = 50 + + +class _AsyncSessionShim: + """The two calls the order repository makes, over a sync SQLAlchemy session.""" + + def __init__(self, session): + self._session = session + + async def execute(self, query): + return self._session.execute(query) + + +class _DbManager: + def __init__(self, engine): + self._engine = engine + + @asynccontextmanager + async def get_session_context(self): + with Session(self._engine) as session: + yield _AsyncSessionShim(session) + + +def _order(client_order_id, created_at, account="master_account", connector="binance", pair="SOL-USDT"): + return Order( + client_order_id=client_order_id, + account_name=account, + connector_name=connector, + trading_pair=pair, + trade_type="BUY", + order_type="LIMIT", + amount=1, + price=200, + status="FILLED", + filled_amount=1, + created_at=created_at, + updated_at=created_at, + ) + + +# Eleven orders over two accounts and three connectors. Several share a created_at to +# the microsecond, which only the client_order_id tie-breaker can order. +ORDERS = [ + _order("a-01", BASE), + _order("a-02", BASE + timedelta(seconds=1)), + _order("a-03", BASE + timedelta(seconds=1)), + _order("a-04", BASE + timedelta(seconds=1)), + _order("a-05", BASE + timedelta(seconds=2), connector="okx"), + _order("a-06", BASE + timedelta(seconds=3)), + _order("a-07", BASE + timedelta(seconds=4), connector="okx"), + _order("a-08", BASE + timedelta(seconds=5), connector="kraken"), + _order("b-01", BASE + timedelta(seconds=1), account="second_account"), + _order("b-02", BASE + timedelta(seconds=3, microseconds=1), account="second_account"), + _order("b-03", BASE + timedelta(seconds=6), account="second_account", connector="okx"), +] + + +def _newest_first(orders): + return [o.client_order_id for o in sorted(orders, key=lambda o: (o.created_at, o.client_order_id), reverse=True)] + + +@pytest.fixture +def client(): + # One connection for the whole test: each new connection to ":memory:" is a new, + # empty database, and TestClient serves the app from another thread. + engine = create_engine("sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False}) + Order.__table__.create(engine) + with Session(engine) as session: + # Detached copies, so every test inserts the same rows into its own database. + session.add_all( + [_order(o.client_order_id, o.created_at, o.account_name, o.connector_name, o.trading_pair) for o in ORDERS] + ) + session.commit() + + app = FastAPI() + app.include_router(trading.router) + app.dependency_overrides[get_trading_history_service] = lambda: TradingHistoryService(_DbManager(engine)) + app.dependency_overrides[get_connector_service] = lambda: SimpleNamespace( + get_all_trading_connectors=lambda: {"master_account": {}, "second_account": {}} + ) + return TestClient(app) + + +def _walk(client, limit, **filters): + """Follow next_cursor to the end, the way a client does; fail rather than loop.""" + order_ids, envelopes, cursor = [], [], None + for _ in range(MAX_PAGES): + body = {"limit": limit, **filters} + if cursor: + body["cursor"] = cursor + response = client.post("/trading/orders/search", json=body) + assert response.status_code == 200, response.text + envelope = response.json() + envelopes.append(envelope) + order_ids.extend(order["order_id"] for order in envelope["data"]) + cursor = envelope["pagination"]["next_cursor"] + if not cursor: + return order_ids, envelopes + pytest.fail(f"pagination did not end within {MAX_PAGES} pages; last cursor {cursor!r}: {order_ids}") + + +@pytest.mark.parametrize("limit", [1, 2, 3, 4, 10, 11, 100]) +def test_walking_the_cursor_returns_every_order_once_newest_first(client, limit): + order_ids, envelopes = _walk(client, limit) + + assert order_ids == _newest_first(ORDERS) + assert all(len(envelope["data"]) <= limit for envelope in envelopes) + # has_more is only ever true on a page that hands out a cursor. + assert [envelope["pagination"]["has_more"] for envelope in envelopes] == [True] * (len(envelopes) - 1) + [False] + + +def test_the_walk_reaches_orders_older_than_twice_the_page_size(client): + """Each account used to be read as its newest limit * 2 rows, then cut off.""" + order_ids, _ = _walk(client, limit=2, account_names=["master_account"]) + + assert order_ids == _newest_first([o for o in ORDERS if o.account_name == "master_account"]) + assert order_ids[-1] == "a-01" + + +def test_a_multi_value_filter_pages_like_an_unfiltered_one(client): + """Two connector_names used to be filtered in memory after the rows were capped.""" + wanted = [o for o in ORDERS if o.connector_name in ("okx", "kraken")] + + order_ids, _ = _walk(client, limit=1, connector_names=["okx", "kraken"]) + + assert order_ids == _newest_first(wanted) + + +def test_total_count_is_every_matching_order_on_every_page(client): + _, envelopes = _walk(client, limit=3, account_names=["second_account"]) + + assert [envelope["pagination"]["total_count"] for envelope in envelopes] == [3] + + +def test_every_page_carries_the_same_total(client): + _, envelopes = _walk(client, limit=4) + + assert {envelope["pagination"]["total_count"] for envelope in envelopes} == {len(ORDERS)} + + +def test_the_cursor_is_not_a_placeholder(client): + """The old route handed out "0:" for every order.""" + response = client.post("/trading/orders/search", json={"limit": 3}) + + cursor = response.json()["pagination"]["next_cursor"] + assert cursor and cursor != "0:" + assert cursor.endswith(response.json()["data"][-1]["order_id"]) + + +@pytest.mark.parametrize("cursor", ["0:", "garbage", "2026-09-11T10:00:00|", "not-a-time|a-01"]) +def test_an_unreadable_cursor_is_refused_not_answered_with_page_one(client, cursor): + """An unknown cursor used to restart the walk at page one, looking like fresh rows.""" + response = client.post("/trading/orders/search", json={"limit": 3, "cursor": cursor}) + + assert response.status_code == 400, response.text + assert "cursor" in response.json()["detail"] From 4dab50a607a3abc23d2ff19c40a0dbcdd4357234 Mon Sep 17 00:00:00 2001 From: cardosofede Date: Mon, 14 Sep 2026 12:34:05 +0300 Subject: [PATCH 2/2] fix(trading): bound the order cursor in the index, and document what a cursor is The keyset cut in OrderRepository.get_orders was written as `created_at < t OR (created_at = t AND client_order_id < id)`. PostgreSQL runs that as a filter over a backward scan of ix_orders_created_at, so every page first walks and discards each order newer than the cursor: the O(offset) cost keyset pagination is meant to remove. At 1M orders a page 95% deep spent 107 ms discarding 950k rows. The same cut as a row-value comparison, `(created_at, client_order_id) < (t, id)`, lets the planner derive `created_at <= t` as an index condition on the existing index. The same deep page now runs in 0.06-0.08 ms, 5 rows removed by the filter, and a repository call costs the same at page one and at 95% depth. A composite (created_at, client_order_id) index was tried too. The planner ignored it, so it is not added. The route's docstring claimed any cursor it did not hand out gets a 400. It only checks the format. A well-formed cursor is a keyset position and pages back from that point whether or not the order it names exists. The docs now say that, and a test pins it. Verified against PostgreSQL 16 with 1M seeded orders: EXPLAIN plans at 5/50/95% depth, and a 2,000-order walk at limit 7 over tied created_at values that matches a direct ordered read exactly, with no repeats. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PjWZzXVi2w1hrbY3Ru9U1q --- database/repositories/order_repository.py | 14 ++++++-------- routers/trading.py | 10 +++++++--- test/test_order_history_cursor.py | 10 ++++++++++ 3 files changed, 23 insertions(+), 11 deletions(-) diff --git a/database/repositories/order_repository.py b/database/repositories/order_repository.py index 11fa67b9..69ef5153 100644 --- a/database/repositories/order_repository.py +++ b/database/repositories/order_repository.py @@ -3,7 +3,7 @@ from decimal import Decimal from typing import Dict, List, Optional, Tuple -from sqlalchemy import and_, func, or_, select +from sqlalchemy import func, select, tuple_ from sqlalchemy.ext.asyncio import AsyncSession from database.models import Order @@ -150,13 +150,11 @@ async def get_orders(self, account_names: Optional[List[str]] = None, select(Order), account_names, connector_names, trading_pairs, status, start_time, end_time ) if before is not None: - created_at, client_order_id = before - query = query.where( - or_( - Order.created_at < created_at, - and_(Order.created_at == created_at, Order.client_order_id < client_order_id), - ) - ) + # A row-value comparison, not the equivalent OR: PostgreSQL derives + # `created_at <= cursor` from it as a range bound on ix_orders_created_at, so a + # deep page starts at the cursor. The OR runs as a filter that walks every + # newer order first, making each page cost linear in how deep it is. + query = query.where(tuple_(Order.created_at, Order.client_order_id) < tuple_(*before)) query = query.order_by(Order.created_at.desc(), Order.client_order_id.desc()).limit(limit) diff --git a/routers/trading.py b/routers/trading.py index 57c16887..f622a4b9 100644 --- a/routers/trading.py +++ b/routers/trading.py @@ -251,7 +251,9 @@ async def get_orders( Orders come newest first. `pagination.next_cursor` is the cursor of the page's last order; pass it back as `cursor` to get the orders older than it. It is `null` on the - last page. A cursor this route did not hand out is refused with a 400. + last page. A cursor is a position, not a token: any `|` pages + from that point back, whether or not the order exists. A cursor not in that format is + refused with a 400. Args: filter_request: JSON payload with filtering criteria @@ -310,8 +312,10 @@ def _order_cursor(order: Dict) -> str: def _parse_order_cursor(cursor: str) -> Tuple[datetime, str]: """Split a cursor from `_order_cursor` back into (created_at, client_order_id). - Anything else is refused rather than read as "start over": an unrecognised cursor - used to serve page one again, which a client walking the history cannot tell from + Only the format is checked. A well-formed cursor is a keyset position, and the order it + names need not exist, the same way `before` works in the repository. Anything + unreadable is refused rather than read as "start over": an unrecognised cursor used + to serve page one again, which a client walking the history cannot tell from genuinely older orders. """ created_at, _, client_order_id = cursor.partition(_ORDER_CURSOR_SEPARATOR) diff --git a/test/test_order_history_cursor.py b/test/test_order_history_cursor.py index cb48256e..5a072d3c 100644 --- a/test/test_order_history_cursor.py +++ b/test/test_order_history_cursor.py @@ -188,3 +188,13 @@ def test_an_unreadable_cursor_is_refused_not_answered_with_page_one(client, curs assert response.status_code == 400, response.text assert "cursor" in response.json()["detail"] + + +def test_a_well_formed_cursor_is_a_position_even_for_an_order_that_does_not_exist(client): + """Only the format is checked: the cursor pages back from its point, like a keyset bound.""" + cursor = f"{(BASE + timedelta(seconds=1)).isoformat()}|a-03x" + + response = client.post("/trading/orders/search", json={"limit": 100, "cursor": cursor}) + + assert response.status_code == 200, response.text + assert [order["order_id"] for order in response.json()["data"]] == ["a-03", "a-02", "a-01"]