572 lines
31 KiB
Python
572 lines
31 KiB
Python
# -*- 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)
|