Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improve number parsing and handle a few more events #193

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
+167 −32
Open

Improve number parsing and handle a few more events #193

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a77e2db
Improve number parsing and handle a few more events
RealCLanger Feb 14, 2025
9c608a0
Improve locale selection for number parsing and add possibility to du…
RealCLanger Feb 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ $ uvx --with git+https://github.com/pytr-org/pytr.git pytr

<!-- runcmd code:console COLUMNS=120 uv run --python 3.13 pytr -->
```console
usage: pytr [-h] [-v {warning,info,debug}] [-V]
usage: pytr [-h] [-v {warning,info,debug}] [-V] [--dump-debug-data-to-file DEBUG_DATA_FILE]
{help,login,dl_docs,portfolio,details,get_price_alarms,set_price_alarms,export_transactions,completion} ...

Use "pytr command_name --help" to get detailed help to a specific command
Expand All @@ -87,6 +87,8 @@ Options:
-v, --verbosity {warning,info,debug}
Set verbosity level (default: info)
-V, --version Print version information and quit
--dump-debug-data-to-file DEBUG_DATA_FILE
Dump debug data to a given file
```
<!-- end runcmd -->

Expand Down
153 changes: 123 additions & 30 deletions pytr/event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import pprint
import re
from dataclasses import dataclass
from datetime import datetime
Expand All @@ -6,6 +7,8 @@

from babel.numbers import NumberFormatError, parse_decimal

from pytr.utils import dump, dump_enabled, get_logger


class EventType(Enum):
pass
Expand Down Expand Up @@ -51,15 +54,14 @@ class PPEventType(EventType):
# Dividends
"CREDIT": PPEventType.DIVIDEND,
"ssp_corporate_action_invoice_cash": PPEventType.DIVIDEND,
# Failed card transactions
"card_failed_transaction": PPEventType.REMOVAL,
# Interests
"INTEREST_PAYOUT": PPEventType.INTEREST,
"INTEREST_PAYOUT_CREATED": PPEventType.INTEREST,
# Removals
"OUTGOING_TRANSFER": PPEventType.REMOVAL,
"OUTGOING_TRANSFER_DELEGATION": PPEventType.REMOVAL,
"PAYMENT_OUTBOUND": PPEventType.REMOVAL,
"card_failed_transaction": PPEventType.REMOVAL,
"card_order_billed": PPEventType.REMOVAL,
"card_successful_atm_withdrawal": PPEventType.REMOVAL,
"card_successful_transaction": PPEventType.REMOVAL,
Expand All @@ -72,10 +74,37 @@ class PPEventType(EventType):
"ORDER_EXECUTED": ConditionalEventType.TRADE_INVOICE,
"SAVINGS_PLAN_EXECUTED": ConditionalEventType.TRADE_INVOICE,
"SAVINGS_PLAN_INVOICE_CREATED": ConditionalEventType.TRADE_INVOICE,
"trading_savingsplan_executed": ConditionalEventType.TRADE_INVOICE,
"benefits_spare_change_execution": ConditionalEventType.TRADE_INVOICE,
"TRADE_INVOICE": ConditionalEventType.TRADE_INVOICE,
"TRADE_CORRECTED": ConditionalEventType.TRADE_INVOICE,
}

log = get_logger(__name__)

events_known_ignored = [
"CUSTOMER_CREATED",
"DEVICE_RESET",
"DOCUMENTS_ACCEPTED",
"DOCUMENTS_CREATED",
"EMAIL_VALIDATED",
"ORDER_CANCELED",
"ORDER_CREATED",
"ORDER_EXPIRED",
"ORDER_REJECTED",
"PUK_CREATED",
"REFERENCE_ACCOUNT_CHANGED",
"SECURITIES_ACCOUNT_CREATED",
"VERIFICATION_TRANSFER_ACCEPTED",
"card_failed_verification",
"card_successful_verification",
"current_account_activated",
"new_tr_iban",
"ssp_corporate_action_informative_notification",
"ssp_corporate_action_invoice_shares",
"ssp_dividend_option_customer_instruction",
]


@dataclass
class Event:
Expand Down Expand Up @@ -110,9 +139,16 @@ def from_dict(cls, event_dict: Dict[Any, Any]):

@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_dict.get("status", "").lower() == "canceled":
event_type = None
eventTypeStr = event_dict.get("eventType", "")
event_type: Optional[EventType] = tr_event_type_mapping.get(eventTypeStr, None)
if event_type is not None:
if event_dict.get("status", "").lower() == "canceled":
event_type = None
else:
if eventTypeStr not in events_known_ignored:
log.warning(f"Ignoring unknown event {eventTypeStr}")
if dump_enabled():
dump(f"Unknown event {eventTypeStr}: {pprint.pformat(event_dict, indent=4)}")
Comment on lines +150 to +151
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this could be handled with logging just as well, don't you think? And formatting the event_dict as JSON would make it easier to handle, for example to include as sample data in tests.

Suggested change
if dump_enabled():
dump(f"Unknown event {eventTypeStr}: {pprint.pformat(event_dict, indent=4)}")
self._log.debug("Unknown event %r: %s", eventTypeStr, json.dumps(event_dict, indent=4)))

Where the --dump-debug-data-to-file option is handled, we can add a FileHandler to the root logger.

handler = logging.FileHandler(args.dump_debug_data_to_file)
handler.setLevel(logging.DEBUG)
handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
logging.root.addHandler(handler)

And maybe shorten the option name to --debug-logfile?

Alternatively, we could forego the entire option and add a --debug or --verbose option to set the log level to DEBUG such that these messages show up in the console and people can grab it from there; or alternatively have a --debug --logfile myfile.log as a combination to achieve the same effect.

Copy link
Collaborator Author

@RealCLanger RealCLanger Mar 9, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK, I'll mull about this. The reason for the if dump_enabled(): construct is that always pretty printing (or json formatting) the data is quite expensive.

return event_type

@classmethod
Expand Down Expand Up @@ -182,17 +218,37 @@ def _parse_shares_and_fees(cls, event_dict: Dict[Any, Any]) -> Tuple[Optional[fl
Returns:
Tuple[Optional[float]]: shares, fees
"""
return_vals = {}
shares, fees = None, None
dump_dict = {"eventType": event_dict["eventType"], "id": event_dict["id"]}

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):
return_vals[key] = cls._parse_float_from_detail(elem_dict)
return return_vals.get("shares"), return_vals.get("fees")
transaction_dicts = filter(lambda x: x["title"] in ["Transaktion"], sections)
for transaction_dict in transaction_dicts:
dump_dict["maintitle"] = transaction_dict["title"]
data = transaction_dict.get("data", [{}])
shares_dicts = filter(lambda x: x["title"] in ["Aktien", "Anteile"], data)
for shares_dict in shares_dicts:
dump_dict["subtitle"] = shares_dict["title"]
dump_dict["type"] = "shares"
pref_locale = (
"en"
if event_dict["eventType"] in ["benefits_saveback_execution", "benefits_spare_change_execution"]
and shares_dict["title"] == "Aktien"
else "de"
)
shares = cls._parse_float_from_detail(shares_dict, dump_dict, pref_locale)
break

fees_dicts = filter(lambda x: x["title"] == "Gebühr", data)
for fees_dict in fees_dicts:
dump_dict["subtitle"] = fees_dict["title"]
dump_dict["type"] = "fees"
fees = cls._parse_float_from_detail(fees_dict, dump_dict)
break

break

return shares, fees

@classmethod
def _parse_taxes(cls, event_dict: Dict[Any, Any]) -> Optional[float]:
Expand All @@ -204,23 +260,26 @@ def _parse_taxes(cls, event_dict: Dict[Any, Any]) -> Optional[float]:
Returns:
Optional[float]: taxes
"""
# taxes keywords
taxes_keys = {"Steuer", "Steuern"}
# Gather all section dicts
parsed_taxes_val = None
dump_dict = {"eventType": event_dict["eventType"], "id": event_dict["id"]}
pref_locale = "en" if event_dict["eventType"] in ["INTEREST_PAYOUT", "trading_savingsplan_executed"] else "de"

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)
transaction_dicts = filter(lambda x: x["title"] in ["Transaktion", "Geschäft"], sections)
for transaction_dict in transaction_dicts:
# Filter for taxes dicts
dump_dict["maintitle"] = transaction_dict["title"]
data = transaction_dict.get("data", [{}])
taxes_dicts = filter(lambda x: x["title"] in taxes_keys, data)
taxes_dicts = filter(lambda x: x["title"] in ["Steuer", "Steuern"], data)
# Iterate over dicts containing tax information and parse each one
for taxes_dict in taxes_dicts:
parsed_taxes_val = cls._parse_float_from_detail(taxes_dict)
if parsed_taxes_val is not None:
return parsed_taxes_val
dump_dict["subtitle"] = taxes_dict["title"]
dump_dict["type"] = "taxes"
parsed_taxes_val = cls._parse_float_from_detail(taxes_dict, dump_dict, pref_locale)
break
break

return None
return parsed_taxes_val

@staticmethod
def _parse_card_note(event_dict: Dict[Any, Any]) -> Optional[str]:
Expand All @@ -237,7 +296,11 @@ def _parse_card_note(event_dict: Dict[Any, Any]) -> Optional[str]:
return None

@staticmethod
def _parse_float_from_detail(elem_dict: Dict[str, Any]) -> Optional[float]:
def _parse_float_from_detail(
elem_dict: Dict[str, Any],
dump_dict={"eventType": "Unknown", "id": "Unknown", "type": "Unknown"},
pref_locale="de",
) -> Optional[float]:
"""Parses a "detail" dictionary potentially containing a float in a certain locale format

Args:
Expand All @@ -247,13 +310,15 @@ def _parse_float_from_detail(elem_dict: Dict[str, Any]) -> Optional[float]:
Optional[float]: parsed float value or None
"""
unparsed_val = elem_dict.get("detail", {}).get("text", "")
if unparsed_val == "":
return None
parsed_val = re.sub(r"[^\,\.\d-]", "", unparsed_val)

# Try the locale that will fail more likely first.
if "." not in parsed_val:
locales = ("en", "de")
else:
# Try the preferred locale first
if pref_locale == "de":
locales = ("de", "en")
else:
locales = ("en", "de")

try:
result = float(parse_decimal(parsed_val, locales[0], strict=True))
Expand All @@ -262,4 +327,32 @@ def _parse_float_from_detail(elem_dict: Dict[str, Any]) -> Optional[float]:
result = float(parse_decimal(parsed_val, locales[1], strict=True))
except NumberFormatError:
return None
log.warning(
f"Number {parsed_val} parsed as {locales[1]} although preference was {locales[0]}. ({dump_dict['eventType']}, {dump_dict['id']}, {dump_dict['type']})"
)
if dump_enabled():
dump(
f"Number {parsed_val} parsed as as {locales[1]} although preference was {locales[0]}: {pprint.pformat(dump_dict, indent=4)}"
)
return None if result == 0.0 else result

alternative_result = None
if "," in parsed_val or "." in parsed_val:
try:
alternative_result = float(parse_decimal(parsed_val, locales[1], strict=True))
except NumberFormatError:
pass

if alternative_result is None:
if dump_enabled():
dump(f"Number {parsed_val} parsed as {locales[0]}: {pprint.pformat(dump_dict, indent=4)}")
else:
log.debug(
f"Number {parsed_val} as {locales[0]} but could also be parsed as {locales[1]}. ({dump_dict['eventType']}, {dump_dict['id']}, {dump_dict['type']})"
)
if dump_enabled():
dump(
f"Number {parsed_val} as {locales[0]} but could also be parsed as {locales[1]}: {pprint.pformat(dump_dict, indent=4)}"
)

return None if result == 0.0 else result
11 changes: 10 additions & 1 deletion pytr/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
from pytr.dl import DL
from pytr.portfolio import Portfolio
from pytr.transactions import export_transactions
from pytr.utils import check_version, get_logger
from pytr.utils import check_version, enable_debug_dump, get_logger


def get_main_parser():
Expand Down Expand Up @@ -45,6 +45,14 @@ def formatter(prog):
help="Print version information and quit",
action="store_true",
)
parser.add_argument(
"--dump-debug-data-to-file",
help="Dump debug data to a given file",
metavar="DEBUG_DATA_FILE",
type=Path,
default=None,
)

parser_cmd = parser.add_subparsers(help="Desired action to perform", dest="command")

# help
Expand Down Expand Up @@ -238,6 +246,7 @@ def main():
args = parser.parse_args()
# print(vars(args))

enable_debug_dump(args.dump_debug_data_to_file)
log = get_logger(__name__, args.verbosity)
log.setLevel(args.verbosity.upper())
log.debug("logging is set to debug")
Expand Down
22 changes: 22 additions & 0 deletions pytr/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,28 @@
import requests
from packaging import version

dump_file = None


def enable_debug_dump(df):
global dump_file
dump_file = df
if df is not None:
with open(dump_file, "w"):
pass


def dump_enabled():
return dump_file is not None


def dump(str):
if dump_file is None:
return
with open(dump_file, "a") as file:
file.write(str + "\n")


log_level = None


Expand Down
9 changes: 9 additions & 0 deletions tests/test_parsing.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,15 @@ def test(data: Dict[str, Any], number: float):
9400.0,
)

test(
{
"title": "Anteile",
"detail": {"text": "1,875", "trend": None, "action": None, "type": "text"},
"style": "plain",
},
1.875,
)

test(
{
"title": "Aktien",
Expand Down
Loading