personal-finance/budget
2
1
Fork 0
Code Issues 11 Pull Requests Projects 6 Releases 2 Wiki Activity

[Refactor] Creates nullifier transformer

Browse Source 
Removes nullifier code from Categorizer.
Fixes bug in nullifier rules assertion.
This commit is contained in:
Luís Murta 2023-04-20 22:59:16 +01:00
parent b9b38db99a
commit 2882bfdada
Signed by: satprog
GPG Key ID: 169EF1BBD7049F94
6 changed files with 156 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
pfbudget/core/manager.py
View File

@ -28,6 +28,7 @@ from pfbudget.extract.nordigen import NordigenClient, NordigenCredentialsManager
from pfbudget.extract.parsers import parse_data
from pfbudget.extract.psd2 import PSD2Extractor
from pfbudget.transform.categorizer import Categorizer
from pfbudget.transform.nullifier import Nullifier
class Manager:
@ -100,6 +101,10 @@ class Manager:
)
categories = session.get(Category)
tags = session.get(Tag)
null_rules = [cat.rules for cat in categories if cat.name == "null"]
Nullifier(null_rules).transform_inplace(uncategorized)
Categorizer().rules(uncategorized, categories, tags, params[0])
case Operation.BankMod:

45
pfbudget/transform/categorizer.py
View File

@ -1,5 +1,4 @@
from codetiming import Timer
from datetime import timedelta
from typing import Sequence
import pfbudget.db.model as t
@ -29,55 +28,11 @@ class Categorizer:
tags (Sequence[Tag]): currently available tags
"""
if nullify:
try:
null = next(cat for cat in categories if cat.name == "null")
print("Nullifying")
self._nullify(transactions, null)
except StopIteration:
print("Null category not defined")
categories = [cat for cat in categories if cat.name != "null"]
self._rule_based_categories(transactions, categories)
self._rule_based_tags(transactions, tags)
@Timer(name="nullify")
def _nullify(self, transactions: Sequence[t.BankTransaction], null: t.Category):
count = 0
matching = []
for transaction in transactions:
for cancel in (
cancel
for cancel in transactions
if (
transaction.date - timedelta(days=self.options["null_days"])
<= cancel.date
<= transaction.date + timedelta(days=self.options["null_days"])
and cancel != transaction
and cancel.bank != transaction.bank
and cancel.amount == -transaction.amount
and transaction not in matching
and cancel not in matching
and all(r.matches(transaction) for r in null.rules)
and all(r.matches(cancel) for r in null.rules)
)
):
transaction.category = t.TransactionCategory(
name="null",
selector=t.CategorySelector(t.Selector_T.nullifier),
)
cancel.category = t.TransactionCategory(
name="null",
selector=t.CategorySelector(t.Selector_T.nullifier),
)
matching.extend([transaction, cancel])
count += 2
break
print(f"Nullified {count} of {len(transactions)} transactions")
@Timer(name="categoryrules")
def _rule_based_categories(
self,

2
pfbudget/transform/exceptions.py Normal file
View File

@ -0,0 +1,2 @@
class MoreThanOneMatchError(Exception):
pass

94
pfbudget/transform/nullifier.py Normal file
View File

@ -0,0 +1,94 @@
from copy import deepcopy
import datetime as dt
from typing import Sequence
from .exceptions import MoreThanOneMatchError
from .transform import Transformer
from pfbudget.db.model import (
CategorySelector,
Selector_T,
Transaction,
TransactionCategory,
)
class Nullifier(Transformer):
NULL_DAYS = 4
def __init__(self, rules=None):
self.rules = rules
def transform(self, transactions: Sequence[Transaction]) -> Sequence[Transaction]:
"""transform
Find transactions that nullify each others, e.g. transfers between banks or
between bank and credit cards.
Args:
transactions (Sequence[Transaction]): ordered sequence of transactions
Raises:
MoreThanOneMatchError: if there is more than a match for a single transation
Returns:
Sequence[Transaction]: nullified sequence of transactions
"""
result = deepcopy(transactions)
for i, transaction in enumerate(result[:-1]):
if matches := [t for t in result[i + 1 :] if self._cancels(transaction, t)]:
if len(matches) > 1:
raise MoreThanOneMatchError(f"{transaction} -> {matches}")
match = matches[0]
transaction = self._nullify(transaction)
match = self._nullify(match)
return result
def transform_inplace(self, transactions: Sequence[Transaction]) -> None:
"""_summary_
Find transactions that nullify each others, e.g. transfers between banks or
between bank and credit cards.
Args:
transactions (Sequence[Transaction]): ordered sequence of transactions that
will be modified inplace
Raises:
MoreThanOneMatchError: if there is more than a match for a single transation
"""
for transaction in transactions:
if matches := [t for t in transactions if self._cancels(transaction, t)]:
if len(matches) > 1:
raise MoreThanOneMatchError(f"{transaction} -> {matches}")
match = matches[0]
transaction = self._nullify(transaction)
match = self._nullify(match)
def _cancels(self, transaction: Transaction, cancel: Transaction):
return (
transaction.date
<= cancel.date
<= transaction.date + dt.timedelta(days=self.NULL_DAYS)
and cancel != transaction
and cancel.bank != transaction.bank
and cancel.amount == -transaction.amount
and (not cancel.category or cancel.category.name != "null")
and (
any(r.matches(transaction) for r in self.rules) if self.rules else True
)
and (any(r.matches(cancel) for r in self.rules) if self.rules else True)
)
def _nullify(self, transaction: Transaction) -> Transaction:
transaction.category = TransactionCategory(
"null", selector=CategorySelector(Selector_T.nullifier)
)
return transaction

4
pfbudget/transform/transform.py
View File

@ -8,3 +8,7 @@ class Transformer(ABC):
@abstractmethod
def transform(self, _: Sequence[Transaction]) -> Sequence[Transaction]:
raise NotImplementedError
@abstractmethod
def transform_inplace(self, _: Sequence[Transaction]) -> None:
raise NotImplementedError

60
tests/test_transform.py
View File

@ -6,29 +6,71 @@ import mocks.categories as mock
from pfbudget.db.model import (
Bank,
BankTransaction,
CategoryRule,
CategorySelector,
Selector_T,
TransactionCategory,
)
from pfbudget.transform.categorizer import Categorizer
from pfbudget.transform.nullifier import Nullifier
from pfbudget.transform.transform import Transformer
class TestTransform:
def test_nullify(self):
def test_nullifier(self):
transactions = [
BankTransaction(
date(2023, 1, 1), "", Decimal("-500"), Bank("Bank#1", "", "")
),
BankTransaction(
date(2023, 1, 2), "", Decimal("500"), Bank("Bank#2", "", "")
),
BankTransaction(date(2023, 1, 1), "", Decimal("-500"), "Bank#1"),
BankTransaction(date(2023, 1, 2), "", Decimal("500"), "Bank#2"),
]
for t in transactions:
assert not t.category
categorizer = Categorizer()
categorizer.rules(transactions, [mock.category_null], [])
categorizer: Transformer = Nullifier()
transactions = categorizer.transform(transactions)
for t in transactions:
assert t.category == TransactionCategory(
"null", CategorySelector(Selector_T.nullifier)
)
def test_nullifier_inplace(self):
transactions = [
BankTransaction(date(2023, 1, 1), "", Decimal("-500"), "Bank#1"),
BankTransaction(date(2023, 1, 2), "", Decimal("500"), "Bank#2"),
]
for t in transactions:
assert not t.category
categorizer: Transformer = Nullifier()
categorizer.transform_inplace(transactions)
for t in transactions:
assert t.category == TransactionCategory(
"null", CategorySelector(Selector_T.nullifier)
)
def test_nullifier_with_rules(self):
transactions = [
BankTransaction(date(2023, 1, 1), "", Decimal("-500"), "Bank#1"),
BankTransaction(date(2023, 1, 2), "", Decimal("500"), "Bank#2"),
]
for t in transactions:
assert not t.category
rules = [CategoryRule(None, None, None, None, "Bank#1", None, None, "null")]
categorizer: Transformer = Nullifier(rules)
transactions = categorizer.transform(transactions)
for t in transactions:
assert not t.category
rules.append(CategoryRule(None, None, None, None, "Bank#2", None, None, "null"))
categorizer: Transformer = Nullifier(rules)
transactions = categorizer.transform(transactions)
for t in transactions:
assert t.category == TransactionCategory(