Skip to content

Commit

Permalink
Refactor parsing and formatting (#122)
Browse files Browse the repository at this point in the history 
* Update .po files to replace event.py's "DIVIDENDS" type with Portfolio Performance's "DIVIDEND" type

* Buggy commit to refactor the event parsing and csf file formatting to include taxes, fees, saveback correctly
1. Removes card_failed_transaction type due to it not being relevant for PP
2. Adds INCOMING_TRANSFER_DELEGATION and from issue #116 (card_successful_oct, TAX_REFUND, benefits_saveback_execution) to event.py
3. Introduces enums and dataclasses in event.py to a. structure and facilitate type matching b. reduce Event object's memory size c. differentiate between pp_types and types that require further csv formating.
4. Introduces hacky tax and fee parsers in event.py
5. Introduces event_formatter.py to generate multiple csv lines from certain events (saveback, taxed income, transactions with fees)
Requires extensive testing (functionality + python version)

* Fixes various parsing and formatting errors.
What remains: 1. Check if still compatible with python 3.8
2. Add Fees field in csv file to avoid having to generate a new Fee-type line for each fee occurence
3. Improve parsers with more logs

* Clarify certain comments; minor fixes; simplify tax parsing to only include tax value and no further info (KapESt, Soli, Kirche)

* Further simplify tax parsing

* Restores language translation
Adds a cli argument to dl_docs and export_transactions to chronologically sort the exported csv transactions
Removes unused imports
Various small modifications

* Adds taxes and fees columns to the csv file instead of generating stand-alone taxes and fees events

* 1. Changes fees and taxes sign to negative 2. Maps card_failed_transaction events with executed status to REMOVALs 3. Renames UnprocessedEventType to ConditionalEventType for more clarity
  • Loading branch information
@pinzutu
pinzutu authored Oct 3, 2024
1 parent 196b42b commit b3e70bf
Show file tree
Hide file tree
Showing 20 changed files with 382 additions and 145 deletions.
2 changes: 2 additions & 0 deletions pytr/dl.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ def __init__(
history_file='pytr_history',
max_workers=8,
universal_filepath=False,
sort_export=False
):
'''
tr: api object
Expand All @@ -33,6 +34,7 @@ def __init__(
self.filename_fmt = filename_fmt
self.since_timestamp = since_timestamp
self.universal_filepath = universal_filepath
self.sort_export = sort_export

self.session = FuturesSession(max_workers=max_workers, session=self.tr._websession)
self.futures = []
Expand Down
343 changes: 248 additions & 95 deletions pytr/event.py
View file
Original file line number Diff line number Diff line change
@@ -1,100 +1,253 @@
from dataclasses import dataclass
from datetime import datetime
from enum import auto, Enum
import re

tr_eventType_to_pp_type = {
'INCOMING_TRANSFER': 'DEPOSIT',
'PAYMENT_INBOUND': 'DEPOSIT',
'PAYMENT_INBOUND_GOOGLE_PAY': 'DEPOSIT',
'PAYMENT_INBOUND_SEPA_DIRECT_DEBIT': 'DEPOSIT',
'card_refund': 'DEPOSIT',

'CREDIT': 'DIVIDENDS',
'ssp_corporate_action_invoice_cash': 'DIVIDENDS',

'INTEREST_PAYOUT_CREATED': 'INTEREST',

'OUTGOING_TRANSFER_DELEGATION': 'REMOVAL',
'PAYMENT_OUTBOUND': 'REMOVAL',
'card_failed_transaction': 'REMOVAL',
'card_order_billed': 'REMOVAL',
'card_successful_atm_withdrawal': 'REMOVAL',
'card_successful_transaction': 'REMOVAL',

'ORDER_EXECUTED': 'TRADE_INVOICE',
'SAVINGS_PLAN_EXECUTED': 'TRADE_INVOICE',
'SAVINGS_PLAN_INVOICE_CREATED': 'TRADE_INVOICE',
'TRADE_INVOICE': 'TRADE_INVOICE'
from typing import Any, Dict, Optional, Tuple, Union


class ConditionalEventType(Enum):
"""Events that conditionally map to None or one/multiple PPEventType events"""

FAILED_CARD_TRANSACTION = auto()
SAVEBACK = auto()
TRADE_INVOICE = auto()


class PPEventType(Enum):
"""PP Event Types"""

BUY = "BUY"
DEPOSIT = "DEPOSIT"
DIVIDEND = "DIVIDEND"
FEES = "FEES" # Currently not mapped to
FEES_REFUND = "FEES_REFUND" # Currently not mapped to
INTEREST = "INTEREST"
INTEREST_CHARGE = "INTEREST_CHARGE" # Currently not mapped to
REMOVAL = "REMOVAL"
SELL = "SELL"
TAXES = "TAXES" # Currently not mapped to
TAX_REFUND = "TAX_REFUND"
TRANSFER_IN = "TRANSFER_IN" # Currently not mapped to
TRANSFER_OUT = "TRANSFER_OUT" # Currently not mapped to


class EventType(Enum):
PP_EVENT_TYPE = PPEventType
CONDITIONAL_EVENT_TYPE = ConditionalEventType


tr_event_type_mapping = {
# Deposits
"INCOMING_TRANSFER": PPEventType.DEPOSIT,
"INCOMING_TRANSFER_DELEGATION": PPEventType.DEPOSIT,
"PAYMENT_INBOUND": PPEventType.DEPOSIT,
"PAYMENT_INBOUND_GOOGLE_PAY": PPEventType.DEPOSIT,
"PAYMENT_INBOUND_SEPA_DIRECT_DEBIT": PPEventType.DEPOSIT,
"card_refund": PPEventType.DEPOSIT,
"card_successful_oct": PPEventType.DEPOSIT,
# Dividends
"CREDIT": PPEventType.DIVIDEND,
"ssp_corporate_action_invoice_cash": PPEventType.DIVIDEND,
# Failed card transactions
"card_failed_transaction": ConditionalEventType.FAILED_CARD_TRANSACTION,
# Interests
"INTEREST_PAYOUT": PPEventType.INTEREST,
"INTEREST_PAYOUT_CREATED": PPEventType.INTEREST,
# Removals
"OUTGOING_TRANSFER_DELEGATION": PPEventType.REMOVAL,
"PAYMENT_OUTBOUND": PPEventType.REMOVAL,
"card_order_billed": PPEventType.REMOVAL,
"card_successful_atm_withdrawal": PPEventType.REMOVAL,
"card_successful_transaction": PPEventType.REMOVAL,
# Saveback
"benefits_saveback_execution": ConditionalEventType.SAVEBACK,
# Tax refunds
"TAX_REFUND": PPEventType.TAX_REFUND,
# Trade invoices
"ORDER_EXECUTED": ConditionalEventType.TRADE_INVOICE,
"SAVINGS_PLAN_EXECUTED": ConditionalEventType.TRADE_INVOICE,
"SAVINGS_PLAN_INVOICE_CREATED": ConditionalEventType.TRADE_INVOICE,
"TRADE_INVOICE": ConditionalEventType.TRADE_INVOICE,
}


@dataclass
class Event:
def __init__(self, event_json):
self.event = event_json
self.shares = ""
self.isin = ""

self.pp_type = tr_eventType_to_pp_type.get(self.event["eventType"], "")
self.body = self.event.get("body", "")
self.process_event()

@property
def date(self):
dateTime = datetime.fromisoformat(self.event["timestamp"][:19])
return dateTime.strftime("%Y-%m-%d")

@property
def is_pp_relevant(self):
if self.event["eventType"] == "card_failed_transaction":
if self.event["status"] == "CANCELED":
return False
return self.pp_type != ""

@property
def amount(self):
return str(self.event["amount"]["value"])

@property
def note(self):
if self.event["eventType"].find("card_") == 0:
return self.event["eventType"]
else:
return ""

@property
def title(self):
return self.event["title"]

def determine_pp_type(self):
if self.pp_type == "TRADE_INVOICE":
if self.event["amount"]["value"] < 0:
self.pp_type = "BUY"
else:
self.pp_type = "SELL"

def determine_shares(self):
if self.pp_type == "TRADE_INVOICE":
sections = self.event.get("details", {}).get("sections", [{}])
for section in sections:
if section.get("title") == "Transaktion":
amount = section.get("data", [{}])[0]["detail"]["text"]
amount = re.sub("[^\,\d-]", "", amount)
self.shares = amount.replace(",", ".")

def determine_isin(self):
if self.pp_type in ("DIVIDENDS", "TRADE_INVOICE"):
sections = self.event.get("details", {}).get("sections", [{}])
self.isin = self.event.get("icon", "")
self.isin = self.isin[self.isin.find("/") + 1 :]
self.isin = self.isin[: self.isin.find("/")]
isin2 = self.isin
for section in sections:
action = section.get("action", None)
if action and action.get("type", {}) == "instrumentDetail":
isin2 = section.get("action", {}).get("payload")
break
if self.isin != isin2:
self.isin = isin2

def process_event(self):
self.determine_shares()
self.determine_isin()
self.determine_pp_type()
date: datetime
title: str
event_type: Optional[EventType]
fees: Optional[float]
isin: Optional[str]
note: Optional[str]
shares: Optional[float]
taxes: Optional[float]
value: Optional[float]

@classmethod
def from_dict(cls, event_dict: Dict[Any, Any]):
"""Deserializes the event dictionary into an Event object
Args:
event_dict (json): _description_
Returns:
Event: Event object
"""
date: datetime = datetime.fromisoformat(event_dict["timestamp"][:19])
event_type: Optional[EventType] = cls._parse_type(event_dict)
title: str = event_dict["title"]
value: Optional[float] = (
v
if (v := event_dict.get("amount", {}).get("value", None)) is not None
and v != 0.0
else None
)
fees, isin, note, shares, taxes = cls._parse_type_dependent_params(
event_type, event_dict
)
return cls(date, title, event_type, fees, isin, note, shares, taxes, value)

@staticmethod
def _parse_type(event_dict: Dict[Any, Any]) -> Optional[EventType]:
event_type: Optional[EventType] = tr_event_type_mapping.get(
event_dict.get("eventType", ""), None
)
if event_type == ConditionalEventType.FAILED_CARD_TRANSACTION:
event_type = (
PPEventType.REMOVAL
if event_dict.get("status", "").lower() == "executed"
else None
)
return event_type

@classmethod
def _parse_type_dependent_params(
cls, event_type: EventType, event_dict: Dict[Any, Any]
) -> Tuple[Optional[Union[str, float]]]:
"""Parses the fees, isin, note, shares and taxes fields
Args:
event_type (EventType): _description_
event_dict (Dict[Any, Any]): _description_
Returns:
Tuple[Optional[Union[str, float]]]]: fees, isin, note, shares, taxes
"""
isin, shares, taxes, note, fees = (None,) * 5

if event_type is PPEventType.DIVIDEND:
isin = cls._parse_isin(event_dict)
taxes = cls._parse_taxes(event_dict)

elif isinstance(event_type, ConditionalEventType):
isin = cls._parse_isin(event_dict)
shares, fees = cls._parse_shares_and_fees(event_dict)
taxes = cls._parse_taxes(event_dict)

elif event_type is PPEventType.INTEREST:
taxes = cls._parse_taxes(event_dict)

elif event_type in [PPEventType.DEPOSIT, PPEventType.REMOVAL]:
note = cls._parse_card_note(event_dict)

return fees, isin, note, shares, taxes

@staticmethod
def _parse_isin(event_dict: Dict[Any, Any]) -> str:
"""Parses the isin
Args:
event_dict (Dict[Any, Any]): _description_
Returns:
str: isin
"""
sections = event_dict.get("details", {}).get("sections", [{}])
isin = event_dict.get("icon", "")
isin = isin[isin.find("/") + 1 :]
isin = isin[: isin.find("/")]
isin2 = isin
for section in sections:
action = section.get("action", None)
if action and action.get("type", {}) == "instrumentDetail":
isin2 = section.get("action", {}).get("payload")
break
if isin != isin2:
isin = isin2
return isin

@staticmethod
def _parse_shares_and_fees(event_dict: Dict[Any, Any]) -> Tuple[Optional[float]]:
"""Parses the amount of shares and the applicable fees
Args:
event_dict (Dict[Any, Any]): _description_
Returns:
Tuple[Optional[float]]: [shares, fees]
"""
return_vals = {}
sections = event_dict.get("details", {}).get("sections", [{}])
for section in sections:
if section.get("title") == "Transaktion":
data = section["data"]
shares_dicts = list(
filter(lambda x: x["title"] in ["Aktien", "Anteile"], data)
)
fees_dicts = list(filter(lambda x: x["title"] == "Gebühr", data))
titles = ["shares"] * len(shares_dicts) + ["fees"] * len(fees_dicts)
for key, elem_dict in zip(titles, shares_dicts + fees_dicts):
elem_unparsed = elem_dict.get("detail", {}).get("text", "")
elem_parsed = re.sub("[^\,\.\d-]", "", elem_unparsed).replace(
",", "."
)
return_vals[key] = (
None
if elem_parsed == "" or float(elem_parsed) == 0.0
else float(elem_parsed)
)
return return_vals["shares"], return_vals["fees"]

@staticmethod
def _parse_taxes(event_dict: Dict[Any, Any]) -> Tuple[Optional[float]]:
"""Parses the levied taxes
Args:
event_dict (Dict[Any, Any]): _description_
Returns:
Tuple[Optional[float]]: [taxes]
"""
# taxes keywords
taxes_keys = {"Steuer", "Steuern"}
# Gather all section dicts
sections = event_dict.get("details", {}).get("sections", [{}])
# Gather all dicts pertaining to transactions
transaction_dicts = filter(
lambda x: x["title"] in {"Transaktion", "Geschäft"}, sections
)
for transaction_dict in transaction_dicts:
# Filter for taxes dicts
data = transaction_dict.get("data", [{}])
taxes_dicts = filter(lambda x: x["title"] in taxes_keys, data)
# Iterate over dicts containing tax information and parse each one
for taxes_dict in taxes_dicts:
unparsed_taxes_val = taxes_dict.get("detail", {}).get("text", "")
parsed_taxes_val = re.sub("[^\,\.\d-]", "", unparsed_taxes_val).replace(
",", "."
)
if parsed_taxes_val != "" and float(parsed_taxes_val) != 0.0:
return float(parsed_taxes_val)

@staticmethod
def _parse_card_note(event_dict: Dict[Any, Any]) -> Optional[str]:
"""Parses the note associated with card transactions
Args:
event_dict (Dict[Any, Any]): _description_
Returns:
Optional[str]: note
"""
if event_dict.get("eventType", "").startswith("card_"):
return event_dict["eventType"]
Loading

0 comments on commit b3e70bf

Please sign in to comment.