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
|
import datetime
from typing import ClassVar
from dataclasses import dataclass
import re
import pandas as pd
from serenitas.utils.env import DAILY_DIR
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",
columns=(
"date",
"fund",
"account_name",
"account_number",
"currency_code",
"balance",
),
):
date: datetime.date
fund: Fund
custodian: ClassVar[Custodian]
def __init_subclass__(cls, custodian, **kwargs):
cls.custodian = custodian
cls._registry[custodian] = cls
def get_report(self):
report_dir = get_dir(self.date)
pattern = f"{self.custodian}_CASH_{self.fund}_"
reports = [
f
for f in report_dir.iterdir()
if f.name.startswith(pattern) and self.get_ts(f.name).date() == self.date
]
p = max(
reports,
key=lambda f: self.get_ts(f.name),
default=None,
)
if p:
return p
else:
raise MissingDataError(
f"Report not ready {self.date}: {self.custodian} {self.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"):
def __iter__(self):
df = pd.read_csv(self.get_report(), on_bad_lines="warn")
df = df[df["T-NARR-LONG"] == "CLOSING BALANCE"]
return (
(prev_business_day(self.date), self.fund, *t)
for t in df[
["Consolidation", "Account Number", "Currency code", "A-TRAN-AMT"]
].itertuples(index=False)
)
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}"
)
|
