diff --git a/fusion_accounting_assets/__manifest__.py b/fusion_accounting_assets/__manifest__.py index 7df0c866..6bb31a7e 100644 --- a/fusion_accounting_assets/__manifest__.py +++ b/fusion_accounting_assets/__manifest__.py @@ -1,6 +1,6 @@ { 'name': 'Fusion Accounting Assets', - 'version': '19.0.1.0.3', + 'version': '19.0.1.0.4', 'category': 'Accounting/Accounting', 'summary': 'AI-augmented asset management with depreciation schedules.', 'description': """ diff --git a/fusion_accounting_assets/services/__init__.py b/fusion_accounting_assets/services/__init__.py index 7be9de01..52bf6ea0 100644 --- a/fusion_accounting_assets/services/__init__.py +++ b/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 diff --git a/fusion_accounting_assets/services/anomaly_detection.py b/fusion_accounting_assets/services/anomaly_detection.py new file mode 100644 index 00000000..48ae85d9 --- /dev/null +++ b/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", + ) diff --git a/fusion_accounting_assets/tests/__init__.py b/fusion_accounting_assets/tests/__init__.py index a070ccd1..c3630e6a 100644 --- a/fusion_accounting_assets/tests/__init__.py +++ b/fusion_accounting_assets/tests/__init__.py @@ -1,3 +1,4 @@ from . import test_depreciation_methods from . import test_prorate from . import test_salvage_value +from . import test_asset_anomaly_detection diff --git a/fusion_accounting_assets/tests/test_asset_anomaly_detection.py b/fusion_accounting_assets/tests/test_asset_anomaly_detection.py new file mode 100644 index 00000000..20ecd2c9 --- /dev/null +++ b/fusion_accounting_assets/tests/test_asset_anomaly_detection.py @@ -0,0 +1,71 @@ +from odoo.tests.common import TransactionCase +from odoo.tests import tagged +from odoo.addons.fusion_accounting_assets.services.anomaly_detection import ( + detect_schedule_variance, detect_low_utilization, AssetAnomaly, +) + + +@tagged('post_install', '-at_install') +class TestAssetAnomalyDetection(TransactionCase): + + def test_schedule_variance_within_threshold_returns_none(self): + # 5% variance < 10% threshold + result = detect_schedule_variance( + asset_id=1, asset_name='Truck', expected_accumulated=10000, + actual_accumulated=10500, + ) + self.assertIsNone(result) + + def test_schedule_variance_behind_schedule_low_severity(self): + # 15% behind: low severity, behind_schedule + result = detect_schedule_variance( + asset_id=1, asset_name='Truck', expected_accumulated=10000, + actual_accumulated=8500, + ) + self.assertIsNotNone(result) + self.assertEqual(result.anomaly_type, 'behind_schedule') + self.assertEqual(result.severity, 'low') + + def test_schedule_variance_ahead_high_severity(self): + # 60% ahead: high severity + result = detect_schedule_variance( + asset_id=2, asset_name='Server', expected_accumulated=10000, + actual_accumulated=16000, + ) + self.assertIsNotNone(result) + self.assertEqual(result.anomaly_type, 'ahead_of_schedule') + self.assertEqual(result.severity, 'high') + + def test_schedule_variance_zero_expected_returns_none(self): + result = detect_schedule_variance( + asset_id=1, asset_name='Truck', expected_accumulated=0, + actual_accumulated=500, + ) + self.assertIsNone(result) + + def test_low_utilization_flags_when_underused(self): + # 60% deficit -> high severity + result = detect_low_utilization( + asset_id=3, asset_name='Mill', expected_units=1000, actual_units=400, + ) + self.assertIsNotNone(result) + self.assertEqual(result.anomaly_type, 'low_utilization') + self.assertEqual(result.severity, 'high') + + def test_low_utilization_within_tolerance_returns_none(self): + # 95% used: within 10% tolerance + result = detect_low_utilization( + asset_id=3, asset_name='Mill', expected_units=1000, actual_units=950, + ) + self.assertIsNone(result) + + def test_anomaly_to_dict_round_trip(self): + anomaly = AssetAnomaly( + asset_id=1, asset_name='X', anomaly_type='behind_schedule', + severity='medium', expected=100.0, actual=70.0, variance_pct=30.0, + detail='example', + ) + d = anomaly.to_dict() + self.assertEqual(d['asset_id'], 1) + self.assertEqual(d['anomaly_type'], 'behind_schedule') + self.assertEqual(d['severity'], 'medium')