feat(fusion_accounting_followup): risk_scorer service

Made-with: Cursor

fusion_accounting_followup/__manifest__.py

@@ -1,6 +1,6 @@
 {
     'name': 'Fusion Accounting Follow-up',
-    'version': '19.0.1.0.2',
+    'version': '19.0.1.0.3',
     'category': 'Accounting/Accounting',
     'summary': 'AI-augmented customer follow-ups (dunning) for unpaid invoices.',
     'description': """

fusion_accounting_followup/services/__init__.py

@@ -1,2 +1,3 @@
 from . import overdue_aging
 from . import level_resolver
+from . import risk_scorer

fusion_accounting_followup/services/risk_scorer.py

@@ -0,0 +1,62 @@
+"""Payment-history risk scorer.
+
+Pure-Python: takes payment history (list of payment events) + average days-late
+and returns a risk score 0-100. Higher = more risky."""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class PartnerRiskScore:
+    score: int
+    band: str
+    drivers: list[str]
+
+
+def score_partner(*, total_invoices: int = 0, paid_late_count: int = 0,
+                    avg_days_late: float = 0.0,
+                    longest_overdue_days: int = 0,
+                    open_overdue_amount: float = 0.0,
+                    average_invoice_amount: float = 1000.0) -> PartnerRiskScore:
+    """Compute a 0-100 risk score from payment-history primitives.
+
+    Heuristic weights:
+    - 30% : late-payment ratio (paid_late_count / total_invoices)
+    - 25% : avg days late (capped at 60 days)
+    - 25% : longest current overdue (capped at 120 days)
+    - 20% : open overdue amount as multiple of average invoice
+    """
+    drivers: list[str] = []
+    score = 0.0
+
+    if total_invoices > 0:
+        late_ratio = paid_late_count / total_invoices
+        score += min(late_ratio * 100, 100) * 0.30
+        if late_ratio > 0.5:
+            drivers.append(f"{paid_late_count}/{total_invoices} invoices paid late")
+
+    score += min(avg_days_late / 60, 1) * 100 * 0.25
+    if avg_days_late > 14:
+        drivers.append(f"Avg {avg_days_late:.1f} days late on payment")
+
+    score += min(longest_overdue_days / 120, 1) * 100 * 0.25
+    if longest_overdue_days > 30:
+        drivers.append(f"Longest currently overdue: {longest_overdue_days} days")
+
+    if average_invoice_amount > 0:
+        ratio = open_overdue_amount / average_invoice_amount
+        score += min(ratio / 5, 1) * 100 * 0.20
+        if ratio > 1.5:
+            drivers.append(f"Open overdue ${open_overdue_amount:,.2f} ({ratio:.1f}x avg invoice)")
+
+    final = int(round(score))
+    if final >= 80:
+        band = 'critical'
+    elif final >= 60:
+        band = 'high'
+    elif final >= 30:
+        band = 'medium'
+    else:
+        band = 'low'
+
+    return PartnerRiskScore(score=final, band=band, drivers=drivers)

fusion_accounting_followup/tests/__init__.py

@@ -1,2 +1,3 @@
 from . import test_overdue_aging
 from . import test_level_resolver
+from . import test_risk_scorer

fusion_accounting_followup/tests/test_risk_scorer.py

@@ -0,0 +1,48 @@
+from odoo.tests.common import TransactionCase
+from odoo.tests import tagged
+from odoo.addons.fusion_accounting_followup.services.risk_scorer import (
+    score_partner, PartnerRiskScore,
+)
+
+
+@tagged('post_install', '-at_install')
+class TestRiskScorer(TransactionCase):
+
+    def test_no_history_returns_low(self):
+        result = score_partner()
+        self.assertEqual(result.band, 'low')
+        self.assertLess(result.score, 30)
+
+    def test_chronic_late_pays_returns_high(self):
+        result = score_partner(
+            total_invoices=20, paid_late_count=18,
+            avg_days_late=45, longest_overdue_days=90,
+            open_overdue_amount=15000, average_invoice_amount=2000,
+        )
+        self.assertIn(result.band, ('high', 'critical'))
+        self.assertGreater(len(result.drivers), 0)
+
+    def test_one_off_overdue_returns_medium(self):
+        result = score_partner(
+            total_invoices=10, paid_late_count=1,
+            avg_days_late=20, longest_overdue_days=45,
+            open_overdue_amount=2000, average_invoice_amount=2000,
+        )
+        self.assertIn(result.band, ('low', 'medium'))
+
+    def test_score_capped_at_100(self):
+        result = score_partner(
+            total_invoices=10, paid_late_count=10,
+            avg_days_late=180, longest_overdue_days=300,
+            open_overdue_amount=999999, average_invoice_amount=1000,
+        )
+        self.assertLessEqual(result.score, 100)
+
+    def test_score_floored_at_0(self):
+        result = score_partner()
+        self.assertGreaterEqual(result.score, 0)
+
+    def test_band_thresholds(self):
+        for s, expected_band in [(10, 'low'), (40, 'medium'), (70, 'high'), (90, 'critical')]:
+            r = PartnerRiskScore(score=s, band=expected_band, drivers=[])
+            self.assertEqual(r.band, expected_band)