108 lines
5.1 KiB
Python
108 lines
5.1 KiB
Python
# ============================================================
|
||
# 蒙特卡洛可配置分布/相关性/置信区间黄金测试（plan.md §9.9 / 矩阵 88 行剩余项）
|
||
# 覆盖：分布可配置（normal/uniform/triangular）、需求-效率相关度、
|
||
# 置信区间输出、默认行为与种子确定性兼容。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from server.aps_domain.robustness import run_monte_carlo
|
||
from server.state.seed import seed_world
|
||
|
||
|
||
def _demo_world() -> dict:
|
||
"""种子世界（RULE 引擎可排产，种子确定性）。"""
|
||
return seed_world()
|
||
|
||
|
||
def test_default_behavior_stable_and_ci_present():
|
||
"""默认参数：输出含既有字段 + 新增均值/标准差/置信区间/配置元信息。"""
|
||
report = run_monte_carlo(_demo_world(), trials=12, seed=2026)
|
||
dist = report["distribution"]
|
||
assert report["method"] == "fixed-seed-monte-carlo"
|
||
assert dist["tardinessP50"] <= dist["tardinessP90"] <= dist["tardinessP95"] <= dist["tardinessMax"]
|
||
assert "tardinessMean" in dist and "tardinessStd" in dist
|
||
lo, hi = dist["tardinessCI"]
|
||
assert lo <= dist["tardinessMean"] <= hi
|
||
assert dist["confidenceLevel"] == 0.95
|
||
cfg = report["config"]
|
||
assert cfg["correlation"] == 0.0
|
||
assert cfg["distributions"]["quantity"]["dist"] == "normal"
|
||
assert cfg["distributions"]["shockProbability"] == 0.15
|
||
assert "outcomes" in report
|
||
|
||
|
||
def test_seed_determinism_preserved_with_defaults():
|
||
"""默认配置下同种子输出完全一致（既有测试语义兼容）。"""
|
||
a = run_monte_carlo(_demo_world(), trials=8, seed=1234)
|
||
b = run_monte_carlo(_demo_world(), trials=8, seed=1234)
|
||
assert a["distribution"] == b["distribution"]
|
||
assert a["outcomes"] == b["outcomes"]
|
||
|
||
|
||
def test_uniform_distribution_bounds_and_determinism():
|
||
"""可配置 uniform 分布：样本都在 [low, high] 内且种子确定。"""
|
||
dist_cfg = {
|
||
"quantity": {"dist": "uniform", "low": 0.90, "high": 1.10},
|
||
"efficiency": {"dist": "uniform", "low": 0.95, "high": 1.05},
|
||
"shockProbability": 0.0,
|
||
}
|
||
a = run_monte_carlo(_demo_world(), trials=16, seed=7, distributions=dist_cfg)
|
||
b = run_monte_carlo(_demo_world(), trials=16, seed=7, distributions=dist_cfg)
|
||
assert a["outcomes"] == b["outcomes"]
|
||
for outcome in a["outcomes"]:
|
||
assert 0.90 <= outcome["quantityScale"] <= 1.10
|
||
assert 0.95 <= outcome["efficiencyScale"] <= 1.05
|
||
assert a["config"]["distributions"]["quantity"]["dist"] == "uniform"
|
||
|
||
|
||
def test_triangular_distribution_mode():
|
||
"""可配置 triangular 分布：样本在区间内且众数附近密度高（粗验）。"""
|
||
dist_cfg = {
|
||
"quantity": {"dist": "triangular", "low": 0.80, "high": 1.20, "mode": 1.0},
|
||
"shockProbability": 0.0,
|
||
}
|
||
report = run_monte_carlo(_demo_world(), trials=24, seed=99, distributions=dist_cfg)
|
||
scales = [o["quantityScale"] for o in report["outcomes"]]
|
||
assert all(0.80 <= s <= 1.20 for s in scales)
|
||
near_mode = sum(1 for s in scales if 0.95 <= s <= 1.05)
|
||
assert near_mode >= len(scales) * 0.4
|
||
|
||
|
||
def test_positive_correlation_moves_scales_together():
|
||
"""相关度 rho>0：quantity 与 efficiency 扰动同向（样本相关性为正）。"""
|
||
dist_cfg = {"shockProbability": 0.0}
|
||
report = run_monte_carlo(_demo_world(), trials=48, seed=42,
|
||
distributions=dist_cfg, correlation=0.8)
|
||
qs = [o["quantityScale"] for o in report["outcomes"]]
|
||
es = [o["efficiencyScale"] for o in report["outcomes"]]
|
||
mean_q, mean_e = sum(qs) / len(qs), sum(es) / len(es)
|
||
cov = sum((q - mean_q) * (e - mean_e) for q, e in zip(qs, es)) / len(qs)
|
||
std_q = (sum((q - mean_q) ** 2 for q in qs) / len(qs)) ** 0.5
|
||
std_e = (sum((e - mean_e) ** 2 for e in es) / len(es)) ** 0.5
|
||
sample_rho = cov / (std_q * std_e) if std_q > 0 and std_e > 0 else 0.0
|
||
assert sample_rho > 0.3, f"期望正相关，实际 {sample_rho:.3f}"
|
||
assert report["config"]["correlation"] == 0.8
|
||
|
||
|
||
def test_zero_correlation_keeps_independence():
|
||
"""相关度 rho=0：样本相关性接近 0（默认独立性语义不变）。"""
|
||
dist_cfg = {"shockProbability": 0.0}
|
||
report = run_monte_carlo(_demo_world(), trials=48, seed=42,
|
||
distributions=dist_cfg, correlation=0.0)
|
||
qs = [o["quantityScale"] for o in report["outcomes"]]
|
||
es = [o["efficiencyScale"] for o in report["outcomes"]]
|
||
mean_q, mean_e = sum(qs) / len(qs), sum(es) / len(es)
|
||
cov = sum((q - mean_q) * (e - mean_e) for q, e in zip(qs, es)) / len(qs)
|
||
std_q = (sum((q - mean_q) ** 2 for q in qs) / len(qs)) ** 0.5
|
||
std_e = (sum((e - mean_e) ** 2 for e in es) / len(es)) ** 0.5
|
||
sample_rho = cov / (std_q * std_e) if std_q > 0 and std_e > 0 else 0.0
|
||
assert abs(sample_rho) < 0.3
|
||
|
||
|
||
def test_shock_probability_configurable():
|
||
"""shockProbability=0 时无产能冲击事件。"""
|
||
dist_cfg = {"shockProbability": 0.0}
|
||
report = run_monte_carlo(_demo_world(), trials=12, seed=5, distributions=dist_cfg)
|
||
assert all(o["capacityShockLineId"] is None for o in report["outcomes"])
|
||
assert report["config"]["distributions"]["shockProbability"] == 0.0
|