feat(perps): expose server time

src/polymarket/_internal/actions/perps/public.py

@@ -5,6 +5,8 @@
 from datetime import datetime
 from typing import Any, cast
 
+from pydantic import field_validator
+
 from polymarket._internal.actions.perps.paging import (
     ONE_DAY_MS,
     as_json_dict,
@@ -18,6 +20,10 @@
 )
 from polymarket.clients._transport import AsyncTransport
 from polymarket.errors import UnexpectedResponseError, UserInputError
+from polymarket.models.base import BaseModel
+from polymarket.models.perps._validators import (
+    _require_epoch_ms,  # pyright: ignore[reportPrivateUsage]
+)
 from polymarket.models.perps.market import (
     PerpsBook,
     PerpsCandle,
@@ -40,6 +46,15 @@
 _KLINE_INTERVALS = ("1s", "1m", "5m", "15m", "1h", "4h", "1d", "1w")
 
 
+class _PerpsServerTimeResponse(BaseModel):
+    time: datetime
+
+    @field_validator("time", mode="before")
+    @classmethod
+    def _parse_time(cls, value: object) -> object:
+        return _require_epoch_ms(value)
+
+
 def _validate_instrument_id(instrument_id: object, *, optional: bool = False) -> int | None:
     if instrument_id is None and optional:
         return None
@@ -50,6 +65,12 @@ def _validate_instrument_id(instrument_id: object, *, optional: bool = False) ->
     return instrument_id
 
 
+async def get_server_time(perps: AsyncTransport) -> datetime:
+    """Return the Perps server clock as an aware UTC datetime."""
+    response = _PerpsServerTimeResponse.parse_response(await perps.get_json("/v1/info/time"))
+    return response.time
+
+
 async def fetch_instruments(
     perps: AsyncTransport,
     *,

src/polymarket/clients/async_public.py

@@ -3,6 +3,7 @@
 import contextlib
 import logging
 from collections.abc import Sequence
+from datetime import datetime
 from decimal import Decimal
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, Literal, Self, assert_never, cast, overload
@@ -136,8 +137,6 @@
 )
 
 if TYPE_CHECKING:
-    from datetime import datetime
-
     from polymarket._internal.streams.clob.market import ClobMarketStreamManager
     from polymarket._internal.streams.perps.market import PerpsMarketStreamManager
     from polymarket._internal.streams.rtds.manager import RtdsStreamManager
@@ -1427,6 +1426,14 @@ async def fetch_perps_instruments(
             self._ctx.perps, instrument_id=instrument_id, category=category
         )
 
+    async def get_server_time(self) -> datetime:
+        """Experimental: This API may change in a breaking way in any release,
+        including patch releases.
+
+        Get the current Perps server time as an aware UTC datetime.
+        """
+        return await _perps_actions.get_server_time(self._ctx.perps)
+
     async def fetch_perps_ticker(self, *, instrument_id: int) -> PerpsTicker:
         """Experimental: This API may change in a breaking way in any release,
         including patch releases.

src/polymarket/clients/async_secure.py

@@ -3,6 +3,7 @@
 import logging
 import time
 from collections.abc import Awaitable, Callable, Mapping, Sequence
+from datetime import datetime
 from decimal import Decimal
 from types import TracebackType
 from typing import (
@@ -281,7 +282,7 @@
 from polymarket.types import EvmAddress, HexString, TransactionHash
 
 if TYPE_CHECKING:
-    from datetime import datetime, timedelta
+    from datetime import timedelta
 
     from polymarket._internal.perps_session import PerpsSession
     from polymarket._internal.rfq import RfqQuoterSession
@@ -3301,6 +3302,14 @@ async def fetch_perps_instruments(
             self._ctx.perps, instrument_id=instrument_id, category=category
         )
 
+    async def get_server_time(self) -> datetime:
+        """Experimental: This API may change in a breaking way in any release,
+        including patch releases.
+
+        Get the current Perps server time as an aware UTC datetime.
+        """
+        return await _perps_actions.get_server_time(self._ctx.perps)
+
     async def fetch_perps_ticker(self, *, instrument_id: int) -> PerpsTicker:
         """Experimental: This API may change in a breaking way in any release,
         including patch releases.

tests/integration/test_perps.py

@@ -115,6 +115,17 @@ async def test_public_perps_reads() -> None:
         assert candles.items
 
 
+@pytest.mark.integration
+async def test_get_server_time() -> None:
+    before = datetime.now(UTC) - timedelta(seconds=5)
+    async with AsyncPublicClient() as client:
+        server_time = await client.get_server_time()
+    after = datetime.now(UTC) + timedelta(seconds=5)
+
+    assert server_time.tzinfo is not None
+    assert before <= server_time <= after
+
+
 @pytest.mark.integration
 async def test_perps_book_subscription_receives_an_event() -> None:
     async with AsyncPublicClient() as client:

tests/unit/test_perps_public_actions.py

@@ -4,16 +4,18 @@
 import base64
 import json
 from collections.abc import Callable
+from datetime import datetime
 from decimal import Decimal
-from typing import Any
+from typing import Any, get_type_hints
 from urllib.parse import parse_qs, urlparse
 
 import httpx
 import pytest
 
 from polymarket._internal.actions.perps import public as perps_public
+from polymarket.clients import AsyncPublicClient, AsyncSecureClient
 from polymarket.clients._transport import AsyncTransport
-from polymarket.errors import UserInputError
+from polymarket.errors import UnexpectedResponseError, UserInputError
 
 _BASE_URL = "https://perps.test"
 
@@ -33,6 +35,36 @@ def _cursor(state: dict[str, Any]) -> str:
     return base64.b64encode(json.dumps(state, separators=(",", ":")).encode()).decode()
 
 
+@pytest.mark.parametrize("time", [1_786_697_208_022, True, "1786697208022", None])
+def test_get_server_time_parses_only_epoch_milliseconds(time: object) -> None:
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json={"time": time})
+
+    async def run() -> None:
+        transport = _transport(handler)
+        try:
+            if isinstance(time, int) and not isinstance(time, bool):
+                server_time = await perps_public.get_server_time(transport)
+                assert server_time.tzinfo is not None
+                assert int(server_time.timestamp() * 1000) == time
+            else:
+                with pytest.raises(UnexpectedResponseError):
+                    await perps_public.get_server_time(transport)
+        finally:
+            await transport.close()
+
+    asyncio.run(run())
+    assert [request.url.path for request in requests] == ["/v1/info/time"]
+
+
+def test_get_server_time_public_return_types_are_runtime_resolvable() -> None:
+    assert get_type_hints(AsyncPublicClient.get_server_time)["return"] is datetime
+    assert get_type_hints(AsyncSecureClient.get_server_time)["return"] is datetime
+
+
 def test_list_candles_steps_forward_by_interval() -> None:
     requests: list[dict[str, str]] = []