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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
import pandas as pd
from serenitas.utils.db import dbconn
from process_queue import rename_keys
import datetime
from collections import defaultdict
from io import StringIO
import csv
from serenitas.utils.env import DAILY_DIR
from serenitas.utils.remote import SftpClient
from psycopg2.errors import UniqueViolation
columns = [
"DealType",
"DealId",
"Action",
"Client",
"Fund",
"Portfolio/Business Unit",
"Strategy",
"Custodian",
"CashAccount",
"Counterparty",
"Comments",
"State",
"TradeDate",
"SettlementDate",
"Reserved",
"InstrumentType",
"ExpirationDate",
"CallNoticeIndicator",
"TransactionIndicator",
"StartMoney",
"Currency",
"Rate",
"Commission",
"DealFunction",
"FromAccount",
"ClientReference",
"Basis",
"MarginType",
"ClearingFacility" "CcpTradeRef",
"BlockId",
"BlockAmount",
"ExecutionDateTimeStamp",
"Collateralized",
"TradeDateFX",
]
_brokers = {
"BAML_ISDA": "BOANNY",
"CS": "CSITLN",
"GS": "GOLINY",
"BNP": "BNPBNY",
"MS": "MSCILN",
"JPM": "JPCBNY",
"GS_FCM": "GOLDNY",
}
insert_query = """INSERT INTO iam_tickets(trade_date, action, strategy, counterparty, maturity, start_money, currency, "offset") VALUES (%s, %s, %s, %s, %s, %s, %s, %s);"""
sql_query = "SELECT broker, amount, strategy FROM (SELECT *, rank() OVER(PARTITION BY broker,fund ORDER BY date desc) FROM strategy_im si WHERE fund = 'BOWDST' AND date<=%s ORDER BY date DESC) test WHERE RANK=1;"
select_query = "UPDATE iam_tickets set uploaded=True where maturity is null and trade_date =%s and action='NEW' and not uploaded returning *"
cancel_query = "UPDATE iam_tickets set maturity=%s where trade_date != %s and maturity is null and action='NEW' returning *"
conn = dbconn("dawndb")
def new_iam_process(obj, action):
rename_keys(
obj,
{
"dealid": "DealId",
"action": "Action",
"strategy": "Strategy",
"counterparty": "Counterparty",
"currency": "Currency",
"start_money": "StartMoney",
"trade_date": "TradeDate",
},
)
if action == "UPDATE":
obj["ExpirationDate"] = trade_date
obj["Action"] = "UPDATE"
elif action == "NEW":
obj["CallNoticeIndicator"] = "24H"
elif action == "CANCEL":
obj["Action"] = "CANCEL"
if obj["StartMoney"] > 0:
obj["TransactionIndicator"] = "DEPOSIT"
else:
obj["TransactionIndicator"] = "LOAN"
obj["StartMoney"] = abs(obj["StartMoney"])
if obj["Strategy"] == "CSH_CASH":
obj["Strategy"] = "M_CSH_CASH"
# static values
obj["DealType"] = "IamDeal"
obj["Client"] = "HEDGEMARK"
obj["Fund"] = "BOS_PAT_BOWDOIN"
obj["Custodian"] = "BNY"
obj["CashAccount"] = 751254
obj["State"] = "Valid"
obj["SettlementDate"] = obj["TradeDate"]
obj["Basis"] = "ACT/360"
obj["MarginType"] = "Variation"
return [obj.get(h, None) for h in columns]
with conn.cursor() as c:
trade_date = datetime.date(2022, 2, 25)
c.execute(sql_query, (trade_date,))
data = []
offsets = defaultdict(int)
for row in c:
data.append(
(
trade_date,
"NEW",
row.strategy,
_brokers[row.broker],
None,
row.amount,
"USD",
False,
)
)
offsets[_brokers[row.broker]] += row.amount
for broker, amount in offsets.items():
if broker == "GOLDNY":
continue
data.append((trade_date, "NEW", "CSH_CASH", broker, None, -amount, "USD", True))
csv_data = []
try:
c.executemany(insert_query, data)
except UniqueViolation:
conn.rollback()
c.execute(
"DELETE FROM iam_tickets where trade_date=%s returning *", (trade_date,)
)
for row in c:
csv_data.append(new_iam_process(row._asdict(), "CANCEL"))
c.executemany(insert_query, data)
c.execute(select_query, (trade_date,))
for row in c:
csv_data.append(new_iam_process(row._asdict(), "NEW"))
c.execute(cancel_query, (trade_date, trade_date))
for row in c:
csv_data.append(new_iam_process(row._asdict(), "UPDATE"))
buf = StringIO()
csvwriter = csv.writer(buf)
csvwriter.writerow(columns)
csvwriter.writerows(csv_data)
buf = buf.getvalue().encode()
dest = (
DAILY_DIR
/ str(datetime.date.today())
/ f"Bowdst.ALL.{datetime.datetime.now():%Y%m%d.%H%M%S}.IamDeal.csv"
)
dest.write_bytes(buf)
sftp = SftpClient.from_creds("hm_globeop")
sftp.client.chdir("incoming")
sftp.put(buf, dest.name)
conn.commit()
|
