You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
121 lines
4.1 KiB
Python
121 lines
4.1 KiB
Python
"""Experiment manifest loading, config merging, and run-ID generation."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from libs.backtest.domain import BacktestConfig, ExperimentManifest
|
|
from libs.backtest.snapshots import resolve_snapshot
|
|
from libs.common.ids import sha256_checksum_str
|
|
from libs.common.logging import get_logger
|
|
from libs.common.time_utils import utc_now
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
def load_base_config(config_path: str | Path) -> dict[str, Any]:
|
|
"""Load a base config JSON file."""
|
|
p = Path(config_path)
|
|
if not p.exists():
|
|
raise FileNotFoundError(f"Base config not found: {p}")
|
|
return json.loads(p.read_text())
|
|
|
|
|
|
def deep_merge(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
|
|
"""Recursively merge overrides into base (overrides win on conflict)."""
|
|
result = dict(base)
|
|
for key, value in overrides.items():
|
|
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
|
|
result[key] = deep_merge(result[key], value)
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def load_manifest(manifest_path: str | Path) -> ExperimentManifest:
|
|
"""Load and validate an experiment manifest JSON file."""
|
|
p = Path(manifest_path)
|
|
if not p.exists():
|
|
raise FileNotFoundError(f"Manifest not found: {p}")
|
|
raw = json.loads(p.read_text())
|
|
return ExperimentManifest.model_validate(raw)
|
|
|
|
|
|
def resolve_config(
|
|
manifest: ExperimentManifest,
|
|
config_root: str | Path | None = None,
|
|
snapshot_id_override: str | None = None,
|
|
) -> BacktestConfig:
|
|
"""Load base config, apply manifest overrides, validate into BacktestConfig.
|
|
|
|
Args:
|
|
manifest: The experiment manifest.
|
|
config_root: Root directory for resolving relative config paths.
|
|
snapshot_id_override: If provided, overrides the dataset_snapshot_id.
|
|
"""
|
|
base_config_path = manifest.base_config
|
|
if config_root is not None:
|
|
resolved_path = Path(config_root) / base_config_path
|
|
if resolved_path.exists():
|
|
base_config_path = str(resolved_path)
|
|
|
|
base = load_base_config(base_config_path)
|
|
merged = deep_merge(base, manifest.overrides)
|
|
|
|
requested_snapshot_id = snapshot_id_override or manifest.dataset_snapshot_id
|
|
resolution = resolve_snapshot(requested_snapshot_id)
|
|
merged["dataset_snapshot_id"] = resolution.canonical_snapshot_id
|
|
merged["requested_snapshot_id"] = requested_snapshot_id
|
|
merged["canonical_snapshot_id"] = resolution.canonical_snapshot_id
|
|
if manifest.strategy_engines:
|
|
merged["strategy_engines"] = [
|
|
engine.model_dump(mode="json")
|
|
for engine in manifest.strategy_engines
|
|
]
|
|
|
|
return BacktestConfig.model_validate(merged)
|
|
|
|
|
|
def _safe_slug(text: str, max_len: int = 20) -> str:
|
|
"""Convert text to safe alphanumeric slug."""
|
|
slug = re.sub(r"[^a-zA-Z0-9_-]", "_", text.strip())
|
|
return slug[:max_len]
|
|
|
|
|
|
def generate_run_id(
|
|
config: BacktestConfig,
|
|
strategy_override: str | None = None,
|
|
) -> str:
|
|
"""Generate a unique, deterministic run ID.
|
|
|
|
Format: bt_{safe_strategy}_{safe_snapshot[:12]}_{timestamp_us}_{config_hash[:8]}
|
|
"""
|
|
strategy = _safe_slug(strategy_override or config.strategy_name)
|
|
snapshot = _safe_slug(config.dataset_snapshot_id, max_len=12)
|
|
timestamp = utc_now().strftime("%Y%m%d%H%M%S%f")
|
|
config_json = config.model_dump_json(indent=None)
|
|
config_hash = sha256_checksum_str(config_json)[:8]
|
|
return f"bt_{strategy}_{snapshot}_{timestamp}_{config_hash}"
|
|
|
|
|
|
def save_resolved_config(
|
|
config: BacktestConfig,
|
|
run_dir: Path,
|
|
) -> Path:
|
|
"""Write resolved_config.json to the run directory."""
|
|
out = run_dir / "resolved_config.json"
|
|
out.write_text(config.model_dump_json(indent=2))
|
|
return out
|
|
|
|
|
|
def save_manifest(
|
|
manifest: ExperimentManifest,
|
|
run_dir: Path,
|
|
) -> Path:
|
|
"""Write manifest.json (copy of experiment manifest) to the run directory."""
|
|
out = run_dir / "manifest.json"
|
|
out.write_text(manifest.model_dump_json(indent=2))
|
|
return out
|