finsec_fund_analysis_step03.py

# --- Functions for mapping CUSIPs to Tickers ---
def write_cusips_to_txt(cusip_mappings, cusip_mapping_file = 'cusips.txt'):
    '''
    Write cusip mappings to txt. Minimizing the need for API calls. 
    '''
    with open(cusip_mapping_file, 'w') as f:
        f.write(str(cusip_mappings))
    return

def read_cusips_from_txt(cusip_mapping_file='cusips.txt'):
    '''
    Read cusip mappings from txt. Minimizing the need for API calls.
    '''
    try:
        with open(cusip_mapping_file) as f:
            cusip_mappings = eval(f.read())
    except:
        return []
    return cusip_mappings

def map_cusips_openfigi(cusips:list):
    '''
    Splits all openFIGI API calls down to lots of 100 so that they can be processed (maximum number of cusips in a POST is 100). 
    Returns cusip_mappings which is a list of tuples containing CUSIP and Ticker.
    '''
    cusips_lots = [cusips[x:x+100] for x in range(0, len(cusips), 100)]
    mappings = []
    for index, lot in enumerate(cusips_lots):
        try:
            query = [{"idType": "ID_CUSIP", "idValue": str(cusip)} for cusip in lot]
            open_figi_resp = json.loads(requests.post(OPEN_FIGI_API, json=query, headers=HEADERS).text)
            tickers = []
            for resp in open_figi_resp:
                if resp.get("warning"):
                    tickers.append('')
                else:
                    tickers.append(resp["data"][0].get('ticker'))
            mappings.append(list(zip(lot, tickers)))
        except:
            print("Exception caught on lot: {}".format(index))
            time.sleep(6)
        time.sleep(0.26)
    mappings = [item for sublist in mappings for item in sublist]   # Flatten list of lists to single list of tuples.
    no_cusip_mappings = [item for item in mappings if item[1] == '']
    cusip_mappings = [item for item in mappings if item[1] != '']
    return cusip_mappings, no_cusip_mappings

def map_cusips_cns(cusips, cns_file):
    '''
    Map cusips based on 'Fails to Deliver' Continuous Net Settlement (CNS) data. Fails to Deliver data contains the list of all shares that failed to be delivered as of a particular settlement date as recorded in the National Securities Clearing Corporation's ("NSCC") Continuous Net Settlement (CNS) system aggregated over all NSCC members. Typically this list contains thousands of potential mappings as it is common place for shares to fail to deliver. 

    Further information can be found here: https://www.sec.gov/data/foiadocsfailsdatahtm
    '''
    mapped_cusips_cns = []
    
    with open(cns_file) as f:
        cns = f.read().split('\n')
        for line in cns:
            components = line.split('|')
            if len(components)>3:
                mapped_cusips_cns.append((components[1],components[2]))
    mapped_cusips_cns = [t for t in (set(tuple(i) for i in mapped_cusips_cns))]   # Distinct mappings

    mapped_cusips_cns = [cns for cusip in cusips for cns in mapped_cusips_cns if cusip == cns[0]]     # Check if the list of unkowns is covered by cns
    mapped_cusips_cns_cusip_list = [item[0] for item in mapped_cusips_cns]  # Look at the list of CUSIPs where we have found mappings. 
    no_cusip_mappings = set(cusips) - set(mapped_cusips_cns_cusip_list)
    no_cusip_mappings = [(item,'') for item in no_cusip_mappings]

    return mapped_cusips_cns, no_cusip_mappings

# --- Get list of CUSIPs from all manager holdings ---
cusips = list(dict.fromkeys(all_this_qtr_holdings['CUSIP'].tolist() + all_last_qtr_holdings['CUSIP'].tolist()))

# --- Remove any CUSIPs that have been previously mapped (contained within the cusips text file) ---
existing_cusip_mappings = read_cusips_from_txt()
existing_cusip = [item[0] for item in existing_cusip_mappings]
cusips = list(set(cusips) - set(existing_cusip))

# --- Check openFIGI then SEC CNS records for CUSIP mappings ---
openfigi_cusip_mappings, no_cusip_mappings = map_cusips_openfigi(cusips)
cns_cusip_mappings, no_cusip_mappings = map_cusips_cns(no_cusip_mappings, 'cnsfails202301a')   # Pass in any cusips that had no openFIGI mappings, check if the cns function can find tickers for these. 
cusip_mappings = openfigi_cusip_mappings + cns_cusip_mappings + existing_cusip_mappings # Bring all of the cusip->ticker mappings together (openFIGI, CNS and any pre-existing mappings).
write_cusips_to_txt(cusip_mappings) # Write/overwrite the cusip mappings to a text file.