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
163
164
165
166
|
import datetime
from typing import ClassVar
from dataclasses import dataclass, field
import re
import pandas as pd
from serenitas.utils.env import DAILY_DIR
from serenitas.utils.db2 import dbconn
from serenitas.analytics.dates import prev_business_day
from serenitas.analytics.exceptions import MissingDataError
from serenitas.ops.trade_dataclasses import Ccy
from serenitas.ops.dataclass_mapping import Fund
from .misc import Custodian, get_dir
from .base import Report
@dataclass
class CashReport(Report, table_name="cash_balances"):
date: datetime.date
fund: Fund
account_name: str
account_number: str
currency_code: Ccy
balance: float
custodian: ClassVar[Custodian] = field(metadata={"insert": False})
_registry: ClassVar = field(default={}, metadata={"insert": False})
def __init_subclass__(cls, custodian):
cls.custodian = custodian
cls._registry[custodian] = cls
def __class_getitem__(cls, key):
return cls._registry[key]
@classmethod
def get_report(cls, knowledge_date, fund):
report_dir = get_dir(knowledge_date)
pattern = f"{cls.custodian}_CASH_{fund}_"
reports = [
f
for f in report_dir.iterdir()
if f.name.startswith(pattern)
and cls.get_ts(f.name).date() == knowledge_date
]
p = max(
reports,
key=lambda f: cls.get_ts(f.name),
default=None,
)
if p:
return p
else:
raise MissingDataError(
f"Report not ready {knowledge_date}: {cls.custodian} {fund}"
)
@staticmethod
def get_ts(s):
m = re.search(r"\d{12}", s)
return datetime.datetime.strptime(m[0], "%Y%m%d%H%M")
class NTCashReport(CashReport, custodian="NT"):
@classmethod
def yield_rows(cls, knowledge_date, fund):
df = pd.read_csv(cls.get_report(knowledge_date, fund), on_bad_lines="warn")
df = df[df["T-NARR-LONG"] == "CLOSING BALANCE"]
yield from df.to_dict(orient="records")
@classmethod
def from_report_line(cls, d):
return cls(
date=prev_business_day(d["knowledge_date"]),
fund=d["fund"],
account_name=d["Consolidation"],
account_number=d["Account Number"],
currency_code=d["Currency code"],
balance=d["A-TRAN-AMT"],
)
class UMBCashReport(CashReport, custodian="UMB"):
@classmethod
def yield_rows(cls, knowledge_date, fund):
df = pd.read_excel(cls.get_report(knowledge_date, fund), skiprows=3)
yield from df.groupby(["Portfolio #", "Currency", "Portfolio Name"]).sum(
numeric_only=True
)["Current Balance"].reset_index().to_dict(orient="records")
@classmethod
def from_report_line(cls, d):
return cls(
date=prev_business_day(d["knowledge_date"]),
fund=d["fund"],
account_name=d["Portfolio Name"],
account_number=d["Portfolio #"],
currency_code=d["Currency"],
balance=d["Current Balance"],
)
class BNYCashReport(CashReport, custodian="BNY"):
@staticmethod
def get_ts(s):
m = re.search(r"\d{14}", s)
return datetime.datetime.strptime(m[0], "%Y%m%d%H%M%S")
@classmethod
def yield_rows(cls, knowledge_date, fund):
df = pd.read_csv(cls.get_report(knowledge_date, fund))
df["Beginning Balance Local"] = df["Beginning Balance Local"].apply(
lambda s: "-" + s[1:-1] if s.startswith("(") else s
)
df["Beginning Balance Local"] = pd.to_numeric(
df["Beginning Balance Local"].str.replace(",", "")
)
yield from df.groupby(
["Account Number", "Account Name", "Local Currency Code"]
).sum(numeric_only=True).reset_index().to_dict(orient="records")
@classmethod
def from_report_line(cls, d):
return cls(
date=prev_business_day(d["knowledge_date"]),
fund=d["fund"],
account_name=d["Account Name"],
account_number=d["Account Number"],
currency_code=d["Local Currency Code"],
balance=d["Beginning Balance Local"],
)
class ScotiaCashReport(CashReport, custodian="SCOTIA"):
@classmethod
def yield_rows(cls, knowledge_date, fund):
p = cls.get_report(knowledge_date, fund)
df = pd.read_excel(p, skipfooter=1)
if df.empty: # No wires, so we can't skip the footer
df = pd.read_excel(p)
yield df.to_dict(orient="records")[0]
@classmethod
def from_report_line(cls, d):
return cls(
date=prev_business_day(d["knowledge_date"]),
fund=d["fund"],
account_name=d["Account Name"],
account_number=d["Account"],
currency_code="USD",
balance=d["Opening Bal."],
)
@classmethod
def get_report(cls, knowledge_date, fund):
REPORT_DIR = DAILY_DIR / "Selene" / "Scotia_reports"
try:
return next(
REPORT_DIR.glob(
f"IsoSelene_{prev_business_day(knowledge_date):%d-%b-%Y}_*_xlsx.JOAAPKO3.JOAAPKO1"
)
)
except StopIteration as e:
raise MissingDataError(
f"Report not ready {knowledge_date}: {cls.custodian} {fund}"
)
|
