ReNunney/tests/test_track1_fit.py

201 lines
6.9 KiB
Python

import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SRC_DIR = ROOT / "src"
if str(SRC_DIR) not in sys.path:
sys.path.insert(0, str(SRC_DIR))
import renunney.track1_api as api
import renunney.track1_fit as fit
def _synthetic_run_rows():
return [
{
"seed": 1,
"K": 500,
"N0": 500,
"n": 1,
"u": 0.005,
"M": 5.0,
"R": 10.0,
"T": 20,
"epochs": 2,
"p": 0.5,
"generations_recorded": 25,
"extinction_occurred": False,
"first_extinction_t": None,
"final_extinct": False,
"final_N": 300,
"min_N": 200,
"max_N": 520,
"mean_N": 360.0,
"final_mean_allele_value": 1.8,
"final_target_value": 1.9,
"final_tracking_gap": -0.1,
"mean_abs_tracking_gap": 0.2,
"max_abs_tracking_gap": 0.4,
"first_nonzero_allele_t": -3,
"last_nonzero_allele_t": 19,
"stayed_zero_after_initialization": False,
"first_productivity_below_replacement_t": None,
"fraction_generations_below_replacement": 0.1,
"longest_zero_mutation_streak": 0,
"cumulative_expected_mutations": 90.0,
"cumulative_realized_mutations": 110,
"cumulative_mutation_shortfall": 6.0,
},
{
"seed": 2,
"K": 500,
"N0": 500,
"n": 1,
"u": 0.001,
"M": 1.0,
"R": 10.0,
"T": 10,
"epochs": 2,
"p": 0.5,
"generations_recorded": 25,
"extinction_occurred": True,
"first_extinction_t": 15,
"final_extinct": True,
"final_N": 0,
"min_N": 0,
"max_N": 500,
"mean_N": 140.0,
"final_mean_allele_value": 0.2,
"final_target_value": 1.9,
"final_tracking_gap": -1.7,
"mean_abs_tracking_gap": 1.1,
"max_abs_tracking_gap": 1.8,
"first_nonzero_allele_t": 2,
"last_nonzero_allele_t": 8,
"stayed_zero_after_initialization": False,
"first_productivity_below_replacement_t": -1,
"fraction_generations_below_replacement": 0.9,
"longest_zero_mutation_streak": 7,
"cumulative_expected_mutations": 12.0,
"cumulative_realized_mutations": 3,
"cumulative_mutation_shortfall": 9.0,
},
{
"seed": 3,
"K": 500,
"N0": 20,
"n": 1,
"u": 0.001,
"M": 1.0,
"R": 10.0,
"T": 10,
"epochs": 2,
"p": 0.5,
"generations_recorded": 25,
"extinction_occurred": True,
"first_extinction_t": 12,
"final_extinct": True,
"final_N": 0,
"min_N": 0,
"max_N": 200,
"mean_N": 70.0,
"final_mean_allele_value": 0.0,
"final_target_value": 1.9,
"final_tracking_gap": -1.9,
"mean_abs_tracking_gap": 1.3,
"max_abs_tracking_gap": 2.0,
"first_nonzero_allele_t": None,
"last_nonzero_allele_t": None,
"stayed_zero_after_initialization": True,
"first_productivity_below_replacement_t": -3,
"fraction_generations_below_replacement": 1.0,
"longest_zero_mutation_streak": 12,
"cumulative_expected_mutations": 8.0,
"cumulative_realized_mutations": 1,
"cumulative_mutation_shortfall": 7.0,
},
{
"seed": 4,
"K": 500,
"N0": 500,
"n": 1,
"u": 0.005,
"M": 5.0,
"R": 10.0,
"T": 10,
"epochs": 2,
"p": 0.5,
"generations_recorded": 25,
"extinction_occurred": False,
"first_extinction_t": None,
"final_extinct": False,
"final_N": 380,
"min_N": 180,
"max_N": 520,
"mean_N": 350.0,
"final_mean_allele_value": 1.7,
"final_target_value": 1.9,
"final_tracking_gap": -0.2,
"mean_abs_tracking_gap": 0.25,
"max_abs_tracking_gap": 0.6,
"first_nonzero_allele_t": -2,
"last_nonzero_allele_t": 19,
"stayed_zero_after_initialization": False,
"first_productivity_below_replacement_t": 1,
"fraction_generations_below_replacement": 0.2,
"longest_zero_mutation_streak": 1,
"cumulative_expected_mutations": 80.0,
"cumulative_realized_mutations": 90,
"cumulative_mutation_shortfall": 10.0,
},
]
def test_fit_extinction_run_model_returns_model_and_summary():
model, summary = fit.fit_extinction_run_model(_synthetic_run_rows())
assert model.converged is True
assert model.feature_names == list(fit.DEFAULT_RUN_FEATURES)
assert len(model.feature_means) == len(model.feature_names)
assert len(model.feature_scales) == len(model.feature_names)
assert len(model.coefficients) == len(model.feature_names)
assert summary.sample_count == 4
assert summary.extinction_count == 2
assert 0.0 <= summary.brier_score <= 1.0
def test_fit_payload_from_jsonl_and_api_mode(tmp_path: Path):
path = tmp_path / "run_rows.jsonl"
with path.open("w", encoding="utf-8") as handle:
for row in _synthetic_run_rows():
handle.write(json.dumps(row, sort_keys=True) + "\n")
payload = fit.fit_payload_from_jsonl(path)
assert payload["summary"]["sample_count"] == 4
assert payload["model"]["converged"] is True
config = api.Track1RunConfig(mode="extinction_fit", run_rows_path=str(path))
api_payload = api.run_config(config)
assert api_payload["mode"] == "extinction_fit"
assert api_payload["fit_status"] == "ok"
assert api_payload["summary"]["extinction_count"] == 2
def test_api_extinction_fit_reports_insufficient_outcome_variation(tmp_path: Path):
path = tmp_path / "run_rows.jsonl"
only_survivors = _synthetic_run_rows()[:1] + [_synthetic_run_rows()[3]]
with path.open("w", encoding="utf-8") as handle:
for row in only_survivors:
row = dict(row)
row["extinction_occurred"] = False
row["final_extinct"] = False
row["first_extinction_t"] = None
handle.write(json.dumps(row, sort_keys=True) + "\n")
config = api.Track1RunConfig(mode="extinction_fit", run_rows_path=str(path))
payload = api.run_config(config)
assert payload["mode"] == "extinction_fit"
assert payload["fit_status"] == "insufficient_outcome_variation"
assert payload["model"] is None
assert payload["summary"]["sample_count"] == 2