diff options
Diffstat (limited to 'python')
| -rw-r--r-- | python/markit/__main__.py | 3 | ||||
| -rw-r--r-- | python/markit/import_quotes.py | 252 |
2 files changed, 136 insertions, 119 deletions
diff --git a/python/markit/__main__.py b/python/markit/__main__.py index 00a05d96..797bc6e6 100644 --- a/python/markit/__main__.py +++ b/python/markit/__main__.py @@ -13,7 +13,7 @@ from .loans import ( download_recupdates, ) from .rates import downloadMarkitIRData -from .import_quotes import insert_cds, insert_index, insert_tranche +from .import_quotes import copy_curves_forward, insert_cds, insert_index, insert_tranche from pandas.tseries.offsets import BDay from sqlalchemy import create_engine from utils import SerenitasFileHandler @@ -92,6 +92,7 @@ elif args.cds: download_composite_data(payload, workdate, historical) serenitasdb = dbconn("serenitasdb") insert_cds(serenitasdb, workdate) + copy_curves_forward(serenitasdb, workdate) serenitasdb.close() if not args.insert_only: engine = create_engine("postgresql://serenitas_user@debian/serenitasdb") diff --git a/python/markit/import_quotes.py b/python/markit/import_quotes.py index 2fd09bc3..62cf1531 100644 --- a/python/markit/import_quotes.py +++ b/python/markit/import_quotes.py @@ -123,6 +123,42 @@ def get_defaulted(mappings, default_table, workdate): return defaulted +def csv_file_gen(workdate): + CDS_DIR = BASE_DIR / "Tranche_data" / "CDS" / f"{workdate:%Y}" + csv_file = CDS_DIR / f"{workdate}_fixed.csv.lz4" + if csv_file.exists(): + fixed = True + else: + csv_file = CDS_DIR / f"{workdate}_parspread.csv.lz4" + if csv_file.exists(): + fixed = False + else: + raise FileNotFoundError + yield fixed + with lz4.frame.open(csv_file, "rt") as fh: + if not fixed: + next(fh) + next(fh) + csvreader = csv.DictReader(fh) + if fixed: + for l in csvreader: + yield (l, int(float(l["RunningCoupon"]) * 10000)) + else: + # we repeat each line with both values + yield from product(csvreader, (100, 500)) + + +def copy_curves_forward(conn, workdate: datetime.date): + sql_str = ( + "INSERT INTO cds_curves " + "SELECT %s, company_id, seniority, redcode, curve " + "FROM cds_curves WHERE date=%s" + ) + with conn.cursor() as c: + c.execute(sql_str, (workdate + BDay(1), workdate)) + conn.commit() + + def insert_cds(database, workdate: datetime.date): """insert Markit index quotes into the database @@ -144,7 +180,9 @@ def insert_cds(database, workdate: datetime.date): coupon_100 = np.full(8, 0.01) coupon_500 = np.full(8, 0.05) tenors = np.array([0.5, 1, 2, 3, 4, 5, 7, 10]) - yc_dict = {curr: get_curve(workdate, curr) for curr in ["USD", "JPY", "EUR"]} + yc_dict = {curr: get_curve(workdate, curr) for curr in ("USD", "EUR")} + if workdate >= datetime.date(2015, 1, 1): + yc_dict["JPY"] = get_curve(workdate, "JPY") seniority_mapping = { "SNRFOR": 0, "SUBLT2": 1, @@ -156,129 +194,107 @@ def insert_cds(database, workdate: datetime.date): default_table = { (cid, Seniority[seniority]): event_date for cid, seniority, event_date in c } - CDS_DIR = BASE_DIR / "Tranche_data" / "CDS" / f"{workdate:%Y}" - csv_file = CDS_DIR / f"{workdate}_fixed.csv.lz4" - if csv_file.exists(): - fixed = True - else: - csv_file = CDS_DIR / f"{workdate}_parspread.csv.lz4" - if csv_file.exists(): - fixed = False - else: - raise FileNotFoundError + g = csv_file_gen(workdate) + fixed = next(g) + with database.cursor() as c: + for line, spread in g: + k = CurveKey( + line["Ticker"], line["Tier"], line["Ccy"], line["DocClause"], spread, + ) + if mappings := markit_bbg_mapping.get(k, False): + if fixed: + upfront_rates = np.array([convert(line[c]) / 100 for c in col_upf]) + recovery_rates = np.full(8, convert(line["RealRecovery"]) / 100) + coupon_rates = coupon_100 if spread == 100 else coupon_500 + else: + upfront_rates = np.zeros(8) + recovery_rates = np.full(8, convert(line["Recovery"]) / 100) + coupon_rates = np.array( + [convert(line[c]) / 100 for c in col_spread] + ) + defaulted = get_defaulted(mappings, default_table, workdate) - with lz4.frame.open(csv_file, "rt") as fh: - if not fixed: - next(fh) - next(fh) - csvreader = csv.DictReader(fh) - if fixed: - g = ((l, int(float(l["RunningCoupon"]) * 10000)) for l in csvreader) - else: - # we repeat each line with both values - g = product(csvreader, (100, 500)) - with database.cursor() as c: - for line, spread in g: - k = CurveKey( - line["Ticker"], - line["Tier"], - line["Ccy"], - line["DocClause"], - spread, - ) - if mappings := markit_bbg_mapping.get(k, False): - if fixed: - upfront_rates = np.array( - [convert(line[c]) / 100 for c in col_upf] + try: + sc = SpreadCurve( + workdate, + yc_dict[k.currency], + None, + None, + None, + tenors, + coupon_rates, + upfront_rates, + recovery_rates, + ticker=k.ticker, + seniority=seniority_mapping[k.tier], + doc_clause=DocClause[k.short_code], + defaulted=defaulted, + ) + except ValueError: + logging.error(f"couldn't build curve for {k.ticker}") + else: + buf = sc.as_buffer(True) + for (cid, sen), curves in mappings: + c.execute( + "INSERT INTO cds_curves VALUES(%s, %s, %s, %s, %s) " + "ON CONFLICT (date, company_id, seniority) " + "DO UPDATE SET curve=excluded.curve, redcode=excluded.redcode", + (workdate, cid, sen.name, line["RedCode"], buf), ) - recovery_rates = np.full(8, convert(line["RealRecovery"]) / 100) - coupon_rates = coupon_100 if spread == 100 else coupon_500 - else: - upfront_rates = np.zeros(8) - recovery_rates = np.full(8, convert(line["Recovery"]) / 100) - coupon_rates = np.array( - [convert(line[c]) / 100 for c in col_spread] + c.executemany( + sqlstr, + [ + ( + workdate, + t, + upf * 100, + upf * 100, + spread, + spread, + "MKIT", + recovery_rates[0], + ) + for t, upf in zip(curves, upfront_rates) + ], ) - defaulted = get_defaulted(mappings, default_table, workdate) + tickers_found.add(k) + database.commit() - try: - sc = SpreadCurve( - workdate, - yc_dict[k.currency], - None, - None, - None, - tenors, - coupon_rates, - upfront_rates, - recovery_rates, - ticker=k.ticker, - seniority=seniority_mapping[k.tier], - doc_clause=DocClause[k.short_code], - defaulted=defaulted, + # handle missing tickers + tickers_missing = markit_bbg_mapping.keys() - tickers_found + jtiuk = CurveKey("JTIUK", "SNRFOR", "EUR", "CR14", 100) + if jtiuk in tickers_missing: + tickers_missing.remove(jtiuk) + + with database.cursor() as c: + for curve_key in tickers_missing: + logger.warning(f"{curve_key.full_ticker} missing for {workdate}") + for (cid, sen), e in markit_bbg_mapping[curve_key]: + c.execute( + "SELECT date, redcode, curve FROM cds_curves " + "WHERE company_id=%s AND seniority=%s AND date <= %s " + "ORDER BY date desc", + (cid, sen.name, workdate), + ) + try: + date, redcode, curve = c.fetchone() + except TypeError: + logger.error(f"{curve_key.full_ticker} never existed") + else: + if (workdate - date).days < 20: # we copy over the old curve + c.execute( + "INSERT INTO cds_curves VALUES(%s, %s, %s, %s, %s) " + "ON CONFLICT (date, company_id, seniority) " + "DO UPDATE SET curve=excluded.curve, redcode=excluded.redcode", + (workdate, cid, sen.name, redcode, curve), ) - except ValueError: - logging.error(f"couldn't build curve for {k.ticker}") + logger.info(f"Using {date} curve for {curve_key.ticker}") else: - buf = sc.as_buffer(True) - for (cid, sen), curves in mappings: - c.execute( - "INSERT INTO cds_curves VALUES(%s, %s, %s, %s, %s) " - "ON CONFLICT (date, company_id, seniority) " - "DO UPDATE SET curve=excluded.curve, redcode=excluded.redcode", - (workdate, cid, sen.name, line["RedCode"], buf), - ) - c.executemany( - sqlstr, - [ - ( - workdate, - t, - upf * 100, - upf * 100, - spread, - spread, - "MKIT", - recovery_rates[0], - ) - for t, upf in zip(curves, upfront_rates) - ], - ) - tickers_found.add(k) - database.commit() - tickers_missing = markit_bbg_mapping.keys() - tickers_found - jtiuk = CurveKey("JTIUK", "SNRFOR", "EUR", "CR14", 100) - if jtiuk in tickers_missing: - tickers_missing.remove(jtiuk) - with database.cursor() as c: - for curve_key in tickers_missing: - logger.warning(f"{curve_key.full_ticker} missing for {workdate}") - for (cid, sen), e in markit_bbg_mapping[curve_key]: - c.execute( - "SELECT date, redcode, curve FROM cds_curves " - "WHERE company_id=%s AND seniority=%s AND date <= %s " - "ORDER BY date desc", - (cid, sen.name, workdate), - ) - try: - date, redcode, curve = c.fetchone() - except TypeError: - logger.error(f"{curve_key.full_ticker} never existed") - else: - if (workdate - date).days < 20: # we copy over the old curve - c.execute( - "INSERT INTO cds_curves VALUES(%s, %s, %s, %s, %s) " - "ON CONFLICT (date, company_id, seniority) " - "DO UPDATE SET curve=excluded.curve, redcode=excluded.redcode", - (workdate, cid, sen.name, redcode, curve), - ) - logger.info(f"Using {date} curve for {curve_key.ticker}") - else: - logger.error( - "Could not find suitable curve for " - f"{curve_key.full_ticker} even looking back 20 days" - ) - database.commit() + logger.error( + "Could not find suitable curve for " + f"{curve_key.full_ticker} even looking back 20 days" + ) + database.commit() def get_date(f): |