diff --git a/.dockerignore b/.dockerignore index 49d24c8..0275dd6 100644 --- a/.dockerignore +++ b/.dockerignore @@ -6,18 +6,3 @@ dist **/__pycache__ **/*.pyc tests -vendor/pineforge-engine/.git -vendor/pineforge-engine/build -vendor/pineforge-engine/benchmarks -vendor/pineforge-engine/corpus -vendor/pineforge-engine/docs -vendor/pineforge-engine/scripts -vendor/pineforge-engine/tests -vendor/pineforge-engine/tutorial -vendor/pineforge-codegen-oss/.git -vendor/pineforge-codegen-oss/.github -vendor/pineforge-codegen-oss/docs -vendor/pineforge-codegen-oss/gate -vendor/pineforge-codegen-oss/npm -vendor/pineforge-codegen-oss/scripts -vendor/pineforge-codegen-oss/tests diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1eb159f..6d3b351 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -33,8 +33,6 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 - with: - submodules: true - uses: actions/setup-python@v6 with: python-version: "3.13" diff --git a/.gitmodules b/.gitmodules deleted file mode 100644 index 2a67fe7..0000000 --- a/.gitmodules +++ /dev/null @@ -1,6 +0,0 @@ -[submodule "vendor/pineforge-codegen-oss"] - path = vendor/pineforge-codegen-oss - url = https://github.com/pineforge-4pass/pineforge-codegen-oss.git -[submodule "vendor/pineforge-engine"] - path = vendor/pineforge-engine - url = https://github.com/pineforge-4pass/pineforge-engine.git diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index e3507ec..a933585 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -4,11 +4,11 @@ Community data providers are the reason this repository exists. Keep the core contracts small and put vendor behavior behind Python provider modules. This repository does not accept a separate C++ provider implementation surface. -Docker and initialized codegen/engine submodules are required for raw-Pine -integration work: +Docker is required for raw-Pine integration work. Engine and codegen are +consumed only through the pinned `pineforge-release` image; do not add source +submodules or duplicate their build logic here. ```bash -git submodule update --init docker version ``` diff --git a/README.md b/README.md index 5bca1c3..4fedbdc 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,7 @@ currently forming candle, and polls public trades into a strictly increasing per-stream sequence. ```bash -pip install -e '.[ccxt]' +pip install 'pineforge-data[ccxt]' ``` ```python @@ -87,30 +87,20 @@ confirmed OHLCV through a data provider and runs this pinned pipeline: ```text raw .pine + provider OHLCV - ↓ read-only mount -Docker: Python codegen → C++ strategy → pineforge-engine → JSON report + ↓ local read-only mount or FastAPI request +pineforge-release → generated C++ → cached/compiled strategy → JSON report ``` Docker is a prerequisite. A host C++ compiler and a precompiled strategy -library are not required. - -Clone with both pinned runtime dependencies: - -```bash -git clone --recurse-submodules https://github.com/pineforge-4pass/pineforge-data.git -cd pineforge-data -python3 -m venv .venv -.venv/bin/pip install -e '.[ccxt]' -``` - -For an existing checkout: +library are not required. Install the package without cloning engine or codegen +repositories: ```bash -git submodule update --init +pip install 'pineforge-data[ccxt]' ``` ```bash -.venv/bin/pineforge-backtest \ +pineforge-backtest \ --pine strategy.pine \ --provider ccxt \ --venue kraken \ @@ -122,38 +112,88 @@ git submodule update --init --pretty ``` -The first invocation builds a local image tagged from the container source and -the exact codegen and engine submodule commits. Later invocations reuse it. -Pass `--rebuild-image` to force a rebuild or `--no-image-build` to require a -prebuilt local image. +The first local invocation pulls an immutable, multi-architecture +`pineforge-release` image pinned by both version and OCI digest. It never builds +engine or codegen locally. Use `--pull-policy never` for offline runs or opt in +to the rolling channel with: + +```bash +pineforge-backtest ... \ + --runtime-image ghcr.io/pineforge-4pass/pineforge-release:latest \ + --pull-policy always +``` + +`latest` is convenient for development but not deterministic. The report +records the resolved image digest and component versions when Docker exposes +them. + Compilation and execution run as a non-root user with networking disabled, all Linux capabilities dropped, a read-only root filesystem, and only a read-only temporary input mount. -The JSON report contains data provenance, processed-bar counts, every closed -trade, all/long/short trade statistics, equity statistics, security-feed -diagnostics, optional trace values, and the complete equity curve. Unix -millisecond timestamps can be used instead of ISO-8601 values. +The JSON report contains provider and market provenance, the release runtime +identity and fingerprint, processed-bar counts, every closed trade, +all/long/short statistics, equity statistics, diagnostics, and the complete +equity curve. Unix millisecond timestamps can be used instead of ISO-8601 +values. The pinned `pineforge-release` does not currently expose trace +collection; `--trace` fails explicitly rather than silently omitting it. Use `--provider-config config.json` for CCXT constructor options and -`--strategy-params inputs.json` for Pine input overrides. The provider config -file may contain credentials, so keep it outside version control. +`--strategy-params inputs.json` for Pine inputs. Use `--strategy-overrides` for +`strategy()` header overrides. The provider config file may contain +credentials, so keep it outside version control. -The report records the Pine source hash, generated C++ hash, transpile and -compile timings, and the exact codegen and engine commits. The OSS codegen is -source-available under its own PolyForm Noncommercial license and supplemental -terms; review `vendor/pineforge-codegen-oss/LICENSE` before distribution or -commercial use. The engine remains Apache-2.0. +The generated C++ hash and exact engine/codegen versions are recorded in the +release fingerprint. The combined runtime and its component licensing are +owned by [`pineforge-release`](https://github.com/pineforge-4pass/pineforge-release), +not vendored into this repository. Provider implementations in this repository are Python-only. The compiled C++ strategy and engine stay behind the Docker/runtime boundary; broker SDKs and provider-specific types do not cross into `pineforge-engine`. +## Concurrent FastAPI server + +The server image derives from the same pinned `pineforge-release` image. It +admits a bounded number of compiler/backtest processes, keeps a bounded queue, +isolates every request in its own temporary directory, and optionally requires +a bearer token. + +```bash +docker build -f docker/server.Dockerfile -t pineforge-data-server . +docker volume create pineforge-compile-cache +docker run --rm -p 127.0.0.1:8000:8000 \ + --read-only \ + --tmpfs /tmp:rw,exec,nosuid,nodev,size=512m \ + --cap-drop ALL \ + --security-opt no-new-privileges \ + --mount type=volume,src=pineforge-compile-cache,dst=/cache \ + -e PINEFORGE_SERVER_API_KEY=change-me \ + pineforge-data-server +``` + +Point the same harness at it without putting the token on the command line: + +```bash +export PINEFORGE_SERVER_URL=http://127.0.0.1:8000 +export PINEFORGE_SERVER_API_KEY=change-me +pineforge-backtest --pine strategy.pine --venue kraken --symbol BTC/USD \ + --timeframe 15m --start 2026-07-01T00:00:00Z --end 2026-07-08T00:00:00Z +``` + +The server always transpiles Pine deterministically and hashes the generated +C++. Its cache stores the compiled `.so` under a key containing that C++ hash +plus the release, engine, architecture, and compile flags. Concurrent misses +for the same key compile once; subsequent requests skip compilation. Cache +hit/key/hash are included in response provenance. See +[docs/server.md](docs/server.md) for endpoints, limits, deployment, and cache +settings. + ## Development ```bash python3 -m venv .venv -.venv/bin/pip install -e '.[dev,ccxt]' +.venv/bin/pip install -e '.[dev,ccxt,server]' .venv/bin/ruff check . .venv/bin/mypy src .venv/bin/pytest diff --git a/docker/Dockerfile b/docker/Dockerfile deleted file mode 100644 index 719d8d1..0000000 --- a/docker/Dockerfile +++ /dev/null @@ -1,63 +0,0 @@ -# syntax=docker/dockerfile:1@sha256:87999aa3d42bdc6bea60565083ee17e86d1f3339802f543c0d03998580f9cb89 - -FROM debian:bookworm-slim@sha256:60eac759739651111db372c07be67863818726f754804b8707c90979bda511df AS engine-builder - -RUN apt-get update && apt-get install -y --no-install-recommends \ - cmake \ - g++ \ - libeigen3-dev \ - ninja-build \ - && rm -rf /var/lib/apt/lists/* - -COPY vendor/pineforge-engine /src/pineforge-engine - -RUN cmake -G Ninja -S /src/pineforge-engine -B /tmp/pineforge-build \ - -DCMAKE_BUILD_TYPE=Release \ - -DCMAKE_INSTALL_PREFIX=/opt/pineforge \ - -DPINEFORGE_BUILD_TESTS=OFF \ - -DPINEFORGE_BUILD_TUTORIAL=OFF \ - -DPINEFORGE_BUILD_CORPUS_STRATEGIES=OFF \ - && cmake --build /tmp/pineforge-build --target pineforge -j2 \ - && cmake --install /tmp/pineforge-build - -FROM python:3.13-slim-bookworm@sha256:fcbd8dfc2605ba7c2eca646846c5e892b2931e41f6227985154a596f26ab8ed7 - -ARG ENGINE_SHA=unknown -ARG CODEGEN_SHA=unknown -ARG DATA_SOURCE_DIGEST=unknown - -LABEL org.opencontainers.image.title="pineforge-data-backtest" \ - org.opencontainers.image.description="Raw PineScript and provider OHLCV to PineForge report" \ - org.opencontainers.image.source="https://github.com/pineforge-4pass/pineforge-data" \ - org.opencontainers.image.licenses="Apache-2.0 AND LicenseRef-PolyForm-Noncommercial-1.0.0" \ - io.pineforge.engine.commit="${ENGINE_SHA}" \ - io.pineforge.codegen.commit="${CODEGEN_SHA}" \ - io.pineforge.data.source-digest="${DATA_SOURCE_DIGEST}" - -RUN apt-get update && apt-get install -y --no-install-recommends \ - g++ \ - libeigen3-dev \ - && rm -rf /var/lib/apt/lists/* /usr/share/doc /usr/share/man - -COPY --from=engine-builder /opt/pineforge /opt/pineforge -COPY vendor/pineforge-codegen-oss/pineforge_codegen /opt/pineforge-codegen/pineforge_codegen -COPY vendor/pineforge-codegen-oss/VERSION /opt/pineforge-codegen/VERSION -COPY src/pineforge_data /opt/pineforge-data/pineforge_data -COPY docker/entrypoint.py /opt/pineforge/bin/pineforge-data-backtest -COPY LICENSE /opt/licenses/pineforge-data.LICENSE -COPY vendor/pineforge-engine/LICENSE /opt/licenses/pineforge-engine.LICENSE -COPY vendor/pineforge-codegen-oss/LICENSE /opt/licenses/pineforge-codegen.LICENSE - -RUN printf '%s\n' "${ENGINE_SHA}" > /opt/pineforge/ENGINE_COMMIT \ - && printf '%s\n' "${CODEGEN_SHA}" > /opt/pineforge/CODEGEN_COMMIT \ - && printf '%s\n' "${DATA_SOURCE_DIGEST}" > /opt/pineforge/DATA_SOURCE_DIGEST \ - && chmod +x /opt/pineforge/bin/pineforge-data-backtest \ - && useradd --system --uid 10001 --shell /usr/sbin/nologin pineforge - -ENV PYTHONPATH=/opt/pineforge-data:/opt/pineforge-codegen \ - PINEFORGE_PREFIX=/opt/pineforge \ - PYTHONDONTWRITEBYTECODE=1 \ - PYTHONUNBUFFERED=1 - -USER 10001 -ENTRYPOINT ["/opt/pineforge/bin/pineforge-data-backtest"] diff --git a/docker/entrypoint.py b/docker/entrypoint.py deleted file mode 100644 index 5d9b8fd..0000000 --- a/docker/entrypoint.py +++ /dev/null @@ -1,170 +0,0 @@ -#!/usr/bin/env python3 -"""Container entrypoint: raw Pine + normalized OHLCV -> JSON backtest report.""" - -from __future__ import annotations - -import csv -import hashlib -import json -import os -import subprocess -import sys -import tempfile -import time -from pathlib import Path -from typing import cast - -from pineforge_codegen import transpile -from pineforge_codegen.errors import CompileError - -from pineforge_data import ( - BacktestOptions, - Bar, - Instrument, - MagnifierDistribution, - PineForgeBacktestRunner, -) -from pineforge_data.backtest import JsonValue - -WORK = Path("/work") -PREFIX = Path(os.environ.get("PINEFORGE_PREFIX", "/opt/pineforge")) - - -def _load_request() -> dict[str, object]: - value = json.loads((WORK / "request.json").read_text(encoding="utf-8")) - if not isinstance(value, dict): - raise ValueError("request.json must contain an object") - return cast(dict[str, object], value) - - -def _load_bars(path: Path, instrument: Instrument, source: str) -> list[Bar]: - bars: list[Bar] = [] - with path.open(newline="", encoding="utf-8") as handle: - for row in csv.DictReader(handle): - bars.append( - Bar( - instrument, - int(row["timestamp"]), - float(row["open"]), - float(row["high"]), - float(row["low"]), - float(row["close"]), - float(row["volume"]), - source, - ) - ) - return bars - - -def _read_commit(name: str) -> str: - return (PREFIX / name).read_text(encoding="utf-8").strip() - - -def _compile(generated: Path, strategy_library: Path) -> float: - command = [ - "g++", - "-std=c++17", - "-O2", - "-ffp-contract=off", - "-fPIC", - "-shared", - f"-I{PREFIX / 'include'}", - "-I/usr/include/eigen3", - str(generated), - "-Wl,--whole-archive", - str(PREFIX / "lib/libpineforge.a"), - "-Wl,--no-whole-archive", - "-o", - str(strategy_library), - ] - started = time.perf_counter() - completed = subprocess.run(command, text=True, capture_output=True, check=False) - elapsed = time.perf_counter() - started - if completed.returncode != 0: - detail = completed.stderr.strip() or completed.stdout.strip() or "unknown compiler error" - raise RuntimeError(f"strategy compilation failed:\n{detail}") - return elapsed - - -def main() -> int: - pine_path = WORK / "strategy.pine" - ohlcv_path = WORK / "ohlcv.csv" - if not pine_path.is_file() or not ohlcv_path.is_file(): - print( - "error: /work must contain strategy.pine, ohlcv.csv, and request.json", - file=sys.stderr, - ) - return 2 - - try: - request = _load_request() - instrument_data = cast(dict[str, object], request["instrument"]) - options_data = cast(dict[str, object], request["options"]) - instrument = Instrument( - symbol=str(instrument_data["symbol"]), - venue=str(instrument_data.get("venue", "")), - timezone=str(instrument_data.get("timezone", "UTC")), - session=str(instrument_data.get("session", "24x7")), - volume_unit=str(instrument_data.get("volume_unit", "base")), - ) - bars = _load_bars(ohlcv_path, instrument, str(request.get("source", "provider"))) - options = BacktestOptions( - input_timeframe=str(options_data.get("input_timeframe", "")), - script_timeframe=str(options_data.get("script_timeframe", "")), - bar_magnifier=bool(options_data.get("bar_magnifier", False)), - magnifier_samples=int(options_data.get("magnifier_samples", 4)), - magnifier_distribution=MagnifierDistribution( - int(options_data.get("magnifier_distribution", 3)) - ), - trace_enabled=bool(options_data.get("trace_enabled", False)), - chart_timezone=cast(str | None, options_data.get("chart_timezone")), - ) - strategy_params = cast(dict[str, JsonValue], request.get("strategy_params", {})) - pine_source = pine_path.read_text(encoding="utf-8") - - with tempfile.TemporaryDirectory(prefix="pineforge-") as temporary: - temp = Path(temporary) - generated = temp / "strategy.cpp" - strategy_library = temp / "strategy.so" - transpile_started = time.perf_counter() - generated_source = transpile(pine_source, filename=pine_path.name) - transpile_seconds = time.perf_counter() - transpile_started - generated.write_text(generated_source, encoding="utf-8") - compile_seconds = _compile(generated, strategy_library) - report = PineForgeBacktestRunner.load(strategy_library).run( - bars, - instrument=instrument, - options=options, - strategy_params=strategy_params, - ) - - output: dict[str, object] = { - "runtime": { - "data_source_digest": _read_commit("DATA_SOURCE_DIGEST"), - "engine_commit": _read_commit("ENGINE_COMMIT"), - "codegen_commit": _read_commit("CODEGEN_COMMIT"), - "codegen_version": ( - Path("/opt/pineforge-codegen/VERSION").read_text(encoding="utf-8").strip() - ), - }, - "transpile": { - "seconds": round(transpile_seconds, 6), - "pine_sha256": hashlib.sha256(pine_source.encode()).hexdigest(), - "generated_cpp_sha256": hashlib.sha256(generated_source.encode()).hexdigest(), - "generated_cpp_bytes": len(generated_source.encode()), - }, - "compile": {"seconds": round(compile_seconds, 6)}, - "backtest": report.to_dict(), - } - print(json.dumps(output, separators=(",", ":"), allow_nan=False)) - return 0 - except CompileError as exc: - print(f"error: PineScript transpilation failed: {exc}", file=sys.stderr) - return 5 - except Exception as exc: - print(f"error: {exc}", file=sys.stderr) - return 4 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/docker/server.Dockerfile b/docker/server.Dockerfile new file mode 100644 index 0000000..313f6b1 --- /dev/null +++ b/docker/server.Dockerfile @@ -0,0 +1,43 @@ +# syntax=docker/dockerfile:1@sha256:87999aa3d42bdc6bea60565083ee17e86d1f3339802f543c0d03998580f9cb89 + +ARG PINEFORGE_RELEASE_IMAGE=ghcr.io/pineforge-4pass/pineforge-release:0.1.12@sha256:312b9d908390b828484617472c749d5815feb75507da87eae2f6902cfe3d47b1 + +FROM python:3.11-slim-bookworm@sha256:f5cf0344c9886ff24d34797578d5d7dd6e8911ae0fe5962bb55d0f89603ec361 AS server-builder +RUN python -m pip install --no-cache-dir --target /opt/pineforge-server-deps \ + 'fastapi>=0.139,<0.140' \ + 'uvicorn>=0.51,<0.52' + +FROM ${PINEFORGE_RELEASE_IMAGE} +ARG PINEFORGE_RELEASE_IMAGE + +USER root +COPY --from=server-builder /opt/pineforge-server-deps /opt/pineforge-server-deps +COPY src/pineforge_data /opt/pineforge-server/pineforge_data +RUN mkdir -p /cache \ + && chown 10001:10001 /cache + +ENV PYTHONPATH=/opt/pineforge-server:/opt/pineforge-server-deps:/opt/pineforge/pycodegen \ + PINEFORGE_RELEASE_IMAGE=${PINEFORGE_RELEASE_IMAGE} \ + PINEFORGE_RELEASE_ENTRYPOINT=/opt/pineforge/bin/entrypoint.sh \ + PINEFORGE_RELEASE_RUN_JSON=/opt/pineforge/bin/run_json.py \ + PINEFORGE_SERVER_CACHE_DIR=/cache \ + PINEFORGE_SERVER_CONCURRENCY=2 \ + PINEFORGE_SERVER_MAX_QUEUE=8 \ + PINEFORGE_SERVER_QUEUE_TIMEOUT=30 \ + PINEFORGE_SERVER_EXECUTION_TIMEOUT=300 \ + PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +LABEL org.opencontainers.image.title="pineforge-data-backtest-server" \ + org.opencontainers.image.description="Concurrent FastAPI service backed by pineforge-release" \ + org.opencontainers.image.source="https://github.com/pineforge-4pass/pineforge-data" \ + org.opencontainers.image.licenses="Apache-2.0" + +VOLUME ["/cache"] +EXPOSE 8000 +HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \ + CMD ["python3", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/readyz', timeout=2).read()"] + +USER 10001 +ENTRYPOINT ["python3", "-m", "uvicorn"] +CMD ["pineforge_data.server:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"] diff --git a/docs/server.md b/docs/server.md new file mode 100644 index 0000000..77a620c --- /dev/null +++ b/docs/server.md @@ -0,0 +1,81 @@ +# FastAPI backtest server + +The server is a thin concurrent control plane layered on the published +`pineforge-release` image. Market data remains a host-side provider concern; +the harness sends normalized OHLCV, raw PineScript, syminfo, runtime options, +and strategy inputs to `POST /v1/backtests`. + +## Endpoints + +- `GET /healthz` — process liveness. +- `GET /readyz` — release tools, concurrency counters, and cache statistics. +- `POST /v1/backtests` — synchronous backtest request; concurrent HTTP requests + execute independently up to the configured process limit. +- `GET /docs` — generated OpenAPI documentation. + +Clients may supply `X-Request-ID`; otherwise the server creates one. When +`PINEFORGE_SERVER_API_KEY` is set, the backtest endpoint requires +`Authorization: Bearer `. Health endpoints intentionally remain +unauthenticated for container orchestration. + +## Concurrency and overload behavior + +Run one Uvicorn worker per container. The service owns a process-wide semaphore; +additional Uvicorn workers would multiply the configured compiler limit. +Scale horizontally with multiple containers when more capacity is required. + +| Environment variable | Default | Meaning | +|---|---:|---| +| `PINEFORGE_SERVER_CONCURRENCY` | `2` in the image | simultaneous compile/backtest processes | +| `PINEFORGE_SERVER_MAX_QUEUE` | `8` in the image | admitted requests waiting for a slot | +| `PINEFORGE_SERVER_QUEUE_TIMEOUT` | `30` | seconds before a queued request returns 503 | +| `PINEFORGE_SERVER_EXECUTION_TIMEOUT` | `300` | total transpile, compile, and backtest deadline | + +When running and queued capacity is full, new requests receive HTTP 429. A +transpile, compile, or engine failure returns HTTP 422 with a structured phase, +code, message, and request ID. Execution timeouts return HTTP 504. + +## Compile cache + +Pine source is transpiled for each request. The server hashes the generated C++ +and combines that digest with the release image, engine/codegen versions, +architecture, and parity-critical compiler flags. The resulting key identifies +the compiled `.so` artifact. + +This deliberately does not key compiled artifacts by Pine source: codegen +changes can alter generated C++ for identical Pine, while different Pine inputs +can theoretically produce the same translation unit. Writes are atomic and +concurrent misses for one key are deduplicated. + +| Environment variable | Default | Meaning | +|---|---:|---| +| `PINEFORGE_SERVER_CACHE_DIR` | `/cache` | compiled artifact directory | +| `PINEFORGE_SERVER_CACHE_MAX_ENTRIES` | `1024` | maximum retained `.so` files | +| `PINEFORGE_SERVER_CACHE_MAX_BYTES` | `2147483648` | maximum retained bytes | + +Mount `/cache` as a named volume to preserve compiled strategies across server +restarts. The cache metadata returned with each result includes `key`, `hit`, +and `generated_cpp_sha256`. + +## Runtime channel + +The Dockerfile defaults to a semver-and-digest-pinned release image. Override it +at build time for a rolling development server: + +```bash +docker build -f docker/server.Dockerfile \ + --build-arg PINEFORGE_RELEASE_IMAGE=ghcr.io/pineforge-4pass/pineforge-release:latest \ + -t pineforge-data-server:latest . +``` + +Do not use the rolling channel for reproducibility-sensitive production runs. +The cache key includes component/release identity, so a runtime upgrade cannot +reuse a compiled artifact from a different engine version. + +## Deployment boundary + +The server accepts CPU- and memory-intensive work. Bind it to a private network, +set an API key, enforce request-size and rate limits at the ingress, and do not +expose it directly to the public internet. The container runs as UID 10001; +deploy it with a read-only root filesystem, an executable `/tmp` tmpfs, dropped +capabilities, `no-new-privileges`, and a writable cache volume. diff --git a/pyproject.toml b/pyproject.toml index 7abea11..c9a4e9b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,14 +24,22 @@ dependencies = [] [project.scripts] pineforge-backtest = "pineforge_data.cli.backtest:main" +pineforge-backtest-server = "pineforge_data.server:main" [project.optional-dependencies] ccxt = ["ccxt>=4.5.64,<5"] +server = [ + "fastapi>=0.139,<0.140", + "uvicorn>=0.51,<0.52", +] dev = [ "build>=1.2", + "fastapi>=0.139,<0.140", + "httpx2>=2.5,<2.6", "mypy>=1.13", "pytest>=8.3", "ruff>=0.8", + "uvicorn>=0.51,<0.52", ] [project.urls] diff --git a/src/pineforge_data/__init__.py b/src/pineforge_data/__init__.py index c3821a5..18c9c77 100644 --- a/src/pineforge_data/__init__.py +++ b/src/pineforge_data/__init__.py @@ -7,11 +7,11 @@ MagnifierDistribution, PineForgeBacktestRunner, ) +from .compile_cache import CompileCache from .docker_runtime import ( DockerBacktestRuntime, DockerExecutionError, DockerPrerequisiteError, - discover_repository_root, ) from .engine import EngineStreamSink, PfBar, PfTradeTick, pack_bars, pack_trade_ticks from .errors import EngineStreamError @@ -45,12 +45,16 @@ create_provider, default_registry, ) +from .release_contract import DEFAULT_RELEASE_IMAGE, ReleaseContractError from .requests import BarRequest, MacroRequest, MarketQuery, TradeSubscription +from .server_client import BacktestServerError, FastApiBacktestClient __all__ = [ + "DEFAULT_RELEASE_IMAGE", "AssetClass", "BacktestOptions", "BacktestReport", + "BacktestServerError", "Bar", "BarRequest", "CcxtCapabilityError", @@ -58,6 +62,7 @@ "CcxtDependencyError", "CcxtError", "CcxtProvider", + "CompileCache", "ContractSpec", "DockerBacktestRuntime", "DockerExecutionError", @@ -65,6 +70,7 @@ "EngineBacktestError", "EngineStreamError", "EngineStreamSink", + "FastApiBacktestClient", "HistoricalBarProvider", "Instrument", "LiveTradeProvider", @@ -86,11 +92,11 @@ "ProviderNotFoundError", "ProviderRegistry", "ProviderRegistryError", + "ReleaseContractError", "TradeSubscription", "TradeTick", "create_provider", "default_registry", - "discover_repository_root", "pack_bars", "pack_trade_ticks", ] diff --git a/src/pineforge_data/cli/backtest.py b/src/pineforge_data/cli/backtest.py index e6431e5..a3858e6 100644 --- a/src/pineforge_data/cli/backtest.py +++ b/src/pineforge_data/cli/backtest.py @@ -5,6 +5,7 @@ import argparse import asyncio import json +import os import re import sys from dataclasses import replace @@ -13,10 +14,12 @@ from typing import cast from ..backtest import BacktestOptions, JsonValue -from ..docker_runtime import DockerBacktestRuntime, discover_repository_root +from ..docker_runtime import DockerBacktestRuntime from ..models import MarketListing from ..providers import create_provider +from ..release_contract import DEFAULT_RELEASE_IMAGE from ..requests import BarRequest +from ..server_client import FastApiBacktestClient def parse_timestamp(value: str) -> int: @@ -129,16 +132,39 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--script-timeframe", default="") parser.add_argument("--provider-config", type=Path, help="provider options JSON file") parser.add_argument("--strategy-params", type=Path, help="strategy parameters JSON file") + parser.add_argument( + "--strategy-overrides", type=Path, help="strategy() header overrides JSON file" + ) parser.add_argument("--bar-magnifier", action="store_true") parser.add_argument("--magnifier-samples", type=int, default=4) parser.add_argument("--trace", action="store_true") - parser.add_argument("--image", help="Docker image override; default tag follows submodule pins") - parser.add_argument("--repository-root", type=Path, help="pineforge-data checkout path") - parser.add_argument("--rebuild-image", action="store_true") parser.add_argument( - "--no-image-build", - action="store_true", - help="fail instead of building when the pinned image is absent", + "--runtime-image", + "--image", + dest="runtime_image", + default=DEFAULT_RELEASE_IMAGE, + help="pineforge-release image; defaults to the package's pinned digest", + ) + parser.add_argument( + "--pull-policy", + choices=("always", "missing", "never"), + default="missing", + help="local release-image pull policy", + ) + parser.add_argument( + "--execution-timeout", + type=float, + default=300.0, + help="local or server request timeout in seconds", + ) + parser.add_argument( + "--server-url", + help="FastAPI base URL; also read from PINEFORGE_SERVER_URL", + ) + parser.add_argument( + "--server-api-key-env", + default="PINEFORGE_SERVER_API_KEY", + help="environment variable containing the server bearer token", ) parser.add_argument("--output", type=Path, help="report path; defaults to stdout") parser.add_argument("--pretty", action="store_true", help="pretty-print report JSON") @@ -150,6 +176,7 @@ async def run_harness(args: argparse.Namespace) -> dict[str, JsonValue]: provider_config = _load_json_object(args.provider_config) strategy_params = _load_json_object(args.strategy_params) + strategy_overrides = _load_json_object(args.strategy_overrides) provider = create_provider(args.provider, args.venue, config=provider_config) try: listing = await provider.resolve_market(args.symbol) @@ -184,25 +211,47 @@ async def run_harness(args: argparse.Namespace) -> dict[str, JsonValue]: pine_path = args.pine.expanduser().resolve() if not pine_path.is_file(): raise FileNotFoundError(f"PineScript file not found: {pine_path}") - repository_root = discover_repository_root(args.repository_root) - runtime = DockerBacktestRuntime( - repository_root, - image=args.image, - rebuild=args.rebuild_image, - build_if_missing=not args.no_image_build, - ) - container = runtime.run( - pine_path.read_text(encoding="utf-8"), - bars, - instrument=instrument, - source=provider_name, - options=options, - strategy_params=strategy_params, - ) - required_sections = ("runtime", "transpile", "compile", "backtest") + pine_source = pine_path.read_text(encoding="utf-8") + server_url = args.server_url or os.environ.get("PINEFORGE_SERVER_URL") + if server_url: + api_key = os.environ.get(args.server_api_key_env) if args.server_api_key_env else None + client = FastApiBacktestClient( + server_url, + timeout_seconds=args.execution_timeout + 30, + api_key=api_key, + ) + container = await asyncio.to_thread( + client.run, + pine_source, + bars, + instrument=instrument, + source=provider_name, + options=options, + strategy_params=strategy_params, + strategy_overrides=strategy_overrides, + ) + else: + runtime = DockerBacktestRuntime( + image=args.runtime_image, + pull_policy=args.pull_policy, + timeout_seconds=args.execution_timeout, + ) + container = await asyncio.to_thread( + runtime.run, + pine_source, + bars, + instrument=instrument, + source=provider_name, + options=options, + strategy_params=strategy_params, + strategy_overrides=strategy_overrides, + ) + required_sections = ("runtime", "backtest") if any(section not in container for section in required_sections): raise RuntimeError("Docker report is missing a required section") return { + "schema_version": 1, + "request_id": cast(JsonValue, container.get("request_id")), "provider": { "name": provider_name, "adapter": args.provider, @@ -223,8 +272,6 @@ async def run_harness(args: argparse.Namespace) -> dict[str, JsonValue]: "script_timeframe": options.script_timeframe, }, "runtime": cast(JsonValue, container["runtime"]), - "transpile": cast(JsonValue, container["transpile"]), - "compile": cast(JsonValue, container["compile"]), "backtest": cast(JsonValue, container["backtest"]), } diff --git a/src/pineforge_data/compile_cache.py b/src/pineforge_data/compile_cache.py new file mode 100644 index 0000000..682237c --- /dev/null +++ b/src/pineforge_data/compile_cache.py @@ -0,0 +1,171 @@ +"""Concurrency-safe compiled-strategy cache keyed by generated C++.""" + +from __future__ import annotations + +import asyncio +import os +from collections import defaultdict +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from dataclasses import dataclass +from pathlib import Path +from uuid import uuid4 + + +@dataclass(slots=True) +class _LockState: + lock: asyncio.Lock + references: int = 0 + + +class CompileCache: + """Store `.so` artifacts atomically and deduplicate concurrent compiles.""" + + def __init__(self, root: Path, *, max_entries: int, max_bytes: int) -> None: + if max_entries <= 0 or max_bytes <= 0: + raise ValueError("cache limits must be positive") + self.root = root + self.max_entries = max_entries + self.max_bytes = max_bytes + self.root.mkdir(parents=True, exist_ok=True) + self._lock_guard = asyncio.Lock() + self._files_guard = asyncio.Lock() + self._locks: dict[str, _LockState] = {} + self._active: defaultdict[str, int] = defaultdict(int) + self.hits = 0 + self.misses = 0 + self.compiles = 0 + + def path(self, key: str) -> Path: + return self.root / f"{key}.so" + + def temporary_path(self, key: str) -> Path: + return self.root / f".{key}.{uuid4().hex}.tmp" + + @asynccontextmanager + async def compile_lock(self, key: str) -> AsyncIterator[None]: + async with self._lock_guard: + state = self._locks.get(key) + if state is None: + state = _LockState(asyncio.Lock()) + self._locks[key] = state + state.references += 1 + await state.lock.acquire() + try: + yield + finally: + state.lock.release() + async with self._lock_guard: + state.references -= 1 + if state.references == 0: + self._locks.pop(key, None) + + @asynccontextmanager + async def use(self, key: str) -> AsyncIterator[Path]: + path = await self.acquire(key) + if path is None: + raise FileNotFoundError(self.path(key)) + try: + yield path + finally: + await self.release(key) + + async def lookup(self, key: str) -> Path | None: + path = self.path(key) + exists = await asyncio.to_thread(path.is_file) + async with self._files_guard: + if exists: + self.hits += 1 + else: + self.misses += 1 + return path if exists else None + + async def commit(self, key: str, temporary: Path) -> Path: + target = self.path(key) + await asyncio.to_thread(os.replace, temporary, target) + await asyncio.to_thread(target.chmod, 0o555) + async with self._files_guard: + self.compiles += 1 + return target + + async def acquire(self, key: str) -> Path | None: + """Atomically look up and reserve an artifact against eviction.""" + + async with self._files_guard: + path = self.path(key) + if not path.is_file(): + self.misses += 1 + return None + self.hits += 1 + self._active[key] += 1 + await asyncio.to_thread(os.utime, path, None) + return path + + async def commit_and_acquire(self, key: str, temporary: Path) -> Path: + """Atomically publish and reserve a newly compiled artifact.""" + + async with self._files_guard: + target = self.path(key) + await asyncio.to_thread(os.replace, temporary, target) + await asyncio.to_thread(target.chmod, 0o555) + self.compiles += 1 + self._active[key] += 1 + return target + + async def release(self, key: str) -> None: + async with self._files_guard: + self._active[key] -= 1 + if self._active[key] == 0: + self._active.pop(key, None) + + async def status(self) -> dict[str, int | str]: + async with self._files_guard: + files, sizes = await asyncio.to_thread(self._snapshot_sync) + return { + "directory": str(self.root), + "entries": len(files), + "bytes": sizes, + "hits": self.hits, + "misses": self.misses, + "compiles": self.compiles, + } + + async def trim(self) -> None: + async with self._files_guard: + active = set(self._active) + await asyncio.to_thread(self._trim_sync, active) + + def _snapshot_sync(self) -> tuple[list[Path], int]: + files: list[Path] = [] + size = 0 + for path in self.root.glob("*.so"): + try: + stat = path.stat() + except FileNotFoundError: + continue + files.append(path) + size += stat.st_size + return files, size + + def _trim_sync(self, active: set[str]) -> None: + entries = [] + for path in self.root.glob("*.so"): + try: + stat = path.stat() + except FileNotFoundError: + continue + entries.append((stat.st_mtime_ns, stat.st_size, path)) + entries.sort() + total = sum(size for _, size, _ in entries) + count = len(entries) + for _, size, path in entries: + if count <= self.max_entries and total <= self.max_bytes: + break + if path.stem in active: + continue + try: + path.unlink() + except FileNotFoundError: + continue + count -= 1 + total -= size diff --git a/src/pineforge_data/docker_runtime.py b/src/pineforge_data/docker_runtime.py index b90025b..dd0961e 100644 --- a/src/pineforge_data/docker_runtime.py +++ b/src/pineforge_data/docker_runtime.py @@ -1,9 +1,7 @@ -"""Docker boundary for raw PineScript compilation and engine execution.""" +"""Run backtests through the published pineforge-release container.""" from __future__ import annotations -import csv -import hashlib import json import os import shutil @@ -12,33 +10,28 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass from pathlib import Path +from typing import Literal from .backtest import BacktestOptions, JsonValue from .models import Bar, Instrument +from .release_contract import ( + DEFAULT_RELEASE_IMAGE, + ReleaseContractError, + parse_release_report, + release_environment, + release_response, + write_release_inputs, +) + +PullPolicy = Literal["always", "missing", "never"] class DockerPrerequisiteError(RuntimeError): - """Docker or an initialized source submodule is unavailable.""" + """Docker or the configured release image is unavailable.""" class DockerExecutionError(RuntimeError): - """The isolated transpile, compile, or backtest command failed.""" - - -def discover_repository_root(start: Path | None = None) -> Path: - """Find a checkout containing the Dockerfile and both pinned submodules.""" - - configured = os.environ.get("PINEFORGE_DATA_ROOT") - origin = Path(configured).expanduser() if configured else start - if origin is None: - origin = Path(__file__).resolve() - candidates = [origin, *origin.parents] - for candidate in candidates: - if (candidate / "docker/Dockerfile").is_file() and (candidate / ".gitmodules").is_file(): - return candidate.resolve() - raise DockerPrerequisiteError( - "pineforge-data checkout not found; set PINEFORGE_DATA_ROOT to a cloned repository" - ) + """The isolated release-container backtest failed.""" def _completed_error(completed: subprocess.CompletedProcess[str]) -> str: @@ -47,57 +40,22 @@ def _completed_error(completed: subprocess.CompletedProcess[str]) -> str: @dataclass(slots=True) class DockerBacktestRuntime: - """Build and execute the pinned raw-Pine backtest image.""" - - repository_root: Path - image: str | None = None - rebuild: bool = False - build_if_missing: bool = True - - def _submodule_commit(self, relative_path: str) -> str: - path = self.repository_root / relative_path - if not (path / ".git").exists() or not any(path.iterdir()): - raise DockerPrerequisiteError( - f"missing submodule {relative_path}; run `git submodule update --init`" - ) - completed = subprocess.run( - ["git", "-C", str(path), "rev-parse", "HEAD"], - text=True, - capture_output=True, - check=False, - ) - if completed.returncode != 0: - raise DockerPrerequisiteError(_completed_error(completed)) - return completed.stdout.strip() + """Pull and execute an immutable pineforge-release image.""" - def _pins(self) -> tuple[str, str]: - engine = self._submodule_commit("vendor/pineforge-engine") - codegen = self._submodule_commit("vendor/pineforge-codegen-oss") - return engine, codegen + image: str = DEFAULT_RELEASE_IMAGE + pull_policy: PullPolicy = "missing" + timeout_seconds: float = 300.0 - def _source_digest(self) -> str: - digest = hashlib.sha256() - paths = [ - self.repository_root / "docker/Dockerfile", - self.repository_root / "docker/entrypoint.py", - *sorted((self.repository_root / "src/pineforge_data").rglob("*.py")), - ] - for path in paths: - relative = path.relative_to(self.repository_root).as_posix() - digest.update(relative.encode()) - digest.update(b"\0") - digest.update(path.read_bytes()) - digest.update(b"\0") - return digest.hexdigest() + def __post_init__(self) -> None: + if not self.image.strip(): + raise ValueError("image must not be empty") + if self.pull_policy not in ("always", "missing", "never"): + raise ValueError(f"invalid pull policy: {self.pull_policy}") + if self.timeout_seconds <= 0: + raise ValueError("timeout_seconds must be positive") def resolved_image(self) -> str: - """Return the explicit image or a tag derived from both submodule pins.""" - - if self.image: - return self.image - engine, codegen = self._pins() - source = self._source_digest() - return f"pineforge-data-backtest:d{source[:12]}-e{engine[:12]}-c{codegen[:12]}" + return self.image def _check_docker(self) -> None: if shutil.which("docker") is None: @@ -115,9 +73,9 @@ def _check_docker(self) -> None: f"Docker daemon is unavailable: {_completed_error(completed)}" ) - def _image_exists(self, image: str) -> bool: + def _image_exists(self) -> bool: completed = subprocess.run( - ["docker", "image", "inspect", image], + ["docker", "image", "inspect", self.image], text=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, @@ -125,38 +83,69 @@ def _image_exists(self, image: str) -> bool: ) return completed.returncode == 0 + def _pull_image(self) -> None: + completed = subprocess.run( + ["docker", "pull", self.image], + text=True, + capture_output=True, + check=False, + ) + if completed.returncode != 0: + raise DockerPrerequisiteError( + f"failed to pull pineforge-release image: {_completed_error(completed)}" + ) + + def _image_identity(self) -> dict[str, object]: + completed = subprocess.run( + ["docker", "image", "inspect", self.image], + text=True, + capture_output=True, + check=False, + ) + if completed.returncode != 0: + return {} + try: + values = json.loads(completed.stdout) + value = values[0] + labels = value.get("Config", {}).get("Labels", {}) + digests = value.get("RepoDigests", []) + except (IndexError, AttributeError, json.JSONDecodeError, TypeError): + return {} + if not isinstance(labels, dict) or not isinstance(digests, list): + return {} + return { + "resolved_digest": str(digests[0]) if digests else None, + "release_version": labels.get("org.opencontainers.image.version"), + "engine_version": labels.get("io.pineforge.engine.version"), + "codegen_version": labels.get("io.pineforge.codegen.version"), + } + def ensure_image(self) -> str: - """Check Docker and build the pin-addressed image when necessary.""" + """Apply the configured pull policy and return the immutable image ref.""" self._check_docker() - engine, codegen = self._pins() - source = self._source_digest() - image = self.resolved_image() - if not self.rebuild and self._image_exists(image): - return image - if not self.build_if_missing: - raise DockerPrerequisiteError(f"Docker image is not available locally: {image}") - command = [ - "docker", - "build", - "--file", - str(self.repository_root / "docker/Dockerfile"), - "--tag", - image, - "--build-arg", - f"ENGINE_SHA={engine}", - "--build-arg", - f"CODEGEN_SHA={codegen}", - "--build-arg", - f"DATA_SOURCE_DIGEST={source}", - str(self.repository_root), - ] - completed = subprocess.run(command, text=True, check=False) - if completed.returncode != 0: - raise DockerExecutionError( - f"Docker image build failed with exit {completed.returncode}" + exists = self._image_exists() + if self.pull_policy == "always" or (self.pull_policy == "missing" and not exists): + self._pull_image() + exists = True + if not exists: + raise DockerPrerequisiteError( + f"pineforge-release image is not available locally: {self.image}" + ) + return self.image + + @staticmethod + def _remove_timed_out_container(cid_path: Path) -> None: + if not cid_path.is_file(): + return + container_id = cid_path.read_text(encoding="utf-8").strip() + if container_id: + subprocess.run( + ["docker", "rm", "--force", container_id], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, ) - return image def run( self, @@ -167,91 +156,71 @@ def run( source: str, options: BacktestOptions, strategy_params: Mapping[str, JsonValue] | None = None, + strategy_overrides: Mapping[str, JsonValue] | None = None, ) -> dict[str, object]: - """Run raw Pine and normalized OHLCV inside the isolated image.""" + """Run PineScript and normalized OHLCV in pineforge-release.""" - if not pine_source.strip(): - raise ValueError("PineScript source must not be empty") - if not bars: - raise ValueError("bars must not be empty") image = self.ensure_image() with tempfile.TemporaryDirectory(prefix="pineforge-data-") as temporary: workspace = Path(temporary) - (workspace / "strategy.pine").write_text(pine_source, encoding="utf-8") - with (workspace / "ohlcv.csv").open("w", newline="", encoding="utf-8") as handle: - writer = csv.writer(handle) - writer.writerow(("timestamp", "open", "high", "low", "close", "volume")) - for bar in bars: - writer.writerow( - ( - bar.timestamp_ms, - bar.open, - bar.high, - bar.low, - bar.close, - bar.volume, - ) - ) - request = { - "instrument": { - "symbol": instrument.symbol, - "venue": instrument.venue, - "timezone": instrument.timezone, - "session": instrument.session, - "volume_unit": instrument.volume_unit, - }, - "source": source, - "options": { - "input_timeframe": options.input_timeframe, - "script_timeframe": options.script_timeframe, - "bar_magnifier": options.bar_magnifier, - "magnifier_samples": options.magnifier_samples, - "magnifier_distribution": int(options.magnifier_distribution), - "trace_enabled": options.trace_enabled, - "chart_timezone": options.chart_timezone, - }, - "strategy_params": dict(strategy_params or {}), - } - request_path = workspace / "request.json" - request_path.write_text( - json.dumps(request, separators=(",", ":"), allow_nan=False), - encoding="utf-8", - ) - for input_path in ( - workspace / "strategy.pine", - workspace / "ohlcv.csv", - request_path, - ): - input_path.chmod(0o644) - workspace.chmod(0o755) - completed = subprocess.run( - [ - "docker", - "run", - "--rm", - "--network", - "none", - "--read-only", - "--tmpfs", - "/tmp:rw,exec,nosuid,nodev,size=512m", - "--cap-drop", - "ALL", - "--security-opt", - "no-new-privileges", - "--mount", - f"type=bind,src={workspace},dst=/work,readonly", - image, - ], - text=True, - capture_output=True, - check=False, + write_release_inputs(workspace, pine_source, bars, instrument) + environment = release_environment( + "/in", + instrument, + options, + strategy_params, + strategy_overrides, ) + environment["PINEFORGE_DATA_SOURCE"] = source + cid_path = workspace / "container.cid" + command = [ + "docker", + "run", + "--rm", + "--cidfile", + str(cid_path), + "--network", + "none", + "--read-only", + "--tmpfs", + "/tmp:rw,exec,nosuid,nodev,size=512m", + "--cap-drop", + "ALL", + "--security-opt", + "no-new-privileges", + "--mount", + f"type=bind,src={workspace},dst=/in,readonly", + ] + for key, value in sorted(environment.items()): + command.extend(("--env", f"{key}={value}")) + command.append(image) + try: + completed = subprocess.run( + command, + text=True, + capture_output=True, + check=False, + timeout=self.timeout_seconds, + env=os.environ.copy(), + ) + except subprocess.TimeoutExpired as exc: + self._remove_timed_out_container(cid_path) + raise DockerExecutionError( + f"pineforge-release exceeded {self.timeout_seconds:g} seconds" + ) from exc if completed.returncode != 0: - raise DockerExecutionError(_completed_error(completed)) + phase = {2: "input", 3: "compile", 4: "backtest", 5: "transpile"}.get( + completed.returncode, "container" + ) + raise DockerExecutionError(f"{phase} failed: {_completed_error(completed)}") try: - payload = json.loads(completed.stdout) - except json.JSONDecodeError as exc: - raise DockerExecutionError("container returned invalid JSON") from exc - if not isinstance(payload, dict): - raise DockerExecutionError("container report must be a JSON object") - return payload + report = parse_release_report(completed.stdout) + except ReleaseContractError as exc: + raise DockerExecutionError(str(exc)) from exc + return release_response( + report, + release_image=image, + mode="local-container", + request_id=None, + runtime_metadata=self._image_identity(), + ) diff --git a/src/pineforge_data/release_contract.py b/src/pineforge_data/release_contract.py new file mode 100644 index 0000000..8b98806 --- /dev/null +++ b/src/pineforge_data/release_contract.py @@ -0,0 +1,150 @@ +"""Shared input and output contract for the published PineForge release image.""" + +from __future__ import annotations + +import csv +import json +from collections.abc import Mapping, Sequence +from itertools import pairwise +from pathlib import Path + +from .backtest import BacktestOptions, JsonValue +from .models import Bar, Instrument + +DEFAULT_RELEASE_IMAGE = ( + "ghcr.io/pineforge-4pass/pineforge-release:0.1.12@" + "sha256:312b9d908390b828484617472c749d5815feb75507da87eae2f6902cfe3d47b1" +) +RELEASE_ENTRYPOINT = "/opt/pineforge/bin/entrypoint.sh" +RESPONSE_SCHEMA_VERSION = 1 + + +class ReleaseContractError(ValueError): + """A request or release-image response violates the integration contract.""" + + +def _validate_request(pine_source: str, bars: Sequence[Bar]) -> None: + if not pine_source.strip(): + raise ReleaseContractError("PineScript source must not be empty") + if not bars: + raise ReleaseContractError("bars must not be empty") + if any(left.timestamp_ms >= right.timestamp_ms for left, right in pairwise(bars)): + raise ReleaseContractError("bar timestamps must be strictly increasing") + + +def write_release_inputs( + workspace: Path, + pine_source: str, + bars: Sequence[Bar], + instrument: Instrument, +) -> None: + """Write the immutable `/in` files consumed by pineforge-release.""" + + _validate_request(pine_source, bars) + workspace.mkdir(parents=True, exist_ok=True) + pine_path = workspace / "strategy.pine" + ohlcv_path = workspace / "ohlcv.csv" + syminfo_path = workspace / "syminfo.json" + pine_path.write_text(pine_source, encoding="utf-8") + with ohlcv_path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.writer(handle) + writer.writerow(("timestamp", "open", "high", "low", "close", "volume")) + for bar in bars: + writer.writerow( + ( + bar.timestamp_ms, + bar.open, + bar.high, + bar.low, + bar.close, + bar.volume, + ) + ) + syminfo_path.write_text( + json.dumps( + { + "syminfo": { + "ticker": instrument.symbol, + "timezone": instrument.timezone, + "session": instrument.session, + } + }, + separators=(",", ":"), + ), + encoding="utf-8", + ) + for path in (pine_path, ohlcv_path, syminfo_path): + path.chmod(0o644) + workspace.chmod(0o755) + + +def release_environment( + input_directory: str, + instrument: Instrument, + options: BacktestOptions, + strategy_params: Mapping[str, JsonValue] | None = None, + strategy_overrides: Mapping[str, JsonValue] | None = None, +) -> dict[str, str]: + """Translate PineForge options into the release entrypoint environment.""" + + if options.trace_enabled: + raise ReleaseContractError("pineforge-release 0.1.12 does not expose trace collection") + if options.bar_magnifier and options.magnifier_samples < 2: + raise ReleaseContractError("bar magnifier requires at least two samples") + return { + "PINEFORGE_IN_DIR": input_directory, + "PINEFORGE_INPUTS": json.dumps( + dict(strategy_params or {}), separators=(",", ":"), allow_nan=False + ), + "PINEFORGE_OVERRIDES": json.dumps( + dict(strategy_overrides or {}), separators=(",", ":"), allow_nan=False + ), + "PINEFORGE_INPUT_TF": options.input_timeframe, + "PINEFORGE_SCRIPT_TF": options.script_timeframe, + "PINEFORGE_BAR_MAGNIFIER": "true" if options.bar_magnifier else "false", + "PINEFORGE_MAGNIFIER_SAMPLES": str(options.magnifier_samples), + "PINEFORGE_MAGNIFIER_DIST": options.magnifier_distribution.name.lower(), + "PINEFORGE_CHART_TZ": options.chart_timezone or "", + "PINEFORGE_SYMINFO": f"{input_directory.rstrip('/')}/syminfo.json", + "PINEFORGE_DATA_SYMBOL": instrument.symbol, + "PINEFORGE_DATA_VENUE": instrument.venue, + } + + +def parse_release_report(stdout: str) -> dict[str, object]: + """Parse and minimally validate the canonical release-image report.""" + + try: + value = json.loads(stdout) + except json.JSONDecodeError as exc: + raise ReleaseContractError("pineforge-release returned invalid JSON") from exc + if not isinstance(value, dict): + raise ReleaseContractError("pineforge-release report must be a JSON object") + required = ("summary", "trades", "metrics", "diagnostics") + missing = [field for field in required if field not in value] + if missing: + raise ReleaseContractError(f"pineforge-release report is missing: {', '.join(missing)}") + return value + + +def release_response( + report: Mapping[str, object], + *, + release_image: str, + mode: str, + request_id: str | None, + runtime_metadata: Mapping[str, object] | None = None, +) -> dict[str, object]: + """Wrap the release report in the stable pineforge-data transport envelope.""" + + runtime: dict[str, object] = { + "mode": mode, + "release_image": release_image, + } + runtime.update(runtime_metadata or {}) + return { + "schema_version": RESPONSE_SCHEMA_VERSION, + "request_id": request_id, + "runtime": runtime, + "backtest": dict(report), + } diff --git a/src/pineforge_data/server.py b/src/pineforge_data/server.py new file mode 100644 index 0000000..3f5b902 --- /dev/null +++ b/src/pineforge_data/server.py @@ -0,0 +1,642 @@ +"""Bounded-concurrency FastAPI service backed by pineforge-release.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import platform +import re +import secrets +import shutil +import signal +import tempfile +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Annotated, Literal, TypeAlias +from uuid import uuid4 + +from fastapi import FastAPI, Header +from fastapi.responses import JSONResponse +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from .backtest import BacktestOptions +from .compile_cache import CompileCache +from .models import Bar, Instrument +from .release_contract import ( + DEFAULT_RELEASE_IMAGE, + RELEASE_ENTRYPOINT, + ReleaseContractError, + parse_release_report, + release_environment, + release_response, + write_release_inputs, +) + +InputValue: TypeAlias = str | int | float | bool +MagnifierName = Literal[ + "uniform", + "cosine", + "triangle", + "endpoints", + "front_loaded", + "back_loaded", +] +_REQUEST_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") + + +class ApiInstrument(BaseModel): + model_config = ConfigDict(extra="forbid") + + symbol: str = Field(min_length=1, max_length=256) + venue: str = Field(default="", max_length=128) + timezone: str = Field(default="UTC", min_length=1, max_length=128) + session: str = Field(default="24x7", min_length=1, max_length=256) + volume_unit: str = Field(default="base", min_length=1, max_length=64) + + +class ApiBar(BaseModel): + model_config = ConfigDict(extra="forbid", allow_inf_nan=False) + + timestamp_ms: int = Field(ge=0, le=2**63 - 1) + open: float = Field(gt=0) + high: float = Field(gt=0) + low: float = Field(gt=0) + close: float = Field(gt=0) + volume: float = Field(ge=0) + + @model_validator(mode="after") + def validate_ohlc(self) -> ApiBar: + if self.high < max(self.open, self.close, self.low): + raise ValueError("high must be greater than or equal to OHLC values") + if self.low > min(self.open, self.close, self.high): + raise ValueError("low must be less than or equal to OHLC values") + return self + + +class ApiBacktestOptions(BaseModel): + model_config = ConfigDict(extra="forbid") + + input_timeframe: str = Field(default="", max_length=32) + script_timeframe: str = Field(default="", max_length=32) + bar_magnifier: bool = False + magnifier_samples: int = Field(default=4, ge=1, le=10_000) + magnifier_distribution: MagnifierName = "endpoints" + trace_enabled: bool = False + chart_timezone: str | None = Field(default=None, max_length=128) + + +class BacktestApiRequest(BaseModel): + model_config = ConfigDict(extra="forbid", allow_inf_nan=False) + + pine_source: str = Field(min_length=1, max_length=2_000_000) + bars: list[ApiBar] = Field(min_length=1, max_length=1_000_000) + instrument: ApiInstrument + source: str = Field(default="provider", min_length=1, max_length=256) + options: ApiBacktestOptions = Field(default_factory=ApiBacktestOptions) + strategy_params: dict[str, InputValue] = Field(default_factory=dict) + strategy_overrides: dict[str, InputValue] = Field(default_factory=dict) + + @model_validator(mode="after") + def validate_timestamps(self) -> BacktestApiRequest: + if any( + left.timestamp_ms >= right.timestamp_ms + for left, right in zip(self.bars, self.bars[1:], strict=False) + ): + raise ValueError("bar timestamps must be strictly increasing") + return self + + def domain_values( + self, + ) -> tuple[str, list[Bar], Instrument, BacktestOptions]: + instrument = Instrument( + self.instrument.symbol, + venue=self.instrument.venue, + timezone=self.instrument.timezone, + session=self.instrument.session, + volume_unit=self.instrument.volume_unit, + ) + bars = [ + Bar( + instrument, + value.timestamp_ms, + value.open, + value.high, + value.low, + value.close, + value.volume, + self.source, + ) + for value in self.bars + ] + from .backtest import MagnifierDistribution + + options = BacktestOptions( + input_timeframe=self.options.input_timeframe, + script_timeframe=self.options.script_timeframe, + bar_magnifier=self.options.bar_magnifier, + magnifier_samples=self.options.magnifier_samples, + magnifier_distribution=MagnifierDistribution[ + self.options.magnifier_distribution.upper() + ], + trace_enabled=self.options.trace_enabled, + chart_timezone=self.options.chart_timezone, + ) + return self.pine_source, bars, instrument, options + + +class BacktestServiceError(RuntimeError): + def __init__( + self, + code: str, + message: str, + *, + status_code: int, + phase: str, + ) -> None: + super().__init__(message) + self.code = code + self.status_code = status_code + self.phase = phase + + +@dataclass(frozen=True, slots=True) +class ExecutionResult: + report: Mapping[str, object] + runtime_metadata: Mapping[str, object] + + +Executor = Callable[[BacktestApiRequest], Awaitable[ExecutionResult]] + + +def _positive_env(name: str, default: int) -> int: + raw = os.environ.get(name, str(default)) + try: + value = int(raw) + except ValueError as exc: + raise RuntimeError(f"{name} must be an integer") from exc + if value <= 0: + raise RuntimeError(f"{name} must be positive") + return value + + +def _non_negative_env(name: str, default: int) -> int: + raw = os.environ.get(name, str(default)) + try: + value = int(raw) + except ValueError as exc: + raise RuntimeError(f"{name} must be an integer") from exc + if value < 0: + raise RuntimeError(f"{name} must be non-negative") + return value + + +class BacktestService: + """Admit a bounded number of jobs and execute them in isolated directories.""" + + def __init__( + self, + *, + max_concurrency: int | None = None, + max_queue: int | None = None, + queue_timeout_seconds: float | None = None, + execution_timeout_seconds: float | None = None, + release_image: str | None = None, + entrypoint: Path | None = None, + run_json_path: Path | None = None, + cache: CompileCache | None = None, + executor: Executor | None = None, + ) -> None: + cpu_default = max(1, min(4, os.cpu_count() or 1)) + self.max_concurrency = ( + max_concurrency + if max_concurrency is not None + else _positive_env("PINEFORGE_SERVER_CONCURRENCY", cpu_default) + ) + self.max_queue = ( + max_queue + if max_queue is not None + else _non_negative_env("PINEFORGE_SERVER_MAX_QUEUE", self.max_concurrency * 2) + ) + self.queue_timeout_seconds = ( + queue_timeout_seconds + if queue_timeout_seconds is not None + else float(os.environ.get("PINEFORGE_SERVER_QUEUE_TIMEOUT", "30")) + ) + self.execution_timeout_seconds = ( + execution_timeout_seconds + if execution_timeout_seconds is not None + else float(os.environ.get("PINEFORGE_SERVER_EXECUTION_TIMEOUT", "300")) + ) + if self.max_concurrency <= 0 or self.max_queue < 0: + raise ValueError("concurrency must be positive and max_queue non-negative") + if self.queue_timeout_seconds <= 0 or self.execution_timeout_seconds <= 0: + raise ValueError("queue and execution timeouts must be positive") + self.release_image = release_image or os.environ.get( + "PINEFORGE_RELEASE_IMAGE", DEFAULT_RELEASE_IMAGE + ) + self.entrypoint = entrypoint or Path( + os.environ.get("PINEFORGE_RELEASE_ENTRYPOINT", RELEASE_ENTRYPOINT) + ) + self.run_json_path = run_json_path or Path( + os.environ.get("PINEFORGE_RELEASE_RUN_JSON", "/opt/pineforge/bin/run_json.py") + ) + self.cache = cache or CompileCache( + Path(os.environ.get("PINEFORGE_SERVER_CACHE_DIR", "/tmp/pineforge-compile-cache")), + max_entries=_positive_env("PINEFORGE_SERVER_CACHE_MAX_ENTRIES", 1_024), + max_bytes=_positive_env("PINEFORGE_SERVER_CACHE_MAX_BYTES", 2 * 1_024 * 1_024 * 1_024), + ) + self._executor = executor or self._execute_release + self._requires_entrypoint = executor is None + self._semaphore = asyncio.Semaphore(self.max_concurrency) + self._state_lock = asyncio.Lock() + self._admitted = 0 + self._running = 0 + + def ready(self) -> bool: + return not self._requires_entrypoint or ( + self.entrypoint.is_file() + and os.access(self.entrypoint, os.X_OK) + and self.run_json_path.is_file() + and shutil.which("g++") is not None + ) + + async def status(self) -> dict[str, object]: + cache_status = await self.cache.status() + async with self._state_lock: + return { + "ready": self.ready(), + "running": self._running, + "queued": self._admitted - self._running, + "max_concurrency": self.max_concurrency, + "max_queue": self.max_queue, + "release_image": self.release_image, + "compile_cache": cache_status, + } + + async def _admit(self) -> None: + async with self._state_lock: + capacity = self.max_concurrency + self.max_queue + if self._admitted >= capacity: + raise BacktestServiceError( + "server_overloaded", + "backtest capacity is full; retry later", + status_code=429, + phase="queue", + ) + self._admitted += 1 + + async def _release_admission(self) -> None: + async with self._state_lock: + self._admitted -= 1 + + async def run(self, request: BacktestApiRequest, request_id: str) -> dict[str, object]: + await self._admit() + acquired = False + try: + try: + await asyncio.wait_for( + self._semaphore.acquire(), timeout=self.queue_timeout_seconds + ) + acquired = True + except TimeoutError as exc: + raise BacktestServiceError( + "queue_timeout", + "backtest did not reach an execution slot before the queue timeout", + status_code=503, + phase="queue", + ) from exc + async with self._state_lock: + self._running += 1 + try: + result = await self._executor(request) + finally: + async with self._state_lock: + self._running -= 1 + return release_response( + result.report, + release_image=self.release_image, + mode="fastapi-server", + request_id=request_id, + runtime_metadata=result.runtime_metadata, + ) + finally: + if acquired: + self._semaphore.release() + await self._release_admission() + + async def _run_command( + self, + command: list[str], + *, + environment: Mapping[str, str], + phase: str, + deadline: float, + ) -> bytes: + remaining = deadline - asyncio.get_running_loop().time() + if remaining <= 0: + raise BacktestServiceError( + "execution_timeout", + f"backtest exceeded {self.execution_timeout_seconds:g} seconds", + status_code=504, + phase=phase, + ) + try: + process = await asyncio.create_subprocess_exec( + *command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=environment, + start_new_session=True, + ) + except OSError as exc: + raise BacktestServiceError( + "runtime_unavailable", + f"cannot start {phase} process: {exc}", + status_code=503, + phase=phase, + ) from exc + try: + stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=remaining) + except TimeoutError as exc: + try: + os.killpg(process.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError): + process.kill() + await process.communicate() + raise BacktestServiceError( + "execution_timeout", + f"backtest exceeded {self.execution_timeout_seconds:g} seconds", + status_code=504, + phase=phase, + ) from exc + if process.returncode != 0: + detail = stderr.decode("utf-8", "replace").strip() or "pineforge-release failed" + raise BacktestServiceError( + f"{phase}_failed", + detail, + status_code=422 if phase in ("input", "transpile", "compile", "backtest") else 500, + phase=phase, + ) + return stdout + + def _cache_key(self, generated_cpp: bytes) -> tuple[str, str]: + cpp_sha256 = hashlib.sha256(generated_cpp).hexdigest() + identity = "\0".join( + ( + "pineforge-compiled-strategy-v1", + cpp_sha256, + self.release_image, + os.environ.get("PINEFORGE_ENGINE_VERSION", "unknown"), + os.environ.get("PINEFORGE_CODEGEN_VERSION", "unknown"), + os.environ.get("PINEFORGE_RELEASE_VERSION", "unknown"), + platform.machine(), + "g++-std=c++17-O2-ffp-contract=off-fPIC-shared-whole-archive", + ) + ) + return hashlib.sha256(identity.encode()).hexdigest(), cpp_sha256 + + async def _execute_release(self, request: BacktestApiRequest) -> ExecutionResult: + pine_source, bars, instrument, options = request.domain_values() + deadline = asyncio.get_running_loop().time() + self.execution_timeout_seconds + with tempfile.TemporaryDirectory(prefix="pineforge-server-") as temporary: + workspace = Path(temporary) + try: + await asyncio.to_thread( + write_release_inputs, + workspace, + pine_source, + bars, + instrument, + ) + release_environment( + str(workspace), + instrument, + options, + request.strategy_params, + request.strategy_overrides, + ) + except (ReleaseContractError, ValueError) as exc: + raise BacktestServiceError( + "invalid_request", + str(exc), + status_code=400, + phase="input", + ) from exc + environment = os.environ.copy() + environment.update( + { + "PINEFORGE_IN_DIR": str(workspace), + "PINEFORGE_TRANSPILE_ONLY": "1", + } + ) + generated_cpp = await self._run_command( + [str(self.entrypoint)], + environment=environment, + phase="transpile", + deadline=deadline, + ) + generated_path = workspace / "strategy.cpp" + await asyncio.to_thread(generated_path.write_bytes, generated_cpp) + cache_key, cpp_sha256 = self._cache_key(generated_cpp) + inputs = json.dumps(request.strategy_params, separators=(",", ":"), allow_nan=False) + overrides = json.dumps( + request.strategy_overrides, separators=(",", ":"), allow_nan=False + ) + cache_hit = False + async with self.cache.compile_lock(cache_key): + strategy_library = await self.cache.acquire(cache_key) + if strategy_library is None: + temporary_library = self.cache.temporary_path(cache_key) + prefix = os.environ.get("PINEFORGE_PREFIX", "/opt/pineforge") + compile_command = [ + "g++", + "-std=c++17", + "-O2", + "-ffp-contract=off", + "-fPIC", + "-shared", + f"-I{prefix}/include", + "-I/usr/include/eigen3", + str(generated_path), + "-Wl,--whole-archive", + f"{prefix}/lib/libpineforge.a", + "-Wl,--no-whole-archive", + "-o", + str(temporary_library), + ] + try: + await self._run_command( + compile_command, + environment=os.environ.copy(), + phase="compile", + deadline=deadline, + ) + strategy_library = await self.cache.commit_and_acquire( + cache_key, temporary_library + ) + finally: + if temporary_library.exists(): + temporary_library.unlink() + else: + cache_hit = True + if strategy_library is None: + raise BacktestServiceError( + "compile_failed", + "compiled strategy cache artifact was not created", + status_code=500, + phase="compile", + ) + command = [ + "python3", + str(self.run_json_path), + "--so", + str(strategy_library), + "--ohlcv", + str(workspace / "ohlcv.csv"), + "--inputs", + inputs, + "--overrides", + overrides, + "--input-tf", + options.input_timeframe, + "--script-tf", + options.script_timeframe, + "--bar-magnifier", + "true" if options.bar_magnifier else "false", + "--magnifier-samples", + str(options.magnifier_samples), + "--magnifier-dist", + options.magnifier_distribution.name.lower(), + "--generated-cpp", + str(generated_path), + "--transpiled", + "true", + "--syminfo", + str(workspace / "syminfo.json"), + "--chart-tz", + options.chart_timezone or "", + ] + try: + stdout = await self._run_command( + command, + environment=os.environ.copy(), + phase="backtest", + deadline=deadline, + ) + finally: + await self.cache.release(cache_key) + await self.cache.trim() + try: + report = parse_release_report(stdout.decode("utf-8", "replace")) + except ReleaseContractError as exc: + raise BacktestServiceError( + "invalid_release_response", + str(exc), + status_code=502, + phase="response", + ) from exc + return ExecutionResult( + report, + { + "release_version": os.environ.get("PINEFORGE_RELEASE_VERSION", "unknown"), + "engine_version": os.environ.get("PINEFORGE_ENGINE_VERSION", "unknown"), + "codegen_version": os.environ.get("PINEFORGE_CODEGEN_VERSION", "unknown"), + "compile_cache": { + "key": cache_key, + "hit": cache_hit, + "generated_cpp_sha256": cpp_sha256, + }, + }, + ) + + +def _request_id(value: str | None) -> str: + if value is None: + return uuid4().hex + if not _REQUEST_ID.fullmatch(value): + raise BacktestServiceError( + "invalid_request_id", + "X-Request-ID must contain 1-128 safe identifier characters", + status_code=400, + phase="request", + ) + return value + + +def create_app( + service: BacktestService | None = None, + *, + api_key: str | None = None, +) -> FastAPI: + runtime = service or BacktestService() + expected_api_key = ( + api_key if api_key is not None else os.environ.get("PINEFORGE_SERVER_API_KEY", "") + ) + application = FastAPI( + title="PineForge Backtest Server", + version="1", + docs_url="/docs", + redoc_url=None, + ) + + @application.get("/healthz") + async def healthz() -> dict[str, str]: + return {"status": "ok"} + + @application.get("/readyz") + async def readyz() -> JSONResponse: + status = await runtime.status() + return JSONResponse(status_code=200 if status["ready"] else 503, content=status) + + @application.post("/v1/backtests") + async def backtest( + request: BacktestApiRequest, + authorization: Annotated[str | None, Header()] = None, + x_request_id: Annotated[str | None, Header()] = None, + ) -> JSONResponse: + request_id = "" + try: + request_id = _request_id(x_request_id) + if expected_api_key: + scheme, _, credential = (authorization or "").partition(" ") + supplied = credential if scheme.casefold() == "bearer" else "" + if not secrets.compare_digest(supplied, expected_api_key): + raise BacktestServiceError( + "unauthorized", + "a valid bearer token is required", + status_code=401, + phase="authorization", + ) + result = await runtime.run(request, request_id) + return JSONResponse(status_code=200, content=result) + except BacktestServiceError as exc: + return JSONResponse( + status_code=exc.status_code, + content={ + "error": { + "code": exc.code, + "phase": exc.phase, + "message": str(exc), + "request_id": request_id or None, + } + }, + ) + + return application + + +app = create_app() + + +def main() -> None: + import uvicorn + + uvicorn.run( + "pineforge_data.server:app", + host=os.environ.get("PINEFORGE_SERVER_HOST", "0.0.0.0"), + port=int(os.environ.get("PINEFORGE_SERVER_PORT", "8000")), + workers=1, + ) diff --git a/src/pineforge_data/server_client.py b/src/pineforge_data/server_client.py new file mode 100644 index 0000000..cde6371 --- /dev/null +++ b/src/pineforge_data/server_client.py @@ -0,0 +1,141 @@ +"""Standard-library client for the PineForge FastAPI backtest service.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from urllib.error import HTTPError, URLError +from urllib.parse import urlsplit +from urllib.request import Request, urlopen +from uuid import uuid4 + +from .backtest import BacktestOptions, JsonValue +from .models import Bar, Instrument + + +class BacktestServerError(RuntimeError): + """The configured FastAPI server rejected or failed a backtest.""" + + +def _scalar_inputs(values: Mapping[str, JsonValue] | None, field: str) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in (values or {}).items(): + if not isinstance(value, (str, int, float, bool)): + raise ValueError(f"{field}.{key} must be a scalar value") + result[key] = value + return result + + +@dataclass(frozen=True, slots=True) +class FastApiBacktestClient: + """Submit normalized bars to a bounded remote PineForge runtime.""" + + base_url: str + timeout_seconds: float = 330.0 + api_key: str | None = None + max_response_bytes: int = 128 * 1_024 * 1_024 + + def __post_init__(self) -> None: + parsed = urlsplit(self.base_url) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + raise ValueError("server URL must be an absolute http:// or https:// URL") + if self.timeout_seconds <= 0 or self.max_response_bytes <= 0: + raise ValueError("client timeout and response limit must be positive") + + def run( + self, + pine_source: str, + bars: Sequence[Bar], + *, + instrument: Instrument, + source: str, + options: BacktestOptions, + strategy_params: Mapping[str, JsonValue] | None = None, + strategy_overrides: Mapping[str, JsonValue] | None = None, + ) -> dict[str, object]: + payload = { + "pine_source": pine_source, + "bars": [ + { + "timestamp_ms": bar.timestamp_ms, + "open": bar.open, + "high": bar.high, + "low": bar.low, + "close": bar.close, + "volume": bar.volume, + } + for bar in bars + ], + "instrument": { + "symbol": instrument.symbol, + "venue": instrument.venue, + "timezone": instrument.timezone, + "session": instrument.session, + "volume_unit": instrument.volume_unit, + }, + "source": source, + "options": { + "input_timeframe": options.input_timeframe, + "script_timeframe": options.script_timeframe, + "bar_magnifier": options.bar_magnifier, + "magnifier_samples": options.magnifier_samples, + "magnifier_distribution": options.magnifier_distribution.name.lower(), + "trace_enabled": options.trace_enabled, + "chart_timezone": options.chart_timezone, + }, + "strategy_params": _scalar_inputs(strategy_params, "strategy_params"), + "strategy_overrides": _scalar_inputs(strategy_overrides, "strategy_overrides"), + } + headers = { + "Content-Type": "application/json", + "Accept": "application/json", + "X-Request-ID": uuid4().hex, + } + if self.api_key: + headers["Authorization"] = f"Bearer {self.api_key}" + request = Request( + f"{self.base_url.rstrip('/')}/v1/backtests", + data=json.dumps(payload, separators=(",", ":"), allow_nan=False).encode(), + headers=headers, + method="POST", + ) + try: + with urlopen(request, timeout=self.timeout_seconds) as response: + body = response.read(self.max_response_bytes + 1) + except HTTPError as exc: + body = exc.read(self.max_response_bytes + 1) + detail = _error_message(body) or f"HTTP {exc.code}" + raise BacktestServerError(detail) from exc + except URLError as exc: + raise BacktestServerError(f"backtest server is unavailable: {exc.reason}") from exc + if len(body) > self.max_response_bytes: + raise BacktestServerError("backtest server response exceeded the configured limit") + try: + value = json.loads(body) + except json.JSONDecodeError as exc: + raise BacktestServerError("backtest server returned invalid JSON") from exc + if not isinstance(value, dict): + raise BacktestServerError("backtest server response must be a JSON object") + if value.get("schema_version") != 1: + raise BacktestServerError("unsupported backtest server response schema") + if not isinstance(value.get("runtime"), dict) or not isinstance( + value.get("backtest"), dict + ): + raise BacktestServerError("backtest server response is missing runtime data") + return value + + +def _error_message(body: bytes) -> str: + try: + value = json.loads(body) + except json.JSONDecodeError: + return body.decode("utf-8", "replace").strip() + if not isinstance(value, dict) or not isinstance(value.get("error"), dict): + return str(value) + error = value["error"] + phase = error.get("phase", "server") + message = error.get("message", "request failed") + request_id = error.get("request_id") + suffix = f" (request {request_id})" if request_id else "" + return f"{phase}: {message}{suffix}" diff --git a/tests/test_backtest.py b/tests/test_backtest.py index 866f6e7..10d2671 100644 --- a/tests/test_backtest.py +++ b/tests/test_backtest.py @@ -196,3 +196,29 @@ def test_cli_accepts_generic_provider_and_venue_names() -> None: ) assert (args.provider, args.venue) == ("community-broker", "paper") + + +def test_cli_can_route_harness_to_concurrent_server() -> None: + args = build_parser().parse_args( + [ + "--pine", + "strategy.pine", + "--venue", + "kraken", + "--symbol", + "BTC/USD", + "--timeframe", + "15m", + "--start", + "1000", + "--end", + "2000", + "--server-url", + "http://127.0.0.1:8000", + "--execution-timeout", + "60", + ] + ) + + assert args.server_url == "http://127.0.0.1:8000" + assert args.execution_timeout == 60 diff --git a/tests/test_compile_cache.py b/tests/test_compile_cache.py new file mode 100644 index 0000000..96a2e36 --- /dev/null +++ b/tests/test_compile_cache.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import asyncio + +from pineforge_data import CompileCache + + +def test_compile_cache_commits_and_reports_hits(tmp_path) -> None: + async def run() -> None: + cache = CompileCache(tmp_path, max_entries=2, max_bytes=1_000) + assert await cache.lookup("abc") is None + temporary = cache.temporary_path("abc") + temporary.write_bytes(b"compiled") + target = await cache.commit("abc", temporary) + + assert target.read_bytes() == b"compiled" + assert await cache.lookup("abc") == target + status = await cache.status() + assert status["hits"] == 1 + assert status["misses"] == 1 + assert status["compiles"] == 1 + + asyncio.run(run()) + + +def test_compile_lock_deduplicates_same_key(tmp_path) -> None: + async def run() -> None: + cache = CompileCache(tmp_path, max_entries=10, max_bytes=10_000) + active = 0 + maximum = 0 + + async def worker() -> None: + nonlocal active, maximum + async with cache.compile_lock("same"): + active += 1 + maximum = max(maximum, active) + await asyncio.sleep(0.01) + active -= 1 + + await asyncio.gather(*(worker() for _ in range(4))) + assert maximum == 1 + + asyncio.run(run()) + + +def test_trim_preserves_reserved_artifacts(tmp_path) -> None: + async def run() -> None: + cache = CompileCache(tmp_path, max_entries=1, max_bytes=1_000) + first_temp = cache.temporary_path("first") + first_temp.write_bytes(b"first") + first = await cache.commit_and_acquire("first", first_temp) + second_temp = cache.temporary_path("second") + second_temp.write_bytes(b"second") + await cache.commit("second", second_temp) + + await cache.trim() + + assert first.is_file() + assert not cache.path("second").exists() + await cache.release("first") + + asyncio.run(run()) diff --git a/tests/test_docker_integration.py b/tests/test_docker_integration.py index 8972d99..f77cb89 100644 --- a/tests/test_docker_integration.py +++ b/tests/test_docker_integration.py @@ -1,8 +1,12 @@ from __future__ import annotations import os +import subprocess +import time +from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import cast +from urllib.request import urlopen import pytest @@ -10,46 +14,22 @@ BacktestOptions, Bar, DockerBacktestRuntime, + FastApiBacktestClient, Instrument, - discover_repository_root, ) pytestmark = pytest.mark.skipif( os.environ.get("PINEFORGE_DOCKER_TEST") != "1", - reason="set PINEFORGE_DOCKER_TEST=1 to build and exercise the pinned image", + reason="set PINEFORGE_DOCKER_TEST=1 to exercise pineforge-release", ) +ROOT = Path(__file__).resolve().parents[1] -def test_raw_pine_transpiles_compiles_and_backtests_in_docker() -> None: - root = discover_repository_root(Path(__file__).resolve()) - pine = (root / "tests/fixtures/sma_cross.pine").read_text(encoding="utf-8") + +def fixture_values() -> tuple[str, Instrument, list[Bar]]: + pine = (ROOT / "tests/fixtures/sma_cross.pine").read_text(encoding="utf-8") instrument = Instrument("TEST/USD", venue="fixture") - closes = [ - 10, - 11, - 12, - 13, - 12, - 11, - 10, - 9, - 10, - 11, - 12, - 13, - 12, - 11, - 10, - 9, - 10, - 11, - 12, - 13, - 12, - 11, - 10, - 9, - ] + closes = [10, 11, 12, 13, 12, 11, 10, 9] * 3 bars = [ Bar( instrument, @@ -63,7 +43,13 @@ def test_raw_pine_transpiles_compiles_and_backtests_in_docker() -> None: ) for index, close in enumerate(closes) ] - result = DockerBacktestRuntime(root).run( + return pine, instrument, bars + + +def test_local_runtime_uses_published_release_image() -> None: + pine, instrument, bars = fixture_values() + + result = DockerBacktestRuntime().run( pine, bars, instrument=instrument, @@ -74,9 +60,106 @@ def test_raw_pine_transpiles_compiles_and_backtests_in_docker() -> None: runtime = cast(dict[str, object], result["runtime"]) backtest = cast(dict[str, object], result["backtest"]) summary = cast(dict[str, object], backtest["summary"]) - transpile = cast(dict[str, object], result["transpile"]) - assert runtime["engine_commit"] == "9734d48ce32ed61c0a1d0285166276f110e9afaa" - assert runtime["codegen_commit"] == "9aa99e4f0d4734cbfd4a01e4ceb058774580f1ee" - assert len(cast(str, runtime["data_source_digest"])) == 64 - assert summary["input_bars_processed"] == len(bars) - assert int(cast(int, transpile["generated_cpp_bytes"])) > 1_000 + assert runtime["mode"] == "local-container" + assert "pineforge-release:0.1.12@sha256:" in cast(str, runtime["release_image"]) + assert summary["bars_processed"] == len(bars) + + +def _wait_for_server(container_id: str) -> str: + port_result = subprocess.run( + ["docker", "port", container_id, "8000/tcp"], + text=True, + capture_output=True, + check=True, + ) + port = port_result.stdout.strip().rsplit(":", 1)[1] + url = f"http://127.0.0.1:{port}" + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + try: + with urlopen(f"{url}/readyz", timeout=1) as response: + if response.status == 200: + return url + except OSError: + time.sleep(0.25) + logs = subprocess.run( + ["docker", "logs", container_id], + text=True, + capture_output=True, + check=False, + ) + raise AssertionError(f"server did not become ready:\n{logs.stdout}\n{logs.stderr}") + + +def test_server_handles_concurrent_requests_and_reuses_compile_cache() -> None: + image = "pineforge-data-backtest-server:integration" + build = subprocess.run( + [ + "docker", + "build", + "--file", + str(ROOT / "docker/server.Dockerfile"), + "--tag", + image, + str(ROOT), + ], + text=True, + capture_output=True, + check=False, + ) + assert build.returncode == 0, f"server image build failed:\n{build.stdout}\n{build.stderr}" + container = subprocess.run( + [ + "docker", + "run", + "--detach", + "--rm", + "--publish", + "127.0.0.1::8000", + "--read-only", + "--tmpfs", + "/tmp:rw,exec,nosuid,nodev,size=512m", + "--cap-drop", + "ALL", + "--security-opt", + "no-new-privileges", + image, + ], + text=True, + capture_output=True, + check=True, + ).stdout.strip() + try: + url = _wait_for_server(container) + pine, instrument, bars = fixture_values() + client = FastApiBacktestClient(url) + + def submit() -> dict[str, object]: + return client.run( + pine, + bars, + instrument=instrument, + source="fixture", + options=BacktestOptions(input_timeframe="1", script_timeframe="1"), + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + first, second = executor.map(lambda _index: submit(), range(2)) + + cache_results = [ + cast(dict[str, object], cast(dict[str, object], result["runtime"])["compile_cache"]) + for result in (first, second) + ] + assert sorted(cast(bool, result["hit"]) for result in cache_results) == [False, True] + third = submit() + third_cache = cast( + dict[str, object], cast(dict[str, object], third["runtime"])["compile_cache"] + ) + assert third_cache["hit"] is True + finally: + subprocess.run( + ["docker", "rm", "--force", container], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) diff --git a/tests/test_docker_runtime.py b/tests/test_docker_runtime.py index a1f20e5..8ed3b03 100644 --- a/tests/test_docker_runtime.py +++ b/tests/test_docker_runtime.py @@ -1,40 +1,58 @@ from __future__ import annotations -from pathlib import Path +import subprocess import pytest -from pineforge_data import DockerBacktestRuntime, DockerPrerequisiteError +from pineforge_data import ( + DEFAULT_RELEASE_IMAGE, + DockerBacktestRuntime, + DockerPrerequisiteError, +) -def test_image_tag_is_derived_from_both_pins(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr( - DockerBacktestRuntime, - "_pins", - lambda _self: ("a" * 40, "b" * 40), - ) - monkeypatch.setattr(DockerBacktestRuntime, "_source_digest", lambda _self: "c" * 64) - runtime = DockerBacktestRuntime(Path("/tmp/repository")) +def test_default_runtime_is_a_digest_pinned_release_image() -> None: + runtime = DockerBacktestRuntime() - assert runtime.resolved_image() == ( - f"pineforge-data-backtest:d{'c' * 12}-e{'a' * 12}-c{'b' * 12}" - ) + assert runtime.resolved_image() == DEFAULT_RELEASE_IMAGE + assert "pineforge-release:0.1.12@sha256:" in runtime.resolved_image() -def test_explicit_image_does_not_replace_pin_validation(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr( - DockerBacktestRuntime, - "_pins", - lambda _self: ("a" * 40, "b" * 40), +def test_latest_is_an_explicit_rolling_channel() -> None: + runtime = DockerBacktestRuntime( + image="ghcr.io/pineforge-4pass/pineforge-release:latest", + pull_policy="always", ) - monkeypatch.setattr(DockerBacktestRuntime, "_source_digest", lambda _self: "c" * 64) - runtime = DockerBacktestRuntime(Path("/tmp/repository"), image="example/image:test") - assert runtime.resolved_image() == "example/image:test" + assert runtime.resolved_image().endswith(":latest") + assert runtime.pull_policy == "always" + + +def test_missing_image_with_never_policy_is_actionable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime = DockerBacktestRuntime(pull_policy="never") + monkeypatch.setattr(DockerBacktestRuntime, "_check_docker", lambda _self: None) + monkeypatch.setattr(DockerBacktestRuntime, "_image_exists", lambda _self: False) + + with pytest.raises(DockerPrerequisiteError, match="not available locally"): + runtime.ensure_image() + + +def test_missing_policy_pulls_once(monkeypatch: pytest.MonkeyPatch) -> None: + runtime = DockerBacktestRuntime(pull_policy="missing") + pulled: list[bool] = [] + monkeypatch.setattr(DockerBacktestRuntime, "_check_docker", lambda _self: None) + monkeypatch.setattr(DockerBacktestRuntime, "_image_exists", lambda _self: False) + monkeypatch.setattr(DockerBacktestRuntime, "_pull_image", lambda _self: pulled.append(True)) + + assert runtime.ensure_image() == DEFAULT_RELEASE_IMAGE + assert pulled == [True] + +def test_completed_error_prefers_stderr() -> None: + from pineforge_data.docker_runtime import _completed_error -def test_missing_submodule_has_actionable_error(tmp_path: Path) -> None: - runtime = DockerBacktestRuntime(tmp_path) + completed = subprocess.CompletedProcess(["docker"], 1, "stdout", "stderr") - with pytest.raises(DockerPrerequisiteError, match="git submodule update --init"): - runtime._submodule_commit("vendor/pineforge-engine") + assert _completed_error(completed) == "stderr" diff --git a/tests/test_release_contract.py b/tests/test_release_contract.py new file mode 100644 index 0000000..e54936e --- /dev/null +++ b/tests/test_release_contract.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import csv +import json +from pathlib import Path + +import pytest + +from pineforge_data import ( + DEFAULT_RELEASE_IMAGE, + BacktestOptions, + Bar, + Instrument, + ReleaseContractError, +) +from pineforge_data.release_contract import ( + parse_release_report, + release_environment, + write_release_inputs, +) + + +def sample_bars(instrument: Instrument) -> list[Bar]: + return [ + Bar(instrument, 1_000, 10, 12, 9, 11, 5, "fixture"), + Bar(instrument, 61_000, 11, 13, 10, 12, 6, "fixture"), + ] + + +def test_writes_release_mount_contract(tmp_path) -> None: + instrument = Instrument("BTC/USD", venue="kraken", timezone="UTC", session="24x7") + + write_release_inputs(tmp_path, "strategy('x')", sample_bars(instrument), instrument) + + assert (tmp_path / "strategy.pine").read_text() == "strategy('x')" + with (tmp_path / "ohlcv.csv").open(newline="") as handle: + rows = list(csv.reader(handle)) + assert rows[0] == ["timestamp", "open", "high", "low", "close", "volume"] + assert rows[1][0] == "1000" + syminfo = json.loads((tmp_path / "syminfo.json").read_text()) + assert syminfo["syminfo"]["ticker"] == "BTC/USD" + + +def test_release_environment_maps_runtime_options() -> None: + instrument = Instrument("BTC/USD", venue="kraken") + environment = release_environment( + "/in", + instrument, + BacktestOptions( + input_timeframe="15", + script_timeframe="60", + bar_magnifier=True, + magnifier_samples=8, + ), + {"Length": 14}, + {"commission_value": 0.1}, + ) + + assert environment["PINEFORGE_INPUTS"] == '{"Length":14}' + assert environment["PINEFORGE_OVERRIDES"] == '{"commission_value":0.1}' + assert environment["PINEFORGE_INPUT_TF"] == "15" + assert environment["PINEFORGE_SCRIPT_TF"] == "60" + assert environment["PINEFORGE_SYMINFO"] == "/in/syminfo.json" + + +def test_release_contract_rejects_unsupported_trace() -> None: + with pytest.raises(ReleaseContractError, match="does not expose trace"): + release_environment( + "/in", + Instrument("BTC/USD"), + BacktestOptions(trace_enabled=True), + ) + + +def test_release_report_requires_canonical_sections() -> None: + with pytest.raises(ReleaseContractError, match="missing"): + parse_release_report('{"summary":{}}') + + +def test_server_image_and_python_runtime_share_the_release_pin() -> None: + root = Path(__file__).resolve().parents[1] + dockerfile = (root / "docker/server.Dockerfile").read_text(encoding="utf-8") + + assert f"PINEFORGE_RELEASE_IMAGE={DEFAULT_RELEASE_IMAGE}" in dockerfile diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..701254f --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import asyncio + +import pytest +from fastapi.testclient import TestClient + +from pineforge_data import CompileCache +from pineforge_data.server import ( + BacktestApiRequest, + BacktestService, + BacktestServiceError, + ExecutionResult, + create_app, +) + + +def request_payload() -> dict[str, object]: + return { + "pine_source": "//@version=6\nstrategy('test')", + "bars": [ + { + "timestamp_ms": 1_000, + "open": 10, + "high": 12, + "low": 9, + "close": 11, + "volume": 5, + }, + { + "timestamp_ms": 61_000, + "open": 11, + "high": 13, + "low": 10, + "close": 12, + "volume": 6, + }, + ], + "instrument": {"symbol": "BTC/USD", "venue": "kraken"}, + "source": "ccxt:kraken", + "options": {"input_timeframe": "1", "script_timeframe": "1"}, + "strategy_params": {"Length": 14}, + } + + +def canonical_report() -> dict[str, object]: + return { + "summary": {"total_trades": 1}, + "trades": [], + "metrics": {}, + "diagnostics": {"input_bars_processed": 2}, + } + + +def test_api_authenticates_and_preserves_request_id(tmp_path) -> None: + async def execute(_request: BacktestApiRequest) -> ExecutionResult: + return ExecutionResult(canonical_report(), {"compile_cache": {"hit": True}}) + + service = BacktestService( + executor=execute, + cache=CompileCache(tmp_path, max_entries=10, max_bytes=10_000), + ) + client = TestClient(create_app(service, api_key="secret")) + + unauthorized = client.post("/v1/backtests", json=request_payload()) + response = client.post( + "/v1/backtests", + json=request_payload(), + headers={"Authorization": "Bearer secret", "X-Request-ID": "job-123"}, + ) + + assert unauthorized.status_code == 401 + assert response.status_code == 200 + assert response.json()["request_id"] == "job-123" + assert response.json()["runtime"]["compile_cache"]["hit"] is True + + +def test_service_enforces_global_concurrency(tmp_path) -> None: + async def run() -> None: + active = 0 + maximum = 0 + + async def execute(_request: BacktestApiRequest) -> ExecutionResult: + nonlocal active, maximum + active += 1 + maximum = max(maximum, active) + await asyncio.sleep(0.02) + active -= 1 + return ExecutionResult(canonical_report(), {}) + + service = BacktestService( + max_concurrency=2, + max_queue=4, + executor=execute, + cache=CompileCache(tmp_path, max_entries=10, max_bytes=10_000), + ) + request = BacktestApiRequest.model_validate(request_payload()) + results = await asyncio.gather( + *(service.run(request, f"job-{index}") for index in range(6)) + ) + + assert maximum == 2 + assert len(results) == 6 + + asyncio.run(run()) + + +def test_service_rejects_requests_beyond_queue_capacity(tmp_path) -> None: + async def run() -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def execute(_request: BacktestApiRequest) -> ExecutionResult: + started.set() + await release.wait() + return ExecutionResult(canonical_report(), {}) + + service = BacktestService( + max_concurrency=1, + max_queue=0, + executor=execute, + cache=CompileCache(tmp_path, max_entries=10, max_bytes=10_000), + ) + request = BacktestApiRequest.model_validate(request_payload()) + first = asyncio.create_task(service.run(request, "first")) + await started.wait() + with pytest.raises(BacktestServiceError, match="capacity is full"): + await service.run(request, "second") + release.set() + await first + + asyncio.run(run()) + + +def test_compile_cache_key_uses_generated_cpp_and_runtime_identity(tmp_path) -> None: + async def execute(_request: BacktestApiRequest) -> ExecutionResult: + return ExecutionResult(canonical_report(), {}) + + service = BacktestService( + release_image="release@sha256:one", + executor=execute, + cache=CompileCache(tmp_path, max_entries=10, max_bytes=10_000), + ) + + first, first_cpp = service._cache_key(b"generated C++ A") + same, same_cpp = service._cache_key(b"generated C++ A") + different, different_cpp = service._cache_key(b"generated C++ B") + + assert first == same + assert first_cpp == same_cpp + assert different != first + assert different_cpp != first_cpp + + +def test_request_rejects_unsorted_bars() -> None: + payload = request_payload() + bars = payload["bars"] + assert isinstance(bars, list) + bars.reverse() + + with pytest.raises(ValueError, match="strictly increasing"): + BacktestApiRequest.model_validate(payload) diff --git a/tests/test_server_client.py b/tests/test_server_client.py new file mode 100644 index 0000000..5c67185 --- /dev/null +++ b/tests/test_server_client.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +import json +from typing import cast +from urllib.request import Request + +import pytest + +from pineforge_data import BacktestOptions, Bar, FastApiBacktestClient, Instrument + + +class FakeResponse: + def __init__(self, value: dict[str, object]) -> None: + self.body = json.dumps(value).encode() + + def __enter__(self) -> FakeResponse: + return self + + def __exit__(self, *_exc: object) -> None: + return None + + def read(self, _limit: int) -> bytes: + return self.body + + +def test_client_posts_normalized_request(monkeypatch: pytest.MonkeyPatch) -> None: + observed: list[Request] = [] + + def fake_urlopen(request: Request, timeout: float) -> FakeResponse: + assert timeout == 30 + observed.append(request) + return FakeResponse( + { + "schema_version": 1, + "request_id": "one", + "runtime": {"mode": "fastapi-server"}, + "backtest": {"summary": {}}, + } + ) + + monkeypatch.setattr("pineforge_data.server_client.urlopen", fake_urlopen) + instrument = Instrument("BTC/USD", venue="kraken") + bars = [Bar(instrument, 1_000, 10, 12, 9, 11, 5, "ccxt:kraken")] + client = FastApiBacktestClient("http://localhost:8000", timeout_seconds=30, api_key="secret") + + result = client.run( + "strategy('test')", + bars, + instrument=instrument, + source="ccxt:kraken", + options=BacktestOptions(input_timeframe="1"), + strategy_params={"Length": 14}, + ) + + assert result["schema_version"] == 1 + assert len(observed) == 1 + assert observed[0].get_header("Authorization") == "Bearer secret" + body = json.loads(cast(bytes, observed[0].data)) + assert body["bars"][0]["timestamp_ms"] == 1_000 + assert body["strategy_params"] == {"Length": 14} + + +def test_client_rejects_non_scalar_strategy_values() -> None: + client = FastApiBacktestClient("http://localhost:8000") + instrument = Instrument("BTC/USD") + bars = [Bar(instrument, 1_000, 10, 12, 9, 11, 5, "fixture")] + + with pytest.raises(ValueError, match="must be a scalar"): + client.run( + "strategy('test')", + bars, + instrument=instrument, + source="fixture", + options=BacktestOptions(), + strategy_params={"nested": {"bad": True}}, + ) diff --git a/vendor/pineforge-codegen-oss b/vendor/pineforge-codegen-oss deleted file mode 160000 index 9aa99e4..0000000 --- a/vendor/pineforge-codegen-oss +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 9aa99e4f0d4734cbfd4a01e4ceb058774580f1ee diff --git a/vendor/pineforge-engine b/vendor/pineforge-engine deleted file mode 160000 index 9734d48..0000000 --- a/vendor/pineforge-engine +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 9734d48ce32ed61c0a1d0285166276f110e9afaa