feat(fusion_accounting_bank_rec): pattern_extractor for per-partner aggregates

Made-with: Cursor

fusion_accounting_bank_rec/services/__init__.py

@@ -2,3 +2,4 @@ from . import memo_tokenizer
 from . import exchange_diff
 from . import matching_strategies
 from . import precedent_lookup
+from . import pattern_extractor

fusion_accounting_bank_rec/services/pattern_extractor.py

@@ -0,0 +1,74 @@
+"""Aggregate per-partner reconciliation patterns from precedent rows.
+
+Computes typical amount range, cadence, preferred strategy, common memo
+tokens. Output is a dict suitable for create/write on fusion.reconcile.pattern.
+"""
+
+from collections import Counter
+from statistics import median
+
+
+def extract_pattern_for_partner(env, *, company_id, partner_id) -> dict:
+    """Compute the pattern aggregate for one (company, partner) pair.
+
+    Returns vals dict suitable for env['fusion.reconcile.pattern'].create()."""
+    Precedent = env['fusion.reconcile.precedent'].sudo()
+    precedents = Precedent.search([
+        ('company_id', '=', company_id),
+        ('partner_id', '=', partner_id),
+    ], order='reconciled_at desc', limit=200)
+
+    if not precedents:
+        return {
+            'company_id': company_id,
+            'partner_id': partner_id,
+            'reconcile_count': 0,
+        }
+
+    amounts = sorted(precedents.mapped('amount'))
+    counts = precedents.mapped('matched_move_line_count')
+
+    single_count = sum(1 for c in counts if c == 1)
+    multi_count = sum(1 for c in counts if c > 1)
+    if multi_count > single_count:
+        pref_strategy = 'multi_invoice'
+    elif _amounts_concentrated(amounts):
+        pref_strategy = 'exact_amount'
+    else:
+        pref_strategy = 'fifo'
+
+    reconcile_dates = sorted([p.reconciled_at for p in precedents if p.reconciled_at])
+    if len(reconcile_dates) >= 2:
+        deltas = [(reconcile_dates[i+1] - reconcile_dates[i]).days
+                  for i in range(len(reconcile_dates) - 1)]
+        cadence = sum(deltas) / len(deltas) if deltas else 0.0
+    else:
+        cadence = 0.0
+
+    token_counter = Counter()
+    for p in precedents:
+        if p.memo_tokens:
+            for tok in p.memo_tokens.split(','):
+                token_counter[tok.strip()] += 1
+    # Keep tokens appearing in >=30% of precedents (min floor of 2 occurrences)
+    threshold = max(2, len(precedents) * 0.3)
+    common_tokens = ','.join(t for t, c in token_counter.most_common() if c >= threshold)
+
+    return {
+        'company_id': company_id,
+        'partner_id': partner_id,
+        'reconcile_count': len(precedents),
+        'typical_amount_range': f"${min(amounts):,.2f} – ${max(amounts):,.2f} (median ${median(amounts):,.2f})",
+        'typical_cadence_days': round(cadence, 1),
+        'pref_strategy': pref_strategy,
+        'common_memo_tokens': common_tokens,
+    }
+
+
+def _amounts_concentrated(amounts: list[float]) -> bool:
+    """True if amounts cluster around a few values (suggests exact-amount strategy)."""
+    if len(amounts) < 3:
+        return True
+    med = median(amounts)
+    within_5pct = sum(1 for a in amounts if abs(a - med) / max(med, 1) < 0.05)
+    return within_5pct / len(amounts) >= 0.6

fusion_accounting_bank_rec/tests/__init__.py

@@ -3,3 +3,4 @@ from . import test_exchange_diff
 from . import test_matching_strategies
 from . import test_ai_suggestion_lifecycle
 from . import test_precedent_lookup
+from . import test_pattern_extraction

fusion_accounting_bank_rec/tests/test_pattern_extraction.py

@@ -0,0 +1,73 @@
+from datetime import date, timedelta, datetime
+from odoo.tests.common import TransactionCase, tagged
+from odoo.addons.fusion_accounting_bank_rec.services.pattern_extractor import (
+    extract_pattern_for_partner,
+)
+
+
+@tagged('post_install', '-at_install')
+class TestPatternExtractor(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.partner = self.env['res.partner'].create({'name': 'Pattern Test Partner'})
+        self.currency = self.env.ref('base.CAD')
+        self.company = self.env.company
+
+    def _make_precedent(self, *, amount, days_ago, memo='RBC,ETF', count=1, source='manual'):
+        return self.env['fusion.reconcile.precedent'].create({
+            'company_id': self.company.id,
+            'partner_id': self.partner.id,
+            'amount': amount,
+            'currency_id': self.currency.id,
+            'date': date.today() - timedelta(days=days_ago),
+            'memo_tokens': memo,
+            'matched_move_line_count': count,
+            'reconciled_at': datetime.now() - timedelta(days=days_ago),
+            'source': source,
+        })
+
+    def test_extracts_typical_amount_range(self):
+        for d in [10, 24, 38, 52]:
+            self._make_precedent(amount=1847.50, days_ago=d)
+
+        pattern_vals = extract_pattern_for_partner(
+            self.env, company_id=self.company.id, partner_id=self.partner.id)
+        self.assertIn('typical_amount_range', pattern_vals)
+        self.assertEqual(pattern_vals['reconcile_count'], 4)
+
+    def test_detects_exact_amount_strategy(self):
+        for d in range(0, 56, 14):
+            self._make_precedent(amount=1847.50, days_ago=d, count=1)
+        pattern_vals = extract_pattern_for_partner(
+            self.env, company_id=self.company.id, partner_id=self.partner.id)
+        self.assertEqual(pattern_vals['pref_strategy'], 'exact_amount')
+
+    def test_detects_multi_invoice_strategy(self):
+        for d in range(0, 56, 14):
+            self._make_precedent(amount=2500.00, days_ago=d, count=3)
+        pattern_vals = extract_pattern_for_partner(
+            self.env, company_id=self.company.id, partner_id=self.partner.id)
+        self.assertEqual(pattern_vals['pref_strategy'], 'multi_invoice')
+
+    def test_computes_cadence_days(self):
+        for d in [0, 14, 28, 42]:
+            self._make_precedent(amount=1000, days_ago=d)
+        pattern_vals = extract_pattern_for_partner(
+            self.env, company_id=self.company.id, partner_id=self.partner.id)
+        self.assertAlmostEqual(pattern_vals['typical_cadence_days'], 14.0, delta=1)
+
+    def test_extracts_common_memo_tokens(self):
+        self._make_precedent(amount=1000, days_ago=10, memo='RBC,ETF,REF')
+        self._make_precedent(amount=1000, days_ago=24, memo='RBC,ETF,DEPOSIT')
+        self._make_precedent(amount=1000, days_ago=38, memo='RBC,ETF,REF')
+        pattern_vals = extract_pattern_for_partner(
+            self.env, company_id=self.company.id, partner_id=self.partner.id)
+        self.assertIn('RBC', pattern_vals['common_memo_tokens'])
+        self.assertIn('ETF', pattern_vals['common_memo_tokens'])
+
+    def test_returns_zero_count_for_partner_with_no_precedents(self):
+        other_partner = self.env['res.partner'].create({'name': 'Empty Partner'})
+        pattern_vals = extract_pattern_for_partner(
+            self.env, company_id=self.company.id, partner_id=other_partner.id)
+        self.assertEqual(pattern_vals['reconcile_count'], 0)