Contents

代码小整理2

Python 实现差分进化算法

用于解决机场选址优化问题:在 20 个城市中放置 3 个机场,最小化所有城市到最近机场的距离总和

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import random
import numpy as np
import matplotlib.pyplot as plt


def distance(x1, x2):
    """计算两个坐标点的欧氏距离平方(无需开方以优化性能)"""
    return (x1[0] - x2[0]) ** 2 + (x1[1] - x2[1]) ** 2


def total_distance(airports, cities):
    """计算所有城市到最近机场的距离总和(使用向量化计算)"""
    airports_arr = np.array(airports)
    cities_arr = np.array(cities)
    # 计算每个城市到所有机场的平方距离
    diff = cities_arr[:, None, :] - airports_arr[None, :, :]
    dist_sq = np.sum(diff**2, axis=2)
    # 获取每个城市到最近机场的距离
    min_dists = np.min(dist_sq, axis=1)
    return np.sum(min_dists)


# 城市坐标数据集
cities = [
    [91, 492], [400, 327], [253, 288], [165, 299], [562, 293],
    [305, 449], [375, 270], [534, 350], [473, 506], [165, 379],
    [168, 339], [406, 537], [131, 571], [320, 368], [233, 410],
    [207, 457], [94, 410], [456, 350], [509, 444], [108, 531]
]

# 算法参数
POP_SIZE = 50  # 种群规模
MAX_ITER = 10000  # 最大迭代次数

# 记录最佳解和迭代信息
best_global = float("inf")
best_solution = None
no_improve = 0
convergence_data = []
MAX_NO_IMPROVE = 100  # 连续无改进的最大迭代次数

# 获取城市坐标边界
x_coords = [c[0] for c in cities]
y_coords = [c[1] for c in cities]
x_min, x_max = min(x_coords), max(x_coords)
y_min, y_max = min(y_coords), max(y_coords)

# 初始化种群(每个个体包含3个机场坐标)
population = []
for _ in range(POP_SIZE):
    airport1 = [
        round(random.uniform(x_min, x_max), 2),
        round(random.uniform(y_min, y_max), 2),
    ]
    airport2 = [
        round(random.uniform(x_min, x_max), 2),
        round(random.uniform(y_min, y_max), 2),
    ]
    airport3 = [
        round(random.uniform(x_min, x_max), 2),
        round(random.uniform(y_min, y_max), 2),
    ]
    population.append([airport1, airport2, airport3])

# 差分进化主循环
for iter in range(MAX_ITER):
    pop_array = np.array(population)
    new_population = []
    no_improve_flag = True

    # 自适应差分权重(随迭代次数递减)
    F = 0.9 * (1 - iter / MAX_ITER) + 0.4 * (iter / MAX_ITER)

    for individual in population:
        # 变异操作:生成变异向量
        a, b, c = random.choices(population, k=3)
        mutant = a + F * (np.array(b) - np.array(c))
        mutant = mutant.tolist()

        # 交叉操作:生成试验向量
        trial = []
        for j in range(3):
            if random.random() <= 0.5 or j == random.randint(0, 2):
                trial.append(mutant[j])
            else:
                trial.append(individual[j])

        # 选择操作:保留更优解
        current_fitness = total_distance(individual, cities)
        trial_fitness = total_distance(trial, cities)
        if trial_fitness < current_fitness:
            new_population.append(trial)
            # 更新全局最优解
            if trial_fitness < best_global:
                best_global = trial_fitness
                best_solution = np.copy(trial)
                no_improve = 0
                no_improve_flag = False
        else:
            new_population.append(individual)

    population = new_population
    # 记录收敛数据
    convergence_data.append(best_global)

    if no_improve_flag:
        no_improve += 1
        # 检查提前终止条件
        if no_improve >= MAX_NO_IMPROVE:
            print(f"提前终止于第{iter}代,已连续{no_improve}代无改进")
            break

# 结果提取与验证
# 记录最优个体的下标
best_idx = 0
for i in range(len(population)):
    if total_distance(population[i], cities) < total_distance(
        population[best_idx], cities
    ):
        best_idx = i
# 记录次优个体的下标
second_best_idx = 0
for i in range(len(population)):
    if i != best_idx and total_distance(population[i], cities) < total_distance(
        population[second_best_idx], cities
    ):
        second_best_idx = i

# 计算最优个体与次优个体的差值,以判断收敛情况
diff = total_distance(population[second_best_idx], cities) - total_distance(
    population[best_idx], cities
)
print(f"最优解与次优解距离差: {diff:.2f}")
print(f"最优机场坐标: {population[best_idx]}")
print(f"最小距离和: {total_distance(population[best_idx], cities):.2f}")
  • 可视化结果
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
# 显示中文和负号
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

# 绘制收敛曲线
plt.figure(figsize=(12, 10))
plt.subplot(2, 1, 1)
plt.plot(convergence_data)
plt.title("收敛曲线")
plt.xlabel("迭代次数")
plt.ylabel("最优距离和")
plt.grid(True)

# 可视化结果
plt.figure(figsize=(10, 8))
# 绘制城市
for i, (x, y) in enumerate(cities):
    plt.scatter(x, y, c="b")
    plt.text(x, y, f"({x},{y})", ha="center", va="bottom", fontsize=10)
lines = [
    [12, 19], [19, 0], [12, 15], [0, 15], [0, 16],
    [16, 9], [9, 10], [10, 3], [3, 2], [15, 14],
    [15, 5], [14, 2], [14, 13], [13, 2], [5, 1],
    [13, 1], [1, 6], [1, 17], [17, 7], [7, 4],
    [17, 18], [18, 8], [8, 11]
]
for i in range(len(x_coords)):
    for j in range(len(y_coords)):
        if [i, j] in lines:
            plt.plot([x_coords[i], x_coords[j]], [y_coords[i], y_coords[j]])

# 绘制机场
airport_x = [p[0] for p in population[best_idx]]
airport_y = [p[1] for p in population[best_idx]]
plt.scatter(airport_x, airport_y, c="r", marker="s", s=100)

# 机场-地点连线线
for city in cities:
    closest = min(population[best_idx], key=lambda a: distance(a, city))
    plt.plot([city[0], closest[0]], [city[1], closest[1]], ":", c="gray", alpha=0.5)
plt.title("机场选址优化结果")
plt.xlabel("X 坐标")
plt.ylabel("Y 坐标")
plt.grid(True)
plt.show()

状态空间模型与深度优先搜索

问题的背景是有三名传教士和三名食人族在河的一侧。他们想要过河,但只有一条船,船每次只能搭载两个人
如果在任何时候食人族的数量超过传教士的数量,食人族就会吃掉传教士。此外,船必须有人在河一侧划船才能返回,船不能自己回来
为了解决这个问题,我们将搜索所有可能移动的状态空间,最终只求找到一种可行解就好

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
from typing import List, Set

# 船搭载的所有可能为:(1,1)、(1,0)、(0,1)、(2,0)、(0,2),注:(食人族,传教士)
cannibal_moves = [1, 1, 0, 2, 0]
missionary_moves = [1, 0, 1, 0, 2]


class State:
    """
    表示河岸状态的状态空间模型
    """

    def __init__(
        self,
        left_m: int,  # 左岸传教士数量
        left_c: int,  # 左岸食人族数量
        right_m: int,  # 右岸传教士数量
        right_c: int,  # 右岸食人族数量
        boat: int,  # 船位置(0=左岸, 1=右岸)
    ):
        self.left_m = left_m
        self.left_c = left_c
        self.right_m = right_m
        self.right_c = right_c
        self.boat = boat

    def __eq__(self, other) -> bool:
        """
        判等
        """
        return (
            self.left_m == other.left_m
            and self.left_c == other.left_c
            and self.right_m == other.right_m
            and self.right_c == other.right_c
            and self.boat == other.boat
        )

    def __hash__(self) -> int:
        """
        用于确保 set 中状态的唯一
        """
        return hash(
            f"{self.left_m}{self.left_c}{self.right_m}{self.right_c}{self.boat}"
        )

    def __str__(self) -> str:
        """
        将结果转为字符串,方便输出查看
        """
        return (
            f"【左岸】: {self.left_m}位传教士, {self.left_c}位食人族 "
            f"【右岸】: {self.right_m}位传教士, {self.right_c}位食人族 "
            f"【船】: {'左' if self.boat == 0 else '右'}岸"
        )


def is_valid(state: State, visited: Set[State]) -> bool:
    """
    检查状态是否合理(安全且未访问过)
    """
    # 检查是否存在食人族多于传教士的危险情况
    left_unsafe = state.left_c > state.left_m > 0
    right_unsafe = state.right_c > state.right_m > 0

    # 检查是否有负数
    has_negative = any(
        [state.left_m < 0, state.left_c < 0, state.right_m < 0, state.right_c < 0]
    )

    # 不合理的情况则返回 false
    # state in visited 说明之前出现过这种状态,进行剪枝
    if left_unsafe or right_unsafe or has_negative or state in visited:
        return False
    else:
        return True


def dfs(state: State, visited: Set[State], path: List[State]) -> bool:
    """
    深度优先搜索解决方案
    返回True表示找到可行解
    """
    visited.add(state)
    path.append(state)

    # 终止条件:所有人在右岸
    if state.left_m == 0 and state.left_c == 0:
        return True

    # 根据船位置生成下一步状态
    next_states = []
    for i in range(5):
        if state.boat == 0:  # 船在左岸
            new_state = State(
                left_m=state.left_m - missionary_moves[i],
                left_c=state.left_c - cannibal_moves[i],
                right_m=state.right_m + missionary_moves[i],
                right_c=state.right_c + cannibal_moves[i],
                boat=1,
            )
        else:  # 船在右岸
            new_state = State(
                left_m=state.left_m + missionary_moves[i],
                left_c=state.left_c + cannibal_moves[i],
                right_m=state.right_m - missionary_moves[i],
                right_c=state.right_c - cannibal_moves[i],
                boat=0,
            )
        if is_valid(new_state, visited):
            next_states.append(new_state)

    # 递归搜索
    for ns in next_states:
        if dfs(ns, visited, path):
            return True

    # 回溯
    path.pop()
    return False


def print_solution(path: List[State]):
    """
    输出解决方案的详细步骤
    """
    if not path:
        print("无可行解")
        return

    print("可行渡河方案:\n")
    for step, (s1, s2) in enumerate(zip(path[:-1], path[1:])):
        direction = "从左向右移动 →" if s1.boat == 0 else "从右向左移动 ←"
        delta_m = abs(s2.right_m - s1.right_m)
        delta_c = abs(s2.right_c - s1.right_c)

        print(f"步骤{step+1}: 船上有{delta_m}位传教士, {delta_c}位食人族, {direction} ")
        print(f"    当前状态: {s2}\n")


if __name__ == "__main__":
    start = State(50, 49, 0, 0, 0)
    visited = set()
    solution_path = []

    if dfs(start, visited, solution_path):
        print_solution(solution_path)
    else:
        print("未找到可行解决方案")

gradio 界面设置

以搭建 codemoss 的聊天机器人为例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
import gradio as gr


def user(chatbot, message):
    """处理用户输入"""
    # 在此处添加用户消息处理逻辑
    return chatbot + [[message, None]], ""


def bot(chatbot):
    """生成AI回复"""
    # 在此处添加AI回复生成逻辑
    return chatbot + [[None, "这是AI回复示例"]]


def reset_memory():
    """清除聊天记忆"""
    return []


with gr.Blocks(title="Yiuios ChatGPT") as Yiuios_chatbot:
    with gr.Column():
        # 聊天记录显示区域
        chatbot = gr.Chatbot(value=[], elem_id="chatbot").style(height=350)

        with gr.Row():
            # 用户输入框
            message = gr.Textbox(
                placeholder="输入消息 (Shift+Enter换行, Enter发送)", show_label=False
            ).style(container=False)

            # 控制按钮
            clear_btn = gr.Button("清除对话历史")

        # 提交处理链
        message.submit(
            fn=user, inputs=[chatbot, message], outputs=[chatbot, message]
        ).then(fn=bot, inputs=chatbot, outputs=chatbot)

        # 清除按钮事件
        clear_btn.click(fn=reset_memory, inputs=None, outputs=chatbot)

    # 预设提示词示例
    with gr.Accordion("提示词示例库", open=False, visible=True):
        gr.Markdown(
            "我想让你充当我的程序设计中 **命名变量** 的工作,我给你中文变量名,你返回英文变量名,不需要写解释,只需要返回变量名即可,变量名格式为大驼峰和小驼峰以及简写,记得换行,不需要其他反馈"
        )
        gr.Markdown(
            "我想让你充当 **前端开发专家**。我将提供一些关于Js、Node等前端代码问题的具体信息,而你的工作就是想出为我解决问题的策略。这可能包括建议代码、代码逻辑思路策略。我的第一个请求是“我需要能够动态监听某个元素节点距离当前电脑设备屏幕的左上角的X和Y轴,通过拖拽移动位置浏览器窗口和改变大小浏览器窗口。”"
        )
        gr.Markdown(
            "我想让你担任 **Android开发工程师面试官**。我将成为候选人,您将向我询问Android开发工程师职位的面试问题。我希望你只作为面试官回答。不要一次写出所有的问题。我希望你只对我进行采访。问我问题,等待我的回答。不要写解释。像面试官一样一个一个问我,等我回答。我的第一句话是“面试官你好”"
        )
        gr.Markdown(
            "我希望你担任 **UX/UI 开发人员**。我将提供有关应用程序、网站或其他数字产品设计的一些细节,而你的工作就是想出创造性的方法来改善其用户体验。这可能涉及创建原型设计原型、测试不同的设计并提供有关最佳效果的反馈。我的第一个请求是“我需要帮助为我的新移动应用程序设计一个直观的导航系统。”"
        )
        gr.Markdown(
            "我想让你充当 **软件开发人员**。我将提供一些关于 Web 应用程序要求的具体信息,您的工作是提出用于使用 Golang 和 Angular 开发安全应用程序的架构和代码。我的第一个要求是'我想要一个允许用户根据他们的角色注册和保存他们的车辆信息的系统,并且会有管理员,用户和公司角色。我希望系统使用 JWT 来确保安全。"
        )
        gr.Markdown(
            "我想让你充当 **英英词典**,对于给出的英文单词,你要给出其中文意思以及英文解释,并且给出一个例句,此外不要有其他反馈,第一个单词是“Hello”"
        )
        gr.Markdown(
            "我想让你做一个 **旅游指南**。我会把我的位置写给你,你会推荐一个靠近我的位置的地方。在某些情况下,我还会告诉您我将访问的地方类型。您还会向我推荐靠近我的第一个位置的类似类型的地方。我的第一个建议请求是“我在上海,我只想参观博物馆。”"
        )
        gr.Markdown(
            "我希望你能 **扮演一个角色**,表现得像 {series} 中的 {character}。我希望你像 {character} 一样回应和回答。不要写任何解释。只回答像 {character}。你必须知道 {character} 的所有知识。其中 {series} 《超凡蜘蛛侠》,{character} 是彼得帕克。我的第一句话是“你好”"
        )
        gr.Markdown(
            "我希望你担任 **人生教练**。请总结这本非小说类书籍,[作者] [书名]。以孩子能够理解的方式简化核心原则。另外,你能给我一份关于如何将这些原则实施到我的日常生活中的可操作步骤列表吗?"
        )
        gr.Markdown(
            "作为一名 **营养师**,我想为 2 人设计一份素食食谱,每份含有大约 500 卡路里的热量并且血糖指数较低。你能提供一个建议吗?"
        )
        gr.Markdown(
            "我要你担任 **哲学老师**。我会提供一些与哲学研究相关的话题,你的工作就是用通俗易懂的方式解释这些概念。这可能包括提供示例、提出问题或将复杂的想法分解成更容易理解的更小的部分。我的第一个请求是“我需要帮助来理解不同的哲学理论如何应用于日常生活。”"
        )
        gr.Markdown(
            "我想让你扮演一名 **数学老师**。我将提供一些数学方程式或概念,你的工作是用易于理解的术语来解释它们。这可能包括提供解决问题的分步说明、用视觉演示各种技术或建议在线资源以供进一步研究。我的第一个请求是“我需要帮助来理解概率是如何工作的。”"
        )
        gr.Markdown(
            "我要你把我写的 **句子翻译成表情符号**。我会写句子,你会用表情符号表达它。我只是想让你用表情符号来表达它。除了表情符号,我不希望你回复任何内容。当我需要用英语告诉你一些事情时,我会用 {like this} 这样的大括号括起来。我的第一句话是“你好,请问你的职业是什么?”"
        )
        gr.Markdown(
            "我要你充当 **格言书**。您将为我提供明智的建议、鼓舞人心的名言和意味深长的名言,以帮助指导我的日常决策。此外,如有必要,您可以提出将此建议付诸行动或其他相关主题的实用方法。我的第一个请求是“我需要关于如何在逆境中保持积极性的指导”。"
        )
        gr.Markdown(
            "我希望你充当 **IT 专家**。我会向您提供有关我的技术问题所需的所有信息,而您的职责是解决我的问题。你应该使用你的计算机科学、网络基础设施和 IT 安全知识来解决我的问题。在您的回答中使用适合所有级别的人的智能、简单和易于理解的语言将很有帮助。用要点逐步解释您的解决方案很有帮助。尽量避免过多的技术细节,但在必要时使用它们。我希望您回复解决方案,而不是写任何解释。我的第一个问题是“我的笔记本电脑出现蓝屏错误”"
        )

reference = gr.Markdown("yiuios chatgpt")
demo = gr.TabbedInterface(
    interface_list=[Yiuios_chatbot, reference],
    title="Yiuios",
    tab_names=["聊天机器人", "更多"],
)
demo.queue()
# 启动应用
demo.launch(server_name="0.0.0.0", server_port=7860, share=False)

注意:当使用 Fiddler 等抓包工具时,Python 的 requests 请求可能失败,因为
Fiddler 会作为系统代理拦截所有请求,需要在代码中显式配置代理或关闭 Fiddler

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import requests

# 正确配置代理示例
proxies = {
    'http': 'http://127.0.0.1:8888',  # Fiddler 默认端口
    'https': 'http://127.0.0.1:8888'
}

response = requests.get(
    "https://api.example.com",
    proxies=proxies,  # 启用代理
    verify=False      # 忽略SSL证书验证(仅调试用)
)

林少的 chat_vits_sd

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
// 创建音频播放器组件
const audioPlayer = document.createElement("audio");
audioPlayer.id = "dynamicAudio";
audioPlayer.controls = true;

const audioSource = document.createElement("source");
audioSource.type = "audio/wav";

audioPlayer.appendChild(audioSource);
document.getElementById("rightContainer").appendChild(audioPlayer);

// 创建音频更新按钮
const updateButton = document.createElement("button");
updateButton.textContent = "刷新音频";
updateButton.addEventListener("click", refreshAudio);

document.getElementById("controlPanel").appendChild(updateButton);

function refreshAudio() {
  const timestamp = Date.now(); // 生成唯一时间戳防止缓存
  const audio = document.getElementById("dynamicAudio");

  // 更新音频源(添加时间戳参数)
  audioSource.src = `./static/audio.wav?t=${timestamp}`;

  // 重新加载并播放(音量设为40%)
  audio.volume = 0.4;
  audio.load();
  audio.play();

  console.log(`音频已更新: ${timestamp}`);
}

// stable diffusion 服务跳转功能
document.getElementById("sd-service-btn").addEventListener("click", () => {
  window.open("http://localhost:7860", "_blank");
});

图像相似度检测

使用图片哈希值判断图像是否重复

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from PIL import Image
from io import BytesIO
import imagehash


def calculate_image_hash(image_data: bytes) -> str:
    """
    计算图像的感知哈希值
    适用于重复图片检测
    """
    # 从二进制数据创建图像对象
    img = Image.open(BytesIO(image_data))

    # 生成平均哈希(其他选项:phash, dhash, whash)
    img_hash = imagehash.average_hash(img)

    return str(img_hash)


# 使用示例
with open("image.jpg", "rb") as f:
    image_data = f.read()

hash_value = calculate_image_hash(image_data)
print(f"图片哈希值: {hash_value}")

# 相似度比较(汉明距离)
hash1 = imagehash.hex_to_hash("abcdef123456")
hash2 = imagehash.hex_to_hash("abcdff123456")
similarity_score = 1 - (hash1 - hash2) / len(hash1.hash) ** 2
print(f"图片相似度: {similarity_score:.2%}")

imagehash 中的哈希算法:

  • average_hash:计算速度快,适合一般场景
  • phash:感知哈希,对缩放旋转更鲁棒
  • dhash:差异哈希,适合检测内容变化
  • whash:小波哈希,保留更多频域信息

📋 使用 threading 下载图片

多线程获取图片直链 → parse_page() 多线程下载图片 → download_image() 线程安全:使用 threading.Lock() 防止并发写入冲突

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
import requests
from bs4 import BeautifulSoup
import threading
import json
import pathlib
import queue
import os
import re
import time

# 创建线程锁(用于多线程同步)
lock = threading.Lock()

# 存储图片直链的队列
links_queue = queue.Queue()

# 请求头
headers = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/114.0.0.0 Safari/537.36"
    )
}

# 参数配置
# 需要爬取的页数、每页图片数量、下载线程数
want_pagenum = 5
page_size = 10
num_threads = 5

# 待爬取的 API 地址列表
urls = [
    f"https://api.huaban.com/pins/recommend/winnow?"
    f"page_num={p}&page_size={page_size+1}&position=discovery_recommend"
    for p in range(1, want_pagenum + 1)
]


def parse_page() -> None:
    """
    从 API 中解析出图片的完整 URL 与标题,并放入 `links_queue`
    """
    while True:
        try:
            if not urls:
                print("直链获取完毕✔")
                break
            url = urls.pop(0)
            resp = requests.get(url, headers=headers, timeout=10)
            soup = BeautifulSoup(resp.text, "html.parser")
            data = json.loads(soup.get_text())

            base = "https://gd-hbimg.huaban.com/"
            img_urls = [base + pic["file"]["key"] for pic in data["pins"]]
            titles = [
                f"{pic['board']['title']}_{pic['file_id']}" for pic in data["pins"]
            ]

            for t, l in zip(titles, img_urls):
                links_queue.put((t, l))

            print(f"【当前线程】{threading.current_thread().name}")
        except Exception as e:
            print("解析错误 →", e)


def download_image() -> None:
    """
    从 `links_queue` 取出图片链接并保存到本地
    """
    while True:
        try:
            # 取出任务时加锁,防止空队列导致阻塞
            with lock:
                file_name, link = links_queue.get(timeout=2)
            resp = requests.get(link, timeout=10)

            # 替换文件名中的非法字符
            safe_name = re.sub(r"[/*]", "_", file_name).replace(" ", "")
            save_path = os.path.join(
                os.path.normpath("./pic_downloaded"),
                f"{safe_name}.png",
            )
            with open(save_path, "wb") as f:
                f.write(resp.content)

            print(
                f"{safe_name} 下载完成✔ "
                f"【当前线程】{threading.current_thread().name}"
            )
        except queue.Empty:
            break
        except Exception as e:
            print("下载错误 →", e)


# 主函数
def main() -> None:
    # ① 启动获取直链的线程
    get_links_thread = threading.Thread(name="get_links", target=parse_page)
    get_links_thread.start()

    # ② 启动图片下载线程
    download_threads = []
    for i in range(num_threads):
        t = threading.Thread(name=f"download_{i+1}", target=download_image)
        t.start()
        download_threads.append(t)

    # 等待全部线程结束
    get_links_thread.join()
    for t in download_threads:
        t.join()


if __name__ == "__main__":
    pathlib.Path("./pic_downloaded").mkdir(parents=True, exist_ok=True)
    start = time.time()
    main()
    print(f"总耗时:{time.time() - start:.2f}s")
    # 30.6689453125s
    # 39.74273657798767s

多线程适用于 I/O 密集型 任务(如图片下载),对 CPU 负载影响小
线程数建议设为 CPU 核数的 2–4 倍,避免资源争抢

通过 you-get 实现视频爬取

环境准备 pip install you-get PyQt5

目录结构:

1
2
3
4
youget_gui/
│─ yougetdownload.py          # you-get 封装类
│─ yougetdownload_pyqt.py     # GUI 主窗口
│─ main.py                    # 程序入口
  • yougetdownload.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import subprocess
import re
from PyQt5.QtCore import QObject, pyqtSignal


class YouGetDownload(QObject):
    """封装 you-get 命令行,提供进度信号。"""

    progress = pyqtSignal(int)  # 下载进度(%)
    finished = pyqtSignal(str)  # 完成或错误信息
    format_error = pyqtSignal(str)  # 格式错误

    def __init__(
        self,
        url: str,
        needformat: str = None,
        filename: str = None,
        dirname: str = None,
    ) -> None:
        super().__init__()
        self.url = url
        self.needformat = needformat
        self.filename = filename or ""
        self.dirname = dirname or "."
        self.output = ""

    def get_from_url(self) -> None:
        """
        获取视频信息(you-get -i)
        """
        cmd = ["you-get", "-i", self.url]
        self.output = subprocess.check_output(cmd, encoding="utf-8", errors="ignore")

    def download_url(self) -> None:
        """
        执行下载并实时发送进度信号
        """
        if self.needformat not in self.formats():
            self.format_error.emit("格式不正确")
            return

        cmd = [
            "you-get",
            f"--format={self.needformat}",
            self.url,
            "-O",
            self.filename,
            "-o",
            self.dirname,
            "--debug",
        ]
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, universal_newlines=True, encoding="utf-8"
        )

        last = -1
        while True:
            line = proc.stdout.readline()
            if line == "" and proc.poll() is not None:
                break
            if "%" in line:
                m = re.search(r"(\d+(?:\.\d+)?)%", line)
                if m:
                    cur = round(float(m.group(1)))
                    if cur != last:
                        last = cur
                        self.progress.emit(cur)

        proc.wait()
        if last == -1:
            self.finished.emit("出现问题,请检查文件名是否含有特殊字符")
        else:
            self.finished.emit("下载完成")


if __name__ == '__main__':
    downloader = YouGetDownload(url='https://www.bilibili.com/video/BV1Th4y1V7hh/?spm_id_from=333.1007.tianma.1-1-1.click')
    downloader.get_from_url()
    downloader.download_url()

formats() 方法需自行实现返回支持的格式列表(上述代码已省略)
通过 pyqtSignal 将进度实时推送到 UI,符合 PyQt 多线程安全的最佳实践

  • yougetdownload_pyqt.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
import sys
from PyQt5.QtWidgets import (
    QApplication,
    QMainWindow,
    QWidget,
    QVBoxLayout,
    QLabel,
    QLineEdit,
    QPushButton,
    QTextEdit,
    QComboBox,
    QProgressBar,
)
from PyQt5.QtCore import QTimer, QThread
from yougetdownload import YouGetDownload


class DownloadThread(QThread):
    """将下载函数封装成 QThread 类, 即放入子线程, 防止界面卡顿"""

    def __init__(self, url, fmt, name, path):
        super().__init__()
        self.downloader = YouGetDownload(url, fmt, name, path)

    def run(self):
        self.downloader.download_url()


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("基于 you-get 的爬取程序")
        self.timer = QTimer(singleShot=True)
        self.timer.timeout.connect(self._fetch_info)

        # 初始化子线程(占位)
        self.download_thread = DownloadThread("", None, None, None)

        # 绑定信号
        self.download_thread.downloader.progress.connect(self._set_progress)
        self.download_thread.downloader.finished.connect(self._log_finished)
        self.download_thread.downloader.format_error.connect(self._log_error)

        # 设置界面 UI
        self._setup_ui()

    # -----------------------------------------------------------------
    def _setup_ui(self):
        central = QWidget(self)
        # 设置居中
        self.setCentralWidget(central)
        # 盒子布局
        layout = QVBoxLayout(central)

        # URL 输入
        layout.addWidget(QLabel("链接:"))
        self.url_input = QLineEdit()
        # 当文本框中的内容变化后,开始计时
        self.url_input.textChanged.connect(lambda: self.timer.start(200))
        layout.addWidget(self.url_input)

        # 网站类型、标题
        self.site_label = QLabel("网站类型: -")
        self.title_label = QLabel("标题: -")
        layout.addWidget(self.site_label)
        layout.addWidget(self.title_label)

        # 格式选择
        layout.addWidget(QLabel("质量:"))
        self.format_box = QComboBox()
        layout.addWidget(self.format_box)

        # 文件名、路径
        layout.addWidget(QLabel("文件名:"))
        self.filename_input = QLineEdit()
        layout.addWidget(self.filename_input)

        layout.addWidget(QLabel("文件路径:"))
        self.dirname_input = QLineEdit()
        layout.addWidget(self.dirname_input)

        # 下载按钮
        btn = QPushButton("下载")
        btn.clicked.connect(self._start_download)
        layout.addWidget(btn)

        # 进度条
        self.progress = QProgressBar()
        self.progress.setRange(0, 100)
        layout.addWidget(self.progress)

        # 日志输出
        self.log = QTextEdit()
        self.log.setReadOnly(True)
        layout.addWidget(self.log)

    # -----------------------------------------------------------------
    def _fetch_info(self):
        """在用户输入 URL 稍后获取视频信息并更新 UI。"""
        self.download_thread.downloader.url = self.url_input.text()
        self.download_thread.downloader.get_from_url()
        info = self.download_thread.downloader.download_info()  # 需自行实现

        self.site_label.setText(f"网站类型: {info['site']}")
        self.title_label.setText(f"标题: {info['title']}")
        self.format_box.clear()
        self.format_box.addItems(info["formats"].keys())
        self.filename_input.setText(info["title"])

    # -----------------------------------------------------------------
    def _start_download(self):
        url = self.url_input.text()
        fmt = self.format_box.currentText()
        name = self.filename_input.text()
        path = self.dirname_input.text() or "."

        # 重新实例化子线程, 确保参数最新
        self.download_thread = DownloadThread(url, fmt, name, path)
        self.download_thread.downloader.progress.connect(self._set_progress)
        self.download_thread.downloader.finished.connect(self._log_finished)
        self.download_thread.downloader.format_error.connect(self._log_error)
        self.log.clear()
        self.download_thread.start()

    # -----------------------------------------------------------------
    def _set_progress(self, value: int):
        self.progress.setValue(value)

    def _log_finished(self, msg: str):
        self.log.append(msg)

    def _log_error(self, msg: str):
        self.log.append(msg)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    win = MainWindow()
    win.show()
    sys.exit(app.exec_())

数据分析实战(涵少)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.express as px

# 机器学习相关库
from sklearn.tree import DecisionTreeClassifier
from sklearn.neighbors import KNeighborsClassifier
from sklearn.svm import SVC
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_validate, KFold, train_test_split, cross_val_score
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, auc, roc_auc_score, roc_curve
from sklearn.feature_selection import f_classif
from sklearn.compose import make_column_selector, make_column_transformer, ColumnTransformer
from sklearn.metrics import classification_report, confusion_matrix

探索性分析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# 加载数据
data = pd.read_csv('222(1).csv')

# 数据初步探索
print("数据形状:", data.shape)
print("\n前5行数据:")
print(data.head())
print("\n数据基本信息:")
print(data.info())
print("\n列名:")
print(data.columns)

将数据字段分为数值类型和类别类型

1
2
3
4
5
6
7
8
# 数值型变量
num_col = ['age', 'height', 'weight', 'ap_hi', 'ap_lo']

# 类别型变量
cate_col = ['gender', 'cholesterol', 'gluc', 'smoke', 'alco', 'active']

# 目标变量
target_col = 'cardio'

绘制 类别型变量目标变量 的交叉表

1
2
3
4
5
for col in cate_col:
    cross_table = pd.crosstab(data[col], data[target_col], normalize='index')
    print(f"\n{col} 与目标变量的交叉表:")
    print(cross_table)
    print("-" * 50)

数值型变量目标变量 的相关性–方差分析

1
2
3
4
5
6
7
8
9
X = data[num_col]
y = data[target_col]

f_scores, p_values = f_classif(X, y)
print("方差分析结果:")
for i, col in enumerate(num_col):
    print(f"{col}: F值={f_scores[i]:.2f}, P值={p_values[i]:.4f}")

# 解释:P值小于0.05说明变量与目标变量显著相关

数值型变量 之间的相关性

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
# 绘制相关性热力图
plt.figure(figsize=(12, 8))

# 创建一个上三角矩阵掩码,用于在绘制相关性热力图时隐藏上三角部分(包括对角线),只显示下三角
mask = np.triu(np.ones_like(data[num_col].corr(), dtype=bool))

sns.heatmap(data[num_col].corr(),
            mask=mask,
            annot=True,
            cmap='coolwarm',
            center=0,
            square=True)
plt.title('数值型变量相关性热力图')
plt.tight_layout()
plt.show()

数据分布检查
查看数据是否平衡,通过绘制直方图展示

1
2
3
4
5
6
7
8
# 检查目标变量是否平衡
plt.figure(figsize=(8, 6))
data[target_col].value_counts().plot(kind='bar')
plt.title('目标变量分布')
plt.xlabel('cardio')
plt.ylabel('数量')
plt.xticks(rotation=0)
plt.show()
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# 检查类别型变量分布
fig, axes = plt.subplots(2, 3, figsize=(15, 10))
axes = axes.ravel()

for i, col in enumerate(cate_col):
    data[col].value_counts().plot(kind='bar', ax=axes[i])
    axes[i].set_title(f'{col} 分布')
    axes[i].tick_params(axis='x', rotation=45)

plt.tight_layout()
plt.show()
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# 检查数值型变量分布(箱线图)
plt.figure(figsize=(15, 10))
for i, col in enumerate(num_col, 1):
plt.subplot(2, 3, i)
plt.boxplot(data[col].dropna())
plt.title(f'{col} 箱线图')
plt.ylabel(col)

plt.tight_layout()
plt.show()

数据预处理与特征工程

删除相关性不高的性别和稀疏特征 alco

1
2
3
4
5
data.drop(['gender', 'alco'], axis=1, inplace=True)

# 更新类别型变量列表
cate_col = [col for col in cate_col if col not in ['gender', 'alco']]
print("更新后的类别型变量:", cate_col)

异常值处理, 使用 3σ 原则处理异常值(盖帽法)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
for col in ['height', 'weight', 'ap_hi', 'ap_lo']:
    print(f'正在处理 {col} 字段的异常值')

    upper_bound = data[col].mean() + 3 * data[col].std()
    lower_bound = data[col].mean() - 3 * data[col].std()

    # 盖帽法处理异常值
    data.loc[data[col] > upper_bound, col] = upper_bound
    data.loc[data[col] < lower_bound, col] = lower_bound

    print(f'{col} 处理完成: 上限={upper_bound:.2f}, 下限={lower_bound:.2f}')

独热编码和数据标准化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 创建列转换器
ct = ColumnTransformer([
    ('num', StandardScaler(), num_col),
    ('cat', OneHotEncoder(), cate_col)
])

# 应用转换
train_data_new = ct.fit_transform(data.drop(target_col, axis=1))

# 获取特征名称
if hasattr(ct, "get_feature_names_out"):
    feature_names = ct.get_feature_names_out()
else:
    # 兼容旧版本
    num_features = num_col
    cat_features = ct.named_transformers_['cat'].get_feature_names_out(cate_col)
    feature_names = np.concatenate([num_features, cat_features])

# 创建转换后的DataFrame
train_data_new = pd.DataFrame(train_data_new, columns=feature_names)
print("转换后的数据形状:", train_data_new.shape)
print("特征名称:", feature_names)

模型建立

划分训练集和测试集

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
X = train_data_new
y = data[target_col]

X_train, X_test, y_train, y_test = train_test_split(
    X, y,
    test_size=0.2,
    random_state=42,
    stratify=y  # 分层抽样确保类别比例
)

print(f"训练集形状: {X_train.shape}")
print(f"测试集形状: {X_test.shape}")

选择算法模型

  • 基础决策树
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# 初始化决策树模型
dt = DecisionTreeClassifier(random_state=42)

# 训练模型
dt.fit(X_train, y_train)

# 预测
dt_pred = dt.predict(X_test)

# 评估
accuracy = accuracy_score(y_test, dt_pred)
print(f"测试集准确率: {accuracy:.2%}")

# 查看默认参数
print("决策树默认参数:")
print(dt.get_params())
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# 超参数优化(随机搜索)
from sklearn.model_selection import RandomizedSearchCV

# 定义参数空间
param_distributions = {
    'max_depth': range(2, 8),
    'max_features': range(2, min(10, X_train.shape[1])),
    'min_samples_leaf': range(2, 10),
    'min_samples_split': range(2, 10)
}

# 随机搜索
random_search = RandomizedSearchCV(
    estimator=dt,
    param_distributions=param_distributions,
    n_iter=100,
    scoring='f1',
    cv=5,
    random_state=42,
    n_jobs=-1
)

random_search.fit(X_train, y_train)

# 输出最优参数
print("最优超参数:", random_search.best_params_)
print("最佳交叉验证F1分数: {:.2%}".format(random_search.best_score_))

# # 网格搜索
# from sklearn.model_selection import GridSearchCV
# grid_search = GridSearchCV(
#     estimator=dt,
#     param_grid=param_grid,
#     scoring='f1',
#     cv=5
# )
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# 使用最优参数训练新模型
dt_optimized = DecisionTreeClassifier(**random_search.best_params_, random_state=42)
dt_optimized.fit(X_train, y_train)

# 预测
dt_pred_optimized = dt_optimized.predict(X_test)

# 评估
accuracy_optimized = accuracy_score(y_test, dt_pred_optimized)
print(f"优化后测试集准确率: {accuracy_optimized:.2%}")
  • 遗传算法优化(可选)
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
# 注意:遗传算法需要安装 deap 库: pip install deap
try:
    from deap import creator, base, tools, algorithms
    import random

    # 定义适应度函数
    def fitness_function(individual):
        clf = DecisionTreeClassifier(
            max_depth=individual[0],
            max_features=individual[1],
            min_samples_leaf=individual[2],
            min_samples_split=individual[3],
            random_state=42
        )
        score = np.mean(cross_val_score(clf, X_train, y_train, cv=3, scoring='f1'))
        return score,

    # 遗传算法优化(代码较长,实际使用时建议单独运行)
    print("遗传算法优化功能已就绪")

except ImportError:
    print("deap 库未安装,遗传算法优化不可用")

# 定义参数范围
parameter_range = {
    # 'max_depth': list(range(2, 8)),
    # 'max_features': list(range(2, 10)),
    # 'min_samples_leaf': list(range(2, 10)),
    # 'min_samples_split': list(range(2, 10))
    'max_depth': [2, 7],
    'max_features': [2, 9],
    'min_samples_leaf': [2, 9],
    'min_samples_split': [2, 9]
}

# 定义遗传算法参数
toolbox = base.Toolbox()
creator.create("FitnessMax", base.Fitness, weights=(1.0,))    #创建FitnessMax来扩展Fitness类
creator.create("Individual", list, fitness=creator.FitnessMax)
toolbox.register("attr_int", random.randint)
toolbox.register("individual", tools.initCycle, creator.Individual,
                 [lambda params=values:toolbox.attr_int(*params) for _, values in parameter_range.items()], n=1)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)

toolbox.register("evaluate", fitness_function)
toolbox.register("mate", tools.cxUniform, indpb=0.5)
toolbox.register("mutate", tools.mutUniformInt, low=2, up=9, indpb=0.2)
toolbox.register("select", tools.selTournament, tournsize=3)

pop = toolbox.population(n=50)
hof = tools.HallOfFame(1)
stats = tools.Statistics(lambda ind: ind.fitness.values)
stats.register("avg", np.mean)
stats.register("max", np.max)

# 运行遗传算法优化参数
pop, log = algorithms.eaSimple(pop, toolbox, cxpb=0.5, mutpb=0.2, ngen=20,
                               stats=stats, halloffame=hof, verbose=True)

# 输出最优参数和模型评分
print("Best parameters: ", hof[0])
print("Train f1 score: ", hof[0].fitness.values[0])
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
# 可视化决策树
plt.figure(figsize=(20, 12))
plot_tree(
    dt_optimized,
    feature_names=feature_names,
    class_names=['0', '1'],
    filled=True,
    rounded=True,
    max_depth=3,  # 限制深度以便查看
    proportion=True
)
plt.title('决策树结构(前3层)')
plt.tight_layout()
plt.savefig("decision_tree.png", dpi=300, bbox_inches='tight')
plt.show()
  • 随机森林
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import RandomizedSearchCV


# 定义参数网格
param_grid_rf = {
    'max_depth': range(2, 8),
    'min_samples_leaf': range(2, 10),
    'min_samples_split': range(2, 10),
    'n_estimators': range(100, 600, 100)
}

# 初始化随机森林
rf = RandomForestClassifier(random_state=42, n_jobs=-1)

# 随机搜索优化
grid_search_rf = RandomizedSearchCV(
    estimator=rf,
    param_distributions=param_grid_rf,
    scoring='f1',
    cv=3,
    n_iter=50,
    verbose=1,
    random_state=42,
    n_jobs=-1
)

grid_search_rf.fit(X_train, y_train)

# 最佳模型
best_rf_model = grid_search_rf.best_estimator_
print("随机森林最佳参数:", grid_search_rf.best_params_)

# 预测
rf_pred = best_rf_model.predict(X_test)
rf_pred_proba = best_rf_model.predict_proba(X_test)[:, 1]
  • SVM 模型
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
svm_model = SVC(
    probability=True,
    C=0.1,
    random_state=42,
    kernel='rbf'  # 默认使用RBF核
)

svm_model.fit(X_train, y_train)

# 预测
svm_pred = svm_model.predict(X_test)
svm_pred_proba = svm_model.predict_proba(X_test)[:, 1]

print("SVM 模型训练完成")

SVM 的随机搜索最佳参数十分耗时

  • KNN 模型(K 近邻模型)
1
2
3
4
5
6
7
8
knn_model = KNeighborsClassifier(n_neighbors=5)
knn_model.fit(X_train, y_train)

# 预测
knn_pred = knn_model.predict(X_test)
knn_pred_proba = knn_model.predict_proba(X_test)[:, 1]

print("KNN 模型训练完成")

模型评估与比较

分类报告

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
# 决策树分类报告
print("=" * 50)
print("决策树模型分类报告:")
print("=" * 50)
print(classification_report(y_test, dt_pred_optimized))

# 随机森林分类报告
print("=" * 50)
print("随机森林模型分类报告:")
print("=" * 50)
print(classification_report(y_test, rf_pred))

# SVM分类报告
print("=" * 50)
print("SVM模型分类报告:")
print("=" * 50)
print(classification_report(y_test, svm_pred))

绘制混淆矩阵

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
##混淆矩阵
import itertools
from sklearn.metrics import confusion_matrix
plt.rcParams['font.family'] = ['sans-serif']
plt.rcParams['font.sans-serif'] = ['SimHei']

def plot_confusion_matrix(cm, classes, title='Confusion Matrix', cmap=plt.cm.Blues):
    """
    绘制混淆矩阵
    """
    plt.imshow(cm, interpolation='nearest', cmap=cmap)
    plt.title(title, fontsize=14)
    plt.colorbar()
    tick_marks = np.arange(len(classes))
    plt.xticks(tick_marks, classes, rotation=0)
    plt.yticks(tick_marks, classes)

    # 添加数值标签
    thresh = cm.max() / 2.
    for i, j in np.ndindex(cm.shape):
        plt.text(j, i, format(cm[i, j], 'd'),
                 horizontalalignment="center",
                 color="white" if cm[i, j] > thresh else "black")

    plt.ylabel('True Label', fontsize=12)
    plt.xlabel('Predicted Label', fontsize=12)
    plt.tight_layout()

# 绘制各个模型的混淆矩阵
models = {
    '决策树': dt_pred_optimized,
    '随机森林': rf_pred,
    'SVM': svm_pred
}

plt.figure(figsize=(15, 5))
for i, (name, pred) in enumerate(models.items(), 1):
    plt.subplot(1, 3, i)
    cm = confusion_matrix(y_test, pred)
    plot_confusion_matrix(cm, classes=['0', '1'], title=f'{name} - 混淆矩阵')

plt.tight_layout()
plt.show()

绘制 ROC 曲线

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 收集各模型的预测概率
model_probas = {
    '决策树': dt_optimized.predict_proba(X_test)[:, 1],
    '随机森林': rf_pred_proba,
    'SVM': svm_pred_proba,
    'KNN': knn_pred_proba
}

# 绘制ROC曲线
plt.figure(figsize=(12, 8))

for name, proba in model_probas.items():
    fpr, tpr, _ = roc_curve(y_test, proba)
    roc_auc = roc_auc_score(y_test, proba)
    plt.plot(fpr, tpr, label=f'{name} (AUC = {roc_auc:.3f})', linewidth=2)

# 绘制随机猜测线
plt.plot([0, 1], [0, 1], 'k--', linewidth=2, label='随机猜测 (AUC = 0.500)')

plt.xlabel('假正率 (False Positive Rate)', fontsize=12)
plt.ylabel('真正率 (True Positive Rate)', fontsize=12)
plt.title('各模型ROC曲线对比', fontsize=14)
plt.legend(loc='lower right', fontsize=11)
plt.grid(True, alpha=0.3)
plt.xlim([0.0, 1.0])
plt.ylim([0.0, 1.05])
plt.tight_layout()
plt.show()
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# 计算各模型的主要指标
models_performance = []

for name, pred in models.items():
    accuracy = accuracy_score(y_test, pred)
    precision = precision_score(y_test, pred)
    recall = recall_score(y_test, pred)
    f1 = f1_score(y_test, pred)
    roc_auc = roc_auc_score(y_test, model_probas[name])

    models_performance.append({
        'Model': name,
        'Accuracy': accuracy,
        'Precision': precision,
        'Recall': recall,
        'F1-Score': f1,
        'ROC-AUC': roc_auc
    })

# 创建性能比较表格
performance_df = pd.DataFrame(models_performance)
performance_df.set_index('Model', inplace=True)

print("各模型性能比较:")
print("=" * 80)
print(performance_df.round(4))

# 可视化性能比较
performance_df.plot(kind='bar', figsize=(14, 8), colormap='Set3')
plt.title('各模型性能指标对比', fontsize=14)
plt.ylabel('分数', fontsize=12)
plt.xticks(rotation=45)
plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
plt.tight_layout()
plt.show()
  1. 数据预处理:成功处理了异常值,进行了特征编码和标准化
  2. 特征工程:移除了不相关的特征,保留了有预测能力的变量
  3. 模型比较:多个模型都取得了较好的性能,可根据具体需求选择:
    • 如果需要可解释性:选择决策树
    • 如果需要最高精度:选择随机森林
    • 如果数据量很大:考虑 SVM(但需要调参)
  4. 后续优化方向
    • 尝试更多的特征工程方法
    • 使用更复杂的集成方法
    • 进行更细致的超参数调优
    • 考虑类别不平衡问题(如有需要)