blob: b477c35548d0f87fb4e1829fb80f0c77719a776e (
plain)
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
|
from serenitas.utils.remote import SftpClient
from trade_dataclasses import Deal, DealType, BbgDeal
from stat import S_ISREG
import re
import time
import logging
from paramiko.ssh_exception import SSHException
logger = logging.getLogger(__name__)
def get_bbg_id(s):
if m := re.match("(CDX|BOND)(?:BLOCK)?-[^_]*_([^$]*)", s):
return m.groups()
if "DEAL" in s:
return "FX", s.split("_")[3]
def run():
sftp = SftpClient.from_creds("bbg")
while True:
try:
for f in sftp.client.listdir_iter("/"):
if S_ISREG(f.st_mode):
try:
deal_type, bbg_id = get_bbg_id(f.filename)
except TypeError:
continue
if bbg_id not in BbgDeal._cache:
with sftp.client.open(f.filename) as fh:
try:
Deal[DealType(deal_type)].process(fh, bbg_id)
except ValueError as e:
logger.warning(e)
pass
else:
BbgDeal._cache[bbg_id] = None
except (SSHException, OSError):
sftp.client.close()
sftp = SftpClient.from_creds("bbg")
time.sleep(60)
if __name__ == "__main__":
run()
|
