数学重学 - 31 数学建模思维

这是 数学重学路线图 阶段六的子页面

📌 建模 = 把现实问题翻译成数学语言,求解后再翻译回来

📌 这一页是”工具箱”思维:不求精通每个工具,而是知道什么场景用什么工具


什么是数学建模

直觉

比喻:你要从北京去上海,数学建模就是先画地图(建模)→ 找最短路(求解)→ 实际开车验证(验证)

现实世界太复杂 → 提取关键要素 → 用数学描述 → 求解 → 验证 → 调整

一句话

数学语言描述现实问题 → 用数学方法求解 → 把结果翻译回现实

为什么程序员需要建模思维

不是每个问题都有现成的库可以调

性能优化、容量规划、风险评估都需要建模

建模能力 = 把模糊问题变成可解问题的能力


建模流程

六步法

1. 问题定义(最难的一步!)

“到底要解决什么?”

比喻:病人说”我不舒服”,医生要确定”到底哪里不舒服”

常见错误:问题太大太模糊 → 需要拆解

例:不是”优化系统性能”,而是”在高峰期将API P99延迟降到200ms以下”

2. 假设简化

忽略次要因素,让问题变得可解

例:排队模型假设”请求到达服从泊松分布”(现实中近似成立)

关键:假设要明确列出,后面可以逐步放松

3. 选择模型

从工具箱中选合适的数学工具

不一定要最复杂的,够用就好

“所有模型都是错的,但有些是有用的” —— George Box

4. 求解

解方程、跑仿真、做优化

这一步通常最”数学”,但其实现代工具(Python/scipy)能帮你完成大部分

5. 验证

预测 vs 现实对比

模型在什么范围内准确?什么时候会失效?

使用历史数据回测

6. 应用/迭代

模型上线 → 收集反馈 → 发现不准 → 回到步骤2调整假设

建模是一个迭代过程,不是一锤子买卖


常见建模工具箱

速查表

| 工具 | 适用场景 | 核心思想 |
| 线性模型 | 趋势、简单关系 | y = ax + b |
| 概率模型 | 不确定性、风险 | P(事件) |
| 图模型 | 网络、依赖关系 | 节点+边 |
| 优化模型 | 资源分配、调度 | min/max 目标函数 |
| 排队论 | 等待时间、容量 | 到达率 vs 服务率 |
| 微分方程 | 动态变化过程 | dx/dt = f(x) |
| 马尔可夫链 | 状态转移、序列 | 下一步只看当前 |

选择原则

先简单后复杂:能用线性回归解决的不要上深度学习

先理解后建模:不理解问题就瞎套模型 = 垃圾进垃圾出

先验证后信任:任何模型在上线前都需要验证


案例1:排队论 —— 系统容量规划

直觉

比喻:超市收银台前排队

顾客来得快(到达率高)+ 收银员慢(服务率低)→ 队伍越排越长

到达率 > 服务率 → 队伍无限增长 → 系统崩溃

M/M/1 队列模型

M = Markov(无记忆/指数分布),1 = 单服务台

到达率 λ:平均每秒来多少请求(泊松过程)

服务率 μ:平均每秒处理多少请求(指数分布服务时间)

核心公式

利用率 ρ = λ/μ(必须 < 1,否则系统永远处理不完)

平均队列长度 L = ρ/(1-ρ)

平均等待时间 W = 1/(μ-λ)

系统中平均请求数 Ls = λ/(μ-λ)

关键洞察:非线性暴涨

| 利用率 ρ | 平均队列长度 L |
| 50% | 1 |
| 70% | 2.33 |
| 80% | 4 |
| 90% | 9 |
| 95% | 19 |
| 99% | 99 |

从 80% 到 90%:队列长度翻了一倍多

从 90% 到 99%:队列长度翻了 10 倍!

这就是为什么系统 80% 负载就该扩容 —— 不是线性增长,是指数级恶化

应用

线程池大小:线程池 = 服务台数量,请求 = 顾客

数据库连接池:连接数 = 服务能力,查询 = 到达请求

消息队列消费者数:消费者太少 → 消息堆积 → 延迟暴涨

负载均衡:多台服务器 = M/M/c 模型(多服务台)


案例2:SIR 模型 —— 蠕虫/病毒传播

直觉

传染病怎么传播?计算机蠕虫的传播规律惊人地相似

SIR 三个状态

**S (Susceptible)**:易感者(没打补丁的机器)

**I (Infected)**:感染者(被感染的机器)

**R (Recovered)**:恢复者/免疫者(打了补丁的机器)

微分方程

dS/dt = -β × S × I (易感者被感染的速率)

dI/dt = β × S × I - γ × I (新增感染 - 恢复)

dR/dt = γ × I (恢复速率)

其中:

β = 感染率(一个感染者每次接触传播的概率)

γ = 恢复率(感染者恢复的速率,1/γ = 平均感染持续时间)

R₀ 基本传染数

R₀ = β/γ = 一个感染者平均能传染多少人

R₀ > 1:疫情爆发(指数增长阶段)

R₀ < 1:疫情衰退

R₀ = 1:边界状态

例子:

流感 R₀ ≈ 1.5

COVID 原始株 R₀ ≈ 2.5

2003 SQL Slammer 蠕虫:10分钟感染了全球 75000 台服务器

安全防御的对应关系

| 生物世界 | 计算机安全 |
| 疫苗 | 安全补丁 |
| 隔离 | 防火墙/网络隔离 |
| 口罩 | 流量过滤/IDS |
| 群体免疫 | 大规模补丁部署 |
| 变异 | 漏洞利用变种 |

降低 R₀ 的策略

降低 β(感染率):网络分段、访问控制、最小权限

提高 γ(恢复率):快速检测+自动修复

减少 S(易感者):提前打补丁、安全加固


案例3:马尔可夫链 —— 状态转移

直觉

比喻:天气预报简化版 —— 今天晴天,明天有 70% 概率还是晴天,30% 概率下雨

关键:明天的天气只取决于今天,跟昨天无关(无记忆性/马尔可夫性)

定义

马尔可夫性:P(X_{n+1} | X_n, X_{n-1}, …) = P(X_{n+1} | X_n)

下一状态只取决于当前状态,与历史无关

状态转移矩阵

例:天气模型(晴/雨两个状态)

1
2
3
   晴    雨
[ 0.7 0.3 ]
[ 0.4 0.6 ]

含义:

晴天→晴天: 70%, 晴天→雨天: 30%

雨天→晴天: 40%, 雨天→雨天: 60%

每行之和 = 1(从一个状态出发,总要到某个状态)

稳态分布

经过足够多步后,各状态概率趋于稳定

不管初始状态是什么,最终比例固定

解 πP = π(π 是稳态分布向量)

天气例子:稳态 → 57% 晴天、43% 雨天

应用

PageRank

网页 = 状态,链接 = 转移

“随机游走者”点击链接浏览网页

稳态分布 = 每个网页的”重要性”

Google 的核心算法就是这个

用户行为预测

状态:浏览→加购→下单→离开

转移概率从日志数据中统计

预测用户下一步行为

文本生成

状态 = 词,转移概率 = 词之后出现另一个词的概率

简单但有趣,ChatGPT 的远古祖先


案例4:背包问题 —— 资源分配

直觉

你要出差,行李箱只能装 20kg

每件东西有重量和价值,怎么选使总价值最大?

0/1 背包的数学描述

n 个物品,每个物品 i 有重量 w_i 和价值 v_i

背包容量 W

决策变量 x_i ∈ {0, 1}(带或不带)

目标:max Σ(v_i × x_i)

约束:Σ(w_i × x_i) ≤ W

动态规划解法思路

定义 dp[i][j] = 前 i 个物品、容量为 j 时的最大价值

转移方程:

不选物品i:dp[i][j] = dp[i-1][j]

选物品i(如果 j >= w_i):dp[i][j] = dp[i-1][j-w_i] + v_i

取较大值

时间复杂度 O(n × W),空间可优化到 O(W)

安全预算分配

安全措施 = 物品

成本 = 重量

风险降低量 = 价值

总预算 = 背包容量

例:

WAF: 成本 50万, 风险降低 30%

安全培训: 成本 10万, 风险降低 15%

渗透测试: 成本 20万, 风险降低 20%

SOC: 成本 80万, 风险降低 40%

预算 100万,怎么选?

其他应用

服务器采购:不同配置不同价格不同性能,预算有限

广告投放:不同渠道不同成本不同转化率,预算有限

功能排期:不同功能不同开发时间不同商业价值,人力有限


应用汇总

安全领域

威胁建模:STRIDE 模型(Spoofing, Tampering, Repudiation, Information Disclosure, DoS, Elevation)

攻击树:根节点是攻击目标,叶子节点是具体攻击手段,树结构表示攻击路径

风险评估量化:风险 = 概率 × 影响

概率来自历史数据 + 专家判断

影响用金额量化(数据泄露平均损失 445 万美元 —— IBM 2024 报告)

安全投资优化:用背包模型分配安全预算

大数据领域

用户流失预测:马尔可夫链建模用户状态转移

推荐系统建模:探索-利用权衡(30-博弈论入门 中提到的 MAB)

流量预测:时间序列模型,排队论做容量规划

A/B 测试:假设检验决定哪个版本更好

后端领域

容量规划:排队论 M/M/1 / M/M/c 模型

性能建模:利特尔定律 L = λW(系统中请求数 = 到达率 × 平均逗留时间)

负载均衡策略:随机、轮询、最少连接 —— 不同模型不同最优

超时设置:排队论告诉你合理的超时时间


Python 代码实战

排队论 M/M/1 模拟

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
import random
import math

def mm1_simulation(lam, mu, n_requests=10000):
"""
M/M/1 排队模型蒙特卡洛模拟
lam: 到达率 (请求/秒)
mu: 服务率 (处理/秒)
"""
rho = lam / mu
print(f"=== M/M/1 模拟 ===")
print(f"到达率 λ={lam}, 服务率 μ={mu}, 利用率 ρ={rho:.2f}")

if rho >= 1:
print("警告: ρ >= 1, 系统不稳定, 队列将无限增长!")
return

# 理论值
L_theory = rho / (1 - rho) # 平均队列长度
W_theory = 1 / (mu - lam) # 平均等待时间
Ls_theory = lam / (mu - lam) # 系统中平均请求数

# 模拟
clock = 0.0
queue = [] # 每个元素是 (到达时间,)
server_free_at = 0.0

wait_times = []
queue_lengths = []

for i in range(n_requests):
# 下一个请求到达(指数分布间隔)
inter_arrival = random.expovariate(lam)
clock += inter_arrival
arrival_time = clock

# 开始服务的时间
start_service = max(arrival_time, server_free_at)

# 服务时间(指数分布)
service_time = random.expovariate(mu)

# 完成时间
finish_time = start_service + service_time
server_free_at = finish_time

# 等待时间 = 开始服务 - 到达
wait = start_service - arrival_time
wait_times.append(wait)

# 记录当前 "在系统中" 的数量(简化)
queue_lengths.append(max(0, wait * lam))

avg_wait = sum(wait_times) / len(wait_times)

print(f"\n理论值:")
print(f" 平均等待时间 W = {W_theory:.4f}s")
print(f" 平均队列长度 L = {L_theory:.2f}")

print(f"\n模拟值 ({n_requests} 个请求):")
print(f" 平均等待时间 = {avg_wait:.4f}s")

# 等待时间分布(文本直方图)
print(f"\n等待时间分布:")
max_wait = max(wait_times)
n_bins = 10
bin_width = max_wait / n_bins if max_wait > 0 else 1
bins = [0] * n_bins
for w in wait_times:
idx = min(int(w / bin_width), n_bins - 1)
bins[idx] += 1

for i, count in enumerate(bins):
bar = '█' * (count * 40 // max(bins))
lo = i * bin_width
hi = (i + 1) * bin_width
print(f" {lo:6.3f}-{hi:6.3f}s |{bar} ({count})")

# 演示不同负载下的表现
mm1_simulation(lam=8, mu=10) # ρ=0.8, 还行
print()
mm1_simulation(lam=9.5, mu=10) # ρ=0.95, 快崩了

SIR 模型微分方程求解

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
from scipy.integrate import odeint
import numpy as np

def sir_model(y, t, beta, gamma):
"""SIR 微分方程"""
S, I, R = y
dSdt = -beta * S * I
dIdt = beta * S * I - gamma * I
dRdt = gamma * I
return [dSdt, dIdt, dRdt]

def simulate_sir(beta=0.3, gamma=0.1, I0=0.01, days=160):
"""
模拟 SIR 传播模型
beta: 感染率
gamma: 恢复率
I0: 初始感染比例
"""
R0 = beta / gamma
print(f"=== SIR 传播模型 ===")
print(f"感染率 β={beta}, 恢复率 γ={gamma}")
print(f"基本传染数 R₀ = β/γ = {R0:.2f}")
print(f"R₀ {'>' if R0 > 1 else '<'} 1 → {'会爆发' if R0 > 1 else '会衰退'}")

# 初始条件
S0 = 1.0 - I0
R0_init = 0.0
y0 = [S0, I0, R0_init]

# 时间点
t = np.linspace(0, days, days * 10)

# 求解 ODE
solution = odeint(sir_model, y0, t, args=(beta, gamma))
S, I, R = solution.T

# 找感染峰值
peak_idx = np.argmax(I)
peak_day = t[peak_idx]
peak_infected = I[peak_idx]

print(f"\n感染峰值: 第 {peak_day:.0f} 天, 感染比例 {peak_infected:.2%}")
print(f"最终未感染比例: {S[-1]:.2%}")
print(f"最终恢复比例: {R[-1]:.2%}")

# 文本可视化(每10天一个数据点)
print(f"\n{'日':>4} | {'S(易感)':>8} | {'I(感染)':>8} | {'R(恢复)':>8} | 感染曲线")
print("-" * 65)
for day in range(0, days, 5):
idx = day * 10
if idx < len(t):
s, i, r = S[idx], I[idx], R[idx]
bar = '█' * int(i * 50)
print(f"{day:4d} | {s:7.2%} | {i:7.2%} | {r:7.2%} | {bar}")

# 安全比喻
print(f"\n--- 安全启示 ---")
print(f"如果这是蠕虫传播 (R₀={R0:.1f}):")
print(f" - 第 {peak_day:.0f} 天达到感染峰值 ({peak_infected:.0%} 设备感染)")
print(f" - 必须在峰值前部署补丁(降低 β)或隔离(提高 γ)")
herd_immunity = 1 - 1/R0
print(f" - 群体免疫阈值: {herd_immunity:.0%} 设备需要打补丁")

simulate_sir(beta=0.3, gamma=0.1) # R0=3, 严重爆发
print()
simulate_sir(beta=0.15, gamma=0.1) # R0=1.5, 温和传播

马尔可夫链文本生成器

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
import random
from collections import defaultdict

def build_markov_chain(text, order=2):
"""
构建 n 阶马尔可夫链
order: 阶数(看前几个词预测下一个)
"""
words = text.split()
chain = defaultdict(list)

for i in range(len(words) - order):
state = tuple(words[i:i + order])
next_word = words[i + order]
chain[state].append(next_word)

return chain

def generate_text(chain, order=2, length=50):
"""用马尔可夫链生成文本"""
# 随机选一个起始状态
state = random.choice(list(chain.keys()))
result = list(state)

for _ in range(length):
if state not in chain:
break
next_word = random.choice(chain[state])
result.append(next_word)
state = tuple(result[-order:])

return ' '.join(result)

# 用一段技术文本做训练语料
corpus = """
the server receives a request and processes it
the server sends a response back to the client
the client sends a request to the server
the server checks the cache before querying the database
the database returns the result to the server
the server stores the result in the cache
the client receives the response and renders the page
the load balancer distributes requests to multiple servers
the server processes the request and returns a response
the cache stores frequently accessed data for fast retrieval
the database stores persistent data and handles queries
the server handles concurrent requests using a thread pool
the client sends multiple requests in parallel
the server returns an error if the request is invalid
the load balancer checks server health before routing
"""

chain = build_markov_chain(corpus, order=2)

print("=== 二阶马尔可夫链文本生成 ===\n")
print("训练语料: 后端技术描述文本\n")
print("生成结果(每次不同):")
for i in range(5):
text = generate_text(chain, order=2, length=15)
print(f" {i+1}. {text}")

print("\n--- 原理 ---")
print("每个词只看前2个词决定 → 马尔可夫性")
print("阶数越高 → 越像原文 → 但创造性越低")
print("阶数越低 → 越随机 → 可能语法不通")

简单 0/1 背包动态规划

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
def knapsack_01(items, capacity):
"""
0/1 背包问题 - 动态规划
items: [(名称, 重量/成本, 价值), ...]
capacity: 背包容量/总预算
返回: 最大价值, 选中的物品
"""
n = len(items)
# dp[j] = 容量为 j 时的最大价值
dp = [0] * (capacity + 1)
# 记录选择(用于回溯)
chosen = [[False] * (capacity + 1) for _ in range(n)]

for i in range(n):
name, weight, value = items[i]
# 从大到小遍历(0/1背包的关键!避免重复选择)
for j in range(capacity, weight - 1, -1):
if dp[j - weight] + value > dp[j]:
dp[j] = dp[j - weight] + value
chosen[i][j] = True

# 回溯找选了哪些物品
selected = []
j = capacity
for i in range(n - 1, -1, -1):
if chosen[i][j]:
selected.append(items[i])
j -= items[i][1]

return dp[capacity], selected

# === 安全预算分配示例 ===
print("=== 安全预算分配(0/1 背包问题)===\n")

# (安全措施, 成本(万), 风险降低分数)
security_items = [
("WAF 防火墙", 50, 30),
("安全培训", 10, 15),
("渗透测试", 20, 20),
("SOC 安全运营中心", 80, 40),
("代码审计工具", 30, 25),
("DDoS 防护", 40, 18),
("零信任网络", 60, 35),
("数据加密", 15, 12),
]

budget = 100 # 100万预算

print(f"总预算: {budget} 万")
print(f"\n可选安全措施:")
print(f"{'措施':<16} {'成本(万)':>8} {'风险降低':>8} {'性价比':>8}")
print("-" * 44)
for name, cost, value in security_items:
ratio = value / cost
print(f"{name:<16} {cost:>8} {value:>8} {ratio:>8.2f}")

max_value, selected = knapsack_01(security_items, budget)
total_cost = sum(item[1] for item in selected)

print(f"\n最优方案 (总预算 {budget} 万):")
print(f"{'='*40}")
for name, cost, value in selected:
print(f" ✓ {name}: {cost}万 → 风险降低 {value}")
print(f"{'='*40}")
print(f"总成本: {total_cost} 万")
print(f"总风险降低: {max_value}")
print(f"剩余预算: {budget - total_cost} 万")

练习题

题1:排队论计算

一个 API 服务每秒收到 100 个请求(λ=100),每个请求平均处理 8ms(μ=125/秒)

计算利用率 ρ、平均等待时间 W、平均队列长度 L

如果请求增加到 120/秒呢?
答案
ρ = 100/125 = 0.8

W = 1/(125-100) = 0.04s = 40ms

L = 0.8/(1-0.8) = 4

如果 λ=120: ρ=0.96, W=1/(125-120)=0.2s=200ms, L=0.96/0.04=24

从 100→120(仅 20% 增长),等待时间从 40ms→200ms(5倍增长!)

题2:SIR 参数调节

一个蠕虫 β=0.5, γ=0.1(R₀=5)。你有两种防御措施:

a) 部署 IDS 将 β 降到 0.2

b) 自动修复将 γ 提高到 0.3

哪种更有效?计算新的 R₀
答案
a) R₀ = 0.2/0.1 = 2(仍然会爆发但慢很多)

b) R₀ = 0.5/0.3 = 1.67(仍然爆发)

两者组合: R₀ = 0.2/0.3 = 0.67 < 1 → 不会爆发!

教训:组合防御比单一措施有效得多

题3:马尔可夫链手算

用户状态转移矩阵:

1
2
3
4
5
     浏览   加购   下单   离开
浏览 [ 0.5 0.2 0.0 0.3 ]
加购 [ 0.1 0.3 0.4 0.2 ]
下单 [ 0.0 0.0 0.0 1.0 ]
离开 [ 0.0 0.0 0.0 1.0 ]

一个用户当前在”浏览”状态,2 步后下单的概率是多少?
答案
路径:浏览→加购→下单 = 0.2 × 0.4 = 0.08

路径:浏览→浏览→下单 = 0.5 × 0.0 = 0(浏览不能直接下单)

其他经过下单的路径也为 0(从浏览和离开不能到下单)

2 步后下单概率 = 0.08 = 8%

题4:背包问题变形

你有 200 万预算,用上面的安全措施列表,但增加一个约束:WAF 和 DDoS 防护必须一起买(买了一个就必须买另一个)

最优方案是什么?
提示
方法:把 WAF+DDoS 合并成一个物品(成本90万,价值48)再跑背包

或者分两种情况讨论:选WAF+DDoS vs 不选WAF+DDoS

题5:综合建模

你负责一个微服务系统,双十一前需要决定是否扩容。已知:

日常 QPS = 500,双十一预计 QPS = 3000

单台服务器处理能力 μ = 800 QPS

当前 4 台服务器

问题:需要扩容到几台?目标是 P99 延迟 < 100ms
提示
每台分到的 QPS = 3000/n(假设均匀负载均衡)

用 M/M/1 模型估算每台的等待时间

目标:W = 1/(μ - λ/n) 的某个百分位 < 100ms

别忘了留余量(ρ < 0.8 原则)

最低需要 3000/(800×0.8) ≈ 4.7 → 至少 5 台

建议 6 台(留冗余应对突发流量和故障)


上一章 目录 下一章
30-博弈论入门 数学重学路线图