major refactor

all the logic now in the max_pain script

max_pain.py

@@ -1,15 +1,111 @@
 #!/usr/bin/env python
+from decimal import (
+    Decimal,
+)
 import trio
 import tractor
+from pprint import pformat
 from piker.brokers.deribit.api import (
+    get_client,
     maybe_open_oi_feed,
 )
 
 
 async def max_pain_daemon(
 ) -> None:
-    async with maybe_open_oi_feed() as oi_feed:
+    expiry_date: str = '13DEC24'
-        print('Im in...')
+    instruments: list[Symbol] = []
+    oi_by_strikes: dict[str, dict[str, Decimal]]
+
+    def check_if_complete(
+        oi: dict[str, dict[str, Decimal | None]],
+    
+        ) -> bool:
+            for strike in oi:
+                if (
+                    oi[strike]['C'] == None 
+                    or
+                    oi[strike]['P'] == None
+                ):
+                    return False
+    
+            return True
+
+    def get_max_pain(
+        oi_by_strikes: dict[str, dict[str, Decimal]]
+    ) -> dict[str, str | Decimal]:
+        '''
+        This method requires only the strike_prices and oi for call
+        and puts, the closes list are the same as the strike_prices
+        the idea is to sum all the calls and puts cash for each strike
+        and the ITM strikes from that strike, the lowest value is what we 
+        are looking for the intrinsic value.
+
+        '''
+
+        nonlocal timestamp
+        # We meed to find the lowest value, so we start at 
+        # infinity to ensure that, and the max_pain must be 
+        # an amount greater than zero.
+        total_intrinsic_value: Decimal = Decimal('Infinity')
+        max_pain: Decimal = Decimal(0)
+
+        intrinsic_values: dict[str, dict[str, Decimal]] = {}
+        closes: list[str] = sorted(oi_by_strikes.keys())
+        for strike in oi_by_strikes:
+            s: Decimal = Decimal(f'{strike}')
+            call_cash: Decimal = Decimal(0)
+            put_cash: Decimal = Decimal(0)
+            for close in closes:
+                c: Decimal = Decimal(f'{close}')
+                call_cash += max(0, (s - c) * oi_by_strikes[f'{close}']['C'])
+                put_cash += max(0, (c - s) * oi_by_strikes[f'{close}']['P'])
+    
+            intrinsic_values[strike] = {
+                'C': call_cash,
+                'P': put_cash,
+                'total': call_cash + put_cash,
+            }
+
+        for strike in intrinsic_values:
+            if intrinsic_values[f'{strike}']['total'] < total_intrinsic_value:
+                total_intrinsic_value = intrinsic_values[f'{strike}']['total']
+                max_pain = strike\
+
+        return {
+            'timestamp': timestamp,
+            'expiry_date': expiry_date,
+            'total_intrinsic_value': total_intrinsic_value,
+            'max_pain': max_pain,
+        }
+
+                
+    async with get_client(
+    ) as client:
+        instruments = await client.get_instruments(
+            expiry_date=expiry_date,
+        )
+        oi_by_strikes = client.get_strikes_dict(instruments)
+
+    async with maybe_open_oi_feed(
+        instruments,
+    ) as oi_feed:
+        async for msg in oi_feed:
+            if 'oi' == msg[0]:
+                timestamp = msg[1]['timestamp']
+                strike_price = msg[1]['strike_price']
+                option_type = msg[1]['option_type']
+                open_interest = msg[1]['open_interest']
+                oi_by_strikes[f'{strike_price}'][f'{option_type}'] = open_interest
+
+            if check_if_complete(oi_by_strikes):
+                max_pain = get_max_pain(oi_by_strikes)
+                print('-----------------------------------------------')
+                print(f'timestamp:             {max_pain['timestamp']}')
+                print(f'expiry_date:           {max_pain['expiry_date']}')
+                print(f'max_pain:              {max_pain['max_pain']}')
+                print(f'total intrinsic value: {max_pain['total_intrinsic_value']}')
+                print('-----------------------------------------------')
 
 
 async def main():

piker/brokers/deribit/api.py

@@ -246,20 +246,6 @@ def get_config() -> dict[str, Any]:
 
     return section
 
-def check_if_complete(
-    oi: dict[str, dict[str, Decimal | None]],
-
-    ) -> bool:
-        for strike in oi:
-            if (
-                oi[strike]['C'] == None 
-                or
-                oi[strike]['P'] == None
-            ):
-                return False
-
-        return True
-                
 
 class Client:
     '''
@@ -826,16 +812,10 @@ async def maybe_open_price_feed(
 
 async def aio_open_interest_feed_relay(
     fh: FeedHandler,
-    instruments: list,
+    instruments: list[Symbol],
-    oi_by_strikes: dict[str, dict[str, Decimal]],
     from_trio: asyncio.Queue,
     to_trio: trio.abc.SendChannel,
 ) -> None:
-    
-    intrinsic_values: dict[str, dict[str, Decimal]] = {}
-    total_intrinsic_value: Decimal = Decimal('Infinity')
-    max_pain: Decimal = Decimal(0)
-
     async def _trade(
         trade: Trade,  # cryptofeed, NOT ours from `.venues`!
         receipt_timestamp: int,
@@ -857,11 +837,6 @@ async def aio_open_interest_feed_relay(
         Proxy-thru `cryptofeed.FeedHandler` "oi" to `piker`-side.
 
         '''
-        nonlocal intrinsic_values
-        nonlocal oi_by_strikes
-        nonlocal total_intrinsic_value
-        nonlocal max_pain
-
         symbol: Symbol = str_to_cb_sym(oi.symbol)
         piker_sym: str = cb_sym_to_deribit_inst(symbol)
         (
@@ -872,36 +847,13 @@ async def aio_open_interest_feed_relay(
         ) = tuple(
             piker_sym.split('-')
         )
-        open_interest: Decimal = oi.open_interest
+        msg = {
-        oi_by_strikes[f'{strike_price}'][f'{option_type}'] = open_interest
+            'timestamp': oi.timestamp,
-
+            'strike_price': strike_price,
-        if check_if_complete(oi_by_strikes):
+            'option_type': option_type,
-            closes: list[str] = sorted(oi_by_strikes.keys())
+            'open_interest': Decimal(oi.open_interest),
-            for strike in oi_by_strikes:
+        }
-                s: Decimal = Decimal(f'{strike}')
+        to_trio.send_nowait(('oi', msg))
-                call_cash: Decimal = Decimal(0)
-                put_cash: Decimal = Decimal(0)
-                for close in closes:
-                    c: Decimal = Decimal(f'{close}')
-                    call_cash += max(0, (s - c) * oi_by_strikes[f'{close}']['C'])
-                    put_cash += max(0, (c - s) * oi_by_strikes[f'{close}']['P'])
-    
-                intrinsic_values[strike] = {
-                    'C': call_cash,
-                    'P': put_cash,
-                    'total': call_cash + put_cash,
-                }
-
-            for strike in intrinsic_values:
-                if intrinsic_values[f'{strike}']['total'] < total_intrinsic_value:
-                    total_intrinsic_value = intrinsic_values[f'{strike}']['total']
-                    max_pain = strike
-            
-            print('-----------------------------------------------')
-            print(f'expiry date:           {expiry_date}')
-            print(f'max_pain:              {max_pain}')
-            print(f'total intrinsic value: {total_intrinsic_value}')
-            print('-----------------------------------------------')
 
 
     channels = [TRADES, OPEN_INTEREST]
@@ -929,17 +881,8 @@ async def aio_open_interest_feed_relay(
 
 @acm
 async def open_oi_feed(
+    instruments: list[Symbol], 
 ) -> to_asyncio.LinkedTaskChannel:
-    expiry_date: str = '20DEC24'
-    instruments: list[Symbol] = []
-    oi_by_strikes: dict[str, dict[str, Decimal]]
-
-    async with get_client(
-    ) as client:
-        instruments = await client.get_instruments(
-            expiry_date=expiry_date,
-        )
-        oi_by_strikes = client.get_strikes_dict(instruments)
 
     fh: FeedHandler
     first: None
@@ -951,7 +894,6 @@ async def open_oi_feed(
                 aio_open_interest_feed_relay,
                 fh,
                 instruments,
-                oi_by_strikes,
             )
         ) as (first, chan)
     ):
@@ -960,12 +902,18 @@ async def open_oi_feed(
 
 @acm
 async def maybe_open_oi_feed(
+    instruments: list[Symbol], 
 ) -> trio.abc.ReceiveStream:
 
     # TODO: add a predicate to maybe_open_context
     feed: to_asyncio.LinkedTaskChannel
     async with maybe_open_context(
         acm_func=open_oi_feed,
+        kwargs={
+            'instruments': instruments
+        },
+        key=f'{instruments[0]}',
+
     ) as (cache_hit, feed):
         if cache_hit:
             yield broadcast_receiver(feed, 10)