path: root/python/parse_emails.py

import pandas as pd
import re
import os
import pdb
from download_emails import update_emails
import datetime
import logging
import pickle
import sys


logging.basicConfig(filename=os.path.join(os.getenv("LOG_DIR"), 'emails_parsing.log'),
                    level=logging.WARNING,
                    format='%(asctime)s %(message)s')

def makedf(r, indextype, quote_source):
    if indextype=='IG':
        cols = ['strike', 'rec_bid', 'rec_offer', 'delta_rec', 'pay_bid',
                'pay_offer', 'delta_pay', 'vol']
    else:
        cols = ['strike', 'rec_bid', 'rec_offer', 'delta_rec', 'pay_bid',
                'pay_offer', 'delta_pay', 'vol', 'price_vol']
    if quote_source == "BAML":
        cols.append('gamma')
    if quote_source == "GS":
        cols.append("tail")
    df = pd.DataFrame.from_records(r, columns = cols)
    for col in ['delta_rec', 'delta_pay', 'vol', 'price_vol', 'gamma', 'tail']:
        if col in df:
            df[col] = df[col].str.strip("%").astype('float')/100
    if quote_source == "GS":
        for col in ["pay_bid", "pay_offer", "rec_bid", "rec_offer"]:
            df[col] = df[col].str.strip('-')
        df['delta_pay'] *= -1
    for k in df:
        if df.dtypes[k] == 'object':
            try:
                df[k] = pd.to_numeric(df[k])
            except ValueError:
                pdb.set_trace()
    df['quote_source'] = quote_source
    df.set_index('strike', inplace=True)
    return df

def parse_quotedate(fh, date_received):
    for line in fh:
        line = line.rstrip()
        if line.startswith("At"):
            for p in ['%m/%d/%y %H:%M:%S', '%b  %d %Y %H:%M:%S', '%m/%d  %H:%M:%S']:
                try:
                    quotedate = pd.to_datetime(line, format=p, exact=False)
                except ValueError:
                    continue
                else:
                    if quotedate.year == 1900: # p='%m/%d  %H:%M:%S'
                        quotedate = quotedate.replace(year=date_received.year)
                    break
            else:
                raise RuntimeError("can't parse date")
            return quotedate

def parse_refline(line):
    regex = "Ref:(?P<ref>\S+)\s+(?:Fwd Px:(?P<fwdprice>\S+)\s+)?" \
                    "Fwd(?: Spd)?:(?P<fwdspread>\S+)\s+Fwd Bpv:(?P<fwdbpv>\S+)" \
                    "\s+Expiry:(?P<expiry>\S+)"
    m = re.match(regex, line)
    try:
        d = m.groupdict()
        d['expiry'] = pd.to_datetime(d['expiry'], format='%d-%b-%y')
    except AttributeError:
        logging.error("something wrong with " + fh.name)
    return d

def parse_baml(fh, indextype, series, quotedate, *args):
    option_stack = {}
    fwd_index = []
    line = ""
    while True:
        if line == "":
            try:
                line = next(fh)
            except StopIteration:
                break
        if line.startswith("Ref"):
            d = parse_refline(line)
            d.update({'quotedate': quotedate, 'index': indextype, 'series': series})
            df, line = parse_baml_block(fh, indextype)
            option_stack[d['expiry']] = df
            fwd_index.append(d)
        else:
            line = ""
    if option_stack:
        fwd_index = pd.DataFrame.from_records(fwd_index,
                                              index='quotedate')
        return option_stack, fwd_index
    else:
        raise RuntimeError("empty email: " + fh.name)

def parse_baml_block(fh, indextype):
    next(fh) ## skip header
    r = []
    line = ""
    for line in fh:
        line = line.strip()
        if line.startswith("Ref") or line == "":
            break
        line = re.sub("[/|]", " ", line)
        vals = re.sub(" +", " ", line).rstrip().split(" ")
        if len(vals) < 3: ## something went wrong
            line = ""
            break
        r.append(vals)
    return makedf(r, indextype, "BAML"), line

def parse_ms_block(fh, indextype):
    next(fh) ## skip header
    r = []
    for line in fh:
        line = line.rstrip()
        if line == "":
            break
        strike, payer, receiver, vol = line.split("|")
        strike = strike.strip()
        if indextype == "HY":
            strike = strike.split()[0]
        pay_bid, pay_offer, pay_delta = payer.strip().split()
        rec_bid, rec_offer, rec_delta = receiver.strip().split()

        vals = [strike, rec_bid, rec_offer, rec_delta,
                pay_bid, pay_offer, pay_delta]
        vol = vol.strip()
        if indextype == "HY":
            price_vol, vol = vol.replace("[","").replace("]","").split()
            vals += [vol, price_vol]
        else:
            vals += [vol]
        r.append(vals)
    return makedf(r, indextype, "MS")

def parse_nomura_block(fh, indextype):
    next(fh) ## skip header
    r = []
    for line in fh:
        line = line.rstrip()
        if line == "":
            break
        strike, receiver, payer, vol, _ = line.split("|", 4)
        strike = strike.strip()
        pay, pay_delta = payer.strip().split()
        rec, rec_delta = receiver.strip().split()
        pay_bid, pay_offer = pay.split("/")
        rec_bid, rec_offer = rec.split("/")
        vol = vol.strip()
        vals = [strike, rec_bid, rec_offer, rec_delta,
                pay_bid, pay_offer, pay_delta, vol]
        if indextype == "HY": # we don't have price vol
            vals.append(None)
        r.append(vals)
    return makedf(r, indextype, "NOM")

def parse_gs_block(fh, indextype):
    next(fh)
    r = []
    for line in fh:
        line = line.rstrip()
        if line == "":
            break
        vals = line.split()
        if indextype=='HY':
            vals.pop(2)
            vals.pop(9)
        else:
            vals.pop(1)
            vals.pop(8)
        strike = vals.pop(0)
        if indextype == "HY":
            vals.pop(0) #pop the spread
        pay, pay_delta = vals[:2]
        pay_bid, pay_offer = pay.split("/")
        rec_bid, rec_offer = vals[2].split("/")
        vol = vals[3]
        tail = vals[6]
        vals = [strike, rec_bid, rec_offer, None, pay_bid, pay_offer, pay_delta, vol]
        if indextype == "HY":
            vals.append(None)
        vals.append(tail)
        r.append(vals)
    return makedf(r, indextype, "GS")

def parse_ms(fh, indextype):
    option_stack = {}
    for line in fh:
        line = line.rstrip()
        if "EXPIRY" in line:
            expiry = line.split(" ")[1]
            expiry = pd.to_datetime(expiry, format="%d-%b-%Y")
            option_stack[expiry] = parse_ms_block(fh, indextype)
    return option_stack

def parse_nomura(fh, indextype):
    option_stack = {}
    for line in fh:
        line = line.rstrip()
        if "EXPIRY" in line:
            expiry = line.split(" ")[0]
            expiry = pd.to_datetime(expiry, format="%d-%b-%y")
            option_stack[expiry] = parse_nomura_block(fh, indextype)
    return option_stack

def parse_gs(fh, indextype, series, quotedate, ref):
    option_stack = {}
    fwd_index = []
    d = {'quotedate': quotedate, 'index': indextype,
         'series': series, 'ref': ref}
    for line in fh:
        line = line.rstrip()
        if line.startswith("Expiry"):
            m = re.match("Expiry (\d{2}\w{3}\d{2}) \((?:([\S]+) )?([\S]+)\)", line)
            if m:
                expiry, fwdprice, fwdspread = m.groups()
                expiry = pd.to_datetime(expiry, format='%d%b%y')
                d.update({'fwdspread': fwdspread, 'fwdprice': fwdprice,
                          'expiry': expiry})
                fwd_index.append(d)
            option_stack[expiry] = parse_gs_block(fh, indextype)
    fwd_index = pd.DataFrame.from_records(fwd_index,
                                          index='quotedate')
    return option_stack, fwd_index

subject_baml = re.compile("(?:Fwd:){0,2}(?:BAML )?(\w{2})([0-9]{1,2})\s")
subject_ms = re.compile("[^$]*\$\$ MS CDX OPTIONS: (IG|HY)(\d{2})[^-]*- REF[^\d]*([\d.]+)")
subject_nomura = re.compile("(?:Fwd:)?CDX (IG|HY)(\d{2}).*- REF:[^\d]*([\d.]+)")
subject_gs = re.compile("GS (IG|HY)(\d{2}) 5y.*- Ref [^\d]*([\d.]+)")

def parse_email(email):
    with open(email.path, "rt") as fh:
        date_received = datetime.datetime.fromtimestamp(int(fh.readline())/1000)
        subject = next(fh)
        for source in ['BAML', 'MS', 'NOMURA', 'GS']:
            m = globals()['subject_'+source.lower()].match(subject)
            if m:
                if source == 'BAML':
                    indextype, series = m.groups()
                else:
                    indextype, series, ref = m.groups()
                    ref = float(ref)

                series = int(series)
                quotedate = parse_quotedate(fh, date_received)
                parse_fun = globals()['parse_'+source.lower()]
                if source == 'BAML':
                    return (quotedate, indextype, series), \
                        parse_fun(fh, indextype, series, quotedate)
                elif source == "GS":
                     return (quotedate, indextype, series), \
                         parse_fun(fh, indextype, series, quotedate, ref)
                else:
                    option_stack = parse_fun(fh, indextype)
                    fwd_index = pd.DataFrame({'quotedate': quotedate,
                                              'ref': ref,
                                              'index': indextype,
                                              'series': series,
                                              'expiry': list(option_stack.keys())})
                    fwd_index.set_index('quotedate', inplace = True)
                    return (quotedate, indextype, series), (option_stack, fwd_index)
        raise RuntimeError("can't parse subject line: {0} for email {1}".format(
            subject, email.name))

def write_todb(swaption_stack, index_data):
    from sqlalchemy import MetaData, Table
    from db import dbengine, nan_to_null
    import psycopg2
    serenitasdb  = dbengine('serenitasdb')
    psycopg2.extensions.register_adapter(float, nan_to_null)
    meta = MetaData(bind=serenitasdb)
    swaption_quotes = Table('swaption_quotes', meta, autoload=True)
    for r in swaption_stack.to_dict(orient='records'):
        serenitasdb.execute(swaption_quotes.insert(), r)
    #ins = swaption_quotes.insert().values(swaption_stack.to_dict(orient='records')).execute()
    index_data.to_sql('swaption_ref_quotes', serenitasdb, if_exists='append', index=False)

def get_email_list(date):
    """returns a list of email file names for a given date

    Parameters
    ----------
    date : string
    """
    with open(".pickle", "rb") as fh:
        already_uploaded = pickle.load(fh)
    df = pd.DataFrame.from_dict(already_uploaded, orient='index')
    df.columns = ['quotedate']
    df = df.reset_index().set_index('quotedate')
    return df.loc[date,'index'].tolist()

def pickle_drop_date(date):
    with open(".pickle", "rb") as fh:
        already_uploaded = pickle.load(fh)
    newdict = {k: v for k, v in already_uploaded.items() if v.date() != date}
    with open(".pickle", "wb") as fh:
        pickle.dump(newdict, fh)

if __name__=="__main__":
    update_emails()
    data_dir = os.path.join(os.getenv("DATA_DIR"), "swaptions")
    emails = [f for f in os.scandir(data_dir) if f.is_file()]
    swaption_stack = {}
    index_data = pd.DataFrame()
    try:
        with open(".pickle", "rb") as fh:
            already_uploaded = pickle.load(fh)
    except FileNotFoundError:
        already_uploaded = {}
    for f in emails:
        if f.name in already_uploaded:
            continue
        else:
            try:
                key, (option_stack, fwd_index) = parse_email(f)
            except RuntimeError as e:
                logging.error(e)
            else:
                if key[0] is None:
                    logging.error("Something wrong with email: {}".format(f.name))
                    continue
                swaption_stack[key] = pd.concat(option_stack, names=['expiry', 'strike'])
                index_data = index_data.append(fwd_index)
                already_uploaded[f.name] = key[0]
    if index_data.empty:
        sys.exit()
    for col in ['fwdbpv', 'fwdprice', 'fwdspread', 'ref']:
        if col in index_data:
            index_data[col] = index_data[col].astype('float')
    index_data['index'] = index_data['index'].astype('category')

    swaption_stack = pd.concat(swaption_stack, names=['quotedate', 'index', 'series'])
    # import feather
    # feather.write_dataframe(swaption_stack, '../../data/swaptions.fth')
    # feather.write_dataframe(index_data, '../../data/index_data.fth')

    swaption_stack = swaption_stack.reset_index()
    swaption_stack = swaption_stack.drop_duplicates(['quotedate', 'index', 'series', 'expiry', 'strike'])
    index_data = index_data.reset_index()
    index_data = index_data.drop_duplicates(['quotedate', 'index', 'series', 'expiry'])
    write_todb(swaption_stack, index_data)
    with open(".pickle", "wb") as fh:
       pickle.dump(already_uploaded, fh)