PyCatalyst

Make the data, grade the data, train the model, ship the artifact.

PyCatalyst is a configuration-driven toolkit for synthetic data and model training. Three artifacts, all YAML, all versioned:

A REST API, an optional web UI, schema inference, and substitutable storage (SQLite, Postgres, MongoDB) sit around that core.

Sixty seconds

pip install pycatalyst
pycatalyst recipe new orders        # a starter recipe you can edit
pycatalyst generate -r orders.yaml -n 5

{
  "id": "a31c06bd-463e-4923-bc1a-adbde48b1697",
  "name": "Bruno Barros",
  "email": "ben.chen518@example.com",
  "quantity": 72,
  "status": "pending",
  "created_at": "2024-08-20T01:46:55+00:00"
}

The same recipe also checks data, so fixtures and ingest validation cannot drift apart:

import pycatalyst

built = pycatalyst.build_recipe(pycatalyst.parse_recipe_file("orders.yaml"))
violations = pycatalyst.validate_records(incoming_rows, built.schema)

Start from real data instead, and get a recipe that captures its statistical character (distributions, categories and their frequencies, formats, and the correlations between columns) without copying rare values into it:

pycatalyst infer -i customers.csv -o customers.yaml
pycatalyst recipe validate customers.yaml --data customers.csv
pycatalyst generate -r customers.yaml -n 100000 -o synthetic.csv --manifest synthetic.json

Fields can be correlated (correlations: {age: 0.6}) so a model has something to learn, and recipes can reference each other (references: customers.id) so generated tables join:

pycatalyst generate -r customers.yaml -r orders.yaml -o out/ --format csv

A field can be computed from the others, and from the record before it, so totals add up, balances run and ids stay unique:

- {name: total, type: float, formula: "round(quantity * unit_price, 2)"}
- {name: balance, type: float, formula: "round(prev(balance, 1000) + total, 2)"}
- {name: id, type: string, formula: "'ORD-' + str(row() + 1)"}

A sink can retry a batch it could not deliver and keep what still fails (delivery: {attempts: 4, dead_letter: failed.jsonl}).

Documentation

• Published (GitHub Pages): optophi.github.io/pycatalyst — start with Getting Started → Quick Start.

• Local docs (from repo root):

  pip install "pycatalyst[docs]"
  pycatalyst docs serve
  

  Opens at http://127.0.0.1:5005 by default. Static site: pycatalyst docs build → site/.

• Live API (when the server is running): http://127.0.0.1:8005/docs (Swagger).

Tutorial index in the doc site: Quick Start, API authentication, generate, recipes, infer schema, workbench, and the distillation reference.

Installation

pip install pycatalyst

API server, DB, and common extras:

pip install "pycatalyst[api,db,inference]"

From source (editable):

git clone https://github.com/optophi/pycatalyst
cd pycatalyst
pip install -e ".[dev]"

Quick Start (CLI)

pycatalyst recipe new orders                     # scaffold a recipe
pycatalyst recipe validate orders.yaml           # check it without running it
pycatalyst generate -r orders.yaml -n 100        # generate
pycatalyst generate -r orders.yaml --format ndjson | jq .
pycatalyst api                                   # serve on 127.0.0.1:8005

Every command takes -v/--verbose, -q/--quiet, --log-level and --debug, and --help carries worked examples.

See the Quick Start page in the docs for database setup, UI, and REST examples.

Training and distillation

pycatalyst train   -c experiment.yaml            # sklearn / PyTorch / HuggingFace
pycatalyst distill -c distill.yaml --dry-run     # validate; spend nothing
pycatalyst distill -c distill.yaml --stages generate
pycatalyst distill -c distill.yaml               # generate, train, bench, export

Distillation declares its data contract once, and that one declaration renders the teacher prompt, grades the completion, splits the datasets and scores the bench — so the prompt and the grader cannot drift apart. Stages are separable because their costs differ by orders of magnitude, and --dry-run plus a cost estimate come before anything is spent. See Distillation.

Streaming mock data

Python — same sequence as batch generation for a fixed seed:

from pycatalyst import GenerationEngine, SchemaBuilder

engine = GenerationEngine()
schema = SchemaBuilder("events").add_uuid("id").add_string("kind").build()
for row in engine.iter_records(schema, seed=123, limit=1000):
    process(row)

HTTP (Server-Sent Events) — POST /api/v1/stream/sse with the same field list as /generate, plus max_records and optional interval_ms. Use curl -N and a Bearer token when auth is enabled:

curl -N -H "Authorization: Bearer YOUR_JWT" -H "Content-Type: application/json" \
  -d '{"name":"x","fields":[{"name":"n","type":"int"}],"max_records":5,"seed":1}' \
  http://127.0.0.1:8005/api/v1/stream/sse

CLI — NDJSON to stdout (one JSON object per line):

pycatalyst stream ndjson --schema my.json -n 500 --seed 1

CLI — Kafka (pip install pycatalyst[kafka]):

export PYCATALYST_STREAM_KAFKA_BOOTSTRAP=localhost:9092
export PYCATALYST_STREAM_KAFKA_TOPIC=test-topic
pycatalyst stream kafka --schema my.json -n 2000 --batch 100

Server caps (optional): PYCATALYST_STREAM_MAX_RECORDS, PYCATALYST_STREAM_MAX_INTERVAL_MS.

Development

• Lint & format: ruff check . && ruff format .
• Type check: mypy src/
• Tests: pytest
• Coverage: pytest --cov=pycatalyst --cov-report=term-missing

Publishing

1. Bump version in pyproject.toml and CHANGELOG.md.
2. Create a release tag: git tag v0.1.0 && git push origin v0.1.0.
3. The GitHub Action uses PyPI Trusted Publishing; configure the publisher on PyPI for this repo, then the workflow will publish on tag push.

License

MIT