import os import pandas as pd from sklearn.neighbors import KNeighborsRegressor # 读取第一个文件夹中的所有csv文件 folder1_path = "/path/to/folder1" files1 = os.listdir(folder1_path) dfs1 = [] for file1 in files1: if file1.endswith(".csv"): file1_path = os.path.join(folder1_path, file1) df1 = pd.read_csv(file1_path, usecols=[1,2,3,4]) dfs1.append(df1) # 将第一个文件夹中的所有数据合并为一个DataFrame df_X = pd.concat(dfs1, ignore_index=True) # 读取第二个文件夹中的所有csv文件 folder2_path = "/path/to/folder2" files2 = os.listdir(folder2_path) dfs2 = [] for file2 in files2: if file2.endswith(".csv"): file2_path = os.path.join(folder2_path, file2) df2 = pd.read_csv(file2_path, usecols=[1]) dfs2.append(df2) # 将第二个文件夹中的所有数据合并为一个DataFrame df_X["X5"] = pd.concat(dfs2, ignore_index=True) # 读取第三个文件夹中的所有csv文件 folder3_path = "/path/to/folder3" files3 = os.listdir(folder3_path) dfs3 = [] for file3 in files3: if file3.endswith(".csv"): file3_path = os.path.join(folder3_path, file3) df3 = pd.read_csv(file3_path, usecols=[2,6]) dfs3.append(df3) # 将第三个文件夹中的所有数据合并为一个DataFrame df_y = pd.concat(dfs3, ignore_index=True) # 训练k邻近回归模型 k = 5 model = KNeighborsRegressor(n_neighbors=k) model.fit(df_X, df_y) 这段代码之后的步骤是什么

这段代码没有划分训练集和测试集进行训练再做预测，而是直接将所有数据合并为一个DataFrame进行训练和预测。可能是因为数据量比较小，划分训练集和测试集不会影响模型的性能评估，或者是因为数据量比较大，划分训练集和测试集会增加代码的复杂度和运行时间，而合并所有数据后训练和预测可以简化代码，并且可以利用更多的数据训练模型，提高模型的泛化能力。 然而，在实际应用中，为了有效评估模型的性能，一般还是需要将数据集划分为训练集和测试集进行训练和预测，并采用交叉验证等方法来进一步验证模型的鲁棒性和泛化能力。

好的！以下是将你的代码拆分为两个Python文件的具体步骤： **第一个文件 `data_split.py`** 用于完成数据集的划分（训练集、验证集和测试集）。 **第二个文件 `jda_processing.py`** 则专注于通过 Joint Distribution Alignment (JDA) 对齐分布并处理数据。 --- ### **File 1: 数据分割 (`data_split.py`)** 此脚本主要用于加载原始数据并将其划分为训练集、验证集和测试集： ```python import os import pandas as pd import numpy as np from sklearn.model_selection import train_test_split # 加载数据函数（与原代码一致） def load_data(main_folder='C:/Users/Lenovo/Desktop/crcw不同端12k在0负载下/风扇端'): data_list = [] label_list = [] for folder_name in sorted(os.listdir(main_folder)): folder_path = os.path.join(main_folder, folder_name) if os.path.isdir(folder_path): csv_files = [f for f in os.listdir(folder_path) if f.endswith('.csv')] for filename in sorted(csv_files): file_path = os.path.join(folder_path, filename) csv_data = pd.read_csv(file_path, header=None) if csv_data.shape[1] >= 4: csv_data = csv_data.iloc[:, [0, 1, 2]].values else: print(f"Skipping file {filename}, unexpected shape: {csv_data.shape}") continue data_list.append(csv_data.flatten()) # 展平为一维向量 if '内圈故障' in folder_name: class_label = 0 elif '球故障' in folder_name: class_label = 1 else: continue label_list.append(class_label) if data_list and label_list: X = np.array(data_list) # 特征矩阵 y = np.array(label_list) # 标签 return X, y else: raise ValueError("No valid data to process.") # 数据集划分函数 def split_dataset(X, y, test_size=0.2, val_size=0.25, random_state=42): """ :param X: 特征数据 :param y: 标签数据 :param test_size: 测试集比例 :param val_size: 验证集占剩余数据的比例 :return: 训练集、验证集和测试集 """ X_train_val, X_test, y_train_val, y_test = train_test_split( X, y, test_size=test_size, stratify=y, random_state=random_state ) X_train, X_val, y_train, y_val = train_test_split( X_train_val, y_train_val, test_size=val_size, stratify=y_train_val, random_state=random_state ) print(f"Training set size: {X_train.shape[0]}") print(f"Validation set size: {X_val.shape[0]}") print(f"Test set size: {X_test.shape[0]}") return X_train, X_val, X_test, y_train, y_val, y_test if __name__ == "__main__": main_folder = 'C:/Users/Lenovo/Desktop/crcw不同端12k在0负载下/风扇端' try: X, y = load_data(main_folder) X_train, X_val, X_test, y_train, y_val, y_test = split_dataset(X, y) # 存储到本地以供后续使用 np.savez('dataset.npz', X_train=X_train, X_val=X_val, X_test=X_test, y_train=y_train, y_val=y_val, y_test=y_test) print("Dataset saved successfully.") except Exception as e: print(e) ``` --- ### **File 2: JDA 处理 (`jda_processing.py`)** 此部分实现 Joint Distribution Alignment 并对齐源域和目标域的数据分布： ```python import numpy as np from sklearn.neighbors import KNeighborsClassifier from scipy.linalg import sqrtm class JDA: def __init__(self, n_components=3, lambd=1.0): self.n_components = n_components self.lambd = lambd def fit(self, Xs, Xt, ys): ns, _ = Xs.shape nt, _ = Xt.shape Z = np.vstack((Xs, Xt)) Z_mean = np.mean(Z, axis=0) Xs_centered = Xs - np.mean(Xs, axis=0) Xt_centered = Xt - np.mean(Xt, axis=0) C_s = np.cov(Xs_centered.T) / ns C_t = np.cov(Xt_centered.T) / nt Cs_inv_sqrt = invsqrt(C_s + self.lambd * np.eye(len(Z_mean))) Ct_inv_sqrt = invsqrt(C_t + self.lambd * np.eye(len(Z_mean))) M = np.dot(Cs_inv_sqrt, Ct_inv_sqrt).T U, S, V = np.linalg.svd(M[:ns], full_matrices=False) W = np.dot(U[:, :self.n_components], V[:self.n_components]) self.Xs_new = np.dot(Xs_centered, W) self.Xr_new = np.dot(np.concatenate([Xs_centered, Xt_centered]), W) return self def transform(self, X): return np.dot(X - np.mean(X, axis=0), self.W) @staticmethod def invsqrt(matrix): u, s, v = np.linalg.svd(matrix) return np.dot(u, np.dot(np.diag(1.0 / np.sqrt(s)), v)) # 主程序入口 if __name__ == '__main__': dataset = np.load('dataset.npz') X_train_source = dataset['X_train'] X_train_target = dataset['X_val'] # 假设用验证集作为目标域 y_train_source = dataset['y_train'] jda = JDA(n_components=3, lambd=1e-6) jda.fit(X_train_source, X_train_target, y_train_source) X_train_aligned = jda.transform(X_train_source) X_val_aligned = jda.transform(X_train_target) clf = KNeighborsClassifier(n_neighbors=3) clf.fit(X_train_aligned, y_train_source) accuracy = clf.score(jda.transform(dataset['X_test']), dataset['y_test']) print(f"Accuracy on test set after JDA alignment: {accuracy:.4f}") print("Joint Distribution Alignment completed.") ``` --- ###