path: root/python/parse_emails.py

import pandas as pd
import re
import os
from db import dbconn
import psycopg2.sql as sql
from download_emails import save_emails, errors
import datetime
import logging
import pickle
import sys
from quantlib.time.imm import next_date
from quantlib.time.api import Date, pydate_from_qldate

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


def list_imm_dates(date):
    d = Date.from_datetime(date)
    r = []
    for i in range(10):
        d = next_date(d, False)
        r.append(pydate_from_qldate(d))
    return r


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':
            df[k] = df[k].str.replace(",", "")
            try:
                df[k] = pd.to_numeric(df[k])
            except ValueError:
                breakpoint()
    df.set_index('strike', inplace=True)
    return df


def parse_quotedate(fh, date_received):
    for line in fh:
        line = line.rstrip()
        if "At:" in line:
            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)
                    quotedate = quotedate.tz_localize("America/New_York")
                    break
            else:
                raise RuntimeError("can't parse date from {line}")
            return quotedate
    else:
        raise RuntimeError("no date received in the email")


def parse_refline(line):
    regex = r"Ref:(?P<ref>\S+)\s+(?:Fwd Px:(?P<fwdprice>\S+)\s+)?" \
        r"Fwd(?: Spd)?:(?P<fwdspread>\S+)\s+Fwd Bpv:(?P<fwdbpv>\S+)" \
        r"\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:
        raise RuntimeError(f"can't parse refline {line}")
    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')
        fwd_index['quote_source'] = 'BAML'
        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):
    line = next(fh)  # skip header
    if line.strip() == "":  # empty block
        return None
    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]
        try:
            pay_bid, pay_offer, pay_delta = payer.strip().split()
            rec_bid, rec_offer, rec_delta = receiver.strip().split()
        except ValueError:
            try:
                pay_mid, pay_delta = payer.strip().split()
                rec_mid, rec_delta = receiver.strip().split()
                pay_bid, pay_offer = pay_mid, pay_mid
                rec_bid, rec_offer = rec_mid, rec_mid
            except ValueError:
                raise RuntimeError("Couldn't parse line: {line}")

        vals = [strike, rec_bid, rec_offer, rec_delta,
                pay_bid, pay_offer, pay_delta]
        vol = vol.strip()
        if indextype == "HY":
            try:
                price_vol, vol = vol.replace("[", "").replace("]", "").split()
            except ValueError:
                price_vol, vol, vol_change, be = (vol.replace("[", "").
                                                  replace("]", "").split())
            vals += [vol, price_vol]
        else:
            if " " in vol:
                vol, vol_change, be = vol.split()
            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 "EXPIRY" in line or 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)
    else:
        return None, makedf(r, indextype, "NOM")
    return line, makedf(r, indextype, "NOM")


def parse_sg_block(fh, indextype, expiration_dates):
    r = []
    for line in fh:
        line = line.rstrip()
        if line == "":
            break
        if indextype == "IG":
            option_type, strike, price, delta, vol, expiry = line.split()
        else:
            option_type, strike, strike_spread, price, delta, vol, expiry = line.split()

        expiry_month = datetime.datetime.strptime(expiry, "%b-%y").month
        expiry = next(pd.Timestamp(d) for d in expiration_dates if d.month == expiry_month)
        if option_type == "Rec":
            rec_bid, rec_offer = price.split("/")
            pay_bid, pay_offer = None, None
            rec_delta, pay_delta = delta, None
        else:
            pay_bid, pay_offer = price.split("/")
            rec_bid, rec_offer = None, None
            rec_delta, pay_delta = None, delta
        vals = [strike, rec_bid, rec_offer, rec_delta, pay_bid,
                pay_offer, pay_delta, vol]
        if indextype == "HY":
            vals.append(None)
        r.append(vals)
    return expiry, makedf(r, indextype, "SG")


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_citi_block(fh, indextype):
    next(fh) #skip header
    r = []
    for line in fh:
        line = line.rstrip()
        if line == "":
            break
        if indextype == "HY":
            strike, payers, receivers, vol, price_vol = line.split("|")
        else:
            strike, payers, receivers, vol = line.split("|")
        strike = strike.strip()
        pay_bid, pay_offer = payers.split("/")
        pay_bid = pay_bid.strip()
        pay_offer = pay_offer.strip()
        pay_offer, pay_delta = pay_offer.split()
        rec_bid, rec_offer = receivers.split("/")
        rec_bid = rec_bid.strip()
        rec_offer = rec_offer.strip()
        rec_offer, rec_delta = rec_offer.split()
        vol = vol.strip()
        vol = vol.split()[0]
        if indextype == "HY":
            price_vol = price_vol.strip()
            r.append([strike, rec_bid, rec_offer, rec_delta,
                      pay_bid, pay_offer, pay_delta, vol, price_vol])
        else:
            r.append([strike, rec_bid, rec_offer, rec_delta,
                      pay_bid, pay_offer, pay_delta, vol])
    return makedf(r, indextype, "CITI")

def parse_ms(fh, indextype, *args):
    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")
            block = parse_ms_block(fh, indextype)
            if block is None or block.empty:
                logging.warning("MS: block is empty for {expiry} expiry")
            else:
                option_stack[expiry] = block
    return option_stack


def parse_nom(fh, indextype, *args):
    option_stack = {}

    def aux(line, fh, indextype, option_stack):
        expiry = line.split(" ")[0]
        expiry = pd.to_datetime(expiry, format="%d-%b-%y")
        next_line, df = parse_nomura_block(fh, indextype)
        option_stack[expiry] = df
        if next_line:
            if "EXPIRY" in next_line:
                aux(next_line, fh, indextype, option_stack)
            else:
                raise RuntimeError(f"Don't know what to do with {line}.")
    for line in fh:
        line = line.rstrip()
        if "EXPIRY" in line:
            aux(line, fh, indextype, option_stack)
    return option_stack


def parse_sg(fh, indextype, expiration_dates):
    option_stack = {}
    for line in fh:
        line = line.rstrip()
        if line.startswith("Type"):
            expiry, df = parse_sg_block(fh, indextype, expiration_dates)
            option_stack[expiry] = df
    return option_stack


def parse_gs(fh, indextype, series, quotedate, ref):
    option_stack = {}
    fwd_index = []
    d = {'quotedate': quotedate, 'index': indextype,
         'series': series, 'ref': ref}
    pat = re.compile(r"Expiry (\d{2}\w{3}\d{2}) \((?:([\S]+) )?([\S]+)\)")
    for line in fh:
        line = line.rstrip()
        if line.startswith("Expiry"):
            m = pat.match(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.copy())
                option_stack[expiry] = parse_gs_block(fh, indextype)
            else:
                logging.error("Can't parse expiry line:", line)
    fwd_index = pd.DataFrame.from_records(fwd_index,
                                          index='quotedate')
    fwd_index['quote_source'] = 'GS'
    return option_stack, fwd_index

def parse_citi(fh, indextype, series, quotedate):
    option_stack = {}
    fwd_index = []
    d = {'quotedate': quotedate,
         'index': indextype,
         'series': series}
    pat = re.compile(r"Exp: (\d{2}-\w{3}-\d{2})[^R]*Ref:[^\d]*([\d.]+)")
    for line in fh:
        line = line.strip()
        if line.startswith("Exp"):
            m = pat.match(line)
            if m:
                expiry, ref = m.groups()
                expiry = pd.to_datetime(expiry, format='%d-%b-%y')
                d.update({'ref': ref,
                          'expiry': expiry})
                fwd_index.append(d.copy())
                option_stack[expiry] = parse_citi_block(fh, indextype)
            else:
                logging.error("Cant't parse expiry line:", line)
    fwd_index = pd.DataFrame.from_records(fwd_index,
                                          index='quotedate')
    fwd_index['quote_source'] = 'CITI'
    return option_stack, fwd_index

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

def parse_email(email, date_received):
    with open(email.path, "rt") as fh:
        subject = fh.readline().lstrip()

        for source in ['BAML', 'GS', 'MS', 'NOM', 'SG', 'CITI']:
            m = globals()[f'subject_{source.lower()}'].match(subject)
            if m:
                if source in ['BAML', 'CITI']:
                    indextype, series = m.groups()
                else:
                    indextype, series, ref = m.groups()
                    ref = float(ref)
                series = int(series)
                cur_pos = fh.tell()
                try:
                    quotedate = parse_quotedate(fh, date_received)
                except RuntimeError:
                    logging.warning("couldn't find received date in message: "
                                    f"{email.name}, using {date_received}")
                    quotedate = pd.Timestamp(date_received).tz_localize("America/New_York")
                    fh.seek(cur_pos)

                expiration_dates = list_imm_dates(quotedate)
                parse_fun = globals()[f'parse_{source.lower()}']
                if source in ['BAML', 'CITI']:
                    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, expiration_dates)
                    fwd_index = pd.DataFrame({'quotedate': quotedate,
                                              'ref': ref,
                                              'index': indextype,
                                              'series': series,
                                              'expiry': list(option_stack.keys()),
                                              'quote_source': source})
                    fwd_index.set_index('quotedate', inplace=True)
                    return (quotedate, indextype, series), (option_stack, fwd_index)
        else:
            raise RuntimeError(f"can't parse subject line: {subject} for email {email.name}")

def write_todb(swaption_stack, index_data):
    def gen_sql_str(query, table_name, columns):
        return query.format(sql.Identifier(table_name),
                            sql.SQL(", ").join(sql.Identifier(c) for c in columns),
                            sql.SQL(", ").join(sql.Placeholder() * len(columns)))
    conn = dbconn('serenitasdb')
    query = sql.SQL("INSERT INTO {}({}) VALUES({}) "
                    "ON CONFLICT DO NOTHING RETURNING ref_id")
    sql_str = gen_sql_str(query, "swaption_ref_quotes", index_data.columns)
    query = sql.SQL("INSERT INTO {}({}) VALUES({}) "
                    "ON CONFLICT DO NOTHING")
    with conn.cursor() as c:
        for t in index_data.itertuples(index=False):
            c.execute(sql_str, t)
            try:
                ref_id, = next(c)
            except StopIteration:
                continue
            else:
                try:
                    df = swaption_stack.loc[(t.quotedate, t.index, t.series, t.expiry)]
                except KeyError as e:
                    logging.warning("missing key in swaption_stack: "
                                    f"{t.quotedate}, {t.index}, {t.series}, {t.expiry}")
                    continue
                df['ref_id'] = ref_id
                c.executemany(gen_sql_str(query, "swaption_quotes", df.columns),
                              df.itertuples(index=False))
            conn.commit()


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__":
    try:
        save_emails()
    except (errors.HttpError, FileNotFoundError) as e:
        logging.error(e)
        save_emails(update=False)
    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:
        date_composed, msg_id = f.name.split("_")
        date_composed = datetime.datetime.strptime(date_composed,
                                                   "%Y-%m-%d %H-%M-%S")
        if msg_id in already_uploaded:
            continue
        else:
            try:
                key, (option_stack, fwd_index) = parse_email(f, date_composed)
            except RuntimeError as e:
                logging.error(e)
            else:
                if key[0] is None or len(option_stack) == 0:
                    logging.error(f"Something wrong with email: {f.name}")
                    continue
                swaption_stack[key] = pd.concat(option_stack,
                                                names=['expiry', 'strike'])
                index_data = index_data.append(fwd_index)
                already_uploaded[msg_id] = 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'])
    swaption_stack = swaption_stack.reset_index()
    swaption_stack = swaption_stack.drop_duplicates(['quotedate', 'index', 'series',
                                                     'expiry', 'strike'])
    swaption_stack = swaption_stack.set_index(['quotedate', 'index', 'series', 'expiry'])
    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)