chore: 初始化 MCM/ICM Toolkit 仓库骨架

审查并精简 ChatGPT 的建库建议后落地的第一版骨架,只定接口不写实现。

- 建议审查:14 个包 → 7 个;约 250 个函数 → 约 40 个;5 个 Result 类 → 1 个
  (docs/01-suggestion-review.md)
- docs/:架构、路线图(含永久不做清单)、环境、Git 工作流、函数卡片模板
- src/mcm_toolkit/:core / data / stats / models / solve / decision / viz
  共 23 个模块,均为签名 + docstring + NotImplementedError
- tests/:9 个文件,其中 test_api_surface.py 已通过(20 passed)
- examples/ problems/ notebooks/ figures/ tables/ 骨架
- pyproject.toml(核心依赖 4 个 + 6 组 extras)、environment.yml
- scripts/:setup-venv.ps1(ASCII + LF)、setup-conda.ps1、doctor.ps1
- 环境:.venv(system-site-packages)与 conda 环境 mcm-toolkit 均已建好,
  数值栈版本与 base 完全对齐(numpy 2.4.6 / scipy 1.18.1 / pandas 3.0.6,MKL 后端)
This commit is contained in:
guzhujushi committed 2026-10-01 17:43:48 +08:00
commit 5319589dfd
76 files changed
+6812

No files matched your search

+54
View File
@@ -0,0 +1,54 @@
"""mcm_toolkit —— MCM/ICM 竞赛层工具箱。
设计原则见 docs/02-architecture.md:
* 薄封装优先:能一行转发 SciPy / sklearn / statsmodels / NetworkX / SALib 的,就不自己写。
* 自己实现是有目的的:只写"库里没有、或自己写才学得会"的算法。
* 统一出口:拟合 / 求解 / 模拟 / 评价类函数一律返回 :class:`Result`。
* 可复现:含随机的函数都接受 `random_state`。
顶层只导出子包与 6 个高频快捷函数,不做 `import *`,避免污染命名空间。
"""
from __future__ import annotations
# 子包:按 `import mcm_toolkit as mcm` 之后 `mcm.stats.xxx` 使用
from . import (
core, # noqa: F401
data, # noqa: F401
decision, # noqa: F401
models, # noqa: F401
solve, # noqa: F401
stats, # noqa: F401
viz, # noqa: F401
)
from ._version import __version__
# 高频快捷函数:比赛时最常用的几个,允许 `mcm.monte_carlo(...)` 直接用
from .core import Result, get_rng, set_seed # noqa: F401
from .decision import topsis # noqa: F401
from .models import compare_models # noqa: F401
from .solve import monte_carlo, parameter_sweep # noqa: F401
from .stats import bootstrap_ci # noqa: F401
# 公开 API 清单(不是"导出按钮",只列对外承诺稳定的名字)
__all__ = [
"__version__",
# 子包
"core",
"data",
"stats",
"models",
"solve",
"decision",
"viz",
# 快捷函数
"Result",
"get_rng",
"set_seed",
"bootstrap_ci",
"compare_models",
"monte_carlo",
"parameter_sweep",
"topsis",
]
+3
View File
@@ -0,0 +1,3 @@
"""版本号唯一来源。"""
__version__ = "0.0.1"
+23
View File
@@ -0,0 +1,23 @@
"""core —— 基础设施:结果对象、随机数、参数校验。
其他所有包都依赖本包;本包**不**依赖任何其他子包。
"""
from .random import get_rng, set_seed
from .result import Result
from .validate import (
check_array,
check_probability,
check_same_length,
check_square_matrix,
)
__all__ = [
"Result",
"get_rng",
"set_seed",
"check_array",
"check_probability",
"check_same_length",
"check_square_matrix",
]
+45
View File
@@ -0,0 +1,45 @@
"""随机数约定。
规则(见 docs/02-architecture.md §4.2):
* 所有含随机的公开函数签名都带 `random_state=None`,内部第一行取 rng。
* 禁止使用 `np.random.*` 旧式全局接口,也禁止裸用标准库 `random`。
* 论文附录必须写明 seed —— 提交要能复现。
"""
from __future__ import annotations
import numpy as np
__all__ = ["get_rng", "set_seed"]
def get_rng(random_state: int | None = None) -> np.random.Generator:
"""取一个随机数生成器。
Parameters
----------
random_state:
`int` 时返回固定种子的独立生成器(可复现);
`None` 时返回基于全局种子的生成器;
已经是 :class:`np.random.Generator` 时原样返回(便于函数间串联)。
Examples
--------
>>> rng = get_rng(42)
>>> bool(rng.random() == get_rng(42).random())
True
"""
# TODO(L0)
raise NotImplementedError
def set_seed(seed: int) -> None:
"""设置全局随机种子。
Examples
--------
>>> set_seed(42)
"""
# TODO(L0)
raise NotImplementedError
+51
View File
@@ -0,0 +1,51 @@
"""统一结果对象。
第一版只有**一个** :class:`Result`(原建议里的 5 个 Result 类属于过早分类)。
任何"拟合 / 求解 / 模拟 / 评价"函数都只返回它,不返回裸 tuple。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
@dataclass
class Result:
"""所有计算型函数的统一返回值。
Parameters
----------
name:
产生该结果的方法名,例如 `"monte_carlo"`。
params:
参数估计 / 决策变量取值 / 权重等"输入侧"的结论。
metrics:
评估指标,例如 `{"rmse": 0.31, "r2": 0.94}`。
data:
主要载荷:预测值、模拟轨迹、排序结果、收益数组等。
success:
是否正常完成。失败时仍返回 Result,但 `success=False` 且 `message` 说明原因。
message:
给人看的一句话说明,可直接抄进论文的"方法说明"。
Examples
--------
>>> r = Result("demo", params={"alpha": 0.5}, metrics={"r2": 0.9})
>>> r.success
True
"""
name: str
params: dict[str, Any] | None = None
metrics: dict[str, Any] | None = None
data: Any = None
success: bool = True
message: str = ""
extras: dict[str, Any] = field(default_factory=dict)
# TODO(L0): 实现 __repr__,打印成"方法名 + 指标表"的紧凑形式,
# 方便在 notebook 里直接肉眼检查、或复制进论文草稿。
# TODO(L0): 实现 summary() -> str,返回可直接贴进论文的 Markdown 表格。
def __repr__(self) -> str: # pragma: no cover - 待实现
raise NotImplementedError
+54
View File
@@ -0,0 +1,54 @@
"""参数校验。
目的不是"教用户用库",而是**把错误提前到调用点**:
比赛时最贵的是 debug 时间,一个 `ValueError: 数组长度不一致` 比
十层 traceback 里的 `IndexError` 值钱得多。
"""
from __future__ import annotations
from typing import Any
import numpy as np
__all__ = [
"check_array",
"check_probability",
"check_same_length",
"check_square_matrix",
]
def check_array(x: Any, ndim: int | None = None, finite: bool = True) -> np.ndarray:
"""转成 ndarray 并做基本检查(维度、有限值)。
Examples
--------
>>> check_array([1, 2, 3]).shape
(3,)
"""
# TODO(L0)
raise NotImplementedError
def check_same_length(*arrays: Any) -> None:
"""检查多个数组长度一致,不一致时抛出带长度明细的 ValueError。
Examples
--------
>>> check_same_length([1, 2], [3, 4])
"""
# TODO(L0)
raise NotImplementedError
def check_probability(p: float) -> float:
"""检查 p 落在 [0, 1],返回 float。"""
# TODO(L0)
raise NotImplementedError
def check_square_matrix(a: Any) -> np.ndarray:
"""检查是方阵,返回 ndarray。AHP 判断矩阵、转移矩阵都要用。"""
# TODO(L0)
raise NotImplementedError
+26
View File
@@ -0,0 +1,26 @@
"""data —— 数据进出与清洗(最高频的一层)。
真实赛题数据往往缺列、单位不统一、日期格式混乱。
本包的目标是让"拿到数据 → 能建模"这一步不超过 10 分钟。
"""
from .clean import clean_missing, detect_outliers, impute_missing, winsorize
from .io import load_table, save_table
from .sample import bootstrap_sample, kfold_split, train_test_split
from .transform import log_transform, normalize, one_hot_encode, standardize
__all__ = [
"load_table",
"save_table",
"clean_missing",
"impute_missing",
"detect_outliers",
"winsorize",
"normalize",
"standardize",
"log_transform",
"one_hot_encode",
"train_test_split",
"kfold_split",
"bootstrap_sample",
]
+81
View File
@@ -0,0 +1,81 @@
"""缺失值、离群值、缩尾。
`detect_outliers` 属于 S 档:不只是"算出来",还要能解释给队友听
为什么要踢掉这几个点(论文里必须交代),所以配完整函数卡片
(docs/dev/function-card-template.md)。
"""
from __future__ import annotations
from typing import Any, Literal
__all__ = ["clean_missing", "detect_outliers", "impute_missing", "winsorize"]
def clean_missing(data: Any, axis: int = 0, threshold: float | None = None) -> Any:
"""删除缺失过多的行/列。
Parameters
----------
axis:
0 = 按行删,1 = 按列删。
threshold:
缺失比例超过该值就删;`None` 表示只要有缺失就删。
Examples
--------
>>> clean_missing(df, threshold=0.5) # doctest: +SKIP
"""
# TODO(L0)
raise NotImplementedError
def impute_missing(data: Any, method: str = "mean", **kwargs: Any) -> Any:
"""缺失值填补。
Parameters
----------
method:
`"mean"` / `"median"` / `"mode"` / `"ffill"` / `"interpolate"` / `"knn"`。
Notes
-----
论文里必须说明用了哪种填补、为什么。填 `mean` 之前先想一下
"这个缺失是不是非随机缺失(MNAR)",否则结论可能被填补方式带偏。
"""
# TODO(L0)
raise NotImplementedError
def detect_outliers(data: Any, method: Literal["iqr", "zscore"] = "iqr", **kwargs: Any) -> Any:
"""标出离群点位置(返回布尔掩码,不直接删)。
Parameters
----------
method:
`"iqr"`:Q1-1.5·IQR 之外(`k` 可调,默认 1.5);
`"zscore"`:|z| > threshold(默认 3.0)。
Returns
-------
与原数据同形的布尔 DataFrame/ndarray,`True` 表示该点是离群点。
Notes
-----
**只标不删**是刻意的:删点是有代价的判断,必须由人来做,
并且要在论文里写清楚删了几个、依据是什么。
Examples
--------
>>> mask = detect_outliers([1, 2, 3, 4, 100], method="iqr")
>>> bool(mask[-1])
True
"""
# TODO(L0) ★ S 档:配完整函数卡片
raise NotImplementedError
def winsorize(data: Any, lower: float = 0.01, upper: float = 0.99) -> Any:
"""缩尾:把超出分位数的值压到分位点,而不是删除。"""
# TODO(L1)
raise NotImplementedError
+37
View File
@@ -0,0 +1,37 @@
"""读写表格。
只保留两个函数:`load_table` / `save_table` 按扩展名分派。
原建议里的 `load_csv` `load_excel` `save_csv` `save_excel` 四个函数
只是在转发 pandas,合并后调用点更短。
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
__all__ = ["load_table", "save_table"]
def load_table(path: str | Path, **kwargs: Any) -> Any:
"""按扩展名自动选择读取方式,返回 DataFrame。
约定:
* `.csv` 默认做编码探测(utf-8 → utf-8-sig → gbk → gb18030),
国内赛题数据常见 GBK;
* `.xlsx` / `.xls` 需要 `openpyxl`(`pip install "mcm-toolkit[excel]"`);
* `na_values` 默认包含空串、`"NA"`、`"N/A"`、`"#N/A"`、`"null"`、`"-"`。
Examples
--------
>>> df = load_table("data.csv") # doctest: +SKIP
>>> df.shape # doctest: +SKIP
"""
# TODO(L0): 编码探测 + 分派;Excel 缺 openpyxl 时给出可读报错。
raise NotImplementedError
def save_table(df: Any, path: str | Path, index: bool = False, **kwargs: Any) -> Path:
"""保存 DataFrame,返回实际写入的路径。"""
# TODO(L0)
raise NotImplementedError
+46
View File
@@ -0,0 +1,46 @@
"""数据集划分与重采样。
原建议里的 `sample_with_replacement` / `sample_without_replacement`
就是 `rng.choice(..., replace=)` 一行,已删;真正需要的是"划分"和"bootstrap"。
"""
from __future__ import annotations
from typing import Any
__all__ = ["bootstrap_sample", "kfold_split", "train_test_split"]
def train_test_split(
*arrays: Any,
test_size: float = 0.2,
shuffle: bool = True,
stratify: Any = None,
random_state: int | None = None,
) -> list[Any]:
"""按行划分训练/测试集。
Examples
--------
>>> Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.2, random_state=0) # doctest: +SKIP
"""
# TODO(L0)
raise NotImplementedError
def kfold_split(n_samples: int, n_splits: int = 5, shuffle: bool = True, random_state: int | None = None) -> list[tuple[Any, Any]]:
"""返回 `n_splits` 组 (train_idx, test_idx)。"""
# TODO(L1)
raise NotImplementedError
def bootstrap_sample(data: Any, n_resamples: int = 10000, random_state: int | None = None) -> Any:
"""有放回重采样,返回形状 `(n_resamples, n)` 的重采样矩阵。
.. warning::
`n_resamples=10000` 且 `n` 很大时这个矩阵会吃内存。
只是想算区间的话用 :func:`mcm_toolkit.stats.bootstrap_ci`,
它内部按块处理,不落地整个矩阵。
"""
# TODO(L1)
raise NotImplementedError
+58
View File
@@ -0,0 +1,58 @@
"""变换与无量纲化。
**重要约定(见 docs/02-architecture.md §4.4)**:
本模块只做"无量纲化",**不改变指标方向**。
效益型 / 成本型的方向问题归 `decision` 包处理,用 `benefit_columns` 显式声明,
不允许默认假设"越大越好"——评价类题目翻车基本都翻在这里。
"""
from __future__ import annotations
from typing import Any
__all__ = ["log_transform", "normalize", "one_hot_encode", "standardize"]
def normalize(x: Any, method: str = "minmax", feature_range: tuple[float, float] = (0.0, 1.0)) -> Any:
"""无量纲化到给定区间。
Parameters
----------
method:
`"minmax"`:`(x - min) / (max - min)`,线性压缩到 `feature_range`;
`"vector"`:`x / ||x||`,保持"份额"含义(熵权法前处理常用);
`"max"`:`x / max(x)`。
Notes
-----
minmax 对离群值极度敏感:一个 100 倍大的异常值会把其余所有点压成一条线。
先跑 `detect_outliers` 再决定。
Examples
--------
>>> normalize([1, 2, 3]).tolist()
[0.0, 0.5, 1.0]
"""
# TODO(L0)
raise NotImplementedError
def standardize(x: Any, ddof: int = 0) -> Any:
"""标准化 `(x - mean) / std`,输出均值 0、标准差 1。"""
# TODO(L0)
raise NotImplementedError
def log_transform(x: Any, offset: float | None = None) -> Any:
"""对数变换,用于压缩右偏分布。
`offset=None` 时自动取 `1 - min(x)`(含 0 或负值时)保证定义域。
"""
# TODO(L1)
raise NotImplementedError
def one_hot_encode(data: Any, columns: list[str] | None = None, drop_first: bool = False) -> Any:
"""类别变量独热编码。"""
# TODO(L1)
raise NotImplementedError
+26
View File
@@ -0,0 +1,26 @@
"""decision —— 评价与决策。
中文赛高频模块(AHP / TOPSIS / 熵权 / 灰色)。
原建议列了 8 个子模块,其中 `dea_efficiency`(要解一堆 LP,低频)与
`fuzzy_comprehensive_evaluation`(隶属度函数没有通用形式,做不了通用封装)已删。
**使用顺序很重要**:先 `normalize`(无量纲化)→ 再定方向 `benefit_columns`
→ 再 `entropy_weights`/`ahp_weights` → 最后 `topsis`。
跳过第一步直接算权重,是评价类题目最常见的翻车方式。
"""
from .ahp import ahp_consistency_ratio, ahp_weights
from .grey import gm11_fit_predict, grey_relational_analysis
from .topsis import topsis
from .weighting import critic_weights, entropy_weights, equal_weights
__all__ = [
"equal_weights",
"entropy_weights",
"critic_weights",
"ahp_weights",
"ahp_consistency_ratio",
"topsis",
"grey_relational_analysis",
"gm11_fit_predict",
]
+57
View File
@@ -0,0 +1,57 @@
"""层次分析法(AHP)。
不过度设计:原建议第 22 节给出的两个函数就是全部需要的东西。
关键在**一致性检验**——CR 不过关的权重不能拿去用,
这是 AHP 唯一真正的技术含量所在。
"""
from __future__ import annotations
from typing import Any
__all__ = ["ahp_consistency_ratio", "ahp_weights"]
def ahp_weights(matrix: Any) -> Any:
"""由判断矩阵求权重。
Parameters
----------
matrix:
正互反判断矩阵(`a_ij = 1/a_ji`,`a_ii = 1`),Saaty 1-9 标度。
Returns
-------
`Result`:`params={"weights"}`,`metrics={"lambda_max", "CI", "CR", "consistent"}`。
Notes
-----
取权重有两种做法:
* **特征向量法**(`eigenvector`):求最大特征值对应的特征向量并归一化,理论严谨;
* **几何平均法**(`geometric`):各行元素连乘开 n 次方再归一化,手算/答辩时好用,
与特征向量法结果几乎一致。
本函数默认特征向量法,`method="geometric"` 可切换,两者都返回以便对照。
**CR > 0.1 一般判定为不一致**(n=2 时 CR 无意义,直接跳过)。
有人会"改矩阵直到 CR 过关",这在论文里必须交代调整依据,否则是学术不端。
Examples
--------
>>> import numpy as np
>>> A = np.array([[1, 3, 5], [1/3, 1, 2], [1/5, 0.5, 1]])
>>> r = ahp_weights(A)
>>> bool(r.metrics["CR"] < 0.1)
True
"""
# TODO(L1) ★ S 档:配完整函数卡片
raise NotImplementedError
def ahp_consistency_ratio(matrix: Any) -> Any:
"""只算一致性指标,不返回权重。
单独提供它是因为:**拿到判断矩阵后第一步就该看 CR**,
不过关时先回去调整矩阵,权重算出来也用不了。
"""
# TODO(L1)
raise NotImplementedError
+71
View File
@@ -0,0 +1,71 @@
"""灰色系统:关联分析与 GM(1,1) 预测。
中文建模赛特色方法。原建议第 23 节的方向没问题,函数名做了合并
(`gm11_fit` + `gm11_predict` → `gm11_fit_predict`,因为 GM(1,1) 极少只用一半)。
"""
from __future__ import annotations
from typing import Any
__all__ = ["gm11_fit_predict", "grey_relational_analysis"]
def grey_relational_analysis(X: Any, reference: Any = None, rho: float = 0.5) -> Any:
"""灰色关联分析。
Parameters
----------
reference:
参考序列。`None` 时按"每个指标取最优值"自动构造。
rho:
分辨系数,通常取 0.5。
Returns
-------
`Result`:`data` 是各方案与参考序列的关联系数、关联度与排名。
Notes
-----
关联度对 `rho` 敏感:`rho` 越小,分辨力越强但稳定性越差。
论文里应说明 `rho` 取值并做一次敏感性(`rho = 0.3 / 0.5 / 0.7`)。
另外它**要求所有指标同向且无量纲**,前置处理同熵权法。
Examples
--------
>>> r = grey_relational_analysis([[1, 2], [3, 1], [2, 3]]) # doctest: +SKIP
"""
# TODO(L1)
raise NotImplementedError
def gm11_fit_predict(series: Any, steps: int = 1) -> Any:
"""GM(1,1) 灰色预测(拟合 + 预测一步到位)。
流程:一次累加生成(1-AGO)→ 建立白化微分方程 `dx/dt + ax = b`
→ 最小二乘估 `a, b` → 还原预测序列。
Parameters
----------
steps:
向前预测步数。GM(1,1) 适合**短序列(4~10 个点)**的短期预测。
Returns
-------
`Result`:`params={"a", "b"}`,`metrics={"c_ratio", "precision"}`,
`data` 是含 `actual / fitted / forecast` 三列的 DataFrame。
Notes
-----
* **必须做后验差检验**:后验差比值 `C = S2/S1`(残差标准差 / 原序列标准差)
和 小误差概率 `P`。`C < 0.35` 一级,`< 0.5` 合格,`>= 0.65` 不能用。
只给预测值不给 `C`、`P` 是论文里的硬伤。
* GM(1,1) 本质是指数增长外推,对**先增后减**或波动序列效果很差,
用之前先看序列形状。
Examples
--------
>>> r = gm11_fit_predict([2.87, 3.05, 3.24, 3.43, 3.64]) # doctest: +SKIP
"""
# TODO(L1) ★ S 档:配完整函数卡片
raise NotImplementedError
+59
View File
@@ -0,0 +1,59 @@
"""TOPSIS 逼近理想解排序。
原建议把标准化(`benefit_normalize` 等 4 个函数)单列一节,
这里合并进 `topsis` 的 `method` 参数——因为它们是同一个流程的步骤,
拆开反而容易出现"标准化用了一套、权重算的时候又换一套"的错位。
"""
from __future__ import annotations
from typing import Any
__all__ = ["topsis"]
def topsis(
X: Any,
weights: Any = None,
benefit_columns: Any = None,
method: str = "vector",
labels: Any = None,
) -> Any:
"""TOPSIS 综合评价。
Parameters
----------
X:
决策矩阵,行 = 待评价方案,列 = 指标。
weights:
权重。`None` 时用等权(并在 `message` 里提示)。
benefit_columns:
效益型指标的列名/下标,其余按成本型处理。
**必填或显式全选**,不允许默认假设——方向搞反会让排名完全颠倒。
method:
列向量的无量纲化方式:`"vector"`(`x/√Σx²`,TOPSIS 原始做法,默认)
或 `"minmax"`。
labels:
方案名,用于结果表。
Returns
-------
`Result`:`data` 是含 `D+ / D- / closeness / rank` 四列、按 `rank` 排好的表,
`params["weights"]` 回显实际使用的权重。
Notes
-----
* 正理想解 = 各指标最优值,负理想解 = 各指标最劣值;
贴近度 `C = D- / (D+ + D-)`,**C 越大越好**。
* TOPSIS 的排名对权重敏感。论文里应附"权重扰动 ±10% 后排名是否稳定"的
敏感性分析——直接调 `solve.robustness_test`。
Examples
--------
>>> import numpy as np
>>> r = topsis(np.array([[1., 2.], [3., 1.], [2., 3.]]), benefit_columns=[0, 1])
>>> list(r.data["rank"])
[2, 3, 1]
"""
# TODO(L1) ★ S 档:配完整函数卡片
raise NotImplementedError
+71
View File
@@ -0,0 +1,71 @@
"""客观赋权:熵权法、CRITIC。
★ `entropy_weights` 是 S 档,建议自己实现。
熵权法是中文建模赛出现频率最高的赋权方法,实现只有 20 行,
但"指标方向、无量纲化、对数 0 处理"三个坑必须理解清楚才能写对。
"""
from __future__ import annotations
from typing import Any
__all__ = ["critic_weights", "entropy_weights", "equal_weights"]
def equal_weights(n: int) -> Any:
"""等权,即 `1/n`。作为赋权结果的对照基线,论文里应该报。"""
# TODO(L1)
raise NotImplementedError
def entropy_weights(
X: Any,
benefit_columns: Any = None,
eps: float = 1e-12,
) -> Any:
"""熵权法。
Parameters
----------
X:
决策矩阵,行 = 方案,列 = 指标。**必须已无量纲化且已做正向化**。
benefit_columns:
效益型(越大越好)指标的列名/下标。其余按成本型处理。
`None` 时假设全部是效益型,并在 `message` 里提示——避免静默搞错方向。
Returns
-------
`Result`:`params={"weights": 权重向量}`,
`extras["entropy"]` 是各指标的熵值,
`extras["redundancy"]` 是差异系数 `1 - e`(权重就来自它的归一化)。
Notes
-----
三个必须处理的细节:
1. **方向**:成本型指标要先正向化(`max - x` 或 `1/x`),否则权重完全错。
2. **量纲**:熵权法要求 `x >= 0`,且对量纲敏感——先做 min-max 或向量归一。
3. **零值**:`p·ln(p)` 在 `p=0` 时按 0 处理(极限为 0),不能直接算,否则出 NaN。
熵权法的缺点是**只看数据离散程度,不看指标的实际重要性**。
实践上常和 AHP 做组合赋权:`w = α·w_entropy + (1-α)·w_ahp`。
Examples
--------
>>> import numpy as np
>>> r = entropy_weights(np.array([[1., 2.], [3., 1.], [2., 3.]]))
>>> bool(np.isclose(r.params["weights"].sum(), 1.0))
True
"""
# TODO(L1) ★ S 档:配完整函数卡片
raise NotImplementedError
def critic_weights(X: Any, benefit_columns: Any = None) -> Any:
"""CRITIC 赋权。
与熵权法的区别:熵权法只看"对比强度"(标准差),
CRITIC 还乘上"冲突性"(与其他指标的相关性)。
`C_j = σ_j · Σ_k (1 - r_jk)`,权重 = `C_j / ΣC`。
"""
# TODO(L1)
raise NotImplementedError
+41
View File
@@ -0,0 +1,41 @@
"""models —— 建模:回归 / 时间序列 / 机器学习(薄封装)。
原建议把 `regression/` 和 `ml/` 拆成两个包,但两边的
`fit_linear` 与 `fit_linear_regression` 明显重复。
这里合并为一个包,共享一套 `fit / predict / score` 协议和一份指标实现。
"""
from .metrics import (
adjusted_r2,
compare_models,
mae,
mape,
r2_score,
rmse,
smape,
)
from .ml import cross_validate, fit_forest, fit_logistic, kmeans, pca
from .regression import fit_linear, fit_nonlinear, fit_polynomial, predict
from .timeseries import exponential_smoothing, fit_arima, forecast
__all__ = [
"fit_linear",
"fit_polynomial",
"fit_nonlinear",
"predict",
"exponential_smoothing",
"fit_arima",
"forecast",
"fit_forest",
"fit_logistic",
"kmeans",
"pca",
"cross_validate",
"rmse",
"mae",
"mape",
"smape",
"r2_score",
"adjusted_r2",
"compare_models",
]
+31
View File
@@ -0,0 +1,31 @@
"""模型协议。
所有 `fit_*` 返回的对象都满足这个协议,于是:
`compare_models({"linear": m1, "rf": m2}, X, y)`
才能对一批异构模型统一执行、统一打分。
"""
from __future__ import annotations
from typing import Any, Protocol, runtime_checkable
import numpy as np
__all__ = ["Model"]
@runtime_checkable
class Model(Protocol):
"""`fit / predict / score` 三件套。"""
def fit(self, X: Any, y: Any, **kwargs: Any) -> Model:
"""拟合,返回 self(支持链式调用)。"""
...
def predict(self, X: Any) -> np.ndarray:
"""预测。"""
...
def score(self, X: Any, y: Any) -> dict[str, float]:
"""返回指标 dict(不是单个 float),便于直接拼成对比表。"""
...
+94
View File
@@ -0,0 +1,94 @@
"""回归/分类指标 + 模型对比。
原建议把这批函数同时放在 `regression/`、`utils/metrics.py` 和第 26 节的
`analysis/` 三处,这里统一到一处。
注意:本模块被 `models` 与 `solve` 共用,属于"公共数学",但为了少建一个包,
放在 `models` 下;`solve` 单向依赖 `models` 是允许的。
"""
from __future__ import annotations
from typing import Any
__all__ = [
"adjusted_r2",
"compare_models",
"mae",
"mape",
"r2_score",
"rmse",
"smape",
]
def rmse(y_true: Any, y_pred: Any) -> float:
"""均方根误差。与 y 同量纲,最常用。"""
# TODO(L0)
raise NotImplementedError
def mae(y_true: Any, y_pred: Any) -> float:
"""平均绝对误差。对离群值比 RMSE 稳健。"""
# TODO(L0)
raise NotImplementedError
def mape(y_true: Any, y_pred: Any) -> float:
"""平均绝对百分比误差。无量纲,但 y 含 0 时会炸(用 smape 替代)。"""
# TODO(L1)
raise NotImplementedError
def smape(y_true: Any, y_pred: Any) -> float:
"""对称 MAPE,y 含 0 也安全。"""
# TODO(L1)
raise NotImplementedError
def r2_score(y_true: Any, y_pred: Any) -> float:
"""决定系数 R²。注意:加变量一定会升高,跨模型比较要看 adjusted_r2。"""
# TODO(L0)
raise NotImplementedError
def adjusted_r2(y_true: Any, y_pred: Any, n_features: int) -> float:
"""调整 R²,惩罚变量个数。"""
# TODO(L1)
raise NotImplementedError
def compare_models(
models: dict[str, Any],
X: Any,
y: Any,
metrics: tuple[str, ...] = ("rmse", "mae", "r2"),
cv: int = 5,
random_state: int | None = None,
) -> Any:
"""把若干模型放在一起打分,返回对比表。
这是赛题里最省时间的函数之一。队友问"要不要上复杂模型"时,
跑一次这个就能用数字回答,不用争论半小时。
Parameters
----------
models:
`{"linear": m1, "poly3": m2, "random_forest": m3}`。
每个模型需满足 :class:`mcm_toolkit.models.base.Model` 协议(有 `fit`/`predict`)。
cv:
交叉验证折数;`cv=0` 表示在全部数据上拟合后打分(只看拟合能力时用)。
Returns
-------
DataFrame:行 = 模型,列 = 指标;按第一列指标升序(越小越好)或降序排列。
可直接 `save_table(result, "compare.tex", fmt="latex")` 贴进论文。
Examples
--------
>>> tbl = compare_models({"linear": m1, "poly": m2}, X, y) # doctest: +SKIP
>>> tbl.index[0] # doctest: +SKIP
'poly'
"""
# TODO(L1)
raise NotImplementedError
+88
View File
@@ -0,0 +1,88 @@
"""机器学习薄封装。
对应原建议的 B 类。规模要控制住:
第一版只封 commonly-used 的四个模型 + 两个工具函数,
不追求把 sklearn 的 estimator 列表抄一遍。
"""
from __future__ import annotations
from typing import Any
__all__ = ["cross_validate", "fit_forest", "fit_logistic", "kmeans", "pca"]
def fit_forest(X: Any, y: Any, task: str = "auto", n_estimators: int = 300, random_state: int | None = None, **kwargs: Any) -> Any:
"""随机森林(分类或回归)。
`task="auto"`:按 y 的取值类型自动判断分类还是回归。
Notes
-----
随机森林最大的价值不是精度,而是 **feature_importances_**——
赛题论文里"哪些因素最重要"这一节经常靠它。
返回值里带上排序后的重要性表。
Examples
--------
>>> r = fit_forest(X, y)
>>> list(r.metrics) # doctest: +SKIP
['r2', 'rmse', 'mae']
"""
# TODO(L1)
raise NotImplementedError
def fit_logistic(X: Any, y: Any, C: float = 1.0, random_state: int | None = None, **kwargs: Any) -> Any:
"""逻辑回归(二分类)。
Notes
-----
`metrics` 里除了准确率,必须同时给 **混淆矩阵、精确率/召回率、AUC**。
只报准确率在不平衡数据上会给出误导性结论。
"""
# TODO(L1)
raise NotImplementedError
def kmeans(X: Any, n_clusters: int = 3, standardize_first: bool = True, random_state: int | None = None, **kwargs: Any) -> Any:
"""K-means 聚类。
Parameters
----------
standardize_first:
默认先标准化。**量纲不同的特征不做标准化,K-means 的结果等于随机**。
Notes
-----
`metrics` 给轮廓系数(silhouette)而不是 inertia——
轮廓系数有界且可比,inertia 只能同数据集内比。
选 `n_clusters` 请画肘部图或轮廓系数曲线,不要拍脑袋。
"""
# TODO(L1)
raise NotImplementedError
def pca(X: Any, n_components: int | float | None = None, standardize_first: bool = True) -> Any:
"""主成分分析。
`n_components` 为 `float`(如 0.95)时表示"保留 95% 方差"。
Returns
-------
`Result`:`data` 是投影后的得分矩阵,`extras["loadings"]` 是载荷矩阵
(论文里解释"第一主成分代表什么"要用它)。
"""
# TODO(L1)
raise NotImplementedError
def cross_validate(model: Any, X: Any, y: Any, cv: int = 5, scoring: str | None = None, random_state: int | None = None) -> Any:
"""交叉验证,返回每折分数与均值±标准差。
Notes
-----
报结果时**必须带标准差**。只报均值会把"5 折里 1 折崩了"这种事掩盖掉。
"""
# TODO(L1)
raise NotImplementedError
+70
View File
@@ -0,0 +1,70 @@
"""回归(自己封装,底层调用成熟库)。
对应原建议的 B 类:学的是"怎样把成熟工具变成符合 MCM 工作流的接口",
不是重写最小二乘。
"""
from __future__ import annotations
from typing import Any
__all__ = ["fit_linear", "fit_nonlinear", "fit_polynomial", "predict"]
def fit_linear(X: Any, y: Any, intercept: bool = True, weights: Any | None = None, alpha: float = 0.0) -> Any:
"""线性回归。
Parameters
----------
weights:
加权最小二乘的权重(数据点可信度不同、或想消除异方差时用)。
alpha:
> 0 时做岭回归(共线性严重时用),= 0 是普通最小二乘。
Returns
-------
`Result`:`params={"coef", "intercept"}`,`metrics={"r2", "adjusted_r2", "rmse", "mae"}`。
Examples
--------
>>> r = fit_linear(X, y)
>>> r.metrics["r2"] < 1.0
True
"""
# TODO(L0)
raise NotImplementedError
def fit_polynomial(X: Any, y: Any, degree: int = 2, alpha: float = 0.0) -> Any:
"""多项式回归。`degree` 大了会过拟合,务必看 `rs.plot_residuals` 和 adjusted_r2。"""
# TODO(L1)
raise NotImplementedError
def fit_nonlinear(func: Any, X: Any, y: Any, p0: Any = None, bounds: Any = None, **kwargs: Any) -> Any:
"""非线性最小二乘拟合(底层 `scipy.optimize.curve_fit`)。
Parameters
----------
func:
`f(x, *params) -> y` 形式的目标函数,例如 logistic / Gompertz 生长曲线。
p0:
初值。非线性拟合**对初值极度敏感**,给不出好初值时要先画图估一个。
Notes
-----
返回的 `metrics` 里额外给 `pcov`(参数协方差矩阵)与参数标准误,
论文里报参数时要带标准误,否则不严谨。
"""
# TODO(L1)
raise NotImplementedError
def predict(model: Any, X: Any) -> Any:
"""对任意 `fit_*` 返回的 `Result` 或符合协议的模型做预测。
存在的意义:让 `predict` 既能吃 `Result` 也能吃 sklearn 原生模型,
统一比赛代码的写法。
"""
# TODO(L1)
raise NotImplementedError
+67
View File
@@ -0,0 +1,67 @@
"""时间序列(MCM C 必备)。
原建议列了 `moving_average` `weighted_moving_average` `holt_trend`
`seasonal_decompose` `lag_features` `forecast_interval` 等一堆函数,
但 statsmodels 已经全部提供,按它自己的原则("做薄封装,不要重写框架")
只保留三个真正有封装价值的入口。
"""
from __future__ import annotations
from typing import Any
__all__ = ["exponential_smoothing", "fit_arima", "forecast"]
def exponential_smoothing(series: Any, alpha: float | None = None, trend: str | None = None, seasonal: str | None = None, periods: int | None = None) -> Any:
"""指数平滑(底层 `statsmodels.tsa.holtwinters`)。
`alpha=None` 时自动用极大似然估计平滑系数——比自己拍脑袋定 `alpha` 靠谱。
Examples
--------
>>> r = exponential_smoothing(y, trend="add") # doctest: +SKIP
"""
# TODO(L1)
raise NotImplementedError
def fit_arima(series: Any, order: tuple[int, int, int] = (1, 1, 1), seasonal_order: tuple[int, int, int, int] | None = None, auto: bool = False) -> Any:
"""ARIMA 拟合。
Parameters
----------
auto:
`True` 时按 AIC 网格搜索 (p, d, q),省去手动定阶;
数据量大时较慢,比赛里建议先手动看一眼 ACF/PACF。
order:
`(p, d, q)`——自回归阶数、差分阶数、移动平均阶数。
Returns
-------
`Result`:`metrics={"aic", "bic"}`,`extras["model"]` 是 statsmodels 结果对象
(需要 `summary()`、残差诊断时直接拿)。
Notes
-----
ARIMA 要求序列**平稳**。`d` 是差分阶数,但差分会丢信息,
非平稳序列更推荐先看是不是有结构突变或季节成分。
"""
# TODO(L1)
raise NotImplementedError
def forecast(model: Any, steps: int, alpha: float = 0.05) -> Any:
"""向后预测 `steps` 步,并给出预测区间。
Returns
-------
`Result`:`data` 是含 `forecast / lower / upper` 三列的 DataFrame。
Notes
-----
**预测区间随步长急剧变宽是正常的**,不要只画中心线——
论文里只画一条线而不给区间,是很常见的扣分点。
"""
# TODO(L1)
raise NotImplementedError
+58
View File
@@ -0,0 +1,58 @@
"""solve —— 求解:优化 / 模拟 / 敏感性 / 图 / 动力学。
竞赛主力包。原建议把 `optimization` `simulation` `sensitivity` 拆成三个包,
但它们本质是同一件事的三种形态——"把模型反复算"——共享
`Result`、`rng`、并行与缓存约定。合成一个包,这些约定只写一遍。
`graph` 与 `dynamics` 是 v0.2 内容(低频,且 NetworkX / SymPy 已覆盖大部分需求)。
"""
from .optimize import (
dynamic_programming,
integer_program,
knapsack_dp,
linear_program,
minimize,
pareto_front,
simulated_annealing,
)
from .sensitivity import (
local_sensitivity,
morris_analyze,
one_at_a_time,
parameter_sweep,
robustness_test,
sobol_analyze,
)
from .simulate import (
markov_simulate,
markov_steady_state,
mm1_metrics,
monte_carlo,
monte_carlo_integral,
monte_carlo_probability,
random_walk,
)
__all__ = [
"minimize",
"linear_program",
"integer_program",
"dynamic_programming",
"knapsack_dp",
"simulated_annealing",
"pareto_front",
"monte_carlo",
"monte_carlo_probability",
"monte_carlo_integral",
"random_walk",
"markov_simulate",
"markov_steady_state",
"mm1_metrics",
"local_sensitivity",
"one_at_a_time",
"parameter_sweep",
"robustness_test",
"sobol_analyze",
"morris_analyze",
]
+79
View File
@@ -0,0 +1,79 @@
"""微分方程与动力系统(v0.2)。
原建议把 `dynamics/` 做成独立包。考虑到:
1. 只有 MCM A 用得上,一年一题;
2. 底层 SymPy(解析)+ SciPy(数值)已经覆盖全部需求;
降级为一个模块。依赖:`pip install "mcm-toolkit[symbolic]"`(sympy)。
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
__all__ = [
"find_equilibrium",
"local_stability",
"solve_ode",
"solve_system",
]
def solve_ode(
derivative: Callable[..., Any],
t_span: tuple[float, float],
y0: Any,
t_eval: Any = None,
method: str = "RK45",
**kwargs: Any,
) -> Any:
"""解 ODE 初值问题(底层 `scipy.integrate.solve_ivp`)。
Parameters
----------
method:
`"RK45"` 通用;`"LSODA"` 刚性系统自动切换(种群/化学反应常见);
`"Radau"` / `"BDF"` 强刚性。
Notes
-----
**刚性是 ODE 最常踩的坑**:非刚性方法遇到刚性问题会跑到天荒地老
且以为是自己代码慢。判据:解里同时存在快慢两个时间尺度。
Examples
--------
>>> r = solve_ode(lambda t, y: -y, t_span=(0, 5), y0=[1.0]) # doctest: +SKIP
"""
# TODO(v0.2)
raise NotImplementedError
def solve_system(equations: Any, variables: Any, **kwargs: Any) -> Any:
"""解方程组(底层 `sympy.solve` / `scipy.optimize.root`)。
`equations` 是符号表达式列表时走 SymPy 解析解,是数值函数时走 root 数值解。
"""
# TODO(v0.2)
raise NotImplementedError
def find_equilibrium(f: Callable[..., Any], x0: Any, n_starts: int = 10, random_state: int | None = None) -> Any:
"""找平衡点。
**必须多初值搜索**:单个初值只能找到最近的平衡点,
而"系统有几个平衡点、分别是什么类型"才是题目真正要问的。
"""
# TODO(v0.2)
raise NotImplementedError
def local_stability(jacobian_matrix: Any, equilibrium: Any = None) -> Any:
"""按 Jacobian 的特征值判定平衡点的局部稳定性。
判据(连续系统):实部全负 → 稳定结点/焦点;有正实部 → 不稳定;
纯虚根 → 中心(临界情形,线性化失效,需要非线性分析)。
`equilibrium` 用于把特征值和解的位置一起报出来。
"""
# TODO(v0.2)
raise NotImplementedError
+86
View File
@@ -0,0 +1,86 @@
"""图与网络(v0.2)。
原建议把 `network/` 做成一个独立包(5 个子模块、20+ 函数),
但里面绝大多数是 NetworkX 的一行转发,且只有 ICM D 用得上。
降级为一个模块,只留真正有封装价值的入口。
依赖:`pip install "mcm-toolkit[graph]"`(networkx)。
"""
from __future__ import annotations
from typing import Any
__all__ = [
"betweenness_centrality",
"build_graph",
"community_detection",
"max_flow",
"min_spanning_tree",
"pagerank",
"shortest_path",
]
def build_graph(edges: Any, directed: bool = False, weighted: bool = True) -> Any:
"""从边表建图。
Parameters
----------
edges:
`[(u, v), ...]` 或 `[(u, v, w), ...]`,也可以是 DataFrame(列名 `source/target/weight`)。
Examples
--------
>>> G = build_graph([("A", "B"), ("B", "C")]) # doctest: +SKIP
"""
# TODO(v0.2)
raise NotImplementedError
def shortest_path(G: Any, source: Any, target: Any = None, weight: str | None = None) -> Any:
"""最短路径。`target=None` 时返回从 source 出发到所有点的路径。"""
# TODO(v0.2)
raise NotImplementedError
def min_spanning_tree(G: Any, weight: str = "weight") -> Any:
"""最小生成树(铺管网、修路这类"连通成本最小"题目)。"""
# TODO(v0.2)
raise NotImplementedError
def betweenness_centrality(G: Any, normalized: bool = True) -> Any:
"""介数中心性。
节点的"关键程度"有三种主流度量(度/接近/介数),
论文里选哪一个要说明理由:**介数**衡量"必经之路",
**度**衡量"连接多",**特征向量**衡量"连到重要节点"。
原建议把四个都列了,这里只留最常用的介数,其余直接调 networkx。
"""
# TODO(v0.2)
raise NotImplementedError
def pagerank(G: Any, alpha: float = 0.85, max_iter: int = 100) -> Any:
"""PageRank(有向图的节点重要性)。"""
# TODO(v0.2)
raise NotImplementedError
def max_flow(G: Any, source: Any, sink: Any, capacity: str = "capacity") -> Any:
"""最大流 / 最小割(网络容量、瓶颈分析)。"""
# TODO(v0.2)
raise NotImplementedError
def community_detection(G: Any, method: str = "louvain", random_state: int | None = None) -> Any:
"""社区发现。
Returns
-------
`Result`:`data` 是 `{节点: 社区编号}`,`metrics={"modularity"}`。
**必须报模块度 Q**,否则社区划分好不好无从判断。
"""
# TODO(v0.2)
raise NotImplementedError
+234
View File
@@ -0,0 +1,234 @@
"""优化。
分层(对应原建议的 A/B/C 三级):
* **B 类(薄封装)**:`minimize` `linear_program` `integer_program` —— 底层 SciPy。
* **A 类(自己实现,S 档)**:`dynamic_programming` `knapsack_dp` `simulated_annealing`。
* **C 类(直接用库)**:NSGA-II 等,用 pymoo,本模块只提供 Pareto 前沿的**后处理**。
原建议里的 `maximize` 已删:它就是 `minimize` 取负,用 `sense="max"` 参数表达。
`value_iteration` / `policy_iteration` 也已删:与 `dynamic_programming` 同族,
第一版同时做三个属于过度设计。
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
__all__ = [
"dynamic_programming",
"integer_program",
"knapsack_dp",
"linear_program",
"minimize",
"pareto_front",
"simulated_annealing",
]
def minimize(
objective: Callable[..., float],
x0: Any = None,
bounds: Any = None,
constraints: Any = None,
method: str = "auto",
sense: str = "min",
random_state: int | None = None,
**kwargs: Any,
) -> Any:
"""连续优化(底层 `scipy.optimize.minimize`)。
Parameters
----------
method:
`"auto"` 时按"有无约束 / 有无梯度"自动选:
无约束 → BFGS;有边界 → L-BFGS-B;有约束 → SLSQP。
sense:
`"min"` / `"max"`。取 max 时内部对目标函数取负并还原结果,
调用方不用自己写负号(负号写错是这类题最常见的手滑)。
Returns
-------
`Result`:`params={"x": 最优解}`,`metrics={"objective": 目标值}`。
Notes
-----
**结果必须在论文里报"是否收敛"和"初值敏感性"**:
用一个 `x0` 跑出来的局部最优,和换 10 个随机初值跑出来的全局最优,
结论可能完全不同。返回值里带上 `success` 与 `message` 就是为了这个。
Examples
--------
>>> f = lambda v: (v[0] - 3) ** 2 + (v[1] + 1) ** 2
>>> r = minimize(f, x0=[0, 0])
>>> round(r.metrics["objective"], 6)
0.0
"""
# TODO(L1)
raise NotImplementedError
def linear_program(
c: Any,
A_ub: Any = None,
b_ub: Any = None,
A_eq: Any = None,
b_eq: Any = None,
bounds: Any = None,
sense: str = "min",
) -> Any:
"""线性规划(底层 `scipy.optimize.linprog`)。
Parameters
----------
c:
目标函数系数。`sense="max"` 时内部取负。
bounds:
`None` 表示默认 `x >= 0`;要无下界显式写 `(None, None)`。
**这是最容易踩的坑**:忘记声明变量非负,解出来的方案可能物理上不可行。
Examples
--------
>>> r = linear_program(c=[1, 2], A_ub=[[1, 1]], b_ub=[4], bounds=[(0, None)] * 2)
>>> r.metrics["objective"]
4.0
"""
# TODO(L1)
raise NotImplementedError
def integer_program(
c: Any,
integrality: Any,
A_ub: Any = None,
b_ub: Any = None,
A_eq: Any = None,
b_eq: Any = None,
bounds: Any = None,
sense: str = "min",
) -> Any:
"""整数规划(底层 `scipy.optimize.milp`)。
`integrality`:0 = 连续,1 = 整数,2 = 0-1 变量。
Notes
-----
整数规划比 LP 慢得多。变量数到几百以上时先想想能不能用
线性松弛 + 取整 + 邻域搜索近似,别硬等求解器。
"""
# TODO(L1)
raise NotImplementedError
def dynamic_programming(
states: Any,
actions: Any,
transition: Callable[..., Any],
reward: Callable[..., float],
horizon: int,
discount: float = 1.0,
) -> Any:
"""通用有限期动态规划(A 类,自己实现)。
Parameters
----------
transition:
`f(state, action, rng) -> next_state`。
reward:
`f(state, action, next_state) -> float`。
horizon:
期数。`discount < 1` 时按 `γ^t` 折算。
Returns
-------
`Result`:`data` 是"最优值函数表 + 最优策略表",`extras["policy"]` 可直接查。
Notes
-----
复杂度 `O(horizon × |S| × |A|)`。**状态空间是唯一瓶颈**——
如果 `|S| × holiday` 超过内存,要么精简状态离散化粒度,要么改成近似 DP。
Examples
--------
>>> # 背包问题用 knapsack_dp 更直接;本函数用于"逐期决策"类题
>>> r = dynamic_programming(states, actions, trans, rew, horizon=10) # doctest: +SKIP
"""
# TODO(L1) ★ S 档:配完整函数卡片
raise NotImplementedError
def knapsack_dp(weights: Any, values: Any, capacity: float) -> Any:
"""0-1 背包(动态规划模板,A 类,自己实现)。
复杂度 `O(n × capacity)`。`capacity` 在题面里是小数时先做单位缩放取整,
否则 DP 表会大到装不下。
Returns
-------
`Result`:`params={"selected": 选中的下标}`,`metrics={"total_value", "total_weight"}`。
Examples
--------
>>> r = knapsack_dp(weights=[1, 3, 4, 5], values=[1, 4, 5, 7], capacity=7)
>>> r.metrics["total_value"]
9.0
"""
# TODO(L1) ★ S 档
raise NotImplementedError
def simulated_annealing(
objective: Callable[..., float],
x0: Any,
bounds: Any = None,
temp0: float = 1.0,
cooling: float = 0.95,
n_iter: int = 1000,
sense: str = "min",
random_state: int | None = None,
**kwargs: Any,
) -> Any:
"""模拟退火(A 类,自己实现)。
Parameters
----------
cooling:
降温系数,每步 `T *= cooling`。`0.95` 配 `n_iter=1000` 是保守默认值。
temp0:
初始温度。**经验做法:取"随机扰动的典型劣化幅度"同量级**,
太大等于随机游走,太小等于爬山。
Returns
-------
`Result`:`extras["history"]` 是每步的最优值轨迹,可直接喂给 `viz.plot_convergence`。
Notes
-----
* 必须记录收敛曲线并画出来——论文里"算法收敛性分析"那一节就靠它。
* 结果对随机种子敏感,**至少跑 10 个种子**,报最好值 + 分布,
只报一个种子的结果是不可信的。
Examples
--------
>>> r = simulated_annealing(lambda v: sum(x * x for x in v), x0=[3.0, 4.0], random_state=0) # doctest: +SKIP
"""
# TODO(L1) ★ S 档:配完整函数卡片
raise NotImplementedError
def pareto_front(points: Any, sense: str = "min") -> Any:
"""从解集里筛出 Pareto 非支配解。
Returns
-------
`Result`:`params={"mask": 布尔掩码, "rank": 每个解的支配层}`。
Notes
-----
多目标问题的**后处理**入口。真正的 NSGA-II 求解器用 pymoo
(`pip install "mcm-toolkit[moo]"`),本函数只负责"给定一组解,挑出前沿"。
它也是画 Pareto 前沿图的必要前置步骤。
"""
# TODO(L1)
raise NotImplementedError
+214
View File
@@ -0,0 +1,214 @@
"""敏感性分析与鲁棒性检验。
★ 建议做成招牌模块。原建议这一节方向完全正确,只是函数名偏多,
已合并同义项、删掉 `grid_sweep`(`parameter_sweep` 的特例)。
论文里的"敏感性分析"一节基本就是这四个函数的组合:
局部扰动 → OAT 单因素扫描 → 多因素参数扫描 → 全局 Sobol/Morris。
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
__all__ = [
"local_sensitivity",
"one_at_a_time",
"parameter_sweep",
"robustness_test",
]
def local_sensitivity(
model: Callable[..., Any],
params: Any,
step: float = 1e-4,
relative: bool = True,
**kwargs: Any,
) -> Any:
"""局部敏感性:各参数的偏导 / 弹性。
Parameters
----------
relative:
`True` 时算弹性 `(∂y/∂p)·(p/y)`——**无量纲,可以跨参数比较**,
这是局部敏感性的正确做法;`False` 时是原始偏导(有量纲,只有同量纲参数之间可比)。
Returns
-------
`Result`:`data` 是参数 → 弹性 的表,按绝对值降序。
Notes
-----
有限差分的步长是精度与截断误差的权衡:太大截断误差大,
太小被浮点误差淹没。`1e-4` 是稳健默认值;参数本身量级差很多时用相对步长。
Examples
--------
>>> s = local_sensitivity(lambda p: p["a"] ** 2 * p["b"], {"a": 2.0, "b": 3.0}) # doctest: +SKIP
"""
# TODO(L1) ★ S 档
raise NotImplementedError
def one_at_a_time(
model: Callable[..., Any],
params: Any,
parameter: str | None = None,
values: Any = None,
n_points: int = 20,
span: float = 0.5,
) -> Any:
"""OAT(一次只动一个参数)。
Parameters
----------
parameter:
`None` 时对所有参数依次扫描。
span:
相对基准值的扫描范围,`0.5` 表示 ±50%。
Returns
-------
`Result`:`data` 是长表 `(parameter, value, output)`,配 `viz.plot` 直接出蛛网图/敏感性曲线。
Notes
-----
OAT 的**已知缺陷**:它看不见参数之间的交互作用。
如果两个参数是相乘关系(很常见),OAT 会严重低估整体敏感性——
论文里必须说明这一点,并补一个全局敏感性作为交叉验证。
"""
# TODO(L1)
raise NotImplementedError
def parameter_sweep(
model: Callable[..., Any],
ranges: dict[str, Any],
n_points: int = 20,
method: str = "grid",
n_samples: int = 1000,
random_state: int | None = None,
) -> Any:
"""多参数扫描(合并了原建议的 `grid_sweep` 与 `random_parameter_sweep`)。
Parameters
----------
ranges:
`{"alpha": (0.1, 0.9), "beta": (0, 5)}`。
method:
`"grid"`:笛卡尔积,点数 = `n_points^d`,**维度一高就爆**(d=5, n=20 是 320 万次);
`"lhs"`:拉丁超立方,维度高时用这个;
`"random"`:纯随机。
Returns
-------
`Result`:`data` 是 `(n_points^d, d + 1)` 的结果表,
`params["best"]` 是让目标最优的参数组合。
Notes
-----
`d >= 4` 时默认改走 `"lhs"`,否则 `n_points` 会被指数放大到跑不完。
Examples
--------
>>> r = parameter_sweep(lambda a, b: -(a - 0.3) ** 2 - (b - 2) ** 2,
... {"a": (0, 1), "b": (0, 4)}, n_points=11) # doctest: +SKIP
"""
# TODO(L0)
raise NotImplementedError
def robustness_test(
model: Callable[..., Any],
scenarios: Any = None,
params: Any = None,
perturbation: float = 0.1,
n_samples: int = 1000,
random_state: int | None = None,
) -> Any:
"""鲁棒性检验(合并了原建议的 `perturbation_test`)。
在参数上施加 ±`perturbation` 的随机扰动,看结论是否翻盘。
Returns
-------
`Result`:`metrics` 含 `"rank_stability"`(原结论在前 X% 的比例)、
`"conclusion_flip_rate"`(结论反转频率)。**这两个数字才是审稿人想看的。**
Notes
-----
鲁棒性检验和敏感性分析不是一回事:
敏感性问"参数变了输出变多少",鲁棒性问"参数变了**结论**还成立吗"。
论文里两节都要有,别混着写。
Examples
--------
>>> r = robustness_test(model, params={"alpha": 0.5}, random_state=0) # doctest: +SKIP
"""
# TODO(L1)
raise NotImplementedError
def sobol_analyze(
model: Callable[..., Any],
bounds: Any,
n_samples: int = 1024,
calc_second_order: bool = False,
random_state: int | None = None,
) -> Any:
"""Sobol 全局敏感性(底层 SALib,自己不要重写)。
Parameters
----------
bounds:
`{"alpha": (0, 1), "beta": (0, 5)}` 形式的参数上下界。
n_samples:
SALib 的基样本数 N;**实际模型调用次数是 `N·(2d + 2)`**
(二阶时更多),d=5、N=1024 就是 12288 次调用。先估一下模型单次耗时。
Returns
-------
`Result`:`data` 含 `S1`(一阶)、`ST`(总效应)、`S2`(二阶,可选)三列。
Notes
-----
* `ST - S1` 的差距就是**交互作用**的大小——这正是 OAT 看不见的东西。
两者差得多,说明不能只看单因素。
* 需要 `pip install "mcm-toolkit[sensitivity]"`;缺 SALib 时本函数给出
可读的安装提示,而不是裸抛 `ModuleNotFoundError`。
* 费用高:**先做 OAT 缩小参数范围,再对少数关键参数做 Sobol**,
不要一上来就对全部参数做全局敏感性。
Examples
--------
>>> r = sobol_analyze(model, bounds={"a": (0, 1), "b": (0, 5)}) # doctest: +SKIP
"""
# TODO(L1) ★ S 档:配完整函数卡片
raise NotImplementedError
def morris_analyze(
model: Callable[..., Any],
bounds: Any,
n_trajectories: int = 10,
n_levels: int = 4,
random_state: int | None = None,
) -> Any:
"""Morris 筛选法(底层 SALib)。
调用次数只与轨迹数成正比(`(d + 1) · n_trajectories`),比 Sobol 便宜一个量级,
适合**参数多但只需要找出"哪几个重要"**的筛选阶段。
Returns
-------
`Result`:`data` 含 `mu`、`mu_star`(推荐看的)、`sigma` 三列。
`sigma` 大表示该参数的效应是非线性的或与其他参数有交互。
Examples
--------
>>> r = morris_analyze(model, bounds={"a": (0, 1), "b": (0, 5)}) # doctest: +SKIP
"""
# TODO(L1)
raise NotImplementedError
+218
View File
@@ -0,0 +1,218 @@
"""随机模拟:Monte Carlo / 马尔可夫 / 随机游走 / 排队。
★ 本模块的 `monte_carlo` 与 `monte_carlo_integral` 是 S 档,建议自己实现。
原建议里的 `brownian_motion` `geometric_brownian_motion` 已删(niche),
`simulate_queue` 也已删(等于自己写一个离散事件仿真引擎,成本远高于收益)。
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
__all__ = [
"markov_simulate",
"markov_steady_state",
"mm1_metrics",
"monte_carlo",
"monte_carlo_integral",
"monte_carlo_probability",
"random_walk",
]
def monte_carlo(
sampler: Callable[..., Any],
simulator: Callable[..., Any],
n: int = 10000,
random_state: int | None = None,
**kwargs: Any,
) -> Any:
"""通用蒙特卡洛:采样 → 仿真 → 汇总。
Parameters
----------
sampler:
`f(rng, **kwargs) -> 一组参数`,例如从先验分布里抽参数。
simulator:
`f(params) -> float`,例如"给定参数下的年收益"。
n:
模拟次数。**必须向量化**:`simulator` 内部不要写 Python 循环,
否则 10^4 次能跑几分钟,10^6 次直接放弃。
Returns
-------
`Result`:`data` 是 `n` 个结果组成的数组,
`metrics={"mean", "std", "median", "p05", "p95", "ci_low", "ci_high"}`。
Notes
-----
比赛里 Monte Carlo 的收敛速度是 `O(1/√n)`——
想少一半误差要 4 倍样本。所以论文里要给"区间随 n 收敛"的图,
证明 `n` 已经够大(这也是 `viz.plot_convergence` 的典型用法)。
Examples
--------
>>> import numpy as np
>>> r = monte_carlo(lambda rng, **k: rng.normal(0, 1), np.mean, n=1000, random_state=0)
"""
# TODO(L1) ★ S 档
raise NotImplementedError
def monte_carlo_probability(
event: Callable[..., bool],
sampler: Callable[..., Any],
n: int = 10000,
confidence: float = 0.95,
random_state: int | None = None,
) -> Any:
"""估计事件发生概率,并给出区间。
Returns
-------
`Result`:`metrics={"probability", "ci_low", "ci_high", "std_error"}`。
Notes
-----
概率估计的相对误差 ≈ `1/√(n·p)`。**小概率事件(p < 1e-3)用朴素 MC 是不可行的**,
需要重要性采样或方差缩减——遇到这类题要提前意识到,别硬跑。
Examples
--------
>>> import numpy as np
>>> r = monte_carlo_probability(lambda v: v > 0, lambda rng, **k: rng.normal(0, 1), n=2000, random_state=0)
>>> abs(r.metrics["probability"] - 0.5) < 0.1
True
"""
# TODO(L1) ★ S 档
raise NotImplementedError
def monte_carlo_integral(
func: Callable[..., Any],
bounds: Any,
n: int = 10000,
random_state: int | None = None,
) -> Any:
"""用蒙特卡洛求多重积分。
Parameters
----------
bounds:
`[(lo, hi), ...]`,维度即积分重数。
Notes
-----
高维积分是 MC 相对网格法的**唯一绝对优势**:网格法误差 `O(h^{-d})` 随维度爆炸,
MC 始终是 `O(1/√n)`。三维以上优先考虑 MC。
Examples
--------
>>> import numpy as np
>>> r = monte_carlo_integral(lambda X: X[:, 0] ** 2, bounds=[(0, 1)], n=200000, random_state=0)
>>> abs(r.data - 1 / 3) < 0.01
True
"""
# TODO(L1) ★ S 档:配完整函数卡片
raise NotImplementedError
def random_walk(
n_steps: int,
n_paths: int = 1,
step_sampler: Callable[..., Any] | None = None,
start: float = 0.0,
random_state: int | None = None,
) -> Any:
"""随机游走轨迹。
Parameters
----------
step_sampler:
`f(rng, size) -> 增量`,默认为 ±1 等概率(对称随机游走)。
Returns
-------
`Result`:`data` 形状 `(n_paths, n_steps + 1)`。
Examples
--------
>>> r = random_walk(n_steps=100, n_paths=5, random_state=0)
>>> r.data.shape
(5, 101)
"""
# TODO(L1)
raise NotImplementedError
def markov_simulate(
transition: Any,
initial: Any,
steps: int = 100,
n_paths: int = 1,
random_state: int | None = None,
) -> Any:
"""马尔可夫链模拟,返回状态序列。"""
# TODO(L1)
raise NotImplementedError
def markov_steady_state(transition: Any, max_iter: int = 10000, tol: float = 1e-10) -> Any:
"""平稳分布。
Parameters
----------
transition:
行随机矩阵(每行和为 1)。
Returns
-------
`Result`:`data` 是平稳分布向量,`metrics={"converged", "iterations"}`。
Notes
-----
用迭代法而不是特征分解:迭代法能顺便告诉你"收敛快不快",
而收敛速度本身就是论文里值得写一句的性质。
**不可约 + 非周期**才有唯一平稳分布,周期链会不收敛,要检查。
Examples
--------
>>> import numpy as np
>>> P = np.array([[0.9, 0.1], [0.2, 0.8]])
>>> r = markov_steady_state(P)
>>> bool(np.allclose(r.data, [2 / 3, 1 / 3], atol=1e-6))
True
"""
# TODO(L1)
raise NotImplementedError
def mm1_metrics(arrival_rate: float, service_rate: float) -> Any:
"""M/M/1 排队系统的闭式指标。
Parameters
----------
arrival_rate, service_rate:
到达率 λ 与服务率 μ。`λ >= μ` 时系统不稳定,队列长度无界——
这种情况下函数返回 `success=False` 并说明原因,而不是返回 `inf`。
Returns
-------
`Result`:`metrics={"rho", "L", "Lq", "W", "Wq"}`
(系统利用率、系统内平均顾客数、队列长度、平均逗留时间、平均等待时间)。
Notes
-----
M/M/1 是唯一值得手写的排队模型:公式是闭式的,10 行搞定。
多服务员(M/M/c)和一般服务时间分布(M/G/1)在美赛出现频率极低,
真遇到时现场用仿真做,不做通用封装。
Examples
--------
>>> r = mm1_metrics(arrival_rate=2, service_rate=3)
>>> round(r.metrics["rho"], 4)
0.6667
"""
# TODO(L1)
raise NotImplementedError
+32
View File
@@ -0,0 +1,32 @@
"""stats —— 描述统计、统计推断、bootstrap。
S 级模块。`bootstrap_ci` 建议自己实现一遍(S 档,配完整函数卡片):
它是赛题里"没有解析解时怎么给区间"的通用答案。
"""
from .bootstrap import bootstrap_ci, bootstrap_statistic
from .descriptive import coefficient_of_variation, describe, zscore
from .inference import (
anova,
chi_square_test,
confidence_interval,
correlation_matrix,
fit_distribution,
goodness_of_fit,
t_test,
)
__all__ = [
"describe",
"zscore",
"coefficient_of_variation",
"correlation_matrix",
"confidence_interval",
"t_test",
"anova",
"chi_square_test",
"fit_distribution",
"goodness_of_fit",
"bootstrap_statistic",
"bootstrap_ci",
]
+71
View File
@@ -0,0 +1,71 @@
"""Bootstrap 重采样。
★ S 档模块:建议自己完整实现一遍。
它是"没有解析解时怎么给区间"的通用答案,赛题里出现频率极高,
而且实现只有几十行——投入产出比最高的一个算法。
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
__all__ = ["bootstrap_ci", "bootstrap_statistic"]
def bootstrap_statistic(
x: Any,
statistic: Callable[..., float],
n_resamples: int = 10000,
random_state: int | None = None,
) -> Any:
"""返回 `n_resamples` 个重采样统计量组成的数组。
Examples
--------
>>> import numpy as np
>>> vals = bootstrap_statistic([1, 2, 3, 4, 5], np.mean, n_resamples=200, random_state=0)
"""
# TODO(L1) ★ S 档
raise NotImplementedError
def bootstrap_ci(
x: Any,
statistic: Callable[..., float] | None = None,
confidence: float = 0.95,
n_resamples: int = 10000,
method: str = "percentile",
random_state: int | None = None,
) -> Any:
"""用重采样给出任意统计量的置信区间。
Parameters
----------
statistic:
任意把样本映射成标量的函数,默认 `np.mean`。
换成 `np.median` / `np.std` / 自定义的 `lambda s: s.max() - s.min()` 都能用,
这是 bootstrap 相对于 `mean ± z·s/√n` 的核心优势。
method:
`"percentile"`:直接取经验分布的 α/2 与 1-α/2 分位(最常用,默认);
`"bca"`:偏差校正 + 加速,区间更准但慢,样本有偏时用。
Returns
-------
`Result`:`params={"estimate", "low", "high", "confidence", "n_resamples"}`。
Notes
-----
* 重采样时必须**按块处理**,不要一次生成 `(10000, n)` 的整矩阵(n 大时会爆内存)。
* 样本量 n < 10 时 bootstrap 不可靠,先补数据。
* 比赛写论文时要注明:n_resamples、seed、用的是 percentile 还是 BCa。
Examples
--------
>>> import numpy as np
>>> r = bootstrap_ci([1, 2, 3, 4, 5], np.median, random_state=0)
>>> r.params["low"] < 3 < r.params["high"]
True
"""
# TODO(L0) ★ S 档:配完整函数卡片
raise NotImplementedError
+40
View File
@@ -0,0 +1,40 @@
"""描述统计。
每个函数都只是"一句话数学含义 + 一句话什么时候用"(A 档),
不单独写文档卡片。
"""
from __future__ import annotations
from typing import Any
__all__ = ["coefficient_of_variation", "describe", "zscore"]
def describe(x: Any, weights: Any | None = None) -> Any:
"""一张表看完集中趋势、离散程度、偏度峰度、缺失情况。
`weights` 非空时给出加权均值与加权标准差(权重是"频数"或"重要性"时用)。
Returns
-------
DataFrame,行是指标,列是被描述的变量。
Examples
--------
>>> describe(df).loc["mean"] # doctest: +SKIP
"""
# TODO(L0)
raise NotImplementedError
def zscore(x: Any, ddof: int = 0) -> Any:
"""标准化到 z 分数,顺带回答了"这个值离均值几个标准差"。"""
# TODO(L0)
raise NotImplementedError
def coefficient_of_variation(x: Any) -> float:
"""变异系数 `std / mean` —— 跨量纲比较"谁更波动"时的唯一正确做法。"""
# TODO(L1)
raise NotImplementedError
+114
View File
@@ -0,0 +1,114 @@
"""统计推断:区间估计、假设检验、分布拟合。
原建议里的 `mean_test`/`proportion_test`/`distribution_pdf`/`distribution_cdf`/
`distribution_ppf`/`sample_distribution` 都是 scipy 的一行转发,已删。
需要 pdf/cdf/ppf 时,用 `fit_distribution(...).dist.pdf(...)` 即可。
"""
from __future__ import annotations
from typing import Any
__all__ = [
"anova",
"chi_square_test",
"confidence_interval",
"correlation_matrix",
"fit_distribution",
"goodness_of_fit",
"t_test",
]
def correlation_matrix(data: Any, method: str = "pearson", min_periods: int = 1) -> Any:
"""相关矩阵 + 显著性星标。
Returns
-------
DataFrame,值 = 相关系数;`min_periods` 控制缺失值容差。
Examples
--------
>>> correlation_matrix(df).round(3) # doctest: +SKIP
"""
# TODO(L0)
raise NotImplementedError
def confidence_interval(
x: Any,
confidence: float = 0.95,
method: str = "normal",
) -> tuple[float, float]:
"""均值或比例的置信区间。
Parameters
----------
method:
`"normal"`:`mean ± z·s/√n`(n 大时用);
`"t"`:小样本用 t 分布(更保守,n < 30 建议用这个);
`"proportion"`:比例的标准误 `√(p(1-p)/n)`。
Examples
--------
>>> lo, hi = confidence_interval([1, 2, 3, 4, 5], confidence=0.95)
"""
# TODO(L0)
raise NotImplementedError
def t_test(a: Any, b: Any | None = None, value: float = 0.0, equal_var: bool = False, paired: bool = False, alpha: float = 0.05) -> Any:
"""t 检验。`b=None` 时是单样本检验。
Returns
-------
dict:`{"statistic", "pvalue", "df", "significant", "conclusion"}`。
`conclusion` 是一句能直接抄进论文的中文结论。
"""
# TODO(L1)
raise NotImplementedError
def anova(*groups: Any, alpha: float = 0.05) -> Any:
"""单因素方差分析:多组均值是否相同。
Returns
-------
dict:`{"F", "pvalue", "conclusion"}`。
"""
# TODO(L1)
raise NotImplementedError
def chi_square_test(observed: Any, expected: Any | None = None, alpha: float = 0.05) -> Any:
"""卡方检验:观察频数与期望频数是否一致(分类变量的独立性 / 拟合优度)。"""
# TODO(L1)
raise NotImplementedError
def fit_distribution(x: Any, candidates: tuple[str, ...] = ("norm", "lognorm", "gamma", "expon", "uniform")) -> Any:
"""对候选分布逐个拟合,按 AIC(或 KS 统计量)排序。
Returns
-------
`Result`:`data` 是排好序的候选表;`params["best"]` 是最优分布名,
`extras["dist"]` 是可直接调 `.pdf / .cdf / .ppf / `.rvs` 的 scipy 分布对象。
Notes
-----
这是"反正态题"的标准做法:先看图 → 候选分布拟合 → 比较 AIC → 做 KS 检验。
Examples
--------
>>> r = fit_distribution(data) # doctest: +SKIP
>>> r.params["best"] # doctest: +SKIP
'lognorm'
"""
# TODO(L1)
raise NotImplementedError
def goodness_of_fit(x: Any, dist: str = "norm", method: str = "ks") -> Any:
"""拟合优度检验:`"ks"`(连续)/ `"chi2"`(分箱)。"""
# TODO(L1)
raise NotImplementedError
+21
View File
@@ -0,0 +1,21 @@
"""viz —— 出图与出表(论文产出的最后一公里)。
原建议把 `visualization/` 拆成 5 个文件、13 个 `plot_*` 函数。
但每个 `plot_*` 本质上都是"在 ax 上加两行",拆开后
**改一次全局样式要改 13 处**,而且 13 个名字谁也记不住。
合并为 `plot(kind=...)` + 3 个有真实逻辑的专用图。
"""
from .export import save_figure, save_table
from .plot import plot, plot_convergence, plot_correlation_matrix, plot_residuals
from .style import mcm_style
__all__ = [
"mcm_style",
"plot",
"plot_correlation_matrix",
"plot_residuals",
"plot_convergence",
"save_figure",
"save_table",
]
+74
View File
@@ -0,0 +1,74 @@
"""图表导出。
工具箱的最终消费者是**论文**,所以这一层不是"附赠功能",
而是原建议里最被低估的部分(它只给了一句 `save_figure`)。
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
__all__ = ["save_figure", "save_table"]
def save_figure(
fig: Any,
path: str | Path,
dpi: int = 300,
formats: tuple[str, ...] = ("png", "pdf"),
bbox_inches: str = "tight",
transparent: bool = False,
) -> list[Path]:
"""保存图,默认同时出 PNG(预览用)和 PDF(LaTeX 插图用)。
Parameters
----------
formats:
美赛论文如果用 LaTeX 排版,**矢量图(PDF/EPS)比 PNG 清晰得多**,
所以默认两种都出。
Returns
-------
实际写出的文件路径列表。
Notes
-----
建议配合项目的 figure 目录约定:`figures/fig{序号}_{短名}.pdf`,
图号与正文引用一一对应,最后核对时不容易错位。
"""
# TODO(L0)
raise NotImplementedError
def save_table(
df: Any,
path: str | Path,
fmt: tuple[str, ...] = ("csv", "latex", "markdown"),
caption: str | None = None,
label: str | None = None,
float_format: str = "%.4f",
index: bool = True,
) -> list[Path]:
r"""导出结果表,默认同时出 CSV + LaTeX + Markdown。
这是**节省时间最多的一个函数**:`compare_models`、`parameter_sweep`、
`topsis` 的结果都能直接 `save_table(r.data, "tab3_model_compare")`,
然后把 LaTeX 片段粘进论文,不用手工敲表格。
Parameters
----------
caption / label:
写进 LaTeX 的 `\caption{}` 与 `\label{}`,
表号与正文 `\ref{}` 的对应关系就不会断。
Returns
-------
实际写出的文件路径列表。
Examples
--------
>>> paths = save_table(df, "tables/tab2", caption="模型对比") # doctest: +SKIP
"""
# TODO(L1)
raise NotImplementedError
+79
View File
@@ -0,0 +1,79 @@
"""绘图。
`plot` 是一个通用入口,按 `kind` 分派;三个专用图有真实的额外逻辑
(显著性星标、残差诊断、收敛曲线),所以单独保留。
"""
from __future__ import annotations
from typing import Any
__all__ = ["plot", "plot_convergence", "plot_correlation_matrix", "plot_residuals"]
def plot(
x: Any = None,
y: Any = None,
kind: str = "line",
ax: Any = None,
labels: Any = None,
title: str | None = None,
xlabel: str | None = None,
ylabel: str | None = None,
**kwargs: Any,
) -> Any:
"""通用绘图入口。
Parameters
----------
kind:
`"line"` / `"scatter"` / `"hist"` / `"box"` / `"bar"` / `"heatmap"` / `"pie"`。
ax:
传入已有 axes 就在上面画(多子图场景);`None` 时新建 figure。
Returns
-------
符合"创建—绘制—保存"三段的 `(fig, ax)`,交给 :func:`save_figure`。
Examples
--------
>>> fig, ax = plot([1, 2, 3], [1, 4, 9], kind="line") # doctest: +SKIP
"""
# TODO(L1)
raise NotImplementedError
def plot_correlation_matrix(data: Any, method: str = "pearson", annot: bool = True, significance: bool = True) -> Any:
"""相关矩阵热图。
`significance=True` 时在不显著的格子(`p > 0.05`)上打叉。
这比只画颜色有意义得多——**不显著的相关性画出来是误导**。
"""
# TODO(L1)
raise NotImplementedError
def plot_residuals(y_true: Any, y_pred: Any, n_features: int | None = None) -> Any:
"""残差诊断四联图:残差-拟合值、Q-Q 图、直方图、残差-序。
拟合完模型**第一件事就是看这个**:
* 残差-拟合值有漏斗形 → 异方差,考虑加权最小二乘或对数变换;
* Q-Q 图偏离直线 → 正态性假设不成立,区间估计要改用 bootstrap;
* 残差有周期性 → 漏了时间/结构项。
这三个诊断结论是论文"模型检验"一节的素材,比只报一个 R² 有说服力。
"""
# TODO(L1)
raise NotImplementedError
def plot_convergence(history: Any, ax: Any = None, labels: Any = None, log: bool = False, reference: float | None = None) -> Any:
"""收敛曲线。
`history` 可以是迭代轨迹数组、也接受 `Result.extras["history"]`
(`simulated_annealing` 与 `monte_carlo` 都会填这个字段)。
`reference` 传入解析解时会把真值画成虚线,
用于"MC 积分误差随 n 收敛"这类图——这是证明 `n` 选得合理的标准做法。
"""
# TODO(L1)
raise NotImplementedError
+40
View File
@@ -0,0 +1,40 @@
"""统一图样式。
美赛送稿要求:**双栏排版、字号不能太小、图必须能在黑白打印下区分**。
把这些要求固化成一次调用,比每次手调 rcParams 靠谱。
"""
from __future__ import annotations
from typing import Any
__all__ = ["mcm_style"]
def mcm_style(palette: str = "colorblind", dpi: int = 300, font_size: float = 9.0, **overrides: Any) -> Any:
"""设置全局绘图样式,返回 matplotlib rcParams(便于 `with` 临时切换)。
Parameters
----------
palette:
`"colorblind"`(默认,色盲友好且黑白打印可辨)/ `"muted"` / `"deep"`。
dpi:
默认 300 —— 美赛送稿的 DPI 底线,低于这个值的图放大后会糊。
font_size:
默认 9pt。**双栏排版下字号小于 8pt 就会被评审认为是"塞图"**。
Notes
-----
* 中文标签需要设置中文字体(`SimHei` / `Microsoft YaHei`),
否则 matplotlib 会画一堆方块;本函数会尝试自动选择可用中文字体。
* 论文用图建议**只用线型 + 标记区分**,不要只靠颜色。
Examples
--------
>>> import matplotlib
>>> mcm_style()
>>> matplotlib.rcParams["figure.dpi"]
300.0
"""
# TODO(L1)
raise NotImplementedError