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
|
from serenitas.utils.remote import SftpClient
from serenitas.ops.trade_dataclasses import (
SpotDeal,
FxSwapDeal,
BondDeal,
IRSDeal,
CDSDeal,
)
from stat import S_ISREG
import csv
import re
import time
import logging
from paramiko.ssh_exception import SSHException
from psycopg.errors import UniqueViolation
from lru import LRU
from typing import ClassVar
logger = logging.getLogger(__name__)
class Bbg:
sftp = SftpClient.from_creds("bbg")
_cache: ClassVar[LRU] = LRU(128)
_registry = {}
_insert_queue = []
def __init_subclass__(cls, ticket_type):
cls._registry[ticket_type] = cls
def __class_getitem__(cls, ticket_type):
return cls._registry[ticket_type]
@classmethod
def run(cls):
while True:
try:
for f in cls.sftp.client.listdir_iter("/"):
if S_ISREG(f.st_mode):
try:
ticket_type, bbg_id = cls.get_bbg_id(f.filename)
except TypeError:
continue
if bbg_id not in cls._cache:
with cls.sftp.client.open(f.filename) as fh:
dr = csv.DictReader(fh)
try:
trade = Bbg[ticket_type].process(dr, bbg_id)
except (ValueError, UnboundLocalError) as e:
logger.warning(f"problem with file {f.filename}")
logger.warning(e)
continue
else:
cls._cache[bbg_id] = None
Bbg[ticket_type].commit(trade)
except (SSHException, OSError):
cls.sftp.client.close()
cls.sftp = SftpClient.from_creds("bbg")
time.sleep(60)
@classmethod
def commit(cls, trade):
with trade._conn.cursor() as c:
try:
c.executemany(cls._sql_insert, cls._insert_queue)
except UniqueViolation as e:
logger.warning(e)
trade._conn.rollback()
else:
c.executemany(trade._sql_insert, trade._insert_queue)
trade._conn.commit()
finally:
cls._insert_queue.clear()
trade._insert_queue.clear()
@staticmethod
def get_bbg_id(s):
if m := re.match("(CDX|BOND|IRS)(?:BLOCK)?-[^_]*_([^$]*)", s):
return m.groups()
if "DEAL" in s:
return "FX", s.split("_")[3]
@classmethod
def process(cls, reader, bbg_id):
for row in reader:
for k, v in row.items():
if v == "":
row[k] = None
if row.get("Block Status") in (
"Rejected",
"Covered",
"Passed",
"Other",
"TrAway",
):
raise ValueError("Rejected trade")
line = {"bbg_ticket_id": bbg_id, **row}
orig_row = line.copy()
for trade in cls.trade_class(row).from_bbg_line(line):
trade.stage()
if "trade" in locals():
cls._insert_queue.append(list(orig_row.values()))
return trade
class BondTicket(Bbg, ticket_type="BOND"):
_sql_insert = f"INSERT INTO bond_tickets VALUES({','.join(['%s'] * 23)})"
@staticmethod
def trade_class(row):
return BondDeal
class IRSTicket(Bbg, ticket_type="IRS"):
_sql_insert = f"INSERT INTO irs_tickets VALUES({','.join(['%s'] * 23)})"
@staticmethod
def trade_class(row):
return IRSDeal
class CDSTicket(Bbg, ticket_type="CDX"):
_sql_insert = f"INSERT INTO cds_tickets VALUES({','.join(['%s'] * 22)})"
@staticmethod
def trade_class(row):
return CDSDeal
@classmethod
def process(cls, reader, bbg_id):
block_file = "Price (Dec)" not in reader.fieldnames
for row in reader:
line = {"bbg_ticket_id": bbg_id, **row}
for trade in CDSDeal.from_bbg_line(line):
trade.stage()
if block_file:
values = [bbg_id, *[None] * 21]
values[14], values[15] = trade.fund, trade.account_code
else:
values = list(line.values())
cls._insert_queue.append(values)
return trade
class FXTicket(Bbg, ticket_type="FX"):
_sql_insert = f"INSERT INTO fx_tickets VALUES({','.join(['%s'] * 211)})"
@staticmethod
def trade_class(row):
return SpotDeal if row["Deal Type"] in ("2", "4") else FxSwapDeal
if __name__ == "__main__":
Bbg.run()
|
