Removed unnecessary manual addition of pytest-cov in Makefile.

* Made a more thorough check of use of datetimes throughout codebase, replaced some instances.
* Added some tests to make sure comparisons don't raise exceptions.

.dockerignore

@@ -6,3 +6,4 @@
 wallet.json
 id.json
 private.py
+__pycache__

Dockerfile

@@ -16,5 +16,6 @@ RUN poetry install --no-dev --no-root
 
 # Have these as the last steps since the code here is the most-frequently changing
 COPY . /app/
+RUN python3 -m compileall /app
 ARG LAST_COMMIT=""
 RUN echo ${LAST_COMMIT} > /app/data/.version

Makefile

@@ -4,7 +4,6 @@ commands := $(wildcard bin/*)
 setup: ## Install all the build and lint dependencies
 	pip --no-cache-dir install poetry
 	poetry install --no-interaction
-	poetry add pytest-cov
 
 upgrade: ## Upgrade all the build and lint dependencies
 	poetry upgrade --no-interaction

bin/show-open-orders

@@ -48,8 +48,8 @@ with mango.ContextBuilder.from_command_line_parameters(args) as context:
             open_orders = mango.OpenOrders.load(
                 context,
                 slot.spot_open_orders,
-                slot.base_token_bank.token.decimals,
-                slot.quote_token_bank.token.decimals,
+                slot.base_token_bank.token,
+                slot.quote_token_bank.token,
             )
             mango.output(slot.base_instrument)
             mango.output(open_orders)

bin/show-serum-open-orders

@@ -37,8 +37,8 @@ with mango.ContextBuilder.from_command_line_parameters(args) as context:
         market.address,
         address,
         context.serum_program_address,
-        market.base.decimals,
-        market.quote.decimals,
+        market.base,
+        market.quote,
     )
     mango.output(
         f"Found {len(all_open_orders_for_market)} Serum OpenOrders account(s) for market {market.symbol}."

docs/MarketmakingIntroduction.md

@@ -170,7 +170,7 @@ def pulse(self, context: mango.Context, model_state: ModelState):
         settle = self.market_instruction_builder.build_settle_instructions()
         (payer + cancellations + place_orders + crank + settle).execute(context, on_exception_continue=True)
 
-        self.pulse_complete.on_next(datetime.now())
+        self.pulse_complete.on_next(mango.local_now())
     except Exception as exception:
         self._logger.error(f"[{context.name}] Market-maker error on pulse: {exception} - {traceback.format_exc()}")
         self.pulse_error.on_next(exception)

mango/__init__.py

@@ -60,6 +60,8 @@ from .constants import WARNING_DISCLAIMER_TEXT as WARNING_DISCLAIMER_TEXT
 from .constants import version as version
 from .context import Context as Context
 from .contextbuilder import ContextBuilder as ContextBuilder
+from .datetimes import datetime_from_chain as datetime_from_chain
+from .datetimes import datetime_from_timestamp as datetime_from_timestamp
 from .datetimes import local_now as local_now
 from .datetimes import utc_now as utc_now
 from .encoding import decode_binary as decode_binary

mango/account.py

@@ -34,6 +34,7 @@ from .instructions import (
 from .instrumentvalue import InstrumentValue
 from .layouts import layouts
 from .metadata import Metadata
+from .observables import Disposable
 from .openorders import OpenOrders
 from .orders import Side
 from .perpaccount import PerpAccount
@@ -44,6 +45,10 @@ from .tokenaccount import TokenAccount
 from .tokenbank import TokenBank
 from .version import Version
 from .wallet import Wallet
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 ReferrerMemory class
@@ -102,6 +107,20 @@ class ReferrerMemory(AddressableAccount):
             raise Exception(f"ReferrerMemory account not found at address '{address}'")
         return referrer_memory
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["ReferrerMemory"], None],
+    ) -> Disposable:
+        subscription = WebSocketAccountSubscription(
+            context, self.address, ReferrerMemory.parse
+        )
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def __str__(self) -> str:
         return f"""« ReferrerMemory [{self.version}] {self.address}
     {self.meta_data}
@@ -563,6 +582,24 @@ class Account(AddressableAccount):
 
         return accounts[0]
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["Account"], None],
+    ) -> Disposable:
+        group: Group = Group.load(context, self.group_address)
+        cache: Cache = group.fetch_cache(context)
+
+        def __parser(account_info: AccountInfo) -> Account:
+            return Account.parse(account_info, group, cache)
+
+        subscription = WebSocketAccountSubscription(context, self.address, __parser)
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def deposit(
         self, context: Context, wallet: Wallet, value: InstrumentValue
     ) -> typing.Sequence[str]:
@@ -688,8 +725,8 @@ class Account(AddressableAccount):
                 ]
                 oo = OpenOrders.parse(
                     account_info,
-                    slot.base_instrument.decimals,
-                    self.shared_quote.base_instrument.decimals,
+                    Token.ensure(slot.base_instrument),
+                    Token.ensure(self.shared_quote.base_instrument),
                 )
                 spot_open_orders[str(slot.spot_open_orders)] = oo
         return spot_open_orders

mango/accountinfoconverter.py

@@ -24,6 +24,7 @@ from .account import Account
 from .accountinfo import AccountInfo
 from .addressableaccount import AddressableAccount
 from .cache import Cache
+from .constants import SYSTEM_PROGRAM_ADDRESS
 from .context import Context
 from .group import Group
 from .layouts import layouts
@@ -64,9 +65,19 @@ def build_account_info_converter(
 
         return account_loader
     elif account_type_upper == "OPENORDERS":
-        return lambda account_info: OpenOrders.parse(
-            account_info, Decimal(6), Decimal(6)
+        base = Token(
+            "FAKEBASE",
+            "Fake Base Token",
+            Decimal(6),
+            SYSTEM_PROGRAM_ADDRESS,
         )
+        quote = Token(
+            "FAKEQUOTE",
+            "Fake Quote Token",
+            Decimal(6),
+            SYSTEM_PROGRAM_ADDRESS,
+        )
+        return lambda account_info: OpenOrders.parse(account_info, base, quote)
     elif account_type_upper == "CACHE":
         return lambda account_info: Cache.parse(account_info)
     elif account_type_upper == "ROOTBANK":

mango/addressableaccount.py

@@ -16,10 +16,14 @@
 
 import abc
 import logging
+import typing
 
 from solana.publickey import PublicKey
 
 from .accountinfo import AccountInfo
+from .context import Context
+from .observables import Disposable
+from .websocketsubscription import WebSocketSubscriptionManager
 
 
 # # 🥭 AddressableAccount class
@@ -40,5 +44,16 @@ class AddressableAccount(metaclass=abc.ABCMeta):
     def address(self) -> PublicKey:
         return self.account_info.address
 
+    @abc.abstractmethod
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["AddressableAccount"], None],
+    ) -> Disposable:
+        raise NotImplementedError(
+            "AddressableAccount.subscribe is not implemented on the base class."
+        )
+
     def __repr__(self) -> str:
         return f"{self}"

mango/cache.py

@@ -25,8 +25,13 @@ from .context import Context
 from .instrumentvalue import InstrumentValue
 from .layouts import layouts
 from .metadata import Metadata
+from .observables import Disposable
 from .tokens import Instrument, Token
 from .version import Version
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 PriceCache class
@@ -217,6 +222,18 @@ class Cache(AddressableAccount):
             raise Exception(f"Cache account not found at address '{address}'")
         return Cache.parse(account_info)
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["Cache"], None],
+    ) -> Disposable:
+        subscription = WebSocketAccountSubscription(context, self.address, Cache.parse)
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def market_cache_for_index(self, index: int) -> MarketCache:
         return MarketCache(
             self.price_cache[index],

mango/datetimes.py

@@ -17,20 +17,25 @@ from datetime import datetime, timezone
 
 # # 🥭 Datetimes
 #
-# This file contains a few useful datetime functions.
+# Comparing datetime values in Python can lead to the following error:
+#   TypeError: can't compare offset-naive and offset-aware datetimes
 #
-# They exist solely to make clear what is being used when called. There are 2 general cases:
-# 1. Logging/time tracking/user output - local time is what should be used.
-# 2. Comparison with dates stored on-chain - UTC should be used.
+# To avoid that, we use these functions to consistently create the datetime in a way
+# that has the correct value and that can be compared with other datetimes generated
+# by these methods.
 #
-# Getting a UTC time that is comparable with on-chain datetimes isn't intuitive, so putting
-# it in one place and using that consistently should make it clearer in the calling code
-# what is going on, without the unnecessary complications.
+# However, getting a comparable UTC time using this mechanism:
+#   return datetime.utcnow().astimezone(timezone.utc)
+#
+# works fine on Linux and Mac but fails on Windows. So we need to avoid that method and
+# be careful of different platform issues.
+#
+# If you use these methods to generate the datetime values, you _should_ be safe.
 #
 
 
 def local_now() -> datetime:
-    return datetime.now()
+    return datetime.now().astimezone()
 
 
 # Getting a comparable UTC time using this line:
@@ -40,3 +45,15 @@ def local_now() -> datetime:
 # on Windows as well as Mac and Linux.
 def utc_now() -> datetime:
     return datetime.utcnow().replace(tzinfo=timezone.utc)
+
+
+# All timestamp values _should be_ in UTC. All we need to do is load the value as a UTC
+# datetime that's comparable on all platforms, so we do this the same way as `utc_now()`.
+def datetime_from_timestamp(timestamp: float) -> datetime:
+    return datetime.utcfromtimestamp(timestamp).replace(tzinfo=timezone.utc)
+
+
+# All on-chain datetime values are in UTC. All we need to do is load the value as a UTC
+# datetime that's comparable on all platforms, so we do this the same way as `utc_now()`.
+def datetime_from_chain(onchain_value: float) -> datetime:
+    return datetime.utcfromtimestamp(onchain_value).replace(tzinfo=timezone.utc)

mango/group.py

@@ -30,9 +30,14 @@ from .layouts import layouts
 from .lotsizeconverter import LotSizeConverter, RaisingLotSizeConverter
 from .marketlookup import MarketLookup
 from .metadata import Metadata
+from .observables import Disposable
 from .tokens import Instrument, Token
 from .tokenbank import TokenBank
 from .version import Version
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 GroupSlotSpotMarket class
@@ -517,6 +522,26 @@ class Group(AddressableAccount):
             account_info, name, context.instrument_lookup, context.market_lookup
         )
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["Group"], None],
+    ) -> Disposable:
+        def __parser(account_info: AccountInfo) -> Group:
+            return Group.parse(
+                account_info,
+                self.name,
+                context.instrument_lookup,
+                context.market_lookup,
+            )
+
+        subscription = WebSocketAccountSubscription(context, self.address, __parser)
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def slot_by_spot_market_address(self, spot_market_address: PublicKey) -> GroupSlot:
         for slot in self.slots:
             if (

mango/layouts/layouts.py

@@ -33,12 +33,13 @@
 
 
 import construct
-import datetime
 import typing
 
+from datetime import datetime
 from decimal import Decimal, Context as DecimalContext
 from solana.publickey import PublicKey
 
+from ..datetimes import datetime_from_chain
 
 # # Adapters
 #
@@ -176,9 +177,7 @@ else:
 #
 if typing.TYPE_CHECKING:
 
-    class DatetimeAdapter(
-        construct.Adapter[datetime.datetime, int, typing.Any, typing.Any]
-    ):
+    class DatetimeAdapter(construct.Adapter[datetime, int, typing.Any, typing.Any]):
         def __init__(self) -> None:
             pass
 
@@ -188,14 +187,10 @@ else:
         def __init__(self) -> None:
             super().__init__(construct.BytesInteger(8, swapped=True))
 
-        def _decode(
-            self, obj: int, context: typing.Any, path: typing.Any
-        ) -> datetime.datetime:
-            return datetime.datetime.fromtimestamp(obj, tz=datetime.timezone.utc)
+        def _decode(self, obj: int, context: typing.Any, path: typing.Any) -> datetime:
+            return datetime_from_chain(obj)
 
-        def _encode(
-            self, obj: datetime.datetime, context: typing.Any, path: typing.Any
-        ) -> int:
+        def _encode(self, obj: datetime, context: typing.Any, path: typing.Any) -> int:
             return int(obj.timestamp())
 
 

mango/liquidationevent.py

@@ -13,9 +13,9 @@
 #   [Github](https://github.com/blockworks-foundation)
 #   [Email](mailto:hello@blockworks.foundation)
 
-import datetime
 import typing
 
+from datetime import datetime
 from solana.publickey import PublicKey
 
 from .instrumentvalue import InstrumentValue
@@ -26,7 +26,7 @@ from .instrumentvalue import InstrumentValue
 class LiquidationEvent:
     def __init__(
         self,
-        timestamp: datetime.datetime,
+        timestamp: datetime,
         liquidator_name: str,
         group_name: str,
         succeeded: bool,
@@ -36,7 +36,7 @@ class LiquidationEvent:
         balances_before: typing.Sequence[InstrumentValue],
         balances_after: typing.Sequence[InstrumentValue],
     ) -> None:
-        self.timestamp: datetime.datetime = timestamp
+        self.timestamp: datetime = timestamp
         self.liquidator_name: str = liquidator_name
         self.group_name: str = group_name
         self.succeeded: bool = succeeded

mango/marketmaking/modelstatebuilder.py

@@ -197,7 +197,7 @@ class SerumPollingModelStateBuilder(PollingModelStateBuilder):
         cache: mango.Cache = mango.Cache.parse(account_infos[1])
         account: mango.Account = mango.Account.parse(account_infos[2], group, cache)
         placed_orders_container: mango.PlacedOrdersContainer = mango.OpenOrders.parse(
-            account_infos[3], self.market.base.decimals, self.market.quote.decimals
+            account_infos[3], self.market.base, self.market.quote
         )
 
         # Serum markets don't accrue MNGO liquidity incentives
@@ -320,8 +320,8 @@ class SpotPollingModelStateBuilder(PollingModelStateBuilder):
                 )
                 open_orders: mango.OpenOrders = mango.OpenOrders.parse(
                     account_info,
-                    basket_token.base_instrument.decimals,
-                    account.shared_quote_token.decimals,
+                    Token.ensure(basket_token.base_instrument),
+                    account.shared_quote_token,
                 )
                 all_open_orders[str(basket_token.spot_open_orders)] = open_orders
 
@@ -440,8 +440,8 @@ class PerpPollingModelStateBuilder(PollingModelStateBuilder):
                 )
                 open_orders: mango.OpenOrders = mango.OpenOrders.parse(
                     account_info,
-                    basket_token.base_instrument.decimals,
-                    account.shared_quote_token.decimals,
+                    Token.ensure(basket_token.base_instrument),
+                    account.shared_quote_token,
                 )
                 all_open_orders[str(basket_token.spot_open_orders)] = open_orders
 

mango/marketmaking/modelstatebuilderfactory.py

@@ -127,8 +127,8 @@ def _polling_serum_model_state_builder_factory(
         market.address,
         wallet.address,
         context.serum_program_address,
-        market.base.decimals,
-        market.quote.decimals,
+        market.base,
+        market.quote,
     )
     if len(all_open_orders) == 0:
         raise Exception(

mango/openorders.py

@@ -28,8 +28,14 @@ from .context import Context
 from .encoding import encode_key
 from .group import Group
 from .layouts import layouts
+from .observables import Disposable
 from .placedorder import PlacedOrder
+from .tokens import Token
 from .version import Version
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 OpenOrders class
@@ -43,6 +49,8 @@ class OpenOrders(AddressableAccount):
         account_flags: AccountFlags,
         market: PublicKey,
         owner: PublicKey,
+        base: Token,
+        quote: Token,
         base_token_free: Decimal,
         base_token_total: Decimal,
         quote_token_free: Decimal,
@@ -56,6 +64,8 @@ class OpenOrders(AddressableAccount):
         self.account_flags: AccountFlags = account_flags
         self.market: PublicKey = market
         self.owner: PublicKey = owner
+        self.base: Token = base
+        self.quote: Token = quote
         self.base_token_free: Decimal = base_token_free
         self.base_token_total: Decimal = base_token_total
         self.quote_token_free: Decimal = quote_token_free
@@ -79,14 +89,14 @@ class OpenOrders(AddressableAccount):
     def from_layout(
         layout: typing.Any,
         account_info: AccountInfo,
-        base_decimals: Decimal,
-        quote_decimals: Decimal,
+        base: Token,
+        quote: Token,
     ) -> "OpenOrders":
         account_flags = AccountFlags.from_layout(layout.account_flags)
         program_address = account_info.owner
 
-        base_divisor = 10**base_decimals
-        quote_divisor = 10**quote_decimals
+        base_divisor = 10**base.decimals
+        quote_divisor = 10**quote.decimals
         base_token_free: Decimal = layout.base_token_free / base_divisor
         base_token_total: Decimal = layout.base_token_total / base_divisor
         quote_token_free: Decimal = layout.quote_token_free / quote_divisor
@@ -110,6 +120,8 @@ class OpenOrders(AddressableAccount):
             account_flags,
             layout.market,
             layout.owner,
+            base,
+            quote,
             base_token_free,
             base_token_total,
             quote_token_free,
@@ -119,9 +131,7 @@ class OpenOrders(AddressableAccount):
         )
 
     @staticmethod
-    def parse(
-        account_info: AccountInfo, base_decimals: Decimal, quote_decimals: Decimal
-    ) -> "OpenOrders":
+    def parse(account_info: AccountInfo, base: Token, quote: Token) -> "OpenOrders":
         data = account_info.data
         if len(data) != layouts.OPEN_ORDERS.sizeof():
             raise Exception(
@@ -129,9 +139,7 @@ class OpenOrders(AddressableAccount):
             )
 
         layout = layouts.OPEN_ORDERS.parse(data)
-        return OpenOrders.from_layout(
-            layout, account_info, base_decimals, quote_decimals
-        )
+        return OpenOrders.from_layout(layout, account_info, base, quote)
 
     @staticmethod
     def load_raw_open_orders_account_infos(
@@ -163,13 +171,13 @@ class OpenOrders(AddressableAccount):
     def load(
         context: Context,
         address: PublicKey,
-        base_decimals: Decimal,
-        quote_decimals: Decimal,
+        base: Token,
+        quote: Token,
     ) -> "OpenOrders":
         open_orders_account = AccountInfo.load(context, address)
         if open_orders_account is None:
             raise Exception(f"OpenOrders account not found at address '{address}'")
-        return OpenOrders.parse(open_orders_account, base_decimals, quote_decimals)
+        return OpenOrders.parse(open_orders_account, base, quote)
 
     @staticmethod
     def load_for_market_and_owner(
@@ -177,8 +185,8 @@ class OpenOrders(AddressableAccount):
         market: PublicKey,
         owner: PublicKey,
         program_address: PublicKey,
-        base_decimals: Decimal,
-        quote_decimals: Decimal,
+        base: Token,
+        quote: Token,
     ) -> typing.Sequence["OpenOrders"]:
         filters = [
             MemcmpOpts(
@@ -197,11 +205,30 @@ class OpenOrders(AddressableAccount):
         )
         return list(
             map(
-                lambda acc: OpenOrders.parse(acc, base_decimals, quote_decimals),
+                lambda acc: OpenOrders.parse(acc, base, quote),
                 account_infos,
             )
         )
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["OpenOrders"], None],
+    ) -> Disposable:
+        def __parser(account_info: AccountInfo) -> OpenOrders:
+            return OpenOrders.parse(
+                account_info,
+                self.base,
+                self.quote,
+            )
+
+        subscription = WebSocketAccountSubscription(context, self.address, __parser)
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def __str__(self) -> str:
         placed_orders = "\n        ".join(map(str, self.placed_orders)) or "None"
 

mango/oracles/ftx/ftx.py

@@ -20,12 +20,11 @@ import rx
 import typing
 import websocket
 
-from datetime import datetime
 from decimal import Decimal
 from rx.subject.subject import Subject
 
 from ...context import Context
-from ...datetimes import utc_now
+from ...datetimes import utc_now, datetime_from_timestamp
 from ...markets import Market
 from ...observables import Disposable, DisposeWrapper
 from ...oracle import (
@@ -97,7 +96,7 @@ class FtxOracle(Oracle):
                 ask = Decimal(data["data"]["ask"])
                 mid = (bid + ask) / Decimal(2)
                 time = data["data"]["time"]
-                timestamp = datetime.fromtimestamp(time)
+                timestamp = datetime_from_timestamp(time)
                 price = Price(
                     self.source,
                     timestamp,

mango/orderbookside.py

@@ -25,9 +25,14 @@ from .addressableaccount import AddressableAccount
 from .context import Context
 from .layouts import layouts
 from .metadata import Metadata
+from .observables import Disposable
 from .orders import Order, OrderType, Side
 from .perpmarketdetails import PerpMarketDetails
 from .version import Version
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 OrderBookSideType enum
@@ -130,6 +135,21 @@ class PerpOrderBookSide(AddressableAccount):
             )
         return PerpOrderBookSide.parse(account_info, perp_market_details)
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["PerpOrderBookSide"], None],
+    ) -> Disposable:
+        def __parser(account_info: AccountInfo) -> PerpOrderBookSide:
+            return PerpOrderBookSide.parse(account_info, self.perp_market_details)
+
+        subscription = WebSocketAccountSubscription(context, self.address, __parser)
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def orders(self) -> typing.Sequence[Order]:
         if self.leaf_count == 0:
             return []

mango/orders.py

@@ -20,13 +20,13 @@ import pyserum.enums
 import typing
 
 from dataclasses import dataclass
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timedelta
 from decimal import Decimal
 from pyserum.market.types import Order as PySerumOrder
 from solana.publickey import PublicKey
 
 from .constants import SYSTEM_PROGRAM_ADDRESS
-from .datetimes import utc_now
+from .datetimes import utc_now, datetime_from_timestamp
 from .lotsizeconverter import LotSizeConverter
 
 
@@ -145,9 +145,7 @@ class OrderType(enum.Enum):
 @dataclass(frozen=True)
 class Order:
     DefaultMatchLimit: typing.ClassVar[int] = 20
-    NoExpiration: typing.ClassVar[datetime] = datetime.fromtimestamp(0).astimezone(
-        timezone.utc
-    )
+    NoExpiration: typing.ClassVar[datetime] = datetime_from_timestamp(0)
     id: int
     client_id: int
     owner: PublicKey

mango/perpeventqueue.py

@@ -26,8 +26,13 @@ from .context import Context
 from .layouts import layouts
 from .lotsizeconverter import LotSizeConverter
 from .metadata import Metadata
+from .observables import Disposable
 from .orders import Side
 from .version import Version
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 PerpEvent class
@@ -263,6 +268,7 @@ class PerpEventQueue(AddressableAccount):
         account_info: AccountInfo,
         version: Version,
         meta_data: Metadata,
+        lot_size_converter: LotSizeConverter,
         head: Decimal,
         count: Decimal,
         sequence_number: Decimal,
@@ -273,6 +279,7 @@ class PerpEventQueue(AddressableAccount):
         self.version: Version = version
 
         self.meta_data: Metadata = meta_data
+        self.lot_size_converter: LotSizeConverter = lot_size_converter
         self.head: Decimal = head
         self.count: Decimal = count
         self.sequence_number: Decimal = sequence_number
@@ -311,6 +318,7 @@ class PerpEventQueue(AddressableAccount):
             account_info,
             version,
             meta_data,
+            lot_size_converter,
             head,
             count,
             seq_num,
@@ -353,6 +361,21 @@ class PerpEventQueue(AddressableAccount):
 
         return distinct
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["PerpEventQueue"], None],
+    ) -> Disposable:
+        def __parser(account_info: AccountInfo) -> PerpEventQueue:
+            return PerpEventQueue.parse(account_info, self.lot_size_converter)
+
+        subscription = WebSocketAccountSubscription(context, self.address, __parser)
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def events_for_account(
         self, mango_account_address: PublicKey
     ) -> typing.Sequence[PerpEvent]:

mango/perpmarket.py

@@ -14,7 +14,6 @@
 #   [Email](mailto:hello@blockworks.foundation)
 
 
-import rx.operators
 import typing
 
 from dataclasses import dataclass
@@ -56,7 +55,6 @@ from .tokenbank import TokenBank
 from .wallet import Wallet
 from .websocketsubscription import (
     IndividualWebSocketSubscriptionManager,
-    WebSocketAccountSubscription,
 )
 
 
@@ -225,26 +223,17 @@ class PerpMarket(LoadedMarket):
             context, self.event_queue_address, self.lot_size_converter
         )
 
-        splitter: UnseenPerpEventChangesTracker = UnseenPerpEventChangesTracker(initial)
-        event_queue_subscription = WebSocketAccountSubscription(
-            context,
-            self.event_queue_address,
-            lambda account_info: PerpEventQueue.parse(
-                account_info, self.lot_size_converter
-            ),
-        )
-        disposer.add_disposable(event_queue_subscription)
-
         manager = IndividualWebSocketSubscriptionManager(context)
         disposer.add_disposable(manager)
-        manager.add(event_queue_subscription)
 
-        publisher = event_queue_subscription.publisher.pipe(
-            rx.operators.flat_map(splitter.unseen)
-        )
+        splitter: UnseenPerpEventChangesTracker = UnseenPerpEventChangesTracker(initial)
 
-        individual_event_subscription = publisher.subscribe(on_next=handler)
-        disposer.add_disposable(individual_event_subscription)
+        def __split_handler(pev: PerpEventQueue) -> None:
+            for event in splitter.unseen(pev):
+                handler(event)
+
+        subscription = initial.subscribe(context, manager, __split_handler)
+        disposer.add_disposable(subscription)
 
         manager.open()
 

mango/perpmarketdetails.py

@@ -27,9 +27,14 @@ from .group import GroupSlot, Group
 from .instrumentvalue import InstrumentValue
 from .layouts import layouts
 from .metadata import Metadata
+from .observables import Disposable
 from .tokens import Instrument, Token
 from .tokenbank import TokenBank
 from .version import Version
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 class LiquidityMiningInfo:
@@ -247,6 +252,21 @@ class PerpMarketDetails(AddressableAccount):
             )
         return PerpMarketDetails.parse(account_info, group)
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["PerpMarketDetails"], None],
+    ) -> Disposable:
+        def __parser(account_info: AccountInfo) -> PerpMarketDetails:
+            return PerpMarketDetails.parse(account_info, self.group)
+
+        subscription = WebSocketAccountSubscription(context, self.address, __parser)
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def __str__(self) -> str:
         liquidity_mining_info: str = f"{self.liquidity_mining_info}".replace(
             "\n", "\n        "

mango/serumeventqueue.py

@@ -23,7 +23,12 @@ from .accountinfo import AccountInfo
 from .addressableaccount import AddressableAccount
 from .context import Context
 from .layouts import layouts
+from .observables import Disposable
 from .version import Version
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 SerumEventFlags class
@@ -233,6 +238,20 @@ class SerumEventQueue(AddressableAccount):
     def capacity(self) -> int:
         return len(self.unprocessed_events) + len(self.processed_events)
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["SerumEventQueue"], None],
+    ) -> Disposable:
+        subscription = WebSocketAccountSubscription(
+            context, self.address, SerumEventQueue.parse
+        )
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def __str__(self) -> str:
         unprocessed_events = (
             "\n        ".join(

mango/serummarket.py

@@ -126,8 +126,8 @@ class SerumMarket(LoadedMarket):
             self.address,
             owner,
             context.serum_program_address,
-            self.base.decimals,
-            self.quote.decimals,
+            self.base,
+            self.quote,
         )
         if len(all_open_orders) == 0:
             return None
@@ -243,8 +243,8 @@ class SerumMarketInstructionBuilder(MarketInstructionBuilder):
             serum_market.address,
             wallet.address,
             context.serum_program_address,
-            serum_market.base.decimals,
-            serum_market.quote.decimals,
+            serum_market.base,
+            serum_market.quote,
         )
         if len(all_open_orders) > 0:
             open_orders_address = all_open_orders[0].address

mango/tokenaccount.py

@@ -30,9 +30,14 @@ from .context import Context
 from .instrumentlookup import InstrumentLookup
 from .instrumentvalue import InstrumentValue
 from .layouts import layouts
+from .observables import Disposable
 from .tokens import Instrument, Token
 from .version import Version
 from .wallet import Wallet
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 TokenAccount class
@@ -208,6 +213,23 @@ class TokenAccount(AddressableAccount):
             account_info, instrument_lookup=context.instrument_lookup
         )
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["TokenAccount"], None],
+    ) -> Disposable:
+        token = Token.ensure(self.value.token)
+
+        def __parser(account_info: AccountInfo) -> TokenAccount:
+            return TokenAccount.parse(account_info, token=token)
+
+        subscription = WebSocketAccountSubscription(context, self.address, __parser)
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def __str__(self) -> str:
         return (
             f"« TokenAccount {self.address}, Owner: {self.owner}, Value: {self.value} »"

mango/tokenbank.py

@@ -28,8 +28,13 @@ from .context import Context
 from .instrumentlookup import InstrumentLookup
 from .layouts import layouts
 from .metadata import Metadata
+from .observables import Disposable
 from .tokens import Instrument, Token
 from .version import Version
+from .websocketsubscription import (
+    WebSocketAccountSubscription,
+    WebSocketSubscriptionManager,
+)
 
 
 # # 🥭 InterestRates class
@@ -115,6 +120,20 @@ class NodeBank(AddressableAccount):
             raise Exception(f"NodeBank account not found at address '{address}'")
         return NodeBank.parse(account_info)
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["NodeBank"], None],
+    ) -> Disposable:
+        subscription = WebSocketAccountSubscription(
+            context, self.address, NodeBank.parse
+        )
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def __str__(self) -> str:
         return f"""« NodeBank [{self.version}] {self.address}
     {self.meta_data}
@@ -261,6 +280,20 @@ class RootBank(AddressableAccount):
 
         return found[0]
 
+    def subscribe(
+        self,
+        context: Context,
+        websocketmanager: WebSocketSubscriptionManager,
+        callback: typing.Callable[["RootBank"], None],
+    ) -> Disposable:
+        subscription = WebSocketAccountSubscription(
+            context, self.address, RootBank.parse
+        )
+        websocketmanager.add(subscription)
+        subscription.publisher.subscribe(on_next=callback)  # type: ignore[call-arg]
+
+        return subscription
+
     def __str__(self) -> str:
         return f"""« RootBank [{self.version}] {self.address}
     {self.meta_data}

mango/transactionscout.py

@@ -15,15 +15,16 @@
 
 
 import base58
-import datetime
 import logging
 import traceback
 import typing
 
+from datetime import datetime
 from decimal import Decimal
 from solana.publickey import PublicKey
 
 from .context import Context
+from .datetimes import datetime_from_timestamp
 from .instructiontype import InstructionType
 from .instrumentvalue import InstrumentValue
 from .logmessages import expand_log_messages
@@ -69,7 +70,7 @@ from .text import indent_collection_as_str, indent_item_by
 class TransactionScout:
     def __init__(
         self,
-        timestamp: datetime.datetime,
+        timestamp: datetime,
         signatures: typing.Sequence[str],
         succeeded: bool,
         group_name: str,
@@ -79,7 +80,7 @@ class TransactionScout:
         pre_token_balances: typing.Sequence[OwnedInstrumentValue],
         post_token_balances: typing.Sequence[OwnedInstrumentValue],
     ) -> None:
-        self.timestamp: datetime.datetime = timestamp
+        self.timestamp: datetime = timestamp
         self.signatures: typing.Sequence[str] = signatures
         self.succeeded: bool = succeeded
         self.group_name: str = group_name
@@ -193,7 +194,7 @@ class TransactionScout:
                 if len(instructions) > 0
                 else "No Group"
             )
-            timestamp = datetime.datetime.fromtimestamp(response["blockTime"])
+            timestamp = datetime_from_timestamp(response["blockTime"])
             signatures = response["transaction"]["signatures"]
             raw_messages = response["meta"]["logMessages"]
             messages = expand_log_messages(raw_messages)

mango/watchers.py

@@ -144,15 +144,15 @@ def build_spot_open_orders_watcher(
         context,
         open_orders_address,
         lambda account_info: OpenOrders.parse(
-            account_info, spot_market.base.decimals, spot_market.quote.decimals
+            account_info, spot_market.base, spot_market.quote
         ),
     )
     manager.add(spot_open_orders_subscription)
     initial_spot_open_orders = OpenOrders.load(
         context,
         open_orders_address,
-        spot_market.base.decimals,
-        spot_market.quote.decimals,
+        spot_market.base,
+        spot_market.quote,
     )
     latest_open_orders_observer = LatestItemObserverSubscriber[OpenOrders](
         initial_spot_open_orders
@@ -176,8 +176,8 @@ def build_serum_open_orders_watcher(
         serum_market.address,
         wallet.address,
         context.serum_program_address,
-        serum_market.base.decimals,
-        serum_market.quote.decimals,
+        serum_market.base,
+        serum_market.quote,
     )
     if len(all_open_orders) > 0:
         initial_serum_open_orders: OpenOrders = all_open_orders[0]
@@ -201,15 +201,15 @@ def build_serum_open_orders_watcher(
         initial_serum_open_orders = OpenOrders.load(
             context,
             open_orders_address,
-            serum_market.base.decimals,
-            serum_market.quote.decimals,
+            serum_market.base,
+            serum_market.quote,
         )
 
     serum_open_orders_subscription = WebSocketAccountSubscription[OpenOrders](
         context,
         open_orders_address,
         lambda account_info: OpenOrders.parse(
-            account_info, serum_market.base.decimals, serum_market.quote.decimals
+            account_info, serum_market.base, serum_market.quote
         ),
     )
 

tests/data.py

@@ -4,6 +4,8 @@ import typing
 
 from decimal import Decimal
 
+from .fakes import fake_seeded_public_key
+
 
 def load_group(filename: str) -> mango.Group:
     account_info: mango.AccountInfo = mango.AccountInfo.load_json(filename)
@@ -42,8 +44,21 @@ def load_account(
 
 def load_openorders(filename: str) -> mango.OpenOrders:
     account_info: mango.AccountInfo = mango.AccountInfo.load_json(filename)
-    # Just hard-code the decimals for now.
-    return mango.OpenOrders.parse(account_info, Decimal(6), Decimal(6))
+    # Just hard-code the tokens for now.
+    base = mango.Token(
+        "FAKEBASE",
+        "Fake Base Token",
+        Decimal(6),
+        fake_seeded_public_key("fake base token"),
+    )
+    quote = mango.Token(
+        "FAKEQUOTE",
+        "Fake Quote Token",
+        Decimal(6),
+        fake_seeded_public_key("fake quote token"),
+    )
+
+    return mango.OpenOrders.parse(account_info, base, quote)
 
 
 def load_cache(filename: str) -> mango.Cache:

tests/fakes.py

@@ -462,6 +462,8 @@ def fake_open_orders(
     market = fake_seeded_public_key("market")
     owner = fake_seeded_public_key("owner")
 
+    base = fake_token("FAKEBASE")
+    quote = fake_token("FAKEQUOTE")
     flags = mango.AccountFlags(
         mango.Version.V1, True, False, True, False, False, False, False, False
     )
@@ -472,6 +474,8 @@ def fake_open_orders(
         flags,
         market,
         owner,
+        base,
+        quote,
         base_token_free,
         base_token_total,
         quote_token_free,

tests/test_addressableaccount.py

@@ -1,9 +1,22 @@
+import typing
+
 from .context import mango
 from .fakes import fake_account_info
 
 
 def test_constructor() -> None:
+    class __derived(mango.AddressableAccount):
+        def subscribe(
+            self,
+            context: mango.Context,
+            websocketmanager: mango.WebSocketSubscriptionManager,
+            callback: typing.Callable[[mango.AddressableAccount], None],
+        ) -> mango.Disposable:
+            raise NotImplementedError(
+                "AddressableAccount.subscribe is not implemented on this test class."
+            )
+
     account_info = fake_account_info()
-    actual = mango.AddressableAccount(account_info)
+    actual = __derived(account_info)
     assert actual is not None
     assert actual.address == account_info.address

tests/test_datetimes.py

@@ -0,0 +1,25 @@
+from .context import mango
+
+
+def test_compare_from_timestamp_to_utc_now() -> None:
+    timestamp = mango.datetime_from_timestamp(1645793955)
+    now = mango.utc_now()
+    assert now > timestamp
+
+
+def test_compare_from_timestamp_to_chain() -> None:
+    timestamp = mango.datetime_from_timestamp(1645793955)
+    chain = mango.datetime_from_chain(1645793955)
+    assert chain == timestamp
+
+
+def test_compare_from_chain_to_utc_now() -> None:
+    chain = mango.datetime_from_chain(1645793955)
+    now = mango.utc_now()
+    assert now > chain
+
+
+def test_compare_from_chain_to_local_now() -> None:
+    chain = mango.datetime_from_chain(1645793955)
+    now = mango.local_now()
+    assert now > chain

tests/test_openorders.py

@@ -1,14 +1,26 @@
 from .context import mango
-from .fakes import fake_account_info, fake_public_key
+from .fakes import fake_account_info, fake_seeded_public_key
 
 from decimal import Decimal
 
 
 def test_constructor() -> None:
     account_info = fake_account_info()
-    program_address = fake_public_key()
-    market = fake_public_key()
-    owner = fake_public_key()
+    program_address = fake_seeded_public_key("program_address")
+    market = fake_seeded_public_key("market")
+    owner = fake_seeded_public_key("owner")
+    base = mango.Token(
+        "FAKEBASE",
+        "Fake Base Token",
+        Decimal(6),
+        fake_seeded_public_key("fake base token"),
+    )
+    quote = mango.Token(
+        "FAKEQUOTE",
+        "Fake Quote Token",
+        Decimal(6),
+        fake_seeded_public_key("fake quote token"),
+    )
 
     flags = mango.AccountFlags(
         mango.Version.V1, True, False, True, False, False, False, False, False
@@ -20,6 +32,8 @@ def test_constructor() -> None:
         flags,
         market,
         owner,
+        base,
+        quote,
         Decimal(0),
         Decimal(0),
         Decimal(0),

tests/test_perpeventqueue.py

@@ -24,6 +24,7 @@ def test_constructor() -> None:
         account_info,
         mango.Version.V1,
         meta_data,
+        mango.NullLotSizeConverter(),
         head,
         count,
         sequence_number,
@@ -57,6 +58,7 @@ def _fake_pev(
         account_info,
         mango.Version.V1,
         meta_data,
+        mango.NullLotSizeConverter(),
         head,
         count,
         sequence_number,