degenbot.uniswap.concentrated.snapshot_readers
==============================================

.. py:module:: degenbot.uniswap.concentrated.snapshot_readers

.. autoapi-nested-parse::

   Generic snapshot readers and liquidity-snapshot facade shared by Uniswap V3/V4.





Module Contents
---------------

.. py:class:: LiquidityMap

   Bases: :py:obj:`TypedDict`


   LiquidityMap class.


   .. py:attribute:: tick_bitmap
      :type:  dict[int, degenbot.uniswap.concentrated.types.BitmapAtWord]


   .. py:attribute:: tick_data
      :type:  dict[int, degenbot.uniswap.concentrated.types.LiquidityAtTick]


.. py:class:: LiquidityEvent

   Bases: :py:obj:`Protocol`


   The fields a liquidity event exposes to the snapshot update flow.


   .. py:attribute:: block_number
      :type:  int


   .. py:attribute:: liquidity
      :type:  int


   .. py:attribute:: tick_lower
      :type:  int


   .. py:attribute:: tick_upper
      :type:  int


.. py:class:: LiquiditySnapshotSource

   Bases: :py:obj:`Protocol`


   A minimal protocol for retrieving pool data from a generic source.

   Any class implementing the protocol must implement these methods,
   transforming data as necessary to return the specified types.


   .. py:attribute:: storage_kind
      :type:  str


   .. py:attribute:: chain_id
      :type:  int


   .. py:method:: get_newest_block() -> degenbot.types.aliases.BlockNumber | None

      Return newest block.

      :returns: The newest block number, or None if unavailable.



   .. py:method:: get_pools() -> set[Any]

      Return pools.

      :returns: The set of pool keys.



.. py:class:: MonolithicJsonFileSnapshotBase(path: pathlib.Path | str)

   A pool liquidity source backed by a single JSON file.

   {
       "snapshot_block": int,
       "chain_id": int,
       "0xPoolKey1": {
           "tick_bitmap": {
               <word>: {
                   'bitmap': <value>,
                   'block': <value>,
               },
               ...
           },
           "tick_data": {
               <tick>: {
                   'liquidity_gross: <value>,
                   'liquidity_net': <value>,
                   'block: <value>,
               }
           }
       },
       "0xPoolKey2": { ... },
       "0xPoolKey3": { ... },
       ...
   }.


   .. py:attribute:: storage_kind
      :value: 'file'



   .. py:attribute:: chain_id
      :type:  int


   .. py:method:: get_newest_block() -> degenbot.types.aliases.BlockNumber | None

      Return newest block.

      :returns: The snapshot block number, or None if unavailable.



   .. py:method:: get_pools() -> set[str]

      Return pools.

      :returns: The set of pool keys from the snapshot.



.. py:class:: IndividualJsonFileSnapshotBase(path: pathlib.Path | str)

   Snapshot source backed by a directory of JSON files with this tree structure.

       /path/to/snapshots/
       |-- _metadata.json              -> { "block": int, "chain_id": int }
       |-- 0xPoolKey1.json             -> { "tick_bitmap": {...}, "tick_data": {...} }
       |-- 0xPoolKey2.json             -> { "tick_bitmap": {...}, "tick_data": {...} }
       |-- 0xPoolKey3.json             -> { "tick_bitmap": {...}, "tick_data": {...} }

   Each pool file contains the same structure as the monolithic snapshot's per-pool entries.


   .. py:attribute:: storage_kind
      :value: 'dir'



   .. py:attribute:: chain_id
      :type:  int


   .. py:method:: get_newest_block() -> degenbot.types.aliases.BlockNumber | None

      Return newest block.

      :returns: The block number from metadata, or None if unavailable.



   .. py:method:: get_pools() -> set[str]

      Return pools.

      :returns: The set of pool keys from file stems.



   .. py:method:: get_liquidity_map(pool_address: str) -> LiquidityMap | None

      Return liquidity map.

      :returns: The liquidity map for the pool, or None if not found.



.. py:class:: DatabaseSnapshotBase[K](chain_id: degenbot.types.aliases.ChainId, *, database_path: pathlib.Path)

   Snapshot source backed by built-in SQLite database.

   Routes every read through the Rust `degenbot-db` core crate via the
   `_EngineSnapshot` PyO3 seam (ADR-005 three-layer architecture). The
   explicit file path is retained as the authority for the lazy Rust handle.


   .. py:attribute:: storage_kind
      :value: 'db'



   .. py:attribute:: database_path


   .. py:attribute:: chain_id


   .. py:method:: close() -> None

      Release the Rust handle and its SQLite connection.

      Closing is idempotent. A closed snapshot cannot lazily reopen its
      database handle.



   .. py:method:: get_all_liquidity_maps() -> dict[K, dict[int, tuple[int, int]]]

      Return all tick data as plain dicts.

      Delegates the bulk read to the Rust core (GIL released during the
      SQLite scan). Returns {pool_key: {tick_index: (liquidity_gross,
      liquidity_net)}}.

      :returns: A dict mapping pool keys to tick data dicts.



   .. py:method:: get_newest_block() -> degenbot.types.aliases.BlockNumber | None

      Return newest block.

      :returns: The newest block number across all exchanges, or None if unavailable.



   .. py:method:: get_pools() -> set[str]

      Return pools.

      :returns: The set of pool keys from the database.



.. py:class:: LiquiditySnapshotBase[K, U, E: LiquidityEvent, S: LiquiditySnapshotSource](source: S)

   Retrieve and maintain liquidity positions for concentrated-liquidity pools.


   .. py:attribute:: newest_block
      :type:  degenbot.types.aliases.BlockNumber


   .. py:property:: chain_id
      :type: int


      Chain id.


   .. py:property:: pools
      :type: set[Any]


      Pools.