piker/max_pain.py

#!/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 = '13DEC24'
    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)