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
|
import datetime
from dataclasses import dataclass
from serenitas.utils.db import dbconn
from serenitas.utils.exchange import ExchangeMessage
from .misc import _sma_recipients, _cc_recipients
from exchangelib import FileAttachment
import pandas as pd
from io import StringIO
from typing import ClassVar
@dataclass
class SMA:
date: datetime.date
fund: ClassVar[str]
_conn: ClassVar = dbconn("dawndb")
_em: ClassVar = ExchangeMessage()
def __init_subclass__(cls, fund):
cls.fund = fund
def get_positions(self):
df_blotter = pd.read_sql_query(
"SELECT * FROM risk_positions(%s, NULL, %s)",
self._conn,
params=(self.date, self.fund),
index_col=["identifier"],
)
cds_positions = pd.read_sql_query(
"SELECT * FROM list_cds_marks_pre(%s, NULL, %s)",
self._conn,
params=(self.date, self.fund),
index_col=["security_id"],
)
tranche_positions = pd.read_sql_query(
"SELECT id, security_id, security_desc, maturity, a.notional, "
"protection, orig_attach, orig_detach, tranche_factor, clean_nav, "
"accrued, cp_code, cpty_id from list_cds(%s, %s) a "
"LEFT JOIN tranche_risk ON id=tranche_id AND date=%s "
"WHERE orig_attach IS NOT NULL",
self._conn,
params=(self.date, self.fund, self.date),
index_col=["id"],
)
return df_blotter, cds_positions, tranche_positions
def email_positions(self):
attachments = []
for name, df in zip(("bonds", "cds", "tranches"), (self.get_positions())):
buf = StringIO()
df.to_csv(buf)
attachments.append(
FileAttachment(
name=f"{self.date} {name}.csv", content=buf.getvalue().encode()
)
)
buf.close()
self._em.send_email(
f"{self.fund} {self.date} EOD positions ",
"",
to_recipients=_sma_recipients[self.fund],
cc_recipients=_cc_recipients[self.fund],
attach=attachments,
)
class IsoselSMA(SMA, fund="ISOSEL"):
pass
class BowdstSMA(SMA, fund="BOWDST"):
pass
|
