Spaces:
Paused
Paused
File size: 3,017 Bytes
5a3b9db | 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 | import pandas as pd
import logging
from typing import Dict, Any
logger = logging.getLogger(__name__)
class RiskFusionEngine:
"""
Module 1 & 2: Enterprise Risk Fusion Engine and Risk Contribution Breakdown.
Consumes Detection Engine anomaly scores and aggregates them into a normalized
Enterprise Risk Score [0-100]. Also categorizes risk into severity levels.
"""
def __init__(self, config: Dict[str, Any]):
self.weights = config.get("risk_weights", {
"rule_contribution": 0.30,
"stat_contribution": 0.10,
"ml_contribution": 0.40,
"behavioral_contribution": 0.10,
"device_contribution": 0.10
})
self.version = "1.0.0"
def calculate_risk(self, df: pd.DataFrame) -> pd.DataFrame:
logger.info("Executing Enterprise Risk Fusion...")
# 1. Extract raw scores, defaulting to 0 if not present in upstream results
# Assuming rule_score, if_score, etc. might exist in detection df
rule_score = df.get("rule_score", pd.Series(0, index=df.index))
stat_score = df.get("stat_score", pd.Series(0, index=df.index))
if_score = df.get("if_score", pd.Series(0, index=df.index))
behavior_score = df.get("behavior_score", pd.Series(0, index=df.index))
device_score = df.get("device_score", pd.Series(0, index=df.index))
# 2. Normalize and apply weights
rule_contrib = rule_score * self.weights.get("rule_contribution", 0.3)
stat_contrib = stat_score * self.weights.get("stat_contribution", 0.1)
ml_contrib = if_score * self.weights.get("ml_contribution", 0.4)
behavior_contrib = behavior_score * self.weights.get("behavioral_contribution", 0.1)
device_contrib = device_score * self.weights.get("device_contribution", 0.1)
# 3. Sum Risk Score and Normalize to 0-100
raw_risk_score = (rule_contrib + stat_contrib + ml_contrib +
behavior_contrib + device_contrib)
final_risk_score = raw_risk_score.clip(lower=0, upper=100)
# 4. Determine Risk Levels
def assign_level(score):
if score <= 15: return "Low"
elif score <= 25: return "Moderate"
elif score <= 30: return "High"
else: return "Critical"
risk_levels = final_risk_score.apply(assign_level)
# 5. Output Preserving Intermediate Contributions
return pd.DataFrame({
"event_id": df["event_id"],
"rule_contribution": rule_contrib,
"stat_contribution": stat_contrib,
"ml_contribution": ml_contrib,
"behavioral_contribution": behavior_contrib,
"device_contribution": device_contrib,
"risk_score": final_risk_score,
"risk_level": risk_levels,
"risk_engine_version": self.version,
"triggered_rules": df.get("triggered_rules", "")
})
|