diff options
| -rw-r--r-- | python/position.py | 37 |
1 files changed, 20 insertions, 17 deletions
diff --git a/python/position.py b/python/position.py index 10a063ba..ded45f5a 100644 --- a/python/position.py +++ b/python/position.py @@ -5,6 +5,7 @@ import pandas as pd import psycopg2 from pandas.tseries.offsets import BDay from pandas import bdate_range +from sqlalchemy import create_engine import re import os import logging @@ -57,7 +58,7 @@ def backpopulate_marks(begin_str='2015-01-15', end_str='2015-07-15'): positions = positions.drop_duplicates() positions.to_sql('position', engine, if_exists='append', index=False) -def update_securities(conn, session, workdate): +def update_securities(engine, session, workdate): field = {'Corp': 'PREV_CPN_DT', 'Mtge': 'START_ACC_DT'} securities = get_list(engine) securities = securities[securities.paid_down.isnull()] @@ -66,6 +67,7 @@ def update_securities(conn, session, workdate): data = pd.DataFrame.from_dict(data, orient='index') data = data[data.CPN_ASOF_DT.isnull() |(data.CPN_ASOF_DT<=workdate)] m = securities.merge(data, left_index=True, right_index=True) + conn = engine.raw_connection() with conn.cursor() as c: for r in m.to_dict('records'): if r[field[r['bbg_type']]] < workdate: @@ -73,6 +75,7 @@ def update_securities(conn, session, workdate): ",coupon=%(CUR_CPN)s WHERE identifier=%(identifier)s".format(field[r['bbg_type']]), r) conn.commit() + conn.close() def init_fx(session, engine, startdate): currencies = ['EURUSD', 'CADUSD'] @@ -153,14 +156,13 @@ def insert_swaption_vol(data, conn): c.execute(sql_str, (k, v.tolist(), v.tolist())) conn.commit() -def update_swaption_vol(session, engine, +def update_swaption_vol(conn, session, tenors=['A', 'C', 'F', 'I'] + list(range(1, 11)) + [15, 20, 25, 30]): tickers = [] for expiry in tenors: for tenor in tenors: tickers.append(f"USSN{expiry:0>2}{tenor} Curncy") - data = retrieve_data(session, tickers, ['PX_LAST', 'LAST_UPDATE_DT']) - conn = engine.raw_connection() + data = retrieve_data(session, tickers, ['PX_YEST_CLOSE', 'PX_CLOSE_DT']) for t in tenors[4:]: sql_str = f'INSERT INTO swaption_vol(date, "{t}y") ' + \ 'VALUES(%s, %s) ON CONFLICT (date)' + \ @@ -170,8 +172,8 @@ def update_swaption_vol(session, engine, for expiry in tenors: ticker = f"USSN{expiry:0>2}{t} Curncy" if data[ticker]: - r.append(data[ticker]['PX_LAST']) - dates.append(data[ticker]['LAST_UPDATE_DT']) + r.append(data[ticker]['PX_YEST_CLOSE']) + dates.append(data[ticker]['PX_CLOSE_DT']) else: r.append(None) dates.append(dates[-1]) @@ -194,12 +196,13 @@ def update_swap_rates(conn, session, c.execute(sql_str, data[ticker]) conn.commit() -def populate_cashflow_history(conn, session, workdate=None): +def populate_cashflow_history(engine, session, workdate=None): securities = get_list(engine, workdate) data = retrieve_data(session, securities.index.tolist(), ['HIST_CASH_FLOW', 'MTG_HIST_CPN', 'FLT_CPN_HIST', 'HIST_INTEREST_DISTRIBUTED']) fixed_coupons = {'XS0306416982 Mtge': 7.62, '91927RAD1 Mtge': 6.77} + conn = engine.raw_connection() for k, v in data.items(): if 'HIST_CASH_FLOW' in v: to_insert = v['HIST_CASH_FLOW'].merge(v['MTG_HIST_CPN'], how='left', @@ -235,26 +238,26 @@ def populate_cashflow_history(conn, session, workdate=None): with conn.cursor() as c: c.execute("REFRESH MATERIALIZED VIEW factors_history") conn.commit() + conn.close() if __name__=="__main__": serenitas_conn = psycopg2.connect(database="serenitasdb", user="serenitas_user", host="debian") - dawn_conn = psycopg2.connect(database="dawndb", - user="dawn_user", - host="debian") + dawn_engine = create_engine('postgresql://dawn_user@debian/dawndb') + dawn_conn = dawn_engine.raw_connection() if len(sys.argv) > 1: workdate = pd.Timestamp(sys.argv[1]) else: workdate = pd.datetime.today() with init_bbg_session(BBG_IP) as session: - # update_securities(dawn_conn, session, workdate) - # populate_cashflow_history(dawn_conn, session, workdate) - # update_fx(dawn_conn, session, ['EURUSD', 'CADUSD']) + update_securities(dawn_engine, session, workdate) + populate_cashflow_history(dawn_engine, session, workdate) + update_fx(dawn_conn, session, ['EURUSD', 'CADUSD']) update_swap_rates(serenitas_conn, session) - #update_swaption_vol(serenitas_conn, session) + update_swaption_vol(serenitas_conn, session) # with init_bbg_session(BBG_IP) as session: # init_fx(session, engine, pd.datetime(2013, 1, 1)) - with init_bbg_session(BBG_IP) as session: - data = init_swaption_vol(session) - insert_swaption_vol(data, serenitas_conn) + # with init_bbg_session(BBG_IP) as session: + # data = init_swaption_vol(session) + # insert_swaption_vol(data, serenitas_conn) |