artifact.py

Go to the documentation of this file.

"""Content-addressed PineScript strategy artifact builder."""
 
from __future__ import annotations
 
from contextlib import AbstractContextManager
import ctypes
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import platform
import shutil
import subprocess
import sys
import tempfile
import time
from typing import Any, Callable, Mapping, Sequence
 
from .transpile import (
    TranspileDiagnostic,
    TranspileFailure,
    codegen_identity,
    transpile_source,
)
 
 
 
CANONICAL_COMPILE_FLAGS = (
    "-std=c++17",
    "-O2",
    "-ffp-contract=off",
    "-fPIC",
    "-shared",
)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
class ArtifactBuildError(RuntimeError):
    """A configuration, transpile, compile, or cache publication failure."""
 
    def __init__(
        self,
        stage: str,
        message: str,
        *,
        command: Sequence[str] = (),
        stdout: str = "",
        stderr: str = "",
        diagnostics: Sequence[TranspileDiagnostic] = (),
    ) -> None:
        self.stage = stage
        self.command = tuple(command)
        self.stdout = stdout
        self.stderr = stderr
        self.diagnostics = tuple(diagnostics)
        super().__init__(message)
 
 
@dataclass(frozen=True)
class StrategyArtifact:
    """Paths and metadata consumed by the native strategy-plugin loader."""
 
    artifact_key: str
    request_key: str
    plugin_path: Path
    generated_cpp_path: Path
    manifest_path: Path
    provenance_path: Path
    cache_hit: bool
    inputs: tuple[dict[str, Any], ...]
    strategy_params: Mapping[str, Any]
 
    @property
    def library_path(self) -> Path:
        """Compatibility alias for callers that call plugins libraries."""
 
        return self.plugin_path
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
class ArtifactBuilder:
    """Transpile and compile a Pine strategy once per compatible request identity."""
 
    def __init__(
        self,
        *,
        engine_root: str | os.PathLike[str],
        cache_dir: str | os.PathLike[str] | None = None,
        compiler: str | os.PathLike[str] | None = None,
        eigen_include: str | os.PathLike[str] | None = None,
        plugin_validator: Callable[[Path, int], Mapping[str, Any]] | None = None,
        lock_timeout_seconds: float = 120.0,
        compile_timeout_seconds: float = 300.0,
    ) -> None:
        self.engine_root = Path(engine_root).expanduser().resolve()
        self.cache_dir = (
            Path(cache_dir).expanduser().resolve()
            if cache_dir
            else _default_cache_dir()
        )
        self.compiler = os.fspath(compiler) if compiler is not None else None
        self.eigen_include = (
            Path(eigen_include).expanduser().resolve()
            if eigen_include is not None
            else None
        )
        self.plugin_validator = plugin_validator or self._validate_plugin
        self.lock_timeout_seconds = lock_timeout_seconds
        self.compile_timeout_seconds = compile_timeout_seconds
 
    def build(self, pine_source: str, *, filename: str = "<input>") -> StrategyArtifact:
        if not isinstance(pine_source, str):
            raise TypeError("pine_source must be a string")
        if self.lock_timeout_seconds <= 0 or self.compile_timeout_seconds <= 0:
            raise ArtifactBuildError("configuration", "build timeouts must be positive")
 
        engine = self._resolve_engine_layout()
        compiler = self._resolve_compiler_identity()
        eigen = self._resolve_eigen_identity(engine)
        try:
            codegen = codegen_identity()
        except ModuleNotFoundError as error:
            raise ArtifactBuildError(
                "configuration",
                "pineforge-codegen is required to compile PineScript; install "
                "pineforge-hpo[transpile] or provide a precompiled artifact",
            ) from error
        output_extension, link_mode = self._platform_link_mode()
        compile_spec = {
            "flags": list(CANONICAL_COMPILE_FLAGS),
            "include_dirs": [
                str(path) for path in (*engine.include_dirs, eigen.include_dir)
            ],
            "link_mode": link_mode,
            "output_extension": output_extension,
        }
        request_identity = {
            "source_sha256": hashlib.sha256(pine_source.encode("utf-8")).hexdigest(),
            "codegen": codegen.to_dict(),
            "engine": engine.identity_dict(),
            "compiler": compiler.to_dict(),
            "eigen": eigen.to_dict(),
            "platform": {
                "system": platform.system(),
                "release": platform.release(),
                "machine": platform.machine(),
                "sys_platform": sys.platform,
                "python_implementation": platform.python_implementation(),
                "python_version": platform.python_version(),
            },
            "compiler_environment": {
                key: os.environ.get(key, "") for key in _COMPILER_ENV_KEYS
            },
            "compile": compile_spec,
        }
        request_key = _json_sha256(request_identity)
 
        hit = self._lookup_request(request_key, request_identity, output_extension)
        if hit is not None:
            return hit
 
        lock_path = self.cache_dir / "locks" / f"{request_key}.lock"
        with _KeyLock(lock_path, self.lock_timeout_seconds):
            hit = self._lookup_request(request_key, request_identity, output_extension)
            if hit is not None:
                return hit
 
            transpile_result = transpile_source(pine_source, filename=filename)
            try:
                generated_cpp = transpile_result.require_success()
            except TranspileFailure as error:
                raise ArtifactBuildError(
                    "transpile",
                    str(error),
                    diagnostics=transpile_result.diagnostics,
                ) from error
 
            generated_cpp_sha256 = hashlib.sha256(
                generated_cpp.encode("utf-8")
            ).hexdigest()
            artifact_identity = {
                "request_identity": request_identity,
                "generated_cpp_sha256": generated_cpp_sha256,
            }
            artifact_key = _json_sha256(artifact_identity)
            existing = self._load_artifact(
                artifact_key=artifact_key,
                request_key=request_key,
                request_identity=request_identity,
                output_extension=output_extension,
                cache_hit=True,
            )
            if existing is not None:
                self._write_request_index(request_key, artifact_key)
                return existing
 
            artifact = self._compile_and_publish(
                artifact_key=artifact_key,
                request_key=request_key,
                request_identity=request_identity,
                output_extension=output_extension,
                generated_cpp=generated_cpp,
                generated_cpp_sha256=generated_cpp_sha256,
                inputs=transpile_result.inputs,
                strategy_params=transpile_result.strategy_params,
                diagnostics=transpile_result.diagnostics,
                source_name=filename,
                compiler=compiler,
                engine=engine,
                eigen=eigen,
            )
            self._write_request_index(request_key, artifact_key)
            return artifact
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
def build_strategy_artifact(
    pine_source: str,
    *,
    engine_root: str | os.PathLike[str],
    cache_dir: str | os.PathLike[str] | None = None,
    compiler: str | os.PathLike[str] | None = None,
    eigen_include: str | os.PathLike[str] | None = None,
    filename: str = "<input>",
) -> StrategyArtifact:
    """Convenience entry point for a one-off CLI artifact build."""
 
    builder = ArtifactBuilder(
        engine_root=engine_root,
        cache_dir=cache_dir,
        compiler=compiler,
        eigen_include=eigen_include,
    )
    return builder.build(pine_source, filename=filename)