guzhujushi/MCM-ICM_Toolkit
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(solve): 同步第三轮入口口径（docstring + 测试 + lint）

Browse Source 
用户第三轮改了 5 处口径：非负校验带容差、initial 收 np.integer、paths 赋值挪进 try、
metrics 换成 terminal_distribution、mm1 放开 lambda=0 并收 numpy 标量。三处算法本体仍正确
（探针复测：终点分布 {0: 0.6644, 1: 0.3356}、lambda=0 给 L=Lq=Wq=0 且 W=1/mu、data is metrics 已为 False）。

Changed（Codex 只做同步，未改口径）
- docstring：markov_simulate 的 Returns 与"遗留"段、markov_steady_state 的 Raises
  （TypeError / ValueError 分工）、mm1_metrics 的 Parameters / Returns / Raises。
- tests/test_solve.py：同步 5 条受影响的测试，新增 2 条（terminal_distribution 只列出现过的状态、
  lambda=0 的空系统）。test_solve.py 64 条。
- ruff：B905（zip 缺 strict）+ 2 条空白 -> 0；rng 之后的双空行 -> 1。
- STATUS.md：决策 56 / 57 / 58 更新到"只剩 5 条小口径"；「下一步」第八轮；
  新增「第八轮备忘：models 与 decision 各是干什么的」（函数清单 / 使用顺序 / 建议开工顺序）。
- CHANGELOG 与 docs/dev/personal-pitfalls.md 易错点 30（口径一改，docstring / 测试 / STATUS 同时过期）。

校验：pytest 204 passed / 28 skipped（232 收集）；ruff 0；pyright 0
This commit is contained in:
guzhujushi committed 2026-10-10 09:31:30 +08:00
1 parent 953e988983
commit 9cb295a362
5 files changed
+190 -107
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/mcm_toolkit/solve/simulate.py
+44 -34
View File
@@ -487,7 +487,8 @@ def markov_simulate(
Returns
-------
`Result`：`data` 是形状 `(n_paths, steps + 1)` 的整数状态序列，第 0 列是初始状态；
`metrics={"mean_terminal", "std_terminal"}`（终点状态的均值与标准差，`std` 用 `ddof=0`）。
`metrics={"mean_terminal", "terminal_distribution"}` —— 前者是终点状态的均值，
后者是 `{状态: 出现频率}` 的 dict（按终点统计，只含**实际出现过**的状态，没出现的状态没有 key）。
Raises
------
@@ -504,9 +505,10 @@ def markov_simulate(
`cum_probs[:, -1] = 1.0` 是为了压掉浮点累加误差，避免随机数落在 `[1 - 1e-16, 1)` 时索引越界。
终点分布随步数收敛到平稳分布：这既是本函数的自检手段，也是论文里值得写一句的性质。
遗留（决策 56 待定）：`initial` 现在靠 `isinstance(initial, int)` 区分"标量状态 / 状态数组"，
`np.int64`、`bool` 会掉进数组分支，"按状态数校验长度"与"按路径数写入"也互相矛盾。
口径定下来后，上面 `initial` 这一节与本段随之改写。
遗留（决策 56）：`initial` 靠 `isinstance(initial, (int, np.integer))` 区分"标量状态 / 状态数组"，
`np.int64` 已收，但 `bool` 仍会被当成状态 1（`isinstance(True, int)` 为真）；
且数组分支"按**状态数**校验长度"与真正写进去时"要按**路径数**"仍互相矛盾 —— 只有 `n_paths == s` 才跑得通。
状态越界（如 `initial=5`、`s=2`）也要等第一轮索引才炸。口径定了这段随之改写。
Examples
--------
@@ -521,8 +523,8 @@ def markov_simulate(
arr, _ = as_array(transition)
if arr.ndim != 2 or arr.shape[0] != arr.shape[1]:
raise ValueError(f"transition 必须是二维方阵，你传入的形状是 {arr.shape}")
if np.any(arr < 0):
raise ValueError("transition 矩阵的元素必须是非负的（概率不能为负）")
if np.any(arr < -1e-12):
raise ValueError("transition 矩阵的元素必须是非负的")
if not np.allclose(arr.sum(axis=1), 1.0):
raise ValueError("transition 每一行的和必须是 1.0")
s = arr.shape[0]
@@ -538,24 +540,25 @@ def markov_simulate(
rng = get_rng(random_state)
if isinstance(initial, int):
if isinstance(initial, (int,np.integer)):
current_states = np.full(n_paths, initial)
initial_param = int(initial)
else:
current_states, _ = as_array(initial, ndim=1)
if len(current_states) != s:
raise TypeError(f"initial 的长度必须是 {s}，而你传入的是 {len(current_states)}")
paths = np.zeros((n_paths, steps + 1), dtype=int)
paths[:, 0] = current_states
initial_param = current_states.tolist()
call_params = {
"n_steps": steps,
"steps": steps,
"n_paths": n_paths,
"initial": initial,
"random_state": random_state,
"initial": initial_param,
"random_state": random_state
}
try:
paths = np.zeros((n_paths, steps + 1), dtype=int)
paths[:, 0] = current_states
for t in range(steps):
probs = arr[current_states]
@@ -569,9 +572,12 @@ def markov_simulate(
current_states = next_states
terminal_positions = paths[:, -1]
unique, counts = np.unique(terminal_positions, return_counts=True)
terminal_dist = {int(state): float(count) / n_paths for state, count in zip(unique, counts, strict=True)}
metrics = {
"mean_terminal": float(np.mean(terminal_positions)),
"std_terminal": float(np.std(terminal_positions)),
"terminal_distribution": terminal_dist
}
return Result(
name="markov_simulate",
@@ -612,9 +618,11 @@ def markov_steady_state(transition: Any, max_iter: int | np.integer = 10000, tol
Raises
------
TypeError:
`transition` 不是二维方阵；`max_iter` 不是整数；`tol` 不是数值。
ValueError:
`transition` 不是二维方阵、或某一行和不等于 1.0；
`max_iter` 不是正整数；`tol` 不是正数。
`transition` 含负元素、或某一行和不等于 1.0；
`max_iter` 不是正整数；`tol` 不是正数（NaN 也算非法）。
Notes
-----
@@ -639,19 +647,19 @@ def markov_steady_state(transition: Any, max_iter: int | np.integer = 10000, tol
"""
arr, _ = as_array(transition)
if arr.ndim != 2 or arr.shape[0] != arr.shape[1]:
raise ValueError(f"transition 必须是二维方阵，你传入的形状是 {arr.shape}")
if np.any(arr < 0):
raise TypeError(f"transition 必须是二维方阵，你传入的形状是 {arr.shape}")
if np.any(arr < -1e-12):
raise ValueError("transition 矩阵的元素必须是非负的（概率不能为负）")
if not np.allclose(arr.sum(axis=1), 1.0):
raise ValueError("transition 每一行的和必须是 1.0")
if isinstance(max_iter, bool) or not isinstance(max_iter, (int, np.integer)):
raise ValueError(f"max_iter 必须是正整数，你传入了 {type(max_iter).__name__}")
raise TypeError(f"max_iter 必须是正整数，你传入了 {type(max_iter).__name__}")
if max_iter <= 0:
raise ValueError(f"max_iter 必须是正整数，你传入了 {max_iter}")
if not isinstance(tol, (int, float)) or isinstance(tol, bool):
raise TypeError(f"tol 必须是浮点数，你传入了 {type(tol).__name__}")
if tol <= 0:
if tol <= 0 or np.isnan(tol):
raise ValueError(f"tol 必须是正数，你传入了 {tol}")
s = arr.shape[0]
@@ -706,21 +714,23 @@ def mm1_metrics(arrival_rate: float, service_rate: float) -> Any:
Parameters
----------
arrival_rate, service_rate:
到达率 λ 与服务率 μ。`λ >= μ` 时系统不稳定，队列长度无界——
这种情况下函数返回 `success=False` 并说明原因，而不是返回 `inf`。
到达率 λ 与服务率 μ，都收 `int` / `float` / numpy 标量。
`λ` 允许为 0（空系统：`L = Lq = Wq = 0`，`W` 取极限值 `1/μ`）；
`λ >= μ` 时系统不稳定、队列长度无界 —— 这种情况返回 `success=False` 并说明原因，而不是返回 `inf`。
Returns
-------
`Result`：`metrics={"rho", "L", "Lq", "W", "Wq"}`
（系统利用率、系统内平均顾客数、队列长度、平均逗留时间、平均等待时间）；
`data` 与 `metrics` 同值（闭式解没有样本数组）。
`data` 与 `metrics` 同值（闭式解没有样本数组），但**不是同一个对象**。
不稳定时 `success=False`、`metrics={{"rho"}}`（只报利用率）、`data=None`。
Raises
------
TypeError:
`arrival_rate` / `service_rate` 不是数值（`bool` 也判非法）。
ValueError:
`arrival_rate` / `service_rate` 不是正数。
`arrival_rate` 是负数；`service_rate` 不是正数。
Notes
-----
@@ -734,27 +744,27 @@ def mm1_metrics(arrival_rate: float, service_rate: float) -> Any:
>>> round(r.metrics["rho"], 4)
0.6667
"""
if not isinstance(arrival_rate, (int, float)) or isinstance(arrival_rate, bool):
if not isinstance(arrival_rate, (int, float, np.number)) or isinstance(arrival_rate, bool):
raise TypeError(f"arrival_rate 必须是浮点数，你传入了 {type(arrival_rate).__name__}")
if arrival_rate <= 0:
raise ValueError(f"arrival_rate 必须是正数，你传入了 {arrival_rate}")
if not isinstance(service_rate, (int, float)) or isinstance(service_rate, bool):
if arrival_rate < 0:
raise ValueError(f"arrival_rate 必须是非负数，你传入了 {arrival_rate}")
if not isinstance(service_rate, (int, float, np.number)) or isinstance(service_rate, bool):
raise TypeError(f"service_rate 必须是浮点数，你传入了 {type(service_rate).__name__}")
if service_rate <= 0:
raise ValueError(f"service_rate 必须是正数，你传入了 {service_rate}")
call_params = {"arrival_rate": arrival_rate, "service_rate": service_rate}
call_params = {"arrival_rate": float(arrival_rate), "service_rate": float(service_rate)}
if service_rate <= arrival_rate:
if arrival_rate >= service_rate:
rho = arrival_rate / service_rate
return Result(
name="mm1_metrics",
params=call_params,
metrics={},
metrics={"rho": float(rho)},
data=None,
success=False,
message="系统不稳定：到达率 λ 大于等于服务率 μ，队列长度无界，无法计算闭式解。"
message=f"系统不稳定：λ >= μ，队列无界。当前利用率 rho={rho:.4f}"
)
try:
rho = arrival_rate / service_rate
L = rho / (1.0 - rho) # 系统内平均顾客数
@@ -775,7 +785,7 @@ def mm1_metrics(arrival_rate: float, service_rate: float) -> Any:
name="mm1_metrics",
params=call_params,
metrics=metrics,
data=metrics,
data=metrics.copy(),
success=True,
message="M/M/1 排队系统指标计算成功。"
)