97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
"""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",
|
|
)
|