study_spec.py

Go to the documentation of this file.

"""Dependency-free StudySpec v1 JSON loader and validator."""
 
from __future__ import annotations
 
from .error import HpoError
 
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(HpoError, 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))
        self.with_failure("hpo_study_spec_invalid", {"reason": "study"})
 
 
@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 | None
    input_tf: str
    script_tf: str
    chart_timezone: str
    ohlcv_bin: Path | None = None
    verify_bin: str = "full"
 
 
@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
    history_switch: int | None = None
    max_threads: int = 0
 
 
@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
    candidates_file: Path | None = None
    scramble: str | 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
    batch_size: int | None = None
    batch_lag: int = 0
    # Schedule selector. "batches" is the default and stays byte-identical to 0.12.0;
    # "rolling" refills a free worker slot at once from whatever has finished.
    schedule: str = "batches"
    pruner: str = "none"
    pruner_rungs: tuple[float, ...] = (0.25, 0.5)
    pruner_eta: int = 2
 
 
@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
    symbol_feeds: Path | Mapping[str, Any] | None = None
 
    @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, continuation: 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}")]
        ).with_failure("hpo_input_file_invalid", {}) 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",
            "symbol_feeds",
        },
        "$",
        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)
 
    symbol_feeds = root.get("symbol_feeds")
    if isinstance(symbol_feeds, str) and symbol_feeds:
        symbol_feeds = (spec_path.parent / Path(symbol_feeds).expanduser()).resolve()
    elif symbol_feeds is not None and not isinstance(symbol_feeds, dict):
        issues.append(
            ValidationIssue(
                "$.symbol_feeds", "must be an index path or inline index object"
            )
        )
 
    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"), spec_path.parent, issues)
    _validate_candidate_policy(strategy, sampler, issues, continuation=continuation)
    execution = _parse_execution(root.get("execution"), issues)
    if sampler.kind == "candidates" and execution.pruner != "none":
        # A pruned row has only a prefix score, so a listed vector could end without a
        # full-window score. The native runner refuses the combination as well.
        issues.append(
            ValidationIssue(
                "$.execution.pruner", "must be none when sampler.kind is candidates"
            )
        )
 
    if (
        sampler.kind == "tpe"
        and execution.batch_lag == 1
        and sampler.config is not None
        and not sampler.config.constant_liar
    ):
        issues.append(
            ValidationIssue(
                "$.execution.batch_lag",
                "lag-one TPE requires sampler.config.constant_liar=true",
            )
        )
 
    if (
        sampler.kind == "tpe"
        and execution.schedule == "rolling"
        and sampler.config is not None
        and not sampler.config.constant_liar
    ):
        # Every rolling proposal is made while other trials are still running, and constant liar
        # is what lets TPE use those outstanding candidates.
        issues.append(
            ValidationIssue(
                "$.execution.schedule",
                "rolling TPE requires sampler.config.constant_liar=true",
            )
        )
 
    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}"
                )
            )
        listed = sampler.candidates_file
        if listed is not None and not listed.is_file():
            # Existence only; the native runner validates the list. The path is not
            # echoed.
            issues.append(
                ValidationIssue(
                    "$.sampler.config.candidates_file",
                    "candidate list file does not exist",
                )
            )
        for index, dataset in enumerate(datasets):
            source = (
                dataset.ohlcv_bin if dataset.ohlcv_bin is not None else dataset.ohlcv
            )
            source_key = "ohlcv_bin" if dataset.ohlcv_bin is not None else "ohlcv"
            if source is not None and not source.is_file():
                issues.append(
                    ValidationIssue(
                        f"$.datasets[{index}].{source_key}",
                        f"file does not exist: {source}",
                    )
                )
 
    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,
        symbol_feeds=symbol_feeds,
    )