herzogfloria/Portfolio_Risk_Management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Kubernetes deployment: shared data directory, write lock, probes

Browse Source 
A third deployment next to the two on the DGX Spark, which stay as they
are: deploy/k8s/ has its own image and Kustomize manifests. The API is a
Deployment with probes, the data a PersistentVolumeClaim, the nightly
update a CronJob; overlays/gpu adds cuOpt on a GPU of the cluster.

Five assumptions of the single-container setup did not hold for pods, so
the service changed (on the Spark nothing changes that one would notice):

- probes: GET /health/live (async, no I/O) and GET /health/ready (data
  directory readable); neither needs cuOpt or market data, so a cuOpt
  outage never gets the API restarted. /health is unchanged
- DataStore replaces every file atomically (temporary file renamed into
  place) and saves the manifest last; the API reloads its market data
  when the manifest changes, so it serves what a CronJob wrote
- one writer at a time across processes and pods: an flock on
  /data/.write.lock, taken by API jobs and the CLI's data commands
  (setup, reselect, update, clean, snapshot); a second writer gets 409 or
  exit status 3, with the holder's name. `clean` now updates the manifest
- job records live in /data/jobs/ and are written at every status
  change: every replica answers GET /jobs/{id}, and a job whose process
  has gone is reported failed ("interrupted")
- the image runs as uid 10001 under a read-only root file system, with
  the library versions pinned to deploy/apptainer/constraints.txt (SciPy
  1.18 releases the GIL while HiGHS solves, so the liveness probe is
  answered during an optimisation)

docs/KUBERNETES.md is the guide (build, deploy, load the data, operate,
troubleshooting); README, USAGE and ARCHITECTURE point to it.

Both kustomizations render (kubectl kustomize); the k3s test of the
deployment is recorded in docs/KUBERNETES.md. 149 tests (148 pass, the
cuOpt round trip skipped without a server).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
herzogflorian 2026-10-29 07:27:43 +01:00
parent 54721801ba
commit c057f8efb7
21 changed files with 986 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
README.md
View File

@ -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=&sector=&country=` · `/universe/{symbol}` | universe with metadata |
| GET | `/prices?symbols=A,B&freq=weekly&currency=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)
```

41
deploy/k8s/Dockerfile Normal file
View File

@ -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 <registry>/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"]

9
deploy/k8s/Dockerfile.dockerignore Normal file
View File

@ -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

128
deploy/k8s/base/cronjobs.yaml Normal file
View File

@ -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: {}

97
deploy/k8s/base/deployment.yaml Normal file
View File

@ -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

23
deploy/k8s/base/kustomization.yaml Normal file
View File

@ -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: <registry>/quantcore
newTag: "0.1.0"

4
deploy/k8s/base/namespace.yaml Normal file
View File

@ -0,0 +1,4 @@
apiVersion: v1
kind: Namespace
metadata:
name: quantcore

16
deploy/k8s/base/pvc.yaml Normal file
View File

@ -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

15
deploy/k8s/base/quantcore.env Normal file
View File

@ -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://<host>: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

17
deploy/k8s/base/service.yaml Normal file
View File

@ -0,0 +1,17 @@
# The API inside the cluster at http://quantcore.quantcore:5001, and from outside at
# http://<node>: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

54
deploy/k8s/overlays/gpu/cuopt.yaml Normal file
View File

@ -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

10
deploy/k8s/overlays/gpu/kustomization.yaml Normal file
View File

@ -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

4
docs/ARCHITECTURE.md
View File

@ -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.

207
docs/KUBERNETES.md Normal file
View File

@ -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 <registry>/quantcore:0.1.0 --push .` and set `newName: <registry>/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://<host>: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://<node>: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://<node>: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://<node>: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 <pod>` | 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.

2
docs/USAGE.md
View File

@ -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) |

147
quantcore/api/app.py
View File

@ -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"])

38
quantcore/cli.py
View File

@ -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

1
quantcore/data/download_service.py
View File

@ -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):

118
quantcore/data/store.py
View File

@ -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/<name>/... 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 ``<data_dir>/.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:

39
tests/test_api.py
View File

@ -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)

39
tests/test_data.py
View File

@ -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