| import joblib | |
| import json | |
| import numpy as np | |
| # Load model | |
| model = joblib.load('isolation_forest.pkl') | |
| # Load feature names | |
| with open('features.json', 'r') as f: | |
| feature_names = json.load(f) | |
| def predict(inputs): | |
| """Run inference on input data.""" | |
| # Handle single input or batch | |
| if isinstance(inputs, dict): | |
| inputs = [inputs] | |
| # Extract features in correct order | |
| features = [] | |
| for input_dict in inputs: | |
| feature_vector = [input_dict.get(feat, 0) for feat in feature_names] | |
| features.append(feature_vector) | |
| # Convert to numpy array | |
| X = np.array(features) | |
| # Get anomaly scores | |
| scores = model.decision_function(X) | |
| # Normalize to 0-1 scale (higher = more anomalous) | |
| normalized_scores = (0.5 - scores) / 1.0 | |
| normalized_scores = np.clip(normalized_scores, 0, 1) | |
| # Return as list | |
| return normalized_scores.tolist() | |
| # For Hugging Face Inference API | |
| def handler(event, context): | |
| """Handler for Hugging Face Inference API.""" | |
| inputs = event.get('inputs', event) | |
| scores = predict(inputs) | |
| return {"score": scores[0] if len(scores) == 1 else scores} | |