"""Experiment management: create, search, tree, diff, migrate, validate, index.""" from __future__ import annotations import fcntl import json import re import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Any from libs.backtest.snapshots import resolve_snapshot _EXPERIMENTS_DIR = Path("configs/experiments") _INDEX_FILE = ".index.json" _VALID_STATUSES = {"draft", "active", "promoted", "retired"} _LEGACY_STATUS_ALIASES = { "archived": "retired", } _META_FIELDS = { "id", "parent", "created_at", "created_by", "status", "generation", "version_family", "changelog", "aliases", "performance_summary", } # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _read_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text()) def _write_json(path: Path, data: dict[str, Any]) -> None: path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") def _exp_path(name: str, configs_dir: Path) -> Path: return configs_dir / f"{name}.json" def normalize_experiment_status(status: str | None) -> str: raw = str(status or "active").strip().lower() return _LEGACY_STATUS_ALIASES.get(raw, raw) def _infer_version_family(name: str) -> str | None: """Extract version family from experiment name, e.g. 'v6new', 'v8', 'v1'.""" m = re.search(r"_(v\d+[a-z]*)\.?\d*$", name) if m: return m.group(1) # Try generic name formats like conviction_v6new.307 m = re.search(r"_(v\d+[a-z]*)\.", name) if m: return m.group(1) return None def _infer_parent_from_description( description: str, experiment_name: str, all_names: set[str], family: str | None, ) -> str | None: """Try to infer parent experiment name from description text. Handles patterns: [v312+] ... -> parent is family.312 [v312] ... -> parent is family.312 v288 + ... -> parent is family.288 v6new.30 + ... -> parent is return_max_long_v6new.30 Derivative of NAME -> parent is NAME (exact name) """ if not description: return None desc = description.strip() # Pattern 1: explicit full name reference "Derivative of return_max_long_v1.274" m = re.search(r"Derivative of ([\w.]+)", desc, re.IGNORECASE) if m: candidate = m.group(1) if candidate in all_names: return candidate # Try adding common prefixes for prefix in ("return_max_long_",): full = prefix + candidate if full in all_names: return full # Pattern 2: bracket notation [v312+] or [v312] m = re.match(r"\[v(\d+)\+?\]", desc) if m and family: version = m.group(1) # Build candidate name by replacing version number in own name base_prefix = re.sub(r"\d+$", "", experiment_name.rsplit(".", 1)[0]) candidate = f"{base_prefix}{version}" if candidate in all_names: return candidate # Try common prefixes candidate2 = f"return_max_long_{family}.{version}" if candidate2 in all_names: return candidate2 # Pattern 3: explicit family.version reference "v6new.30 + ..." or "v6new.312 with ..." or "v6new.30 but ..." m = re.match(r"(v\d+[a-z]*)\.(\d+)\s*(?:\+|with\b|but\b)", desc) if m: ref_family = m.group(1) ref_version = m.group(2) candidate = f"return_max_long_{ref_family}.{ref_version}" if candidate in all_names: return candidate # Pattern 4: short version reference "v288 + ..." or "v288 but ..." or "v288 with ..." m = re.match(r"v(\d+)\s*(?:\+|with\b|but\b)", desc) if m and family: version = m.group(1) candidate = f"return_max_long_{family}.{version}" if candidate in all_names: return candidate return None def _get_git_creation_time(path: Path) -> str | None: """Get ISO-8601 timestamp of the file's first git commit.""" try: result = subprocess.run( ["git", "log", "--follow", "--format=%aI", "--", str(path)], capture_output=True, text=True, cwd=path.parent.parent.parent, ) lines = result.stdout.strip().splitlines() if lines: return lines[-1].strip() # oldest commit date except Exception: pass return None def _compute_generation( parent_name: str | None, all_data: dict[str, dict[str, Any]], depth: int = 0, ) -> int: if depth > 50 or parent_name is None: return depth parent = all_data.get(parent_name, {}) parent_parent = parent.get("parent") if parent_parent is None or parent_parent == parent_name: return depth + 1 return _compute_generation(parent_parent, all_data, depth + 1) # --------------------------------------------------------------------------- # Index management # --------------------------------------------------------------------------- def _index_path(configs_dir: Path) -> Path: return configs_dir / _INDEX_FILE def _is_index_stale(configs_dir: Path) -> bool: idx = _index_path(configs_dir) if not idx.exists(): return True idx_mtime = idx.stat().st_mtime for p in _iter_experiment_files(configs_dir): if p.stat().st_mtime > idx_mtime: return True return False def resolve_experiment_name( id_or_name: str, configs_dir: Path = _EXPERIMENTS_DIR, ) -> str: """Resolve an experiment ID (e.g. '228') or name to the canonical experiment name. Raises KeyError if the ID is not found. """ # If it looks like a plain integer, look it up in the index if id_or_name.isdigit(): target_id = int(id_or_name) idx_path = _index_path(configs_dir) if idx_path.exists(): try: index = _read_json(idx_path) for name, meta in index.get("experiments", {}).items(): if meta.get("id") == target_id: return name except Exception: pass # Fallback: scan files for p in _iter_experiment_files(configs_dir): try: data = _read_json(p) if data.get("id") == target_id: return data.get("experiment_name") or p.stem except Exception: pass raise KeyError(f"No experiment found with ID {target_id}") return id_or_name def next_experiment_id(configs_dir: Path = _EXPERIMENTS_DIR) -> int: """Return the next available sequential experiment ID (max existing + 1).""" max_id = 0 for p in _iter_experiment_files(configs_dir): try: eid = json.loads(p.read_text()).get("id") if isinstance(eid, int) and eid > max_id: max_id = eid except Exception: pass return max_id + 1 def _iter_experiment_files(configs_dir: Path): """Yield experiment JSON files, excluding the index file.""" for p in sorted(configs_dir.glob("*.json")): if p.name.startswith("."): continue yield p def rebuild_experiment_index( configs_dir: Path = _EXPERIMENTS_DIR, journal_path: Path | None = None, ) -> dict[str, Any]: """Scan all experiment JSONs and build .index.json metadata cache.""" experiments: dict[str, Any] = {} # Optionally cross-reference journal for SQS scores journal_sqs: dict[str, float | None] = {} journal_exists: set[str] = set() if journal_path and journal_path.exists(): for line in journal_path.read_text().splitlines(): line = line.strip() if not line: continue try: entry = json.loads(line) name = entry.get("experiment_name", "") journal_exists.add(name) score = entry.get("sqs_score") if score is not None: journal_sqs[name] = score except Exception: pass for p in _iter_experiment_files(configs_dir): try: data = _read_json(p) except Exception: continue name = data.get("experiment_name") or p.stem experiments[name] = { "id": data.get("id"), "parent": data.get("parent"), "version_family": data.get("version_family") or _infer_version_family(name), "generation": data.get("generation"), "status": normalize_experiment_status(data.get("status", "active")), "created_at": data.get("created_at"), "created_by": data.get("created_by"), "tags": data.get("tags", []), "aliases": data.get("aliases", []), "description": data.get("description"), "changelog": data.get("changelog"), "has_journal_entry": name in journal_exists, "sqs_score": journal_sqs.get(name), } index = { "updated_at": datetime.now(timezone.utc).isoformat(), "experiments": experiments, } _write_json(_index_path(configs_dir), index) return index def _load_index( configs_dir: Path = _EXPERIMENTS_DIR, journal_path: Path | None = None, ) -> dict[str, Any]: """Load index, rebuilding if stale or corrupted.""" if _is_index_stale(configs_dir): rebuild_experiment_index(configs_dir, journal_path) return _read_json(_index_path(configs_dir)) try: return _read_json(_index_path(configs_dir)) except (json.JSONDecodeError, ValueError): # Index is corrupted — rebuild it rebuild_experiment_index(configs_dir, journal_path) return _read_json(_index_path(configs_dir)) # --------------------------------------------------------------------------- # Core API # --------------------------------------------------------------------------- def create_experiment( parent_name: str, new_name: str, changelog: str | None = None, created_by: str = "ai_agent", configs_dir: Path = _EXPERIMENTS_DIR, ) -> Path: """Create a new experiment config by copying from a parent experiment. Sets parent, created_at, generation, version_family, status=draft, changelog. Returns the path of the created file. """ parent_path = _exp_path(parent_name, configs_dir) if not parent_path.exists(): raise FileNotFoundError(f"Parent experiment not found: {parent_path}") new_path = _exp_path(new_name, configs_dir) if new_path.exists(): raise FileExistsError(f"Experiment already exists: {new_path}") data = _read_json(parent_path) # Update identity fields data["experiment_name"] = new_name # Set metadata data["parent"] = parent_name data["created_at"] = datetime.now(timezone.utc).isoformat() data["created_by"] = created_by data["status"] = "draft" data["changelog"] = changelog data["version_family"] = _infer_version_family(new_name) or data.get("version_family") # Compute generation from parent parent_gen = data.get("generation") data["generation"] = (parent_gen + 1) if isinstance(parent_gen, int) else None # Clear runtime cache data["performance_summary"] = None # Phase-1 canonical snapshot migration: keep old manifests untouched, but # newly created experiments should point at the canonical snapshot id. requested_snapshot_id = data.get("dataset_snapshot_id") if isinstance(requested_snapshot_id, str) and requested_snapshot_id: try: data["dataset_snapshot_id"] = resolve_snapshot(requested_snapshot_id).canonical_snapshot_id except Exception: pass # Assign ID and write atomically under a file lock to prevent duplicate IDs # when multiple experiments are created concurrently. lock_path = configs_dir / ".id_lock" with open(lock_path, "w") as _lf: fcntl.flock(_lf, fcntl.LOCK_EX) data["id"] = next_experiment_id(configs_dir) _write_json(new_path, data) # Incrementally update index if it exists idx_path = _index_path(configs_dir) if idx_path.exists(): try: index = _read_json(idx_path) index["experiments"][new_name] = { "id": data["id"], "parent": parent_name, "version_family": data.get("version_family"), "generation": data.get("generation"), "status": "draft", "created_at": data["created_at"], "created_by": created_by, "tags": data.get("tags", []), "aliases": data.get("aliases", []), "description": data.get("description"), "changelog": changelog, "has_journal_entry": False, "sqs_score": None, } index["updated_at"] = datetime.now(timezone.utc).isoformat() _write_json(idx_path, index) except Exception: pass return new_path def search_experiments( configs_dir: Path = _EXPERIMENTS_DIR, journal_path: Path | None = None, *, tag: str | None = None, status: str | None = None, include_retired: bool = True, version_family: str | None = None, parent: str | None = None, name_pattern: str | None = None, has_journal_entry: bool | None = None, ) -> list[dict[str, Any]]: """Search experiments by metadata criteria. Returns list of index entries.""" index = _load_index(configs_dir, journal_path) results = [] normalized_status = normalize_experiment_status(status) if status else None for name, meta in index["experiments"].items(): meta_status = normalize_experiment_status(meta.get("status", "active")) if tag and tag not in meta.get("tags", []): continue if normalized_status and meta_status != normalized_status: continue if not include_retired and not normalized_status and meta_status == "retired": continue if version_family and meta.get("version_family") != version_family: continue if parent and meta.get("parent") != parent: continue if name_pattern and not re.search(name_pattern, name): continue if has_journal_entry is not None and meta.get("has_journal_entry") != has_journal_entry: continue results.append({"name": name, **meta, "status": meta_status}) return results def build_lineage_tree( ancestor_name: str, configs_dir: Path = _EXPERIMENTS_DIR, journal_path: Path | None = None, ) -> dict[str, Any]: """Build a lineage tree rooted at ancestor_name. Returns a nested dict: {"name": ..., "meta": ..., "children": [...]} """ index = _load_index(configs_dir, journal_path) experiments = index["experiments"] # Build parent->children map children_map: dict[str, list[str]] = {} for name, meta in experiments.items(): p = meta.get("parent") if p: children_map.setdefault(p, []).append(name) def _build_node(name: str, depth: int = 0) -> dict[str, Any]: meta = experiments.get(name, {}) children = sorted(children_map.get(name, [])) return { "name": name, "meta": meta, "depth": depth, "children": [_build_node(c, depth + 1) for c in children], } if ancestor_name not in experiments: raise KeyError(f"Experiment not found in index: {ancestor_name}") return _build_node(ancestor_name) def format_tree(node: dict[str, Any], prefix: str = "", is_last: bool = True) -> str: """Render a lineage tree as an indented text string.""" lines = [] connector = "└── " if is_last else "├── " meta = node.get("meta", {}) status = normalize_experiment_status(meta.get("status", "active")) sqs = meta.get("sqs_score") sqs_str = f" SQS={sqs:.1f}" if sqs is not None else "" aliases = meta.get("aliases", []) alias_str = f" [{', '.join(aliases)}]" if aliases else "" line = f"{prefix}{connector}{node['name']} [{status}]{sqs_str}{alias_str}" if node.get("depth") == 0: line = f"{node['name']} [{status}]{sqs_str}{alias_str}" lines.append(line) children = node.get("children", []) for i, child in enumerate(children): ext = " " if is_last else "│ " child_prefix = "" if node.get("depth") == 0 else prefix + ext lines.append(format_tree(child, child_prefix, i == len(children) - 1)) return "\n".join(lines) def diff_experiments( name_a: str, name_b: str, configs_dir: Path = _EXPERIMENTS_DIR, ) -> dict[str, tuple[Any, Any]]: """Compare two experiment configs. Returns dict of {path: (value_a, value_b)}. Compares overrides, strategy_engines (by engine_id), tags, dataset_snapshot_id. Metadata fields (parent, created_at, etc.) are excluded from the diff. """ path_a = _exp_path(name_a, configs_dir) path_b = _exp_path(name_b, configs_dir) if not path_a.exists(): raise FileNotFoundError(f"Experiment not found: {name_a}") if not path_b.exists(): raise FileNotFoundError(f"Experiment not found: {name_b}") data_a = _read_json(path_a) data_b = _read_json(path_b) diffs: dict[str, tuple[Any, Any]] = {} # Compare non-metadata top-level fields compare_keys = {"dataset_snapshot_id", "base_config", "overrides", "tags", "notes", "description"} for key in compare_keys: va = data_a.get(key) vb = data_b.get(key) if key == "overrides": _deep_diff(va or {}, vb or {}, f"overrides", diffs) elif va != vb: diffs[key] = (va, vb) # Compare strategy_engines by engine_id engines_a = {e.get("engine_id", i): e for i, e in enumerate(data_a.get("strategy_engines", []))} engines_b = {e.get("engine_id", i): e for i, e in enumerate(data_b.get("strategy_engines", []))} all_ids = sorted(set(engines_a) | set(engines_b)) for eid in all_ids: ea = engines_a.get(eid) eb = engines_b.get(eid) if ea is None: diffs[f"engines.{eid}"] = (None, eb) elif eb is None: diffs[f"engines.{eid}"] = (ea, None) else: _deep_diff(ea, eb, f"engines.{eid}", diffs) return diffs def _deep_diff( a: dict[str, Any], b: dict[str, Any], path: str, result: dict[str, tuple[Any, Any]], ) -> None: all_keys = set(a) | set(b) for k in sorted(all_keys): va = a.get(k) vb = b.get(k) full_path = f"{path}.{k}" if isinstance(va, dict) and isinstance(vb, dict): _deep_diff(va, vb, full_path, result) elif va != vb: result[full_path] = (va, vb) def set_experiment_status( experiment_name: str, new_status: str, configs_dir: Path = _EXPERIMENTS_DIR, ) -> None: """Update the status field of an experiment JSON in place.""" new_status = normalize_experiment_status(new_status) if new_status not in _VALID_STATUSES: raise ValueError(f"Invalid status: {new_status!r}. Must be one of {_VALID_STATUSES}") path = _exp_path(experiment_name, configs_dir) if not path.exists(): raise FileNotFoundError(f"Experiment not found: {path}") data = _read_json(path) data["status"] = new_status _write_json(path, data) # Update index if it exists idx_path = _index_path(configs_dir) if idx_path.exists(): try: index = _read_json(idx_path) if experiment_name in index.get("experiments", {}): index["experiments"][experiment_name]["status"] = new_status index["updated_at"] = datetime.now(timezone.utc).isoformat() _write_json(idx_path, index) except Exception: pass def validate_all_experiments( configs_dir: Path = _EXPERIMENTS_DIR, ) -> list[dict[str, str]]: """Validate all experiment JSON files. Returns list of {file, issue} dicts.""" from libs.backtest.domain import ExperimentManifest from pydantic import ValidationError issues: list[dict[str, str]] = [] all_names: set[str] = {p.stem for p in _iter_experiment_files(configs_dir)} for p in _iter_experiment_files(configs_dir): try: raw = _read_json(p) except Exception as e: issues.append({"file": p.name, "issue": f"JSON parse error: {e}"}) continue # Schema validation try: ExperimentManifest.model_validate(raw) except ValidationError as e: issues.append({"file": p.name, "issue": f"Schema validation failed: {e}"}) continue name = raw.get("experiment_name", "") # Name/filename match if name and name != p.stem: issues.append({"file": p.name, "issue": f"experiment_name '{name}' doesn't match filename '{p.stem}'"}) # base_config exists base = raw.get("base_config", "") if base and not Path(base).exists(): issues.append({"file": p.name, "issue": f"base_config path not found: {base}"}) # parent reference is valid parent = raw.get("parent") if parent and parent not in all_names: issues.append({"file": p.name, "issue": f"parent '{parent}' does not exist"}) # status value status = normalize_experiment_status(raw.get("status", "active")) if status not in _VALID_STATUSES: issues.append({"file": p.name, "issue": f"invalid status: {status!r}"}) return issues def migrate_experiments( configs_dir: Path = _EXPERIMENTS_DIR, dry_run: bool = False, ) -> list[dict[str, Any]]: """Backfill metadata fields for existing experiment JSON files. For each file missing metadata: - version_family: inferred from filename - parent: inferred from description patterns - created_at: from git log (first commit) or file mtime - generation: computed from parent chain - changelog: extracted from description after parent ref - status: defaulted to "active" if missing - created_by: defaulted to "human" Returns list of migration actions (what was inferred/set). """ all_paths = list(_iter_experiment_files(configs_dir)) all_names: set[str] = {p.stem for p in all_paths} # First pass: load all data all_data: dict[str, dict[str, Any]] = {} for p in all_paths: try: all_data[p.stem] = _read_json(p) except Exception: pass actions: list[dict[str, Any]] = [] # Second pass: infer metadata for each file inferred_parents: dict[str, str | None] = {} for name, data in all_data.items(): family = data.get("version_family") or _infer_version_family(name) desc = data.get("description") or "" inferred_parent = data.get("parent") or _infer_parent_from_description( desc, name, all_names, family ) inferred_parents[name] = inferred_parent # Third pass: compute generations def _get_gen(name: str, visited: set[str] | None = None) -> int: if visited is None: visited = set() if name in visited: return 0 # cycle protection visited.add(name) p = inferred_parents.get(name) if p is None or p not in all_data: return 0 return _get_gen(p, visited) + 1 # Compute ID assignments for files missing them (sorted by filename for determinism) existing_ids: set[int] = { d["id"] for d in all_data.values() if isinstance(d.get("id"), int) } next_id = max(existing_ids, default=0) + 1 id_assignments: dict[str, int] = {} for name in sorted(all_data): if all_data[name].get("id") is None: id_assignments[name] = next_id next_id += 1 # Fourth pass: write updates for name, data in all_data.items(): p = _exp_path(name, configs_dir) changes: dict[str, Any] = {} if data.get("id") is None: changes["id"] = id_assignments[name] family = _infer_version_family(name) if not data.get("version_family") and family: changes["version_family"] = family inferred_parent = inferred_parents.get(name) if not data.get("parent") and inferred_parent: changes["parent"] = inferred_parent if not data.get("created_at"): ts = _get_git_creation_time(p) if ts: changes["created_at"] = ts else: # Fall back to file mtime mtime = datetime.fromtimestamp(p.stat().st_mtime, tz=timezone.utc) changes["created_at"] = mtime.isoformat() if not data.get("created_by"): changes["created_by"] = "human" if data.get("status") is None: changes["status"] = "active" elif normalize_experiment_status(data.get("status")) != data.get("status"): changes["status"] = normalize_experiment_status(data.get("status")) computed_gen = _get_gen(name) if data.get("generation") != computed_gen: changes["generation"] = computed_gen if not data.get("changelog"): desc = data.get("description") or "" cl = _extract_changelog(desc) if cl: changes["changelog"] = cl if changes: actions.append({"name": name, "changes": changes}) if not dry_run: data.update(changes) _write_json(p, data) # Rebuild index after migration if not dry_run: rebuild_experiment_index(configs_dir) return actions def _extract_changelog(description: str) -> str | None: """Extract the delta text from a description after the parent reference.""" if not description: return None desc = description.strip() # Remove bracket prefix [v312+] desc = re.sub(r"^\[v\d+\+?\]\s*", "", desc) # Remove "vNNN + " prefix desc = re.sub(r"^v\d+\s*\+\s*", "", desc) # Remove "v6new.30 + " prefix desc = re.sub(r"^v\d+[a-z]*\.\d+\s*\+\s*", "", desc) # Remove "Derivative of NAME" prefix desc = re.sub(r"^Derivative of [\w.]+\s*(?:with)?\s*", "", desc, flags=re.IGNORECASE) desc = desc.strip() return desc if desc else None def get_ancestor_chain( experiment_name: str, configs_dir: Path = _EXPERIMENTS_DIR, ) -> list[dict[str, Any]]: """Return the full ancestor chain from root down to experiment_name (inclusive). Each entry: {"name": str, "id": int|None, "description": str|None, "changelog": str|None} """ chain: list[dict[str, Any]] = [] visited: set[str] = set() current = experiment_name while current and current not in visited: visited.add(current) p = _exp_path(current, configs_dir) if not p.exists(): break try: data = _read_json(p) except Exception: break chain.append({ "name": current, "id": data.get("id"), "description": data.get("description"), "changelog": data.get("changelog"), "status": normalize_experiment_status(data.get("status", "active")), }) current = data.get("parent") # type: ignore[assignment] chain.reverse() # root first return chain def update_performance_summary( experiment_name: str, summary: dict[str, Any], configs_dir: Path = _EXPERIMENTS_DIR, activate_if_draft: bool = True, ) -> None: """Cache performance metrics in the experiment JSON and optionally activate draft experiments.""" path = _exp_path(experiment_name, configs_dir) if not path.exists(): return data = _read_json(path) data["performance_summary"] = summary if activate_if_draft and data.get("status") == "draft": data["status"] = "active" _write_json(path, data) # Update index if it exists idx_path = _index_path(configs_dir) if idx_path.exists(): try: index = _read_json(idx_path) entry = index.get("experiments", {}).get(experiment_name) if entry is not None: entry["sqs_score"] = summary.get("sqs_score") if activate_if_draft and entry.get("status") == "draft": entry["status"] = "active" index["updated_at"] = datetime.now(timezone.utc).isoformat() _write_json(idx_path, index) except Exception: pass