File size: 3,942 Bytes
821cc4b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | #!/usr/bin/env python3
"""
GENESIS Risk ML Engine v10.1
Predictive risk scoring based on infrastructure metrics.
Features:
- Multi-feature risk prediction (CPU, Memory, Network, Disk, Error Rate)
- Gradient Boosted model for non-linear risk patterns
- Confidence intervals for regulatory reporting
- JSON output for audit trail integration
"""
import json
import sys
from datetime import datetime, timezone
try:
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import cross_val_score
except ImportError:
print("Installing dependencies...")
import subprocess
subprocess.check_call([sys.executable, "-m", "pip", "install",
"numpy", "scikit-learn", "pandas"])
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import cross_val_score
TRAINING_DATA = {
"cpu": [20, 30, 40, 50, 55, 60, 65, 70, 75, 80, 85, 90, 92, 95, 98],
"memory": [15, 25, 30, 40, 45, 50, 55, 60, 65, 70, 75, 80, 85, 90, 95],
"network_io": [10, 15, 20, 25, 30, 35, 40, 45, 50, 60, 65, 70, 75, 80, 90],
"disk_usage": [20, 25, 30, 35, 40, 45, 50, 55, 60, 65, 70, 75, 80, 85, 92],
"error_rate": [ 0, 1, 2, 3, 4, 5, 7, 9, 12, 15, 20, 30, 40, 60, 80],
"risk_score": [ 5, 8, 12, 18, 22, 28, 35, 42, 52, 65, 72, 82, 88, 94, 99],
}
X = np.array([
TRAINING_DATA["cpu"],
TRAINING_DATA["memory"],
TRAINING_DATA["network_io"],
TRAINING_DATA["disk_usage"],
TRAINING_DATA["error_rate"],
]).T
y = np.array(TRAINING_DATA["risk_score"])
model = GradientBoostingRegressor(
n_estimators=100,
max_depth=4,
learning_rate=0.1,
random_state=42
)
model.fit(X, y)
cv_scores = cross_val_score(model, X, y, cv=3, scoring="r2")
current_metrics = {
"cpu": float(sys.argv[1]) if len(sys.argv) > 1 else 75.0,
"memory": float(sys.argv[2]) if len(sys.argv) > 2 else 65.0,
"network_io": float(sys.argv[3]) if len(sys.argv) > 3 else 50.0,
"disk_usage": float(sys.argv[4]) if len(sys.argv) > 4 else 60.0,
"error_rate": float(sys.argv[5]) if len(sys.argv) > 5 else 12.0,
}
input_vector = np.array([[
current_metrics["cpu"],
current_metrics["memory"],
current_metrics["network_io"],
current_metrics["disk_usage"],
current_metrics["error_rate"],
]])
risk_score = float(np.clip(model.predict(input_vector)[0], 0, 100))
feature_importance = dict(zip(
["cpu", "memory", "network_io", "disk_usage", "error_rate"],
[round(float(x), 4) for x in model.feature_importances_]
))
if risk_score >= 80:
risk_level = "CRITICAL"
elif risk_score >= 60:
risk_level = "HIGH"
elif risk_score >= 40:
risk_level = "MEDIUM"
elif risk_score >= 20:
risk_level = "LOW"
else:
risk_level = "MINIMAL"
result = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"version": "10.1.0",
"engine": "GradientBoostingRegressor",
"input_metrics": current_metrics,
"risk_score": round(risk_score, 2),
"risk_level": risk_level,
"confidence": {
"cv_r2_mean": round(float(cv_scores.mean()), 4),
"cv_r2_std": round(float(cv_scores.std()), 4),
},
"feature_importance": feature_importance,
"model_params": {
"n_estimators": 100,
"max_depth": 4,
"training_samples": len(y),
},
"recommendation": (
"IMMEDIATE ACTION REQUIRED" if risk_level == "CRITICAL" else
"Escalate to operations team" if risk_level == "HIGH" else
"Monitor closely" if risk_level == "MEDIUM" else
"Normal operations"
),
}
with open("risk_score.json", "w") as f:
json.dump(result, f, indent=2)
with open("risk_score.txt", "w") as f:
f.write(str(round(risk_score)))
print(json.dumps(result, indent=2))
|