Python Type Hint Finance Use Case: Implied Volatility and Greeks of Index 'At The Money' Options: Part 2

    	
            
import refinitiv.data as rd  # This is LSEG's Data and Analytics' API wrapper, called the Refinitiv Data Library for Python. You can update this library with the comand `!pip install refinitiv-data --upgrade`
from refinitiv.data.content import historical_pricing  # We will use this Python Class in `rd` to show the Implied Volatility data already available before our work.
from refinitiv.data.content import search  # We will use this Python Class in `rd` to fid the instrument we are after, closest to At The Money.
import refinitiv.data.content.ipa.financial_contracts as rdf  # We're going to need thtis to use the content layer of the RD library and the calculators of greeks and Impl Volat in Instrument Pricing Analytics (IPA) and Exchange Traded Instruments (ETI)
from refinitiv.data.content.ipa.financial_contracts import option  # We're going to need thtis to use the content layer of the RD library and the calculators of greeks and Impl Volat in IPA & ETI
 
import numpy as np  # We need `numpy` for mathematical and array manipilations.
import pandas as pd  # We need `pandas` for datafame and array manipilations.
import calendar  # We use `calendar` to identify holidays and maturity dates of intruments of interest.
import pytz  # We use `pytz` to manipulate time values aiding `calendar` library. to import its types, you might need to run `!python3 -m pip install types-pytz`
import pandas_market_calendars as mcal  # Used to identify holidays. See `https://github.com/rsheftel/pandas_market_calendars/blob/master/examples/usage.ipynb` for info on this market calendar library
from datetime import datetime, timedelta, timezone  # We use these to manipulate time values
from dateutil.relativedelta import relativedelta  # We use `relativedelta` to manipulate time values aiding `calendar` library.
import requests  # We'll need this to send requests to servers vie a the delivery layer - more on that below
 
# `plotly` is a library used to render interactive graphs:
import plotly.graph_objects as go
import plotly.express as px  # This is just to see the implied vol graph when that field is available
import matplotlib.pyplot as plt  # We use `matplotlib` to just in case users do not have an environment suited to `plotly`.
from IPython.display import clear_output, display  # We use `clear_output` for users who wish to loop graph production on a regular basis. We'll use this to `display` data (e.g.: pandas data-frames).
from plotly import subplots
import plotly

        
        Copy
    

    	
            
try:
    rd.open_session(
        name="desktop.workspace4",  # "platform.rdph", "desktop.workspace4"
        config_name="C:/Example.DataLibrary.Python-main/Configuration/refinitiv-data.config.json")
    print("We're on 'desktop.workspace4' Session")
except:
    rd.open_session()
    print("We're on Desktop Session")
# For more info on the session, use `rd.get_config().as_dict()`

        
        Copy
    

    	
            
tst = rd.content.historical_pricing.summaries.Definition(
    universe='STXE42500D3.EX^D23', start='2021-08-29', end='2023-04-21', interval='PT10M', fields=['TRDPRC_1', 'SETTLE', 'BID', 'ASK']).get_data().data.df

        
        Copy
    

    	
            tst
        
        Copy
    

    	
            
for i, j in zip(
    ['refinitiv.data', 'numpy', 'pandas', 'pandas_market_calendars' 'pytz', 'requests', 'plotly'],
    [rd, np, pd, mcal, pytz, requests, plotly]):
    print(f"{i} used in this code is version {j.__version__}")

        
        Copy
    

    	
            !python -V
        
        Copy
    

    	
            
calc_time, underlying = "2022-04-01", ".STOXX50E"
calc_time_datetime = datetime.strptime(calc_time, '%Y-%m-%d')
current_underlying_prc = rd.get_history(
    universe=[underlying],
    start=calc_time,  # , end: "OptDateTime"=None
    fields=["TRDPRC_1"],
    interval="tick").iloc[-1][0]
current_underlying_prc

        
        Copy
    

    	
            
if underlying == ".STOXX50E":
    exchange, exchangeRIC, mcal_cal = "EUX", "STX", "EUREX"
elif underlying == ".SPX":
    exchange, exchangeRIC, mcal_cal = "OPQ", "SPX", "CBOE_Futures"
exchange, exchangeRIC, mcal_cal

        
        Copy
    

    	
            
def Get_exp_dates(year, days=True, mcal_get_calendar='EUREX'):
 
    # get CBOE market holidays
    EUREXCal = mcal.get_calendar(mcal_get_calendar)
    holidays = EUREXCal.holidays().holidays
 
    # set calendar starting from Saturday
    c = calendar.Calendar(firstweekday=calendar.SATURDAY)
 
    # get the 3rd Friday of each month
    exp_dates = {}
    for i in range(1, 13):
        monthcal = c.monthdatescalendar(year, i)
        date = monthcal[2][-1]
        # check if found date is an holiday and get the previous date if it is
        if date in holidays:
            date = date + timedelta(-1)
        # append the date to the dictionary
        if year in exp_dates:
            ### Changed from original code from here on by Jonathan Legrand on 2022-10-11
            if days: exp_dates[year].append(date.day)
            else: exp_dates[year].append(date)
        else:
            if days: exp_dates[year] = [date.day]
            else: exp_dates[year] = [date]
    return exp_dates

        
        Copy
    

    	
            
calc_full_dates_time = Get_exp_dates(
    year=2022, days=False,
    mcal_get_calendar=mcal_cal)
calc_full_dates_time

        
        Copy
    

    	
            
calc_full_dates_time_datetime = [
    datetime(i.year, i.month, i.day)
    for i in calc_full_dates_time[list(calc_full_dates_time.keys())[0]]]
calc_full_dates_time_datetime

        
        Copy
    

    	
            x = 15
        
        Copy
    

    	
            
expiry_date_of_intrst = [i for i in calc_full_dates_time_datetime
                    if i > calc_time_datetime + relativedelta(days=x)][0]
expiry_date_of_intrst

        
        Copy
    

    	
            
!python -Vdef Get_exp_month(exp_date, opt_type):
 
    # define option expiration identifiers
    ident = {
        '1':  {'exp': 'A', 'C': 'A', 'P': 'M'},
        '2':  {'exp': 'B', 'C': 'B', 'P': 'N'},
        '3':  {'exp': 'C', 'C': 'C', 'P': 'O'},
        '4':  {'exp': 'D', 'C': 'D', 'P': 'P'},
        '5':  {'exp': 'E', 'C': 'E', 'P': 'Q'},
        '6':  {'exp': 'F', 'C': 'F', 'P': 'R'},
        '7':  {'exp': 'G', 'C': 'G', 'P': 'S'},
        '8':  {'exp': 'H', 'C': 'H', 'P': 'T'},
        '9':  {'exp': 'I', 'C': 'I', 'P': 'U'},
        '10': {'exp': 'J', 'C': 'J', 'P': 'V'},
        '11': {'exp': 'K', 'C': 'K', 'P': 'W'},
        '12': {'exp': 'L', 'C': 'L', 'P': 'X'}}
 
    # get expiration month code for a month
    if opt_type.upper() == 'C':
        exp_month = ident[str(exp_date.month)]['C']
    elif opt_type.upper() == 'P':
        exp_month = ident[str(exp_date.month)]['P']
 
    return ident, exp_month

        
        Copy
    

    	
            
def Check_RIC(
        ric,
        maturity,
        ident,
        interval=rd.content.historical_pricing.Intervals.DAILY):
 
    exp_date = pd.Timestamp(maturity)
 
    # get start and end date for get_historical_price_summaries
    # query (take current date minus 90 days period)
    sdate = (datetime.now() - timedelta(90)).strftime('%Y-%m-%d')
    edate = datetime.now().strftime('%Y-%m-%d')
 
    # check if option is matured. If yes, add expiration syntax and recalculate
    # start and end date of the query (take expiration day minus 90 days period)
    if pd.Timestamp(maturity) < datetime.now():
        ric = ric + '^' + ident[str(exp_date.month)]['exp'] + str(exp_date.year)[-2:]
        sdate = (exp_date - timedelta(90)).strftime('%Y-%m-%d')
        edate = exp_date.strftime('%Y-%m-%d')
 
    # Now things are getting tricky.
    # Certain Expiered Options do not have 'TRDPRC_1' data historically. Some don't have 'SETTLE'. Some have both...
    # The below should capture 'SETTLE' when it is available, then 'TRDPRC_1' if not, then MID (calculated with 'ASK' and 'BID') otherwise.
    prices = rd.content.historical_pricing.summaries.Definition(
            universe=ric,
            start=sdate,
            end=edate,
            interval=interval,
            fields=['TRDPRC_1', 'SETTLE', 'BID', 'ASK']
            ).get_data().data.df
    pr_cnt = prices.count()
    if pr_cnt.TRDPRC_1 > 0:
        prices = pd.DataFrame(
            data={'TRDPRC_1': prices.TRDPRC_1}).dropna()
    elif pr_cnt.SETTLE > 0:
        prices = pd.DataFrame(
            data={'SETTLE': prices.SETTLE}).dropna()
    elif pr_cnt.BID > 0:
        prices = pd.DataFrame(
            data={'MID': (prices.BID + prices.ASK)/2}).dropna()
    prices.columns.name = ric
 
    return ric, prices

        
        Copy
    

    	
            
def Get_RIC_eurex(asset, maturity, strike, opt_type, interval='PT10M'):
    exp_date = pd.Timestamp(maturity)
 
    if asset[0] == '.':
        asset_name = asset[1:]
        if asset_name == 'FTSE':
            asset_name = 'OTUK'
        elif asset_name == 'SSMI':
            asset_name = 'OSMI'
        elif asset_name == 'GDAXI':
            asset_name = 'GDAX'
        elif asset_name == 'ATX':
            asset_name = 'FATXA'
        elif asset_name == 'STOXX50E':
            asset_name = 'STXE'
    else:
        asset_name = asset.split('.')[0]
 
    ident, exp_month = Get_exp_month(
        exp_date=exp_date, opt_type=opt_type)
 
    if type(strike) == float:
        int_part = int(strike)
        dec_part = str(str(strike).split('.')[1])[0]
    else:
        int_part = int(strike)
        dec_part = '0'
 
    if len(str(int(strike))) == 1:
        strike_ric = '0' + str(int_part) + dec_part
    else:
        strike_ric = str(int_part) + dec_part
 
    possible_rics = []
    generations = ['', 'a', 'b', 'c', 'd']
    for gen in generations:
        ric = asset_name + strike_ric + gen + exp_month + str(exp_date.year)[-1:] + '.EX'
        ric, prices = Check_RIC(ric=ric, maturity=maturity, ident=ident,
                                interval=rd.content.historical_pricing.Intervals.DAILY)
        if prices is not None:
            return ric, prices
        else:
            possible_rics.append(ric)
    print(f'Here is a list of possible RICs {possible_rics}, however we could not find any prices for those!')
    return ric, prices

        
        Copy
    

    	
            print(f"{current_underlying_prc} as int: {int(round(current_underlying_prc, -2))}")
        
        Copy
    

    	
            
instrument, instrument_pr = Get_RIC_eurex(
    asset='.STOXX50E', opt_type='P',
    maturity=expiry_date_of_intrst.strftime('%Y-%m-%d'),
    strike=int(round(current_underlying_prc, -2)))

        
        Copy
    

    	
            instrument
        
        Copy
    

    	
            instrument_pr
        
        Copy
    

    	
            # pip install --trusted-host files.pythonhosted.org nb_mypy # This line can be used to install `nb_mypy`.
        
        Copy
    

    	
            
import nb_mypy  # !pip3 install nb_mypy --trusted-host pypi.org # https://pypi.org/project/nb-mypy/ # https://gitlab.tue.nl/jupyter-projects/nb_mypy/-/blob/master/Nb_Mypy.ipynb
%load_ext nb_mypy
%reload_ext nb_mypy
%nb_mypy On
%nb_mypy DebugOff

        
        Copy
    

    	
            
def myfunc(myparam: int) -> None:
    print(myparam)
 
myfunc('test')

        
        Copy
    

    	
            
import refinitiv.data as rd  # This is LSEG's Data and Analytics' API wrapper, called the Refinitiv Data Library for Python. You can update this library with the comand `!pip install refinitiv-data --upgrade`
from refinitiv.data.content import historical_pricing  # We will use this Python Class in `rd` to show the Implied Volatility data already available before our work.
from refinitiv.data.content import search  # We will use this Python Class in `rd` to fid the instrument we are after, closest to At The Money.
import refinitiv.data.content.ipa.financial_contracts as rdf  # We're going to need this to use the content layer of the RD library and the calculators of greeks and Impl Volat in Instrument Pricing Analytics (IPA) and Exchange Traded Instruments (ETI)
from refinitiv.data.content.ipa.financial_contracts import option  # We're going to need thtis to use the content layer of the RD library and the calculators of greeks and Impl Volat in IPA & ETI
 
import numpy as np  # We need `numpy` for mathematical and array manipilations.
import pandas as pd  # We need `pandas` for datafame and array manipilations.
import calendar  # We use `calendar` to identify holidays and maturity dates of intruments of interest.
import pytz  # We use `pytz` to manipulate time values aiding `calendar` library. to import its types, you might need to run `!python3 -m pip install types-pytz`. To get the Type Hints for this library, try `pip install types-pytz`.
import pandas_market_calendars as mcal  # Used to identify holidays. See `https://github.com/rsheftel/pandas_market_calendars/blob/master/examples/usage.ipynb` for info on this market calendar library
from datetime import datetime, date, timedelta, timezone  # We use these to manipulate time values
from dateutil.relativedelta import relativedelta  # We use `relativedelta` to manipulate time values aiding `calendar` library. May also need `pip install types-python-dateutil`.
import requests  # We'll need this to send requests to servers vie a the delivery layer - more on that below. Might also need `pip install types-requests`.
 
# `plotly` is a library used to render interactive graphs:
import plotly.graph_objects as go
import plotly.express as px  # This is just to see the implied vol graph when that field is available
import matplotlib.pyplot as plt  # We use `matplotlib` to just in case users do not have an environment suited to `plotly`.
from IPython.display import clear_output, display  # We use `clear_output` for users who wish to loop graph production on a regular basis. We'll use this to `display` data (e.g.: pandas data-frames).
from plotly import subplots
import plotly
 
from typing import Generator, Any
from types import ModuleType  # FrameType, TracebackType

        
        Copy
    

    	
            
try:
    rd.open_session(
        name="desktop.workspace",
        config_name="C:/Example.DataLibrary.Python-main/Configuration/refinitiv-data.config.json")
except:
    rd.open_session()
# For more info on the session, use `rd.get_config().as_dict()`

        
        Copy
    

    	
            
i_str: str
i_ModType: ModuleType
 
for i_str, i_ModType in zip(
    ['refinitiv.data', 'numpy', 'pandas', 'pandas_market_calendars' 'pytz', 'requests', 'plotly'],
    [rd, np, pd, mcal, pytz, requests, plotly]):
    print(f"{i_str} used in this code is version {i_ModType.__version__}")

        
        Copy
    

    	
            
calc_time: str = "2022-04-01"
underlying: str = ".STOXX50E"
 
calc_time_datetime: datetime = datetime.strptime(calc_time, '%Y-%m-%d')
current_underlying_prc: float = rd.get_history(
    universe=[underlying],
    start=calc_time,  # , end: "OptDateTime"=None
    fields=["TRDPRC_1"],
    interval="tick").iloc[-1][0]
 
current_underlying_prc

        
        Copy
    

    	
            
exchange: str
exchangeRIC: str
mcal_cal: str
if underlying == ".STOXX50E":
    exchange, exchangeRIC, mcal_cal = "EUX", "STX", "EUREX"
elif underlying == ".SPX":
    exchange, exchangeRIC, mcal_cal = "OPQ", "SPX", "CBOE_Futures"
exchange, exchangeRIC, mcal_cal

        
        Copy
    

    	
            
def Get_exp_dates(year: int,
                  days: bool = True,
                  mcal_get_calendar: str = 'EUREX'
                  ) -> dict[int, list[Any]]:
 
    # get CBOE market holidays
    holidays: tuple[date] = mcal.get_calendar(mcal_get_calendar).holidays().holidays
 
    # set calendar starting from Saturday
    c: calendar.Calendar = calendar.Calendar(firstweekday=calendar.SATURDAY)
 
    # get the 3rd Friday of each month
    exp_dates: dict[int, list[Any]]
    _date: date
    i_int: int
 
    _date = c.monthdatescalendar(year, 1)[2][-1]
    if _date in holidays: # check if found date is an holiday and get the previous date if it is
        _date = _date + timedelta(-1)
    if days:
        exp_dates = {year:[_date.day]}
    else:
        exp_dates = {year: [_date]}
    for i_int in range(2, 13):
        _date = c.monthdatescalendar(year, i_int)[2][-1]
        if _date in holidays: # check if found date is an holiday and get the previous date if it is
            _date = _date + timedelta(-1)
        # append the date to the dictionary
        if days:
            exp_dates[year].append(_date.day)
        else:
            exp_dates[year].append(_date)
 
    return exp_dates

        
        Copy
    

    	
            Get_exp_dates.__annotations__
        
        Copy
    

    	
            
calc_full_dates_time: dict[int, list[Any]] = Get_exp_dates(
    year=2022, days=False,
    mcal_get_calendar=mcal_cal)

        
        Copy
    

    	
            calc_full_dates_time[list(calc_full_dates_time.keys())[0]]
        
        Copy
    

    	
            
x: int = 15
expiry_date_of_intrst: datetime = [
    datetime(i_date.year, i_date.month, i_date.day)
    for i_date in calc_full_dates_time[list(calc_full_dates_time.keys())[0]]
    if datetime(i_date.year, i_date.month, i_date.day) > calc_time_datetime + relativedelta(days=x)][0]
expiry_date_of_intrst

        
        Copy
    

    	
            
def Get_exp_month(
    exp_date: datetime,
    opt_type: str
    ) -> tuple[dict[int, dict[str, str]], str]:
 
    # define option expiration identifiers
    ident: dict[int, dict[str, str]] = {
        1:  {'exp': 'A', 'C': 'A', 'P': 'M'},
        2:  {'exp': 'B', 'C': 'B', 'P': 'N'},
        3:  {'exp': 'C', 'C': 'C', 'P': 'O'},
        4:  {'exp': 'D', 'C': 'D', 'P': 'P'},
        5:  {'exp': 'E', 'C': 'E', 'P': 'Q'},
        6:  {'exp': 'F', 'C': 'F', 'P': 'R'},
        7:  {'exp': 'G', 'C': 'G', 'P': 'S'},
        8:  {'exp': 'H', 'C': 'H', 'P': 'T'},
        9:  {'exp': 'I', 'C': 'I', 'P': 'U'},
        10: {'exp': 'J', 'C': 'J', 'P': 'V'},
        11: {'exp': 'K', 'C': 'K', 'P': 'W'},
        12: {'exp': 'L', 'C': 'L', 'P': 'X'}}
 
    # get expiration month code for a month
    exp_month: str
    if opt_type.upper() == 'C' or opt_type.upper() == 'CALL':
        exp_month = ident[int(exp_date.month)]['C']
    elif opt_type.upper() == 'P' or opt_type.upper() == 'PUT':
        exp_month = ident[int(exp_date.month)]['P']
 
    return ident, exp_month

        
        Copy
    

    	
            
Get_exp_month(
    exp_date=datetime(2022, 5, 20, 0, 0),
    opt_type='P'
)

        
        Copy
    

    	
            datetime.strptime(expiry_date_of_intrst.strftime('%Y-%m-%d'), '%Y-%m-%d')
        
        Copy
    

    	
            (datetime.strptime(expiry_date_of_intrst.strftime('%Y-%m-%d'), '%Y-%m-%d')  - timedelta(90)).strftime('%Y-%m-%d')
        
        Copy
    

    	
            
def Check_RIC(
        ric: str,
        maturity: str,
        ident: dict[int, dict[str, str]],
        interval: rd.content.historical_pricing.Intervals = rd.content.historical_pricing.Intervals.DAILY
        ) -> tuple[str, pd.DataFrame]:
 
    exp_date: pd.Timestamp = pd.Timestamp(maturity)
 
    # get start and end date for get_historical_price_summaries
    # query (take maturity date minus 90 days period)
    sdate: str = (datetime.strptime(maturity, '%Y-%m-%d') - timedelta(90)).strftime('%Y-%m-%d')
    edate: str = maturity
 
    # check if option is matured. If yes, add expiration syntax and recalculate
    # start and end date of the query (take expiration day minus 90 days period)
    if pd.Timestamp(maturity) < datetime.now():
        ric = ric + '^' + ident[int(exp_date.month)]['exp'] + str(exp_date.year)[-2:]
        sdate = (exp_date - timedelta(90)).strftime('%Y-%m-%d')
        edate = exp_date.strftime('%Y-%m-%d')
 
    # Now things are getting tricky.
    # Certain Expiered Options do not have 'TRDPRC_1' data historically. Some don't have 'SETTLE'. Some have both...
    # The below should capture 'SETTLE' when it is available, then 'TRDPRC_1' if not, then MID (calculated with 'ASK' and 'BID') otherwise.
    prices: pd.DataFrame | None
    prices = rd.content.historical_pricing.summaries.Definition(
            universe=ric,
            start=sdate,
            end=edate,
            interval=interval,
            fields=['TRDPRC_1', 'SETTLE', 'BID', 'ASK']
            ).get_data().data.df
    pr_cnt: pd.Series = prices.count()
    if pr_cnt.TRDPRC_1 > 0:
        prices = pd.DataFrame(
            data={'TRDPRC_1': prices.TRDPRC_1}).dropna()
    elif pr_cnt.SETTLE > 0:
        prices = pd.DataFrame(
            data={'SETTLE': prices.SETTLE}).dropna()
    elif pr_cnt.BID > 0:
        prices = pd.DataFrame(
            data={'MID': (prices.BID + prices.ASK)/2}).dropna()
    prices.columns.name = ric
 
    return ric, prices

        
        Copy
    

    	
            
def Get_RIC_eurex(
        asset: str,
        maturity: str,
        strike: int,
        opt_type: str,
        interval: rd.content.historical_pricing.Intervals = rd.content.historical_pricing.Intervals.DAILY
        ) -> tuple[str, pd.DataFrame]:
    exp_date = pd.Timestamp(maturity)
 
    asset_name: str
    ident: dict[int, dict[str, str]]
    exp_month: str
    dec_part: str
    strike_ric: str
 
    if asset[0] == '.':
        asset_name = asset[1:]
        if asset_name == 'FTSE':
            asset_name = 'OTUK'
        elif asset_name == 'SSMI':
            asset_name = 'OSMI'
        elif asset_name == 'GDAXI':
            asset_name = 'GDAX'
        elif asset_name == 'ATX':
            asset_name = 'FATXA'
        elif asset_name == 'STOXX50E':
            asset_name = 'STXE'
    else:
        asset_name = asset.split('.')[0]
 
    ident, exp_month = Get_exp_month(
        exp_date=exp_date, opt_type=opt_type)
 
    if type(strike) == float:
        int_part = int(strike)
        dec_part = str(str(strike).split('.')[1])[0]
    else:
        int_part = int(strike)
        dec_part = '0'
 
    if len(str(int(strike))) == 1:
        strike_ric = '0' + str(int_part) + dec_part
    else:
        strike_ric = str(int_part) + dec_part
 
    possible_rics: list = []
    generations: list[str] = ['', 'a', 'b', 'c', 'd']
    prices: pd.DataFrame
    for gen in generations:
        ric: str = asset_name + strike_ric + gen + exp_month + str(exp_date.year)[-1:] + '.EX'
        ric, prices = Check_RIC(ric=ric, maturity=maturity, ident=ident, interval=interval)
        if prices is not None:
            return ric, prices
        else:
            possible_rics.append(ric)
    print(f'Here is a list of possible RICs {possible_rics}, however we could not find any prices for those!')
    return ric, prices

        
        Copy
    

    	
            
instrument: str
instrument_pr: pd.DataFrame
instrument, instrument_pr = Get_RIC_eurex(
    asset='.STOXX50E', opt_type='P',
    maturity=expiry_date_of_intrst.strftime('%Y-%m-%d'),
    strike=int(round(current_underlying_prc, -2)))

        
        Copy
    

    	
            instrument
        
        Copy
    

    	
            instrument_pr
        
        Copy
    

    	
            
HSI_underlying_RIC: str = '.HSI'
HSI_curr: pd.DataFrame = rd.get_data(
    universe=HSI_underlying_RIC,
    fields=["CF_CURR"])
HSI_curr

        
        Copy
    

    	
            
time_of_calc3: str = "2023-02-02"
HSI_test0 = OptionRIC(
    maturity=time_of_calc3,
    strike=20200,  # could be: `int(round(rd.get_history(universe=[SPX_underlying_RIC], start=time_of_calc3, fields=["TRDPRC_1"], interval="tick").iloc[-1][0], -2))` but this would get the price right now, which may not be appropriate for a 'past'/expired option.
    opt_type='P',
    asset=HSI_underlying_RIC,
    debug=False)
HSI_test: dict[str, list[dict[str, pd.DataFrame]] | list] | pd.DataFrame
HSI_test = HSI_test0.Construct_RIC()
HSI_test

        
        Copy
    

    	
            
HSI_test_df: pd.DataFrame
HSI_test_df = HSI_test['valid_ric'][0][list(HSI_test['valid_ric'][0].keys())[0]]
HSI_test_df.head()

        
        Copy
    

    	
            
# # 1st: get the current underlying price
time_of_calc4: str = '2022-02-10'
SPX_underlying_RIC: str = ".SPX"
# time_of_calc_datetime3: date = datetime.strptime(time_of_calc4, '%Y-%m-%d')
current_underlying_SPX_prc: float = rd.get_history(
    universe=[SPX_underlying_RIC],
    start=time_of_calc4,  # , end: "OptDateTime"=None
    fields=["TRDPRC_1"],
    interval="tick").iloc[-1][0]
current_underlying_SPX_prc

        
        Copy
    

    	
            
# # 2nd: get data via the `OptionRIC` function we created
SPX_test: dict[str, list[dict[str, pd.DataFrame]] | list] | pd.DataFrame = OptionRIC(
    maturity='2022-07-15',
    strike=int(round(current_underlying_SPX_prc, -2)),
    opt_type='P',
    asset=SPX_underlying_RIC,
    debug=False,
    interval=rd.content.historical_pricing.Intervals.DAILY
    ).Construct_RIC()
SPX_test

        
        Copy
    

    	
            
# # test df for verification:
# tdf: pd.DataFrame = rd.content.historical_pricing.summaries.Definition(
#     universe="SPXs152245000.U^G22",
#     start='2020-11-22T09:00:00.000',
#     end="2022-07-15T20:40:00.000",
#     fields=['CLOSE', 'TRDPRC_1', 'BID', 'ASK'],
#     interval=rd.content.historical_pricing.Intervals.DAILY).get_data().data.df
# tdf

        
        Copy
    

    	
            
# # Note that the output is a dictionary:
SPX_test['valid_ric'][0].keys()

        
        Copy
    

    	
            
# # You can find the keys of this dictionary:
list(SPX_test['valid_ric'][0].keys())[0]

        
        Copy
    

    	
            
# # With dictionary keys, you can itterate through what `OptionRIC` found:
SPX_test['valid_ric'][0][list(SPX_test['valid_ric'][0].keys())[0]]

        
        Copy
    

    	
            
# Now we will try and find the strike price for the option found.
# If we were to use the logic above, in the function `OptionRIC().Get_strike()`, we would find a list of all the possible prices for options on this underlying, which is too large of a group.
# We will use the name of the outputed option, which includes the strike:
import re  # native Python library that allows us to manipulate strings
hist_opt_found_strk_pr: str = re.findall(
    '(\d+|[A-Za-z]+)',  # This will split the string out of its numerical and non-numerical characters.
    list(HSI_test['valid_ric'][0].keys())[0])[1] #`[1]` here skips through 'HSI' and to the numbers.
hist_opt_found_strk_pr

        
        Copy
    

    	
            
def hk_rf_f():
    return 100 - rd.get_history(
    universe=['HK3MT=RR'],  # other examples could include: `HK10YGB=EODF`, `HKGOV3MZ=R`, `HK3MT=RR`
    fields=['TR.MIDPRICE'],
    start=HSI_test_df.index[0].strftime('%Y-%m-%d %H:%M:%S'),
    end=HSI_test_df.index[-1].strftime('%Y-%m-%d %H:%M:%S'))

        
        Copy
    

    	
            
# try:
hk_rf: pd.DataFrame = 100 - rd.get_history(
universe=['HK3MT=RR'],  # other examples could include: `HK10YGB=EODF`, `HKGOV3MZ=R`, `HK3MT=RR`
fields=['TR.MIDPRICE'],
start=HSI_test_df.index[0].strftime('%Y-%m-%d %H:%M:%S'),
end=HSI_test_df.index[-1].strftime('%Y-%m-%d %H:%M:%S'))  # .iloc[::-1]  # `.iloc[::-1]` is here so that the resulting data-frame is the same order as `HSI_test_df1` so we can merge them later
# except RDError:
#     hk_rf = 100 - rd.get_history(
#     universe=['HK3MT=RR'],
#     fields=['TR.MIDPRICE'],
#     start=HSI_test_df.index[0].strftime('%Y-%m-%d %H:%M:%S'),
#     end=HSI_test_df.index[-1].strftime('%Y-%m-%d %H:%M:%S'))
hk_rf.head(3)

        
        Copy
    

    	
            
def Insert_daily_data(
        intra_day_df: pd.DataFrame,
        intra_day_col: str,
        daily_df: pd.DataFrame,
        daily_df_col: str
        ) -> pd.DataFrame:
    df: pd.DataFrame = intra_day_df.copy()
    df[intra_day_col] = [pd.NA for i_int in range(len(df))]
    i_int: int; j_int: int
    for i_int in range(len(df)):
        for j_int in range(len(daily_df)):
            if daily_df.index[j_int].strftime('%Y-%m-%d') in df.index[i_int].strftime('%Y-%m-%d'):
                df[intra_day_col].iloc[i_int] = daily_df[daily_df_col].iloc[j_int]
    return df

        
        Copy
    

    	
            
HSI_test_df1: pd.DataFrame = Insert_daily_data(
    intra_day_df=HSI_test_df, intra_day_col="RfRatePrct",
    daily_df=hk_rf, daily_df_col="Mid Price")
# Rename the field name 'TRDPRC_1' with something that makes a litle more sense, like 'OptionPrice'
HSI_test_df1: pd.DataFrame = HSI_test_df1.rename(
    columns={"TRDPRC_1": "OptionPrice"})
HSI_test_df1.head()

        
        Copy
    

    	
            
# Now let's get the underlying price:
hist_HSI_undrlying_pr: pd.DataFrame = rd.get_history(
    universe=[HSI_underlying_RIC],
    fields=["TRDPRC_1"],
    # interval="1D",
    start=HSI_test_df.index[0].strftime('%Y-%m-%d'),
    end=HSI_test_df.index[-1].strftime('%Y-%m-%d'))  # .iloc[::-1]  # `.iloc[::-1]` is here so that the resulting data-frame is the same order as `HSI_test_df1` so we can merge them later
hist_HSI_undrlying_pr.head(2)

        
        Copy
    

    	
            
# Now let's put it all together:
HSI_test_df2: pd.DataFrame = Insert_daily_data(
    intra_day_df=HSI_test_df1, intra_day_col="UndrlyingPr",
    daily_df=hist_HSI_undrlying_pr, daily_df_col="TRDPRC_1")
HSI_test_df2

        
        Copy
    

    	
            
HSI_test2_exp_date: str = HSI_test0.maturity.strftime('%Y-%m-%d')
HSI_test2_exp_date

        
        Copy
    

    	
            
hist_daily_universe_l: list[option._definition.Definition] = [
    option.Definition( # You can find details on this with `help(option.Definition)`.
        # instrument_tag="Option",
        # instrument_code='STXE42000D3.EX',  # 'STXE42000D3.EX' #  'HSI19300N3.HF^B23',  # list(HSI_test2['valid_ric'][0].keys())[0], # # `instrument_code` is ambiguous here because we ought to use our expired option RIC, but when we use it we automatically get ann NaNs back. Putting in a live option's RIC resolves the problem, but we're not investigating a live option... So you can simply not define it; which is what we're doing here.
        strike=float(hist_opt_found_strk_pr),
        buy_sell='Buy',
        call_put='Call',
        exercise_style="EURO", # optional, str # AMER, EURO or BERM. AMER: the owner has the right to exercise on any date before the option expires, EURO: the owner has the right to exercise only on EndDate, BERM: the owner has the right to exercise on any of several specified dates before the option expires.
        end_date=datetime.strptime(HSI_test2_exp_date, '%Y-%m-%d').strftime('%Y-%m-%dT%H:%M:%SZ'), # optional, date-time # The maturity or expiry date of the instrument. The value is expressed in ISO 8601 format: YYYY-MM-DDT[hh]:[mm]:[ss]Z (e.g., '2021-01-01T00:00:00Z'). Mandatory for OTC ETI options and FX options (if tenor is not defined).,
        lot_size=1.0, # optional, double (i.e.: float) # The number of the underlying asset unit on which the option is written. It can be overriden only for Commodity options. If instrumentCode of listed ETI option is defined the value comes from the instrument reference data. The default value is '1' for OTC ETI options.
        deal_contract=1, # optional, int # The number of contracts bought or sold in the deal. The default value is '1'.
        time_zone_offset=0, # optional, int # The offset in minutes between UTC and the time of the exchange where the contract is traded. No default value applies.
        underlying_type=option.UnderlyingType.ETI,
        underlying_definition=option.EtiUnderlyingDefinition(
            instrument_code=HSI_underlying_RIC),
        # tenor=str((HSI_test1.maturity - HSI_test_start).days/365),  # Expecting this to `YearsToExpiry` which is in 'DaysToExpiry / 365'.
        # notional_ccy='HKD',
        # notional_amount=None,
        # asian_definition=None,
        # barrier_definition=None,
        # binary_definition=None,
        # double_barrier_definition=None,
        # double_binary_definition=None,
        # dual_currency_definition=None,
        # forward_start_definition=None,
        # underlying_definition=None,
        # delivery_date=HSI_test1.maturity.strftime('%Y-%m-%d'),
        # cbbc_definition=None,
        # double_barriers_definition=None,
        # end_date_time=None,
        # offset=None,
        # extended_params=None,
        pricing_parameters=option.PricingParameters( # mode on this with `help(option.PricingParameters)`
            valuation_date=HSI_test_df2.dropna().index[i_int].strftime('%Y-%m-%dT%H:%M:%SZ'), # The date at which the instrument is valued. the value is expressed in iso 8601 format: yyyy-mm-ddt[hh]:[mm]:[ss]z (e.g., '2021-01-01t00:00:00z'). by default, marketdatadate is used. if marketdatadate is not specified, the default value is today.
            report_ccy='HKD',
            market_value_in_deal_ccy=float(HSI_test_df2.dropna()['OptionPrice'][i_int]),
            # market_value_in_report_ccy=None,
            pricing_model_type='BlackScholes',
            # dividend_type=None,  # APPARENTLY NOT APPLICABLE IN IPA/QA IN RD LIB. None, 'ForecastTable', 'HistoricalYield', 'ForecastYield', 'ImpliedYield', or 'ImpliedTable'.
            # dividend_yield_percent=None,
            # volatility_percent=None,  # The degree of the underlying asset's price variations over a specified time period, used for the option pricing. the value is expressed in percentages. it is used to compute marketvalueindealccy.if marketvalueindealccy is defined, volatilitypercent is not taken into account. optional. by default, it is computed from marketvalueindealccy. if volsurface fails to return a volatility, it defaults to '20'.
            risk_free_rate_percent=float(HSI_test_df2.dropna()['RfRatePrct'][i_int]),
            underlying_price=float(HSI_test_df2.dropna()['UndrlyingPr'][i_int]),
            volatility_type='Implied', # 'option.OptionVolatilityType.IMPLIED,  # option.OptionVolatilityType.IMPLIED, 'Implied'
            option_price_side='Mid', # option.PriceSide.LAST,  # 'bid', 'ask', 'mid','last', option.PriceSide.LAST
            underlying_time_stamp='Close', # option.TimeStamp.SETTLE,  # 'Close', 'Default',
            # underlying_price_side='Last' # option.PriceSide.LAST
            # volatility_model=option.VolatilityModel.SVI
        ))
    for i_int in range(len(HSI_test_df2.dropna()))]

        
        Copy
    

    	
            
def Chunks(lst, n) -> Generator[list[Any], None, None]:
    """Yield successive n-sized chunks from lst."""
    i_int: int
    for i_int in range(0, len(lst), n):
        yield lst[i_int:i_int + n]

        
        Copy
    

    	
            request_fields: list[str] = ['ErrorMessage', 'AverageSoFar', 'AverageType', 'BarrierLevel', 'BarrierType', 'BreakEvenDeltaAmountInDealCcy', 'BreakEvenDeltaAmountInReportCcy', 'BreakEvenPriceInDealCcy', 'BreakEvenPriceInReportCcy', 'CallPut', 'CbbcOptionType', 'CbbcType', 'CharmAmountInDealCcy', 'CharmAmountInReportCcy', 'ColorAmountInDealCcy', 'ColorAmountInReportCcy', 'ConversionRatio', 'DailyVolatility', 'DailyVolatilityPercent', 'DaysToExpiry', 'DealCcy', 'DeltaAmountInDealCcy', 'DeltaAmountInReportCcy', 'DeltaExposureInDealCcy', 'DeltaExposureInReportCcy', 'DeltaHedgePositionInDealCcy', 'DeltaHedgePositionInReportCcy', 'DeltaPercent', 'DividendType', 'DividendYieldPercent', 'DvegaDtimeAmountInDealCcy', 'DvegaDtimeAmountInReportCcy', 'EndDate', 'ExerciseStyle', 'FixingCalendar', 'FixingDateArray', 'FixingEndDate', 'FixingFrequency', 'FixingNumbers', 'FixingStartDate', 'ForecastDividendYieldPercent', 'GammaAmountInDealCcy', 'GammaAmountInReportCcy', 'GammaPercent', 'Gearing', 'HedgeRatio', 'InstrumentCode', 'InstrumentDescription', 'InstrumentTag', 'Leverage', 'LotSize', 'LotsUnits', 'MarketDataDate', 'MarketValueInDealCcy', 'MoneynessAmountInDealCcy', 'MoneynessAmountInReportCcy', 'OptionPrice', 'OptionPriceSide', 'OptionTimeStamp', 'OptionType', 'PremiumOverCashInDealCcy', 'PremiumOverCashInReportCcy', 'PremiumOverCashPercent', 'PremiumPerAnnumInDealCcy', 'PremiumPerAnnumInReportCcy', 'PremiumPerAnnumPercent', 'PremiumPercent', 'PricingModelType', 'PricingModelTypeList', 'ResidualAmountInDealCcy', 'ResidualAmountInReportCcy', 'RhoAmountInDealCcy', 'RhoAmountInReportCcy', 'RhoPercent', 'RiskFreeRatePercent', 'SevenDaysThetaAmountInDealCcy', 'SevenDaysThetaAmountInReportCcy', 'SevenDaysThetaPercent', 'SpeedAmountInDealCcy', 'SpeedAmountInReportCcy', 'Strike', 'ThetaAmountInDealCcy', 'ThetaAmountInReportCcy', 'ThetaPercent', 'TimeValueInDealCcy', 'TimeValueInReportCcy', 'TimeValuePercent', 'TimeValuePerDay', 'TotalMarketValueInDealCcy', 'TotalMarketValueInDealCcy', 'TotalMarketValueInReportCcy', 'TotalMarketValueInReportCcy', 'UltimaAmountInDealCcy', 'UltimaAmountInReportCcy', 'UnderlyingCcy', 'UnderlyingPrice', 'UnderlyingPriceSide', 'UnderlyingRIC', 'UnderlyingTimeStamp', 'ValuationDate', 'VannaAmountInDealCcy', 'VannaAmountInReportCcy', 'VegaAmountInDealCcy', 'VegaAmountInReportCcy', 'VegaPercent', 'Volatility', 'VolatilityPercent', 'VolatilityType', 'VolgaAmountInDealCcy', 'VolgaAmountInReportCcy', 'YearsToExpiry', 'ZommaAmountInDealCcy', 'ZommaAmountInReportCcy']
        
        Copy
    

    	
            
# # Collect data from IPA in batches (since it only accepts a max. of 100 requests per call)
batch_of: int = 100
hist_daily_universe_l_ch: list[list[option._definition.Definition]] = [
    i for i in Chunks(hist_daily_universe_l, batch_of)]
i_rd_o_def: option._definition.Definition
response2: rd.content.ipa.financial_contracts._definition.Definitions
_request_fields: list[str]
for i_int, i_rd_o_def in enumerate(hist_daily_universe_l_ch):
    print(f"Batch of {len(i_rd_o_def)} requests no. {i_int+1}/{len(hist_daily_universe_l_ch)} started")
    # Example request with Body Parameter - Symbology Lookup
    response2def = rdf.Definitions(universe=i_rd_o_def, fields=request_fields)
    try:  # One issue we may encounter here is that the field 'ErrorMessage' in `request_fields` may break the `get_data()` call and therefore the for loop. we may therefore have to remove 'ErrorMessage' in the call; ironically when it is most useful.
        response2 = response2.get_data()
    except:  # https://stackoverflow.com/questions/11520492/difference-between-del-remove-and-pop-on-lists
        _request_fields = request_fields.copy()
        _request_fields.remove('ErrorMessage')
        response2 = rdf.Definitions(universe=i_rd_o_def, fields=_request_fields).get_data()
    if i_int == 0:
        _response2df: pd.DataFrame = response2.data.df
    else:
        _response2df = _response2df.append(response2.data.df, ignore_index=True)
    # print(f"Batch of {len(i_rd_o_def)} requests no. {i_int+1}/{len(hist_daily_universe_l_ch)} ended")
# # Rename the column 'Volatility' with 'ImpliedVolatility', since that's the Volatility we asked for
response2df: pd.DataFrame = _response2df.rename(columns={'Volatility': 'ImpliedVolatility'})
# # Keep only the columns we're after
try:
    response2df = response2df[
        ['ErrorMessage', 'ImpliedVolatility', 'MarketValueInDealCcy',
        'RiskFreeRatePercent', 'UnderlyingPrice', 'DeltaPercent',
        'GammaPercent', 'RhoPercent', 'ThetaPercent', 'VegaPercent']]
except:
    response2df = response2df[
        ['ImpliedVolatility', 'MarketValueInDealCcy',
        'RiskFreeRatePercent', 'UnderlyingPrice', 'DeltaPercent',
        'GammaPercent', 'RhoPercent', 'ThetaPercent', 'VegaPercent']]
# # Add an index of dates
response2df.index = HSI_test_df2.dropna().index
# # Drop the na (not applicable) values
response2df = response2df.dropna()
# # Give a name to our data frame
response2df.columns.name = HSI_test_df2.columns.name
response2df

        
        Copy
    

    	
            
HSI_test_df3: pd.DataFrame = response2df.copy()
# Get timezone of the exchange where option is traded:
i_pd_mltyidx: pd.core.indexes.multi.MultiIndex
j_pd_mltyidx: pd.core.indexes.multi.MultiIndex
k_pd_mltyidx: pd.core.indexes.multi.MultiIndex
HSI_test_df3.index = pd.MultiIndex.from_tuples(
    [(i_pd_mltyidx, j_pd_mltyidx, k_pd_mltyidx) for i_pd_mltyidx, j_pd_mltyidx, k_pd_mltyidx in zip(
        HSI_test_df3.index,
        HSI_test_df3.index.tz_localize('GMT').tz_convert('Europe/Berlin'),
        HSI_test_df3.index.tz_localize('GMT').tz_convert('Asia/Hong_Kong'))],
    names=["gmt", "cet", "localHSI/HK"])

        
        Copy
    

    	
            
# Get only trading hours
mrkt_exhng_open_time: str = '9:30'
mrkt_exhng_lnch_time_start: str = '12:00'
mrkt_exhng_lnch_time_end: str = '13:00'
mrkt_exhng_close_time: str = '12:00'
HSI_test_morn_df3: pd.DataFrame = HSI_test_df3.droplevel(["gmt", "cet"] # unfortunately, we have to drop levels we just added to apply the useful `between_time` function.
    ).between_time(mrkt_exhng_open_time, mrkt_exhng_lnch_time_start)
HSI_test_afternoon_df3: pd.DataFrame = HSI_test_df3.droplevel(["gmt", "cet"]
    ).between_time(mrkt_exhng_lnch_time_end, mrkt_exhng_close_time)
HSI_test_th_df3: pd.DataFrame = HSI_test_morn_df3.append(HSI_test_afternoon_df3)
 
HSI_test_th_df3.index = pd.MultiIndex.from_tuples(
    [(i_pd_mltyidx, j_pd_mltyidx, k_pd_mltyidx) for i_pd_mltyidx, j_pd_mltyidx, k_pd_mltyidx in zip(
        HSI_test_th_df3.index,
        HSI_test_th_df3.index.tz_convert('GMT'),
        HSI_test_th_df3.index.tz_convert('Europe/Berlin'))],
    names=["localHSI/HK", "gmt", "cet"])
HSI_test_th_df3

        
        Copy
    

    	
            
# We will pre-alocate space for list objects to speed up its populating:
HSI_test_df2_del_lay_list: list[dict[str, str | float | int | dict[str, str | float | int | dict[str, str | float | int]]]] = [
    {'0': '0'}] * len(HSI_test_df2.dropna())
for i_int in range(len(HSI_test_df2.dropna())):
    HSI_test_df2_del_lay_list[i_int] = {
        "instrumentType": "Option",
        "instrumentDefinition": {
            # "instrumentTag": "HSITest6DelLay", # optional, str
            # "instrumentCode": "STXE44000G3.EX", # list(HSI_test2['valid_ric'][0].keys())[0],  # "instrumentCode": None # optional, str
            "strike": float(hist_opt_found_strk_pr), # optional, double (i.e.: float)
            "buySell": "Buy", # optional, str
            "callPut": "Call", # optional, str
            "exerciseStyle": "EURO", # optional, str # AMER, EURO or BERM. AMER: the owner has the right to exercise on any date before the option expires, EURO: the owner has the right to exercise only on EndDate, BERM: the owner has the right to exercise on any of several specified dates before the option expires.
            "endDate": datetime.strptime(HSI_test2_exp_date, '%Y-%m-%d').strftime('%Y-%m-%dT%H:%M:%SZ'), # optional, date-time # The maturity or expiry date of the instrument. The value is expressed in ISO 8601 format: YYYY-MM-DDT[hh]:[mm]:[ss]Z (e.g., '2021-01-01T00:00:00Z'). Mandatory for OTC ETI options and FX options (if tenor is not defined).
            "lotSize": 1.0, # optional, double (i.e.: float) # The number of the underlying asset unit on which the option is written. It can be overriden only for Commodity options. If instrumentCode of listed ETI option is defined the value comes from the instrument reference data. The default value is '1' for OTC ETI options.
            "dealContract": 1, # optional, int # The number of contracts bought or sold in the deal. The default value is '1'.
            "timeZoneOffset": 0, # optional, int # The offset in minutes between UTC and the time of the exchange where the contract is traded. No default value applies.
            "underlyingType": "Eti", # optional, str # Eti or Fx; Eti: ETI (Exchanged Traded Instruments) options, Fx: FX options.
            "underlyingDefinition": {
                "instrumentCode": HSI_underlying_RIC # optional, str # The code (a RIC) used to define the underlying asset. Mandatory for OTC ETI options. No default value applies.
            }
            # "asianDefinition": ""
            # "barrierDefinition": ""
            # "binaryDefinition": ""
            # "cbbcDefinition": ""
        },
        "pricingParameters": {
            "reportCcy": "HKD",
            "marketValueInDealCcy": float(HSI_test_df2.dropna()['OptionPrice'][i_int]),  # 14361.694905
            "valuationDate": HSI_test_df2.dropna().index[i_int].strftime('%Y-%m-%dT%H:%M:%SZ'),
            # # "marketDataDate": str(HSI_test_df2.dropna().index[i]).replace(" ", "T") + "Z", # optional, date-time
            "pricingModelType": "BlackScholes",
            "dividendType": "None",
            # # "dividendYieldPercent": 0.0, # optional, double (i.e.: float) # The ratio of annualized dividends to the underlying asset's price. The value is expressed in percentages.
            # # "marketValueInReportCcy": float(HSI_test_df2.dropna()['OptionPrice'][i]) * 1, # optional, double (i.e.: float) # The market value (premium) of the instrument. It is computed as [MarketValueInDealCcy × FxSpot]. The value is expressed in the reporting currency.
            # # "volatilityPercent": , # optional, double (i.e.: float) # The degree of the underlying asset's price variations over a specified time period, used for the option pricing. The value is expressed in percentages.It is used to compute MarketValueInDealCcy. If marketValueInDealCcy is defined, volatilityPercent is not taken into account. By default, it is computed from MarketValueInDealCcy; if VolSurface fails to return a volatility, it defaults to '20'.
            "riskFreeRatePercent": float(HSI_test_df2.dropna()['RfRatePrct'][i_int]),
            "underlyingPrice": float(HSI_test_df2.dropna()['UndrlyingPr'][i_int]),
            "volatilityType": "Implied", # optional, str # Implied, SVISurface, Historical. The value 'Implied' is available only for listed options. If volatilityPercent is defined, volatilityType is not taken into account.
            "optionPriceSide": "Mid", # optional, str # Bid, Ask, Mid, Last. Ask: if buySell is set to 'Buy', Bid: if buySell is set to 'Sell', Last: if buySell is not provided.
            "optionTimeStamp": "Close" # optional, str # Open: the opening value of valuationDate, or if it is not available, the close of the previous day is used; Close: the close value of valuationDate is used; Default: the latest snapshot is used when valuationDate is today, and the close price when valuationDate is in the past.
            # # "underlyingPriceSide": "Mid", # optional, str # Bid, Ask, Mid, Last. Ask: if buySell is set to 'Buy', Bid: if buySell is set to 'Sell', Last: if buySell is not provided.
            # # "underlyingTimeStamp": "Close" # optional, str # The mode of the underlying asset's timestamp selection. The possible values are: Open: the opening value of valuationDate, or if it is not available, the close of the previous day is used; Close: the close value of valuationDate is used; 
        }
      }

        
        Copy
    

    	
            
batch_of: int = 100
i_chunks: Any # dict[str, str | float | int | dict[str, str | float | int | dict[str, str | float | int]]]
HSI_test_df2_chunks: list[Any] = [
    i_chunks for i_chunks in Chunks(HSI_test_df2_del_lay_list, batch_of)]
response3: rd.delivery._data._response.Response
headers_name: list[str]
_response3df: pd.DataFrame
for i_int, i_chunks in enumerate(HSI_test_df2_chunks):
    if len(HSI_test_df2_chunks) > 50 and i_int % 50 == 0: # This is there just in case you have a lot of calls
        print(f"Batch of {batch_of} requests no.{str(i_int+1)}/{str(len(HSI_test_df2_chunks))} started")
    # Example request with Body Parameter - Symbology Lookup
    request_definition = rd.delivery.endpoint_request.Definition(
        method=rd.delivery.endpoint_request.RequestMethod.POST,
        url='https://api.refinitiv.com/data/quantitative-analytics/v1/financial-contracts',
        body_parameters={
            "fields": request_fields,
            "universe": i_chunks,
            "outputs": ["Data", "Headers"]})
 
    # print({"fields": request_fields, "outputs": ["Data", "Headers"], "universe": j})
 
    response3 = request_definition.get_data()
 
    if response3.is_success:
        headers_name = [h['name'] for h in response3.data.raw['headers']]
        if i_int == 0:
            response3df = pd.DataFrame(
                data=response3.data.raw['data'], columns=headers_name)
            # print({"fields": request_fields, "outputs": ["Data", "Headers"], "universe": j})
        else:
            _response3df = pd.DataFrame(
                data=response3.data.raw['data'], columns=headers_name)
            response3df = response3df.append(_response3df, ignore_index=True)
        # display(response3df)
        if len(HSI_test_df2_chunks) > 50 and i_int % 50 == 0: # This is there just in case you have a lot of calls
            print(f"Batch of {batch_of} requests no.{str(i_int+1)}/{str(len(HSI_test_df2_chunks))} ended")
    # else:
    #     display(response3)

        
        Copy
    

    	
            response3df
        
        Copy
    

    	
            
# Rename the column 'Volatility' with 'ImpliedVolatility' since that's the one we calculated
HSI_IPA_df: pd.DataFrame = response3df.rename(columns={'Volatility': 'ImpliedVolatility'})
# Name our data frame
HSI_IPA_df.columns.name = HSI_test_df2.columns.name
# Get timezone of the exchange where option is traded:
HSI_IPA_df.index = pd.MultiIndex.from_tuples(
    [(i_pd_mltyidx, j_pd_mltyidx, k_pd_mltyidx) for i_pd_mltyidx, j_pd_mltyidx, k_pd_mltyidx in zip(
        HSI_test_df2.dropna().index,
        HSI_test_df2.dropna().index.tz_localize('GMT').tz_convert('Europe/Berlin'),
        HSI_test_df2.dropna().index.tz_localize('GMT').tz_convert('Asia/Hong_Kong'))],
    names=["gmt", "cet", "localHSI/HK"])
# Get only trading hours
HSI_IPA_morn_df: pd.DataFrame = HSI_IPA_df.droplevel(["gmt", "cet"] # unfortunately, we have to drop levels we just added to apply the useful `between_time` function.
    ).between_time(mrkt_exhng_open_time, mrkt_exhng_lnch_time_start)
HSI_IPA_afternoon_df: pd.DataFrame = HSI_IPA_df.droplevel(["gmt", "cet"]
    ).between_time(mrkt_exhng_lnch_time_end, mrkt_exhng_close_time)
HSI_IPA_th_df: pd.DataFrame = HSI_IPA_morn_df.append(HSI_IPA_afternoon_df)
HSI_IPA_th_df.index = pd.MultiIndex.from_tuples(
    [(i_pd_mltyidx, j_pd_mltyidx, k_pd_mltyidx) for i_pd_mltyidx, j_pd_mltyidx, k_pd_mltyidx in zip(
        HSI_IPA_th_df.index,
        HSI_IPA_th_df.index.tz_convert('GMT'),
        HSI_IPA_th_df.index.tz_convert('Europe/Berlin'))],
    names=["localHSI/HK", "gmt", "cet"])

        
        Copy
    

    	
            
HSI_IPA_df_graph0: pd.DataFrame = HSI_IPA_th_df.droplevel(
    ["gmt", "cet"])[
        ['ImpliedVolatility', 'MarketValueInDealCcy',
        'RiskFreeRatePercent', 'UnderlyingPrice', 'DeltaPercent',
        'GammaPercent', 'RhoPercent', 'ThetaPercent', 'VegaPercent']].dropna()
HSI_IPA_df_graph0

        
        Copy
    

    	
            
def Get_weekend_days(
    df: pd.DataFrame,
    tz: str | None = None) -> list[list[date]]:
    """Get_weekend_days(df, tz=None)
 
    df (pandas dataframe):
        Must have dates as `pd.Timestamp`s in its index.
    
    returns:
        List of list of two objects: the start second of each weekend day (saturnday 1st, sunday 2nd).
    """
    weekends: list[list[date]] = []
    i_pd_dtindx: pd.core.indexes.datetimes.DatetimeIndex
    bdate_range: list[i_pd_dtindx]
    for i_int in range(len(df)-1):
        if len(pd.bdate_range(start=df.index[i_int], end=df.index[i_int+1], freq="C", weekmask="Sat Sun")) > 0:
            bdate_range = [j for j in pd.bdate_range(start=df.index[i_int], end=df.index[i_int+1], freq="C", weekmask="Sat")]
            for saturday in bdate_range:
                # Append with the the first second of that Saturday
                weekends.append([
                    pd.Timestamp(
                        year=saturday.year,
                        month=saturday.month,
                        day=saturday.day,
                        tzinfo=tz),
                    pd.Timestamp(
                        year=saturday.year,
                        month=saturday.month,
                        day=saturday.day,
                        hour=23,
                        minute=59,
                        second=59,
                        tzinfo=tz)])
            bdate_range = [j for j in pd.bdate_range(start=df.index[i_int], end=df.index[i_int+1], freq="C", weekmask="Sun")]
            for sunday in bdate_range:
                weekends.append([
                    pd.Timestamp(
                        year=sunday.year,
                        month=sunday.month,
                        day=sunday.day,
                        tzinfo=tz),
                    pd.Timestamp(
                        year=sunday.year,
                        month=sunday.month,
                        day=sunday.day,
                        hour=23,
                        minute=59,
                        second=59,
                        tzinfo=tz)])
            
    return weekends

        
        Copy
    

    	
            
# # Create Graph's Dataframe
 
for i_str in HSI_IPA_df_graph0.columns:
    HSI_IPA_df_graph0[i_str] = pd.to_numeric(HSI_IPA_df_graph0[i_str])
 
 
# # Create plot layout
 
fig: plotly.graph_objs._figure.Figure = px.line(HSI_IPA_df_graph0)  # This is just to see the implied vol graph when that field is available
 
# Format Graph: https://plotly.com/python/tick-formatting/
fig.update_layout(
    title=list(HSI_test['valid_ric'][0].keys())[0],
    template='plotly_dark')
 
# Make it so that only one line is shown by default: # https://stackoverflow.com/questions/73384807/plotly-express-plot-subset-of-dataframe-columns-by-default-and-the-rest-as-opt
fig.for_each_trace(lambda t: t.update(
    visible=True if t.name in HSI_IPA_df_graph0.columns[:1] else "legendonly"))
 
 
# # Add shade in Non Trading Hours
 
# Create purple areas for weekends:
shapes: list[dict[str, str | dict[str, str | int] | pd._libs.tslibs.timestamps.Timestamp]] = [
    {'type': "rect", 'xref': 'x', 'yref': 'paper',
     'fillcolor': 'purple', 'opacity': 0.35,
     "line": {"color": "purple", "width": 0},
     'x0': i[0], 'y0': 0,
     'x1': i[1], 'y1': 1}
     for i in Get_weekend_days(df=HSI_IPA_df_graph0, tz=None)]
 
fig.update_layout(shapes=shapes,)
 
fig.show()

        
        Copy
    

    	
            
session: rd.session.Session
# session = rd.open_session(
#     name="desktop.workspace4",  # name="desktop.workspace4", "platform.rdph"
#     config_name="C:/Example.DataLibrary.Python-main/Configuration/refinitiv-data.config.json")
try:
    session = rd.open_session(
        name="desktop.workspace4",  # name="desktop.workspace4", "platform.rdph"
        config_name="C:/Example.DataLibrary.Python-main/Configuration/refinitiv-data.config.json")
except:
    try:
        app_key: str; rd_login:str; rd_password: str
        # # You could specify your session with:
        session = rd.session.platform.Definition(
            app_key=app_key,
            grant=rd.session.platform.GrantPassword(
                username=rd_login, password=rd_password),
                ).get_session()
        session.open()
    except:
        session = rd.open_session()
        print("We're on Desktop Session")
rd.session.set_default(session) # Letting the RD API know that this is the session we're on
# session.open()

        
        Copy
    

    	
            
print("Index Implied Volatility aNd Greeks Instrument Pricing Analytics Calculation (IndxImpVolatNGreeksIPACalc) (IIVNGIPAC) expiered option test:")
IIVNGIPAC_exp_optn_test1 = IndxImpVolatNGreeksIPACalc(
    time_of_calc_datetime=datetime.strptime("2023-04-01", '%Y-%m-%d'),
    after=15,
    index_underlying=".STOXX50E",
    call_put='Call')
IIVNGIPAC_exp_optn_test2: IndxImpVolatNGreeksIPACalc = IIVNGIPAC_exp_optn_test1.search_index_opt_ATM(
    debug=False,
    call_or_put='Put',
    search_fields=["ExchangeCode", "UnderlyingQuoteName"],
    include_weekly_opts=False,
    top_nu_srch_results=10)
IIVNGIPAC_exp_optn_test3: IndxImpVolatNGreeksIPACalc = IIVNGIPAC_exp_optn_test2.IPA_calc(
    debug=False)
print(f"IIVNGIPAC_exp_optn_test3.instrument: {IIVNGIPAC_exp_optn_test3.instrument}")
print(f"IIVNGIPAC_exp_optn_test3.ATM_opt: {IIVNGIPAC_exp_optn_test3.ATM_opt}")
print(f"IIVNGIPAC_exp_optn_test3.maturity: {IIVNGIPAC_exp_optn_test3.maturity}")
if pd.to_datetime(IIVNGIPAC_exp_optn_test3.maturity) > datetime.now():
    print(f"IIVNGIPAC_exp_optn_test3.instrument_info:")
    display(IIVNGIPAC_exp_optn_test3.instrument_info)

        
        Copy
    

    	
            IIVNGIPAC_exp_optn_test3.IPA_df.dropna()
        
        Copy
    

    	
            
request_fields = [
            "DeltaPercent", "GammaPercent", "RhoPercent",
            "ThetaPercent", "VegaPercent"]
for i_str in ["ErrorMessage", "MarketValueInDealCcy", "RiskFreeRatePercent",
                      "UnderlyingPrice", "Volatility"][::-1]:
            request_fields.insert(0, i_str)
print(request_fields)

        
        Copy
    

    	
            IIVNGIPAC_exp_optn_test3.simple_graph()
        
        Copy
    

    	
            IIVNGIPAC_exp_optn_test3.graph(mrkt_exhng_open_time='9:00', mrkt_exhng_close_time='17:00').overlay().fig
        
        Copy
    

    	
            IIVNGIPAC_exp_optn_test3.graph(mrkt_exhng_open_time='9:00', mrkt_exhng_close_time='17:00').stack3().fig
        
        Copy
    

    	
            
print("Index Implied Volatility aNd Greeks Instrument Pricing Analytics Calculation (IndxImpVolatNGreeksIPACalc) (IIVNGIPAC) expiered option test At Trade only:")
IIVNGIPAC_exp_optn_AT_test1 = IndxImpVolatNGreeksIPACalc(
    time_of_calc_datetime=datetime.strptime("2023-04-01", '%Y-%m-%d'),
    after=15,
    index_underlying=".STOXX50E",
    call_put='Call')
IIVNGIPAC_exp_optn_AT_test2: IndxImpVolatNGreeksIPACalc = IIVNGIPAC_exp_optn_AT_test1.search_index_opt_ATM(
    debug=False,
    call_or_put='Put',
    search_fields=["ExchangeCode", "UnderlyingQuoteName"],
    include_weekly_opts=False,
    top_nu_srch_results=10)
IIVNGIPAC_exp_optn_AT_test3: IndxImpVolatNGreeksIPACalc = IIVNGIPAC_exp_optn_AT_test2.IPA_calc(
    debug=False,
    AT_opn_trade_only=True)
print(f"IIVNGIPAC_exp_optn_AT_test3.instrument: {IIVNGIPAC_exp_optn_AT_test3.instrument}")
print(f"IIVNGIPAC_exp_optn_AT_test3.ATM_opt: {IIVNGIPAC_exp_optn_AT_test3.ATM_opt}")
print(f"IIVNGIPAC_exp_optn_AT_test3.maturity: {IIVNGIPAC_exp_optn_AT_test3.maturity}")
if pd.to_datetime(IIVNGIPAC_exp_optn_AT_test3.maturity) > datetime.now():
    print(f"IIVNGIPAC_exp_optn_AT_test3.instrument_info:")
    display(IIVNGIPAC_exp_optn_AT_test3.instrument_info)

        
        Copy
    

    	
            IIVNGIPAC_exp_optn_AT_test3.simple_graph()
        
        Copy
    

    	
            IIVNGIPAC_exp_optn_AT_test3.graph(mrkt_exhng_open_time='9:00', mrkt_exhng_close_time='17:00').overlay().fig
        
        Copy
    

    	
            IIVNGIPAC_exp_optn_AT_test3.graph(mrkt_exhng_open_time='9:00', mrkt_exhng_close_time='17:00').stack3().fig
        
        Copy
    

    	
            rd.close_session() # close the RD session opend at the start
        
        Copy