feat(fusion_accounting_assets): asset anomaly detection service

Made-with: Cursor

fusion_accounting_assets/services/__init__.py

@@ -1,3 +1,4 @@
 from . import depreciation_methods
 from . import prorate
 from . import salvage_value
+from . import anomaly_detection

fusion_accounting_assets/services/anomaly_detection.py

@@ -0,0 +1,96 @@
+"""Asset utilization anomaly detection.
+
+Flags assets where actual usage / posted depreciation deviates significantly
+from the expected schedule. Three signal types:
+- behind_schedule: actual depreciation < expected by > threshold pct
+- ahead_of_schedule: actual > expected (over-depreciated; scrap or recompute)
+- low_utilization: units_used < expected_units_per_period (waste alert)
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class AssetAnomaly:
+    asset_id: int
+    asset_name: str
+    anomaly_type: str
+    severity: str
+    expected: float
+    actual: float
+    variance_pct: float
+    detail: str
+
+    def to_dict(self):
+        return {
+            'asset_id': self.asset_id,
+            'asset_name': self.asset_name,
+            'anomaly_type': self.anomaly_type,
+            'severity': self.severity,
+            'expected': self.expected,
+            'actual': self.actual,
+            'variance_pct': self.variance_pct,
+            'detail': self.detail,
+        }
+
+
+DEFAULT_LOW_THRESHOLD_PCT = 10.0
+DEFAULT_MEDIUM_THRESHOLD_PCT = 25.0
+DEFAULT_HIGH_THRESHOLD_PCT = 50.0
+
+
+def detect_schedule_variance(*, asset_id: int, asset_name: str,
+                             expected_accumulated: float,
+                             actual_accumulated: float) -> AssetAnomaly | None:
+    """Compare expected accumulated depreciation vs actual posted."""
+    if expected_accumulated <= 0:
+        return None
+    variance_amt = actual_accumulated - expected_accumulated
+    variance_pct = abs(variance_amt) / expected_accumulated * 100
+    if variance_pct < DEFAULT_LOW_THRESHOLD_PCT:
+        return None
+    direction = 'ahead_of_schedule' if variance_amt > 0 else 'behind_schedule'
+    if variance_pct >= DEFAULT_HIGH_THRESHOLD_PCT:
+        severity = 'high'
+    elif variance_pct >= DEFAULT_MEDIUM_THRESHOLD_PCT:
+        severity = 'medium'
+    else:
+        severity = 'low'
+    detail = f"Posted ${actual_accumulated:,.2f} vs expected ${expected_accumulated:,.2f}"
+    return AssetAnomaly(
+        asset_id=asset_id,
+        asset_name=asset_name,
+        anomaly_type=direction,
+        severity=severity,
+        expected=expected_accumulated,
+        actual=actual_accumulated,
+        variance_pct=round(variance_pct, 1),
+        detail=detail,
+    )
+
+
+def detect_low_utilization(*, asset_id: int, asset_name: str,
+                           expected_units: float,
+                           actual_units: float) -> AssetAnomaly | None:
+    """For units-of-production assets: flag low actual usage."""
+    if expected_units <= 0:
+        return None
+    if actual_units >= expected_units * 0.9:
+        return None
+    deficit_pct = (expected_units - actual_units) / expected_units * 100
+    if deficit_pct >= 50:
+        severity = 'high'
+    elif deficit_pct >= 25:
+        severity = 'medium'
+    else:
+        severity = 'low'
+    return AssetAnomaly(
+        asset_id=asset_id,
+        asset_name=asset_name,
+        anomaly_type='low_utilization',
+        severity=severity,
+        expected=expected_units,
+        actual=actual_units,
+        variance_pct=round(deficit_pct, 1),
+        detail=f"Used {actual_units:.0f} of expected {expected_units:.0f} units",
+    )