From fa5c106781b914f0dba9832fc82f2396490ea606 Mon Sep 17 00:00:00 2001 From: Eduard Kerkhoven Date: Wed, 23 Sep 2026 17:19:57 +0200 Subject: [PATCH] Add sweep_tunable_mask_size/recommend_target_impact_share select_tunable_mask's target_impact_share picks a comparable *share* of leverage on any model, but the same share does not sit at a comparable point on the *fit* curve -- two models with similarly shaped leverage curves can have fit curves that diverge sharply. The only way to tell is to tune at a few candidate sizes and look at where the improvement stops. sweep_tunable_mask_size tunes at each of a list of target_impact_shares (and seeds) and reports the resulting fit per point. recommend_target_impact_share picks the smallest share past which further tuning stops paying for itself, using two independent checks against a later point's best result: a relative-improvement tolerance, and, where at least one of the two points compared has more than one seed, a check against their own local seed spread (not a sweep-wide pooled one, since how noisy a comparison is can differ across the sweep). MaskSizePoint.rmse takes min(rmse_trace) per seed rather than the final entry, since rmse_trace only advances at a step that improved the (possibly prior-penalised) objective, per cmaes_kcat_tuning's own docstring -- with the default nonzero prior_penalty_weight the last entry is not guaranteed to be the best one recorded. sweep_tunable_mask_size's default target_impact_shares now starts at 0.3 (previously 0.5) and its default seeds is (0, 1) (previously a single seed), so a default call's noise-floor check has something to compare against instead of silently degrading to the relative check alone. --- src/geckopy/kcat_tuning/evotune/__init__.py | 8 + src/geckopy/kcat_tuning/evotune/mask_size.py | 309 ++++++++++++++++++ tests/test_mask_size.py | 327 +++++++++++++++++++ 3 files changed, 644 insertions(+) create mode 100644 src/geckopy/kcat_tuning/evotune/mask_size.py create mode 100644 tests/test_mask_size.py diff --git a/src/geckopy/kcat_tuning/evotune/__init__.py b/src/geckopy/kcat_tuning/evotune/__init__.py index 78c5a12..2a6e00f 100644 --- a/src/geckopy/kcat_tuning/evotune/__init__.py +++ b/src/geckopy/kcat_tuning/evotune/__init__.py @@ -20,6 +20,11 @@ compute_excarbon, dataset_rmse, ) +from .mask_size import ( + MaskSizePoint, + recommend_target_impact_share, + sweep_tunable_mask_size, +) from .priors import ( UNLABELLED_GROUP, build_sigma0_log, @@ -42,6 +47,7 @@ "EvotuneData", "EvotuneResult", "ConditionSimResult", + "MaskSizePoint", "evotune_distance", "build_sigma0_log", "classify_kcat_source", @@ -50,8 +56,10 @@ "compute_excarbon", "dataset_rmse", "load_evotune_data", + "recommend_target_impact_share", "screen_kcat_leverage", "select_tunable_mask", "simulate_evotune_dataset", + "sweep_tunable_mask_size", "tune_prior_penalty_weight", ] diff --git a/src/geckopy/kcat_tuning/evotune/mask_size.py b/src/geckopy/kcat_tuning/evotune/mask_size.py new file mode 100644 index 0000000..9d9afd8 --- /dev/null +++ b/src/geckopy/kcat_tuning/evotune/mask_size.py @@ -0,0 +1,309 @@ +"""Measure how many free parameters a tuning run actually needs. + +:func:`~.tuning.select_tunable_mask`'s ``target_impact_share`` selects a +comparable *share* of leverage on any model -- that is the point of a +relative cutoff over a fixed count or a fixed leverage value. It does not +follow that the same share sits at a comparable point on the *fit* +curve: how concentrated a model's leverage is, and how quickly its fit +stops improving as more parameters join the search, are different +properties of the model and the data, and they do not move together. +Two ecModels screened and tuned the same way can have leverage curves +of similar shape yet fit curves that diverge sharply -- one still +climbing at ``target_impact_share=0.9``, the other flat since a third of +the way there. The only way to tell them apart is to tune at a few +candidate sizes and look at where the improvement stops, which is what +this module automates. + +The two functions here are meant to be used together but stay separate: +:func:`sweep_tunable_mask_size` only measures (it runs +:func:`~.tuning.cmaes_kcat_tuning` at each requested share and reports +what came back); :func:`recommend_target_impact_share` only decides, +from measurements already in hand. Passing a sweep's own points through +the second lets a different stopping rule be tried without re-tuning. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Callable, Optional, Sequence + +import numpy as np +import pandas as pd + +from .tuning import ( + _resolve_context, cmaes_kcat_tuning, screen_kcat_leverage, select_tunable_mask, +) + +if TYPE_CHECKING: + from ...adapter import ModelAdapter + from ...adapter.params import EvotuneParams + from ...ec_model.ec_model import EcModel + from .data import EvotuneData + + +@dataclass(frozen=True) +class MaskSizePoint: + """Tuned fit at one candidate ``target_impact_share``. + + Attributes + ---------- + target_impact_share + The share this point was tuned at. + n_selected + Free parameters :func:`~.tuning.select_tunable_mask` chose at + this share (tied isozyme copies count once). + rmse + The lowest value reached in each seed's + ``EvotuneResult.rmse_trace`` -- not ``rmse_trace[-1]``, because + ``rmse_trace`` only records plain RMSE at a step that improved + the (possibly prior-penalised) *objective*, per + ``cmaes_kcat_tuning``'s own docstring; with the default nonzero + ``prior_penalty_weight`` a later generation can improve the + objective while its plain RMSE is slightly worse than an + earlier generation's, making the final entry not necessarily + the best one recorded. ``min(rmse_trace)`` is the best that + *was* recorded either way, though a candidate that improved + plain RMSE without improving the objective is invisible to + both and to ``rmse_trace`` itself -- set + ``params.prior_penalty_weight=0`` before a sweep meant to + measure fit alone, if that residual gap matters. Kept per-seed + rather than pre-averaged so a caller can look at the spread + directly. + seeds + The seeds each entry of ``rmse`` came from. + """ + + target_impact_share: float + n_selected: int + rmse: tuple[float, ...] + seeds: tuple[int, ...] + + @property + def rmse_mean(self) -> float: + return float(np.mean(self.rmse)) + + @property + def rmse_sd(self) -> float: + """Sample standard deviation across seeds; ``nan`` with fewer + than two -- a single seed carries no information about noise, + and pretending otherwise (e.g. reporting 0) understates it.""" + return float(np.std(self.rmse, ddof=1)) if len(self.rmse) > 1 else float("nan") + + +def sweep_tunable_mask_size( + model: "EcModel", + *, + adapter: Optional["ModelAdapter"] = None, + params: Optional["EvotuneParams"] = None, + evotune_data: Optional["EvotuneData"] = None, + okp_method: Optional[str] = None, + bio_rxn: Optional[str] = None, + make_anaerobic: Optional[Callable[["EcModel"], None]] = None, + change_protein_biomass: Optional[Callable[["EcModel", float], None]] = None, + screen: Optional[pd.DataFrame] = None, + target_impact_shares: Sequence[float] = (0.3, 0.5, 0.7, 0.8, 0.9, 0.95), + seeds: Sequence[int] = (0, 1), + n_proc: Optional[int] = None, + verbose: bool = True, +) -> list[MaskSizePoint]: + """Tune at each of ``target_impact_shares`` and report the fit. + + The screen is computed once (or reused, via ``screen``) and shared + across every share and seed; only :func:`~.tuning.cmaes_kcat_tuning` + reruns per point. Each run starts from ``model``'s own current + ``ec.kcat`` -- ``model`` is not mutated (a fresh copy is tuned each + time), unlike :func:`~.tuning.cmaes_kcat_tuning` itself. + + Cost is one screen (comparable to a full tuning run) plus + ``len(target_impact_shares) * len(seeds)`` tuning runs, each sized + by however many parameters that share selects -- the later points + in an increasing sequence of shares cost more, both because CMA-ES + scales its population with dimension and because a wider search + space usually takes longer to explore. + + The defaults are a coarse scout, not a guarantee of finding the + true optimum: five or six points cannot resolve a curve whose real + knee sits inside one of the gaps between them, and how wide that + gap needs to be to matter is exactly what a first sweep is for + finding out. Look at ``n_selected`` after a first sweep -- if + consecutive shares jump by a large multiple in parameter count + (``target_impact_share`` is closer to the leverage tail than to its + bulk there, e.g. the jump from 0.9 to 0.95 was 5x on one real + model), or if the recommendation lands next to the widest such gap, + that gap is where the true knee could be hiding; rerun with a few + shares added inside it. The same applies at the low end: if the + smallest share measured isn't clearly worse than the best one, it + has not necessarily bracketed the true minimum, and a smaller share + is worth adding to confirm it. + + ``seeds`` defaults to two, not one, doubling the cost of the + default call: :func:`recommend_target_impact_share`'s noise-floor + check needs at least two seeds at a point to have anything to + check against, and with only one it silently falls back to a + cruder rule everywhere. Pass a single seed for the very first, + cheapest possible scout of the shape; widen to more once a + promising region is known, since that is where seed noise is most + likely to matter (fit values close enough together for it to + decide the comparison). + + Parameters + ---------- + model + EcModel with a populated ``ec.kcat``, used as the tuning + starting point at every share/seed. Not mutated. + target_impact_shares + Candidate shares to measure, any order (sorted on return). + seeds + Seeds tuned at every share. + screen + A precomputed :func:`~.tuning.screen_kcat_leverage` report. + Computed once, fresh, if not given. + adapter, params, evotune_data, okp_method, bio_rxn, make_anaerobic, + change_protein_biomass, n_proc, verbose + Forwarded to :func:`~.tuning.screen_kcat_leverage` and + :func:`~.tuning.cmaes_kcat_tuning`; see their docstrings. + + Returns + ------- + list of :class:`MaskSizePoint`, one per share, sorted by + ``target_impact_share`` ascending. + """ + params, evotune_data, bio_rxn, okp_method = _resolve_context( + model, adapter, params, evotune_data, bio_rxn, okp_method) + + if screen is None: + screen = screen_kcat_leverage( + model, adapter=adapter, params=params, evotune_data=evotune_data, + okp_method=okp_method, bio_rxn=bio_rxn, + make_anaerobic=make_anaerobic, change_protein_biomass=change_protein_biomass, + n_proc=n_proc, + ) + + points = [] + for share in sorted(target_impact_shares): + mask = select_tunable_mask(model, screen, target_impact_share=share) + n_selected = int(mask.sum()) + rmses = [] + for seed in seeds: + run_model = model.copy() + result = cmaes_kcat_tuning( + run_model, adapter=adapter, params=params, evotune_data=evotune_data, + okp_method=okp_method, bio_rxn=bio_rxn, + make_anaerobic=make_anaerobic, change_protein_biomass=change_protein_biomass, + tunable_mask=mask, n_proc=n_proc, seed=seed, verbose=False, + ) + rmses.append(float(min(result.rmse_trace))) + if verbose: + print( + f"target_impact_share={share}: {n_selected} selected, " + f"seed {seed}: rmse {rmses[-1]:.4f}", + flush=True, + ) + points.append(MaskSizePoint( + target_impact_share=float(share), n_selected=n_selected, + rmse=tuple(rmses), seeds=tuple(seeds), + )) + return points + + +def _local_sd(a: MaskSizePoint, b: MaskSizePoint) -> Optional[float]: + """Pool ``a`` and ``b``'s own seed sd, using whichever of the two + actually has one (both, if both do). Deliberately not a sweep-wide + pooled sd: how noisy a comparison is can genuinely differ across + the sweep (e.g. a larger parameter count giving CMA-ES more room to + land on different local optima across seeds), and borrowing an + unrelated point's spread to judge this pair would let noise from + somewhere else in the sweep, that has nothing to do with this + specific comparison, decide it.""" + sds = [p.rmse_sd for p in (a, b) if not np.isnan(p.rmse_sd)] + if not sds: + return None + return float(np.sqrt(np.mean(np.square(sds)))) + + +def recommend_target_impact_share( + points: Sequence[MaskSizePoint], + *, + rel_tol: float = 0.02, + noise_multiplier: float = 2.0, +) -> tuple[float, str]: + """Pick the smallest share past which tuning stops paying for itself. + + Walks ``points`` in increasing share order and returns the first + one, ``p``, whose best later point does not clearly improve on it. + "Clearly" is two independent tests, either enough on its own to + call a gain not worth it: the gain is small in relative terms + (within ``rel_tol`` of ``p``'s own RMSE), or -- only checkable where + at least one of the two points being compared has more than one + seed -- the gain is small next to how much repeating just those two + points varies (within ``noise_multiplier`` times their own pooled + seed standard deviation, not a sweep-wide one; see :func:`_local_sd`). + A comparison where neither point has a repeated seed only gets the + relative check, which alone cannot tell a genuine plateau from a + lucky seed -- ``reason`` says which test(s) applied so that + limitation is visible rather than silent. + + Parameters + ---------- + points + Output of :func:`sweep_tunable_mask_size` (or hand-built), at + least two entries, any order (sorted internally). + rel_tol + A later point counts as "no better" when its RMSE is within + this fraction of the earlier one's. + noise_multiplier + How many local standard deviations a gain must clear, where + checkable, to count as more than noise. + + Returns + ------- + (share, reason) + The recommended ``target_impact_share`` and one sentence on + why -- always the ``target_impact_share`` of one of ``points``, + since only shares actually measured can be recommended. + + Raises + ------ + ValueError + If ``points`` has fewer than two entries. + """ + pts = sorted(points, key=lambda p: p.target_impact_share) + if len(pts) < 2: + raise ValueError( + f"Need at least two points to compare, got {len(pts)}. " + "sweep_tunable_mask_size needs at least two target_impact_shares." + ) + + for i, p in enumerate(pts[:-1]): + best_later = min(pts[i + 1:], key=lambda q: q.rmse_mean) + gain = p.rmse_mean - best_later.rmse_mean + rel_gain = gain / p.rmse_mean if p.rmse_mean > 0 else 0.0 + rel_ok = rel_gain <= rel_tol + sd = _local_sd(p, best_later) + noise_ok = sd is not None and gain <= noise_multiplier * sd + + if rel_ok or noise_ok: + reasons = [] + if rel_ok: + reasons.append( + f"only {100 * rel_gain:.1f}% relative " + f"(<= {100 * rel_tol:.0f}% tolerance)" + ) + if noise_ok: + reasons.append( + f"{gain:.4g} absolute, within {noise_multiplier:g}x the " + f"local seed sd ({sd:.4g}) -- not distinguishable from " + "seed noise" + ) + elif sd is None: + reasons.append( + "no repeated seed at either point to check against noise" + ) + return p.target_impact_share, ( + f"share {best_later.target_impact_share:g}'s best improvement " + f"over this point's {p.rmse_mean:.4g} is " + " and ".join(reasons) + ) + + return pts[-1].target_impact_share, ( + "RMSE kept improving beyond every share measured; " + "widen target_impact_shares to find where it actually flattens" + ) diff --git a/tests/test_mask_size.py b/tests/test_mask_size.py new file mode 100644 index 0000000..0a591e3 --- /dev/null +++ b/tests/test_mask_size.py @@ -0,0 +1,327 @@ +"""Tests for kcat_tuning.evotune.mask_size. + +Reuses the same two-branch toy EcModel shape as test_evotune.py (kept +self-contained rather than imported, matching this test suite's +convention of one fixture builder per file). +""" +from pathlib import Path + +import numpy as np +import pytest +from scipy import sparse + +import cobra +from geckopy import EcModel, ModelAdapter +from geckopy.adapter.params import EvotuneParams +from geckopy.databases.flux_data import FluxData +from geckopy.ec_model.ec_data import EcData +from geckopy.kcat_tuning.evotune.data import EvotuneData +from geckopy.kcat_tuning.evotune.mask_size import ( + MaskSizePoint, + recommend_target_impact_share, + sweep_tunable_mask_size, +) +from geckopy.kcat_tuning.evotune.tuning import screen_kcat_leverage + +_TRUE_KCAT = 2.0 +_START_KCAT = 1.0 +_MW = 100.0 +_TRUE_GROWTH = _TRUE_KCAT * 3600.0 / _MW + + +def _adapter(tmp_path: Path) -> ModelAdapter: + (tmp_path / "model_adapter.toml").write_text( + 'conv_gem = "dummy.xml"\n' + 'org_name = "test"\n' + 'bio_rxn = "biomass"\n' + ) + return ModelAdapter.from_folder(tmp_path) + + +def _build_toy(adapter: ModelAdapter) -> EcModel: + """Two independent branches, one kcat each -- same shape as + test_evotune.py's fixture of the same name.""" + model = EcModel("toy", adapter=adapter) + + glc_e = cobra.Metabolite("glc_e", compartment="e") + eth_e = cobra.Metabolite("eth_e", compartment="e") + prot_pool = cobra.Metabolite("prot_pool", compartment="c") + prot_glc = cobra.Metabolite("prot_Eglc", compartment="c") + prot_eth = cobra.Metabolite("prot_Eeth", compartment="c") + bio_met = cobra.Metabolite("bio_met", compartment="c") + model.add_metabolites([glc_e, eth_e, prot_pool, prot_glc, prot_eth, bio_met]) + + EX_glc = cobra.Reaction("EX_glc") + EX_glc.add_metabolites({glc_e: -1.0}) + EX_glc.bounds = (-1000.0, 0.0) + + EX_eth = cobra.Reaction("EX_eth") + EX_eth.add_metabolites({eth_e: -1.0}) + EX_eth.bounds = (-1000.0, 0.0) + + coeff = _MW / (_START_KCAT * 3600.0) + R_glc = cobra.Reaction("R_glc") + R_glc.add_metabolites({glc_e: -1.0, prot_glc: -coeff, bio_met: 1.0}) + R_glc.bounds = (0.0, 1000.0) + + R_eth = cobra.Reaction("R_eth") + R_eth.add_metabolites({eth_e: -1.0, prot_eth: -coeff, bio_met: 1.0}) + R_eth.bounds = (0.0, 1000.0) + + BIO = cobra.Reaction("biomass") + BIO.add_metabolites({bio_met: -1.0}) + BIO.bounds = (0.0, 1000.0) + + pool_ex = cobra.Reaction("prot_pool_exchange") + pool_ex.add_metabolites({prot_pool: 1.0}) + pool_ex.bounds = (0.0, 1.0) + + usage_glc = cobra.Reaction("usage_prot_Eglc") + usage_glc.add_metabolites({prot_pool: -1.0, prot_glc: 1.0}) + usage_glc.bounds = (0.0, 1000.0) + + usage_eth = cobra.Reaction("usage_prot_Eeth") + usage_eth.add_metabolites({prot_pool: -1.0, prot_eth: 1.0}) + usage_eth.bounds = (0.0, 1000.0) + + model.add_reactions( + [EX_glc, EX_eth, R_glc, R_eth, BIO, pool_ex, usage_glc, usage_eth] + ) + model.objective = "biomass" + + model.ec = EcData( + rxns=["R_glc", "R_eth"], + kcat=np.array([_START_KCAT, _START_KCAT]), + source=["brenda", "dlkcat"], + notes=["", ""], + eccodes=["", ""], + genes=["g_glc", "g_eth"], + enzymes=["Eglc", "Eeth"], + mw=np.array([_MW, _MW]), + sequence=["", ""], + concs=np.array([np.nan, np.nan]), + rxn_enz_mat=sparse.csr_matrix(np.eye(2)), + ) + return model + + +def _evotune_data() -> EvotuneData: + max_grate = FluxData( + conds=["glucose", "ethanol"], + p_tot=np.array([np.nan, np.nan]), + gr_rate=np.array([_TRUE_GROWTH, _TRUE_GROWTH]), + exch_fluxes=np.array([[-1000.0, np.nan], [np.nan, -1000.0]]), + exch_mets=["glucose", "ethanol"], + exch_rxn_ids=["EX_glc", "EX_eth"], + ) + return EvotuneData(flux_data=None, max_grate=max_grate, zero_flux=[]) + + +# --------------------------------------------------------------------------- # +# sweep_tunable_mask_size +# --------------------------------------------------------------------------- # + + +def test_sweep_reports_one_point_per_share_sorted(tmp_path): + adapter = _adapter(tmp_path) + model = _build_toy(adapter) + evotune_data = _evotune_data() + params = EvotuneParams(max_generations=10, rmse_threshold=1e-6) + + points = sweep_tunable_mask_size( + model, adapter=adapter, params=params, evotune_data=evotune_data, + target_impact_shares=[1.0, 0.7, 0.9], seeds=[0], n_proc=1, verbose=False, + ) + + assert [p.target_impact_share for p in points] == [0.7, 0.9, 1.0] + # This toy model's leverage is evenly split across exactly two + # kcats (cum_leverage_share .667/1.0), so every share above two + # thirds keeps both -- chosen to stay clear of + # cmaes_kcat_tuning's own "too few free parameters" floor of 2. + assert [p.n_selected for p in points] == [2, 2, 2] + + +def test_sweep_does_not_mutate_input_model(tmp_path): + adapter = _adapter(tmp_path) + model = _build_toy(adapter) + evotune_data = _evotune_data() + params = EvotuneParams(max_generations=10, rmse_threshold=1e-6) + before = model.ec.kcat.copy() + + sweep_tunable_mask_size( + model, adapter=adapter, params=params, evotune_data=evotune_data, + target_impact_shares=[1.0], seeds=[0], n_proc=1, verbose=False, + ) + + assert np.array_equal(model.ec.kcat, before) + + +def test_sweep_runs_every_seed_at_every_share(tmp_path): + adapter = _adapter(tmp_path) + model = _build_toy(adapter) + evotune_data = _evotune_data() + params = EvotuneParams(max_generations=10, rmse_threshold=1e-6) + + points = sweep_tunable_mask_size( + model, adapter=adapter, params=params, evotune_data=evotune_data, + target_impact_shares=[1.0], seeds=[0, 1, 2], n_proc=1, verbose=False, + ) + + assert len(points) == 1 + assert points[0].seeds == (0, 1, 2) + assert len(points[0].rmse) == 3 + # Different seeds explore a stochastic search differently; on this + # toy problem they need not disagree by much, but nothing forces + # every seed to find bit-identical optima. + assert points[0].rmse_sd >= 0.0 + + +def test_sweep_reuses_a_given_screen(tmp_path): + adapter = _adapter(tmp_path) + model = _build_toy(adapter) + evotune_data = _evotune_data() + params = EvotuneParams(max_generations=10, rmse_threshold=1e-6) + screen = screen_kcat_leverage( + model, adapter=adapter, evotune_data=evotune_data, n_proc=1, + ) + + # select_tunable_mask reads cum_leverage_share directly; dropping it + # breaks a screen that's actually used, but a call that silently + # recomputes its own (correct) screen instead of reusing this + # broken one would not notice. + stripped = screen.drop(columns=["cum_leverage_share"]) + with pytest.raises(KeyError): + sweep_tunable_mask_size( + model, adapter=adapter, params=params, evotune_data=evotune_data, + screen=stripped, target_impact_shares=[1.0], seeds=[0], + n_proc=1, verbose=False, + ) + + +# --------------------------------------------------------------------------- # +# recommend_target_impact_share +# --------------------------------------------------------------------------- # + + +def _pt(share, rmse, seeds=None): + rmse = tuple(rmse) if not isinstance(rmse, tuple) else rmse + return MaskSizePoint( + target_impact_share=share, n_selected=int(share * 100), + rmse=rmse, seeds=seeds or tuple(range(len(rmse))), + ) + + +def test_recommend_picks_the_flattening_point(): + points = [ + _pt(0.5, (1.0,)), + _pt(0.7, (0.5,)), + _pt(0.8, (0.495,)), + _pt(0.9, (0.49,)), + ] + share, reason = recommend_target_impact_share(points, rel_tol=0.05) + assert share == 0.7 + assert "improve" in reason + + +def test_recommend_requires_at_least_two_points(): + with pytest.raises(ValueError): + recommend_target_impact_share([_pt(0.9, (0.5,))]) + + +def test_recommend_returns_largest_share_if_still_improving(): + points = [_pt(0.5, (1.0,)), _pt(0.7, (0.5,)), _pt(0.9, (0.1,))] + share, reason = recommend_target_impact_share(points, rel_tol=0.01) + assert share == 0.9 + assert "kept improving" in reason + + +def test_recommend_treats_a_later_regression_as_no_improvement(): + # A later share that's actually worse must not be picked over an + # earlier, better one -- this is what happened in practice (the + # motivating yeast N-sweep: N=1029 fit worse than N=500). + points = [_pt(0.7, (1.0,)), _pt(0.8, (0.5,)), _pt(0.9, (0.6,))] + share, reason = recommend_target_impact_share(points, rel_tol=0.05) + assert share == 0.8 + + +def test_recommend_noise_floor_stops_a_real_but_small_gain_from_mattering(): + # Two seeds at the earlier point give it a real (if crude) sd + # estimate; a later point barely inside that noise should not look + # like a genuine improvement even though rel_tol alone might allow it. + points = [ + _pt(0.7, (0.70, 0.74)), # mean 0.72, sd 0.028 + _pt(0.9, (0.71,)), # improvement of 0.01, well inside 2*sd + ] + # rel_tol tighter than the ~1.4% relative gain, so only the + # noise-floor check (not the relative-tolerance one) can be why + # this doesn't count as an improvement. + share, reason = recommend_target_impact_share( + points, rel_tol=0.005, noise_multiplier=2.0, + ) + assert share == 0.7 + assert "noise" in reason + + +def test_recommend_sorts_points_regardless_of_input_order(): + points = [_pt(0.9, (0.49,)), _pt(0.5, (1.0,)), _pt(0.7, (0.5,))] + share, _ = recommend_target_impact_share(points, rel_tol=0.05) + assert share == 0.7 + + +# --------------------------------------------------------------------------- # +# MaskSizePoint +# --------------------------------------------------------------------------- # + + +def test_mask_size_point_sd_is_nan_with_one_seed(): + p = _pt(0.9, (0.5,)) + assert np.isnan(p.rmse_sd) + + +def test_mask_size_point_mean_and_sd_with_multiple_seeds(): + p = _pt(0.9, (0.4, 0.5, 0.6)) + assert p.rmse_mean == pytest.approx(0.5) + assert p.rmse_sd == pytest.approx(np.std([0.4, 0.5, 0.6], ddof=1)) + + +# --------------------------------------------------------------------------- # +# rmse_trace's last entry isn't always its lowest (only true when +# prior_penalty_weight=0 -- the default is 0.03, so this is the common +# case, not an edge case): sweep_tunable_mask_size must use the +# trace's minimum, not its last entry. +# --------------------------------------------------------------------------- # + + +def test_sweep_uses_rmse_trace_minimum_not_last_entry(tmp_path, monkeypatch): + from geckopy.kcat_tuning.evotune import mask_size as mask_size_module + from geckopy.kcat_tuning.evotune.tuning import EvotuneResult + + adapter = _adapter(tmp_path) + model = _build_toy(adapter) + evotune_data = _evotune_data() + params = EvotuneParams(max_generations=10, rmse_threshold=1e-6) + + # A later generation can improve the (prior-penalised) objective + # while its own plain RMSE is worse than an earlier generation's -- + # cmaes_kcat_tuning's own code only records rmse_trace alongside an + # objective improvement, so the trace need not be monotonic in RMSE + # alone. 0.71934 here, not the last entry 0.72331, is the value + # that was actually achieved. + fake_trace = [1.0, 0.8, 0.71934, 0.72331] + + def fake_cmaes_kcat_tuning(run_model, **kwargs): + return EvotuneResult( + rxns=["R_glc", "R_eth"], old_kcat=run_model.ec.kcat.copy(), + new_kcat=run_model.ec.kcat.copy(), groups=["brenda", "dlkcat"], + rmse_trace=fake_trace, objective_trace=fake_trace, + n_generations=len(fake_trace), converged=False, + ) + + monkeypatch.setattr(mask_size_module, "cmaes_kcat_tuning", fake_cmaes_kcat_tuning) + + points = sweep_tunable_mask_size( + model, adapter=adapter, params=params, evotune_data=evotune_data, + target_impact_shares=[1.0], seeds=[0], n_proc=1, verbose=False, + ) + + assert points[0].rmse == (0.71934,)