<jupyter_start><jupyter_text>Processing Refinitiv Marketdata   Imports<jupyter_code>import s3fs
import bz2
import boto3
import pandas as pd
import io
SRC_BUCKET='epython'
SRC_PATH='mktdata/refinitiv/raw_bars/'
DST_BUCKET='epython-marketdata'
s3 = s3fs.S3FileSystem()
inst_type = 'IR'<jupyter_output><empty_output><jupyter_text>Getting files by instrument<jupyter_code>def get_files_for_inst(inst_type):
    client = boto3.client('s3')
    paginator = client.get_paginator('list_objects')
    
    page_iterator = paginator.paginate(
        Bucket=SRC_BUCKET,
        Prefix=SRC_PATH)

    for page in page_iterator:
        for content in page['Contents']:
            key = content['Key']
            if key.endswith('bz2') and f'/{inst_type}=' in key:
                yield key<jupyter_output><empty_output><jupyter_text>Let's see a few examples of **IR** files<jupyter_code>### paths = get_files_for_inst('IR')
for _ in range(5):
    print(next(paths))<jupyter_output>mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=20y(Ric_AUDQM3AB20Y=TWEB).csv.bz2
mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=25y(Ric_AUDQM3AB25Y=FMD).csv.bz2
mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=25y(Ric_AUDQM3AB25Y=TWEB).csv.bz2
mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=2y(Ric_AUDQM3AB2Y=BGCP).csv.bz2
mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=2y(Ric_AUDQM3AB2Y=FMD).csv.bz2<jupyter_text>Translate to destination path To help DataLake partition with appropriate columns, we use *Key=Value* subdirectories.<jupyter_code>def get_dest_path(inst_type, src_path):
    parts = src_path.split('/')
    date, filename = parts[3], parts[4]
    if inst_type == 'IR':
        _, ccy, curve, tenor = filename.split('(', 1)[0].split('=')
        keys = [('date', date), ('ccy', ccy), ('curve', curve), ('tenor', tenor)]
    else:
        ## TODO:
        ## Perhaps many instruments has the same format as IR, but need to check
        raise NotImpelmentedError(f'{isnt_type} not yet implemented')
                                  
    subdir = '/'.join('='.join(x) for x in keys)
    return f'refinitiv/{subdir}/{filename}'<jupyter_output><empty_output><jupyter_text>And here's an example<jupyter_code>get_dest_path('IR', 'mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=12y(Ric_AUDQM3AB12Y=FMD).csv.bz2')<jupyter_output><empty_output><jupyter_text>Process the file content<jupyter_code>def process_file(path):
    with s3.open(path) as f:
        data = f.read()

    data = bz2.decompress(data).decode()
    # Nothing fancy for now. Just cut off the '#' in front of the header
    # as the Crawler thinks it's a comment
    return data[1:]<jupyter_output><empty_output><jupyter_text>Visulise the dataNot required, but just for our sanity, let's visualise the data<jupyter_code>def visualise_data(data):
    buffer = io.StringIO(data)
    return pd.read_csv(buffer)

data = process_file('epython/mktdata/refinitiv/raw_bars/20201015/IR=AUD=Swap_6M=25y(Ric_AUDSM6AB25Y=TWEB).csv.bz2')
visualise_data(data)<jupyter_output><empty_output><jupyter_text>End to End  IR Insturments<jupyter_code>paths = get_files_for_inst('IR')

inst_type = 'IR'
for i, path in enumerate(paths):
    src_path = SRC_BUCKET + '/' + path
    print(f'processing file {i}')
    print(f'reading from file: {src_path}')
    data = process_file(src_path)
    
    dst_path = DST_BUCKET + '/' + get_dest_path(inst_type, path)
    compressed = bz2.compress(data.encode())
    print(f'writing to file: {dst_path}')
    with s3.open(dst_path, 'wb') as f:
        f.write(compressed)

    # When we are happy with the results,
    # we take out this and it'll run for all
    if i > 5:
        break<jupyter_output>processing file 0
reading from file: epython/mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=12y(Ric_AUDQM3AB12Y=FMD).csv.bz2
writing to file: epython-marketdata/refinitiv/date=20201014/ccy=AUD/curve=Swap_3M/tenor=12y/IR=AUD=Swap_3M=12y(Ric_AUDQM3AB12Y=FMD).csv.bz2
processing file 1
reading from file: epython/mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=12y(Ric_AUDQM3AB12Y=TWEB).csv.bz2
writing to file: epython-marketdata/refinitiv/date=20201014/ccy=AUD/curve=Swap_3M/tenor=12y/IR=AUD=Swap_3M=12y(Ric_AUDQM3AB12Y=TWEB).csv.bz2
processing file 2
reading from file: epython/mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=15y(Ric_AUDQM3AB15Y=FMD).csv.bz2
writing to file: epython-marketdata/refinitiv/date=20201014/ccy=AUD/curve=Swap_3M/tenor=15y/IR=AUD=Swap_3M=15y(Ric_AUDQM3AB15Y=FMD).csv.bz2
processing file 3
reading from file: epython/mktdata/refinitiv/raw_bars/20201014/IR=AUD=Swap_3M=15y(Ric_AUDQM3AB15Y=TWEB).csv.bz2
writing to file: epython-marketdata/refinitiv/date=20201014/c[...]<jupyter_text>Checking the resultLet's read the results back just to make sure it is okay<jupyter_code>def read_data(path):
    with s3.open(path, 'rb') as f:
        data = f.read()

    return visualise_data(bz2.decompress(data).decode())
    
path = 'epython-marketdata/refinitiv/date=20201014/ccy=AUD/curve=Swap_3M/tenor=20y/IR=AUD=Swap_3M=20y(Ric_AUDQM3AB20Y=FMD).csv.bz2'
read_data(path)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12710207274149393, 'fraction_numerical': 0.05220962064919828, 'mean_word_length': 5.133093525179856, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '450456', 'license': 'MIT', 'n_tokens_mistral': 2266, 'n_tokens_neox': 2064, 'n_words': 323, 'path': 'notebooks/refinitiv.ipynb', 'repo_name': 'juggler99/epython-showcase', 'size': '5114'}