diff options
Diffstat (limited to 'python/process_queue.py')
| -rw-r--r-- | python/process_queue.py | 33 |
1 files changed, 28 insertions, 5 deletions
diff --git a/python/process_queue.py b/python/process_queue.py index c03a8607..87722e02 100644 --- a/python/process_queue.py +++ b/python/process_queue.py @@ -1,7 +1,9 @@ import argparse +import blpapi import csv import datetime import logging +import psycopg2 import pathlib import re import redis @@ -315,7 +317,13 @@ def get_trades(p: redis.client.Pipeline, key: str): yield trades[-1] -def process_list(p: redis.client.Pipeline, key: str, upload: bool) -> None: +def process_list( + p: redis.client.Pipeline, + key: str, + upload: bool, + session: blpapi.session.Session, + conn: psycopg2.extensions.connection, +) -> None: trade_type, fund = key.split("_") trades = get_trades(p, key) if trade_type in ["bond", "cds", "swaption"]: @@ -971,10 +979,25 @@ if __name__ == "__main__": sys.exit("Please set path of daily directory in 'DAILY_DIR'") dawndb = dbconn("dawndb") - for trade_type in ["bond", "cds", "swaption", "future", "wire", "spot", "capfloor"]: - key = f"{trade_type}_{args.fund}" - p_list = partial(process_list, key=key, upload=not args.no_upload) - r.transaction(p_list, key) + with init_bbg_session(BBG_IP) as session: + for trade_type in [ + "bond", + "cds", + "swaption", + "future", + "wire", + "spot", + "capfloor", + ]: + key = f"{trade_type}_{args.fund}" + p_list = partial( + process_list, + key=key, + upload=not args.no_upload, + session=session, + conn=dawndb, + ) + r.transaction(p_list, key) for trade_type in ["cds", "swaption", "capfloor"]: key = f"{trade_type}_termination" t_list = partial( |