OHFFDRL

Optimized Hierarchical Fused Fuzzy Deep Reinforcement Learning.

Example

import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

import OHFFDRL as oh

# === Step 1: Load and preprocess data ===
df = pd.read_csv("MAINDiagnostics.csv")
df = df.drop(columns=["IDFILENAME", "FileName", "Beat"])
df["Gender"] = df["Gender"].map({"MALE": 1, "FEMALE": 0})

normal_group = ["SR", "SB", "ST", "SI", "SAAWR"]
arrhythmia_group = ["AFIB", "AF", "SVT", "AT", "AVNRT", "AVRT"]
df["Rhythm_Binary"] = df["Rhythm"].apply(lambda x: 0 if x in normal_group else (1 if x in arrhythmia_group else np.nan))
df = df.dropna(subset=["Rhythm_Binary"])

X = df.drop(columns=["Rhythm", "Rhythm_Binary"], errors='ignore').values
y = df["Rhythm_Binary"].astype(int).values

scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# === SSO-inspired oversampling for class 1 ===
X_min = X_scaled[y == 1]
X_maj = X_scaled[y == 0]
n_to_generate = len(X_maj) - len(X_min)

def sso_augment(X, n_samples):
    n_features = X.shape[1]
    augmented = []
    for _ in range(n_samples):
        i, j = np.random.choice(len(X), 2, replace=False)
        alpha = np.random.uniform(-1, 1, n_features)
        sample = X[i] + alpha * (X[j] - X[i])
        augmented.append(sample)
    return np.array(augmented)

X_syn = sso_augment(X_min, n_to_generate)
y_syn = np.ones(n_to_generate, dtype=int)

X_bal = np.vstack([X_scaled, X_syn])
y_bal = np.concatenate([y, y_syn])

# === Train/test split ===
X_train, X_val, y_train, y_val = train_test_split(X_bal, y_bal, test_size=0.2, random_state=42, stratify=y_bal)
input_dim = X_train.shape[1]

ohf = oh.OHFFDRL(input_dim, X_train, X_val, y_train, y_val)

dim = 2 * 3 * input_dim
opt_vector = ohf.whho_optimize(dim)
mu_opt = opt_vector[:len(opt_vector)//2].reshape(3, input_dim)
sigma_opt = np.abs(opt_vector[len(opt_vector)//2:].reshape(3, input_dim)) + 1e-2

# === Step 5: Train final model ===
final_model = ohf.build_model(mu_opt, sigma_opt)
final_model.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=30, batch_size=32, verbose=1)

# === Step 6: Evaluate ===
y_pred_final = np.argmax(final_model.predict(X_val), axis=1)
print(classification_report(y_val, y_pred_final, target_names=["Normal", "Arrhythmia"]))

You can find the "MAINDiagnostics.csv" here.

Authors

• Arman Daliri