# -*- coding: utf-8 -*- """O2–O4 统计方法学广度正式 tc 套件（2026-08-05，_DATA_VERSION v2.30，TestClient 真实 API）。 O2 幼年–成年遗传相关 / O3 非线性基因组预测 / O4 约束选择指数 —— 前后端+引擎落地后的 API 回归护栏： [O2] POST /type-b-heredity env_dim=stage（发育阶段当"环境"）→ 判别 fixture（两全同胞家系 + 共享遗传效应跨阶段）落库 r_g>0.5（童期可借力成株）、n_common=全树数、envs_json 双阶段。 [O3] POST /gblup/run method=rrblup/bayesb → 落库 PredictionModel(method=RRBLUP/BayesB)； bayesb 同 seed 两次运行 EBV 逐位一致（可复现，MLOps 铁律）；rrblup 与 gblup EBV 相关>0.9（对偶）。 POST /cv/run method=rrblup/bayesb → KFCV/RRBLUP、KFCV/BayesB 完整 CV（mean_pearson∈[0,1]）。 [O4] POST /selection-index method=restricted（+economic_weights）→ SI_RES + restricted/constraint_mode= ΔG=0 标记 + 受限性状 b≈0（对角 G 时 ΔG=0 ⇔ b=0，判别投影生效）；method=smith_hazel（economic） → asc 性状 direction 翻转 b<0、desc 性状 b>0。 fixture（镜像 e2e_o234_20260805 判别结构）：4 种质亲本（两全同胞家系）+ 2 组合 × 6 株 = 12 株； 基因型数据集 12 SNP 标记（确保多态）全株分型；cX(desc)/cY(asc) 表型独立（P 近对角，r_g=0 可控）； ABLUP 批次手插（reliability=0 → Calo 分母 0 → r_g=0 → G 精确对角，restricted ΔG=0 ⇔ b_受限=0）。 依赖: Redis + PG 正常（TestClient 走真实 lifespan）。运行后自动清理。 """ import os os.environ["ENVIRONMENT"] = "dev" os.environ["PYTHONUTF8"] = "1" import sys, asyncio # noqa: E402 sys.path.insert(0, r"d:\dpb\dpb\backend") import numpy as np # noqa: E402 import main # noqa: E402 from fastapi.testclient import TestClient # noqa: E402 from sqlalchemy import delete, select # noqa: E402 from app.core.database import create_async_engine_and_session # noqa: E402 from app.api.v1.module_system.user.model import UserModel # noqa: E402 (注册 mapper) from app.api.v1.module_bre.target.model import TargetModel # noqa: E402 from app.api.v1.module_bre.trait.model import TraitModel # noqa: E402 from app.api.v1.module_bre.germplasm.model import BreedingGermplasmModel # noqa: E402 from app.api.v1.module_bre.cross_combination.model import CrossCombinationModel # noqa: E402 from app.api.v1.module_bre.tree.model import TreeModel # noqa: E402 from app.api.v1.module_bre.trait_observation.model import TraitObservationModel # noqa: E402 from app.api.v1.module_bre.genotype_dataset.model import GenotypingDatasetModel # noqa: E402 from app.api.v1.module_bre.genotype_sample.model import GenotypeSampleModel # noqa: E402 from app.api.v1.module_bre.genotype_call.model import GenotypeCallModel # noqa: E402 from app.api.v1.module_bre.marker.model import MarkerModel # noqa: E402 from app.api.v1.module_bre.statistics.model import ( # noqa: E402 PredictionModel, PredictionValueModel, TypeBResultModel, CvResultModel, CvFoldModel, SelectionIndexModel, StatisticsJobModel, ) from scripts.breeding_stats import blup, genomic # noqa: E402 create_app = main.create_app TOKEN = None ok, fail = 0, 0 PREFIX = "O234" SEED = 20260805 N_FAM = 2 N_PER = 6 N = N_FAM * N_PER M = 12 GT = {0: "0/0", 1: "0/1", 2: "1/1"} tokens: dict[str, list[int]] = { "trait": [], "tree": [], "combo": [], "germ": [], "target": [], "obs": [], "dataset": [], "sample": [], "call": [], "marker": [], } pred_ids: list[int] = [] cv_ids: list[int] = [] idx_ids: list[int] = [] GS_PREDS: dict[str, list[int]] = {"rr": [], "bb": [], "gb": []} FIX: dict = {} def check(name, cond, detail=""): global ok, fail if cond: ok += 1 print(f" [ok] {name} {detail}") else: fail += 1 print(f" [FAIL] {name} {detail}") def login(client): global TOKEN d = {"username": "super", "password": "123456", "grant_type": "password", "login_type": "PC端"} r = client.post("/api/v1/system/auth/login", data=d) b = r.json() if r.status_code == 200 and b.get("code") == 0: TOKEN = b["data"]["access_token"] return key = client.get("/api/v1/system/auth/captcha/get").json()["data"]["key"] client.post("/api/v1/system/auth/captcha/slider/complete", json={"captcha_key": key}) d["captcha_key"] = key r = client.post("/api/v1/system/auth/login", data=d) b = r.json() assert r.status_code == 200 and b.get("code") == 0, f"LOGIN FAIL {r.status_code} {b}" TOKEN = b["data"]["access_token"] def auth(): return {"Authorization": f"Bearer {TOKEN}"} async def _wipe() -> None: """启动前清理本前缀残留（防上次进程被杀/建 fixture 中途失败留下的脏数据）。""" engine, sf = create_async_engine_and_session() try: async with sf() as db: trait_ids = list((await db.execute(select(TraitModel.id).where( TraitModel.trait_code.like(f"c%_{PREFIX}")))).scalars()) trait_like = [f"c%_{PREFIX}"] for pat in trait_like: await db.execute(delete(TypeBResultModel).where( TypeBResultModel.trait_code.like(pat))) cv_rows = list((await db.execute(select(CvResultModel.id).where( CvResultModel.trait_code.like(f"c%_{PREFIX}")))).scalars()) if cv_rows: await db.execute(delete(CvFoldModel).where(CvFoldModel.cv_result_id.in_(cv_rows))) await db.execute(delete(CvResultModel).where(CvResultModel.id.in_(cv_rows))) await db.execute(delete(SelectionIndexModel).where( SelectionIndexModel.result_json.op("->>")("traits").like(f"%_{PREFIX}%"))) await db.execute(delete(StatisticsJobModel).where( StatisticsJobModel.params_json.op("->>")("trait_code").like(f"c%_{PREFIX}%"))) if trait_ids: await db.execute(delete(PredictionValueModel).where( PredictionValueModel.trait_id.in_(trait_ids))) await db.execute(delete(PredictionModel).where( PredictionModel.trait_id.in_(trait_ids))) obs = list((await db.execute(select(TraitObservationModel.id).where( TraitObservationModel.trait_id.in_(trait_ids)))).scalars()) if obs: await db.execute(delete(TraitObservationModel).where( TraitObservationModel.id.in_(obs))) ds_ids = list((await db.execute(select(GenotypingDatasetModel.id).where( GenotypingDatasetModel.dataset_name == f"GS_{PREFIX}"))).scalars()) if ds_ids: smp = list((await db.execute(select(GenotypeSampleModel.id).where( GenotypeSampleModel.dataset_id.in_(ds_ids)))).scalars()) if smp: await db.execute(delete(GenotypeCallModel).where( GenotypeCallModel.sample_id.in_(smp))) await db.execute(delete(GenotypeSampleModel).where( GenotypeSampleModel.id.in_(smp))) await db.execute(delete(GenotypingDatasetModel).where( GenotypingDatasetModel.id.in_(ds_ids))) m_ids = list((await db.execute(select(MarkerModel.id).where( MarkerModel.marker_name.like(f"MK%_{PREFIX}")))).scalars()) if m_ids: await db.execute(delete(MarkerModel).where(MarkerModel.id.in_(m_ids))) tree_ids = list((await db.execute(select(TreeModel.id).where( TreeModel.tree_no.like(f"{PREFIX}-%")))).scalars()) if tree_ids: await db.execute(delete(TreeModel).where(TreeModel.id.in_(tree_ids))) combo_ids = list((await db.execute(select(CrossCombinationModel.id).where( CrossCombinationModel.combination_code.like(f"C%_{PREFIX}")))).scalars()) if combo_ids: await db.execute(delete(CrossCombinationModel).where( CrossCombinationModel.id.in_(combo_ids))) germ_ids = list((await db.execute(select(BreedingGermplasmModel.id).where( BreedingGermplasmModel.cultivar_name.like(f"%_{PREFIX}")))).scalars()) if germ_ids: await db.execute(delete(BreedingGermplasmModel).where( BreedingGermplasmModel.id.in_(germ_ids))) if trait_ids: await db.execute(delete(TraitModel).where(TraitModel.id.in_(trait_ids))) await db.execute(delete(TargetModel).where(TargetModel.target_name == f"目标{PREFIX}")) await db.commit() print(f"[preclean] o234 tc 前缀残留已清（trait={len(trait_ids)} tree={len(tree_ids)}）") finally: await engine.dispose() async def _build_fixture() -> None: engine, sf = create_async_engine_and_session() try: async with sf() as db: rng = np.random.default_rng(SEED) target = TargetModel(target_name=f"目标{PREFIX}", created_id=1) db.add(target) await db.flush() tokens["target"].append(target.id) cX = TraitModel(trait_code=f"cX_{PREFIX}", trait_name=f"单果重{PREFIX}", data_type="numeric", unit="1", is_core="1", direction="desc", into_ebv="1", default_h2=0.5, created_id=1) cY = TraitModel(trait_code=f"cY_{PREFIX}", trait_name=f"可溶固形物{PREFIX}", data_type="numeric", unit="%", is_core="1", direction="asc", into_ebv="1", default_h2=0.5, created_id=1) cS = TraitModel(trait_code=f"cS_{PREFIX}", trait_name=f"幼成相关{PREFIX}", data_type="numeric", unit="1", is_core="1", direction="desc", into_ebv="1", default_h2=0.6, created_id=1) db.add_all([cX, cY, cS]) await db.flush() tokens["trait"] = [cX.id, cY.id, cS.id] FIX["cX"], FIX["cY"], FIX["cS"] = cX, cY, cS germ_f, germ_m = [], [] for i in range(N_FAM): gf = BreedingGermplasmModel(cultivar_name=f"FM{i}_{PREFIX}", can_be_female=True, created_id=1) gm = BreedingGermplasmModel(cultivar_name=f"MM{i}_{PREFIX}", can_be_male=True, created_id=1) db.add_all([gf, gm]) await db.flush() germ_f.append(gf.id) germ_m.append(gm.id) tokens["germ"] = germ_f + germ_m combos, combo_of = [], {} for i in range(N_FAM): c = CrossCombinationModel( combination_code=f"C{i}_{PREFIX}", bre_target_id=target.id, female_parent_id=germ_f[i], male_parent_id=germ_m[i], design_type="full_diallel", created_id=1) db.add(c) await db.flush() combos.append(c.id) tokens["combo"] = combos trees: list[TreeModel] = [] for i in range(N_FAM): for k in range(N_PER): t = TreeModel(combination_id=combos[i], tree_no=f"{PREFIX}-{i:02d}-{k:02d}", status="alive", stage="seedling", generation="F1", planted_date="2023-03-10", created_id=1) db.add(t) trees.append(t) await db.flush() tokens["tree"] = [t.id for t in trees] FIX["trees"] = trees # 基因型剂量矩阵（确保每列多态：全同列翻一个样本） dos = rng.binomial(2, 0.5, size=(N, M)).astype(int) for j in range(M): if len(set(dos[:, j])) == 1: dos[0, j] = 2 - int(dos[0, j]) # cX 表型挂钩标记 3 剂量（GBLUP/rrBLUP 有信号）；cY 独立（P 近对角） obs_rows = [] for i, t in enumerate(trees): fam = i // N_PER d3 = dos[i, 3] obs_rows.append(TraitObservationModel( tree_id=t.id, trait_id=cX.id, value_numeric=float(100.0 + 4.0 * (d3 - 1) + rng.normal(0, 1.2)), evaluate_year=2025, created_id=1)) obs_rows.append(TraitObservationModel( tree_id=t.id, trait_id=cY.id, value_numeric=float(30.0 + rng.normal(0, 0.8)), evaluate_year=2025, created_id=1)) dev = rng.normal(0, 0.4) g = 6.0 if fam == 0 else 14.0 for stage in ("juvenile", "evaluation"): obs_rows.append(TraitObservationModel( tree_id=t.id, trait_id=cS.id, value_numeric=float(10.0 + g + dev + rng.normal(0, 0.6)), stage=stage, evaluate_year=2025, created_id=1)) db.add_all(obs_rows) await db.flush() tokens["obs"] = [o.id for o in obs_rows] # ---- 基因型数据集：全株分型 ---- ds = GenotypingDatasetModel(dataset_name=f"GS_{PREFIX}", platform="SNP", purpose="GS", created_id=1) db.add(ds) await db.flush() tokens["dataset"].append(ds.id) FIX["dataset_id"] = ds.id markers: list[MarkerModel] = [] for j in range(M): markers.append(MarkerModel(marker_name=f"MK{j}_{PREFIX}", marker_type="SNP", chromosome=str(j % 5), position=j * 100, created_id=1)) db.add_all(markers) await db.flush() tokens["marker"] = [m.id for m in markers] samples, calls = [], [] for i, t in enumerate(trees): s = GenotypeSampleModel(sample_name=f"{PREFIX}-S{i:02d}", dataset_id=ds.id, source_type="tree", source_id=t.id, created_id=1) db.add(s) samples.append(s) await db.flush() tokens["sample"] = [s.id for s in samples] for i, s in enumerate(samples): for j, m in enumerate(markers): calls.append(GenotypeCallModel(sample_id=s.id, marker_id=m.id, allele=GT[int(dos[i, j])], created_id=1)) db.add_all(calls) await db.flush() tokens["call"] = [c.id for c in calls] # ---- ABLUP 批次（手插，供选择指数）：reliability=0 → Calo 分母 0 → r_g=0 → G 对角 ---- ev = blup.ENGINE_VERSION for tc, trait, scale in (("X", cX, 4.0), ("Y", cY, 2.0)): pred = PredictionModel(model_name=f"AB_{PREFIX}_{tc}", trait_id=trait.id, method="ABLUP", heritability=0.5, engine_version=ev, input_hash="a" * 64, is_active=True, created_id=1) db.add(pred) await db.flush() pred_ids.append(pred.id) vals = [] for r_i, t in enumerate(trees): vals.append(PredictionValueModel( prediction_id=pred.id, tree_id=t.id, trait_id=trait.id, predicted_value=float(rng.normal(0, scale)), reliability=0.0, pa=0.0, rank=r_i + 1, created_id=1)) db.add_all(vals) await db.commit() finally: await engine.dispose() def _values_by_tree(client, pred_id: int) -> dict[int, float]: r = client.get(f"/api/v1/bre/statistics/predictions/{pred_id}/values", headers=auth()) b = r.json() assert r.status_code == 200 and b.get("code") == 0, f"values 失败 {r.status_code} {str(r.text)[:120]}" return {int(v["tree_id"]): float(v["predicted_value"]) for v in (b.get("data") or []) if v.get("tree_id") is not None} def _pearson(x: dict, y: dict) -> float: ks = sorted(set(x) & set(y)) a = np.array([x[k] for k in ks]) c = np.array([y[k] for k in ks]) return float(np.corrcoef(a, c)[0, 1]) async def _verify_db() -> None: """DB 层 golden 断言（镜像 ssgblup tc 惯例）：HTTP 已落库，此处核对 method/engine_version/折数。""" engine, sf = create_async_engine_and_session() try: async with sf() as db: for pid in GS_PREDS.get("rr", []): p = await db.get(PredictionModel, pid) check("[DB] rrblup 落库 method=RRBLUP", p is not None and p.method == "RRBLUP", f"method={p.method if p else None}") check("[DB] rrblup engine_version=genomic.ENGINE_VERSION", p is not None and p.engine_version == genomic.ENGINE_VERSION, f"ev={p.engine_version if p else None}") nv = len(list((await db.execute(select(PredictionValueModel).where( PredictionValueModel.prediction_id == pid))).scalars())) check("[DB] rrblup EBV 条数 = 基因型树数", nv == N, f"n={nv}") for pid in GS_PREDS.get("bb", []): p = await db.get(PredictionModel, pid) check("[DB] bayesb 落库 method=BayesB", p is not None and p.method == "BayesB", f"method={p.method if p else None}") for pid in GS_PREDS.get("gb", []): p = await db.get(PredictionModel, pid) check("[DB] gblup 落库 method=GBLUP", p is not None and p.method == "GBLUP", f"method={p.method if p else None}") if cv_ids: rows = list((await db.execute(select(CvResultModel).where( CvResultModel.id.in_(cv_ids)).order_by(CvResultModel.id))).scalars()) expect = ["KFCV/RRBLUP", "KFCV/BayesB"] for c, exp in zip(rows, expect): check(f"[DB] CV 落库 method={exp}", c.method == exp, f"method={c.method}") nf = len(list((await db.execute(select(CvFoldModel).where( CvFoldModel.cv_result_id == c.id))).scalars())) check(f"[DB] {exp} 折数=5", nf == 5, f"n_folds={nf}") check(f"[DB] {exp} mean_pearson∈[−1,1]", c.mean_pearson is not None and -1.0 <= float(c.mean_pearson) <= 1.0, f"mp={c.mean_pearson}") finally: await engine.dispose() async def _cleanup() -> None: engine, sf = create_async_engine_and_session() try: async with sf() as db: if pred_ids or cv_ids: await db.execute(delete(StatisticsJobModel).where( StatisticsJobModel.result_ref.in_(pred_ids + cv_ids))) if idx_ids: await db.execute(delete(SelectionIndexModel).where(SelectionIndexModel.id.in_(idx_ids))) if cv_ids: await db.execute(delete(CvFoldModel).where(CvFoldModel.cv_result_id.in_(cv_ids))) await db.execute(delete(CvResultModel).where(CvResultModel.id.in_(cv_ids))) if pred_ids: await db.execute(delete(PredictionValueModel).where( PredictionValueModel.prediction_id.in_(pred_ids))) await db.execute(delete(PredictionModel).where(PredictionModel.id.in_(pred_ids))) if tokens["call"]: await db.execute(delete(GenotypeCallModel).where(GenotypeCallModel.id.in_(tokens["call"]))) if tokens["sample"]: await db.execute(delete(GenotypeSampleModel).where(GenotypeSampleModel.id.in_(tokens["sample"]))) if tokens["dataset"]: await db.execute(delete(GenotypingDatasetModel).where(GenotypingDatasetModel.id.in_(tokens["dataset"]))) if tokens["marker"]: await db.execute(delete(MarkerModel).where(MarkerModel.id.in_(tokens["marker"]))) if tokens["obs"]: await db.execute(delete(TraitObservationModel).where(TraitObservationModel.id.in_(tokens["obs"]))) if tokens["tree"]: await db.execute(delete(TreeModel).where(TreeModel.id.in_(tokens["tree"]))) if tokens["combo"]: await db.execute(delete(CrossCombinationModel).where(CrossCombinationModel.id.in_(tokens["combo"]))) if tokens["germ"]: await db.execute(delete(BreedingGermplasmModel).where( BreedingGermplasmModel.id.in_(tokens["germ"]))) if tokens["trait"]: await db.execute(delete(TraitModel).where(TraitModel.id.in_(tokens["trait"]))) if tokens["target"]: await db.execute(delete(TargetModel).where(TargetModel.id.in_(tokens["target"]))) await db.commit() print(f"[cleanup] o234 tc 数据已清（pred={len(pred_ids)} cv={len(cv_ids)} idx={len(idx_ids)}）") finally: await engine.dispose() def main_() -> None: asyncio.run(_wipe()) asyncio.run(_build_fixture()) try: with TestClient(create_app()) as client: login(client) H = auth() cX, cY, cS = FIX["cX"], FIX["cY"], FIX["cS"] dsid = FIX["dataset_id"] # ── O2 type_b env_dim=stage（幼-成遗传相关）── r = client.post("/api/v1/bre/statistics/type-b-heredity", json={"trait_id": cS.id, "trait_code": cS.trait_code, "env_dim": "stage", "method": "reml", "year": None}, headers=H) check("[O2] type-b env_dim=stage HTTP 200", r.status_code == 200, f"{r.status_code}") tb = r.json() rid = tb.get("data") if tb.get("code") == 0 else None check("[O2] 返回 id", isinstance(rid, int), f"rid={rid}") det = None if isinstance(rid, int): g = client.get(f"/api/v1/bre/statistics/type-b/{rid}", headers=H) det = g.json().get("data") if g.status_code == 200 and g.json().get("code") == 0 else None check("[O2] env_dim=stage 落库", det is not None and det.get("env_dim") == "stage") if det: envs = det.get("envs_json") or {} check("[O2] 双阶段各 12 株", sorted(envs) == ["evaluation", "juvenile"] and all(v == N for v in envs.values()), f"envs_json={envs}") mg = det.get("matrix_json") or {} rg = mg.get("evaluation", {}).get("juvenile") check(f"[O2] 幼年–成年遗传相关 r_g={rg} > 0.5", rg is not None and float(rg) > 0.5) check("[O2] n_common=12（同树双阶段）", det.get("n_common") == N, f"n_common={det.get('n_common')}") check("[O2] data_version=v2.30", det.get("data_version") == "v2.30", f"{det.get('data_version')}") # ── O3 run_gblup：rrblup / bayesb（可复现）/ gblup（对偶）── pid_rr = None r = client.post("/api/v1/bre/statistics/gblup/run", json={"dataset_id": dsid, "trait_id": cX.id, "trait_code": cX.trait_code, "method": "rrblup", "seed": SEED}, headers=H) b = r.json() check("[O3] gblup/run rrblup HTTP 200", r.status_code == 200, f"{r.status_code}") pid_rr = b.get("data") if b.get("code") == 0 else None if isinstance(pid_rr, int): pred_ids.append(pid_rr) GS_PREDS["rr"].append(pid_rr) v_rr = _values_by_tree(client, pid_rr) check("[O3] rrblup EBV 覆盖全部基因型树", len(v_rr) == N, f"n={len(v_rr)}") pid_bb = [] for _ in range(2): r = client.post("/api/v1/bre/statistics/gblup/run", json={"dataset_id": dsid, "trait_id": cX.id, "trait_code": cX.trait_code, "method": "bayesb", "seed": SEED}, headers=H) b = r.json() pid = b.get("data") if b.get("code") == 0 else None check("[O3] gblup/run bayesb HTTP 200", r.status_code == 200, f"{r.status_code}") if isinstance(pid, int): pred_ids.append(pid) GS_PREDS["bb"].append(pid) pid_bb.append(pid) if len(pid_bb) == 2: v1 = _values_by_tree(client, pid_bb[0]) v2 = _values_by_tree(client, pid_bb[1]) check("[O3] bayesb 同 seed 两次 EBV 逐位一致（可复现）", v1 == v2) r = client.post("/api/v1/bre/statistics/gblup/run", json={"dataset_id": dsid, "trait_id": cX.id, "trait_code": cX.trait_code, "method": "gblup", "seed": SEED}, headers=H) b = r.json() pid_gb = b.get("data") if b.get("code") == 0 else None check("[O3] gblup/run gblup HTTP 200", r.status_code == 200, f"{r.status_code}") if isinstance(pid_gb, int): pred_ids.append(pid_gb) GS_PREDS["gb"].append(pid_gb) v_gb = _values_by_tree(client, pid_gb) if pid_rr is not None: corr = _pearson(v_rr, v_gb) check(f"[O3] rrblup/GBLUP EBV 相关 {corr:.3f} > 0.9（对偶）", corr > 0.9) # method 校验：非法方法 → 409 r = client.post("/api/v1/bre/statistics/gblup/run", json={"dataset_id": dsid, "trait_id": cX.id, "trait_code": cX.trait_code, "method": "gbm", "seed": SEED}, headers=H) check("[O3] method 非法 → 409", r.status_code == 409, f"{r.status_code}") # ── O3 run_cv：rrblup / bayesb GS 分派 ── for meth, label in (("rrblup", "KFCV/RRBLUP"), ("bayesb", "KFCV/BayesB")): r = client.post("/api/v1/bre/statistics/cv/run", json={"trait_id": cX.id, "trait_code": cX.trait_code, "dataset_id": dsid, "method": meth, "k": 5, "split": "random", "seed": SEED}, headers=H) b = r.json() cid = b.get("data") if b.get("code") == 0 else None check(f"[O3] cv/run {meth} HTTP 200", r.status_code == 200, f"{r.status_code}") if isinstance(cid, int): cv_ids.append(cid) g = client.get(f"/api/v1/bre/statistics/cv/{cid}", headers=H) gb2 = g.json().get("data") if g.status_code == 200 and g.json().get("code") == 0 else None res = (gb2 or {}).get("result") or {} check(f"[O3] {label} 落库 + mean_pearson∈[−1,1]", res.get("method") == label and res.get("mean_pearson") is not None and -1.0 <= float(res["mean_pearson"]) <= 1.0, f"method={res.get('method')} pearson={res.get('mean_pearson')}") # ── O4 selection_index：restricted + economic ── bidX = pred_ids[0] # ABLUP_X（手插） bidY = pred_ids[1] # ABLUP_Y（手插） r = client.post("/api/v1/bre/statistics/selection-index", json={"method": "restricted", "aggregate": "tree", "g_method": "calo", "weights": {cX.trait_code: 1.0, cY.trait_code: 1.0}, "batch_ids": {cX.trait_code: bidX, cY.trait_code: bidY}, "restricted_traits": [cX.trait_code], "economic_weights": {cX.trait_code: 1.0, cY.trait_code: 1.0}, "top_n": 50}, headers=H) b = r.json() check("[O4] selection-index restricted HTTP 200", r.status_code == 200, f"{r.status_code}") d = b.get("data") if b.get("code") == 0 else None if isinstance(d, dict) and d.get("id"): idx_ids.append(d["id"]) check("[O4] 标记 SI_RES + restricted + ΔG=0 + economic", d.get("method") == "SI_RES" and d.get("restricted") == [cX.trait_code] and d.get("constraint_mode") == "ΔG=0" and d.get("economic") is True) bw = d.get("b") or {} bx, by = bw.get(cX.trait_code), bw.get(cY.trait_code) # r_g=0（reliability=0 → Calo 分母 0）→ G 精确对角 → ΔG_受限=0 ⇔ b_受限=0 check("[O4] 受限性状 b≈0（ΔG=0 投影生效）", bx is not None and abs(float(bx)) < 1e-4, f"b_cX={bx}") check("[O4] 非受限性状自由响应", by is not None and abs(float(by)) > 0.05, f"b_cY={by}") check("[O4] 排名 top 落库", isinstance(d.get("top"), list) and len(d.get("top")) > 0, f"n_top={len(d.get('top') or [])}") # economic 方向翻转：cY asc（越小越优）→ 经济权重为正时 b<0 r = client.post("/api/v1/bre/statistics/selection-index", json={"method": "smith_hazel", "aggregate": "tree", "g_method": "calo", "weights": {cX.trait_code: 1.0, cY.trait_code: 1.0}, "batch_ids": {cX.trait_code: bidX, cY.trait_code: bidY}, "economic_weights": {cX.trait_code: 2.0, cY.trait_code: 1.0}, "top_n": 50}, headers=H) b = r.json() check("[O4] selection-index smith_hazel+economic HTTP 200", r.status_code == 200, f"{r.status_code}") d = b.get("data") if b.get("code") == 0 else None if isinstance(d, dict) and d.get("id"): idx_ids.append(d["id"]) check("[O4] 标记 SI_SH + economic", d.get("method") == "SI_SH" and d.get("economic") is True) bw = d.get("b") or {} bx, by = bw.get(cX.trait_code), bw.get(cY.trait_code) check("[O4] economic 方向翻转：desc 性状 b>0", bx is not None and float(bx) > 0, f"b_cX={bx}") check("[O4] asc 性状 direction 翻转 b<0", by is not None and float(by) < 0, f"b_cY={by}") # 门禁：restricted 缺 restricted_traits → 409 r = client.post("/api/v1/bre/statistics/selection-index", json={"method": "restricted", "aggregate": "tree", "g_method": "calo", "weights": {cX.trait_code: 1.0, cY.trait_code: 1.0}, "batch_ids": {cX.trait_code: bidX, cY.trait_code: bidY}, "top_n": 50}, headers=H) check("[O4] restricted 缺 restricted_traits → 409", r.status_code == 409, f"{r.status_code}") finally: asyncio.run(_verify_db()) asyncio.run(_cleanup()) print(f"\n===== o234 tc 套件：ok={ok} fail={fail} =====") if __name__ == "__main__": main_() sys.exit(1 if fail else 0)