"""Unit tests for libs/backtest/experiments.py""" from __future__ import annotations import json import tempfile from pathlib import Path import pytest from libs.backtest.experiments import ( _extract_changelog, _infer_parent_from_description, _infer_version_family, build_lineage_tree, create_experiment, diff_experiments, format_tree, migrate_experiments, rebuild_experiment_index, search_experiments, set_experiment_status, update_performance_summary, validate_all_experiments, ) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- def _make_exp(name: str, extra: dict | None = None) -> dict: """Build a minimal valid experiment JSON dict.""" data = { "experiment_name": name, "dataset_snapshot_id": "test-snapshot", "description": f"{name} description", "base_config": "configs/backtest/return_max_long_v1.json", "overrides": {"signal": {"score_threshold": 0.45}}, "strategy_engines": [], "splits": [], "tags": ["return-max", "v6new"], "notes": None, } if extra: data.update(extra) return data @pytest.fixture def tmpdir_experiments(tmp_path: Path) -> Path: """Create a temporary configs/experiments directory with a few experiment files.""" exp_dir = tmp_path / "configs" / "experiments" exp_dir.mkdir(parents=True) # Create base_config file so validation passes base_dir = tmp_path / "configs" / "backtest" base_dir.mkdir(parents=True) (base_dir / "return_max_long_v1.json").write_text("{}") experiments = { "return_max_long_v6new.29": _make_exp("return_max_long_v6new.29", { "description": "[baseline] v29", "tags": ["return-max", "v6new"], "aliases": ["baseline"], "status": "promoted", "version_family": "v6new", "generation": 0, }), "return_max_long_v6new.84": _make_exp("return_max_long_v6new.84", { "description": "v29 + momentum scaler", "tags": ["return-max", "v6new", "momentum"], "parent": "return_max_long_v6new.29", "version_family": "v6new", "generation": 1, "overrides": { "signal": {"score_threshold": 0.50}, "risk": {"per_trade_risk_pct": 0.07}, }, }), "return_max_long_v6new.100": _make_exp("return_max_long_v6new.100", { "description": "v84 + vol gate", "tags": ["return-max", "v6new", "vol"], "parent": "return_max_long_v6new.84", "version_family": "v6new", "generation": 2, }), } for name, data in experiments.items(): (exp_dir / f"{name}.json").write_text(json.dumps(data, indent=2)) return exp_dir # --------------------------------------------------------------------------- # _infer_version_family # --------------------------------------------------------------------------- @pytest.mark.parametrize("name,expected", [ ("return_max_long_v6new.315", "v6new"), ("return_max_long_v8.14", "v8"), ("return_max_long_v1.298", "v1"), ("conviction_v6new.307", "v6new"), ("return_max_long_baseline", None), ]) def test_infer_version_family(name, expected): assert _infer_version_family(name) == expected # --------------------------------------------------------------------------- # _infer_parent_from_description # --------------------------------------------------------------------------- @pytest.mark.parametrize("desc,exp_name,expected_parent", [ ("[v29+] momentum scaler", "return_max_long_v6new.84", "return_max_long_v6new.29"), ("[v84] vol gate", "return_max_long_v6new.100", "return_max_long_v6new.84"), ("v84 + vol gate", "return_max_long_v6new.100", "return_max_long_v6new.84"), ("v6new.29 + momentum", "return_max_long_v6new.84", "return_max_long_v6new.29"), ("Derivative of return_max_long_v6new.29 with changes", "return_max_long_v6new.84", "return_max_long_v6new.29"), ("unrelated description", "return_max_long_v6new.100", None), ]) def test_infer_parent_from_description(desc, exp_name, expected_parent): all_names = { "return_max_long_v6new.29", "return_max_long_v6new.84", "return_max_long_v6new.100", } result = _infer_parent_from_description(desc, exp_name, all_names, "v6new") assert result == expected_parent # --------------------------------------------------------------------------- # _extract_changelog # --------------------------------------------------------------------------- @pytest.mark.parametrize("desc,expected", [ ("[v312+] disable weak + boost broad_oneoff", "disable weak + boost broad_oneoff"), ("v288 + recovery engines early fail day5", "recovery engines early fail day5"), ("v6new.30 + momentum scaler", "momentum scaler"), ("Derivative of v1.274 with tighter stop", "tighter stop"), ("[baseline] 원본 v29", "[baseline] 원본 v29"), # non-version bracket labels are preserved ("", None), ]) def test_extract_changelog(desc, expected): assert _extract_changelog(desc) == expected # --------------------------------------------------------------------------- # create_experiment # --------------------------------------------------------------------------- def test_create_experiment_basic(tmpdir_experiments: Path): new_name = "return_max_long_v6new.200" path = create_experiment( parent_name="return_max_long_v6new.29", new_name=new_name, changelog="add entropy gate", created_by="test", configs_dir=tmpdir_experiments, ) assert path.exists() data = json.loads(path.read_text()) assert data["experiment_name"] == new_name assert data["parent"] == "return_max_long_v6new.29" assert data["created_by"] == "test" assert data["status"] == "draft" assert data["changelog"] == "add entropy gate" assert data["created_at"] is not None # generation should be parent's (0) + 1 assert data["generation"] == 1 assert data["performance_summary"] is None def test_create_experiment_canonicalizes_registered_snapshot(tmp_path: Path): exp_dir = tmp_path / "configs" / "experiments" exp_dir.mkdir(parents=True) base_dir = tmp_path / "configs" / "backtest" base_dir.mkdir(parents=True) (base_dir / "return_max_long_v1.json").write_text("{}") parent_name = "return_max_long_v12.61" parent_payload = _make_exp( parent_name, { "dataset_snapshot_id": "midlarge-liquid-long-v1_bucketfix_full_audit_tier3tech", "version_family": "v12", "generation": 1, }, ) (exp_dir / f"{parent_name}.json").write_text(json.dumps(parent_payload, indent=2)) path = create_experiment( parent_name=parent_name, new_name="return_max_long_v12.62", configs_dir=exp_dir, ) data = json.loads(path.read_text()) assert data["dataset_snapshot_id"] == "midlarge-liquid-long-v1_bucketfix_full_audit_canonical" def test_create_experiment_parent_not_found(tmpdir_experiments: Path): with pytest.raises(FileNotFoundError): create_experiment( parent_name="nonexistent", new_name="return_max_long_v6new.999", configs_dir=tmpdir_experiments, ) def test_create_experiment_already_exists(tmpdir_experiments: Path): with pytest.raises(FileExistsError): create_experiment( parent_name="return_max_long_v6new.29", new_name="return_max_long_v6new.84", # already exists configs_dir=tmpdir_experiments, ) # --------------------------------------------------------------------------- # search_experiments # --------------------------------------------------------------------------- def test_search_by_tag(tmpdir_experiments: Path): results = search_experiments(configs_dir=tmpdir_experiments, tag="momentum") assert len(results) == 1 assert results[0]["name"] == "return_max_long_v6new.84" def test_search_by_status(tmpdir_experiments: Path): results = search_experiments(configs_dir=tmpdir_experiments, status="promoted") assert len(results) == 1 assert results[0]["name"] == "return_max_long_v6new.29" def test_search_by_family(tmpdir_experiments: Path): results = search_experiments(configs_dir=tmpdir_experiments, version_family="v6new") assert len(results) == 3 def test_search_by_parent(tmpdir_experiments: Path): results = search_experiments( configs_dir=tmpdir_experiments, parent="return_max_long_v6new.29", ) assert len(results) == 1 assert results[0]["name"] == "return_max_long_v6new.84" def test_search_by_name_pattern(tmpdir_experiments: Path): results = search_experiments(configs_dir=tmpdir_experiments, name_pattern=r"\.1\d\d$") assert len(results) == 1 assert results[0]["name"] == "return_max_long_v6new.100" def test_search_empty_criteria_returns_all(tmpdir_experiments: Path): results = search_experiments(configs_dir=tmpdir_experiments) assert len(results) == 3 # --------------------------------------------------------------------------- # build_lineage_tree / format_tree # --------------------------------------------------------------------------- def test_build_lineage_tree(tmpdir_experiments: Path): tree = build_lineage_tree("return_max_long_v6new.29", configs_dir=tmpdir_experiments) assert tree["name"] == "return_max_long_v6new.29" assert tree["depth"] == 0 assert len(tree["children"]) == 1 child = tree["children"][0] assert child["name"] == "return_max_long_v6new.84" assert len(child["children"]) == 1 grandchild = child["children"][0] assert grandchild["name"] == "return_max_long_v6new.100" def test_build_lineage_tree_unknown(tmpdir_experiments: Path): with pytest.raises(KeyError): build_lineage_tree("nonexistent", configs_dir=tmpdir_experiments) def test_format_tree(tmpdir_experiments: Path): tree = build_lineage_tree("return_max_long_v6new.29", configs_dir=tmpdir_experiments) rendered = format_tree(tree) assert "return_max_long_v6new.29" in rendered assert "return_max_long_v6new.84" in rendered assert "return_max_long_v6new.100" in rendered # --------------------------------------------------------------------------- # diff_experiments # --------------------------------------------------------------------------- def test_diff_overrides(tmpdir_experiments: Path): diffs = diff_experiments( "return_max_long_v6new.29", "return_max_long_v6new.84", configs_dir=tmpdir_experiments, ) # v84 has score_threshold=0.50 vs v29's 0.45 assert "overrides.signal.score_threshold" in diffs assert diffs["overrides.signal.score_threshold"] == (0.45, 0.50) # v84 has extra risk override (v29 has no risk key, so top-level overrides.risk differs) assert "overrides.risk" in diffs assert diffs["overrides.risk"][0] is None # v29 has no risk overrides assert "per_trade_risk_pct" in diffs["overrides.risk"][1] def test_diff_identical(tmpdir_experiments: Path): # Copy v29 to a temp file, compare with itself import shutil src = tmpdir_experiments / "return_max_long_v6new.29.json" dst = tmpdir_experiments / "return_max_long_v6new.29_copy.json" shutil.copy(src, dst) # Patch experiment_name to match filename data = json.loads(dst.read_text()) data["experiment_name"] = "return_max_long_v6new.29_copy" dst.write_text(json.dumps(data, indent=2)) diffs = diff_experiments( "return_max_long_v6new.29", "return_max_long_v6new.29_copy", configs_dir=tmpdir_experiments, ) # Should only differ in experiment_name (which is not in compare_keys) assert diffs == {} def test_diff_missing_experiment(tmpdir_experiments: Path): with pytest.raises(FileNotFoundError): diff_experiments("return_max_long_v6new.29", "nonexistent", configs_dir=tmpdir_experiments) # --------------------------------------------------------------------------- # set_experiment_status # --------------------------------------------------------------------------- def test_set_experiment_status(tmpdir_experiments: Path): set_experiment_status("return_max_long_v6new.84", "retired", configs_dir=tmpdir_experiments) data = json.loads((tmpdir_experiments / "return_max_long_v6new.84.json").read_text()) assert data["status"] == "retired" def test_set_experiment_status_legacy_archived_alias(tmpdir_experiments: Path): set_experiment_status("return_max_long_v6new.84", "archived", configs_dir=tmpdir_experiments) data = json.loads((tmpdir_experiments / "return_max_long_v6new.84.json").read_text()) assert data["status"] == "retired" def test_set_experiment_status_invalid(tmpdir_experiments: Path): with pytest.raises(ValueError): set_experiment_status("return_max_long_v6new.84", "invalid_status", configs_dir=tmpdir_experiments) # --------------------------------------------------------------------------- # validate_all_experiments # --------------------------------------------------------------------------- def test_validate_all_no_issues(tmpdir_experiments: Path, monkeypatch): # Patch Path to resolve base_config from tmpdir import libs.backtest.experiments as exp_mod original_exists = Path.exists def patched_exists(self): if str(self) == "configs/backtest/return_max_long_v1.json": return True return original_exists(self) monkeypatch.setattr(Path, "exists", patched_exists) issues = validate_all_experiments(configs_dir=tmpdir_experiments) assert issues == [] def test_validate_detects_bad_parent(tmpdir_experiments: Path, monkeypatch): # Add an experiment with a non-existent parent bad = _make_exp("return_max_long_v6new.999", {"parent": "nonexistent_parent"}) (tmpdir_experiments / "return_max_long_v6new.999.json").write_text(json.dumps(bad, indent=2)) import libs.backtest.experiments as exp_mod original_exists = Path.exists def patched_exists(self): if str(self) == "configs/backtest/return_max_long_v1.json": return True return original_exists(self) monkeypatch.setattr(Path, "exists", patched_exists) issues = validate_all_experiments(configs_dir=tmpdir_experiments) issue_files = [i["file"] for i in issues] assert "return_max_long_v6new.999.json" in issue_files # --------------------------------------------------------------------------- # migrate_experiments # --------------------------------------------------------------------------- def test_migrate_infers_version_family(tmp_path: Path): exp_dir = tmp_path / "configs" / "experiments" exp_dir.mkdir(parents=True) # File without version_family data = _make_exp("return_max_long_v6new.50") del data["tags"] # minimal data["tags"] = [] (exp_dir / "return_max_long_v6new.50.json").write_text(json.dumps(data)) actions = migrate_experiments(configs_dir=exp_dir, dry_run=True) assert any("version_family" in a["changes"] for a in actions) v_action = next(a for a in actions if "version_family" in a["changes"]) assert v_action["changes"]["version_family"] == "v6new" def test_migrate_dry_run_does_not_write(tmp_path: Path): exp_dir = tmp_path / "configs" / "experiments" exp_dir.mkdir(parents=True) data = _make_exp("return_max_long_v6new.50") path = exp_dir / "return_max_long_v6new.50.json" path.write_text(json.dumps(data)) original_mtime = path.stat().st_mtime import time time.sleep(0.01) migrate_experiments(configs_dir=exp_dir, dry_run=True) assert path.stat().st_mtime == pytest.approx(original_mtime, abs=0.05) def test_migrate_infers_parent_from_description(tmp_path: Path): exp_dir = tmp_path / "configs" / "experiments" exp_dir.mkdir(parents=True) (exp_dir / "return_max_long_v6new.29.json").write_text(json.dumps(_make_exp("return_max_long_v6new.29"))) child = _make_exp("return_max_long_v6new.84", {"description": "v29 + momentum scaler"}) (exp_dir / "return_max_long_v6new.84.json").write_text(json.dumps(child)) actions = migrate_experiments(configs_dir=exp_dir, dry_run=True) child_action = next((a for a in actions if a["name"] == "return_max_long_v6new.84"), None) if child_action: # Parent may have been inferred assert child_action["changes"].get("parent") == "return_max_long_v6new.29" or \ "parent" not in child_action["changes"] # --------------------------------------------------------------------------- # rebuild_experiment_index # --------------------------------------------------------------------------- def test_rebuild_index(tmpdir_experiments: Path): index = rebuild_experiment_index(configs_dir=tmpdir_experiments) assert "experiments" in index assert len(index["experiments"]) == 3 assert "return_max_long_v6new.29" in index["experiments"] entry = index["experiments"]["return_max_long_v6new.29"] assert entry["version_family"] == "v6new" assert entry["status"] == "promoted" assert entry["generation"] == 0 def test_index_file_created(tmpdir_experiments: Path): rebuild_experiment_index(configs_dir=tmpdir_experiments) assert (tmpdir_experiments / ".index.json").exists() # --------------------------------------------------------------------------- # update_performance_summary # --------------------------------------------------------------------------- def test_update_performance_summary(tmpdir_experiments: Path): update_performance_summary( "return_max_long_v6new.84", {"sqs_score": 72.5, "public_sqs": 68.0, "trade_count_test": 150}, configs_dir=tmpdir_experiments, ) data = json.loads((tmpdir_experiments / "return_max_long_v6new.84.json").read_text()) assert data["performance_summary"]["sqs_score"] == 72.5 def test_update_performance_summary_activates_draft(tmpdir_experiments: Path): # First set to draft set_experiment_status("return_max_long_v6new.84", "draft", configs_dir=tmpdir_experiments) update_performance_summary( "return_max_long_v6new.84", {"sqs_score": 72.5}, configs_dir=tmpdir_experiments, activate_if_draft=True, ) data = json.loads((tmpdir_experiments / "return_max_long_v6new.84.json").read_text()) assert data["status"] == "active" def test_update_performance_summary_nonexistent(tmpdir_experiments: Path): # Should not raise — silently skip missing experiments update_performance_summary( "nonexistent_experiment", {"sqs_score": 72.5}, configs_dir=tmpdir_experiments, ) # --------------------------------------------------------------------------- # ExperimentManifest schema backward compatibility # --------------------------------------------------------------------------- def test_existing_manifest_loads_without_metadata(tmpdir_experiments: Path): """Old-style manifests without metadata fields should load fine.""" from libs.backtest.domain import ExperimentManifest old_style = { "experiment_name": "return_max_long_v6new.50", "dataset_snapshot_id": "test-snapshot", "base_config": "configs/backtest/return_max_long_v1.json", "overrides": {}, "strategy_engines": [], "splits": [], "tags": ["return-max"], } manifest = ExperimentManifest.model_validate(old_style) assert manifest.parent is None assert manifest.status == "active" assert manifest.generation is None assert manifest.aliases == [] assert manifest.performance_summary is None def test_new_manifest_with_metadata(): """New manifests with metadata fields should parse correctly.""" from libs.backtest.domain import ExperimentManifest new_style = { "experiment_name": "return_max_long_v6new.315", "dataset_snapshot_id": "test-snapshot", "base_config": "configs/backtest/return_max_long_v1.json", "overrides": {}, "strategy_engines": [], "splits": [], "tags": ["return-max", "v6new"], "parent": "return_max_long_v6new.312", "created_at": "2026-03-25T10:00:00+00:00", "created_by": "ai_agent", "status": "active", "generation": 5, "version_family": "v6new", "changelog": "disable weak + boost broad_oneoff", "aliases": ["conviction"], "performance_summary": {"sqs_score": 68.5}, } manifest = ExperimentManifest.model_validate(new_style) assert manifest.parent == "return_max_long_v6new.312" assert manifest.status == "active" assert manifest.generation == 5 assert manifest.version_family == "v6new" assert manifest.changelog == "disable weak + boost broad_oneoff" assert manifest.aliases == ["conviction"] assert manifest.performance_summary == {"sqs_score": 68.5}