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
|
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]
_registry: ClassVar[dict] = {}
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"):
def __iter__(self):
df = pd.read_excel(self.get_report(), skiprows=3)
df = df.groupby(
["Portfolio #", "Currency", "Portfolio Name"], as_index=False
).sum(numeric_only=True)
return (
(prev_business_day(self.date), self.fund, *t)
for t in df[
["Portfolio Name", "Portfolio #", "Currency", "Current Balance"]
].itertuples(index=False)
)
class BNYCashReport(CashReport, custodian="BNY"):
def __iter__(self):
df = pd.read_csv(self.get_report(), thousands=",")
if df["Beginning Balance Local"].dtype == "object":
df["Beginning Balance Local"].apply(
lambda s: -float(s[1:-1]) if s.startswith("(") else float(s)
)
df = df.groupby(
["Account Number", "Account Name", "Local Currency Code"], as_index=False
).sum(numeric_only=True)
return (
(prev_business_day(self.date), self.fund, *t)
for t in df[
[
"Account Name",
"Account Number",
"Local Currency Code",
"Beginning Balance Local",
]
].itertuples(index=False)
)
@staticmethod
def get_ts(s):
m = re.search(r"\d{14}", s)
return datetime.datetime.strptime(m[0], "%Y%m%d%H%M%S")
class ScotiaCashReport(CashReport, custodian="SCOTIA"):
def __iter__(self):
p = self.get_report()
df = pd.read_excel(p, skipfooter=1)
if df.empty:
df = pd.read_excel(p)
rec = df.iloc[0]
yield (
prev_business_day(self.date),
self.fund,
rec["Account Name"],
rec["Account"],
"USD",
rec["Opening Bal."],
)
def get_report(self):
REPORT_DIR = DAILY_DIR / "Selene" / "Scotia_reports"
try:
return next(
REPORT_DIR.glob(
f"IsoSelene_{prev_business_day(self.date):%d-%b-%Y}_*_xlsx.JOAAPKO3.JOAAPKO1"
)
)
except StopIteration as e:
raise MissingDataError(f"Report not ready {self.date}: {self.custodian}")
|
