get_lots_of_history function for the Pyhton LSEG Data Library

danieluphromes

Could you please provide some function using the Python LDL's get_history for lots of data, specifically, lots of instruments?

[Deleted User]

Does this work? If you want to loop it, you can do so similarly to in this question.

.

import datetime as dt
import pandas as pd
import warnings

import lseg.data as ld
ld.open_session()

def get_lots_of_history(
    req_univ: list[str],
    fields=None,
    start=dt.datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
    end=(dt.datetime.now() - dt.timedelta(days=7)).strftime('%Y-%m-%dT%H:%M:%S'),
    adjustments=None,
    count=None,
    parameters=None,
    header_type=ld.HeaderType.TITLE,
    batch_of: int = 50,
    warns: bool = True,
    messages: bool = True):

    err = {'Error': [], 'ErrorUniverse': []}

    if not warns:
        warnings.filterwarnings("ignore", category=FutureWarning, module="lseg.data._tools._dataframe")

    def Chunks(lst, n):
        """Yield successive n-sized chunks from lst."""
        for i in range(0, len(lst), n):
            yield lst[i:i + n]

    df_list = []

    for i, j in enumerate(Chunks(lst=req_univ, n=batch_of)):

        if messages:
            print(f"Batch of {batch_of} requests no. {str(i+1)}/{str(len([i for i in Chunks(req_univ, batch_of)]))} started")

        try:
            _df = ld.get_history(
                universe=j,
                fields=fields,
                start=start,
                end=end,
                adjustments=adjustments,
                count=count,
                parameters=parameters,
                header_type=header_type)
            df_list.append(_df)
            df = pd.concat(df_list, ignore_index=True)
        except ld.errors.LDError as e:
            print("There was an error, please note that the following were not added to the df:")
            # print("LDError:")
            print(e)
            err['Error'].append(e)
            err['ErrorUniverse'].append(j)

        if messages:
            print(f"Batch of {batch_of} requests no. {str(i+1)}/{str(len([i for i in Chunks(req_univ, batch_of)]))} ended")

    return df, err

.

If you get an error, you can find info about it with `err['Error'][0].__dict__`. The object `err['ErrorUniverse']` will give you all the universe items missed in the call.