"""Repository for persisting backtest settings configuration.""" from __future__ import annotations import json from datetime import date from pathlib import Path from typing import Any from uuid import UUID from app.models.backtest_settings import BacktestSettings class BacktestSettingsRepository: """Repository for persisting backtest settings configuration. Persists to `.workspaces/{workspace_id}/backtest_settings.json` """ def __init__(self, base_path: Path | str = Path(".workspaces")) -> None: self.base_path = Path(base_path) self.base_path.mkdir(parents=True, exist_ok=True) def load(self, workspace_id: str) -> BacktestSettings | None: """Load backtest settings for a workspace. Args: workspace_id: The workspace ID to load settings for Returns: BacktestSettings: The loaded settings, or None if no settings exist Raises: ValueError: If settings file is invalid """ settings_path = self._settings_path(workspace_id) if not settings_path.exists(): return None try: with open(settings_path) as f: data = json.load(f) return self._settings_from_dict(data) except (json.JSONDecodeError, KeyError) as e: raise ValueError(f"Invalid backtest settings file: {e}") from e def save(self, workspace_id: str, settings: BacktestSettings) -> None: """Save backtest settings for a workspace. Args: workspace_id: The workspace ID to save settings for settings: The settings to save """ settings_path = self._settings_path(workspace_id) settings_path.parent.mkdir(parents=True, exist_ok=True) payload = self._to_dict(settings) tmp_path = settings_path.with_name(f"{settings_path.name}.tmp") with open(tmp_path, "w") as f: json.dump(payload, f, indent=2) f.flush() # Atomic replace tmp_path.replace(settings_path) def _settings_path(self, workspace_id: str) -> Path: """Get the path to the settings file for a workspace.""" return self.base_path / workspace_id / "backtest_settings.json" def _to_dict(self, settings: BacktestSettings) -> dict[str, Any]: """Convert BacktestSettings to a dictionary for serialization.""" return { "settings_id": str(settings.settings_id), "name": settings.name, "data_source": settings.data_source, "dataset": settings.dataset, "schema": settings.schema, "start_date": settings.start_date.isoformat(), "end_date": settings.end_date.isoformat(), "underlying_symbol": settings.underlying_symbol, "start_price": settings.start_price, "underlying_units": settings.underlying_units, "loan_amount": settings.loan_amount, "margin_call_ltv": settings.margin_call_ltv, "template_slugs": list(settings.template_slugs), "cache_key": settings.cache_key, "data_cost_usd": settings.data_cost_usd, "provider_ref": { "provider_id": settings.provider_ref.provider_id, "pricing_mode": settings.provider_ref.pricing_mode, }, } def _settings_from_dict(self, data: dict[str, Any]) -> BacktestSettings: """Create BacktestSettings from a dictionary.""" # Handle potential string dates start_date = data["start_date"] if isinstance(start_date, str): start_date = date.fromisoformat(start_date) end_date = data["end_date"] if isinstance(end_date, str): end_date = date.fromisoformat(end_date) # Import here to avoid circular import issues at module level from app.models.backtest import ProviderRef return BacktestSettings( settings_id=UUID(data["settings_id"]), name=data["name"], data_source=data["data_source"], dataset=data["dataset"], schema=data["schema"], start_date=start_date, end_date=end_date, underlying_symbol=data["underlying_symbol"], start_price=data["start_price"], underlying_units=data["underlying_units"], loan_amount=data["loan_amount"], margin_call_ltv=data["margin_call_ltv"], template_slugs=tuple(data["template_slugs"]), cache_key=data["cache_key"], data_cost_usd=data["data_cost_usd"], provider_ref=ProviderRef( provider_id=data["provider_ref"]["provider_id"], pricing_mode=data["provider_ref"]["pricing_mode"], ), )