fix(trading): page order history by a real keyset cursor, not the same "0:" forever

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 func, select, tuple_
 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,52 @@ 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:
+            # 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)
 
         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,

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,83 +244,98 @@ 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 is a position, not a token: any `<created_at ISO>|<order id>` 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
 
     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 "<created_at ISO timestamp>|<client order id>". 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).
+
+    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)
+    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
 @router.post("/trades", response_model=PaginatedResponse)
 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 +388,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 +492,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 +510,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 +556,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 +627,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,