feat(fusion_accounting_reports): anomaly_detection service

Made-with: Cursor

fusion_accounting_reports/__manifest__.py

@@ -1,6 +1,6 @@
 {
     'name': 'Fusion Accounting Reports',
-    'version': '19.0.1.0.8',
+    'version': '19.0.1.0.9',
     'category': 'Accounting/Accounting',
     'summary': 'AI-augmented financial reports (P&L, balance sheet, trial balance, GL).',
     'description': """

fusion_accounting_reports/services/__init__.py

@@ -4,3 +4,4 @@ from . import totaling
 from . import currency_conversion
 from . import line_resolver
 from . import drill_down_resolver
+from . import anomaly_detection

fusion_accounting_reports/services/anomaly_detection.py

@@ -0,0 +1,81 @@
+"""Anomaly detection for financial reports.
+
+Compares each row's current-period amount to its comparison-period
+amount and flags variances exceeding a threshold. Uses both:
+- Absolute threshold ($X minimum movement)
+- Percentage threshold (Y% min variance)
+
+Pure-Python: callers pass the engine's compute_*() result; we return
+a list of anomaly dicts."""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class Anomaly:
+    row_id: str
+    label: str
+    current_amount: float
+    comparison_amount: float
+    variance_amount: float
+    variance_pct: float
+    severity: str  # 'low', 'medium', 'high'
+    direction: str  # 'increase', 'decrease'
+
+    def to_dict(self):
+        return {
+            'row_id': self.row_id, 'label': self.label,
+            'current_amount': self.current_amount,
+            'comparison_amount': self.comparison_amount,
+            'variance_amount': self.variance_amount,
+            'variance_pct': self.variance_pct,
+            'severity': self.severity, 'direction': self.direction,
+        }
+
+
+# Defaults -- tunable per company via ir.config_parameter
+DEFAULT_MIN_ABSOLUTE_THRESHOLD = 100.0
+DEFAULT_MIN_PCT_THRESHOLD = 10.0  # 10%
+DEFAULT_HIGH_PCT_THRESHOLD = 50.0  # 50%+ flagged 'high'
+
+
+def detect(report_result: dict, *, min_absolute: float = None,
+           min_pct: float = None, high_pct: float = None) -> list[dict]:
+    """Detect anomalies in a report_result dict (engine output).
+
+    Returns list of anomaly dicts ordered by severity desc, variance_amount desc.
+    Returns empty list if no comparison period was computed."""
+    if not report_result.get('comparison_period'):
+        return []
+    min_absolute = min_absolute if min_absolute is not None else DEFAULT_MIN_ABSOLUTE_THRESHOLD
+    min_pct = min_pct if min_pct is not None else DEFAULT_MIN_PCT_THRESHOLD
+    high_pct = high_pct if high_pct is not None else DEFAULT_HIGH_PCT_THRESHOLD
+
+    anomalies = []
+    for row in report_result.get('rows', []):
+        comparison = row.get('amount_comparison')
+        current = row.get('amount', 0.0)
+        if comparison is None:
+            continue
+        variance_amount = current - comparison
+        variance_pct = abs(row.get('variance_pct') or 0.0)
+        if abs(variance_amount) < min_absolute:
+            continue
+        if variance_pct < min_pct:
+            continue
+        severity = 'high' if variance_pct >= high_pct else 'medium' if variance_pct >= min_pct * 2 else 'low'
+        direction = 'increase' if variance_amount > 0 else 'decrease'
+        anomalies.append(Anomaly(
+            row_id=row['id'],
+            label=row.get('label', ''),
+            current_amount=current,
+            comparison_amount=comparison,
+            variance_amount=variance_amount,
+            variance_pct=variance_pct,
+            severity=severity,
+            direction=direction,
+        ).to_dict())
+
+    severity_order = {'high': 0, 'medium': 1, 'low': 2}
+    anomalies.sort(key=lambda a: (severity_order[a['severity']], -abs(a['variance_amount'])))
+    return anomalies

fusion_accounting_reports/tests/__init__.py

@@ -5,3 +5,4 @@ from . import test_line_resolver
 from . import test_drill_down_resolver
 from . import test_fusion_report_engine
 from . import test_seeded_reports
+from . import test_anomaly_detection

fusion_accounting_reports/tests/test_anomaly_detection.py

@@ -0,0 +1,74 @@
+"""Unit tests for anomaly_detection service."""
+
+from odoo.tests.common import TransactionCase, tagged
+from odoo.addons.fusion_accounting_reports.services.anomaly_detection import detect
+
+
+@tagged('post_install', '-at_install')
+class TestAnomalyDetection(TransactionCase):
+
+    def test_returns_empty_when_no_comparison(self):
+        report_result = {
+            'rows': [{'id': 'r1', 'label': 'Test', 'amount': 100,
+                      'amount_comparison': None, 'variance_pct': None}],
+            'comparison_period': None,
+        }
+        self.assertEqual(detect(report_result), [])
+
+    def test_flags_significant_increase(self):
+        report_result = {
+            'rows': [{'id': 'r1', 'label': 'Revenue',
+                      'amount': 12000, 'amount_comparison': 10000,
+                      'variance_pct': 20.0}],
+            'comparison_period': {'date_from': '2025-01-01'},
+        }
+        anomalies = detect(report_result)
+        self.assertEqual(len(anomalies), 1)
+        self.assertEqual(anomalies[0]['direction'], 'increase')
+        self.assertEqual(anomalies[0]['variance_amount'], 2000)
+
+    def test_skips_below_absolute_threshold(self):
+        report_result = {
+            'rows': [{'id': 'r1', 'label': 'Tiny', 'amount': 50,
+                      'amount_comparison': 30, 'variance_pct': 67}],
+            'comparison_period': {'date_from': '2025-01-01'},
+        }
+        # variance is $20 < default $100 minimum
+        self.assertEqual(detect(report_result), [])
+
+    def test_skips_below_pct_threshold(self):
+        report_result = {
+            'rows': [{'id': 'r1', 'label': 'Steady',
+                      'amount': 10500, 'amount_comparison': 10000,
+                      'variance_pct': 5.0}],
+            'comparison_period': {'date_from': '2025-01-01'},
+        }
+        # 5% < default 10%
+        self.assertEqual(detect(report_result), [])
+
+    def test_severity_high_for_50pct_plus(self):
+        report_result = {
+            'rows': [{'id': 'r1', 'label': 'Spike',
+                      'amount': 16000, 'amount_comparison': 10000,
+                      'variance_pct': 60.0}],
+            'comparison_period': {'date_from': '2025-01-01'},
+        }
+        anomalies = detect(report_result)
+        self.assertEqual(anomalies[0]['severity'], 'high')
+
+    def test_orders_by_severity_then_amount(self):
+        report_result = {
+            'rows': [
+                {'id': 'r1', 'label': 'Med', 'amount': 1300,
+                 'amount_comparison': 1000, 'variance_pct': 30.0},
+                {'id': 'r2', 'label': 'High', 'amount': 16000,
+                 'amount_comparison': 10000, 'variance_pct': 60.0},
+                {'id': 'r3', 'label': 'Low', 'amount': 1150,
+                 'amount_comparison': 1000, 'variance_pct': 15.0},
+            ],
+            'comparison_period': {'date_from': '2025-01-01'},
+        }
+        anomalies = detect(report_result)
+        # Should be: High first, then Med, then Low
+        self.assertEqual(anomalies[0]['severity'], 'high')
+        self.assertEqual(anomalies[-1]['severity'], 'low')