更新 function.py
Gitea Actions Test / Challenge-Actions (push) Successful in 53s

This commit is contained in:
2024-11-20 16:12:49 +08:00
parent 56c2bad4d7
commit 9f6ff6c3fe
+89 -96
View File
@@ -1,132 +1,125 @@
# function.py
import numpy as np
data_columns=[]
id = "202210423230"
name = "尹家立"
import pandas as pd
# 这里写自己的代码实现
import numpy as np
from sklearn.impute import SimpleImputer
from sklearn.svm import SVC
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, accuracy_score
from sklearn.preprocessing import StandardScaler
def load_data(data_path):
NAN_THRES=0.5
# 读取数据并指定列
horse_data = pd.read_csv(data_path, header=None, sep=r"\s+")
horse_data = horse_data.loc[:, :22] # 只取到死亡情况(前23列,索引0-22)
horse_data.columns = data_columns # 指定列名
horse_data = horse_data.replace("?", float('NaN')) # 替换缺失值标记
# 数据加载函数
def load_data(filepath):
"""
从文件中加载数据,若不自行正确实现,则无法通过评测,**不记录成绩**。
# 删除无关或缺失值过多的特征
col_del = ["Hospital Number"]
for col in horse_data.columns:
if horse_data[col].isna().sum() > len(horse_data) * NAN_THRES and col not in col_del:
col_del.append(col)
horse_data = horse_data.drop(columns=col_del)
Args:
- directory (str): 数据文件路径。
# 删除标签缺失的样本
horse_data = horse_data.dropna(subset=['outcome']).reset_index(drop=True)
# 分离特征和标签
x_data = horse_data.drop(columns=['outcome'])
y_data = horse_data[['outcome']]
# 标签:保留三分类
y_data['outcome'] = pd.to_numeric(y_data['outcome'], errors='coerce').astype('int32')
# 分离离散和连续特征
x_discrete = x_data[list(set(columns_discrete) - set(col_del))]
x_continue = x_data.drop(columns=x_discrete.columns)
# 离散特征 one-hot 编码
for col in col_oneHot:
if col in x_discrete.columns:
column_onehot = pd.get_dummies(x_discrete[col], prefix=col)
x_discrete = x_discrete.drop(columns=col)
x_discrete = pd.concat([x_discrete, column_onehot], axis=1)
# 用众数填补离散特征缺失值
imp = SimpleImputer(strategy='most_frequent')
x_discrete = pd.DataFrame(imp.fit_transform(x_discrete), columns=x_discrete.columns)
x_discrete = x_discrete.astype('int32')
# 用均值填补连续特征缺失值并归一化
for col in x_continue.columns:
x_continue[col] = pd.to_numeric(x_continue[col], errors='coerce').astype("float64")
x_continue[col] = x_continue[col].fillna(x_continue[col].mean())
x_continue = (x_continue - x_continue.mean()) / (x_continue.max() - x_continue.min())
# 合并特征和标签
horse_data = pd.concat([x_discrete, x_continue, y_data], axis=1)
horse_data = horse_data.sort_index(axis=1) # 排序列名,确保验证集和训练集一致
# 最终返回特征和标签
return horse_data.drop(columns=['outcome']).values, horse_data['outcome'].values
Returns:
- np.array: 特征数据。
- np.array: 标签数据。
"""
data = []
labels = []
with open(filepath, 'r') as file:
for line in file:
fields = line.strip().split()
# 处理缺失值,用0填充
fields = [0 if f == '?' else float(f) for f in fields]
# 标签是倒数第6列
label = int(fields[-6])
# 特征为其他列(不包含倒数第6列标签)
features = np.array(fields[: -6] + fields[-5:], dtype=float)
# 标签映射:将 1->0, 2->1, 3->2
if label == 1:
label = 0
elif label == 2:
label = 1
elif label == 3:
label = 2
else:
print(f"无效标签,跳过此样本:{label}")
continue # 跳过无效标签
# 添加数据和标签
data.append(features)
labels.append(label)
# 转换为numpy数组
data = np.array(data)
labels = np.array(labels)
return data, labels
def train_svm(X, y, learning_rate=0.01, iterations=1000):
def train_svm(X, y, other_params=None):
"""
支持向量机训练函数。
Args:
- X (numpy array): 特征数据。
- y (numpy array): 标签数据。
- learning_rate (float): 学习率 默认值为0.01。
- iterations (int): 训练迭代次数 默认值为1000。
- other_params (str): 其它参数,可选设置。若需设置其它参数,必须设置其默认值,否则将无法通过自动评测,成绩为零。
Returns:
- numpy array: 训练得到的权重。
- SVC: 训练好的SVM模型。
"""
m, n = X.shape
weights = np.zeros(n)
for _ in range(iterations):
for i in range(m):
if y[i] * np.dot(X[i], weights) < 1:
weights += learning_rate * (y[i] * X[i] + (-2 * (1 / iterations) * weights))
else:
weights += learning_rate * (-2 * (1 / iterations) * weights)
return weights
# 使用高斯核(RBF核)训练SVM
model = SVC(kernel='rbf', C=1.0, gamma='scale', random_state=42)
model.fit(X, y)
return model
def predict_svm(inX, weights):
def predict_svm(inX, model):
"""
支持向量机预测函数。
Args:
- inX (numpy array): 输入待分类的样本。
- weights (numpy array): 训练得到的权重。
- model (SVC): 训练得到的支持向量机模型。
Returns:
- int: 预测的分类标签。
"""
return 1 if np.dot(inX, weights) >= 0 else -1
return model.predict(inX.reshape(1, -1))[0]
if __name__ == "__main__":
# 取得列信息
with open("./data/horse-colic.names", 'r', encoding='utf-8') as f:
data_names = f.read()
data_columns = re.findall(r'(?<=[0-9]:).*', data_names)
data_columns = [col.strip() for col in data_columns[:-3]]
data_columns
# 离散列
columns_discrete = ['surgery?', 'Age', 'temperature of extremities', 'peripheral pulse',
'mucous membranes', 'capillary refill time', 'pain - a subjective judgement of the horse\'s pain level', 'peristalsis',
'abdominal distension', 'nasogastric tube', 'nasogastric reflux', 'rectal examination - feces', 'abdomen', 'abdominocentesis appearance']
# 独热编码列
col_oneHot = ['surgery?', 'Age']
# 加载训练集
# 加载数据
X_train, y_train = load_data("./data/horse-colic.data")
# 加载测试集
X_test, y_test = load_data("./data/horse-colic.test")
# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(X_train, y_train, test_size=0.2, random_state=42, stratify=y_train)
# 数据标准化
scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test = scaler.transform(X_test)
# 训练支持向量机模型
weights = train_svm(X_train, y_train, learning_rate=0.01, iterations=1000)
print("正在训练支持向量机模型...")
svm_model = train_svm(X_train, y_train)
# 预测测试集
y_pred = svm_model.predict(X_test)
# 输出评估结果
print("\n分类报告:")
print(classification_report(y_test, y_pred, target_names=["存活", "安乐死", "死亡"]))
# 输出准确率
accuracy = accuracy_score(y_test, y_pred)
print(f"测试集准确率: {accuracy * 100:.2f}%")
# 测试单个数据
y_pred = predict_svm(X_test[11], weights)
print(f"预测结果: {y_pred}, 实际结果: {y_test[11]}")
test_sample = X_test[11] # 选择第12个样本(从0开始计数)
y_pred_single = predict_svm(test_sample, svm_model)
print(f"\n单个样本预测结果: {y_pred_single}, 实际结果: {y_test[11]}")