This script contains the necessary logic to calculate the max pain for a given

expiration date.

examples/max_pain.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python
+from decimal import (
+    Decimal,
+)
+import trio
+import tractor
+from datetime import datetime
+from pprint import pformat
+from piker.brokers.deribit.api import (
+    get_client,
+    maybe_open_oi_feed,
+)
+
+def check_if_complete(
+        oi: dict[str, dict[str, Decimal | None]]
+    ) -> bool:
+    return all(
+        oi[strike]['C'] is not None 
+        and
+        oi[strike]['P'] is not None for strike in oi
+    )
+
+
+async def max_pain_daemon(
+) -> None:
+    expiry_date: str = input('Please enter a valid expiration date (7feb25): ').upper()
+    print('Starting little daemon...')
+    instruments: list[Symbol] = []
+    oi_by_strikes: dict[str, dict[str, Decimal]]
+
+    def update_oi_by_strikes(msg: tuple):
+        nonlocal oi_by_strikes
+        if 'oi' == msg[0]:
+            strike_price = msg[1]['strike_price']
+            option_type = msg[1]['option_type']
+            open_interest = msg[1]['open_interest']
+            oi_by_strikes.setdefault(
+                strike_price, {}
+            ).update(
+                {option_type: open_interest}
+            )
+
+    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)
+        call_cash: Decimal = Decimal(0)
+        put_cash: Decimal = Decimal(0)
+        intrinsic_values: dict[str, dict[str, Decimal]] = {}
+        closes: list = sorted(Decimal(close) for close in oi_by_strikes)
+
+        for strike, oi in oi_by_strikes.items():
+            s = Decimal(strike)
+            call_cash = sum(max(0, (s - c) * oi_by_strikes[str(c)]['C']) for c in closes)
+            put_cash = sum(max(0, (c - s) * oi_by_strikes[str(c)]['P']) for c in closes)
+
+            intrinsic_values[strike] = {
+                'C': call_cash,
+                'P': put_cash,
+                'total': call_cash + put_cash,
+            }
+
+            if intrinsic_values[strike]['total'] < total_intrinsic_value:
+                total_intrinsic_value = intrinsic_values[strike]['total']
+                max_pain = s
+
+        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:
+
+            update_oi_by_strikes(msg)
+            if check_if_complete(oi_by_strikes):
+                if 'oi' == msg[0]:
+                    timestamp = msg[1]['timestamp']
+                    max_pain = get_max_pain(oi_by_strikes)
+                    print('-----------------------------------------------')
+                    print(f'timestamp:             {datetime.fromtimestamp(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():
+
+    async with tractor.open_nursery() as n:
+        
+        p: tractor.Portal = await n.start_actor(
+            'max_pain_daemon',
+            enable_modules=[__name__],
+            infect_asyncio=True,
+        )
+        await p.run(max_pain_daemon)
+
+if __name__ == '__main__':
+    trio.run(main)