study_spec.py

Go to the documentation of this file.

"""Dependency-free StudySpec v1 JSON loader and validator."""
 
from __future__ import annotations
 
from dataclasses import dataclass, field
from fractions import Fraction
import json
import math
from pathlib import Path
from typing import Any, Mapping
 
 
 
 
@dataclass(frozen=True)
class ValidationIssue:
    """One path-qualified StudySpec validation failure."""
 
    path: str
    message: str
 
    def __str__(self) -> str:
        return f"{self.path}: {self.message}"
 
 
class StudySpecError(ValueError):
    """Aggregate error containing every issue found in one StudySpec."""
 
    def __init__(self, issues: list[ValidationIssue] | tuple[ValidationIssue, ...]):
        self.issues = tuple(issues)
        super().__init__("; ".join(str(issue) for issue in self.issues))
 
 
@dataclass(frozen=True)
class ParameterSpec:
    """Normalized integer, real, Boolean, or categorical search dimension."""
 
    kind: str
    low: int | float | None = None
    high: int | float | None = None
    step: int | float | None = None
    log: bool = False
    choices: tuple[JsonScalar, ...] = ()
 
 
@dataclass(frozen=True)
class StrategySpec:
    """One Pine source or compiled artifact and its runtime inputs."""
 
    id: str
    source: Path | None
    artifact: Path | None
    dataset_ids: tuple[str, ...]
    fixed_inputs: Mapping[str, JsonScalar]
    strategy_overrides: Mapping[str, JsonScalar]
    search_space: Mapping[str, ParameterSpec]
 
 
@dataclass(frozen=True)
class DatasetSpec:
    """One OHLCV source and its PineForge timeframe interpretation."""
 
    id: str
    ohlcv: Path
    input_tf: str
    script_tf: str
    chart_timezone: str
 
 
@dataclass(frozen=True)
class ObjectiveSpec:
    """Objective direction, expression or registration, and constraints."""
 
    kind: str
    direction: str
    expression: str | None = None
    name: str | None = None
    constraints: tuple[str, ...] = ()
    requires: tuple[str, ...] = ()
    config: Mapping[str, Any] = field(default_factory=dict)
    nan_policy: str = "fail_trial"
    division_by_zero: str = "fail_trial"
 
 
@dataclass(frozen=True)
class TpeSamplerConfig:
    """Validated native TPE model and batching controls."""
 
    startup_trials: int = 10
    ei_candidates: int = 24
    gamma_fraction: float = 0.10
    gamma_cap: int = 25
    prior_weight: float = 1.0
    constant_liar: bool = True
 
 
@dataclass(frozen=True)
class SamplerSpec:
    """Sampler selection, deterministic seed, budget, and candidate policy."""
 
    kind: str
    seed: int
    trials: int
    candidate_policy: str = "sampler_default"
    config: TpeSamplerConfig | None = None
 
 
@dataclass(frozen=True)
class ExecutionSpec:
    """Native trial worker and failure-policy configuration."""
 
    workers: int
    isolation: str
    timeout_seconds: float | None = None
    fail_fast: bool = False
 
 
@dataclass(frozen=True)
class StudySpec:
    """Fully validated and path-resolved executable StudySpec v1."""
 
    schema_version: int
    mode: str
    strategy: StrategySpec
    datasets: tuple[DatasetSpec, ...]
    objective: ObjectiveSpec
    sampler: SamplerSpec
    execution: ExecutionSpec
    spec_path: Path
 
    @property
    def search_space(self) -> Mapping[str, ParameterSpec]:
        """Return the selected strategy's tunable input dimensions."""
 
        return self.strategy.search_space
 
    @property
    def fixed_inputs(self) -> Mapping[str, JsonScalar]:
        """Return Pine inputs applied unchanged to every trial."""
 
        return self.strategy.fixed_inputs
 
    @property
    def overrides(self) -> Mapping[str, JsonScalar]:
        """Return runtime ``strategy(...)`` overrides applied to every trial."""
 
        return self.strategy.strategy_overrides
 
    @classmethod
    def from_json(cls, path: str | Path, *, require_files: bool = False) -> "StudySpec":
        """Load a StudySpec with the same validation as :func:`load_study_spec`."""
 
        return load_study_spec(path, require_files=require_files)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
def load_study_spec(path: str | Path, *, require_files: bool = False) -> StudySpec:
    """Load StudySpec v1 and resolve every filesystem path relative to its JSON file."""
 
    spec_path = Path(path).expanduser().resolve()
    try:
        source = spec_path.read_text(encoding="utf-8")
    except OSError as error:
        raise StudySpecError(
            [ValidationIssue("$", f"cannot read {spec_path}: {error}")]
        ) from error
    try:
        document = json.loads(
            source,
            object_pairs_hook=_reject_duplicate_keys,
            parse_constant=_reject_json_constant,
        )
    except StudySpecError:
        raise
    except json.JSONDecodeError as error:
        raise StudySpecError(
            [
                ValidationIssue(
                    "$",
                    f"invalid JSON at line {error.lineno}, "
                    f"column {error.colno}: {error.msg}",
                )
            ]
        ) from error
 
    issues: list[ValidationIssue] = []
    root = _object(document, "$", issues)
    _check_unknown(
        root,
        {
            "schema_version",
            "mode",
            "strategies",
            "datasets",
            "objective",
            "sampler",
            "execution",
        },
        "$",
        issues,
    )
    schema_version = root.get("schema_version")
    if schema_version != 1:
        issues.append(ValidationIssue("$.schema_version", "must equal 1"))
    mode = root.get("mode")
    if mode != "single_strategy":
        issues.append(
            ValidationIssue("$.mode", "this loader currently supports single_strategy")
        )
 
    strategies = root.get("strategies")
    strategy_raw: Any = {}
    if not isinstance(strategies, list) or len(strategies) != 1:
        issues.append(
            ValidationIssue("$.strategies", "must contain exactly one strategy")
        )
        if isinstance(strategies, list) and strategies:
            strategy_raw = strategies[0]
    else:
        strategy_raw = strategies[0]
    strategy = _parse_strategy(strategy_raw, spec_path.parent, issues)
 
    datasets_raw = root.get("datasets")
    datasets: list[DatasetSpec] = []
    if not isinstance(datasets_raw, list) or not datasets_raw:
        issues.append(ValidationIssue("$.datasets", "must be a non-empty array"))
    else:
        datasets = [
            _parse_dataset(value, index, spec_path.parent, issues)
            for index, value in enumerate(datasets_raw)
        ]
    dataset_ids = [dataset.id for dataset in datasets if dataset.id]
    if len(set(dataset_ids)) != len(dataset_ids):
        issues.append(ValidationIssue("$.datasets", "dataset ids must be unique"))
    unknown_dataset_ids = sorted(set(strategy.dataset_ids) - set(dataset_ids))
    if unknown_dataset_ids:
        issues.append(
            ValidationIssue(
                "$.strategies[0].datasets",
                "unknown dataset ids: " + ", ".join(unknown_dataset_ids),
            )
        )
 
    objective = _parse_objective(root.get("objective"), issues)
    sampler = _parse_sampler(root.get("sampler"), issues)
    _validate_candidate_policy(strategy, sampler, issues)
    execution = _parse_execution(root.get("execution"), issues)
 
    if require_files:
        strategy_path = strategy.source or strategy.artifact
        if strategy_path is not None and not strategy_path.is_file():
            issues.append(
                ValidationIssue(
                    "$.strategies[0]", f"file does not exist: {strategy_path}"
                )
            )
        for index, dataset in enumerate(datasets):
            if not dataset.ohlcv.is_file():
                issues.append(
                    ValidationIssue(
                        f"$.datasets[{index}].ohlcv",
                        f"file does not exist: {dataset.ohlcv}",
                    )
                )
 
    if issues:
        raise StudySpecError(issues)
    return StudySpec(
        schema_version=1,
        mode="single_strategy",
        strategy=strategy,
        datasets=tuple(datasets),
        objective=objective,
        sampler=sampler,
        execution=execution,
        spec_path=spec_path,
    )