For Ticker MICR.NS we are are not able to get the financial statement date while accessing the workspace using python from the below code.
Can you able to assist us on this.
######eikon_download.py##############
import eikon as ek
import pandas as pd
import yaml
import time
import re
from itertools import repeat
from multiprocessing import Pool
import lseg.data as ld
Initialize session and set Eikon key
ld.open_session()
ek.set_app_key('980d59c17ef047d08e04fd4de1294fd7bf6966fd')
class EikonDownload:
"""
Eikon downloader.
Reads a YAML file for instruments, fields, and parameters to extract data from Eikon.
"""
def __init__(self, download_config):
"""
Initialize with input from YAML file.
:param download_config: Path to YAML file.
"""
self._file_name = download_config
self._yaml_data = None
self.load_config(download_config)
super().__init__()
def set_instrument(self, instrument_list):
"""Set the instrument list to download."""
self._yaml_data['instrument'] = instrument_list
def set_field(self, field_list):
"""Set the list of fields to download from Eikon."""
self._yaml_data['field'] = field_list
def get_download_config(self):
"""
Get instruments, fields, and parameters from the config.
:return: (instrument list, field list, parameters)
"""
return (
self._yaml_data['instrument'],
self._yaml_data['field'],
self._yaml_data['parameter']
)
def load_config(self, file_name):
"""
Load YAML file containing download inputs.
:param file_name: YAML file path.
:return: Tuple (success: bool, error: Exception or None)
"""
try:
with open(file_name, 'r') as stream:
self._yaml_data = yaml.safe_load(stream)
except yaml.YAMLError as ex:
self._yaml_data = None
return False, ex
return True, None
@staticmethod
def download(instrument_list, field_list, parameters, rename_list, key, yaml_file, logger):
"""
Download data as a DataFrame.
:return: (DataFrame, error list)
"""
err_list, data_list = [], []
total_instruments = len(instrument_list)
logger.info(f'Total instruments in {key}: {yaml_file} : {total_instruments}')
counter = 0
for instrument in instrument_list:
succeed = False
while not succeed:
try:
#ld.get_data()
data_tuple = ek.get_data(instrument, field_list, parameters)
counter += 1
logger.info(f'Instrument {counter}/{total_instruments}: {instrument}')
data_list.append(data_tuple[0])
err_list.append(data_tuple[1]) if data_tuple[1] is not None else None
succeed = True
except Exception as e:
logger.info(e)
time.sleep(0.2)
# Remove duplicate DataFrames
def are_equal(df1, df2):
return df1.equals(df2)
unique_data = []
for df in data_list:
if not any(are_equal(df, u_df) for u_df in unique_data):
unique_data.append(df)
logger.info(f"Data list deduplication count: {len(unique_data)}")
# Reset indices
unique_data = [df.reset_index(drop=True) for df in unique_data]
logger.info(f"Before filtering structure: count={len(unique_data)}")
# Identify the most common structure
def most_common_structure(dfs):
structures = [tuple(df.columns) for df in dfs]
return max(set(structures), key=structures.count)
common_structure = most_common_structure(unique_data)
filtered_data = [df for df in unique_data if tuple(df.columns) == common_structure]
logger.info(f"After filtering structure: count={len(filtered_data)}")
result = pd.concat(filtered_data, ignore_index=True)
logger.info(f"Final data:\n{result}")
if rename_list:
result.columns = rename_list
#result.set_axis(rename_list, axis='columns', inplace=True)
return result, err_list
def download_using_config(self, key, yaml_file, logger):
"""
Download data as a DataFrame using the config file.
:return: (DataFrame, error list)
"""
result, err = EikonDownload.download(
self._yaml_data['instrument'],
self._yaml_data['field'],
self._yaml_data['parameter'],
self._yaml_data['rename'],
key,
yaml_file,
logger
)
logger.info("Completed download")
return result, err
