1 files changed, 59 insertions, 43 deletions

diff --git a/python/book_bbg2.py b/python/book_bbg2.py
index ee7ed33a..726224fb 100644
--- a/python/book_bbg2.py
+++ b/python/book_bbg2.py
@@ -5,6 +5,12 @@ import datetime
 import csv
 from trade_dataclasses import CDSDeal, BondDeal
 from decimal import Decimal
+from serenitas.utils.db import dbconn
+from lru import LRU
+from stat import S_ISREG
+import pandas as pd
+from sqlalchemy.exc import SQLAlchemyError
+import time
 
 _funds = {"SERENITAS_CGMF": "SERCGMAST", "BOWDOINST": "BOWDST"}
 _fcms = {"Bank of America, N.A.": "BAML", "Goldman Sachs": "GS"}
@@ -40,22 +46,6 @@ _bond_cp = {
 }
 
 
-def download_files(date: datetime.date):
-    sftp = SftpClient.from_creds("bbg")
-    dst = DAILY_DIR / str(date) / "bbg_tickets"
-    if not dst.exists():
-        dst.mkdir()
-    EST = ZoneInfo("US/Eastern")
-
-    def by_date(f, date):
-        local_dt = datetime.datetime.fromtimestamp(
-            f.st_mtime, tz=datetime.timezone.utc
-        ).astimezone(EST)
-        return local_dt.date() == date
-
-    sftp.download_files("/", dst, (lambda f: by_date(f, date),))
-
-
 def get_indic_data(conn, redcode, tenor):
     sql_str = (
         "SELECT maturity, coupon "
@@ -96,38 +86,64 @@ def cdx_booking_process(path, conn):
         CDSDeal.commit()
 
 
-def bond_booking_process(path, conn):
-    with open(path) as fh:
-        reader = csv.DictReader(fh)
-        for line in reader:
-            trade = BondDeal(
-                faceamount=Decimal(line["Quantity"]),
-                price=Decimal(line["Price (Dec)"]),
-                cp_code=_bond_cp[line["Brkr"]],
-                cusip=line["Cusip"],
-                identifier=line["Cusip"],
-                trade_date=datetime.datetime.strptime(line["Trade Dt"], "%m/%d/%Y"),
-                settle_date=datetime.datetime.strptime(line["SetDt"], "%m/%d/%Y"),
-                portfolio="UNALLOCATED",
-                asset_class=None,
-                description=line["Security"].removesuffix(" Mtge"),
-                buysell=line["Side"] == "B",
-            )
-            trade.stage()
+def bond_booking_process(path, conn, fname, _cache):
+    df = pd.read_csv(path)
+    for _, line in df.iterrows():
+        trade = BondDeal(
+            faceamount=Decimal(line["Quantity"]),
+            price=Decimal(line["Price (Dec)"]),
+            cp_code=_bond_cp[line["Brkr"]],
+            cusip=line["Cusip"],
+            identifier=line["Cusip"],
+            trade_date=datetime.datetime.strptime(line["Trade Dt"], "%m/%d/%Y"),
+            settle_date=datetime.datetime.strptime(line["SetDt"], "%m/%d/%Y"),
+            portfolio="UNALLOCATED",
+            asset_class=None,
+            description=line["Security"].removesuffix(" Mtge"),
+            buysell=line["Side"] == "B",
+        )
+        trade.stage()
+    df["bbg_ticket_id"] = [fname]
+    try:
+        df.to_sql("bond_tickets", dawn_engine, if_exists="append", index=False)
+    except SQLAlchemyError as e:
+        error = str(e.__dict__["orig"])
+        BondDeal._insert_queue.clear()
+        print(error)
+    else:
+        _cache[fname] = None
+    finally:
         BondDeal.commit()
 
 
-def book_trades(conn, date=datetime.date.today()):
-    download_files(date)
-    for p in (DAILY_DIR / str(date) / "bbg_tickets").glob("CDX*"):
-        cdx_booking_process(p, conn)
-    for p in (DAILY_DIR / str(date) / "bbg_tickets").glob("BOND*"):
-        bond_booking_process(p, conn)
+def get_bbg_id(name):
+    return name.split("_", 1)[1]
 
 
 if __name__ == "__main__":
-    from serenitas.utils.db import serenitas_pool
+    from serenitas.utils.db import serenitas_pool, dawn_engine
 
-    d = datetime.date(2022, 2, 7)
     conn = serenitas_pool.getconn()
-    book_trades(conn, d)
+    _cache = LRU(128)
+    while True:
+        d = datetime.date.today()
+        sftp = SftpClient.from_creds("bbg")
+        filters = [lambda f: S_ISREG(f.st_mode)]
+        for f in filter(
+            lambda f: all(filt(f) for filt in filters), sftp.client.listdir_iter("/")
+        ):
+            if ("CDX" in f.filename) or ("BOND" in f.filename):
+                if get_bbg_id(f.filename) in _cache.keys():
+                    continue
+                else:
+                    if "CDX" in f.filename:
+                        # cds_booking_process(sftp.client.open(f"/{f.filename}"), conn, get_bbg_id(f.filename), _cache)
+                        pass
+                    elif "BOND" in f.filename:
+                        bond_booking_process(
+                            sftp.client.open(f"/{f.filename}"),
+                            conn,
+                            get_bbg_id(f.filename),
+                            _cache,
+                        )
+        time.sleep(60)