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
|
###### Citco Submission DataClass
from dataclasses import dataclass, field, fields, Field
from serenitas.ops.trade_dataclasses import Deal
from typing import Literal
import datetime
import csv
def get_file_status(s):
is_processed, fname_short = s.rsplit("_", 1)
is_processed = is_processed.rsplit("-")[1] == "PROCESSED"
fname_short = fname_short.removesuffix(".csv")
return is_processed, fname_short
@dataclass
class CitcoSubmission(
Deal, deal_type=None, table_name="citco_submission", insert_ignore=("submit_date",)
):
fname: str = Field()
identifier_type: Literal["trade", "instrument"]
identifier: str
serenitas_id: str
submit_date: datetime.datetime
@classmethod
def from_citco_line(line, fname):
return cls()
@classmethod
def process(fh, fname):
for row in csv.DictReader(file_handle):
trade = cls.from_citco_line(line, fname)
trade.stage()
|
