1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
|
from serenitas.utils.remote import SftpClient
import datetime
import pandas as pd
from serenitas.utils.db import dbconn
from psycopg2.errors import UniqueViolation
from zipfile import ZipFile
from paramiko.ssh_exception import AuthenticationException
import logging
def latest(f):
return f.removesuffix(".csv.zip").rsplit("_", 1)[-1]
def run(conn, date):
try:
sftp = SftpClient.from_creds("mtm")
except AuthenticationException as e:
logger.warning(e)
pass
files = [f for f in sftp.client.listdir("outbound") if date.strftime("%m%d%Y") in f]
if not files:
return
target_file = max(files, key=latest)
with sftp.client.open(f"outbound/{target_file}") as sftp_handle:
with ZipFile(sftp_handle).open(target_file.removesuffix(".zip")) as zip_handle:
df = pd.read_csv(zip_handle, skiprows=2, na_filter=False)
df = df[
(df["SwapType"].isin(["NEW", "ASGM", "TERM"]))
& (df["ThirdPartyStatus.1"].isin(["Confirmed"]))
]
place_holders = ",".join(["%s"] * 7)
sql_str = (
f"INSERT INTO mtm_submissions VALUES({place_holders}) "
"ON CONFLICT (ticketid, dealid) DO NOTHING"
)
with conn.cursor() as c:
for row in df.itertuples():
cpty_id = row.BrokerSecRef
try:
c.execute(
sql_str,
(
row.TICKETID,
row.Executed if row.Executed else None,
row.TicketNo,
row.ProductType,
row.BrokerId,
row.SwapType,
cpty_id if cpty_id else None,
),
)
except UniqueViolation:
conn.rollback()
else:
if (row.SwapType == "NEW") and cpty_id:
if row.BrokerId == "BNPBNY":
cpty_id = cpty_id.replace("BNPP", "19000")
if row.ProductType == "CDISW":
c.execute(
"UPDATE swaptions SET cpty_id = %s WHERE dealid = %s",
(
cpty_id,
row.TicketNo,
),
)
elif row.ProductType == "TRN":
c.execute(
"UPDATE cds SET cpty_id = %s WHERE dealid = %s",
(
cpty_id,
row.TicketNo,
),
)
elif row.ProductType == "CDI":
c.execute(
"UPDATE trs SET cpty_id = %s WHERE dealid = %s",
(
cpty_id,
row.TicketNo,
),
)
conn.commit()
if __name__ == "__main__":
logger = logging.getLogger(__name__)
conn = dbconn("dawndb")
run(conn, datetime.date.today())
|
