Authors:
Equity Derivatives Intraday Analytics: using Python & IPA to gather intraday insights on live and expired derivatives
You could consider this article to be part 3, after "Python Type Hint Finance Use Case: Implied Volatility and Greeks of Index 'At The Money' Options: Part 2".
In the realm of financial markets, the ability to analyze and predict the behavior of equity derivatives within the trading day can be the difference between profit and loss. This comprehensive guide will walk you through an advanced Python helper file and it's 'associate' Notebook, meticulously crafted to aid traders in making informed decisions.
Intraday analytics for equity derivatives is not just about speed; it’s about precision. The financial markets are unforgiving, and even the smallest oversight can lead to significant consequences. The workflow showcased below shines a light to guide traders through the tumultuous market data with accuracy and insight.
In this article, we will create a series of Python Classes that will gather insights on Equity Options. Specifically it will (i) take in (a) an asset as the underlying and (b) a date as the expiry of the Option one would like to find (and find information/insights on) from which it will (ii) collect market data, (iii) compute Implied Volatilities and Greeks using IPA, and (iv) loop through this logic to construct Smiles; then it will (v) graphically represent all this data all while (vi) allowing savvy users access to said data in raw form (i.e.: in pandas dataframes).
This is all done while heavily leveraging Haykaz's article.
In the fast-paced world of finance, intraday analytics for equity derivatives is a critical tool for traders and analysts. Today, we’re going to explore the function `IPA_Equity_Vola_n_Greeeks` in the Python helper file `IPA_Equity_Vola_n_Greeks_Class.py` designed to provide insights on live and expired derivatives.
Content
1. The `IPA_Equity_Vola_n_Greeks_Class.py` helper file
1.1. `get_options_RIC`
1.1.01. `_get_exchange_code`
1.1.02. `_get_exp_month`
1.1.03. `_check_expiry`
1.1.04. `_request_prices`
1.1.05. `get_ric_opra`
1.1.06. `get_ric_hk`
1.1.07. `get_ric_ose`
1.1.08. `get_ric_eurex`
1.1.09. `get_ric_ieu`
1.1.10 `get_option_ric`
1.1.11 `get_option_ric_through_strike_range`
1.2. `IPA_Equity_Vola_n_Greeeks`
1.2.1. `initiate`
1.2.2. `get_history_mult_times`
1.2.3. `get_data`
1.2.4. `graph`
1.2.5. `cross_moneyness`
2. CodeBook's `IPA_Equity_Vola_n_Greeks.ipynb` functional file
2.1. `Eqty_ATM_Optn_Vola_n_Greeks`
2.2. `Eqty_ATM_Optn_Impli_Vol_Smile`
3. Output
4. Conclusion
1. The `IPA_Equity_Vola_n_Greeks_Class.py` helper file
Dr. Haykaz Aramyan's great article, "Functions to find Option RICs traded on different exchanges", was the basis for this function. Please do not hesitate to read it. For this article, I changed it to in-build `bebug` functionalities.
When it comes to the code, we first need to import libraries:
import datetime as dt
from datetime import datetime, timedelta # We use these to manipulate time values
import pandas as pd
import plotly
import matplotlib # We use `matplotlib` to just in case users do not have an environment suited to `plotly`.
import IPython # 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
import time # This is to pause our code when it needs to slow down
import numpy as np
import refinitiv.data as rd
Now let's authenticate ourselves to the LSEG data service. You can find out more about this in the article "Summary of Common LSEG Refinitiv APIs".
rd.open_session()
1.1. `get_options_RIC`
Now let's create helper functions in the `get_options_RIC` CLass:
class get_options_RIC():
def __init__(self): # Constroctor
return None
In Python, you can create functions (as shown here); you may, however, want to create a series of functions embedded within another function, akin to libraries (e.g.: the function `to_csv` embedded in the library `pandas` that is often called with `pandas.to_csv`). You can create such functions within a Python Class. As per the Python official documentation:
<<The instantiation operation (“calling” a class object) creates an empty object. Many classes like to create objects with instances customized to a specific initial state. Therefore a class may define a special method named __init__(),>>
1.1.01. `_get_exchange_code`
In the code cell below, we define the function `_get_exchange_code` that will live within the Class `get_options_RIC`. The lines in betwen """ are Docstrings, here to outline details about the function in question.
This fuction will take in the `asset` string object as argument; this is the underlying asset for the Option.
def _get_exchange_code(
self,
asset):
"""
This function retrieves the exchange code for a given equity option.
The exchange code is determined by the specific characteristics of the asset, which
may include its market type, geographical location, or other defining attributes.
These attributes are gathered via LSEG Data.
Parameters
-----------------------------------------------
Inputs:
- asset (str): The identifier for the asset for which the exchange code is required.
Output/Returns:
- list[str]: The exchange codes associated with the asset.
"""
Now (in the next code cell), let's use the search functionality to find live options (trading monthly):
response = rd.discovery.search(
query = asset,
filter = "SearchAllCategory eq 'Options' and Periodicity eq 'Monthly' ",
select = 'ExchangeCode',
group_by = "ExchangeCode")
Now, let's create a prompt (raise an exception) if and when no response is returned:
if len(response) == 0:
raise(MyException(ExceptionData(f"It's looking like there might not have been any trades for this option, {asset} on that date. You may want to check with: `rd.discovery.search(query = '{asset}', filter = \"SearchAllCategory eq 'Options' and Periodicity eq 'Monthly' \", select = 'ExchangeCode', group_by = 'ExchangeCode')`")))
If the above prompt did not stop our code and warn the user, let's move on and collect the exchanges on which the option inputed by the user trades:
exchanges = response.drop_duplicates()["ExchangeCode"].to_list()
exchange_codes = []
for exchange in exchanges:
exchange_codes.append(exchange)
return exchange_codes
We then, at the end, return our list of exchanges, as per the `return exchange_codes`. This is all that our function `_get_exchange_code` does. We will, below, use that function simply to help us get this exchange list.
1.1.02. `_get_exp_month`
Now we can create the `_get_exp_month` function that returns two objects, `indent`, which is a dictionary including data on the nomeclacure of expired Options (specifically relating to their expiring month) and `exm_month` which is a dictionary (itself, too) containing the same data as `indent` but only just for our expiry month (i.e.: for the month in which the Option-we-are-interested-in expires)
def _get_exp_month(
self,
maturity,
opt_type,
strike=None,
opra=False):
maturity = pd.to_datetime(maturity)
# 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(maturity.month)]['C']
elif opt_type.upper() == 'P':
exp_month = ident[str(maturity.month)]['P']
if opra and strike > 999.999:
exp_month = exp_month.lower()
return ident, exp_month
Note that the logic for such expiry month data depends on the nature of the Option, be it a 'call' ('C') or a 'put' ('P'); thus the `if opt_type.upper() == 'C'` and `elif opt_type.upper() == 'P'` loops above.
Note also that, simply just for the Opra Exchange, if the Option's strike is at (or above) 1000, the expiry month lettre used in the Option's RIC is not capitalised any more. This is simply there to make the game fun... This is the reason for the `if opra and strike > 999.999` loop.
1.1.03. `_check_expiry`
Now, if an Option is expired, we need to change its RIC; the `_check_expiry` not only checks for expiry, as the name suggests, but also changes the outputed RIC in accordance with whether or not the Option is expired:
def _check_expiry(self, ric, maturity, ident):
maturity = pd.to_datetime(maturity)
if maturity < datetime.now():
ric = ric + '^' + ident[str(maturity.month)]['exp'] + str(maturity.year)[-2:]
return ric
1.1.04. `_request_prices`
Now, this `_request_prices` function simply collects BID, ASK, Trade (TRDPRC_1) and Settlement (SETTLE) prices, if available, via the get_history function:
def _request_prices(self, ric, debug):
prices = []
try:
prices = rd.get_history(ric, fields = ['BID','ASK','TRDPRC_1','SETTLE'])
except rd.errors.RDError as err:
if debug:
print(f'Constructed ric {ric} - {err}')
return prices
1.1.05. `get_ric_opra`
Note that previous functions had an underscore (_) before their names. This is because I considered them as helper functions; i.e.: functions themselves only there to be used by other functions. These differ from other functions that could (and probably should) be used by users.
Such a function is `get_ric_opra`, that... getsthe RIC of the Option in question if it is traded on the Opra Exchange. It also returns price data (since these were collected anyway):
def get_ric_opra(self, asset, maturity, strike, opt_type, debug):
maturity = pd.to_datetime(maturity)
# trim underlying asset's RIC to get the required part for option RIC
if asset[0] == '.': # check if the asset is an index or an equity
asset_name = asset[1:] # get the asset name - we remove "." symbol for index options
else:
asset_name = asset.split('.')[0] # we need only the first part of the RICs for equities
ident, exp_month = self._get_exp_month(
maturity=maturity, opt_type=opt_type, strike=strike, opra=True)
# get strike prrice
if type(strike) == float:
int_part = int(strike)
dec_part = str(str(strike).split('.')[1])
else:
int_part = int(strike)
dec_part = '00'
if len(dec_part) == 1:
dec_part = dec_part + '0'
if int(strike) < 10:
strike_ric = '00' + str(int_part) + dec_part
elif int_part >= 10 and int_part < 100:
strike_ric = '0' + str(int_part) + dec_part
elif int_part >= 100 and int_part < 1000:
strike_ric = str(int_part) + dec_part
elif int_part >= 1000 and int_part < 10000:
strike_ric = str(int_part) + '0'
elif int_part >= 10000 and int_part < 20000:
strike_ric = 'A' + str(int_part)[-4:]
elif int_part >= 20000 and int_part < 30000:
strike_ric = 'B' + str(int_part)[-4:]
elif int_part >= 30000 and int_part < 40000:
strike_ric = 'C' + str(int_part)[-4:]
elif int_part >= 40000 and int_part < 50000:
strike_ric = 'D' + str(int_part)[-4:]
# build ric
ric = asset_name + exp_month + str(maturity.day) + str(maturity.year)[-2:] + strike_ric + '.U'
ric = self._check_expiry(ric, maturity, ident)
prices = self._request_prices(ric, debug=debug)
# return valid ric(s)
if len(prices) == 0:
if debug:
print('RIC with specified parameters is not found')
return ric, prices
1.1.06. `get_ric_hk`
The `get_ric_hk` function functions similarly to the `get_ric_opra` one, but for the Hong Kong Exchange. You can find this code in this article's GitHub Repo.
1.1.07. `get_ric_ose`
The `get_ric_ose` function functions similarly to the `get_ric_opra` one. You can find this code in this article's GitHub Repo.
1.1.08. `get_ric_eurex`
The `get_ric_eurex` function functions similarly to the `get_ric_opra` one. You can find this code in this article's GitHub Repo.
1.1.09. `get_ric_ieu`
The `get_ric_eurex` function functions similarly to the `get_ric_opra` one. You can find this code in this article's GitHub Repo.
1.1.10 `get_option_ric`
Our `get_option_ric` function can now use all the above functions to return the RIC of the expired or live Option:
def get_option_ric(self, asset, maturity, strike, opt_type, debug, exchange_not_supported_message_count=0):
# define covered exchanges along with functions to get RICs from
exchanges = {
'OPQ': self.get_ric_opra,
'IEU': self.get_ric_ieu,
'EUX': self.get_ric_eurex,
'HKG': self.get_ric_hk,
'HFE': self.get_ric_hk,
'OSA': self.get_ric_ose}
# get exchanges codes where the option on the given asset is traded
exchnage_codes = self._get_exchange_code(asset)
# get the list of (from all available and covered exchanges) valid rics and their prices
options_data = {}
for exch in exchnage_codes:
if exch in exchanges.keys():
ric, prices = exchanges[exch](asset, maturity, strike, opt_type, debug)
if len(prices) != 0:
options_data[ric] = prices
if debug:
print(f'Option RIC for {exch} exchange is successfully constructed')
else:
if exchange_not_supported_message_count < 1:
print(f'The {exch} exchange is not supported yet')
return options_data
1.1.11 `get_option_ric_through_strike_range`
I removed the Docstrings from the code cell below to keep it concise. We will be creating Smiles and will want to collect Option RICs for strikes above and below the "current" strike used in our calculations (whatever this strike is exactly at the time the code runs). In line with this, we go through the below `direction`, up (`+`), down (`-`) or with our original strike (`None`) to find such Options:
def get_option_ric_through_strike_range(
self,
asset,
maturity,
strike,
opt_type,
rnge,
rnge_interval,
round_to_nearest,
debug,
direction=None):
if direction == "+":
if debug:
print("strike lookthough direction: +")
new_strike = strike+rnge_interval
if direction == "-":
if debug:
print("strike lookthough direction: -")
new_strike = strike-rnge_interval
if direction == None:
if debug:
print("strike lookthough direction: o")
new_strike = strike
We can then use previously defined functions to collect the information (Option RIC) we're after. In the cell below, we do just that, and loop through prices to try and find a strike for which there exists an Option:
optn_ric = self.get_option_ric(
asset=asset,
maturity=maturity,
strike=new_strike,
opt_type=opt_type,
debug=debug)
range_rounded = round((rnge+rnge_interval)/round_to_nearest)*round_to_nearest
if len(optn_ric) == 0:
# We are going to loop though exchanges to check is the option is traded there.
# We are coding defensively, meaning that we may make several calls for the same exchange.
# To limit the number of 'exchange not supported' messages, if it is not supported, we will use the `exchng_not_sprted_msg_cnt` object.
exchng_not_sprted_msg_cnt = 0
for i in range(0, range_rounded, rnge_interval):
if i < rnge:
if len(optn_ric) == 0 and (direction == None or direction == "+"): # Try and find an Option RIC for a Strike `interval` above the given `strike`:
new_strike = (round(strike/round_to_nearest)*round_to_nearest)+i
optn_ric = get_options_RIC().get_option_ric(
asset=asset, maturity=maturity, opt_type=opt_type, debug=debug,
strike=new_strike, exchange_not_supported_message_count=exchng_not_sprted_msg_cnt)
exchng_not_sprted_msg_cnt =+ 1
if debug:
print(f"{i} new_strike: {new_strike}")
if len(optn_ric) == 0 and (direction == None or direction == "-"): # Try and find an Option RIC for a Strike `interval` below the given `strike`:
new_strike = (round(strike/round_to_nearest)*round_to_nearest)-i
optn_ric = get_options_RIC().get_option_ric(
asset=asset, maturity=maturity, opt_type=opt_type, debug=debug,
strike=new_strike, exchange_not_supported_message_count=exchng_not_sprted_msg_cnt)
if debug:
print(f"{i} new_strike: {new_strike}")
return optn_ric, new_strike
1.2. `IPA_Equity_Vola_n_Greeeks`
Now let's create the main `IPA_Equity_Vola_n_Greeeks` Class:
Our initiating constructor (`__init__`) here is a lot more busy than our previous one (for the above `get_options_RIC` function).
Let's go through it in more detail; specifically, what are these arguments?
- self: `self` is used to rever to all objects previously loded in `self` as shown here. You can think of it like a dictionary of sorts.
- debug: we add lines printing logs through out our code, but we do not want them showing up all the time; if `debug` is set to `False`, they will be silenced.
-underlying: this is the underlying equity for which we are looking for an Option.
- strike: this can be left as `None`, in which case our code will look for the price of the underlying; if the date of interest is in the past (and we are after an expired option), then the price of the underlying at that (past) date is used; if the date of interest is in the future (i.e.: we're after a live option maturing in the future), we will collect today's price. Of course, if a float is entered as `strike`, it superseeds this lolgic, and this entered strike is used instead.
- atm_range: we will be looking for an Option close to the strike by default, but we cannot search indefinitely from - infinity to + infinity; we therefore cap our search from - `atm_range ` to + `atm_range `. What intervals are we going to use going up and down though? well:
- atm_intervals: We will be looking for a strike for which an option exist (for our underlying of choice) in increments of `atm_intervals`.
- atm_round_to_nearest: It just happens that Options can trade at diferent intervals depending on the currency (e.g.: to the nearest 1$? 0.01$? 1 Yen? 100 Yen?). Whichever currency is chosen, we will only look for options with strikes that round up to the nearest `atm_round_to_nearest`.
- atm_direction: this refers back to the `get_option_ric_through_strike_range` function defined above.
- maturity: this refers to the maturity date of the option.
- maturity_format: this refers to the date format entered as `maturity` just above.
- option_type: refers to 'Call' or 'Put' as the option type of interest.
- buy_sell: this is needed for the IPA calculations that will compute Implied Volatilities and Greeks.
- data_retrieval_interval: once an option is found, which data granularity would you like to use to compute Implied Volatilities and Greeks?
- curr: this is needed for the IPA calculations that will compute Implied Volatilities and Greeks.
- exercise_style: this is needed for the IPA calculations that will compute Implied Volatilities and Greeks.
- option_price_side: while this is needed too for the IPA calcs, like `curr` above, we will also use this for option data colection; i.e.: once an option is found, we eould usually try and collect trade data, from which to start our computations, however, you may find little liquidity (i.e.: few trades) but much interest (i.e.: many BIDs and ASKs); if you think it best to focus on this interest, in order to collect more liquid price data, you can use this `option_price_side` as 'BID' or 'ASK'.
- underlying_time_stamp: this is needed for the IPA calculations that will compute Implied Volatilities and Greeks.
- resample: It is true that we collected data as per the the above `data_retrieval_interval` argument, however, we will be resampling data to get homogeneous dataframes (i.e.: we get daily data - via Risk Free Rates (which will be our next argument) - and intraday data - via option prices - and need to combile the two).
- rsk_free_rate_prct: this argument needs to be a string for the RIC of the instrument you would like as a Risk Free Rate from which the Black-Scholes-Merton model will compute Implied Volatilities and Greeks.
- rsk_free_rate_prct_field: this will be the field used to collect price data for the `rsk_free_rate_prct`.
- request_fields: this is the (long) list of fields to be collected from IPA. They are input and output data fields for the IPA computations.
- search_batch_max: were we to look for a live (and not an expired) option, we will use the aformentioned search functionalities, and it needs a max batch, an upper limit, up to which the search will go on.
- slep: this is an abriviation for 'sleep'. We will use this object to let us know how many seconds we should wait, or 'sleep', between calls (e.g.: IPA calls). This is needed so that we don't overload the services we use, since they might kick us out if we do.
- corr: this is short for 'correlation'. We can add correlation analysis to our graphs to enhance the number of insights we can show.
- hist_vol: this is short for 'historical volaitliy'. We can add hist_vol analysis to our graphs to enhance the number of insights we can show.
def __init__(
self,
debug=False,
underlying = "LSEG.L", # ".SPX"
strike = None,
atm_range=200,
atm_intervals=50,
atm_round_to_nearest=100,
atm_direction=None,
maturity = '2024-01-19',
maturity_format = '%Y-%m-%d', # e.g.: '%Y-%m-%d', '%Y-%m-%d %H:%M:%S' or '%Y-%m-%dT%H:%M:%SZ'
option_type = 'Call', # 'Put'
buy_sell = 'Buy', # 'Sell'
data_retrieval_interval = rd.content.historical_pricing.Intervals.HOURLY,
curr = 'USD',
exercise_style = 'EURO',
option_price_side = None, # `None`, `'Bid'` or `'Ask'`.
underlying_time_stamp = 'Close',
resample = '10min', # You can consider this the 'bucket' or 'candles' from which calculations will be made.
rsk_free_rate_prct = 'USDCFCFCTSA3M=', # for `".SPX"`, I go with `'USDCFCFCTSA3M='`; for `".STOXX50E"`, I go with `'EURIBOR3MD='`
rsk_free_rate_prct_field = 'TR.FIXINGVALUE', # for `".SPX"`, I go with `'TR.FIXINGVALUE'`; for `".STOXX50E"`, I go with `'TR.FIXINGVALUE'` too.
request_fields = ['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'],
search_batch_max = 90,
slep= 0.6,
corr=True,
hist_vol=True): # Constroctor
There is little to speak of in this `__init__` constructor, let's move on. You can find the full code on GitHub.
1.2.1. `initiate`
We will now create a simple `initiate` function from which one ought to collect the necessary data, namely the strike of the underlying if none is provided in `IPA_Equity_Vola_n_Greeeks` itself:
def initiate(
self,
direction='default'):
if direction=='default':
direction=self.atm_direction
if self.strike == None:
try_no = 0
while try_no < 3:
try:
if datetime.strptime(self.maturity, '%Y-%m-%d') >= datetime.now():
strike_maturity_pr_call = datetime.now().strftime('%Y-%m-%d')
else:
strike_maturity_pr_call = self.maturity
self.strike = rd.get_history(self.underlying, "TR.ClosePrice", start=strike_maturity_pr_call).values[0][0]
self.strike = round(self.strike, -1)
try_no += 1
except:
try_no += 1
self._strike=self.strike # Let _strike be an 'original press' that we will not change, but keep going back to instead.
Now let's find out Option of interest's RIC:
_undrlying_optn_ric, new_strike = get_options_RIC().get_option_ric_through_strike_range(
asset=self.underlying,
maturity=self.maturity,
strike=self.strike,
opt_type=self.option_type[0],
rnge=self.atm_range,
rnge_interval=self.atm_intervals,
round_to_nearest=self.atm_round_to_nearest,
debug=self.debug,
direction=direction)
We can skip a few debugging and data tidying lines to get to the below. We're still in the `initiate` function. In the code cell below, we try and collect, first, trading data, and if that's not available, 'SETTLE' data, then 'BID':
optn_mrkt_pr_gmt = rd.content.historical_pricing.summaries.Definition(rd.content.historical_pricing.summaries.Definition(
universe=undrlying_optn_ric,
start=sdate,
end=edate,
interval=self.data_retrieval_interval,
fields=fields_lst
).get_data().data.df
if 'TRDPRC_1' in optn_mrkt_pr_gmt.columns:
if len(optn_mrkt_pr_gmt.TRDPRC_1.dropna()) > 0:
optn_mrkt_pr_gmt = pd.DataFrame(
data={'TRDPRC_1': optn_mrkt_pr_gmt.TRDPRC_1}).dropna()
elif 'SETTLE' in optn_mrkt_pr_gmt.columns:
if len(optn_mrkt_pr_gmt.SETTLE.dropna()) > 0:
optn_mrkt_pr_gmt = pd.DataFrame(
data={'SETTLE': optn_mrkt_pr_gmt.SETTLE}).dropna()
elif 'BID' in optn_mrkt_pr_gmt.columns:
if len(optn_mrkt_pr_gmt.BID.dropna()) > 0:
optn_mrkt_pr_gmt = pd.DataFrame(
data={'BID': optn_mrkt_pr_gmt.BID}).dropna()
else:
display(optn_mrkt_pr_gmt)
raise ValueError("Issue with `optn_mrkt_pr_gmt`, as displayed above.")
optn_mrkt_pr_gmt.columns.name = undrlying_optn_ric
column_with_fewest_nas = optn_mrkt_pr_gmt.isna().sum().idxmin()
In the next two code cells, we collect data on our underlying for the data range of interest:
optn_mrkt_pr_field = column_with_fewest_nas
df_strt_dt = optn_mrkt_pr_gmt.index[0]
df_strt_dt_str = df_strt_dt.strftime('%Y-%m-%dT%H:%M:%S.%f')
df_end_dt = optn_mrkt_pr_gmt.index[-1]
df_end_dt_str = df_end_dt.strftime('%Y-%m-%dT%H:%M:%S.%f')
undrlying_mrkt_pr_gmt = rd.content.historical_pricing.summaries.Definition(
universe=self.underlying,
start=df_strt_dt_str,
end=df_end_dt_str,
interval=self.data_retrieval_interval,
fields=optn_mrkt_pr_gmt.columns.array[0]
).get_data().data.df
undrlying_mrkt_pr_gmt_cnt = undrlying_mrkt_pr_gmt.count()
if self.debug:
print("undrlying_mrkt_pr_gmt 1st")
print(f"rd.content.historical_pricing.summaries.Definition(universe='{self.underlying}', start='{df_strt_dt_str}', end='{df_end_dt_str}', interval='{self.data_retrieval_interval}', fields='{optn_mrkt_pr_gmt.columns.array[0]}').get_data().data.df")
display(undrlying_mrkt_pr_gmt)
try:
undrlying_mrkt_pr_gmt = pd.DataFrame(
data={'TRDPRC_1': undrlying_mrkt_pr_gmt.TRDPRC_1}).dropna()
except:
try:
undrlying_mrkt_pr_gmt = pd.DataFrame(
data={'SETTLE': undrlying_mrkt_pr_gmt.SETTLE}).dropna()
except:
try:
undrlying_mrkt_pr_gmt = pd.DataFrame(
data={'BID': undrlying_mrkt_pr_gmt.BID}).dropna()
except:
raise ValueError(f"There seem to be no data for the field {optn_mrkt_pr_gmt.columns.array[0]}. You may want to choose the option 'Let Program Choose' as opposed to {optn_mrkt_pr_gmt.columns.array[0]}. This came from the function of `as per the function `rd.content.historical_pricing.summaries.Definition(universe='{self.underlying}', start='{df_strt_dt_str}', end='{df_end_dt_str}', interval='{self.data_retrieval_interval}', fields='{optn_mrkt_pr_gmt.columns.array[0]}').get_data().data.df`")
undrlying_mrkt_pr_gmt.columns.name = f"{self.underlying}"
if optn_mrkt_pr_gmt.index[-1] >= undrlying_mrkt_pr_gmt.index[-1]:
df_gmt = optn_mrkt_pr_gmt.copy()
underlying_pr_field = f"underlying {self.underlying} {optn_mrkt_pr_gmt.columns.array[0]}"
df_gmt[underlying_pr_field] = undrlying_mrkt_pr_gmt
else:
df_gmt = undrlying_mrkt_pr_gmt.copy()
underlying_pr_field = f"underlying {self.underlying} {optn_mrkt_pr_gmt.columns.array[0]}"
df_gmt.rename(
columns={optn_mrkt_pr_gmt.columns.array[0]: underlying_pr_field},
inplace=True)
df_gmt[optn_mrkt_pr_gmt.columns.array[0]] = optn_mrkt_pr_gmt
df_gmt.columns.name = optn_mrkt_pr_gmt.columns.name
df_gmt.fillna(method='ffill', inplace=True)
You may have seen that we sometimes create objects under this `self` objects, such as in:
`self.strike = rd.get_history(self.underlying, "TR.ClosePrice", start=strike_maturity_pr_call).values[0][0]`
This is an object we created and can refer to in the `self` object from that line on.
We created quite a few useful objects in the lines of code above, and would like to access them, so let's embed them in `self` and return `self`:
self.df_end_dt = df_end_dt
self.df_end_dt_str = df_end_dt_str
self.optn_mrkt_pr_field = optn_mrkt_pr_field
self.df_strt_dt = df_strt_dt
self.df_strt_dt_str = df_strt_dt_str
self.df_end_dt = df_end_dt
self.df_end_dt_str = df_end_dt_str
self.underlying_pr_field = underlying_pr_field
self.df_gmt = df_gmt
self.optn_mrkt_pr_gmt = optn_mrkt_pr_gmt
self.undrlying_optn_ric = undrlying_optn_ric
return self
1.2.2. `get_history_mult_times`
We always have to code defensively when using LSEG APIs. Reason being that no one API can ever flawlessly return all finance data in the same format; e.g.: think of the different fields that exist for Equities and not for Fixed Income (e.g.: Dividends). In this case, we may be sending too many requests to LSEG's services at too quick a succession; we might get "Error code -1" errors, or similar ones. As a result, it is advised to use a function such as the below `get_history_mult_times` to alleviate such errors.
`get_history_mult_times` is a function that will send data requests to the get_history function multiple times; it's that simple. As a result, the arguments for this function are rather self explanatory and are the same as the get_history function apart from `self`, which is explained above, and `trs` which is short for "tries" and limits the number of tries we will go through attempting to collect said data.
def get_history_mult_times(
self,
univ,
flds,
strt,
nd,
trs=6):
while trs:
try:
trs -= 1
_df = rd.get_history(
universe=univ,
fields=flds,
start=strt, end=nd)
trs = 0 # break, because no error
except Exception: # catch all exceptions
time.sleep(self.slep)
_df = None
if _df is None: # If `_df` doesn't exist
print("\n")
print("Please note that the following failed:")
print(_df)
print(f"rd.get_history(universe={univ}, fields={flds},start='{strt}', end='{nd}')")
display(rd.get_history(universe=univ, fields=flds,start=strt, end=nd))
print(f"Please consider another instrument other than {univ}")
raise ValueError(f"Issue with `_df`, itself taken from `rd.get_history(universe={univ}, fields={flds},start='{strt}', end='{nd}')`")
return _df
`get_history_mult_times` only returns a pandas dataframe, and not `self`. It is rather self contained, and maybe it should have had an underscore (_) before it...
1.2.3. `get_data`
Unlike `get_history_mult_times`, our function `get_data` is a lot more specific to our use case and the `IPA_Equity_Vola_n_Greeeks` Python Class. This `get_data` function:
def get_data(self):
(i) Collects the Risk Free price data as per the fields predefined in `self`:
rf_rate_prct = self.get_history_mult_times(
[self.rsk_free_rate_prct],
[self.rsk_free_rate_prct_field],
(self.df_strt_dt - timedelta(days=1)).strftime('%Y-%m-%d'),
(datetime.strptime(self.maturity, self.maturity_format) + timedelta(days=1)).strftime('%Y-%m-%d'))
(ii) The line collecting data for the `rf_rate_prct` object in the code cell above may fail, as the `self.rsk_free_rate_prct`, `self.rsk_free_rate_prct_field` and `self.df_strt_dt` arguments may be incompatible with the `get_history_mult_times` (and by extention, the `rd.get_history`) function. We need to take this permutation of arguments into account with the code cell below which throws a (comprehensible) error if the `rf_rate_prct` dataframe is empty:
if np.amax(self.optn_mrkt_pr_gmt.groupby(self.optn_mrkt_pr_gmt.index.date).count().values) == 1:
raise(MyException(ExceptionData("This function only allows intraday data for now. Only interday data was returned. This may be due to illiquid Options Trading. You may want to ask only for 'Bid' or 'Ask' data as opposed to 'Let Program Choose', if you have not made that choice already, as the latter will prioritise executed trades, of which there may be few.")))
(iii) We can get all this data together in a `merged_df` object:
_df_gmt = self.df_gmt.copy()
# Convert the index of df_gmt to datetime
_df_gmt.index = pd.to_datetime(_df_gmt.index)
# Convert the index of rf_rate_prct to datetime
rf_rate_prct.index = pd.to_datetime(rf_rate_prct.index)
# Resample rf_rate_prct to match the frequency of df_gmt, forward filling the missing values
rf_rate_prct_resampled = rf_rate_prct.resample('T').ffill()
# Merge the dataframes
merged_df = _df_gmt.merge(rf_rate_prct_resampled, left_index=True, right_index=True, how='left')
merged_df.columns.name = self.df_gmt.columns.name
self.df_gmt = merged_df
self.df_gmt.rename(
columns={list(merged_df.columns)[-1]: 'RfRatePrct'},
inplace=True)
self.df_gmt = self.df_gmt.fillna(method='ffill').fillna(method='bfill')
self.df_gmt_no_na = self.df_gmt.dropna(subset=['RfRatePrct'])
(iv) We can finally go forward now that we have all the data needed to compute Implied Volatilities and Greeks:
ipa_univ_requ = [
rd.content.ipa.financial_contracts.option.Definition(
strike=float(self.strike),
buy_sell=self.buy_sell,
underlying_type=rd.content.ipa.financial_contracts.option.UnderlyingType.ETI,
call_put=self.option_type,
exercise_style=self.exercise_style,
end_date=datetime.strptime(self.maturity, self.maturity_format).strftime('%Y-%m-%dT%H:%M:%SZ'), # '%Y-%m-%dT%H:%M:%SZ' for RD version 1.2.0 # self.maturity.strftime('%Y-%m-%dt%H:%M:%Sz') # datetime.strptime(self.maturity, '%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%dt%H:%M:%Sz')
lot_size=1,
deal_contract=1,
time_zone_offset=0,
underlying_definition=rd.content.ipa.financial_contracts.option.EtiUnderlyingDefinition(
instrument_code=self.underlying),
pricing_parameters=rd.content.ipa.financial_contracts.option.PricingParameters(
valuation_date=self.df_gmt_no_na.index[i_int].strftime('%Y-%m-%dT%H:%M:%SZ'), # '%Y-%m-%dT%H:%M:%SZ' for RD version 1.2.0 # '%Y-%m-%dt%H:%M:%Sz' # One version of rd wanted capitals, the other small. Here they are if you want to copy paste.
report_ccy=self.curr,
market_value_in_deal_ccy=float(
self.df_gmt_no_na[self.optn_mrkt_pr_field][i_int]),
pricing_model_type='BlackScholes',
risk_free_rate_percent=float(self.df_gmt_no_na['RfRatePrct'][i_int]),
underlying_price=float(
self.df_gmt_no_na[self.underlying_pr_field][i_int]),
volatility_type='Implied',
option_price_side=self.option_price_side_for_IPA,
underlying_time_stamp=self.underlying_time_stamp))
for i_int in range(len(self.df_gmt_no_na))]
ipa_univ_requ_debug = [ # For debugging.
{"strike": float(self.strike),
"buy_sell": self.buy_sell,
"underlying_type": rd.content.ipa.financial_contracts.option.UnderlyingType.ETI,
"call_put": self.option_type,
"exercise_style": self.exercise_style,
"end_date": datetime.strptime(self.maturity, self.maturity_format).strftime('%Y-%m-%dT%H:%M:%SZ'), # '%Y-%m-%dT%H:%M:%SZ' for RD version 1.2.0 # self.maturity.strftime('%Y-%m-%dt%H:%M:%Sz') # datetime.strptime(self.maturity, '%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%dt%H:%M:%Sz')
"lot_size": 1,
"deal_contract": 1,
"time_zone_offset": 0,
"underlying_definition": {"instrument_code": self.underlying},
"pricing_parameters": {
"valuation_date": self.df_gmt_no_na.index[i_int].strftime('%Y-%m-%dT%H:%M:%SZ'), # '%Y-%m-%dT%H:%M:%SZ' for RD version 1.2.0 # '%Y-%m-%dt%H:%M:%Sz' # One version of rd wanted capitals, the other small. Here they are if you want to copy paste.
"report_ccy": self.curr,
"market_value_in_deal_ccy": float(
self.df_gmt_no_na[self.optn_mrkt_pr_field][i_int]),#
"pricing_model_type": 'BlackScholes',
"risk_free_rate_percent": float(self.df_gmt_no_na['RfRatePrct'][i_int]),
"underlying_price": float(
self.df_gmt_no_na[self.underlying_pr_field][i_int]),
"volatility_type": 'Implied',
"option_price_side": self.option_price_side_for_IPA,
"underlying_time_stamp": self.underlying_time_stamp}}
for i_int in range(len(self.df_gmt_no_na))]
ipa_univ_requ_debug_buckets = [i_rdf_bd_dbg for i_rdf_bd_dbg in [ipa_univ_requ_debug[j_int:j_int+self.search_batch_max] for j_int in range(0, len(ipa_univ_requ_debug), self.search_batch_max)]] # For debugging.
for i_str in ["ErrorMessage", "MarketValueInDealCcy", "RiskFreeRatePercent", "UnderlyingPrice", "Volatility"][::-1]: # We would like to keep a minimum of these fields in the Search Responce in order to construct following graphs.
request_fields = [i_str] + self.request_fields
i_int = 0
_request_fields = request_fields.copy()
no_of_ipa_calls = len([ipa_univ_requ[j_int:j_int+self.search_batch_max] for j_int in range(0, len(ipa_univ_requ), self.search_batch_max)])
if no_of_ipa_calls > 100 or self.debug:
print(f"There are {no_of_ipa_calls} to make. This may take a long while. If this is too long, please consider changing the `IPA_Equity_Vola_n_Greeeks` argument from {self.data_retrieval_interval} to a longer interval")
for enum, i_rdf_bd in enumerate([ipa_univ_requ[j_int:j_int+self.search_batch_max] for j_int in range(0, len(ipa_univ_requ), self.search_batch_max)]): # This list chunks our `ipa_univ_requ` in batches of `search_batch_max`
# IPA may sometimes come back to us saying that Implied VOlatilities cannot be computed. This can happen sometimes due to extreme Moneyness and closeness to expiration. To investigate these issues, we create this 1st try loop:
try:
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.
ipa_df_get_data_return = rd.content.ipa.financial_contracts.Definitions(
universe=i_rdf_bd, fields=request_fields).get_data()
except: # https://stackoverflow.com/questions/11520492/difference-between-del-remove-and-pop-on-lists
_request_fields.remove('ErrorMessage')
ipa_df_get_data_return = rd.content.ipa.financial_contracts.Definitions(
universe=i_rdf_bd, fields=request_fields).get_data()
except:
if self.debug:
print("request_fields")
print(request_fields)
print(f"ipa_univ_requ_debug_buckets[{enum}]")
display(ipa_univ_requ_debug_buckets[enum])
_request_fields.remove('ErrorMessage')
ipa_df_get_data_return = rd.content.ipa.financial_contracts.Definitions(
universe=i_rdf_bd, fields=request_fields).get_data()
_ipa_df_gmt_no_na = ipa_df_get_data_return.data.df
if i_int == 0:
ipa_df_gmt_no_na = _ipa_df_gmt_no_na.drop("ErrorMessage", axis=1) # We only keep "ErrorMessage" for debugging in self._IPA_df.
else:
ipa_df_gmt_no_na = pd.concat(
[ipa_df_gmt_no_na, _ipa_df_gmt_no_na.drop(labels=["ErrorMessage"], axis=1)],
ignore_index=True)
i_int += 1
time.sleep(1)
if self.debug:
print(i_int)
if not self.debug and no_of_ipa_calls > 100:
print(i_int)
ipa_df_gmt_no_na.index = self.df_gmt_no_na.index
ipa_df_gmt_no_na.columns.name = self.df_gmt_no_na.columns.name
There's a lot happening here, in the code cell above, but the gist of it is that we batch our request in buckets of `search_batch_max`, which is set to the max allowed by IPA by default (90) (well, the max is actually 100, but we give a little margin).
(v) We can now add the correlation and such analyses in the code cell below:
self.df = ipa_df_gmt_no_na.copy()
if self.corr:
# silense a PerformanceWarinng
import warnings
warnings.simplefilter(action='ignore', category=pd.errors.PerformanceWarning)
# Now add the new column for correnation
corr_df = self.df['OptionPrice'].ffill().rolling(window=63).corr(self.df['Volatility'].ffill())
self.df['3M(63WorkDay)MovCorr(StkprImpvola)'] = corr_df
if self.hist_vol:
# Resample data to daily frequency by taking the mean of intraday data
self.df_daily = self.df.resample(rule='B').mean()
for i in ["OptionPrice", "UnderlyingPrice"]:
# Forward fill it to get rid on NAs that represent a lack of movement in price
self.df_daily[i] = self.df_daily[i].ffill()
# Calculate the Historical Volatility on a 30 day moving window and implement it in main dataframe
self.df_daily[f'30DHist{i}DailyVolatility'] = self.df_daily[i].rolling(window=30).std()
self.df_daily[f'30DHist{i}DailyVolatilityAnnualized'] = self.df_daily[f'30DHist{i}DailyVolatility'] * np.sqrt(252)
self.df = pd.merge_asof(
self.df,
self.df_daily[[f'30DHist{i}DailyVolatility', f'30DHist{i}DailyVolatilityAnnualized']],
left_index=True, right_index=True, direction='backward')
(vi) Finally, we can enter all our data in `self`:
self._request_fields = _request_fields
self.rf_rate_prct = rf_rate_prct
self.ipa_univ_requ = ipa_univ_requ
self.ipa_df_gmt_no_na = ipa_df_gmt_no_na
return self
1.2.4. `graph`
Now, let's create a function to return a plotly figure of our data. We will need a few required fields, to make sure that relevant data is displayed, thus the `req_flds` argument:
def graph(
self,
title=None,
corr=True,
hist_vol=True,
df=None,
graph_template='plotly_dark',
req_flds=["OptionPrice", "MarketValueInDealCcy", "RiskFreeRatePercent", "UnderlyingPrice", "Volatility", "VolatilityPercent", "DeltaPercent", "GammaPercent", "RhoPercent", "ThetaPercent", "VegaPercent", "Leverage", "Gearing", "HedgeRatio", "DailyVolatility", "DailyVolatilityPercent", "Strike", "YearsToExpiry"]
):
Note that we would like to make this function 'generalised', in that someone with their own data frame should be able to use this function to output a similar graph, thus the `if df == None` loop:
# `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`.
if df == None:
df = self.df
if corr and self.corr:
req_flds.append("3M(63WorkDay)MovCorr(StkprImpvola)")
if hist_vol and self.hist_vol:
for i in ["OptionPrice", "UnderlyingPrice"]:
req_flds.append(f"30DHist{i}DailyVolatility")
req_flds.append(f"30DHist{i}DailyVolatilityAnnualized")
req_flds = list(dict.fromkeys(req_flds)) # remove potential duplicates in the list and keep only unique and different str elements while keepint the original order.
if title == None:
title = self.ipa_df_gmt_no_na.columns.name
df_graph = df[req_flds].fillna(np.nan).astype(float)
fig = px.line(df_graph)
fig.update_layout(
title=title,
template=graph_template)
fig.for_each_trace(
lambda t: t.update(
visible=True if t.name in df.columns[:1] else "legendonly"))
self.df_graph = df_graph
self.fig = fig
return self
1.2.5. `cross_moneyness`
Last but not least, what if we go through the above in a loop? Then we can get data (such as Implied Volatility) through several strikes at any one point in time! Let's try:
def cross_moneyness(
self,
smile_range=4
):
# We're going to append to our 3 lists:
undrlying_optn_ric_lst = []
strikes_lst = []
df_gmt_lst = []
df_lst = []
fig_lst = []
first_strike = self.strike
first_undrlying_optn_ric = self.undrlying_optn_ric
first_df_gmt = self.df_gmt
first_df = self.df
first_fig = self.fig
# Let's itterate down strikes up to the lower limit `smile_range`
for i in range(smile_range):
# try:
self.initiate(direction="-").get_data().graph()
if self.debug:
print("self.strike")
print(self.strike)
print("self._strike")
print(self._strike)
undrlying_optn_ric_lst.append(self.undrlying_optn_ric)
strikes_lst.append(self.strike)
df_gmt_lst.append(self.df_gmt)
df_lst.append(self.df)
fig_lst.append(self.fig)
# except:
# print(f"failed at strike {self.strike}")
# Up to now, we collected data per strike, going down, but we want a list from the lowest strike going up:
strikes_lst = strikes_lst[::-1] # `[::-1]` reverses the order of our list
undrlying_optn_ric_lst = undrlying_optn_ric_lst[::-1]
df_gmt_lst = df_gmt_lst[::-1]
df_lst = df_lst[::-1]
fig_lst = fig_lst[::-1]
# We skipped the strike we already collected data for before calling on `single_date_smile`
strikes_lst.append(first_strike)
undrlying_optn_ric_lst.append(first_undrlying_optn_ric)
df_gmt_lst.append(first_df_gmt)
df_lst.append(first_df)
fig_lst.append(first_fig)
# Now let's itterate up strikes up to the upper limit `smile_range`
self.strike=self._strike+self.atm_intervals
for i in range(smile_range):
# try:
self.initiate(direction="+").get_data().graph()
undrlying_optn_ric_lst.append(self.undrlying_optn_ric)
strikes_lst.append(self.strike)
df_gmt_lst.append(self.df_gmt)
df_lst.append(self.df)
fig_lst.append(self.fig)
# except:
# print(f"failed at strike {self.strike}")
return strikes_lst, undrlying_optn_ric_lst, df_gmt_lst, df_lst, fig_lst
The code cell above goes through all the code above iterating through different strikes, going up and down up to the `smile_range` limit, which is 4 by default.
2. CodeBook's `IPA_Equity_Vola_n_Greeks.ipynb` functional file
The above is quite nice, but rather complicated... Is there a way to streamline arguments to homogenise calls and uses of our `IPA_Equity_Vola_n_Greeeks` Class? Default arguments are extremely useful, but some arguments will have to be entered by users...
In come Codebook.
Within codebook, you can use the "User Interface Library" widgets to streamline code and make it look like an App! You can even have a look at them in the __Examples__ folder in Codebook:
You can find all these examples in "__Examples__/12. User Interface Library". I chose "Horizontal_Box.ipynb" as a very quick example above.
Let's get to our coding. Codebook is an environment like JupyteLab, and you can create your own folder in the navigational panel to the lsef; in which we can put our .py helper file described above, and we can import it in our Notebook (named IPA_Equity_Vola_n_Greeks.ipynb):
from IPA_Equity_Vola_n_Greeks_Class_023 import IPA_Equity_Vola_n_Greeeks
When using Codebook, we can use the library `refinitiv_widgets`:
from refinitiv_widgets import Select, MultiSelect, Button, TextFieldAutosuggest, Calendar, Select, Loader, TextField
Along with our favourite libraries:
%matplotlib inline
from datetime import datetime, timedelta
import plotly.graph_objects as go # `plotly` is a library used to render interactive graphs
import IPython
import ipywidgets as widgets
import copy
import pandas as pd
import numpy as np
from dataclasses import dataclass
Let's create two functions, one for the original graphs (e.g.: of Greeks), `Eqty_ATM_Optn_Vola_n_Greeks` and another if users have the patience to wait for the program to loop through the above through strike prices, `Eqty_ATM_Optn_Impli_Vol_Smile`:
2.1. `Eqty_ATM_Optn_Vola_n_Greeks`
def Eqty_ATM_Optn_Vola_n_Greeks(debug=False):
Let's start with the widgets. We need to define them first; and we'll use several in our case:
eqty_out = widgets.Output()
eqty_choice = TextFieldAutosuggest(placeholder='Equity Underlying', filters=['EQ', 'INDX'])
display(eqty_choice, eqty_out) # This displays the box
c_p_select_out = widgets.Output()
c_p_choice = Select(
placeholder='Call or Put?', width=300.0)
c_p_choice.data = [
{'value': i, 'label': i, 'items': []}
for i in ['Call', 'Put']]
# display(c_p_choice, c_p_select_out) # If you want to display each choice box one above the other, you can use this line. I chose `HBox`s instead.
b_s_select_out = widgets.Output()
b_s_choice = Select(
placeholder='Buy or Sell?', width=300.0)
b_s_choice.data = [
{'value': i, 'label': i, 'items': []}
for i in ['Buy', 'Sell']]
# display(b_s_choice, b_s_select_out) # If you want to display each choice box one above the other, you can use this line. I chose `HBox`s instead.
# Display both choice boxes (`c_p_choice` & `b_s_choice`) horisontally, next to eachother.
display(widgets.HBox([c_p_choice, b_s_choice])) # This will display both choice boxes horisontally, next to eachother.
report_ccy_select_out = widgets.Output()
report_ccy_choice = Select(
placeholder='Report Currency?', width=300.0)
report_ccy_choice.data = [
{'value': i, 'label': i, 'items': []}
for i in ['EUR', 'USD', 'GBP', 'JPY']]
# display(report_ccy_choice, report_ccy_select_out) # If you want to display each choice box one above the other, you can use this line. I chose `HBox`s instead.
option_price_side_select_out = widgets.Output()
option_price_side_choice = Select(
placeholder='Option Price Side?', width=300.0)
option_price_side_choice.data = [
{'value': i, 'label': i, 'items': []}
for i in ['Bid', 'Ask']]
# display(option_price_side_choice, option_price_side_select_out) # If you want to display each choice box one above the other, you can use this line. I chose `HBox`s instead.
# Display both choice boxes (`report_ccy_choice` & `option_price_side_choice`) horisontally, next to eachother.
display(widgets.HBox([report_ccy_choice, option_price_side_choice]))
print("\n")
print("Please enter the RIC of the reference Risk Free Rate, e.g.: for `.SPX`, go with `USDCFCFCTSA3M=`; for `.STOXX50E`, go with `EURIBOR3MD=`")
rsk_free_rate_prct_out = widgets.Output()
rsk_free_rate_prct_choice = TextFieldAutosuggest(placeholder='Risk Free Instrument RIC', filters=['FX'])
display(rsk_free_rate_prct_choice, rsk_free_rate_prct_out) # This displays the box
print("Maturity:")
calendar = Calendar(
max=(datetime.now() + timedelta(days=30*5)).strftime('%Y-%m-%d'),
min=(datetime.now() - timedelta(days=30*5)).strftime('%Y-%m-%d'))
display(calendar)
widgets.DatePicker(
description='Start date:', continuous_update=False, max=(datetime.now() + timedelta(days=30*5)).strftime('%Y-%m-%d'))
# create widgets
button = Button('Create/Update Graph')
button_output = widgets.Output()
Now, let's add a loader, so that it looks good while it's loading:
loader = Loader(visible=False)
loader.visible = not loader.visible
Now let's create our click handler, which will initiate with the click of the boutton `Button('Create/Update Graph')`:
# create click handler
def click_handler(a):
with button_output:
Let's clear the screen to make sure nothing is showing at the moment. This is to make sure nothing is displaying even if the user already ran the code once and is looking at a graph that (s)he wants updated.
IPython.display.clear_output(wait=True)
Now let's display the loader:
display(loader)
Let's check that the user selected appropriate data using the widgets above:
if c_p_choice.value == "" or eqty_choice.value == "" or rsk_free_rate_prct_choice.value == "" or calendar.value == []:
IPython.display.clear_output(wait=True)
raise ValueError("Please make sure to complete all fields before running the program.")
else:
print("This may take a few minutes...")
Now we can let it all run!
ipa_data = IPA_Equity_Vola_n_Greeeks(
debug=debug,
underlying=eqty_choice.value,
strike=None,
maturity=calendar.value[0], # "2024-03-15", # calendar.value,
maturity_format = '%Y-%m-%d', # e.g.: '%Y-%m-%d', '%Y-%m-%d %H:%M:%S' or '%Y-%m-%dT%H:%M:%SZ'
option_type = c_p_choice.value,
buy_sell = b_s_choice.value,
curr = report_ccy_choice.value,
exercise_style = 'EURO',
option_price_side = option_price_side_choice.value,
underlying_time_stamp = 'Close',
resample = '10min', # You can consider this the 'bucket' or 'candles' from which calculations will be made.
rsk_free_rate_prct = rsk_free_rate_prct_choice.value, # for `".SPX"`, I go with `'USDCFCFCTSA3M='`; for `".STOXX50E"`, I go with `'EURIBOR3MD='`
rsk_free_rate_prct_field = 'TR.FIXINGVALUE' # for `".SPX"`, I go with `'TR.FIXINGVALUE'`; for `".STOXX50E"`, I go with `'TR.FIXINGVALUE'` too.
).initiate().get_data()
fig, worked = ipa_data.graph(
title=ipa_data.ipa_df_gmt_no_na.columns.name).fig, True
if worked:
IPython.display.clear_output(wait=True)
fig.show()
strikes_lst, undrlying_optn_ric_lst, df_gmt_lst, df_lst, fig_lst = ipa_data.cross_moneyness(smile_range=8)
# refister click handler for button
print("\n")
button.on_click(click_handler)
display(button)
# display our widgets
display(button_output)
2.2. `Eqty_ATM_Optn_Impli_Vol_Smile`
We can now do the same for our Smile function:
def Eqty_ATM_Optn_Impli_Vol_Smile(debug=False):
eqty_out = widgets.Output()
eqty_choice = TextFieldAutosuggest(placeholder='Equity Underlying', filters=['EQ', 'INDX'])
display(eqty_choice, eqty_out) # This displays the box
c_p_select_out = widgets.Output()
c_p_choice = Select(
placeholder='Call or Put?', width=300.0)
c_p_choice.data = [
{'value': i, 'label': i, 'items': []}
for i in ['Call', 'Put']]
# display(c_p_choice, c_p_select_out) # If you want to display each choice box one above the other, you can use this line. I chose `HBox`s instead.
b_s_select_out = widgets.Output()
b_s_choice = Select(
placeholder='Buy or Sell?', width=300.0)
b_s_choice.data = [
{'value': i, 'label': i, 'items': []}
for i in ['Buy', 'Sell']]
# display(b_s_choice, b_s_select_out) # If you want to display each choice box one above the other, you can use this line. I chose `HBox`s instead.
# Display both choice boxes (`c_p_choice` & `b_s_choice`) horisontally, next to eachother.
display(widgets.HBox([c_p_choice, b_s_choice])) # This will display both choice boxes horisontally, next to eachother.
report_ccy_select_out = widgets.Output()
report_ccy_choice = Select(
placeholder='Report Currency?', width=300.0)
report_ccy_choice.data = [
{'value': i, 'label': i, 'items': []}
for i in ['EUR', 'USD', 'GBP', 'JPY']]
# display(report_ccy_choice, report_ccy_select_out) # If you want to display each choice box one above the other, you can use this line. I chose `HBox`s instead.
option_price_side_select_out = widgets.Output()
option_price_side_choice = Select(
placeholder='Option Price Side?', width=300.0)
option_price_side_choice.data = [
{'value': i, 'label': i, 'items': []}
for i in ['Let Program Choose', 'Bid', 'Ask']]
# display(option_price_side_choice, option_price_side_select_out) # If you want to display each choice box one above the other, you can use this line. I chose `HBox`s instead.
# Display both choice boxes (`report_ccy_choice` & `option_price_side_choice`) horisontally, next to eachother.
display(widgets.HBox([report_ccy_choice, option_price_side_choice]))
print("\n")
print("Please enter the RIC of the reference Risk Free Rate, e.g.: for `.SPX`, go with `USDCFCFCTSA3M=`; for `.STOXX50E`, go with `EURIBOR3MD=`")
rsk_free_rate_prct_out = widgets.Output()
rsk_free_rate_prct_choice = TextFieldAutosuggest(placeholder='Risk Free Instrument RIC', filters=['FX'])
display(rsk_free_rate_prct_choice, rsk_free_rate_prct_out) # This displays the box
smile_rnge_select_out = widgets.Output()
smile_rnge_choice = Select(
placeholder='Smile Moneyness Range', width=300.0)
smile_rnge_choice.data = [
{'value': str(i), 'label': str(i), 'items': []}
for i in range(5)]
display(smile_rnge_choice, smile_rnge_select_out)
print("Maturity (note that most Options mature on the third Friday of the month):")
calendar = Calendar(
max=(datetime.now() + timedelta(days=30*5)).strftime('%Y-%m-%d'),
min=(datetime.now() - timedelta(days=30*5)).strftime('%Y-%m-%d'))
display(calendar)
widgets.DatePicker(
description='Start date:', continuous_update=False, max=(datetime.now() + timedelta(days=30*5)).strftime('%Y-%m-%d'))
# create widgets
button = Button('Create/Update Graph')
button_output = widgets.Output()
loader = Loader(visible=False)
loader.visible = not loader.visible
# create click handler
def click_handler(a):
with button_output:
IPython.display.clear_output(wait=True)
display(loader)
if c_p_choice.value == "" or eqty_choice.value == "" or rsk_free_rate_prct_choice.value == "" or calendar.value == []:
IPython.display.clear_output(wait=True)
raise ValueError("Please make sure to complete all fields before running the program.")
else:
if debug:
print(f"eqty_choice.value: {eqty_choice.value}")
print(f"calendar.value[0]: {calendar.value[0]}")
print(f"c_p_choice.value: {c_p_choice.value}")
print(f"rsk_free_rate_prct_choice.value: {rsk_free_rate_prct_choice.value}")
print("This may take a few minutes...")
# Above, we created an option for the `option_price_side_choice` to allow users to not choose a price side.
# In the if statement below, we translate this choice to one that the `IPA_Equity_Vola_n_Greeeks` funciton will understand.
if option_price_side_choice.value == "Let Program Choose":
option_price_side_choice_val = None
else:
option_price_side_choice_val = option_price_side_choice.value
ipa_data = IPA_Equity_Vola_n_Greeeks(
debug=debug,
underlying=eqty_choice.value,
strike=None,
maturity=calendar.value[0], # "2024-03-15", # calendar.value,
maturity_format = '%Y-%m-%d', # e.g.: '%Y-%m-%d', '%Y-%m-%d %H:%M:%S' or '%Y-%m-%dT%H:%M:%SZ'
option_type = c_p_choice.value,
buy_sell = b_s_choice.value,
curr = report_ccy_choice.value,
exercise_style = 'EURO',
option_price_side = option_price_side_choice_val,
underlying_time_stamp = 'Close',
resample = '10min', # You can consider this the 'bucket' or 'candles' from which calculations will be made.
rsk_free_rate_prct = rsk_free_rate_prct_choice.value, # for `".SPX"`, I go with `'USDCFCFCTSA3M='`; for `".STOXX50E"`, I go with `'EURIBOR3MD='`
rsk_free_rate_prct_field = 'TR.FIXINGVALUE' # for `".SPX"`, I go with `'TR.FIXINGVALUE'`; for `".STOXX50E"`, I go with `'TR.FIXINGVALUE'` too.
).initiate().get_data()
sngl_fig, worked = ipa_data.graph(
title=ipa_data.ipa_df_gmt_no_na.columns.name).fig, True
strikes_lst, undrlying_optn_ric_lst, df_gmt_lst, df_lst, fig_lst = ipa_data.cross_moneyness(
smile_range=int(smile_rnge_choice.value))
if debug:
print(strikes_lst)
display(fig_lst[0])
display(fig_lst[-1])
volatility_result = pd.concat([i.Volatility for i in df_lst], axis=1, join="outer")
volatility_result.columns = [str(int(i)) for i in strikes_lst]
volatility_result.index.name = "ImpliedVolatilities"
volatility_result
df = volatility_result.copy()
# Assuming df is your DataFrame and 'timestamp' is your time column
df['timestamp'] = df.index
df.timestamp = pd.to_datetime(df.timestamp)
df.set_index('timestamp', inplace=True)
# Resample to daily data and compute daily averages
daily_df = df.resample('7D').mean()
# Fill NA/NaN values using the specified method
daily_df_filled = daily_df.fillna(np.nan).astype(float).dropna()
daily_df_filled.index = [str(i) for i in daily_df_filled.index]
daily_df_filled = daily_df_filled.T
# Let's go back to the `sngl_fig` figure created above
if worked:
IPython.display.clear_output(wait=True)
sngl_fig.show()
# Now let's get back to our Smile figure:
smile_fig = go.Figure()
# Add traces (lines) for each column
for col in daily_df_filled.columns:
smile_fig.add_trace(
go.Scatter(
x=daily_df_filled.index,
y=daily_df_filled[col],
mode='lines', name=col))
smile_fig.update_layout(
title="Volatility Smiles",
template="plotly_dark")
smile_fig.show()
# refister click handler for button
print("\n")
button.on_click(click_handler)
display(button)
# display our widgets
display(button_output)
3. Output
We can ask for an Equity (or an Equity Index):
Then select other arguments with drop downs:
Write an intrument you'd like as a risk free metric:
Pick a time for which we'd like our data. Let's test for an expired option:
Our loader should show up:
And boom! Now let's look at our Implied Volatility:
This is extremely powerful. See how it found the expired Option nearest to At The Money for the price of the underlying at the date chosen in our calendar widget above? Then the data was sent to IPA to calculate insightful data to allow access in this graph. This is a very information-rich interactive graph. But that's not all! Let' test Similes!
We can (double) click on any one date (on the right of the last graph) for which we are interested in a smile:
4. Conclusion
The Art and Science of Financial Analytics
This Python helper file is more than just code; it’s a fusion of art and science, intuition and logic, experience and innovation. It empowers traders to navigate the complexities of the financial markets with confidence and clarity.
