fix: improve recurring detection with fuzzy grouping and wider bands
Use fuzzy description grouping (strip digits/symbols) so transactions like "FREEDOM MTG PYMTS 1234" and "FREEDOM MTG PYMTS 5678" are detected together. Switch to median for amounts and intervals for outlier resilience. Widen frequency bands to eliminate gaps (biweekly 10-18, monthly 19-45) and add semi-annual detection. Raise default amount tolerance to 15% and allow 25% outlier payments. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
+50
-16
@@ -1,15 +1,32 @@
|
||||
# src/services/recurring.py
|
||||
from collections import defaultdict
|
||||
import re
|
||||
from collections import Counter, defaultdict
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.models.transaction import Transaction
|
||||
|
||||
|
||||
def _grouping_key(description: str) -> str:
|
||||
"""Derive a fuzzy key that ignores trailing reference numbers / codes."""
|
||||
key = re.sub(r"\d+", "", description)
|
||||
key = re.sub(r"[*#]+", " ", key)
|
||||
return " ".join(key.split()).strip().lower()
|
||||
|
||||
|
||||
def _median(values: list[float]) -> float:
|
||||
s = sorted(values)
|
||||
n = len(s)
|
||||
if n % 2 == 1:
|
||||
return s[n // 2]
|
||||
return (s[n // 2 - 1] + s[n // 2]) / 2
|
||||
|
||||
|
||||
class RecurringDetector:
|
||||
def __init__(self, session: Session):
|
||||
self.session = session
|
||||
|
||||
def detect(self, amount_tolerance: float = 0.10) -> list[dict]:
|
||||
def detect(self, amount_tolerance: float = 0.15) -> list[dict]:
|
||||
txns = (
|
||||
self.session.query(Transaction)
|
||||
.filter(Transaction.amount < 0)
|
||||
@@ -18,36 +35,51 @@ class RecurringDetector:
|
||||
.all()
|
||||
)
|
||||
|
||||
# Fuzzy-group by description so trailing reference numbers don't split groups
|
||||
groups: dict[str, list[Transaction]] = defaultdict(list)
|
||||
for txn in txns:
|
||||
groups[txn.description].append(txn)
|
||||
groups[_grouping_key(txn.description)].append(txn)
|
||||
|
||||
results = []
|
||||
for desc, group in groups.items():
|
||||
for _key, group in groups.items():
|
||||
if len(group) < 2:
|
||||
continue
|
||||
|
||||
amounts = [abs(float(t.amount)) for t in group]
|
||||
avg_amount = sum(amounts) / len(amounts)
|
||||
median_amount = _median(amounts)
|
||||
|
||||
if any(abs(a - avg_amount) / avg_amount > amount_tolerance for a in amounts):
|
||||
if median_amount == 0:
|
||||
continue
|
||||
|
||||
# Allow some outliers: at least 75% of amounts must be within tolerance
|
||||
within = sum(1 for a in amounts if abs(a - median_amount) / median_amount <= amount_tolerance)
|
||||
if within / len(amounts) < 0.75:
|
||||
continue
|
||||
|
||||
dates = sorted(t.date for t in group)
|
||||
intervals = [(dates[i + 1] - dates[i]).days for i in range(len(dates) - 1)]
|
||||
avg_interval = sum(intervals) / len(intervals)
|
||||
if not intervals:
|
||||
continue
|
||||
median_interval = _median(intervals)
|
||||
|
||||
frequency = self._classify_frequency(avg_interval)
|
||||
frequency = self._classify_frequency(median_interval)
|
||||
if frequency is None:
|
||||
continue
|
||||
|
||||
annual_multiplier = {"weekly": 52, "biweekly": 26, "monthly": 12, "quarterly": 4, "annual": 1}
|
||||
annual_multiplier = {
|
||||
"weekly": 52, "biweekly": 26, "monthly": 12,
|
||||
"quarterly": 4, "semi-annual": 2, "annual": 1,
|
||||
}
|
||||
|
||||
# Use the most common original description for display
|
||||
desc_counts = Counter(t.description for t in group)
|
||||
representative_desc = desc_counts.most_common(1)[0][0]
|
||||
|
||||
results.append({
|
||||
"description": desc,
|
||||
"typical_amount": round(avg_amount, 2),
|
||||
"description": representative_desc,
|
||||
"typical_amount": round(median_amount, 2),
|
||||
"frequency": frequency,
|
||||
"annual_cost": round(avg_amount * annual_multiplier[frequency], 2),
|
||||
"annual_cost": round(median_amount * annual_multiplier[frequency], 2),
|
||||
"occurrences": len(group),
|
||||
"last_date": max(dates),
|
||||
})
|
||||
@@ -58,12 +90,14 @@ class RecurringDetector:
|
||||
def _classify_frequency(self, avg_days: float) -> str | None:
|
||||
if 5 <= avg_days <= 9:
|
||||
return "weekly"
|
||||
elif 12 <= avg_days <= 16:
|
||||
elif 10 <= avg_days <= 18:
|
||||
return "biweekly"
|
||||
elif 25 <= avg_days <= 35:
|
||||
elif 19 <= avg_days <= 45:
|
||||
return "monthly"
|
||||
elif 80 <= avg_days <= 100:
|
||||
elif 75 <= avg_days <= 105:
|
||||
return "quarterly"
|
||||
elif 350 <= avg_days <= 380:
|
||||
elif 160 <= avg_days <= 200:
|
||||
return "semi-annual"
|
||||
elif 340 <= avg_days <= 395:
|
||||
return "annual"
|
||||
return None
|
||||
|
||||
@@ -68,7 +68,7 @@ class RecurringView(QWidget):
|
||||
header.setSectionResizeMode(6, QHeaderView.ResizeMode.Fixed) # Confirm btn
|
||||
header.setSectionResizeMode(7, QHeaderView.ResizeMode.Fixed) # Dismiss btn
|
||||
self.table.setColumnWidth(1, 100)
|
||||
self.table.setColumnWidth(2, 100)
|
||||
self.table.setColumnWidth(2, 120)
|
||||
self.table.setColumnWidth(3, 110)
|
||||
self.table.setColumnWidth(4, 100)
|
||||
self.table.setColumnWidth(5, 90)
|
||||
|
||||
Reference in New Issue
Block a user