From f538cad6ae2f35aa814db14af39854d0b7ef03c2 Mon Sep 17 00:00:00 2001 From: andy Date: Tue, 10 Feb 2026 14:56:53 -0500 Subject: [PATCH] feat: cross-account transfer detection Co-Authored-By: Claude Opus 4.6 --- src/services/transfer_detector.py | 85 ++++++++++++++++++++ tests/services/test_transfer_detector.py | 99 ++++++++++++++++++++++++ 2 files changed, 184 insertions(+) create mode 100644 src/services/transfer_detector.py create mode 100644 tests/services/test_transfer_detector.py diff --git a/src/services/transfer_detector.py b/src/services/transfer_detector.py new file mode 100644 index 0000000..eebc84c --- /dev/null +++ b/src/services/transfer_detector.py @@ -0,0 +1,85 @@ +import datetime +from sqlalchemy.orm import Session + +from src.models.transaction import Transaction + + +# Known transfer patterns +TRANSFER_PATTERNS = [ + "CREDIT CRD EPAY", + "Payment Thank You", + "CAPITAL ONE TRANSFER", + "AMEX EPAYMENT", + "AMZ_STORECRD_PMT", +] + + +class TransferDetector: + def __init__(self, session: Session): + self.session = session + + def detect(self, date_tolerance_days: int = 3) -> list[dict]: + """Find matching transfer pairs across accounts.""" + txns = self.session.query(Transaction).filter( + Transaction.is_transfer == False + ).all() + + # Separate positive and negative transactions + positives = [t for t in txns if float(t.amount) > 0] + negatives = [t for t in txns if float(t.amount) < 0] + + pairs = [] + matched_ids = set() + + for neg in negatives: + if neg.id in matched_ids: + continue + # Check if this looks like a transfer + if not self._is_transfer_pattern(neg.description): + continue + + neg_amount = abs(float(neg.amount)) + + for pos in positives: + if pos.id in matched_ids: + continue + if pos.account_id == neg.account_id: + continue # Must be different accounts + + pos_amount = float(pos.amount) + if abs(pos_amount - neg_amount) > 0.01: + continue # Amounts must match + + date_diff = abs((neg.date - pos.date).days) + if date_diff > date_tolerance_days: + continue # Dates must be close + + pairs.append({ + "outgoing_id": neg.id, + "incoming_id": pos.id, + "amount": neg_amount, + "date": neg.date, + }) + matched_ids.add(neg.id) + matched_ids.add(pos.id) + break + + return pairs + + def mark_transfers(self) -> int: + """Detect and mark transfer pairs.""" + pairs = self.detect() + for pair in pairs: + outgoing = self.session.get(Transaction, pair["outgoing_id"]) + incoming = self.session.get(Transaction, pair["incoming_id"]) + if outgoing and incoming: + outgoing.is_transfer = True + incoming.is_transfer = True + outgoing.transfer_pair_id = incoming.id + incoming.transfer_pair_id = outgoing.id + self.session.commit() + return len(pairs) + + def _is_transfer_pattern(self, description: str) -> bool: + desc_upper = description.upper() + return any(p.upper() in desc_upper for p in TRANSFER_PATTERNS) diff --git a/tests/services/test_transfer_detector.py b/tests/services/test_transfer_detector.py new file mode 100644 index 0000000..12f7986 --- /dev/null +++ b/tests/services/test_transfer_detector.py @@ -0,0 +1,99 @@ +import datetime +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from src.db import Base +from src.models import * +from src.services.transfer_detector import TransferDetector + + +def make_session(): + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + return Session(engine) + + +def test_detect_matching_transfer(): + session = make_session() + member = HouseholdMember(name="Andrew", relationship="self") + session.add(member) + session.flush() + checking = Account(name="Checking", institution="WF", account_type="checking", owner_id=member.id) + chase = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id) + session.add_all([checking, chase]) + session.flush() + + # Payment from checking to Chase + txn1 = Transaction( + date=datetime.date(2026, 1, 29), amount=-1461.35, + description="CHASE CREDIT CRD EPAY", raw_description="CHASE CREDIT CRD EPAY 260128 9077835526 ANDREW B CONLON", + account_id=checking.id, + ) + txn2 = Transaction( + date=datetime.date(2026, 1, 28), amount=1461.35, + description="Payment Thank You - Web", raw_description="Payment Thank You - Web", + account_id=chase.id, + ) + session.add_all([txn1, txn2]) + session.commit() + + detector = TransferDetector(session) + pairs = detector.detect() + assert len(pairs) == 1 + assert pairs[0]["amount"] == 1461.35 + + +def test_detect_marks_transfers(): + session = make_session() + member = HouseholdMember(name="Andrew", relationship="self") + session.add(member) + session.flush() + checking = Account(name="Checking", institution="WF", account_type="checking", owner_id=member.id) + chase = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id) + session.add_all([checking, chase]) + session.flush() + + txn1 = Transaction( + date=datetime.date(2026, 1, 29), amount=-1461.35, + description="CHASE CREDIT CRD EPAY", account_id=checking.id, + ) + txn2 = Transaction( + date=datetime.date(2026, 1, 28), amount=1461.35, + description="Payment Thank You - Web", account_id=chase.id, + ) + session.add_all([txn1, txn2]) + session.commit() + + detector = TransferDetector(session) + detector.mark_transfers() + + session.refresh(txn1) + session.refresh(txn2) + assert txn1.is_transfer is True + assert txn2.is_transfer is True + + +def test_no_false_positive_different_amounts(): + session = make_session() + member = HouseholdMember(name="Andrew", relationship="self") + session.add(member) + session.flush() + checking = Account(name="Checking", institution="WF", account_type="checking", owner_id=member.id) + chase = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id) + session.add_all([checking, chase]) + session.flush() + + txn1 = Transaction( + date=datetime.date(2026, 1, 29), amount=-1461.35, + description="CHASE CREDIT CRD EPAY", account_id=checking.id, + ) + txn2 = Transaction( + date=datetime.date(2026, 1, 28), amount=500.00, + description="Payment Thank You - Web", account_id=chase.id, + ) + session.add_all([txn1, txn2]) + session.commit() + + detector = TransferDetector(session) + pairs = detector.detect() + assert len(pairs) == 0