diff --git a/README.md b/README.md index b96d0e6..aa8ec8e 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ never computes a number itself. | **Stock-price models** | `quantcore.models.simulation` | Geometric Brownian motion (multivariate normal) and the empirical-innovation model `eps_t = Sigma_{t-1}^{-1/2} r_t`, `r(j|T) = Sigma_T^{1/2} eps_j` with the latest rolling or EWM covariance. | | **Expected returns** | `quantcore.models.black_litterman` | Market-cap equilibrium returns (broad-market average, market weights), Black–Litterman on a 10-year weekly covariance with **Idzorek confidence** views (diagonal Ω), sentiment → view helper. | | **OptimizationService** | `quantcore.optimization` | Mean–CVaR MILP in integer share numbers: **new portfolio** (from cash) or **rebalancing** (existing holdings, transaction costs); constraints: max 10 % per stock, max per sector, no shorts, cost/turnover limits, optional cardinality; scenario sets with variance reduction (Sobol', moment matching, market-direction importance sampling) that stay identical when risk limits, constraints, costs or views change, tail-row generation for large sets, and an exact Gaussian (mean–variance-equivalent) mode; solved on cuOpt (GPU, over HTTP) with a HiGHS fallback (CPU, compiled into SciPy — runs in-process, nothing to deploy). | -| **HTTP API** | `quantcore.api` | FastAPI app (`/docs`) exposing all of the above; setup/update as background jobs. Runs as a container on the DGX Spark next to cuOpt. | +| **HTTP API** | `quantcore.api` | FastAPI app (`/docs`) exposing all of the above; setup/update as background jobs. Runs as a container on the DGX Spark next to cuOpt, or as a Kubernetes Deployment. | Documentation: **[docs/USAGE.md](docs/USAGE.md) (start here if you are building on this package: the API, the Python client, the `quantcore-api` command line, the conventions and the traps)**, @@ -32,7 +32,8 @@ source `docs/optimization_model.tex`), estimators and shrinkage, GBM and empirical-innovation scenarios, Black–Litterman with Idzorek confidence; source `docs/measurement_and_models.tex`), [docs/SPARK_APPTAINER.md](docs/SPARK_APPTAINER.md) (**students: running it on your own DGX Spark** with -Apptainer, no root), [docs/SPARK_DEPLOYMENT.md](docs/SPARK_DEPLOYMENT.md) (the Docker setup on a Spark with root). +Apptainer, no root), [docs/SPARK_DEPLOYMENT.md](docs/SPARK_DEPLOYMENT.md) (the Docker setup on a Spark with root), +[docs/KUBERNETES.md](docs/KUBERNETES.md) (the Kubernetes deployment: image, manifests, data jobs). --- @@ -176,6 +177,7 @@ res2 = opt.optimize(OptimizationRequest( | Method | Path | Purpose | |---|---|---| | GET | `/health` | store status, cuOpt health | +| GET | `/health/live` · `/health/ready` | liveness / readiness probes (no cuOpt, no market data needed) | | GET | `/universe?region=§or=&country=` · `/universe/{symbol}` | universe with metadata | | GET | `/prices?symbols=A,B&freq=weekly¤cy=base` · `/returns?...` | prices / log-returns | | GET | `/market/index?freq=weekly` · `/market/weights` | cap-weighted market index, weights | @@ -252,13 +254,27 @@ There are two setups. Both run the same code and give the same results. adds a `quantcore` service next to the `cuopt` service of an existing compose file and reaches cuOpt at `http://cuopt:5000`. See [docs/SPARK_DEPLOYMENT.md](docs/SPARK_DEPLOYMENT.md). +## On Kubernetes + +A separate deployment that leaves the two above as they are: its own image +(`deploy/k8s/Dockerfile`) and Kustomize manifests. The API is a Deployment with probes, the data +lives on a PersistentVolumeClaim, and the nightly update is a CronJob. + +```bash +docker build -f deploy/k8s/Dockerfile -t quantcore:0.1.0 . +kubectl apply -k deploy/k8s/base # CPU only (HiGHS); overlays/gpu adds cuOpt on a GPU +``` + +See [docs/KUBERNETES.md](docs/KUBERNETES.md), which also lists what changed in the service for it. + ## Tests ```bash -pytest # 127 tests: reference values for VaR/CVaR, Idzorek closed form vs. numeric, brute-force MILP optimality, +pytest # 149 tests: reference values for VaR/CVaR, Idzorek closed form vs. numeric, brute-force MILP optimality, # exact Gaussian mode vs. brute force, row generation vs. full MILP, sampling/moment matching/tilt, # weighted quantiles and scenario distributions, API (incl. the before/after of a rebalancing), - # the HTTP client and every `quantcore-api` command against the app in-process + # the HTTP client and every `quantcore-api` command against the app in-process, the store's + # write lock and atomic saves, job records shared through the data directory ``` The cuOpt round-trip test runs only when the server is reachable. @@ -282,6 +298,7 @@ quantcore/ install.sh one-step installer/updater for a DGX Spark (wget ... | bash): code, images, data, start deploy/apptainer/ qc.sh (install/start/stop/cli on a Spark without root), quantcore.def, pinned constraints.txt deploy/ Dockerfile, compose service blocks (Spark + a bench-free one), deploy_to_spark.sh -docs/ USAGE.md, ARCHITECTURE.md, METHODOLOGY.md, SPARK_APPTAINER.md, SPARK_DEPLOYMENT.md +deploy/k8s/ Kubernetes: its own Dockerfile, base/ (API, PVC, CronJobs), overlays/gpu/ (+ cuOpt) +docs/ USAGE.md, ARCHITECTURE.md, METHODOLOGY.md, SPARK_APPTAINER.md, SPARK_DEPLOYMENT.md, KUBERNETES.md tests/ pytest suite (synthetic data store fixture) ``` diff --git a/deploy/k8s/Dockerfile b/deploy/k8s/Dockerfile new file mode 100644 index 0000000..2945500 --- /dev/null +++ b/deploy/k8s/Dockerfile @@ -0,0 +1,41 @@ +# quantcore for Kubernetes (deploy/k8s/, guide: docs/KUBERNETES.md). The DGX Spark deployments keep +# their own images: deploy/Dockerfile (Docker Compose) and deploy/apptainer/quantcore.def (Apptainer). +# +# Build context is the repository root; build for the platform of the cluster's nodes: +# docker build -f deploy/k8s/Dockerfile -t quantcore:0.1.0 . # this machine's +# docker buildx build --platform linux/arm64,linux/amd64 -f deploy/k8s/Dockerfile \ +# -t /quantcore:0.1.0 --push . # both, to a registry +# +# What differs from deploy/Dockerfile, all of it for running as a pod: +# * the library versions are pinned (the Apptainer image's constraints.txt): SciPy 1.18 releases +# the GIL while HiGHS solves, so the liveness probe is answered during a long optimisation +# * it runs as an unprivileged user (uid 10001) whose HOME is /tmp: under the pod's read-only root +# file system, /tmp and /data are the only writable paths (yfinance caches its cookies in HOME) +# * no HEALTHCHECK: Kubernetes ignores it; the probes are in base/deployment.yaml +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1 PIP_DISABLE_PIP_VERSION_CHECK=1 \ + PIP_ROOT_USER_ACTION=ignore QUANTCORE_DATA_DIR=/data \ + OPENBLAS_NUM_THREADS=4 OMP_NUM_THREADS=4 MKL_NUM_THREADS=4 +# BLAS threading: the innovation filter runs thousands of small (N x N) eigendecompositions; +# OpenBLAS with all cores is ~50x slower on those than with 1-4 threads. + +RUN apt-get update && apt-get install -y --no-install-recommends tini \ + && rm -rf /var/lib/apt/lists/* \ + && groupadd --system --gid 10001 quantcore \ + && useradd --system --uid 10001 --gid 10001 --home-dir /tmp --no-create-home --shell /usr/sbin/nologin quantcore \ + && mkdir /data && chown 10001:10001 /data + +# the pinned libraries first: this layer survives code changes, so a rebuild only installs quantcore +COPY deploy/apptainer/constraints.txt /src/constraints.txt +RUN pip install -r /src/constraints.txt +COPY pyproject.toml README.md /src/ +COPY quantcore /src/quantcore +RUN pip install -c /src/constraints.txt /src && rm -rf /src + +WORKDIR /app +USER 10001:10001 +EXPOSE 5001 + +ENTRYPOINT ["/usr/bin/tini", "--"] +CMD ["uvicorn", "quantcore.api.app:app", "--host", "0.0.0.0", "--port", "5001", "--workers", "1"] diff --git a/deploy/k8s/Dockerfile.dockerignore b/deploy/k8s/Dockerfile.dockerignore new file mode 100644 index 0000000..09b2e0c --- /dev/null +++ b/deploy/k8s/Dockerfile.dockerignore @@ -0,0 +1,9 @@ +# Build context of deploy/k8s/Dockerfile (BuildKit reads this file for that Dockerfile only): +# just what the image copies; the data directory, .git and build output stay out. +* +!pyproject.toml +!README.md +!quantcore/ +!deploy/apptainer/constraints.txt +**/__pycache__ +**/*.pyc diff --git a/deploy/k8s/base/cronjobs.yaml b/deploy/k8s/base/cronjobs.yaml new file mode 100644 index 0000000..70e825f --- /dev/null +++ b/deploy/k8s/base/cronjobs.yaml @@ -0,0 +1,128 @@ +# The data jobs: the API's image and data volume, the quantcore CLI instead of the server. +# +# While they write they hold the store's write lock, so they never overlap with each other or with +# a job started through the API (POST /jobs/...); a job that finds the lock taken exits with 3 and +# is retried. The API pods pick up the new data by themselves once the job has finished. +--- +# Newest prices and FX, Monday to Friday at 23:15 Zurich time (after the US close). +apiVersion: batch/v1 +kind: CronJob +metadata: + name: quantcore-update +spec: + schedule: "15 23 * * 1-5" + timeZone: Europe/Zurich + concurrencyPolicy: Forbid + startingDeadlineSeconds: 3600 + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 3 + jobTemplate: + spec: + backoffLimit: 3 + activeDeadlineSeconds: 3600 + template: + metadata: + labels: + app.kubernetes.io/name: quantcore + app.kubernetes.io/component: data-job + spec: + restartPolicy: Never + securityContext: + runAsNonRoot: true + runAsUser: 10001 + runAsGroup: 10001 + fsGroup: 10001 + fsGroupChangePolicy: OnRootMismatch + seccompProfile: + type: RuntimeDefault + containers: + - name: update + image: quantcore + imagePullPolicy: IfNotPresent + args: ["quantcore", "update"] + envFrom: + - configMapRef: + name: quantcore-config + resources: + requests: + cpu: 500m + memory: 1Gi + limits: + memory: 4Gi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: data + mountPath: /data + - name: tmp + mountPath: /tmp + volumes: + - name: data + persistentVolumeClaim: + claimName: quantcore-data + - name: tmp + emptyDir: {} +--- +# The one-time setup (~15 min: candidates, metadata, the whole price and FX history). Suspended, so +# it never runs on a schedule; start it when the data volume is empty: +# kubectl -n quantcore create job --from=cronjob/quantcore-setup quantcore-setup-1 +apiVersion: batch/v1 +kind: CronJob +metadata: + name: quantcore-setup +spec: + suspend: true + schedule: "0 0 1 1 *" + concurrencyPolicy: Forbid + jobTemplate: + spec: + backoffLimit: 1 + activeDeadlineSeconds: 7200 + template: + metadata: + labels: + app.kubernetes.io/name: quantcore + app.kubernetes.io/component: data-job + spec: + restartPolicy: Never + securityContext: + runAsNonRoot: true + runAsUser: 10001 + runAsGroup: 10001 + fsGroup: 10001 + fsGroupChangePolicy: OnRootMismatch + seccompProfile: + type: RuntimeDefault + containers: + - name: setup + image: quantcore + imagePullPolicy: IfNotPresent + args: ["quantcore", "setup"] + envFrom: + - configMapRef: + name: quantcore-config + resources: + requests: + cpu: 500m + memory: 1Gi + limits: + memory: 4Gi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: data + mountPath: /data + - name: tmp + mountPath: /tmp + volumes: + - name: data + persistentVolumeClaim: + claimName: quantcore-data + - name: tmp + emptyDir: {} diff --git a/deploy/k8s/base/deployment.yaml b/deploy/k8s/base/deployment.yaml new file mode 100644 index 0000000..0e4da86 --- /dev/null +++ b/deploy/k8s/base/deployment.yaml @@ -0,0 +1,97 @@ +# The API (and the test bench at /ui/): uvicorn with one worker per pod; scale with replicas. +# +# Replicas share the data volume (see pvc.yaml): job records and the write lock live in it, and each +# pod reloads the market data when a data job has written new data. One replica is the default. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: quantcore + labels: + app.kubernetes.io/name: quantcore + app.kubernetes.io/component: api +spec: + replicas: 1 + strategy: + type: RollingUpdate + rollingUpdate: + maxSurge: 1 + maxUnavailable: 0 # the new pod is ready before the old one stops: no downtime on a rollout + selector: + matchLabels: + app.kubernetes.io/name: quantcore + app.kubernetes.io/component: api + template: + metadata: + labels: + app.kubernetes.io/name: quantcore + app.kubernetes.io/component: api + spec: + # SIGTERM lets uvicorn finish the requests in flight: an optimisation can take the whole + # QUANTCORE_CUOPT_TIME_LIMIT (300 s) before it answers + terminationGracePeriodSeconds: 360 + securityContext: + runAsNonRoot: true + runAsUser: 10001 + runAsGroup: 10001 + fsGroup: 10001 # makes the data volume writable for that user + fsGroupChangePolicy: OnRootMismatch + seccompProfile: + type: RuntimeDefault + containers: + - name: api + image: quantcore + imagePullPolicy: IfNotPresent + ports: + - name: http + containerPort: 5001 + envFrom: + - configMapRef: + name: quantcore-config + resources: + requests: + cpu: "1" + memory: 1Gi + limits: + cpu: "4" # = the BLAS threads of quantcore.env + memory: 4Gi + # live: the process answers (no I/O, no cuOpt); ready: the data volume can be read + startupProbe: + httpGet: + path: /health/live + port: http + periodSeconds: 2 + failureThreshold: 60 + livenessProbe: + httpGet: + path: /health/live + port: http + periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 3 + readinessProbe: + httpGet: + path: /health/ready + port: http + periodSeconds: 10 + timeoutSeconds: 5 + lifecycle: + preStop: + exec: + command: ["sleep", "5"] # the Service stops sending requests before uvicorn stops + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: data + mountPath: /data + - name: tmp + mountPath: /tmp + volumes: + - name: data + persistentVolumeClaim: + claimName: quantcore-data + - name: tmp + emptyDir: + sizeLimit: 1Gi diff --git a/deploy/k8s/base/kustomization.yaml b/deploy/k8s/base/kustomization.yaml new file mode 100644 index 0000000..1ac6978 --- /dev/null +++ b/deploy/k8s/base/kustomization.yaml @@ -0,0 +1,23 @@ +# quantcore on Kubernetes, CPU only: kubectl apply -k deploy/k8s/base (guide: docs/KUBERNETES.md) +# +# Runs on any cluster; every optimisation uses HiGHS inside the API pod. overlays/gpu adds cuOpt on +# an NVIDIA GPU of the cluster; a cuOpt that runs elsewhere is one line in quantcore.env. +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +namespace: quantcore +labels: + - pairs: + app.kubernetes.io/part-of: quantcore +resources: + - namespace.yaml + - pvc.yaml + - deployment.yaml + - service.yaml + - cronjobs.yaml +configMapGenerator: + - name: quantcore-config + envs: + - quantcore.env +images: + - name: quantcore # built with deploy/k8s/Dockerfile; a registry: add newName: /quantcore + newTag: "0.1.0" diff --git a/deploy/k8s/base/namespace.yaml b/deploy/k8s/base/namespace.yaml new file mode 100644 index 0000000..c839b1d --- /dev/null +++ b/deploy/k8s/base/namespace.yaml @@ -0,0 +1,4 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: quantcore diff --git a/deploy/k8s/base/pvc.yaml b/deploy/k8s/base/pvc.yaml new file mode 100644 index 0000000..d6dc88d --- /dev/null +++ b/deploy/k8s/base/pvc.yaml @@ -0,0 +1,16 @@ +# The data directory: universe, prices, FX, snapshots and job records (~70 MB, plus snapshots). +# +# ReadWriteOnce, the cluster's default storage class. The API pod and the data-job pods mount it at +# the same time, which works while they run on one node: always so on a one-node cluster (k3s on a +# DGX Spark, Docker Desktop, kind, minikube). With several nodes or replicas, use a ReadWriteMany +# class (NFS, CephFS, Longhorn) -- the write lock in it works across nodes too. +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: quantcore-data +spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 2Gi diff --git a/deploy/k8s/base/quantcore.env b/deploy/k8s/base/quantcore.env new file mode 100644 index 0000000..e36eb1e --- /dev/null +++ b/deploy/k8s/base/quantcore.env @@ -0,0 +1,15 @@ +# Settings of the API pods and the data jobs; every QUANTCORE_* variable of quantcore/config.py works. +# Kustomize turns this file into the ConfigMap quantcore-config, and a change rolls the pods. +# Comments on lines of their own only: text after a value would become part of the value. +QUANTCORE_DATA_DIR=/data +# The Service of overlays/gpu, or a cuOpt outside the cluster (http://:5000). When it cannot +# be reached, as in this CPU-only base, every optimisation runs on HiGHS and the result says so. +QUANTCORE_CUOPT_URL=http://cuopt:5000 +QUANTCORE_CUOPT_TIME_LIMIT=300 +QUANTCORE_LOG_LEVEL=INFO +# false: the JSON API and /docs only, without the test bench at /ui/ +QUANTCORE_ENABLE_UI=true +# small-matrix linear algebra is much slower with more threads; keep at the pod's CPU limit or below +OPENBLAS_NUM_THREADS=4 +OMP_NUM_THREADS=4 +MKL_NUM_THREADS=4 diff --git a/deploy/k8s/base/service.yaml b/deploy/k8s/base/service.yaml new file mode 100644 index 0000000..3b91ae8 --- /dev/null +++ b/deploy/k8s/base/service.yaml @@ -0,0 +1,17 @@ +# The API inside the cluster at http://quantcore.quantcore:5001, and from outside at +# http://:30501 (Docker Desktop: http://localhost:30501). Or, from any machine with kubectl: +# kubectl -n quantcore port-forward svc/quantcore 5001:5001 +apiVersion: v1 +kind: Service +metadata: + name: quantcore +spec: + type: NodePort + selector: + app.kubernetes.io/name: quantcore + app.kubernetes.io/component: api + ports: + - name: http + port: 5001 + targetPort: http + nodePort: 30501 diff --git a/deploy/k8s/overlays/gpu/cuopt.yaml b/deploy/k8s/overlays/gpu/cuopt.yaml new file mode 100644 index 0000000..9c29446 --- /dev/null +++ b/deploy/k8s/overlays/gpu/cuopt.yaml @@ -0,0 +1,54 @@ +# cuOpt, the GPU MILP solver: one pod with one GPU, and the Service "cuopt" that the API reaches at +# http://cuopt:5000. The same image and server command as the Apptainer setup (deploy/apptainer/qc.sh). +apiVersion: apps/v1 +kind: Deployment +metadata: + name: cuopt + labels: + app.kubernetes.io/name: cuopt +spec: + replicas: 1 + strategy: + type: Recreate # one GPU: the old pod has to free it before the new one can start + selector: + matchLabels: + app.kubernetes.io/name: cuopt + template: + metadata: + labels: + app.kubernetes.io/name: cuopt + spec: + runtimeClassName: nvidia + containers: + - name: cuopt + image: nvidia/cuopt:26.8.0-cuda13.3-py3.14 + imagePullPolicy: IfNotPresent + # the image's entrypoint runs its arguments: the cuOpt REST server + args: ["python", "-m", "cuopt_server.cuopt_service", "-i", "0.0.0.0", "-p", "5000", "-l", "info"] + ports: + - name: http + containerPort: 5000 + resources: + limits: + nvidia.com/gpu: 1 + # A startup probe only. cuOpt works on one request at a time and may not answer its health + # check during a long solve: a readiness probe would take the Service away from the solve + # the API is polling, and a liveness probe would restart the server in the middle of it. + startupProbe: + httpGet: + path: /cuopt/health + port: http + periodSeconds: 5 + failureThreshold: 60 +--- +apiVersion: v1 +kind: Service +metadata: + name: cuopt +spec: + selector: + app.kubernetes.io/name: cuopt + ports: + - name: http + port: 5000 + targetPort: http diff --git a/deploy/k8s/overlays/gpu/kustomization.yaml b/deploy/k8s/overlays/gpu/kustomization.yaml new file mode 100644 index 0000000..da48a85 --- /dev/null +++ b/deploy/k8s/overlays/gpu/kustomization.yaml @@ -0,0 +1,10 @@ +# quantcore with cuOpt on an NVIDIA GPU of the cluster (e.g. k3s on a DGX Spark): +# kubectl apply -k deploy/k8s/overlays/gpu +# The cluster needs the NVIDIA device plugin (resource nvidia.com/gpu) and the RuntimeClass +# "nvidia"; see docs/KUBERNETES.md. The base's QUANTCORE_CUOPT_URL already names this Service. +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +namespace: quantcore +resources: + - ../../base + - cuopt.yaml diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3033e02..524bd16 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -123,3 +123,7 @@ the risk limit or the constraints reuses one scenario set. * `quantcore setup` (once), `quantcore update` (scheduled, e.g. nightly), `quantcore snapshot`. * On the Spark the API runs as the `quantcore` container next to `cuopt`, `gurobi`, `jupyter` (see `docs/SPARK_DEPLOYMENT.md`); data lives in `~/services/work/quantcore_data`. +* On Kubernetes (`docs/KUBERNETES.md`) the API is a Deployment, the data a PersistentVolumeClaim and + the update a CronJob. Several processes then share one data directory, so `DataStore` replaces + files atomically, one writer at a time holds its write lock (`/data/.write.lock`), the manifest is + saved last, and the API reloads its market data when the manifest changes. diff --git a/docs/KUBERNETES.md b/docs/KUBERNETES.md new file mode 100644 index 0000000..78f8d56 --- /dev/null +++ b/docs/KUBERNETES.md @@ -0,0 +1,207 @@ +# Running the quant service on Kubernetes + +quantcore has two kinds of deployment. They run the same code and give the same results: + +| Deployment | Where | Files | Guide | +|---|---|---|---| +| **DGX Spark**, Apptainer, no root | a student's Spark | `deploy/apptainer/`, `install.sh` | [SPARK_APPTAINER.md](SPARK_APPTAINER.md) | +| **DGX Spark**, Docker Compose, with root | the lecturer's Spark | `deploy/Dockerfile`, `deploy/docker-compose.*.yml`, `deploy/deploy_to_spark.sh` | [SPARK_DEPLOYMENT.md](SPARK_DEPLOYMENT.md) | +| **Kubernetes** | any cluster: k3s on a Spark, Docker Desktop, kind, minikube, a university or cloud cluster | `deploy/k8s/` (its own image and manifests) | this page | + +The Spark deployments are unchanged. The Kubernetes one has its own Dockerfile and manifests, and +nothing in `deploy/k8s/` is used by the other two. + +## What runs in the cluster + +``` +namespace quantcore + ├─ Deployment quantcore API + test bench, uvicorn, 1 replica, port 5001 + ├─ Service quantcore NodePort 30501 (inside the cluster: http://quantcore.quantcore:5001) + ├─ PVC quantcore-data the data directory /data: universe, prices, FX, snapshots, job records + ├─ ConfigMap quantcore-config generated from deploy/k8s/base/quantcore.env (QUANTCORE_* settings) + ├─ CronJob quantcore-update `quantcore update`, Mon–Fri 23:15 Europe/Zurich + ├─ CronJob quantcore-setup `quantcore setup`, suspended: started by hand, once + └─ Deployment + Service cuopt only with overlays/gpu: cuOpt on one NVIDIA GPU, port 5000 +``` + +``` +deploy/k8s/ + Dockerfile, Dockerfile.dockerignore the image (pinned libraries, uid 10001, read-only root fs) + base/ CPU only: every optimisation runs on HiGHS in the API pod + overlays/gpu/ base + cuOpt on a GPU of the cluster +``` + +## What changed in the service, and why + +A pod is not a container on a host you look after. Kubernetes restarts it when a probe fails, +replaces it on every rollout, may run several copies, and runs the nightly update in a pod of its own. +Five assumptions of the single-container setup did not survive that, and the code changed for them. +These changes apply to the Spark deployments as well, where they change nothing you would notice. + +| Assumption | Why it breaks on Kubernetes | Change | +|---|---|---| +| `/health` is the health check | It asks cuOpt (up to 5 s). As a liveness probe, a cuOpt outage would get healthy API pods restarted | `GET /health/live` (async, no I/O) for liveness, `GET /health/ready` (data volume readable) for readiness. Neither needs cuOpt or market data. `/health` is unchanged | +| The API loads the data once and reloads after its own jobs | The CronJob writes the data from another pod, and the API would keep serving yesterday's prices | Every writer saves `manifest.json` last. The API reloads its market data when that file changes | +| Files are written in place | A reader in another pod could read half a parquet file | `DataStore` writes to a temporary file and renames it into place, an atomic replace | +| One data job at a time, enforced by a lock inside the process | The CronJob, a second replica or the CLI cannot see that lock | A file lock (`flock`) on `/data/.write.lock`, taken by API jobs, CLI data commands and therefore the CronJobs. The kernel frees it when its holder dies. A second writer gets HTTP 409 or exit status 3, with the holder's name | +| Job records live in memory | A restarted pod forgets them, and another replica never knew them | Records are written to `/data/jobs/` at every status change. `GET /jobs/{id}` works from every replica. A job whose pod died is reported `failed` (`interrupted: …`) | + +The Kubernetes image (`deploy/k8s/Dockerfile`) differs from `deploy/Dockerfile` in three ways: + +- **Pinned libraries:** it installs the versions in `deploy/apptainer/constraints.txt`. This matters + for the liveness probe. SciPy 1.18.1 releases the GIL while HiGHS solves: during a 15 s solve the + probe was answered within 0.3 s at worst. SciPy 1.13 blocks every other thread for the whole solve, + and the pod would be restarted in the middle of an optimisation. +- **Unprivileged user:** it runs as uid 10001 with `HOME=/tmp`, under a read-only root file system. + yfinance keeps its cookie cache in `HOME`. +- **No `HEALTHCHECK`:** Kubernetes ignores it. + +## 1. Build the image + +From the repository root, for the platform of the cluster's nodes: + +```bash +docker build -f deploy/k8s/Dockerfile -t quantcore:0.1.0 . +``` + +Then make it available to the cluster: + +| Cluster | Command | +|---|---| +| Docker Desktop | nothing: its Kubernetes uses Docker's images | +| kind | `kind load docker-image quantcore:0.1.0` | +| minikube | `minikube image load quantcore:0.1.0` | +| k3s (e.g. on a Spark; build on the Spark itself, it is arm64) | `docker save quantcore:0.1.0 \| sudo k3s ctr images import -` | +| a registry | `docker buildx build --platform linux/arm64,linux/amd64 -f deploy/k8s/Dockerfile -t /quantcore:0.1.0 --push .` and set `newName: /quantcore` under `images:` in `deploy/k8s/base/kustomization.yaml` | + +The manifests use `imagePullPolicy: IfNotPresent`. A rebuilt image with the same tag is therefore +used only by new pods: after loading it, run `kubectl -n quantcore rollout restart deploy/quantcore`. +With a registry, give every build a new tag instead. + +## 2. Deploy + +**Without a GPU in the cluster:** + +```bash +kubectl apply -k deploy/k8s/base +kubectl -n quantcore rollout status deploy/quantcore +``` + +Every optimisation then runs on HiGHS in the API pod, and the result says `solver: "highs"`. +To use a cuOpt that runs outside the cluster, set `QUANTCORE_CUOPT_URL=http://:5000` in +`deploy/k8s/base/quantcore.env` before applying. For example, the cuOpt container next to the Docker +Compose setup on the lecturer's Spark. + +**With cuOpt on a GPU of the cluster:** + +```bash +kubectl apply -k deploy/k8s/overlays/gpu +``` + +The cluster needs NVIDIA's container runtime and device plugin. On k3s, the RuntimeClass `nvidia` is +created by k3s itself. Then install the device plugin and check that the node offers the GPU: + +```bash +helm repo add nvdp https://nvidia.github.io/k8s-device-plugin +helm upgrade -i nvdp nvdp/nvidia-device-plugin -n nvidia-device-plugin --create-namespace \ + --set runtimeClassName=nvidia +kubectl describe node | grep nvidia.com/gpu # nvidia.com/gpu: 1 +``` + +The cuOpt pod has a startup probe and nothing else. cuOpt works on one request at a time and may not +answer its health check during a long solve. A readiness probe would remove the Service under a solve +the API is still polling, and a liveness probe would restart the solver in the middle of it. + +## 3. Load the data + +The volume starts empty. `GET /health/ready` answers `"market_data": false`, and the data endpoints +answer 503 until there is data. Either copy a prepared data directory into the pod: + +```bash +COPYFILE_DISABLE=1 tar --no-xattrs -C data -cf - manifest.json universe prices fx snapshots \ + | kubectl -n quantcore exec -i deploy/quantcore -- tar -C /data -xf - +``` + +(Leave out `snapshots` if you have none. `COPYFILE_DISABLE=1 --no-xattrs` keeps macOS `tar` from +adding `._*` files; on Linux both are harmless. The command names the entries rather than `.` +because the volume's root directory is not owned by the pod's user.) + +Or build the data from Yahoo Finance, about 15 minutes: + +```bash +kubectl -n quantcore create job --from=cronjob/quantcore-setup quantcore-setup-1 +kubectl -n quantcore logs -f job/quantcore-setup-1 +``` + +`curl -X POST 'http://:30501/jobs/setup?offline=true'` does the same through the API, with the +candidate lists shipped in the package. A job of the API dies with its pod, however, so a rollout +during those 15 minutes interrupts it. The Job above does not have that problem. + +The API picks up the new data by itself. No restart is needed. + +## 4. Use it + +| From | URL | +|---|---| +| outside the cluster | `http://:30501`; Docker Desktop: `http://localhost:30501`; `/docs`, `/ui/` | +| any machine with kubectl (also kind) | `kubectl -n quantcore port-forward svc/quantcore 5001:5001`, then `http://127.0.0.1:5001` | +| another pod in the cluster | `http://quantcore.quantcore:5001` | + +```bash +quantcore-api --url http://:30501 selftest # every endpoint, pass/fail table +``` + +## 5. Operate + +```bash +kubectl -n quantcore get pods,cronjobs,jobs +kubectl -n quantcore logs -f deploy/quantcore +kubectl -n quantcore create job --from=cronjob/quantcore-update update-now # the nightly update, now +kubectl -n quantcore exec deploy/quantcore -- quantcore status +kubectl -n quantcore exec deploy/quantcore -- quantcore snapshot --name demo +``` + +| Task | How | +|---|---| +| New code | build and load the image (section 1); with a new tag, change `newTag` in `base/kustomization.yaml`, then `kubectl apply -k …`. A rollout starts the new pod before it stops the old one, so there is no downtime | +| Change a setting | edit `deploy/k8s/base/quantcore.env` and `kubectl apply -k …`. The ConfigMap gets a new name, which rolls the pods | +| API only, without the test bench | `QUANTCORE_ENABLE_UI=false` in `quantcore.env` | +| Another update time | `schedule` / `timeZone` in `base/cronjobs.yaml` | +| More replicas | `kubectl -n quantcore scale deploy/quantcore --replicas=3`. All replicas mount one volume, so they must share a node (one-node clusters) or the PVC must use a ReadWriteMany storage class. Each replica holds the market data in memory (~0.5–1 GB) | +| Remove it, keep the data | `kubectl -n quantcore delete deploy,svc,cronjob,job --all` | +| Remove everything | `kubectl delete -k deploy/k8s/base` (or `overlays/gpu`). **This deletes the PVC and the data with it** | + +## Settings in the manifests + +| Setting | Value | Why | +|---|---|---| +| API resources | requests 1 CPU / 1 GiB, limits 4 CPU / 4 GiB | measured: 0.75 GB resident after a 150-asset optimisation on the full data set. 4 CPUs match `OPENBLAS_NUM_THREADS=4` | +| `terminationGracePeriodSeconds` | 360 | uvicorn finishes the requests in flight after SIGTERM, and an optimisation can use the whole 300 s time limit | +| `preStop: sleep 5` | | the Service stops routing to the pod before uvicorn stops accepting | +| startup / liveness / readiness | `/health/live` every 2 s up to 120 s / every 20 s, 3 failures / `/health/ready` every 10 s | | +| Pod security | uid 10001, read-only root fs, no capabilities, seccomp `RuntimeDefault` | the API and job pods meet the `restricted` Pod Security Standard; the cuOpt pod runs NVIDIA's image unchanged | +| Update job | `concurrencyPolicy: Forbid`, `backoffLimit: 3`, 1 h deadline | a run that finds the store busy exits with 3 and is retried | + +## Troubleshooting + +| Symptom | Cause / fix | +|---|---| +| `ErrImageNeverPull` / `ImagePullBackOff` | the node does not have the image: load it (section 1), or set `newName` to your registry | +| `PermissionError: '/data/…'` in a pod | the storage class ignores `fsGroup`, so the volume is not writable for uid 10001. Use a class that applies it (local-path, most CSI drivers), or prepare the directory's owner on the node | +| `Multi-Attach error` for `quantcore-data` | a ReadWriteOnce volume is wanted on two nodes (API and job pod, or two replicas). Use a ReadWriteMany class, or keep to one node | +| cuOpt pod `Pending`, `Insufficient nvidia.com/gpu` | the device plugin is missing or the GPU is taken. Without cuOpt the API keeps optimising on HiGHS | +| `409 the data store is busy: … on ` | a data job is running, named with its pod. Wait for it: `kubectl -n quantcore get jobs`, `GET /jobs` | +| a job record says `interrupted` | the pod that ran it was deleted (a rollout, a node drain). Start the job again, preferably as a Kubernetes Job (section 3) | +| `/optimize` answers 504 through an Ingress | the proxy timeout is shorter than the solve. For ingress-nginx: `nginx.ingress.kubernetes.io/proxy-read-timeout: "900"` | +| `/health/ready` answers 503 | the data volume cannot be read: `kubectl -n quantcore describe pod`, and look at the volume | + +## What was tested + +On k3s v1.32.2 (one arm64 node in Docker): + +- `deploy/k8s/base` deployed, and the data imported as in section 3. +- The update CronJob ran as a Job. While it ran, the API answered 409 and named the Job's pod; afterwards it served the new prices without a restart. +- A rollout under load: 547 requests, all 200. +- The pods meet the `restricted` Pod Security Standard. + +The GPU overlay passed a server-side dry run only; that cluster has no GPU. The setup job was not run end to end. diff --git a/docs/USAGE.md b/docs/USAGE.md index e8ed3de..d79fb8e 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -142,6 +142,7 @@ These are not negotiable per request; they are how the whole package thinks. | method | path | what it answers | |---|---|---| | GET | `/health` | is the service up, what data does it hold, is the GPU solver reachable | +| GET | `/health/live`, `/health/ready` | probes for an orchestrator such as Kubernetes: the process answers; the data directory can be read | | GET | `/universe`, `/universe/{symbol}` | the investable stocks and their metadata | | GET | `/data/symbols` | the universe as one table plus filter facets — feeds a stock picker | | GET | `/prices`, `/returns` | adjusted closes and log-returns, daily/weekly/monthly, CHF or local | @@ -325,6 +326,7 @@ that must be identical next week, freeze the data with a snapshot (`POST /snapsh | `503 no market data yet` | the store is empty: `POST /jobs/setup` (~15 min) or mount a prepared data directory | | `422` with a field name | your request does not match the schema; the message names the field. `/docs` has the valid values | | `/health` shows `cuopt.reachable: false` | the GPU solver is down; optimisations fall back to HiGHS and say so | +| `409 the data store is busy: …` | another data job (setup, update, snapshot) is writing, from this service, the CLI or a CronJob; the message names it. Wait until it has finished | | `FeasibleFound` instead of `Optimal` | the time limit hit; raise `time_limit`, or lower `n_scenarios` / `max_assets` | | an optimisation "seems stuck" | it is solving, and the service is single-threaded. `timings` afterwards shows where the time went | | numbers differ slightly from a colleague's | different environment, same inputs: compare `scenario_key`, not `scenario_hash` (§10) | diff --git a/quantcore/api/app.py b/quantcore/api/app.py index 4cadca8..81aca45 100644 --- a/quantcore/api/app.py +++ b/quantcore/api/app.py @@ -4,13 +4,19 @@ Run with ``quantcore serve`` (or ``uvicorn quantcore.api.app:app``). Interactive ``/docs``. Long-running work (setup, data update) runs as background jobs; everything else answers synchronously. All numbers come from the deterministic services -- the API adds no logic of its own. + +Several instances can serve one data directory (replicas of a Kubernetes Deployment): a job +record lives in the data directory next to the data it produces, the store's write lock keeps +the data jobs of all instances one at a time, and each instance reloads its market data when +another process (a CronJob, the CLI) has written new data. """ from __future__ import annotations import json import logging -import threading +import os +import socket import uuid from datetime import datetime, timezone from pathlib import Path @@ -27,7 +33,7 @@ from quantcore.api import schemas as S from quantcore.config import Settings, get_settings from quantcore.data.download_service import DataDownloadService from quantcore.data.market_data import MarketDataService -from quantcore.data.store import DataStore +from quantcore.data.store import DataStore, StoreBusyError, WriteLock from quantcore.measurement.performance import PerformanceService, ewm_performance, rolling_performance from quantcore.measurement.returns import horizon_periods, periods_per_year from quantcore.measurement.risk import RiskService @@ -57,15 +63,26 @@ class State: self.settings = settings or get_settings() self.store = DataStore(self.settings) self._md: MarketDataService | None = None - self.jobs: dict[str, dict] = {} - self.lock = threading.Lock() + self._md_version: tuple | None = None + self.jobs: dict[str, dict] = {} # the jobs of this process; others' are read from jobs_dir + + def _data_version(self) -> tuple | None: + """Identity of the manifest file, which every writer saves last (see DataStore).""" + try: + st = self.store.manifest_path.stat() + except FileNotFoundError: + return None + return st.st_ino, st.st_mtime_ns, st.st_size @property def md(self) -> MarketDataService: - if self._md is None: + version = self._data_version() + if self._md is None or version != self._md_version: if not self.store.has_prices(): raise HTTPException(503, "no market data yet - run the setup job (POST /jobs/setup)") - self._md = MarketDataService(self.settings, self.store) + if self._md is not None: + log.info("the data store has changed; reloading the market data") + self._md, self._md_version = MarketDataService(self.settings, self.store), version return self._md def reload(self) -> None: @@ -128,6 +145,25 @@ def health() -> dict: "cuopt": CuOptSolver(state.settings).health(), "base_currency": state.settings.base_currency}) +@app.get("/health/live", tags=["system"]) +async def health_live() -> dict: + """Liveness: the process answers. Async and free of I/O, so it is answered on the event loop + even while every worker thread is busy optimising, and a down cuOpt or a slow volume never + gets the container restarted.""" + return {"status": "ok"} + + +@app.get("/health/ready", tags=["system"]) +def health_ready() -> dict: + """Readiness: the data directory can be read. Market data is not required (an empty instance + must stay reachable for ``POST /jobs/setup``) and neither is cuOpt (HiGHS takes over).""" + try: + store = state.store.status() + except OSError as exc: + raise HTTPException(503, f"data directory unreadable: {exc}") from exc + return {"status": "ready", "market_data": store["prices"]} + + @app.get("/universe", tags=["data"]) def universe(region: str | None = None, sector: str | None = None, country: str | None = None) -> list[dict]: try: @@ -447,34 +483,66 @@ def optimize(req: S.OptimizeRequest) -> dict: # ------------------------------------------------------------------------------------------------ # jobs: setup and data update (background), snapshots # ------------------------------------------------------------------------------------------------ -def _run_job(job_id: str, fn) -> None: - job = state.jobs[job_id] - job["status"], job["started_at"] = "running", datetime.now(timezone.utc).isoformat(timespec="seconds") +def _now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def _save_job(job: dict) -> None: + """Write the job record to ``jobs_dir``, where every instance (and a restarted one) reads it; + renamed into place, so another replica polling it never reads half a record.""" + path = state.settings.jobs_dir / f"{job['id']}.json" try: - job["result"] = _clean(fn()) - job["status"] = "done" - state.reload() - except Exception as exc: # noqa: BLE001 - log.exception("job %s failed", job_id) - job["status"], job["error"] = "failed", f"{type(exc).__name__}: {exc}" - job["finished_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds") - try: - state.settings.jobs_dir.mkdir(parents=True, exist_ok=True) - (state.settings.jobs_dir / f"{job_id}.json").write_text(json.dumps(job, indent=2, default=str)) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(f".{path.name}.tmp") + tmp.write_text(json.dumps(job, indent=2, default=str)) + os.replace(tmp, path) except OSError: - pass + log.warning("could not write the record of job %s", job["id"]) + + +def _load_job(path: Path) -> dict | None: + """A job record from ``jobs_dir``. One that says ``running`` while no process holds the write + lock (which every data job holds until it has saved its final record) lost its process: the + pod was deleted or the container restarted.""" + try: + job = json.loads(path.read_text()) + except (OSError, ValueError): + return None + if job.get("status") in ("queued", "running") and not state.store.writer_active(): + job["status"], job["error"] = "failed", "interrupted: the process that ran the job has gone" + return job + + +def _run_job(job_id: str, fn, lock: WriteLock) -> None: + job = state.jobs[job_id] + try: + job["status"], job["started_at"] = "running", _now() + _save_job(job) + try: + job["result"] = _clean(fn()) + job["status"] = "done" + state.reload() + except Exception as exc: # noqa: BLE001 + log.exception("job %s failed", job_id) + job["status"], job["error"] = "failed", f"{type(exc).__name__}: {exc}" + job["finished_at"] = _now() + _save_job(job) + finally: + lock.release() def _submit(kind: str, fn, background: BackgroundTasks) -> dict: - with state.lock: - running = [j for j in state.jobs.values() if j["status"] in ("queued", "running") and j["kind"] in ("setup", "update")] - if running and kind in ("setup", "update"): - raise HTTPException(409, f"a {running[0]['kind']} job is already running ({running[0]['id']})") - job_id = uuid.uuid4().hex[:12] - state.jobs[job_id] = {"id": job_id, "kind": kind, "status": "queued", "started_at": None, "finished_at": None, - "result": None, "error": None} - background.add_task(_run_job, job_id, fn) - return state.jobs[job_id] + job_id = uuid.uuid4().hex[:12] + try: # the store's lock, not one of this process: it also covers other replicas, the CLI and CronJobs + lock = state.store.write_lock(f"{kind} job {job_id}") + except StoreBusyError as exc: + raise HTTPException(409, str(exc)) from exc + job = {"id": job_id, "kind": kind, "status": "queued", "host": socket.gethostname(), "created_at": _now(), + "started_at": None, "finished_at": None, "result": None, "error": None} + state.jobs[job_id] = job + _save_job(job) + background.add_task(_run_job, job_id, fn, lock) + return job @app.post("/jobs/setup", tags=["jobs"], status_code=202) @@ -491,22 +559,35 @@ def job_update(background: BackgroundTasks, full: bool = False) -> dict: @app.get("/jobs", tags=["jobs"]) def jobs() -> list[dict]: - return list(state.jobs.values()) + """Every job of this data directory, the oldest first (those of other instances included).""" + records = {} + if state.settings.jobs_dir.is_dir(): + for path in state.settings.jobs_dir.glob("*.json"): + if (record := _load_job(path)) is not None: + records[record["id"]] = record + records.update(state.jobs) + return sorted(records.values(), key=lambda j: j.get("created_at") or j.get("started_at") or "") @app.get("/jobs/{job_id}", tags=["jobs"]) def job(job_id: str) -> dict: - if job_id not in state.jobs: + if job_id in state.jobs: + return state.jobs[job_id] + record = _load_job(state.settings.jobs_dir / f"{job_id}.json") if job_id.isalnum() else None + if record is None: raise HTTPException(404, "unknown job") - return state.jobs[job_id] + return record @app.post("/snapshots", tags=["jobs"], status_code=201) def create_snapshot(req: S.SnapshotRequest) -> dict: try: - return {"path": str(state.store.create_snapshot(req.name))} + with state.store.write_lock(f"snapshot {req.name or ''}".strip()): + return {"path": str(state.store.create_snapshot(req.name))} except FileExistsError as exc: raise HTTPException(409, f"snapshot exists: {exc}") from exc + except StoreBusyError as exc: + raise HTTPException(409, str(exc)) from exc @app.get("/snapshots", tags=["jobs"]) diff --git a/quantcore/cli.py b/quantcore/cli.py index 3eec322..d1d0c7a 100644 --- a/quantcore/cli.py +++ b/quantcore/cli.py @@ -15,6 +15,10 @@ These commands run the services in this process and need the data directory locally. To drive a *running* service over HTTP instead -- from any machine, without the data -- use ``quantcore-api`` (:mod:`quantcore.apicli`), whose ``selftest`` also checks a deployment endpoint by endpoint. + +The commands that write the data (setup, reselect, update, clean, snapshot) take the store's write +lock, so they never run at the same time as each other or as a data job of the API; one that finds +the lock taken exits with status 3 (on Kubernetes the CronJob retries it). """ from __future__ import annotations @@ -23,6 +27,7 @@ import argparse import json import logging import sys +from pathlib import Path import pandas as pd @@ -40,10 +45,18 @@ def _settings(args) -> Settings: return s +def _write_lock(s: Settings, command: str): + from quantcore.data.store import DataStore + + return DataStore(s).write_lock(f"quantcore {command}") + + def cmd_setup(args) -> int: from quantcore.universe.setup_service import SetupService - report = SetupService(_settings(args)).run(allow_web=not args.offline, max_per_region=args.max_per_region) + s = _settings(args) + with _write_lock(s, "setup"): + report = SetupService(s).run(allow_web=not args.offline, max_per_region=args.max_per_region) print(json.dumps(report.as_dict(), indent=2, default=str)) return 0 @@ -51,7 +64,9 @@ def cmd_setup(args) -> int: def cmd_reselect(args) -> int: from quantcore.universe.setup_service import SetupService - report = SetupService(_settings(args)).rebuild_selection() + s = _settings(args) + with _write_lock(s, "reselect"): + report = SetupService(s).rebuild_selection() print(json.dumps(report.as_dict(), indent=2, default=str)) return 0 @@ -59,7 +74,9 @@ def cmd_reselect(args) -> int: def cmd_update(args) -> int: from quantcore.data.download_service import DataDownloadService - report = DataDownloadService(_settings(args)).update(full=args.full) + s = _settings(args) + with _write_lock(s, "update"): + report = DataDownloadService(s).update(full=args.full) print(json.dumps(report.as_dict(), indent=2, default=str)) return 0 @@ -67,7 +84,9 @@ def cmd_update(args) -> int: def cmd_clean(args) -> int: from quantcore.data.download_service import DataDownloadService - report = DataDownloadService(_settings(args)).clean() + s = _settings(args) + with _write_lock(s, "clean"): + report = DataDownloadService(s).clean() print(json.dumps(report.as_dict(), indent=2, default=str)) return 0 @@ -83,7 +102,9 @@ def _rounded(table: pd.DataFrame, digits: int) -> pd.DataFrame: def cmd_snapshot(args) -> int: from quantcore.data.store import DataStore - path = DataStore(_settings(args)).create_snapshot(args.name) + s = _settings(args) + with _write_lock(s, "snapshot"): + path = DataStore(s).create_snapshot(args.name) print(path) return 0 @@ -179,7 +200,7 @@ def cmd_serve(args) -> int: def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(prog="quantcore", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - p.add_argument("--data-dir", help="override QUANTCORE_DATA_DIR") + p.add_argument("--data-dir", type=Path, help="override QUANTCORE_DATA_DIR") p.add_argument("--cuopt-url", help="override QUANTCORE_CUOPT_URL") sub = p.add_subparsers(dest="command", required=True) @@ -253,9 +274,14 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: list[str] | None = None) -> int: + from quantcore.data.store import StoreBusyError + args = build_parser().parse_args(argv) try: return args.func(args) + except StoreBusyError as exc: + print(f"error: {exc}.\n Wait for it to finish (quantcore-api jobs lists the API's jobs).", file=sys.stderr) + return 3 except FileNotFoundError as exc: if "run the setup service first" not in str(exc): raise diff --git a/quantcore/data/download_service.py b/quantcore/data/download_service.py index ee0437e..c8b596d 100644 --- a/quantcore/data/download_service.py +++ b/quantcore/data/download_service.py @@ -161,6 +161,7 @@ class DataDownloadService: self.store.save_prices(prices) fx = clean_fx(self.store.load_fx(), report=report) self.store.save_fx(fx) + self.store.update_manifest(last_clean=date.today().isoformat()) # the manifest last: readers reload return report def snapshot(self, name: str | None = None): diff --git a/quantcore/data/store.py b/quantcore/data/store.py index 5bc3244..0ac46ca 100644 --- a/quantcore/data/store.py +++ b/quantcore/data/store.py @@ -9,16 +9,27 @@ Layout under ``settings.data_dir``:: prices/adj_close.parquet date x symbol adjusted close in the listing currency (ISO) fx/fx_chf.parquet date x currency, CHF per unit snapshots//... frozen copies for reproducible demos and grading + .write.lock held by the one process that writes (see :class:`WriteLock`) Wide parquet frames are cheap to load (1000 columns x 5500 days ~ 40 MB in memory) and are the only format downstream services read. + +Several processes can share one data directory -- the API, the CLI, and on Kubernetes the API pods +and the CronJob that updates the data. So every file is replaced atomically (a reader sees the old +file or the new one), one writer at a time holds the write lock, and a writer saves the manifest +last: the API reloads its market data when the manifest changes. """ from __future__ import annotations +import fcntl import json import logging +import os import shutil +import socket +import time +from collections.abc import Callable from datetime import datetime, timezone from pathlib import Path @@ -29,6 +40,77 @@ from quantcore.config import Settings log = logging.getLogger(__name__) +class StoreBusyError(RuntimeError): + """Another process holds the write lock of the data directory.""" + + +class WriteLock: + """Exclusive lock of a data directory: one writer at a time, across processes and pods. + + An advisory ``flock`` on ``/.write.lock``, so it holds for every process that mounts + the directory -- the API's jobs, the CLI, a Kubernetes CronJob. The kernel releases it when its + holder exits, even when the holder is killed, so a crashed job never blocks the next one. + The holder writes who it is into the file, for the error message of the next one to try. + """ + + def __init__(self, path: Path, holder: str): + self.path, self.holder = path, holder + self._fh = None + + def acquire(self, wait: float = 1.0) -> "WriteLock": + """Take the lock; :class:`StoreBusyError` if it stays taken for ``wait`` seconds. + + The short wait covers :meth:`DataStore.writer_active`, which holds a shared lock for a + moment to look; a real writer holds it for minutes. + """ + self.path.parent.mkdir(parents=True, exist_ok=True) + fh = open(self.path, "a+") + deadline = time.monotonic() + wait + while True: + try: + fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + if time.monotonic() >= deadline: + fh.seek(0) + current = fh.read().strip() or "another process" + fh.close() + raise StoreBusyError(f"the data store is busy: {current}") from None + time.sleep(0.05) + fh.seek(0) + fh.truncate() + since = datetime.now(timezone.utc).isoformat(timespec="seconds") + fh.write(f"{self.holder} on {socket.gethostname()} (pid {os.getpid()}) since {since}") + fh.flush() + self._fh = fh + return self + + def release(self) -> None: + if self._fh is None: + return + try: + self._fh.truncate(0) + finally: + self._fh.close() # closing the file releases the flock + self._fh = None + + def __enter__(self) -> "WriteLock": + return self + + def __exit__(self, *exc: object) -> None: + self.release() + + +def _replace(path: Path, write: Callable[[Path], object]) -> None: + """Write ``path`` through a temporary file renamed into place (an atomic replace).""" + tmp = path.with_name(f".{path.name}.tmp") + try: + write(tmp) + os.replace(tmp, path) + finally: + tmp.unlink(missing_ok=True) + + class DataStore: """File-based store; one instance per data directory (a snapshot is just another directory).""" @@ -69,6 +151,30 @@ class DataStore: def manifest_path(self) -> Path: return self.root / "manifest.json" + @property + def lock_path(self) -> Path: + return self.root / ".write.lock" + + # -- writers ----------------------------------------------------------------------------------- + def write_lock(self, holder: str) -> WriteLock: + """Take the write lock (``with store.write_lock("quantcore update"): ...``). + + Raises :class:`StoreBusyError` when another process is writing. + """ + return WriteLock(self.lock_path, holder).acquire() + + def writer_active(self) -> bool: + """Does some process hold the write lock right now? (A shared lock for a moment: checks + never conflict with each other, and a writer that comes along waits for it.)""" + if not self.lock_path.exists(): + return False + with open(self.lock_path) as fh: + try: + fcntl.flock(fh, fcntl.LOCK_SH | fcntl.LOCK_NB) + except BlockingIOError: + return True + return False + def has_universe(self) -> bool: return self.universe_path.exists() @@ -78,7 +184,7 @@ class DataStore: # -- universe ------------------------------------------------------------------------------ def save_universe(self, df: pd.DataFrame) -> None: self.ensure_dirs() - df.to_csv(self.universe_path, index=False) + _replace(self.universe_path, lambda p: df.to_csv(p, index=False)) def load_universe(self) -> pd.DataFrame: """Universe indexed by symbol (raises ``FileNotFoundError`` before the setup ran).""" @@ -89,14 +195,14 @@ class DataStore: def save_candidates(self, df: pd.DataFrame) -> None: self.ensure_dirs() - df.to_csv(self.candidates_path, index=False) + _replace(self.candidates_path, lambda p: df.to_csv(p, index=False)) def load_candidates(self) -> pd.DataFrame: return pd.read_csv(self.candidates_path, dtype={"symbol": str, "source_ticker": str}) def save_metadata(self, df: pd.DataFrame) -> None: self.ensure_dirs() - df.to_csv(self.metadata_path, index=True) + _replace(self.metadata_path, lambda p: df.to_csv(p, index=True)) def load_metadata(self) -> pd.DataFrame: return pd.read_csv(self.metadata_path, dtype={"symbol": str}).set_index("symbol") @@ -107,7 +213,7 @@ class DataStore: prices = prices.sort_index() prices.index.name = "date" prices.columns = [str(c) for c in prices.columns] - prices.to_parquet(self.prices_path) + _replace(self.prices_path, prices.to_parquet) def load_prices( self, symbols: list[str] | None = None, start: str | None = None, end: str | None = None @@ -138,7 +244,7 @@ class DataStore: self.ensure_dirs() fx = fx.sort_index() fx.index.name = "date" - fx.to_parquet(self.fx_path) + _replace(self.fx_path, fx.to_parquet) def load_fx(self) -> pd.DataFrame: if not self.fx_path.exists(): @@ -158,7 +264,7 @@ class DataStore: manifest = self.read_manifest() manifest.update(entries) manifest["updated_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds") - self.manifest_path.write_text(json.dumps(manifest, indent=2, default=str)) + _replace(self.manifest_path, lambda p: p.write_text(json.dumps(manifest, indent=2, default=str))) return manifest def status(self) -> dict: diff --git a/tests/test_api.py b/tests/test_api.py index 00bf676..325b7d8 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,8 +1,12 @@ +import json + import numpy as np import pytest from fastapi.testclient import TestClient import quantcore.api.app as app_module +from quantcore.config import Settings +from quantcore.data.store import DataStore @pytest.fixture(scope="module") @@ -97,6 +101,41 @@ def test_models_and_optimize(client): assert sum(before["sector_exposure"].values()) == pytest.approx(before["invested"] / reb["wealth"], rel=1e-6) +def test_probes(client): + assert client.get("/health/live").json() == {"status": "ok"} + assert client.get("/health/ready").json() == {"status": "ready", "market_data": True} + + +def test_reloads_data_written_by_another_process(client, store): + md = app_module.state.md + assert app_module.state.md is md # nothing written: the loaded data are kept + store.update_manifest() # what a CronJob or the CLI does last after writing new data + assert app_module.state.md is not md + + +def test_job_records_live_in_the_data_directory(tmp_path, monkeypatch): + """What another replica, or the pod after a restart, sees of a job: its record in jobs_dir.""" + settings = Settings(data_dir=tmp_path, cuopt_url="http://127.0.0.1:9") + monkeypatch.setattr(app_module, "state", app_module.State(settings)) + c = TestClient(app_module.app) + settings.jobs_dir.mkdir(parents=True) + record = {"id": "abc123", "kind": "update", "status": "running", "created_at": "2026-10-22T08:00:00+00:00"} + (settings.jobs_dir / "abc123.json").write_text(json.dumps(record)) + with DataStore(settings).write_lock("update job abc123"): # its process is alive + assert c.get("/jobs/abc123").json()["status"] == "running" + busy = c.post("/jobs/update") + assert busy.status_code == 409 and "update job abc123" in busy.json()["detail"] + gone = c.get("/jobs/abc123").json() # lock released, no final record: the process has gone + assert gone["status"] == "failed" and gone["error"].startswith("interrupted") + # a job of this process: runs, fails (no data), leaves its record and frees the lock + job = c.post("/jobs/update").json() + final = c.get(f"/jobs/{job['id']}").json() + assert final["status"] == "failed" and "setup" in final["error"] + assert json.loads((settings.jobs_dir / f"{job['id']}.json").read_text())["status"] == "failed" + assert [j["id"] for j in c.get("/jobs").json()] == ["abc123", job["id"]] + assert c.post("/jobs/update").status_code == 202 + + def test_jobs_and_snapshots(client): snaps = client.get("/snapshots").json() assert isinstance(snaps, list) diff --git a/tests/test_data.py b/tests/test_data.py index a40d5f5..594a279 100644 --- a/tests/test_data.py +++ b/tests/test_data.py @@ -150,3 +150,42 @@ def test_unknown_symbols_are_summarised_not_logged_one_by_one(monkeypatch, caplo summary = [r.getMessage() for r in caplog.records if "no quote for 2 of 3" in r.getMessage()] assert summary and "GONE1.PA, GONE2.SW" in summary[0] assert logging.getLogger("yfinance").level == logging.NOTSET # restored afterwards + + +def test_write_lock_admits_one_writer_at_a_time(tmp_path): + from quantcore.config import Settings + from quantcore.data.store import DataStore, StoreBusyError, WriteLock + + store = DataStore(Settings(data_dir=tmp_path)) + assert not store.writer_active() + with store.write_lock("quantcore update"): + assert store.writer_active() and store.writer_active() # checks do not disturb each other + with pytest.raises(StoreBusyError, match="quantcore update on"): + WriteLock(store.lock_path, "quantcore setup").acquire(wait=0) + assert not store.writer_active() and store.lock_path.read_text() == "" + store.write_lock("quantcore setup").release() # free again once the holder has gone + + +def test_saves_replace_files_atomically(tmp_path): + """A reader in another process sees the old file or the new one, never one being written.""" + from quantcore.config import Settings + from quantcore.data.store import DataStore + + store = DataStore(Settings(data_dir=tmp_path)) + store.update_manifest(n_universe=1) + first = store.manifest_path.stat().st_ino + store.update_manifest(n_universe=2) + assert store.manifest_path.stat().st_ino != first # a new file renamed into place + assert store.read_manifest()["n_universe"] == 2 + assert not list(tmp_path.rglob("*.tmp")) + + +def test_cli_data_commands_refuse_a_busy_store(tmp_path, capsys): + from quantcore import cli + from quantcore.config import Settings + from quantcore.data.store import DataStore + + with DataStore(Settings(data_dir=tmp_path)).write_lock("update job 1a2b"): + assert cli.main(["--data-dir", str(tmp_path), "snapshot", "--name", "x"]) == 3 + assert "busy: update job 1a2b" in capsys.readouterr().err + assert cli.main(["--data-dir", str(tmp_path), "snapshot", "--name", "x"]) == 0