|
|
import random
|
|
|
import pandas as pd
|
|
|
import numpy as np
|
|
|
import json
|
|
|
import os
|
|
|
import matplotlib.pyplot as plt
|
|
|
from sklearn.decomposition import PCA
|
|
|
from sklearn.preprocessing import StandardScaler
|
|
|
import joblib
|
|
|
|
|
|
# 设置中文显示
|
|
|
plt.rcParams['font.sans-serif'] = ['SimHei'] # 解决中文显示
|
|
|
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示
|
|
|
|
|
|
|
|
|
# ===================== 数据生成模块 =====================
|
|
|
def generate_student_consume_data(
|
|
|
n_train=100, n_test=5,
|
|
|
save_paths={"csv": "student_consume_train.csv", "excel": "student_consume_train.xlsx",
|
|
|
"json": "student_consume_train.json"},
|
|
|
test_path="student_consume_test.csv"
|
|
|
):
|
|
|
"""
|
|
|
生成模拟学生消费数据（含缺失值、异常值）
|
|
|
:param n_train: 训练数据量
|
|
|
:param n_test: 测试数据量
|
|
|
:param save_paths: 训练数据保存路径（csv/excel/json）
|
|
|
:param test_path: 测试数据保存路径
|
|
|
"""
|
|
|
# 生成基础信息
|
|
|
student_ids = [f"2024{str(i).zfill(6)}" for i in range(1, n_train + n_test + 1)]
|
|
|
names = [f"学生{j}" for j in range(1, n_train + n_test + 1)]
|
|
|
|
|
|
# 生成消费数据（10次消费记录，含缺失/异常值）
|
|
|
consume_data = []
|
|
|
for idx, (sid, name) in enumerate(zip(student_ids, names)):
|
|
|
row = [sid, name]
|
|
|
# 生成10次消费金额（正常范围10-50元，随机插入异常值/缺失值）
|
|
|
for _ in range(10):
|
|
|
# 10%概率生成缺失值，5%概率生成异常值（>200或<0）
|
|
|
if random.random() < 0.1:
|
|
|
row.append(None)
|
|
|
elif random.random() < 0.05:
|
|
|
row.append(random.choice([random.randint(-50, 0), random.randint(200, 500)]))
|
|
|
else:
|
|
|
row.append(round(random.uniform(10, 50), 2))
|
|
|
consume_data.append(row)
|
|
|
|
|
|
# 拆分训练/测试数据
|
|
|
train_data = consume_data[:n_train]
|
|
|
test_data = consume_data[n_train:]
|
|
|
|
|
|
# 定义列名
|
|
|
cols = ["学号", "姓名"] + [f"消费{i + 1}" for i in range(10)]
|
|
|
|
|
|
# 保存训练数据（多格式）
|
|
|
# CSV
|
|
|
df_train = pd.DataFrame(train_data, columns=cols)
|
|
|
df_train.to_csv(save_paths["csv"], index=False)
|
|
|
# Excel
|
|
|
df_train.to_excel(save_paths["excel"], index=False)
|
|
|
# JSON（嵌套格式）
|
|
|
json_train = []
|
|
|
for row in train_data:
|
|
|
json_train.append({
|
|
|
"student_info": {"学号": row[0], "姓名": row[1]},
|
|
|
"consume_records": row[2:]
|
|
|
})
|
|
|
with open(save_paths["json"], "w", encoding="utf-8") as f:
|
|
|
json.dump(json_train, f, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# 保存测试数据
|
|
|
df_test = pd.DataFrame(test_data, columns=cols)
|
|
|
df_test.to_csv(test_path, index=False)
|
|
|
|
|
|
print(f"模拟数据生成完成！训练数据{len(train_data)}条，测试数据{len(test_data)}条")
|
|
|
return save_paths, test_path
|
|
|
|
|
|
|
|
|
# ===================== 数据导入模块 =====================
|
|
|
def import_csv_data(file_path):
|
|
|
"""导入CSV文件，缺失值补None，返回DataFrame"""
|
|
|
if not os.path.exists(file_path):
|
|
|
print(f"错误：文件{file_path}不存在！")
|
|
|
return pd.DataFrame()
|
|
|
if os.path.getsize(file_path) == 0:
|
|
|
print(f"提示：文件{file_path}为空！")
|
|
|
return pd.DataFrame()
|
|
|
|
|
|
df = pd.read_csv(file_path, na_values=["", "NaN", "None"], keep_default_na=False)
|
|
|
df = df.where(pd.notna(df), None) # 缺失值统一替换为None
|
|
|
return df
|
|
|
|
|
|
|
|
|
def import_excel_data(file_path):
|
|
|
"""导入Excel文件（兼容xls/xlsx），自动识别表头，缺失值补None"""
|
|
|
if not os.path.exists(file_path):
|
|
|
print(f"错误：文件{file_path}不存在！")
|
|
|
return pd.DataFrame()
|
|
|
if os.path.getsize(file_path) == 0:
|
|
|
print(f"提示：文件{file_path}为空！")
|
|
|
return pd.DataFrame()
|
|
|
|
|
|
df = pd.read_excel(file_path, header=0) # 自动识别首行作表头
|
|
|
df = df.where(pd.notna(df), None) # 缺失值统一替换为None
|
|
|
return df
|
|
|
|
|
|
|
|
|
def import_json_data(file_path):
|
|
|
"""导入JSON文件，解析嵌套字段，缺失值补None"""
|
|
|
if not os.path.exists(file_path):
|
|
|
print(f"错误：文件{file_path}不存在！")
|
|
|
return {}
|
|
|
if os.path.getsize(file_path) == 0:
|
|
|
print(f"提示：文件{file_path}为空！")
|
|
|
return {}
|
|
|
|
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
|
data = json.load(f)
|
|
|
|
|
|
# 解析嵌套字段，标准化格式
|
|
|
parsed_data = []
|
|
|
for item in data:
|
|
|
row = {
|
|
|
"学号": item.get("student_info", {}).get("学号", None),
|
|
|
"姓名": item.get("student_info", {}).get("姓名", None),
|
|
|
"消费记录": item.get("consume_records", [None] * 10)
|
|
|
}
|
|
|
parsed_data.append(row)
|
|
|
return {"data": parsed_data}
|
|
|
|
|
|
|
|
|
# ===================== 数据清洗模块 =====================
|
|
|
def remove_duplicates(data, key_col="学号"):
|
|
|
"""
|
|
|
按主键去重，保留第一条
|
|
|
:param data: DataFrame/字典类型数据
|
|
|
:param key_col: 主键列名
|
|
|
:return: 去重后的数据
|
|
|
"""
|
|
|
if isinstance(data, pd.DataFrame):
|
|
|
if key_col not in data.columns:
|
|
|
print(f"错误：主键列{key_col}不存在！")
|
|
|
return data
|
|
|
cleaned = data.drop_duplicates(subset=[key_col], keep="first")
|
|
|
elif isinstance(data, dict) and "data" in data:
|
|
|
# 处理字典类型数据
|
|
|
seen_keys = set()
|
|
|
cleaned_list = []
|
|
|
for item in data["data"]:
|
|
|
key = item.get(key_col, None)
|
|
|
if key is None:
|
|
|
cleaned_list.append(item)
|
|
|
continue
|
|
|
if key not in seen_keys:
|
|
|
seen_keys.add(key)
|
|
|
cleaned_list.append(item)
|
|
|
cleaned = {"data": cleaned_list}
|
|
|
else:
|
|
|
print("错误：不支持的数据类型！")
|
|
|
return data
|
|
|
print(
|
|
|
f"去重完成！原数据{len(data) if isinstance(data, pd.DataFrame) else len(data['data'])}条，清洗后{len(cleaned) if isinstance(cleaned, pd.DataFrame) else len(cleaned['data'])}条")
|
|
|
return cleaned
|
|
|
|
|
|
|
|
|
def fill_missing_values(data, strategy="mean", key_col="学号"):
|
|
|
"""
|
|
|
缺失值填充（均值/众数/前向填充）
|
|
|
:param data: DataFrame/字典类型数据
|
|
|
:param strategy: 填充策略（mean/median/ffill）
|
|
|
:param key_col: 主键列（用于保留非数值列）
|
|
|
:return: 填充后的数据
|
|
|
"""
|
|
|
if isinstance(data, pd.DataFrame):
|
|
|
# 分离数值列和非数值列
|
|
|
non_numeric_cols = [col for col in data.columns if col in [key_col, "姓名"]]
|
|
|
numeric_cols = [col for col in data.columns if col not in non_numeric_cols]
|
|
|
df_numeric = data[numeric_cols].copy()
|
|
|
|
|
|
# 填充策略
|
|
|
if strategy == "mean":
|
|
|
fill_value = df_numeric.astype(float).mean()
|
|
|
elif strategy == "mode":
|
|
|
fill_value = df_numeric.astype(float).mode().iloc[0]
|
|
|
elif strategy == "ffill":
|
|
|
df_filled = df_numeric.fillna(method="ffill")
|
|
|
else:
|
|
|
print(f"错误：不支持的填充策略{strategy}！")
|
|
|
return data
|
|
|
|
|
|
if strategy != "ffill":
|
|
|
df_filled = df_numeric.fillna(fill_value)
|
|
|
|
|
|
# 合并非数值列
|
|
|
cleaned = pd.concat([data[non_numeric_cols].reset_index(drop=True), df_filled.reset_index(drop=True)], axis=1)
|
|
|
|
|
|
elif isinstance(data, dict) and "data" in data:
|
|
|
# 处理字典类型数据
|
|
|
cleaned_list = []
|
|
|
for item in data["data"]:
|
|
|
consume_records = item.get("消费记录", [None] * 10)
|
|
|
# 过滤None，计算填充值
|
|
|
valid_vals = [v for v in consume_records if v is not None and isinstance(v, (int, float))]
|
|
|
if not valid_vals:
|
|
|
filled_records = [0.0] * 10 # 无有效值时填充0
|
|
|
else:
|
|
|
if strategy == "mean":
|
|
|
fill_val = np.mean(valid_vals)
|
|
|
elif strategy == "mode":
|
|
|
fill_val = np.mode(valid_vals)[0] if len(valid_vals) > 0 else 0.0
|
|
|
elif strategy == "ffill":
|
|
|
# 前向填充
|
|
|
filled_records = []
|
|
|
prev_val = 0.0
|
|
|
for val in consume_records:
|
|
|
if val is None or not isinstance(val, (int, float)):
|
|
|
filled_records.append(prev_val)
|
|
|
else:
|
|
|
filled_records.append(val)
|
|
|
prev_val = val
|
|
|
fill_val = None
|
|
|
else:
|
|
|
print(f"错误：不支持的填充策略{strategy}！")
|
|
|
return data
|
|
|
|
|
|
if strategy != "ffill":
|
|
|
filled_records = [fill_val if v is None else v for v in consume_records]
|
|
|
|
|
|
cleaned_item = {
|
|
|
"学号": item.get("学号"),
|
|
|
"姓名": item.get("姓名"),
|
|
|
"消费记录": filled_records
|
|
|
}
|
|
|
cleaned_list.append(cleaned_item)
|
|
|
cleaned = {"data": cleaned_list}
|
|
|
else:
|
|
|
print("错误：不支持的数据类型！")
|
|
|
return data
|
|
|
|
|
|
print(f"缺失值填充完成（策略：{strategy}）")
|
|
|
return cleaned
|
|
|
|
|
|
|
|
|
def handle_outliers(data, strategy="mean", key_col="学号", alpha=3):
|
|
|
"""
|
|
|
基于3σ准则处理异常值
|
|
|
:param data: DataFrame/字典类型数据
|
|
|
:param strategy: 处理策略（drop/mean/mode/ffill）
|
|
|
:param key_col: 主键列
|
|
|
:param alpha: 3σ准则的α值
|
|
|
:return: 处理后的数据
|
|
|
"""
|
|
|
if isinstance(data, pd.DataFrame):
|
|
|
# 分离数值列和非数值列
|
|
|
non_numeric_cols = [col for col in data.columns if col in [key_col, "姓名"]]
|
|
|
numeric_cols = [col for col in data.columns if col not in non_numeric_cols]
|
|
|
df_numeric = data[numeric_cols].astype(float).copy()
|
|
|
|
|
|
# 计算3σ范围
|
|
|
mean_vals = df_numeric.mean()
|
|
|
std_vals = df_numeric.std()
|
|
|
lower_bound = mean_vals - alpha * std_vals
|
|
|
upper_bound = mean_vals + alpha * std_vals
|
|
|
|
|
|
# 标记异常值
|
|
|
outliers_mask = (df_numeric < lower_bound) | (df_numeric > upper_bound)
|
|
|
|
|
|
if strategy == "drop":
|
|
|
# 删除含异常值的行
|
|
|
row_mask = outliers_mask.any(axis=1)
|
|
|
df_cleaned = df_numeric[~row_mask]
|
|
|
non_numeric_cleaned = data[non_numeric_cols][~row_mask].reset_index(drop=True)
|
|
|
else:
|
|
|
# 填充异常值
|
|
|
df_cleaned = df_numeric.copy()
|
|
|
for col in numeric_cols:
|
|
|
# 计算该列的填充值
|
|
|
valid_vals = df_cleaned[col][~outliers_mask[col]]
|
|
|
if len(valid_vals) == 0:
|
|
|
fill_val = 0.0
|
|
|
elif strategy == "mean":
|
|
|
fill_val = valid_vals.mean()
|
|
|
elif strategy == "mode":
|
|
|
fill_val = valid_vals.mode().iloc[0]
|
|
|
elif strategy == "ffill":
|
|
|
df_cleaned[col] = df_cleaned[col].mask(outliers_mask[col]).fillna(method="ffill")
|
|
|
continue
|
|
|
else:
|
|
|
print(f"错误：不支持的异常值处理策略{strategy}！")
|
|
|
return data
|
|
|
# 填充异常值
|
|
|
if strategy != "ffill":
|
|
|
df_cleaned = df_cleaned.mask(outliers_mask, fill_val)
|
|
|
non_numeric_cleaned = data[non_numeric_cols].reset_index(drop=True)
|
|
|
|
|
|
# 合并数据
|
|
|
cleaned = pd.concat([non_numeric_cleaned, df_cleaned.reset_index(drop=True)], axis=1)
|
|
|
|
|
|
elif isinstance(data, dict) and "data" in data:
|
|
|
# 处理字典类型数据
|
|
|
cleaned_list = []
|
|
|
for item in data["data"]:
|
|
|
consume_records = item.get("消费记录", [])
|
|
|
valid_vals = [v for v in consume_records if v is not None and isinstance(v, (int, float))]
|
|
|
if not valid_vals:
|
|
|
cleaned_list.append(item)
|
|
|
continue
|
|
|
|
|
|
# 计算3σ范围
|
|
|
mean_val = np.mean(valid_vals)
|
|
|
std_val = np.std(valid_vals)
|
|
|
lower = mean_val - alpha * std_val
|
|
|
upper = mean_val + alpha * std_val
|
|
|
|
|
|
# 处理异常值
|
|
|
cleaned_records = []
|
|
|
prev_val = mean_val # 前向填充的初始值
|
|
|
for val in consume_records:
|
|
|
if val is None:
|
|
|
cleaned_records.append(val)
|
|
|
continue
|
|
|
# 判断是否为异常值
|
|
|
is_outlier = val < lower or val > upper
|
|
|
if not is_outlier:
|
|
|
cleaned_records.append(val)
|
|
|
prev_val = val
|
|
|
else:
|
|
|
if strategy == "drop":
|
|
|
break # 丢弃整行
|
|
|
elif strategy == "mean":
|
|
|
cleaned_records.append(mean_val)
|
|
|
elif strategy == "mode":
|
|
|
cleaned_records.append(np.mode(valid_vals)[0] if len(valid_vals) > 0 else 0.0)
|
|
|
elif strategy == "ffill":
|
|
|
cleaned_records.append(prev_val)
|
|
|
else:
|
|
|
cleaned_records.append(val)
|
|
|
|
|
|
if strategy != "drop":
|
|
|
cleaned_item = {
|
|
|
"学号": item.get("学号"),
|
|
|
"姓名": item.get("姓名"),
|
|
|
"消费记录": cleaned_records
|
|
|
}
|
|
|
cleaned_list.append(cleaned_item)
|
|
|
|
|
|
cleaned = {"data": cleaned_list}
|
|
|
else:
|
|
|
print("错误：不支持的数据类型！")
|
|
|
return data
|
|
|
|
|
|
print(f"异常值处理完成（策略：{strategy}，3σ准则）")
|
|
|
return cleaned
|
|
|
|
|
|
|
|
|
# ===================== 数据分析模块 =====================
|
|
|
def statistical_analysis(data, key_col="学号"):
|
|
|
"""
|
|
|
统计特征分析：计算每个学生消费次数、均值、标准差
|
|
|
:param data: 清洗后的DataFrame数据
|
|
|
:return: 统计结果DataFrame
|
|
|
"""
|
|
|
if not isinstance(data, pd.DataFrame):
|
|
|
print("错误：仅支持DataFrame类型数据的统计分析！")
|
|
|
return pd.DataFrame()
|
|
|
|
|
|
# 分离消费列
|
|
|
consume_cols = [col for col in data.columns if "消费" in col]
|
|
|
# 统计每个学生的消费特征
|
|
|
stats = []
|
|
|
for idx, row in data.iterrows():
|
|
|
sid = row[key_col]
|
|
|
name = row["姓名"]
|
|
|
# 提取有效消费记录
|
|
|
consume_vals = [v for v in row[consume_cols] if v is not None and isinstance(v, (int, float))]
|
|
|
# 计算统计特征
|
|
|
count = len(consume_vals)
|
|
|
mean_val = round(np.mean(consume_vals), 2) if count > 0 else 0.0
|
|
|
std_val = round(np.std(consume_vals), 2) if count > 1 else 0.0
|
|
|
|
|
|
stats.append({
|
|
|
"学号": sid,
|
|
|
"姓名": name,
|
|
|
"消费次数": count,
|
|
|
"消费均值": mean_val,
|
|
|
"消费标准差": std_val
|
|
|
})
|
|
|
|
|
|
stats_df = pd.DataFrame(stats)
|
|
|
print("统计特征分析完成！")
|
|
|
return stats_df
|
|
|
|
|
|
|
|
|
def plot_top10_bar(stats_df, save_dir="plots"):
|
|
|
"""
|
|
|
绘制Top10柱状图（次数/均值/标准差）并保存
|
|
|
:param stats_df: 统计特征DataFrame
|
|
|
:param save_dir: 图片保存目录
|
|
|
"""
|
|
|
if not os.path.exists(save_dir):
|
|
|
os.makedirs(save_dir)
|
|
|
|
|
|
# 1. 消费次数Top10
|
|
|
top10_count = stats_df.nlargest(10, "消费次数")
|
|
|
plt.figure(figsize=(12, 4))
|
|
|
plt.bar(top10_count["姓名"], top10_count["消费次数"], color="skyblue")
|
|
|
plt.title("学生消费次数Top10")
|
|
|
plt.xlabel("学生姓名")
|
|
|
plt.ylabel("消费次数")
|
|
|
plt.xticks(rotation=45)
|
|
|
plt.tight_layout()
|
|
|
plt.savefig(os.path.join(save_dir, "top10_count.png"))
|
|
|
plt.show()
|
|
|
|
|
|
# 2. 消费均值Top10
|
|
|
top10_mean = stats_df.nlargest(10, "消费均值")
|
|
|
plt.figure(figsize=(12, 4))
|
|
|
plt.bar(top10_mean["姓名"], top10_mean["消费均值"], color="lightcoral")
|
|
|
plt.title("学生消费均值Top10")
|
|
|
plt.xlabel("学生姓名")
|
|
|
plt.ylabel("消费均值（元）")
|
|
|
plt.xticks(rotation=45)
|
|
|
plt.tight_layout()
|
|
|
plt.savefig(os.path.join(save_dir, "top10_mean.png"))
|
|
|
plt.show()
|
|
|
|
|
|
# 3. 消费标准差Top10
|
|
|
top10_std = stats_df.nlargest(10, "消费标准差")
|
|
|
plt.figure(figsize=(12, 4))
|
|
|
plt.bar(top10_std["姓名"], top10_std["消费标准差"], color="lightgreen")
|
|
|
plt.title("学生消费标准差Top10")
|
|
|
plt.xlabel("学生姓名")
|
|
|
plt.ylabel("消费标准差（元）")
|
|
|
plt.xticks(rotation=45)
|
|
|
plt.tight_layout()
|
|
|
plt.savefig(os.path.join(save_dir, "top10_std.png"))
|
|
|
plt.show()
|
|
|
|
|
|
print(f"Top10柱状图已保存至{save_dir}目录！")
|
|
|
|
|
|
|
|
|
def plot_pca_scatter(stats_df, save_dir="plots"):
|
|
|
"""
|
|
|
PCA降维后绘制散点图并保存
|
|
|
:param stats_df: 统计特征DataFrame
|
|
|
:param save_dir: 图片保存目录
|
|
|
"""
|
|
|
if not os.path.exists(save_dir):
|
|
|
os.makedirs(save_dir)
|
|
|
|
|
|
# 提取数值特征
|
|
|
features = ["消费次数", "消费均值", "消费标准差"]
|
|
|
X = stats_df[features].values
|
|
|
# 标准化
|
|
|
scaler = StandardScaler()
|
|
|
X_scaled = scaler.fit_transform(X)
|
|
|
# PCA降维至2D
|
|
|
pca = PCA(n_components=2)
|
|
|
X_pca = pca.fit_transform(X_scaled)
|
|
|
|
|
|
# 绘制散点图
|
|
|
plt.figure(figsize=(10, 8))
|
|
|
plt.scatter(X_pca[:, 0], X_pca[:, 1], alpha=0.7, c="orange")
|
|
|
# 标注部分学生
|
|
|
for i in range(min(20, len(stats_df))): # 仅标注前20个学生
|
|
|
plt.annotate(stats_df.iloc[i]["姓名"], (X_pca[i, 0], X_pca[i, 1]), fontsize=8)
|
|
|
plt.title("学生消费特征PCA降维散点图")
|
|
|
plt.xlabel("主成分1")
|
|
|
plt.ylabel("主成分2")
|
|
|
plt.grid(alpha=0.3)
|
|
|
plt.tight_layout()
|
|
|
plt.savefig(os.path.join(save_dir, "pca_scatter.png"))
|
|
|
plt.show()
|
|
|
|
|
|
print(f"PCA散点图已保存至{save_dir}目录！")
|
|
|
|
|
|
|
|
|
# ===================== 学生画像模型构建与应用模块 =====================
|
|
|
def build_portrait_model(stats_df, model_path="student_portrait_model.pkl"):
|
|
|
"""
|
|
|
构建学生画像模型（按消费特征分标签）
|
|
|
标签规则：
|
|
|
- 特别困难：消费均值<15 且 消费次数<5
|
|
|
- 困难：消费均值15-20 且 消费次数<7
|
|
|
- 一般困难：消费均值20-25 或 消费次数7-8
|
|
|
- 不困难：其他情况
|
|
|
:param stats_df: 统计特征DataFrame
|
|
|
:param model_path: 模型保存路径
|
|
|
:return: 带标签的DataFrame
|
|
|
"""
|
|
|
|
|
|
def get_label(row):
|
|
|
mean_val = row["消费均值"]
|
|
|
count = row["消费次数"]
|
|
|
if mean_val < 15 and count < 5:
|
|
|
return "特别困难"
|
|
|
elif 15 <= mean_val < 20 and count < 7:
|
|
|
return "困难"
|
|
|
elif (20 <= mean_val < 25) or (7 <= count <= 8):
|
|
|
return "一般困难"
|
|
|
else:
|
|
|
return "不困难"
|
|
|
|
|
|
# 生成标签
|
|
|
stats_df["困难标签"] = stats_df.apply(get_label, axis=1)
|
|
|
# 保存模型（此处简化为保存标签规则对应的统计阈值，实际可替换为机器学习模型）
|
|
|
model = {
|
|
|
"rules": {
|
|
|
"特别困难": {"消费均值<15": 15, "消费次数<5": 5},
|
|
|
"困难": {"15≤消费均值<20": (15, 20), "消费次数<7": 7},
|
|
|
"一般困难": {"20≤消费均值<25": (20, 25), "7≤消费次数≤8": (7, 8)},
|
|
|
"不困难": "其他"
|
|
|
},
|
|
|
"label_counts": stats_df["困难标签"].value_counts().to_dict()
|
|
|
}
|
|
|
joblib.dump(model, model_path)
|
|
|
|
|
|
print(f"画像模型构建完成！标签分布：{model['label_counts']}")
|
|
|
print(f"模型已保存至{model_path}")
|
|
|
return stats_df
|
|
|
|
|
|
|
|
|
def get_aid_strategy(label):
|
|
|
"""根据标签返回精准帮扶策略"""
|
|
|
strategies = {
|
|
|
"特别困难": "提供全额助学金+生活补贴+勤工俭学优先安排",
|
|
|
"困难": "提供半额助学金+生活补贴",
|
|
|
"一般困难": "提供临时补助+学业辅导",
|
|
|
"不困难": "无需资助，可推荐奖学金申请"
|
|
|
}
|
|
|
return strategies.get(label, "暂无适配的帮扶策略")
|
|
|
|
|
|
|
|
|
def apply_portrait_model(test_data, model_path="student_portrait_model.pkl"):
|
|
|
"""
|
|
|
应用画像模型对测试数据打标签并返回资助建议
|
|
|
:param test_data: 测试数据（DataFrame）
|
|
|
:param model_path: 模型路径
|
|
|
:return: 带标签和建议的字典
|
|
|
"""
|
|
|
# 加载模型
|
|
|
if not os.path.exists(model_path):
|
|
|
print(f"错误：模型文件{model_path}不存在！")
|
|
|
return {}
|
|
|
model = joblib.load(model_path)
|
|
|
|
|
|
# 先对测试数据做统计分析
|
|
|
test_stats = statistical_analysis(test_data)
|
|
|
|
|
|
# 打标签+生成建议
|
|
|
results = {}
|
|
|
for idx, row in test_stats.iterrows():
|
|
|
sid = row["学号"]
|
|
|
name = row["姓名"]
|
|
|
mean_val = row["消费均值"]
|
|
|
count = row["消费次数"]
|
|
|
|
|
|
# 应用模型规则
|
|
|
if mean_val < model["rules"]["特别困难"]["消费均值<15"] and count < model["rules"]["特别困难"]["消费次数<5"]:
|
|
|
label = "特别困难"
|
|
|
elif (model["rules"]["困难"]["15≤消费均值<20"][0] <= mean_val < model["rules"]["困难"]["15≤消费均值<20"][
|
|
|
1]) and count < model["rules"]["困难"]["消费次数<7"]:
|
|
|
label = "困难"
|
|
|
elif (model["rules"]["一般困难"]["20≤消费均值<25"][0] <= mean_val <
|
|
|
model["rules"]["一般困难"]["20≤消费均值<25"][1]) or (
|
|
|
model["rules"]["一般困难"]["7≤消费次数≤8"][0] <= count <= model["rules"]["一般困难"]["7≤消费次数≤8"][
|
|
|
1]):
|
|
|
label = "一般困难"
|
|
|
else:
|
|
|
label = "不困难"
|
|
|
|
|
|
# 获取帮扶策略
|
|
|
strategy = get_aid_strategy(label)
|
|
|
|
|
|
results[sid] = {
|
|
|
"姓名": name,
|
|
|
"消费次数": count,
|
|
|
"消费均值": mean_val,
|
|
|
"困难标签": label,
|
|
|
"帮扶策略": strategy
|
|
|
}
|
|
|
|
|
|
print("画像模型应用完成！测试数据标签及帮扶策略如下：")
|
|
|
for sid, info in results.items():
|
|
|
print(f"学号：{sid}，姓名：{info['姓名']}，标签：{info['困难标签']}，策略：{info['帮扶策略']}")
|
|
|
return results
|
|
|
|
|
|
|
|
|
# ===================== 主函数（全流程执行） =====================
|
|
|
def main():
|
|
|
# 1. 生成模拟数据
|
|
|
save_paths = {"csv": "student_consume_train.csv", "excel": "student_consume_train.xlsx",
|
|
|
"json": "student_consume_train.json"}
|
|
|
test_path = "student_consume_test.csv"
|
|
|
generate_student_consume_data(n_train=100, n_test=5, save_paths=save_paths, test_path=test_path)
|
|
|
|
|
|
# 2. 数据导入（以CSV为例，可替换为excel/json）
|
|
|
train_data = import_csv_data(save_paths["csv"])
|
|
|
if train_data.empty:
|
|
|
return
|
|
|
|
|
|
# 3. 数据清洗
|
|
|
# 去重
|
|
|
train_data_dedup = remove_duplicates(train_data)
|
|
|
# 缺失值填充（均值填充）
|
|
|
train_data_fill = fill_missing_values(train_data_dedup, strategy="mean")
|
|
|
# 异常值处理（均值填充）
|
|
|
train_data_clean = handle_outliers(train_data_fill, strategy="mean")
|
|
|
|
|
|
# 4. 数据分析
|
|
|
# 统计特征分析
|
|
|
stats_df = statistical_analysis(train_data_clean)
|
|
|
# 可视化
|
|
|
plot_top10_bar(stats_df)
|
|
|
plot_pca_scatter(stats_df)
|
|
|
|
|
|
# 5. 构建画像模型
|
|
|
stats_df_with_label = build_portrait_model(stats_df)
|
|
|
|
|
|
# 6. 模型应用（测试数据）
|
|
|
test_data = import_csv_data(test_path)
|
|
|
if not test_data.empty:
|
|
|
# 测试数据先清洗
|
|
|
test_data_dedup = remove_duplicates(test_data)
|
|
|
test_data_fill = fill_missing_values(test_data_dedup, strategy="mean")
|
|
|
test_data_clean = handle_outliers(test_data_fill, strategy="mean")
|
|
|
# 应用模型
|
|
|
aid_results = apply_portrait_model(test_data_clean)
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
main() |
