Python 打造智慧資安防護系統：AI 驅動的自動化解決方案實戰指南

import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# 載入惡意軟體特徵資料集
# 資料集包含靜態特徵（檔案大小、區段數量、匯入函數等）和動態特徵（系統呼叫序列、網路連線等）
def load_malware_dataset(file_path):
    """
    載入惡意軟體資料集
    
    參數:
        file_path: CSV 檔案路徑
    
    回傳:
        features: 特徵矩陣
        labels: 標籤向量（0: 良性, 1: 惡意）
    """
    data = pd.read_csv(file_path)
    
    # 分離特徵與標籤
    features = data.drop(['label', 'file_hash'], axis=1)
    labels = data['label']
    
    return features, labels

# 資料預處理與標準化
def preprocess_data(X_train, X_test):
    """
    對特徵資料進行標準化處理
    
    參數:
        X_train: 訓練集特徵
        X_test: 測試集特徵
    
    回傳:
        標準化後的訓練集與測試集
    """
    # 使用 StandardScaler 將特徵標準化為平均值 0、標準差 1
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    return X_train_scaled, X_test_scaled, scaler

# 建構隨機森林分類器
def build_random_forest_model(n_estimators=100, max_depth=20):
    """
    建立隨機森林分類模型
    
    參數:
        n_estimators: 決策樹數量
        max_depth: 每棵樹的最大深度
    
    回傳:
        訓練好的隨機森林模型
    """
    # 隨機森林是集成學習方法，透過多棵決策樹投票來提高準確率
    model = RandomForestClassifier(
        n_estimators=n_estimators,
        max_depth=max_depth,
        random_state=42,
        n_jobs=-1  # 使用所有 CPU 核心進行平行運算
    )
    
    return model

# 建構深度學習分類器
def build_deep_learning_model(input_dim):
    """
    建立深度神經網路分類模型
    
    參數:
        input_dim: 輸入特徵維度
    
    回傳:
        編譯好的 Keras 模型
    """
    model = keras.Sequential([
        # 第一層：全連接層，256 個神經元，ReLU 啟動函數
        layers.Dense(256, activation='relu', input_dim=input_dim),
        layers.Dropout(0.3),  # Dropout 層防止過度擬合
        
        # 第二層：全連接層，128 個神經元
        layers.Dense(128, activation='relu'),
        layers.Dropout(0.3),
        
        # 第三層：全連接層，64 個神經元
        layers.Dense(64, activation='relu'),
        layers.Dropout(0.2),
        
        # 輸出層：單一神經元，sigmoid 啟動函數產生機率值
        layers.Dense(1, activation='sigmoid')
    ])
    
    # 編譯模型：使用 Adam 最佳化器與二元交叉熵損失函數
    model.compile(
        optimizer='adam',
        loss='binary_crossentropy',
        metrics=['accuracy', keras.metrics.Precision(), keras.metrics.Recall()]
    )
    
    return model

# 訓練與評估流程
def train_and_evaluate():
    """
    完整的模型訓練與評估流程
    """
    # 載入資料
    print("載入惡意軟體資料集...")
    features, labels = load_malware_dataset('malware_dataset.csv')
    
    # 分割訓練集與測試集（80% 訓練，20% 測試）
    X_train, X_test, y_train, y_test = train_test_split(
        features, labels, test_size=0.2, random_state=42, stratify=labels
    )
    
    # 資料預處理
    print("進行資料標準化...")
    X_train_scaled, X_test_scaled, scaler = preprocess_data(X_train, X_test)
    
    # 訓練隨機森林模型
    print("\n訓練隨機森林模型...")
    rf_model = build_random_forest_model()
    rf_model.fit(X_train_scaled, y_train)
    
    # 評估隨機森林模型
    rf_predictions = rf_model.predict(X_test_scaled)
    print("\n隨機森林模型評估結果:")
    print(classification_report(y_test, rf_predictions, 
                                target_names=['良性', '惡意']))
    
    # 訓練深度學習模型
    print("\n訓練深度神經網路模型...")
    dl_model = build_deep_learning_model(X_train_scaled.shape[1])
    
    # 使用 Early Stopping 避免過度擬合
    early_stopping = keras.callbacks.EarlyStopping(
        monitor='val_loss',
        patience=10,
        restore_best_weights=True
    )
    
    history = dl_model.fit(
        X_train_scaled, y_train,
        validation_split=0.2,
        epochs=50,
        batch_size=32,
        callbacks=[early_stopping],
        verbose=1
    )
    
    # 評估深度學習模型
    dl_predictions = (dl_model.predict(X_test_scaled) > 0.5).astype(int)
    print("\n深度學習模型評估結果:")
    print(classification_report(y_test, dl_predictions, 
                                target_names=['良性', '惡意']))
    
    return rf_model, dl_model, scaler

# 執行訓練
if __name__ == "__main__":
    rf_model, dl_model, scaler = train_and_evaluate()

import numpy as np
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
import tensorflow as tf
from tensorflow import keras

# 建構自動編碼器模型用於異常偵測
def build_autoencoder(input_dim, encoding_dim=32):
    """
    建立自動編碼器模型進行異常偵測
    
    參數:
        input_dim: 輸入資料維度
        encoding_dim: 編碼後的壓縮維度
    
    回傳:
        autoencoder: 完整的自動編碼器模型
        encoder: 編碼器部分
    """
    # 編碼器部分：將高維度輸入壓縮到低維度表示
    encoder_input = keras.Input(shape=(input_dim,))
    encoded = layers.Dense(128, activation='relu')(encoder_input)
    encoded = layers.Dense(64, activation='relu')(encoded)
    encoded = layers.Dense(encoding_dim, activation='relu')(encoded)
    
    # 解碼器部分：將壓縮表示還原回原始維度
    decoded = layers.Dense(64, activation='relu')(encoded)
    decoded = layers.Dense(128, activation='relu')(decoded)
    decoder_output = layers.Dense(input_dim, activation='sigmoid')(decoded)
    
    # 建立完整的自動編碼器模型
    autoencoder = keras.Model(encoder_input, decoder_output)
    encoder = keras.Model(encoder_input, encoded)
    
    # 編譯模型：使用 MSE 損失函數衡量重建誤差
    autoencoder.compile(optimizer='adam', loss='mse')
    
    return autoencoder, encoder

# 使用 K-means 進行網路流量聚類分析
def cluster_network_traffic(traffic_data, n_clusters=5):
    """
    使用 K-means 將網路流量分組成不同行為模式
    
    參數:
        traffic_data: 網路流量特徵資料
        n_clusters: 群集數量
    
    回傳:
        cluster_labels: 每筆資料的群集標籤
        kmeans_model: 訓練好的 K-means 模型
    """
    # 資料標準化處理
    scaler = StandardScaler()
    traffic_scaled = scaler.fit_transform(traffic_data)
    
    # 訓練 K-means 聚類模型
    kmeans_model = KMeans(
        n_clusters=n_clusters,
        random_state=42,
        n_init=10,
        max_iter=300
    )
    cluster_labels = kmeans_model.fit_predict(traffic_scaled)
    
    return cluster_labels, kmeans_model, scaler

# 使用 PCA 降維並視覺化異常點
def detect_anomalies_with_pca(data, contamination=0.05):
    """
    使用 PCA 降維並識別異常資料點
    
    參數:
        data: 原始高維度資料
        contamination: 預期異常資料比例
    
    回傳:
        異常分數與降維後的資料
    """
    # 資料標準化
    scaler = StandardScaler()
    data_scaled = scaler.fit_transform(data)
    
    # 使用 PCA 將資料降至 2 維以便視覺化
    pca = PCA(n_components=2)
    data_pca = pca.fit_transform(data_scaled)
    
    # 計算每個資料點到群集中心的距離作為異常分數
    distances = np.sum((data_pca - np.mean(data_pca, axis=0))**2, axis=1)
    threshold = np.percentile(distances, (1 - contamination) * 100)
    
    # 標記異常點
    anomalies = distances > threshold
    
    return anomalies, data_pca, distances

# 訓練自動編碼器並偵測異常
def train_autoencoder_for_anomaly_detection(normal_data, test_data):
    """
    訓練自動編碼器並用重建誤差偵測異常
    
    參數:
        normal_data: 僅包含正常行為的訓練資料
        test_data: 包含正常與異常的測試資料
    
    回傳:
        reconstruction_errors: 重建誤差
        threshold: 異常判定閾值
    """
    # 資料標準化
    scaler = StandardScaler()
    normal_scaled = scaler.fit_transform(normal_data)
    test_scaled = scaler.transform(test_data)
    
    # 建立自動編碼器
    autoencoder, encoder = build_autoencoder(normal_scaled.shape[1])
    
    # 僅使用正常資料訓練
    history = autoencoder.fit(
        normal_scaled, normal_scaled,
        epochs=50,
        batch_size=32,
        validation_split=0.2,
        verbose=1
    )
    
    # 計算測試集的重建誤差
    reconstructed = autoencoder.predict(test_scaled)
    reconstruction_errors = np.mean(np.square(test_scaled - reconstructed), axis=1)
    
    # 使用訓練集的重建誤差分佈來設定閾值
    train_reconstructed = autoencoder.predict(normal_scaled)
    train_errors = np.mean(np.square(normal_scaled - train_reconstructed), axis=1)
    threshold = np.percentile(train_errors, 95)  # 使用 95 百分位數作為閾值
    
    # 識別異常資料點
    anomalies = reconstruction_errors > threshold
    
    print(f"偵測到 {np.sum(anomalies)} 個異常資料點，佔測試集的 {np.sum(anomalies)/len(test_data)*100:.2f}%")
    
    return reconstruction_errors, threshold, anomalies

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# PyTorch 實作：自訂威脅偵測神經網路
class ThreatDetectionNet(nn.Module):
    """
    使用 PyTorch 實作的威脅偵測神經網路
    採用多層感知機架構，適用於結構化的網路流量特徵
    """
    def __init__(self, input_size, hidden_sizes, num_classes):
        """
        初始化網路架構
        
        參數:
            input_size: 輸入特徵維度
            hidden_sizes: 隱藏層神經元數量列表
            num_classes: 輸出類別數量
        """
        super(ThreatDetectionNet, self).__init__()
        
        # 動態建立多層全連接網路
        layers_list = []
        prev_size = input_size
        
        for hidden_size in hidden_sizes:
            layers_list.append(nn.Linear(prev_size, hidden_size))
            layers_list.append(nn.ReLU())
            layers_list.append(nn.Dropout(0.3))  # 防止過度擬合
            prev_size = hidden_size
        
        # 輸出層
        layers_list.append(nn.Linear(prev_size, num_classes))
        
        # 將所有層組合成序列模型
        self.model = nn.Sequential(*layers_list)
    
    def forward(self, x):
        """
        前向傳播計算
        
        參數:
            x: 輸入張量
        
        回傳:
            輸出張量
        """
        return self.model(x)

# PyTorch 訓練流程
def train_pytorch_model(model, train_loader, val_loader, epochs=50):
    """
    完整的 PyTorch 模型訓練流程
    
    參數:
        model: PyTorch 模型
        train_loader: 訓練資料載入器
        val_loader: 驗證資料載入器
        epochs: 訓練週期數
    """
    # 設定損失函數與最佳化器
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    
    # 學習率調度器：當驗證損失不再下降時降低學習率
    scheduler = optim.lr_scheduler.ReduceLROnPlateau(
        optimizer, mode='min', factor=0.5, patience=5
    )
    
    # 記錄最佳模型
    best_val_loss = float('inf')
    
    for epoch in range(epochs):
        # 訓練階段
        model.train()
        train_loss = 0.0
        train_correct = 0
        train_total = 0
        
        for inputs, labels in train_loader:
            # 清除梯度
            optimizer.zero_grad()
            
            # 前向傳播
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            
            # 反向傳播與參數更新
            loss.backward()
            optimizer.step()
            
            # 統計訓練指標
            train_loss += loss.item()
            _, predicted = torch.max(outputs.data, 1)
            train_total += labels.size(0)
            train_correct += (predicted == labels).sum().item()
        
        # 驗證階段
        model.eval()
        val_loss = 0.0
        val_correct = 0
        val_total = 0
        
        with torch.no_grad():  # 驗證時不需要計算梯度
            for inputs, labels in val_loader:
                outputs = model(inputs)
                loss = criterion(outputs, labels)
                
                val_loss += loss.item()
                _, predicted = torch.max(outputs.data, 1)
                val_total += labels.size(0)
                val_correct += (predicted == labels).sum().item()
        
        # 計算平均損失與準確率
        avg_train_loss = train_loss / len(train_loader)
        train_accuracy = 100 * train_correct / train_total
        avg_val_loss = val_loss / len(val_loader)
        val_accuracy = 100 * val_correct / val_total
        
        # 調整學習率
        scheduler.step(avg_val_loss)
        
        # 儲存最佳模型
        if avg_val_loss < best_val_loss:
            best_val_loss = avg_val_loss
            torch.save(model.state_dict(), 'best_model.pth')
        
        print(f'Epoch [{epoch+1}/{epochs}]')
        print(f'Train Loss: {avg_train_loss:.4f}, Train Acc: {train_accuracy:.2f}%')
        print(f'Val Loss: {avg_val_loss:.4f}, Val Acc: {val_accuracy:.2f}%')

# TensorFlow 實作：使用 Keras API 建構同等架構
def build_tensorflow_model(input_size, hidden_sizes, num_classes):
    """
    使用 TensorFlow Keras API 建構威脅偵測模型
    
    參數:
        input_size: 輸入特徵維度
        hidden_sizes: 隱藏層神經元數量列表
        num_classes: 輸出類別數量
    
    回傳:
        編譯好的 Keras 模型
    """
    model = keras.Sequential()
    
    # 輸入層
    model.add(layers.Input(shape=(input_size,)))
    
    # 動態新增隱藏層
    for hidden_size in hidden_sizes:
        model.add(layers.Dense(hidden_size, activation='relu'))
        model.add(layers.Dropout(0.3))
    
    # 輸出層：使用 softmax 進行多類別分類
    model.add(layers.Dense(num_classes, activation='softmax'))
    
    # 編譯模型
    model.compile(
        optimizer='adam',
        loss='sparse_categorical_crossentropy',
        metrics=['accuracy']
    )
    
    return model

# TensorFlow 訓練流程
def train_tensorflow_model(model, X_train, y_train, X_val, y_val, epochs=50):
    """
    完整的 TensorFlow 模型訓練流程
    
    參數:
        model: TensorFlow 模型
        X_train, y_train: 訓練資料與標籤
        X_val, y_val: 驗證資料與標籤
        epochs: 訓練週期數
    """
    # 設定回呼函數
    callbacks = [
        # Early Stopping：驗證損失不再改善時提前停止
        keras.callbacks.EarlyStopping(
            monitor='val_loss',
            patience=10,
            restore_best_weights=True
        ),
        # 學習率調度：驗證損失不再改善時降低學習率
        keras.callbacks.ReduceLROnPlateau(
            monitor='val_loss',
            factor=0.5,
            patience=5,
            min_lr=1e-7
        ),
        # 模型檢查點：儲存最佳模型
        keras.callbacks.ModelCheckpoint(
            'best_model.h5',
            monitor='val_loss',
            save_best_only=True
        )
    ]
    
    # 訓練模型
    history = model.fit(
        X_train, y_train,
        validation_data=(X_val, y_val),
        epochs=epochs,
        batch_size=32,
        callbacks=callbacks,
        verbose=1
    )
    
    return history

import shap
import numpy as np
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestClassifier

# 使用 SHAP 解釋隨機森林模型的預測
def explain_model_predictions(model, X_train, X_test, feature_names):
    """
    使用 SHAP 值解釋模型預測結果
    
    參數:
        model: 訓練好的模型
        X_train: 訓練集資料（用於建立解釋器）
        X_test: 要解釋的測試樣本
        feature_names: 特徵名稱列表
    """
    # 建立 SHAP 解釋器
    # TreeExplainer 針對樹型模型最佳化，計算速度快
    explainer = shap.TreeExplainer(model)
    
    # 計算測試樣本的 SHAP 值
    shap_values = explainer.shap_values(X_test)
    
    # 如果是二元分類，選擇正類別的 SHAP 值
    if isinstance(shap_values, list):
        shap_values = shap_values[1]
    
    # 繪製摘要圖：顯示所有特徵的重要性分佈
    plt.figure(figsize=(10, 8))
    shap.summary_plot(shap_values, X_test, feature_names=feature_names, show=False)
    plt.title('特徵重要性摘要圖', fontsize=16)
    plt.tight_layout()
    plt.savefig('shap_summary.png', dpi=300, bbox_inches='tight')
    plt.close()
    
    # 繪製單一樣本的力圖：顯示各特徵如何影響該樣本的預測
    plt.figure(figsize=(12, 4))
    shap.force_plot(
        explainer.expected_value[1] if isinstance(explainer.expected_value, list) else explainer.expected_value,
        shap_values[0],
        X_test[0],
        feature_names=feature_names,
        matplotlib=True,
        show=False
    )
    plt.title('單一樣本預測解釋', fontsize=16)
    plt.tight_layout()
    plt.savefig('shap_force_plot.png', dpi=300, bbox_inches='tight')
    plt.close()
    
    # 計算全域特徵重要性
    feature_importance = np.abs(shap_values).mean(axis=0)
    importance_df = pd.DataFrame({
        '特徵': feature_names,
        '重要性': feature_importance
    }).sort_values('重要性', ascending=False)
    
    print("\n前 10 個最重要特徵:")
    print(importance_df.head(10))
    
    return shap_values, importance_df

# 從決策樹中擷取可讀規則
def extract_decision_rules(tree_model, feature_names, class_names):
    """
    從決策樹模型中擷取人類可讀的決策規則
    
    參數:
        tree_model: 訓練好的決策樹模型
        feature_names: 特徵名稱列表
        class_names: 類別名稱列表
    
    回傳:
        規則列表
    """
    from sklearn.tree import _tree
    
    tree = tree_model.tree_
    rules = []
    
    def recurse(node, depth, condition_list):
        """
        遞迴走訪決策樹並擷取規則
        """
        indent = "  " * depth
        
        if tree.feature[node] != _tree.TREE_UNDEFINED:
            # 當前節點是決策節點
            feature = feature_names[tree.feature[node]]
            threshold = tree.threshold[node]
            
            # 走訪左子樹（特徵值 <= 閾值）
            left_condition = f"{feature} <= {threshold:.4f}"
            recurse(
                tree.children_left[node],
                depth + 1,
                condition_list + [left_condition]
            )
            
            # 走訪右子樹（特徵值 > 閾值）
            right_condition = f"{feature} > {threshold:.4f}"
            recurse(
                tree.children_right[node],
                depth + 1,
                condition_list + [right_condition]
            )
        else:
            # 當前節點是葉節點
            class_idx = np.argmax(tree.value[node])
            class_name = class_names[class_idx]
            samples = tree.n_node_samples[node]
            
            # 建立規則字串
            if condition_list:
                rule = f"如果 {' 且 '.join(condition_list)}, 則預測為 {class_name} (樣本數: {samples})"
            else:
                rule = f"預測為 {class_name} (樣本數: {samples})"
            
            rules.append(rule)
    
    # 從根節點開始走訪
    recurse(0, 0, [])
    
    return rules

# 視覺化決策路徑
def visualize_decision_path(model, X_sample, feature_names):
    """
    視覺化單一樣本在決策樹中的路徑
    
    參數:
        model: 訓練好的決策樹模型
        X_sample: 單一樣本
        feature_names: 特徵名稱列表
    """
    from sklearn.tree import plot_tree
    
    # 取得決策路徑
    decision_path = model.decision_path(X_sample.reshape(1, -1))
    node_indicator = decision_path.toarray()[0]
    
    # 繪製決策樹並標示路徑
    plt.figure(figsize=(20, 10))
    plot_tree(
        model,
        feature_names=feature_names,
        class_names=['正常', '異常'],
        filled=True,
        fontsize=10
    )
    plt.title('決策樹結構與預測路徑', fontsize=16)
    plt.tight_layout()
    plt.savefig('decision_tree_path.png', dpi=300, bbox_inches='tight')
    plt.close()
    
    # 輸出決策路徑資訊
    print("\n決策路徑:")
    for node_id in np.where(node_indicator)[0]:
        if model.tree_.feature[node_id] != _tree.TREE_UNDEFINED:
            feature = feature_names[model.tree_.feature[node_id]]
            threshold = model.tree_.threshold[node_id]
            value = X_sample[model.tree_.feature[node_id]]
            
            if value <= threshold:
                direction = "<="
            else:
                direction = ">"
            
            print(f"節點 {node_id}: {feature} {direction} {threshold:.4f} (實際值: {value:.4f})")