nsedt.derivatives.options

  1""" 
  2get data for Options
  3"""
  4import logging
  5import urllib
  6
  7from datetime import datetime
  8from nsedt import utils
  9from nsedt.resources import constants as cns
 10from nsedt.utils import data_format, exceptions
 11
 12log = logging.getLogger("root")
 13
 14
 15def get_option_chain(
 16    symbol: str,
 17    strike_price: str = None,
 18    expiry_date: str = None,
 19    response_type="panda_df",
 20):
 21    """Get option data of stock and indices
 22
 23    Args:
 24        symbol (str): symbol name
 25        strike_price (str, optional): strike price to apply filter on price. Defaults to None.
 26        expiry_date (str, optional): expiry date to apply filter on date. Defaults to None.
 27        response_type (str, optional): response_type panda_df or json . Defaults to "panda_df".
 28
 29    Returns:
 30        Pandas DataFrame: df containing option data
 31    or
 32        Json: json containing option data
 33
 34    """
 35    params = {}
 36    cookies = utils.get_cookies()
 37    base_url = cns.BASE_URL
 38    symbol = utils.get_symbol(symbol=symbol, get_key="derivatives")
 39
 40    if symbol in cns.INDICES:
 41        event_api = cns.OPTIONS_PRICE_INDICES
 42    else:
 43        event_api = cns.OPTIONS_PRICE_EQUITIES
 44
 45    params["symbol"] = symbol
 46
 47    url = base_url + event_api + urllib.parse.urlencode(params)
 48    data = utils.fetch_url(url, cookies, response_type="json")
 49
 50    if data is None or data == {}:
 51        log.error("symbol is wrong or unable to access API")
 52        raise ValueError
 53
 54    # filtering data
 55
 56    if strike_price and expiry_date:
 57        filtered_data = [
 58            record
 59            for record in data["records"]["data"]
 60            if record["strikePrice"] == strike_price
 61            and record["expiryDate"]
 62            == datetime.strptime(expiry_date, "%d-%m-%Y").strftime("%d-%b-%Y")
 63        ]
 64
 65    elif strike_price:
 66        filtered_data = [
 67            record
 68            for record in data["records"]["data"]
 69            if record["strikePrice"] == strike_price
 70        ]
 71    elif expiry_date:
 72        filtered_data = [
 73            record
 74            for record in data["records"]["data"]
 75            if record["expiryDate"]
 76            == datetime.strptime(expiry_date, "%d-%m-%Y").strftime("%d-%b-%Y")
 77        ]
 78    else:
 79        filtered_data = data["records"]["data"]
 80
 81    return data_format.option_chain(
 82        filtered_data,
 83        response_type=response_type,
 84    )
 85
 86
 87def get_option_chain_expdate(symbol: str) -> list:
 88    """get option  expiry date for stock and indices
 89
 90    Args:
 91        symbol (str): symbol name
 92
 93    Returns:
 94        list: expiry date in list("%d-%m-%Y" format)
 95    """
 96    params = {}
 97    cookies = utils.get_cookies()
 98    base_url = cns.BASE_URL
 99
100    if symbol in cns.INDICES:
101        event_api = cns.OPTIONS_PRICE_INDICES
102    else:
103        event_api = cns.OPTIONS_PRICE_EQUITIES
104
105    params["symbol"] = symbol
106
107    url = base_url + event_api + urllib.parse.urlencode(params)
108    data = utils.fetch_url(url, cookies, response_type="json")
109    ret = []
110    expiry_dates = data.get("records").get("expiryDates")
111    if expiry_dates is None:
112        log.error("expiry_dates is None, symbol is wrong or unable to access API")
113        return []
114    for expiry_date in expiry_dates:
115        ret.append(datetime.strptime(expiry_date, "%d-%b-%Y").strftime("%d-%m-%Y"))
116    return ret
117
118def get_historical_option_data(
119    symbol: str,
120    start_date: str,
121    end_date: str,
122    option_type: str,
123    strike_price: str,
124    year : str,
125    expiry_date: str,
126    response_type: str = "panda_df",
127    columns_drop_list: list = None,
128):
129    """
130    Get historical data for option chain for a given expiry
131    Args:
132        symbol (str): valid scrip name
133        start_date (str): in %d-%m-%Y format
134        end_date (str): in %d-%m-%Y format
135        option_type (str): CE or PE.
136        strike_price (str): valid integer.
137        year (str): in %Y format eg 2024.
138        expiry_date (str): in %d-%m-%Y format
139        response_type (str, optional): either json or pand_df. Defaults to "panda_df".
140        columns_drop_list (list, optional): list of columns to skip. Defaults to None.
141
142    Returns:
143        _type_: either json or pandas df. Defaults to pandas_df
144    """
145    cookies = utils.get_cookies()
146    base_url = cns.BASE_URL
147    event_api = cns.FNO_HISTORY
148    symbol = utils.get_symbol(symbol=symbol, get_key="derivatives")
149
150    if option_type not in ["CE", "PE"]:
151        raise ValueError("Option type must be either CE or PE")
152
153    try:
154        expiry_date = datetime.strptime(expiry_date, "%d-%m-%Y").strftime("%d-%b-%Y")
155    except Exception as e:
156        raise ValueError("Please give expiry date in %d-%b-%Y format") from e
157
158    params = {
159        "symbol": symbol,
160        "from": start_date,
161        "to": end_date,
162        "instrumentType": "OPTSTK",
163        "optionType": option_type,
164        "expiryDate": expiry_date,
165        "strikePrice": strike_price,
166        "year": year,
167    }
168    url = base_url + event_api + urllib.parse.urlencode(params)
169    data = utils.fetch_url(url, cookies, response_type="json")
170
171    if data["data"] == []:
172        raise exceptions.DateStrikePriceOutofRange()
173
174    return data_format.derivaties_options(
175        data,
176        response_type=response_type, columns_drop_list=columns_drop_list,
177    )