Enhance transaction categorization: add category_type to AkahuTransaction and update related methods

Author: Slaymish

src/bank_sync/akahu_client.py

@@ -41,7 +41,7 @@ def from_payload(cls, payload: Dict, *, source: str, account_name: str = "unknow
             source=source,
         )
 
-    def to_row(self, *, category: str, is_transfer: bool, imported_at: datetime) -> List[str]:
+    def to_row(self, *, category: str, category_type: str, is_transfer: bool, imported_at: datetime) -> List[str]:
         """Return the row representation expected by Google Sheets."""
 
         return [
@@ -53,6 +53,7 @@ def to_row(self, *, category: str, is_transfer: bool, imported_at: datetime) ->
             self.description_raw,
             self.merchant_normalised,
             category,
+            category_type,
             str(is_transfer).upper(),
             self.source,
             imported_at.isoformat(),

src/bank_sync/categoriser.py

@@ -36,6 +36,7 @@ class CategoryRule:
     pattern: str
     field: str
     category: str
+    category_type: str = field(default="", compare=False)
     amount_condition: AmountCondition | None = field(default=None, compare=False)
 
     def matches(self, transaction: dict) -> bool:
@@ -66,16 +67,18 @@ def __init__(self, rules: Iterable[dict]):
                     pattern=pattern,
                     field=rule.get("field", "merchant_normalised"),
                     category=rule.get("category", "Uncategorised"),
+                    category_type=rule.get("category_type", ""),
                     amount_condition=_parse_amount_condition(rule.get("amount_condition", "")),
                 )
             )
         self._rules.sort()
 
-    def categorise(self, transaction: dict) -> str:
+    def categorise(self, transaction: dict) -> tuple[str, str]:
+        """Return (category, category_type) for the transaction."""
         for rule in self._rules:
             if rule.matches(transaction):
-                return rule.category
-        return "Uncategorised"
+                return (rule.category, rule.category_type)
+        return ("Uncategorised", "")
 
     @staticmethod
     def detect_transfer(transaction: dict) -> bool:

src/bank_sync/main.py

@@ -158,9 +158,9 @@ def run_sync(dry_run: bool = False, reset_state: bool = False) -> None:
             "merchant_normalised": transaction.merchant_normalised,
             "source": transaction.source,
         }
-        category = categoriser.categorise(transaction_dict)
+        category, category_type = categoriser.categorise(transaction_dict)
         is_transfer = categoriser.detect_transfer(transaction_dict)
-        row = transaction.to_row(category=category, is_transfer=is_transfer, imported_at=imported_at)
+        row = transaction.to_row(category=category, category_type=category_type, is_transfer=is_transfer, imported_at=imported_at)
 
         if transaction.id not in existing_map:
             new_rows.append(row)

src/bank_sync/sheets_client.py

@@ -20,6 +20,7 @@
     "description_raw",
     "merchant_normalised",
     "category",
+    "category_type",
     "is_transfer",
     "source",
     "imported_at",
@@ -69,7 +70,7 @@ def _get_sheet_id(self, sheet_name: str) -> int:
         return self._sheet_id_cache.get(sheet_name, 0)
 
     def fetch_transactions(self) -> List[SheetTransaction]:
-        range_name = f"{self._transactions_tab}!A2:K"
+        range_name = f"{self._transactions_tab}!A2:L"
         response = self._service.spreadsheets().values().get(
             spreadsheetId=self._spreadsheet_id, range=range_name
         ).execute()
@@ -84,7 +85,7 @@ def fetch_transactions(self) -> List[SheetTransaction]:
     def append_transactions(self, rows: Iterable[List[str]]) -> None:
         if not rows:
             return
-        range_name = f"{self._transactions_tab}!A:K"
+        range_name = f"{self._transactions_tab}!A:L"
         body = {"values": list(rows)}
         LOGGER.info("Appending %s new transactions", len(body["values"]))
         self._service.spreadsheets().values().append(
@@ -95,7 +96,7 @@ def append_transactions(self, rows: Iterable[List[str]]) -> None:
         ).execute()
 
     def update_transaction(self, row_index: int, row: List[str]) -> None:
-        range_name = f"{self._transactions_tab}!A{row_index}:K{row_index}"
+        range_name = f"{self._transactions_tab}!A{row_index}:L{row_index}"
         LOGGER.info("Updating row %s", row_index)
         self._service.spreadsheets().values().update(
             spreadsheetId=self._spreadsheet_id,
@@ -111,7 +112,7 @@ def batch_update_transactions(self, updates: List[tuple[int, List[str]]]) -> Non
 
         data = [
             {
-                "range": f"{self._transactions_tab}!A{row_index}:K{row_index}",
+                "range": f"{self._transactions_tab}!A{row_index}:L{row_index}",
                 "values": [row]
             }
             for row_index, row in updates
@@ -153,20 +154,21 @@ def delete_rows(self, row_indices: List[int]) -> None:
             raise
 
     def fetch_category_rules(self) -> List[Dict[str, str]]:
-        range_name = f"{self._category_tab}!A2:E"
+        range_name = f"{self._category_tab}!A2:F"
         response = self._service.spreadsheets().values().get(
             spreadsheetId=self._spreadsheet_id, range=range_name
         ).execute()
         rules = []
         for row in response.get("values", []):
-            padded = row + [""] * (5 - len(row))
+            padded = row + [""] * (6 - len(row))
             rules.append(
                 {
                     "pattern": padded[0],
                     "field": padded[1] or "merchant_normalised",
                     "category": padded[2] or "Uncategorised",
                     "priority": padded[3] or "1000",
                     "amount_condition": padded[4],
+                    "category_type": padded[5],
                 }
             )
         return rules
@@ -177,15 +179,15 @@ def upload_category_rules(self, rows: List[List[str]]) -> None:
             raise ValueError("No rows to upload")
 
         # Clear existing content (keep header row)
-        range_name = f"{self._category_tab}!A2:E"
+        range_name = f"{self._category_tab}!A2:F"
         self._service.spreadsheets().values().clear(
             spreadsheetId=self._spreadsheet_id,
             range=range_name
         ).execute()
         LOGGER.info("Cleared existing category rules")
 
         # Upload new data (including header if first row)
-        range_name = f"{self._category_tab}!A1:E"
+        range_name = f"{self._category_tab}!A1:F"
         body = {"values": rows}
         self._service.spreadsheets().values().update(
             spreadsheetId=self._spreadsheet_id,

tests/test_akahu_client.py

@@ -54,11 +54,12 @@ def test_transaction_from_payload(transaction_payload):
 
 def test_transaction_to_row_formats_numbers(transaction_payload):
     txn = AkahuTransaction.from_payload(transaction_payload, source="akahu_bnz", account_name="Cheque")
-    row = txn.to_row(category="Groceries", is_transfer=False, imported_at=dt.datetime(2023, 9, 2, 10, 0))
+    row = txn.to_row(category="Groceries", category_type="E", is_transfer=False, imported_at=dt.datetime(2023, 9, 2, 10, 0))
     assert row[:4] == ["txn_123", "2023-09-02", "Cheque", "-12.34"]
     assert row[4] == "120.55"
     assert row[7] == "Groceries"
-    assert row[8] == "FALSE"
+    assert row[8] == "E"
+    assert row[9] == "FALSE"
 
 
 def test_fetch_settled_transactions_paginates(transaction_payload):

tests/test_categoriser.py

@@ -8,7 +8,8 @@ def test_categoriser_applies_highest_priority_match():
     ]
     categoriser = Categoriser(rules)
     txn = {"merchant_normalised": "Countdown Ponsonby"}
-    assert categoriser.categorise(txn) == "Groceries"
+    category, category_type = categoriser.categorise(txn)
+    assert category == "Groceries"
 
 
 def test_categoriser_skips_empty_patterns():
@@ -17,7 +18,8 @@ def test_categoriser_skips_empty_patterns():
         {"pattern": "ferry", "category": "Transport"},
     ]
     categoriser = Categoriser(rules)
-    assert categoriser.categorise({"merchant_normalised": "Ferry ride"}) == "Transport"
+    category, category_type = categoriser.categorise({"merchant_normalised": "Ferry ride"})
+    assert category == "Transport"
 
 
 def test_categoriser_honours_amount_conditions():
@@ -36,14 +38,10 @@ def test_categoriser_honours_amount_conditions():
         },
     ]
     categoriser = Categoriser(rules)
-    assert (
-        categoriser.categorise({"merchant_normalised": "New World", "amount": "12.00"})
-        == "Groceries"
-    )
-    assert (
-        categoriser.categorise({"merchant_normalised": "New World", "amount": "5.00"})
-        == "Snacks"
-    )
+    category1, _ = categoriser.categorise({"merchant_normalised": "New World", "amount": "12.00"})
+    assert category1 == "Groceries"
+    category2, _ = categoriser.categorise({"merchant_normalised": "New World", "amount": "5.00"})
+    assert category2 == "Snacks"
 
 
 def test_categoriser_supports_exact_amounts_and_or_conditions():
@@ -68,18 +66,12 @@ def test_categoriser_supports_exact_amounts_and_or_conditions():
         },
     ]
     categoriser = Categoriser(rules)
-    assert (
-        categoriser.categorise({"merchant_normalised": "Coffee", "amount": "4.50"})
-        == "Work Coffee"
-    )
-    assert (
-        categoriser.categorise({"merchant_normalised": "Coffee", "amount": "0"})
-        == "Free Coffee"
-    )
-    assert (
-        categoriser.categorise({"merchant_normalised": "Coffee", "amount": "-4"})
-        == "Discount Coffee"
-    )
+    category1, _ = categoriser.categorise({"merchant_normalised": "Coffee", "amount": "4.50"})
+    assert category1 == "Work Coffee"
+    category2, _ = categoriser.categorise({"merchant_normalised": "Coffee", "amount": "0"})
+    assert category2 == "Free Coffee"
+    category3, _ = categoriser.categorise({"merchant_normalised": "Coffee", "amount": "-4"})
+    assert category3 == "Discount Coffee"
 
 
 def test_categoriser_ignores_amount_condition_when_unparseable():
@@ -91,7 +83,8 @@ def test_categoriser_ignores_amount_condition_when_unparseable():
         }
     ]
     categoriser = Categoriser(rules)
-    assert categoriser.categorise({"merchant_normalised": "Countdown", "amount": "1"}) == "Groceries"
+    category, _ = categoriser.categorise({"merchant_normalised": "Countdown", "amount": "1"})
+    assert category == "Groceries"
 
 
 def test_categoriser_falls_back_when_amount_missing():
@@ -104,10 +97,42 @@ def test_categoriser_falls_back_when_amount_missing():
         {"pattern": "fuel", "category": "Misc"},
     ]
     categoriser = Categoriser(rules)
-    assert categoriser.categorise({"merchant_normalised": "Fuel stop"}) == "Misc"
+    category, _ = categoriser.categorise({"merchant_normalised": "Fuel stop"})
+    assert category == "Misc"
 
 
 def test_detect_transfer_checks_multiple_fields():
     txn = {"description_raw": "Internal Transfer", "merchant_normalised": "BNZ"}
     assert Categoriser.detect_transfer(txn) is True
     assert Categoriser.detect_transfer({"description_raw": "Cafe"}) is False
+
+
+def test_categoriser_returns_category_type():
+    rules = [
+        {
+            "pattern": "countdown",
+            "category": "Groceries",
+            "category_type": "E",
+            "priority": 10,
+        },
+        {
+            "pattern": "sharesies",
+            "category": "Investments",
+            "category_type": "Iv",
+            "priority": 5,
+        },
+    ]
+    categoriser = Categoriser(rules)
+    
+    category, category_type = categoriser.categorise({"merchant_normalised": "Countdown"})
+    assert category == "Groceries"
+    assert category_type == "E"
+    
+    category, category_type = categoriser.categorise({"merchant_normalised": "Sharesies"})
+    assert category == "Investments"
+    assert category_type == "Iv"
+    
+    # Test uncategorised transaction
+    category, category_type = categoriser.categorise({"merchant_normalised": "Unknown Store"})
+    assert category == "Uncategorised"
+    assert category_type == ""