6.6 KiB
6.6 KiB
In [ ]:
from pathlib import Path
import numpy as np
import pandas as pd
DATA_PATH = Path("data/synthetic_security_flows.csv")
RANDOM_STATE = 42
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix
In [ ]:
df = pd.read_csv(DATA_PATH).drop_duplicates()
df.columns = df.columns.str.strip()
df = df.replace([np.inf, -np.inf], np.nan)
X = df.drop(columns="Label").select_dtypes(include="number")
y = df["Label"].astype(str).str.strip().str.upper().ne("BENIGN").astype(int)
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=RANDOM_STATE, stratify=y
)
benign_count = int((y_train == 0).sum())
attack_count = int((y_train == 1).sum())
imbalance_ratio = benign_count / attack_count
print(f"BENIGN={benign_count}, ATTACK={attack_count}")
print(f"BENIGN-to-ATTACK ratio: {imbalance_ratio:.2f}:1")
In [ ]:
def train_and_measure(class_weight):
model = Pipeline([
("imputer", SimpleImputer(strategy="median")),
("model", RandomForestClassifier(
n_estimators=100,
max_depth=12,
class_weight=class_weight,
random_state=RANDOM_STATE,
n_jobs=-1,
)),
])
model.fit(X_train, y_train)
predictions = model.predict(X_test)
tn, fp, fn, tp = confusion_matrix(y_test, predictions).ravel()
return {
"precision": precision_score(y_test, predictions, zero_division=0),
"recall": recall_score(y_test, predictions, zero_division=0),
"f1": f1_score(y_test, predictions, zero_division=0),
"false_positives": int(fp),
"false_negatives": int(fn),
}
In [ ]:
baseline_metrics = train_and_measure(class_weight=None)
balanced_metrics = train_and_measure(class_weight="balanced")
comparison_df = pd.DataFrame(
[baseline_metrics, balanced_metrics],
index=["Baseline", "Balanced"],
)
comparison_df.round(3)
In [ ]:
recommended_configuration = comparison_df["f1"].idxmax()
print("Recommended configuration:", recommended_configuration)
print(comparison_df.loc[recommended_configuration].round(3))
In [ ]:
assert benign_count > attack_count
assert imbalance_ratio > 1
assert list(comparison_df.index) == ["Baseline", "Balanced"]
assert {"precision", "recall", "f1", "false_positives", "false_negatives"}.issubset(comparison_df.columns)
assert comparison_df[["precision", "recall", "f1"]].apply(lambda col: col.between(0, 1).all()).all()
assert recommended_configuration in {"Baseline", "Balanced"}
print("✓ Lab 5 completed")