Datastream Backend

A service to serve and build financial time-series data.

Service

A single Python FastAPI service handles both public API traffic and builds. It listens on port 3000.

Public endpoints

GET /status

GET /datasets

POST /build/{dataset_name}/{dataset_version}?start=<timestamp>&end=<timestamp>&dry-run=<bool>

GET /data/{dataset_name}/{dataset_version}?start=<timestamp>&end=<timestamp>&build-data=<bool>

DELETE /data/{dataset_name}/{dataset_version}?start=<timestamp>&end=<timestamp>

POST /datasets

API authentication

Every endpoint except GET /status requires a valid API key sent as a bearer token:

Authorization: Bearer <raw-key>

/status is intentionally unauthenticated so the Docker healthcheck (which hits /api/v1/status) keeps working.

Key storage. Valid keys live in the API_KEYS env var as comma-separated label:sha256hex pairs, e.g. default:ab12…,team-a:cd34…. Only the sha256 hash of each key is stored, so a leaked config or log never reveals a working key. The label identifies the caller (a team) and is bound to the structlog context as team on every authenticated request. The env format is already a label -> hash map, so extending from one shared key to a key-per-team is just another entry — no format change.

Verification. The verify_api_key dependency (core/auth.py) hashes the presented token and looks it up in the key map. Comparing the hash (not the raw key) is timing-safe: the compared value is already a sha256 of the secret, so timing cannot recover the key. Enforcement is wired once in main.py: public_router (carrying /status) mounts open, and router (everything else) mounts with dependencies=[Depends(verify_api_key)].

Fail-closed startup. The lifespan handler loads API_KEYS and raises if it is empty, so the service refuses to start unauthenticated on a public network. There is no "auth disabled" flag.

Key generation. Mint a key and its env line with:

cd builders/server && uv run python -m core.auth generate <label>

This prints a dsk_-prefixed raw key (hand to the client) and the label:hash line (add to API_KEYS). Rotation is a superset operation: add a new key alongside the old, roll clients over, then remove the old entry.

Scaling assumption. Keys are loaded from process env into an in-memory (cached) map, consistent with the single-uvicorn-worker model used elsewhere in the service.

Status	Meaning
`401`	Missing or invalid API key (on any endpoint except `/status`)

Clients. The Python SDK sends the header automatically when given a key: pass DatastreamClient(api_key=...) (or the module-level get_data(..., api_key=...)), set it globally via configure(api_key=...), or export DATASTREAM_API_KEY. The browser frontend prompts the user for a key and sends it on every request (see SPEC-frontend.md).

CORS

The browser frontend is served from GitHub Pages — a different origin than the API — so main.py adds Starlette's CORSMiddleware:

Allowed origins come from the CORS_ALLOW_ORIGINS env var (comma-separated exact origins). Default: https://wat-street.github.io,http://localhost:5173 (the Pages origin and the local Vite dev server). Origins never include a path — the Pages origin is https://wat-street.github.io, not .../datastream.
The middleware is added last, so it wraps outermost: preflight OPTIONS requests short-circuit before auth and request-context logging.
allow_credentials stays off (auth is a bearer header, not cookies); allowed headers are Authorization and Content-Type; all methods are allowed.
Auth is unchanged. CORS only permits browsers to send the Authorization header cross-origin; every non-public route still requires a valid API key.

Datasets endpoint

GET /datasets returns all datasets pre-loaded into the runtime config registry at startup, annotated with whether each has any data in the database.

Response format:

{
  "datasets": [
    {"name": "mock-ohlc", "version": "0.1.0", "has_data": true},
    {"name": "faang-daily-close", "version": "0.1.0", "has_data": false}
  ]
}

has_data: true when at least one row exists in the DB for that (name, version) pair, false otherwise
datasets are sorted alphabetically by name, then by version
datasets are loaded at startup by scanning SCRIPTS_DIR for config.toml files and caching them in the runtime registry

Status codes:

Status	Meaning
`200`	success
`500`	unexpected failure (filesystem error, DB error)

Data endpoint

GET /data fetches dataset data for a time range. By default it builds missing data before returning, ensuring complete results. Callers can opt out of building with build-data=false for fast read-only access.

Query parameters:

start, end (required): timestamp range
build-data (optional, default true): when true, builds missing data before fetching; when false, returns only existing data

Response format:

{
  "dataset_name": "mock-ohlc",
  "dataset_version": "0.1.0",
  "total_timestamps": 3,
  "returned_timestamps": 3,
  "rows": [
    {
      "timestamp": "2024-01-02T00:00:00",
      "data": [
        {"ticker": "AAPL", "open": 100, "high": 150, "low": 90, "close": 130}
      ]
    }
  ]
}

Each entry in rows contains all data dicts for that timestamp (matching the DB model where multiple rows can share a timestamp). Rows are sorted by timestamp. If no data exists for the range, rows is an empty list.

Metadata fields:

total_timestamps: number of valid calendar timestamps in the requested range (computed via generate_timestamps())
returned_timestamps: number of distinct timestamps actually returned from the DB

Status codes:

Status	Meaning
`200`	Data is complete, or `build-data=true` (default) was used
`206`	`build-data=false` and `returned_timestamps < total_timestamps` (incomplete data)
`400`	Malformed input (invalid version or timestamp)
`401`	Missing or invalid API key
`422`	`build-data=true` but no valid calendar timestamps in range
`500`	Unexpected failure (config not found, DB error)

Delete endpoint

DELETE /data deletes a dataset's rows in a time range. The delete flows through the same three layers as GET /data (API → service → DB): the route parses input, delete_data() in service/builder.py validates and coordinates, and delete_rows_range() in db/datasets.py issues the query.

Query parameters:

start, end (required): timestamp range; rows with start <= timestamp <= end are deleted

Response format:

{
  "dataset_name": "mock-ohlc",
  "dataset_version": "0.1.0",
  "rows_deleted": 42,
  "start": "2024-01-02T00:00:00",
  "end": "2024-01-30T00:00:00"
}

rows_deleted: number of rows removed (each row counts individually, so multi-row timestamps contribute more than one)
start / end: the actual range of deleted rows (min/max deleted timestamp), which may be narrower than the requested range

Semantics:

Atomic. The delete is a single DELETE ... RETURNING timestamp statement wrapped in a transaction (same pattern as insert_rows): either all matching rows are removed or none are.
Safe against concurrent builds. The service holds the dataset's per-(name, version) build lock (from service/locks.py) during the delete, so it cannot interleave with a concurrent build of the same dataset.
Dependents are not checked. Deleting from a dataset is allowed even if datasets that depend on it still have derived data in the range. Callers are responsible for cleaning up derived data if they care about staleness.
No calendar check. Whatever rows exist in the range are deleted; there is no 422 case.

Status codes:

Status	Meaning
`200`	Rows deleted; body reports the count and actual deleted range
`400`	Malformed input (invalid version or timestamp)
`401`	Missing or invalid API key
`404`	Dataset not found in the config registry
`404`	Dataset exists but has no rows in the requested range
`500`	Unexpected failure (DB error)

Dataset proposals endpoint

POST /datasets submits a new dataset for review. The server never writes to its own scripts directory: it validates the submission, generates the dataset files, and opens a GitHub pull request adding builders/scripts/<name>/<version>/ to the repo. The dataset goes live only after the PR is reviewed, merged, and the server restarts. This keeps code review as the trust gate — a builder script is code that executes on the server, and the "internal users write trusted builders" assumption (see "Build behavior") is enforced by human review, not by the API key alone.

Request body (JSON): name, version, calendar, granularity, start_date, schema (field → type map), builder_script, proposer identity (author_name, team, discord_user, description — all required, surfaced in the PR body), and optional dependencies (list of {name, version, lookback?}), env_vars, requirements_txt, env_template.

Validation (core/service/proposals.py), in order:

name must match ^[a-z0-9][a-z0-9_-]*$ (it becomes a directory and branch name); version must parse as SemVer; proposer fields must be non-empty
(name, version) must not already exist in the config registry
the canonical config.toml is generated, then those exact bytes are re-parsed and run through the same validate_config used at startup — the committed config is byte-identical to what was validated
registry cross-checks against live configs: dependencies exist, granularity is not finer than any dependency's, start-date is not before any dependency's (a new dataset is a leaf nothing references, so cycles are impossible)
builder script: AST-parsed, must define a top-level build(dependencies, timestamp) with exactly two positional args
the script is then run through ruff autofix + format server-side (same rule selection as repo CI — kept in sync manually with the root pyproject.toml); unfixable violations reject the proposal with ruff's output. Proposal PRs therefore land pre-linted and cannot fail the CI lint gate. The rejection message combines both of ruff's streams (violations go to stdout, but a ruff-internal failure such as a bad config reports only on stderr) and falls back to the exit code if ruff printed nothing, so the caller and the server log never get a bare "fails lint:" with no diagnostic

PR shape: branch add-dataset/<name>-<version> off main, one commit containing config.toml, builder.py, and (when provided) requirements.txt / .env.template. The real .env is never committed — for env-vars = true datasets the PR body carries a checklist item to place it on the server manually before the first build. Title: feat: add dataset <name>/<version>. Reviewers are auto-requested (best-effort: the PR author is excluded, and a failed batch falls back to per-reviewer requests).

Configuration (env vars, documented in infra/.env.template):

GITHUB_TOKEN (required for this endpoint only): PAT with Contents + Pull requests read/write on the repo. Without it the endpoint returns 502; everything else works.
GITHUB_REPO (default Wat-Street/datastream), GITHUB_API_URL (default https://api.github.com; overridable for testing against a fake), GITHUB_REVIEWERS (default Blackgaurd,Scr4tch587).

Response (200): {"dataset_name", "dataset_version", "pr_url", "branch"}.

Status	Meaning
`400`	Invalid submission (validation or lint failure; `detail` is safe to show in the UI)
`401`	Missing or invalid API key
`409`	Dataset already registered, or a proposal branch for it is already open
`422`	Malformed request body (missing/mistyped fields)
`502`	GitHub unreachable or `GITHUB_TOKEN` missing/invalid

The requesting API key's team label is logged and included in the PR body alongside the proposer fields. A CI test (tests/core/runtime/test_real_scripts.py) validates every real config in builders/scripts/ with the startup validation path, so green CI on a proposal PR means the dataset cannot break server boot.

Build behavior

On a POST /build request, it builds missing data for the requested range and writes it to the database.
It dynamically imports builder scripts to run them (see below).
- All builder scripts will be implemented by internal users, so we may trust that all code is safe.
- But we must not trust that builder scripts will not crash, so each builder invocation runs in an isolated subprocess.
- Each builder runs in its own subprocess.Popen process, communicating via JSON over stdin/stdout. A standalone worker script (isolated_worker.py) handles deserialization, builder import, and result serialization. It uses only stdlib so it works in any venv.
- If the subprocess crashes, the main process catches the failure cleanly without going down.
Builders never have direct access to the database -- all reads and writes are handled by the server. For now, this is enforced by convention. TODO: add a runtime guard to enforce this.
After builder scripts are run, we upload the data to the Postgres database (see below).
Builds use a scheduler/worker/orchestrator architecture (see below). The scheduler computes a topological build plan, the orchestrator executes it level by level, and the worker handles building a single dataset.
All data access during a build (reads and writes) goes through a Store abstraction (see "Store abstraction and dry-run builds" below), so the same build logic runs against Postgres for real builds and an in-memory store for dry runs.
The service is the only public-facing security boundary (auth, rate limiting, input validation).

Dry-run builds

POST /build accepts an optional dry-run query parameter (default false). When true, the entire build runs against an in-memory store and nothing is written to the database — the request rebuilds the whole dependency graph in isolation (the store starts empty, so it never reads real committed data) and returns the produced rows so the caller can validate builder logic.

Store selection happens at one boundary. build_dataset(dry_run=...) is the only place the flag is read: it constructs a PostgresStore (real build) or a fresh MemoryStore (dry run) and passes the ready-made store into run_build. Everything below (run_build → execute_job → _fetch_dep_data) takes a store and never sees the dry_run flag, so build logic is identical across real and dry runs.
No lock. A dry run uses a request-private MemoryStore, so it cannot corrupt real data and must not take the shared per-dataset build lock (which would block production builds). The lock is owned by the store: PostgresStore.build_lock returns the shared lock from service/locks.py; MemoryStore.build_lock returns a nullcontext.
No cleanup. The MemoryStore is garbage-collected when the request ends (even on crash). The real DB was never touched, so there is nothing to roll back.
Response. A real build returns {"status": "ok"}. A dry run returns an envelope with the produced rows for the requested dataset:
```
{
  "dataset_name": "mock-ohlc",
  "dataset_version": "0.1.0",
  "dry_run": true,
  "rows": [
    {"timestamp": "2024-01-02T00:00:00", "data": [{"ticker": "AAPL", "open": 100, "high": 150, "low": 90, "close": 130}]}
  ]
}
```
rows uses the same shape as GET /data (sorted by timestamp; empty list when nothing was produced). Builder and validation failures surface the same 400/422/500 semantics as a real build.

Build architecture: scheduler / worker / orchestrator

Build execution is split into three layers:

Scheduler (service/scheduler.py): computes a topological build plan
Worker (service/worker.py): executes a single build job (one dataset, one time range)
Orchestrator (service/orchestrator.py): coordinates level-by-level execution of the plan

The public entry point is build_dataset() in service/builder.py, which delegates to run_build() in the orchestrator.

Scheduler: dependency graph and topological ordering

The scheduler has two phases:

Graph collection (collect_graph()): DFS walk from the root dataset through its dependencies via registry.get_config(). For each node (name, version), it:

Validates end >= start_date, clamps start to start_date
Records the required [start, end] build range
Expands dependency ranges by lookback_subtract when applicable
Handles diamond dependencies by taking the union of required ranges. If a second visit widens a node's range, its subtree is re-walked so the expansion propagates to grandchildren. Since ranges only ever widen (never shrink), convergence is guaranteed for any DAG.

Returns a DependencyGraph containing:

ranges: dict[Node, tuple[datetime, datetime]] -- required range per node
edges: dict[Node, set[Node]] -- parent -> set of dependencies

Topological scheduling (schedule_build()): Kahn's algorithm (BFS-based topological sort) on the collected graph. Chosen over Tarjan's DFS-based sort because it naturally produces level-order grouping -- nodes are processed in waves by their distance from roots, which directly maps to the barrier model.

Algorithm:

Compute in_degree[node] = number of dependencies for each node
Seed level 0 with all zero-in-degree nodes (roots)
For each level: remove those nodes, decrement in-degrees of their dependents, collect newly zero-in-degree nodes as the next level
Convert each node + range into a JobDescriptor, grouped by level

Returns a BuildPlan where levels[0] = root datasets (build first), levels[-1] = the requested dataset (build last).

Data types (`service/models.py`)

@dataclass(frozen=True)
class JobDescriptor:
    dataset_name: str
    dataset_version: SemVer
    start: datetime
    end: datetime

@dataclass(frozen=True)
class JobResult:
    job: JobDescriptor
    success: bool
    error: str | None = None

@dataclass
class BuildPlan:
    levels: list[list[JobDescriptor]]

JobDescriptor is frozen and hashable for use as dict keys. JobResult is lightweight -- the worker handles its own DB insert, so no need to carry rows back.

Store abstraction and dry-run builds

The build path reads and writes data through a Store interface (service/store.py), an abc.ABC with four data methods plus a build_lock (matching the existing Calendar ABC convention, not a Protocol):

class Store(ABC):
    def get_existing_timestamps(self, name, version, start, end) -> list[datetime]: ...
    def get_rows_range(self, name, version, start, end) -> dict[datetime, list[dict]]: ...
    def get_rows_timestamps(self, name, version, timestamps) -> dict[datetime, list[dict]]: ...
    def insert_rows(self, name, version, rows) -> None: ...
    def build_lock(self, name, version) -> AbstractContextManager: ...

Two concrete backends:

PostgresStore (real builds): a thin shell forwarding each data method to the corresponding core.db.datasets function — same SQL, same behavior. build_lock returns the shared per-dataset lock from service/locks.py.
MemoryStore (dry runs): the same methods backed by a plain dict {(name, version): {timestamp: [rows]}} held in process. insert_rows appends (round-tripping each row through json.dumps/loads to mirror Postgres Jsonb serialization, so non-serializable builder output still fails); the read methods filter the dict by timestamp range. It never opens a DB connection, and build_lock returns a nullcontext.

The store is threaded through run_build → execute_job → _fetch_dep_data; the worker never calls core.db.datasets directly. build_dataset is the single boundary that reads the dry_run flag and constructs the appropriate store (see "Dry-run builds" above).

Worker: single-job execution

execute_job(job, cancelled, store) builds one dataset over one time range:

Generate valid calendar timestamps via generate_timestamps()
Acquire the build lock via store.build_lock(...) (a real lock for PostgresStore, a no-op for MemoryStore)
Check which timestamps already exist via store.get_existing_timestamps(...)
For each missing timestamp: fetch dependency data (store.get_rows_range / store.get_rows_timestamps), run builder subprocess, validate output against schema
Bulk-insert all rows on success via store.insert_rows(...) (atomicity: no partial inserts)
Check cancelled event between timestamps for early termination

store defaults to PostgresStore() when omitted. The worker does NOT handle dependency graph walking or start-date clamping -- those are the scheduler's responsibility.

Orchestrator: level-by-level execution

run_build(dataset_name, version, start, end, store):

Calls schedule_build() to get a BuildPlan
Iterates levels sequentially (barrier model: all level N jobs must complete before level N+1 starts)
Within each level, executes jobs sequentially via execute_job(), passing the store down (MVP single worker -- future: parallelize within levels)
On any job failure: sets cancelled event, raises RuntimeError

store defaults to PostgresStore() when omitted, so the orchestrator stays dry-run agnostic.

Commit model

Data is committed per-level. Each level's data is inserted to the store before the next level starts. Workers read dependency data back from the same store. For real builds the store is Postgres; for dry runs it is the request-private MemoryStore, so the dependency hand-off A → store → B works identically without touching the DB. If level N fails, levels 0 through N-1 remain committed (real builds) or simply retained in the discarded MemoryStore (dry runs).

Within each job, atomicity is preserved: rows are accumulated in memory and bulk-inserted only if all timestamps succeed. If any timestamp fails, no rows are inserted for that job.

Build error responses

Status	Meaning
`400`	Malformed input (invalid version format or unparseable timestamp)
`401`	Missing or invalid API key
`422`	Valid input but no valid calendar timestamps exist in the requested range (e.g. weekday-only dataset requested over a weekend)
`500`	Unexpected failure (config not found, builder crash, DB error)

Server architecture

The server code lives under builders/server/ and is organized into four layers:

builders/server/
├── main.py         # entrypoint: creates FastAPI app, mounts routers
├── log_config.py   # central structlog configuration
├── auth.py         # api-key hashing + verify_api_key bearer dependency
├── api/            # endpoint handlers using APIRouter
│   └── routes.py            # public_router (open /status) + router (authenticated)
├── service/        # build orchestration, scheduling, and execution
│   ├── builder.py        # public API: build_dataset(), get_data()
│   ├── orchestrator.py   # level-by-level plan execution via run_build()
│   ├── scheduler.py      # dependency graph collection + Kahn's algorithm
│   ├── worker.py         # single-job execution via execute_job()
│   ├── store.py          # Store ABC + PostgresStore + MemoryStore (dry-run backend)
│   ├── models.py         # JobDescriptor, JobResult, BuildPlan data types
│   ├── timestamps.py     # generate_timestamps(), NoValidTimestampsError
│   └── locks.py          # per-dataset threading.Lock registry
├── db/             # database connection management and queries
│   ├── connection.py
│   └── datasets.py
├── runtime/        # config loading, subprocess isolation, schema validation, venv management
│   ├── config.py
│   ├── registry.py         # startup preload + in-memory config registry
│   ├── isolated_worker.py  # standalone worker script (stdlib-only, runs in builder subprocesses)
│   ├── loader.py
│   ├── runner.py
│   ├── serialization.py    # JSON serialization for subprocess IPC
│   ├── validator.py
│   └── venv_management.py  # per-builder venv creation and caching
├── utils/          # shared utilities
│   ├── retry.py            # generic retry with exponential backoff
│   └── semver.py
└── tests/          # mirrors the layer structure
    ├── api/
    ├── service/
    ├── db/
    ├── runtime/
    └── utils/

main.py is the uvicorn entrypoint (main:app). It creates the FastAPI app, mounts routers from api/, and runs a lifespan handler that calls load_all_configs(SCRIPTS_DIR) and setup_builder_venvs() on startup. Dependencies flow strictly downward: api -> service -> db/runtime. No layer imports upward.

Scripts directory resolution: SCRIPTS_DIR in runtime/config.py defaults to a path relative to the source file (../scripts from the server package root). This works in Docker where scripts are volume-mounted at /app/scripts. For local dev, the scripts live at builders/scripts (a sibling of builders/server), so the SCRIPTS_DIR env var overrides the default. just backend-dev sets this automatically.

Logging

The server uses structlog for structured logging. Configuration lives in log_config.py and is called once at import time in main.py.

Processor pipeline: merge_contextvars -> add_log_level -> TimeStamper(iso) -> renderer. The renderer is ConsoleRenderer by default (human-readable) or JSONRenderer when LOG_FORMAT=json is set.

stdlib integration: stdlib logging is routed through structlog via ProcessorFormatter, so uvicorn logs flow through the same pipeline.

Request context: a FastAPI middleware in main.py clears contextvars per request and binds a unique request_id. The build endpoint also binds dataset_name and version to context.

What is logged:

main.py: api key count at startup (info)
auth.py: matched team label bound to the request context on each authenticated request
api/routes.py: build failures (exception)
service/scheduler.py: start-date clamping (warning)
service/orchestrator.py: build plan summary (info), level start/complete (info)
service/worker.py: skipped builds (info), build progress (info), insert counts (info)
db/connection.py: new connections (debug)
db/datasets.py: query execution (debug), rows inserted (info)
runtime/runner.py: subprocess start/complete (info), stderr output (warning), timeouts and crashes (error)
runtime/registry.py: config loaded during startup preload (debug)
runtime/loader.py: builder script imported (debug)
runtime/venv_management.py: venv creation progress (info), failures (exception)

CI workflow

Backend checks run in .github/workflows/backend-ci.yml.

The workflow triggers on:

builders/**
pyproject.toml
uv.lock
.github/workflows/backend-ci.yml

This keeps backend CI scoped to builder and shared Python dependency changes.

Benchmarking

End-to-end build benchmarks live under builders/server/benchmarks/. They measure wall-clock time for a full build request through the server (HTTP handler → dependency resolution → subprocess spawns → DB insert) using a testcontainer postgres so as not to pollute the real database.

Two profiling modes:

just bench — runs pytest-benchmark over a 90-day mock-ohlc build, 3 rounds, outputs a timing table (mean/stddev/min/max).
just bench-profile [DAYS] — wraps the standalone benchmarks/bench_build.py script with py-spy record --subprocesses, producing bench-flamegraph.svg. The --subprocesses flag captures the builder subprocess spawns, which dominate build time for simple datasets.

The benchmark test uses benchmark.pedantic(rounds=3, warmup_rounds=0) — explicit rounds because each round is expensive, no warmup because every round must do real work (the DB is truncated between rounds by the clean_db fixture).

The standalone __main__ script does the same testcontainer setup without pytest, so it can be wrapped directly by py-spy. Module patching is done via direct attribute assignment rather than monkeypatch.

The flame graph should reveal the breakdown between: subprocess spawn overhead (subprocess.Popen + interpreter startup), JSON serialization (stdin/stdout IPC), calendar/timestamp generation, and DB operations (get_existing_timestamps + insert_rows).

Containers

The service, Postgres database, and Caddy reverse proxy each run in their own Docker containers.
Containers communicate over a Docker-managed internal network. Only Caddy exposes host ports (80/443). Internal services (postgres, builder, pgweb) have no host port mappings in production.
builders/scripts/ is mounted as a volume into the builder container at runtime, so scripts can be updated without rebuilding the image.

Network topology

Internet --> [Caddy :80/:443] --internal--> [builder:3000]
                                            [postgres:5432]  (no host port)
                                            [pgweb:8080]     (no host port)

Caddy terminates TLS and reverse proxies to the builder over the Docker bridge network. Internal traffic stays plain HTTP.

Caddy reverse proxy

Caddy provides automatic HTTPS via Let's Encrypt with minimal configuration. The domain is set via the DOMAIN env var in infra/.env.

Production: set DOMAIN=datastream.yourdomain.com, Caddy auto-provisions a Let's Encrypt certificate
Local Docker: DOMAIN=localhost uses Caddy's internal CA (self-signed), or DOMAIN=:80 for plain HTTP
Local dev (just backend-dev): unaffected, hits uvicorn on localhost:3000 directly via the dev overlay

Dev overlay

infra/docker-compose.dev.yml re-exposes internal service ports for local development:

postgres:5432, builder:3000, pgweb:8080

Use just docker-up-dev to start with the overlay, or just docker-up for production mode.

Infra directory layout

infra/
  docker-compose.yml      # production compose (only Caddy ports exposed)
  docker-compose.dev.yml  # dev overlay (re-exposes internal ports)
  .env                    # environment variables (DB credentials, DOMAIN, etc.)
  builder/
    Dockerfile
  caddy/
    Caddyfile             # Caddy reverse proxy config
  postgres/
    Dockerfile
    init.sql

Dataset and builder model

The two main components of the service are datasets and builders.

Datasets define the data, they are typically series of (timestamp, data) pairs.
- Datasets are uniquely identified by a (name, version) pair, where name is a string and version is a SemVer (e.g. 1.1.12).
- Datasets may have dependencies on other datasets. For example, a 5 day moving average may depend on daily pricing data.
- Datasets with no dependencies are called "root datasets".
Builders are the logic to create more data. These will be Python scripts that can be called by the main process.

Start date

Each dataset declares a start-date in config.toml (format: YYYY-MM-DD), which defines the earliest date data can be built for. At build time:

If the request's end is before start-date, a ValueError is raised.
If the request's start is before start-date, start is clamped to start-date with a warning log.

The start-date field is required and validated when loading the config.

Overwrite policy

When a build is triggered for a time range, the builder server first queries the DB for all distinct timestamps in that range that already have any rows for the given dataset.
If any rows exist for a timestamp, it is considered fully built and skipped.
Only missing timestamps are built and inserted — existing rows are never overwritten.
This avoids re-running builders unnecessarily; DB writes are plain INSERT (no upsert needed).

Atomicity

Builds are atomic at the per-job level. Each job (one dataset, one time range) accumulates rows in memory and bulk-inserts only if all timestamps succeed. If any timestamp fails, no rows are inserted for that job.

Data is committed per-level: all jobs in level N insert before level N+1 starts. If a job in level N fails, levels 0 through N-1 remain committed.

Build concurrency

The service runs a single uvicorn worker. FastAPI dispatches sync endpoint handlers to a thread pool, so concurrent requests for the same dataset can race between the "check what's missing" DB read and the "insert rows" DB write, producing duplicate rows.

Per-dataset locking: a threading.Lock per (dataset_name, dataset_version) pair serializes the critical section of execute_job() in the worker (steps: check existing timestamps, compute missing, build, validate, insert). Different datasets build concurrently; the same dataset serializes. Locks are created lazily and stored in a module-level registry (service/locks.py).

Lock scope in execute_job():

Generate valid timestamps (outside the lock, deterministic)
Acquire lock
Check existing timestamps in DB, compute missing, build each missing timestamp, validate + insert atomically
Release lock

The scheduler and orchestrator run outside the lock. Each dataset only holds its own lock during its own build phase. This minimizes lock hold time and avoids unnecessary serialization of the dependency tree.

Deadlock safety: the dependency graph is validated as acyclic at startup by registry.load_all_configs(). Since the orchestrator executes levels sequentially (dependencies before dependents), lock ordering follows the topological sort and cannot deadlock.

Scaling assumption: this design assumes a single uvicorn worker (single process). Multi-worker deployments would require Postgres advisory locks or a similar distributed locking mechanism.

Datasets postgres schema

Datasets will be stored in a Postgres database. The table used to store datasets is datasets. Each row contains some timeseries data along with some metadata. It has the following columns:

Metadata columns:

id: primary key, automatically increasing,
created_at: a timestamp, accurate to the microsecond, for when the current row was created at,
dataset_name: a string, the dataset name to which the current row's data belongs to,
dataset_version: a string, the stringified semver of the current row's dataset (validated at the application level, no DB constraint),

The combination of (dataset_name, dataset_version, timestamp) is not unique — multiple rows can share the same triple (e.g. multiple tickers at the same timestamp). A non-unique index on (dataset_name, dataset_version, timestamp) keeps range queries fast. TODO: partition the table by dataset_name or time range for performance at scale.

Data columns:

timestamp: a timestamp, what timestamp this row represents,
data: a JSONB, key-value pairs for the current row's data,
- the JSONB must not be nested for more than one level

Here is an example of what the decoded JSON in the data column is:

{
    "ticker": "AAPL",
    "open": 123,
    "high": 456,
    "low": 100,
    "close": 200
}

Builders layout

Builders are stateless Python scripts. To each dataset there is a builder script, and that script will build only that dataset. Builders are stored under the builders/ directory. Each builder is minimally a builder.py and a config.toml.

The [schema] section in config.toml is used for runtime validation:

After a builder returns its output list, the builder server validates each dict in the list against the schema before inserting into the DB.
Validation checks that all declared keys are present and that values match the declared types.
Validation correctness is the priority over performance.
The builder script for dataset (dataset_name, dataset_version) is under builders/scripts/dataset_name/dataset_version/builder.py. The config is stored under builders/scripts/dataset_name/dataset_version/config.toml.

Here is an example builder.py (subject to change):

from datetime import datetime
from typing import Any

def build(dependencies: dict[str, dict[datetime, list[dict]]], timestamp: datetime) -> list[dict[str, Any]]:
    return [{"ticker": "AAPL", "price": 123}]

Type notes:

timestamp: a datetime.datetime with microsecond precision.
dependencies: maps each dependency's name (not name+version) to a dict keyed by timestamp, where each value is a list of data dicts. For deps without lookback, the dict contains only the current timestamp. For deps with lookback, the dict contains all timestamps in the lookback window [T - lookback + step, T] (N points inclusive, where step is one unit of the lookback duration). Versions are resolved by the builder server using config.toml, so builder scripts never need to reference them directly.
Return value: a list of dicts, where each dict is one row to insert. Single-row datasets return a list of length 1.

And here is an example config.toml (subject to change):

name = "dataset name"
version = "0.0.1"
builder = "builder.py"  # not strictly necessary, here just in case
calendar = "NYSE"  # defines the valid timestamps for this dataset
start-date = "2020-01-01"  # earliest date data can be built for (YYYY-MM-DD)
env-vars = true  # optional, default false; loads .env file into builder subprocess

[schema]
ticker = "str"
price = "int"

[dependencies]
dependency-a = "0.0.2"  # simple: version only (no lookback)
dependency-b = { version = "0.0.1", lookback = "5d" }  # with lookback window

There may be other Python files in the same directory or relative sub-directories, and they will be imported using the Python module system.

Builder environment variables

Builder scripts may need secrets or config (API keys, credentials) passed via environment variables. This is supported through the env-vars field in config.toml.

Config field: env-vars is an optional boolean (default false). When true, the builder server loads a .env file and injects its variables into the builder subprocess.

.env location: The .env file must be in the same directory as the builder script: builders/scripts/<dataset_name>/<version>/.env. This file is not committed to git (see builders/scripts/.gitignore).

.env.template convention: By convention, committed .env.template files document which variables a builder needs. The server does not read these; they exist purely as documentation for humans.

Runtime behavior:

The .env file is validated at build time, not config load time. This means CI can load configs for datasets with env-vars = true without needing the actual .env file present.
If env-vars is true and the .env file is missing at build time, a FileNotFoundError is raised.
The main server process never reads the .env values. The .env file is parsed inside the subprocess only (using a minimal stdlib-only parser in isolated_worker.py), so secrets never enter the parent process memory.
Environment variables are scoped to the subprocess and do not leak to the parent.

Per-builder virtual environments

Builder scripts may need external libraries (pandas, numpy, API clients, etc.) that differ between builders. Each builder can declare its own dependencies via a requirements.txt file in its directory.

Dependency specification: a standard requirements.txt in the builder directory (builders/scripts/<name>/<version>/requirements.txt).

Venv location: .venv/ inside each builder's directory, gitignored via builders/scripts/.gitignore.

Install timing: eager on server startup. The FastAPI lifespan handler calls setup_builder_venvs() which scans all builder directories and creates venvs for any that have a requirements.txt.

Caching: venv creation is skipped if .venv/.requirements_hash (a crc32 hash of requirements.txt) matches the current file. Changing requirements.txt triggers a rebuild on next startup.

Tooling: uv is used for venv creation (uv venv) and package installation (uv pip install).

Builders without requirements.txt: no venv is created, no overhead. The runner uses the system Python (sys.executable) for these builders.

Venv detection: at build time, runner.py checks if script_dir/.venv/bin/python exists. If so, it uses that interpreter for the subprocess; otherwise it falls back to sys.executable.

Error handling: if one builder's venv creation fails, it logs a warning but does not block other builders from being set up.

Updated builder directory layout:

builders/scripts/<dataset_name>/<version>/
  builder.py            # builder script (required)
  config.toml           # dataset config (required)
  requirements.txt      # python dependencies (optional)
  .env                  # environment variables (optional, gitignored)
  .env.template         # documents required env vars (optional)
  .venv/                # per-builder venv (auto-created, gitignored)

Subprocess execution model

Builder subprocesses use subprocess.Popen with JSON over stdin/stdout for IPC:

The runner serializes builder inputs (dependencies, timestamp, paths, env file) to JSON via serialization.py
The subprocess runs isolated_worker.py using the builder's venv python (or system python)
isolated_worker.py is stdlib-only: it deserializes input, imports and runs the builder, serializes output
The runner deserializes the JSON response from stdout

This model supports per-builder venvs since each subprocess uses its own Python interpreter. The worker script has no dependencies on server code.

Retry behavior

Builder subprocess execution is wrapped in automatic retry with exponential backoff via a generic retry_with_backoff() utility in utils/retry.py.

Constants (defined in runtime/runner.py):

RETRY_MAX_RETRIES = 5
RETRY_INITIAL_DELAY = 2.0 seconds
RETRY_BACKOFF_FACTOR = 2.0

Delay progression: 2s, 4s, 8s, 16s, 32s (total worst-case wait: ~62s + subprocess execution time).

What is retried: all subprocess failures, including timeouts (TimeoutExpired), crashes (non-zero exit with no stdout), and worker errors (WorkerError from the isolated worker). The entire subprocess execution is retried from scratch on each attempt.

What is NOT retried: payload serialization (deterministic, runs once before the retry loop), schema validation (runs after run_builder() returns, in the caller), and dependency resolution (runs before run_builder() is called).

Logging: each retry attempt logs a structlog warning with the attempt number, delay, and error message.

Integration test override: integration tests monkeypatch retry constants in tests/integration/conftest.py to keep runtime fast while preserving retry semantics:

RETRY_MAX_RETRIES = 3
RETRY_INITIAL_DELAY = 0.01
RETRY_BACKOFF_FACTOR = 2.0

This keeps integration coverage for transient-recovery and retry-exhaustion behavior without incurring production-scale backoff delays.

Mock builders

The following mock builders exist for testing and development:

mock-ohlc/0.1.0: root dataset, generates single-row OHLC data for AAPL per timestamp
mock-daily-close/0.1.0: depends on mock-ohlc, extracts the close price (single row)
mock-multi-ohlc/0.1.0: root dataset, generates multi-row OHLC data for AAPL, MSFT, GOOG per timestamp
mock-multi-close/0.1.0: depends on mock-multi-ohlc, extracts close prices for each ticker (multi-row)
mock-moving-avg/0.1.0: depends on mock-daily-close with lookback = "5d", computes 5-day moving average of close prices

Lookback dependencies

Certain datasets depend on a time window of historical data from their dependencies. For example, a moving average needs the last 5 days of close prices.

Config format: Dependencies support two formats:

Simple: dep = "0.1.0" (no lookback, builder receives only the current timestamp's data)
Table with lookback: dep = {version = "0.1.0", lookback = "5d"}

Lookback is a duration string using these units: "5d" (days), "24h" (hours), "30m" (minutes), "60s" (seconds). Must be a positive value. After parsing, all dependencies are normalized to {"version": str, "lookback_subtract": timedelta | None}. lookback_subtract is pre-computed as amount - 1 units (e.g. timedelta(days=4) for "5d").

Semantics: Lookback defines an inclusive window of N points. lookback = "5d" means "fetch 5 days of dependency data in [T - 4d, T]" (5 days inclusive). The window start is computed as T - lookback_subtract.

Data format: Builders always receive dict[str, dict[datetime, list[dict]]] — dependency data keyed by name, then by timestamp, then a list of rows. This applies regardless of whether lookback is set.

Build range expansion: When building dependencies recursively, the builder server expands the dependency's build start date using dep_start = start - lookback_subtract so historical data covers the correct inclusive window.

Edge case: Near a dependency's start-date, the lookback window may return fewer data points than usual. Builder scripts are responsible for handling short windows gracefully.

Example config:

name = "dataset name"
version = "0.0.1"
builder = "builder.py"
granularity = "1d"
start-date = "2020-01-01"

[schema]
ticker = "str"
average = "float"

[dependencies]
mock-daily-close = { version = "0.0.1", lookback = "5d" }

Timestamp granularity

Timestamps are stored with microsecond precision (using pandas.Timestamp). In practice, the finest granularity used will be per-second.

Each dataset has a declared granularity (e.g. "1s", "1m", "1d"), which will be a field in config.toml. A dataset may only depend on another dataset whose granularity is finer or equal to its own. For example, a daily dataset may depend on per-second data or another daily dataset, but an hourly dataset cannot depend on a daily one. This constraint is enforced at build time by validate_dependency_graph() before any data is built.

Calendars

Calendars determine which timestamps are valid for a dataset. Each dataset must declare a calendar in config.toml. At build time, generate_timestamps() filters candidate timestamps through the calendar's is_open() method, so only valid dates are built and stored.

Calendar interface

The Calendar ABC lives in builders/server/calendars/interface.py:

name: str — unique identifier (abstract property)
granularity: timedelta — smallest time step (abstract property)
is_open(timestamp: datetime) -> bool — whether a timestamp is valid (abstract method)
next_open(timestamp: datetime) -> datetime | None — returns the next valid datetime >= timestamp, or None if never open again (abstract method)

Calendars are lightweight structures that are allowed to maintain state but should be minimal.

Available calendars

All calendars are registered in CALENDARS_MAP (a dict[str, Calendar]) in builders/server/calendars/registry.py:

everyday — every day is valid (is_open always returns True). This is the default calendar.
weekday — Monday through Friday are valid, Saturday and Sunday are not.
nyse-daily — NYSE trading days only (excludes weekends and all NYSE holidays). Uses exchange_calendars library with the XNYS exchange calendar.

Integration with timestamp generation

generate_timestamps() in service/timestamps.py accepts a Calendar. Each candidate timestamp is checked via calendar.is_open() and excluded if not open. The worker passes cfg.calendar automatically when building each job.

Config integration

DatasetConfig.calendar is of type Calendar (not a string).
During config loading, the calendar string from config.toml is validated against CALENDARS_MAP and resolved to a Calendar instance.
The calendar field is required. Missing it raises a ValueError during config validation.
Unknown calendar names raise a ValueError during config validation.

Layout

builders/server/calendars/
├── __init__.py
├── interface.py      # Calendar ABC
├── utils.py          # utility functions
├── definitions/      # concrete calendar class implementations
└── registry.py       # CALENDARS_MAP registry

Future work

Cross-dataset dependency lookups may use as-of semantics (nearest prior valid timestamp) to handle calendar mismatches.
Additional calendars (e.g. NYSE trading days) can be added by subclassing Calendar and registering in CALENDARS_MAP.

FilesExpand file tree

SPEC-backend.md

Latest commit

History

SPEC-backend.md

File metadata and controls

Datastream Backend

Service

Public endpoints

API authentication

CORS

Datasets endpoint

Data endpoint

Delete endpoint

Dataset proposals endpoint

Build behavior

Dry-run builds

Build architecture: scheduler / worker / orchestrator

Scheduler: dependency graph and topological ordering

Data types (service/models.py)

Store abstraction and dry-run builds

Worker: single-job execution

Orchestrator: level-by-level execution

Commit model

Build error responses

Server architecture

Logging

CI workflow

Benchmarking

Containers

Network topology

Caddy reverse proxy

Dev overlay

Infra directory layout

Dataset and builder model

Start date

Overwrite policy

Atomicity

Build concurrency

Datasets postgres schema

Builders layout

Builder environment variables

Per-builder virtual environments

Subprocess execution model

Retry behavior

Mock builders

Lookback dependencies

Timestamp granularity

Calendars

Calendar interface

Available calendars

Integration with timestamp generation

Config integration

Layout

Future work

Data types (`service/models.py`)