diff options
Diffstat (limited to 'python/ops/process_queue.py')
| -rw-r--r-- | python/ops/process_queue.py | 19 |
1 files changed, 11 insertions, 8 deletions
diff --git a/python/ops/process_queue.py b/python/ops/process_queue.py index 17c81892..2b1f82c2 100644 --- a/python/ops/process_queue.py +++ b/python/ops/process_queue.py @@ -54,7 +54,7 @@ def process_indicative( process_fun = globals().get( f"{trade_type}_trade_process", lambda conn, session, trade: trade ) - mtm = Fund["MTM"]() + mtm = Fund["MTM"] for trade in get_trades(p, trade_type): process_fun(conn, session, trade) fund = trade["fund"] @@ -68,8 +68,10 @@ def process_indicative( "BESPOKE", ): mtm.stage(trade, trade_type=trade_type) - buf, dest = mtm.build_buffer(trade_type) - mtm.upload(buf, dest.name) + if mtm.staging_queue: + buf, dest = mtm.build_buffer(trade_type) + mtm.upload(buf, dest.name) + mtm.staging_queue.clear() p.delete(trade_type) @@ -86,6 +88,7 @@ def process_upload( buf, dest = fund.build_buffer(trade_type) if upload: fund.upload(buf, dest.name) + fund.staging_queue.clear() p.delete(key) @@ -97,16 +100,16 @@ def terminate_list( base_dir: pathlib.Path = DAILY_DIR, ): trade_type, fund, _ = key.split("_") - mtm = Fund["MTM"]() + mtm = Fund["MTM"] f = Fund[fund]() for term in p.lrange(key, 0, -1): obj = DealKind["termination"].from_dict(**loads(term)) mtm.staging_queue.append(obj.to_markit()) f.staging_queue.append(obj.to_globeop()) - # pretty crappy way, but there should be only one - mtm.product_type = obj.product_type - buf, dest = mtm.build_buffer("mtm") - mtm.upload(buf, dest.name) + if mtm.staging_queue: + buf, dest = mtm.build_buffer(trade_type) + mtm.upload(buf, dest.name) + mtm.staging_queue.clear() if upload and f.staging_queue: buf, dest = f.build_buffer((trade_type, "A")) f.upload(buf, dest.name) |