Trading Engine

Run algorithmic trading strategies as supervised tasks, with typed dependencies between them, a live dashboard, per-task logs and restart policies. Venue access goes through the Tribulnation SDK gateway.

A process is a class you write; a task is one running instance of it, keyed by process/id (for example maker/btc). Tasks declare what they require from each other, and the engine starts, waits for, stops and restarts them accordingly.

The normative behaviour is specified in specs/; this README is the practical guide and yields to the specs where they disagree.

Install

pip install tribulnation-engine

tribulnation-engine 0.2.0 requires Python 3.11+, tribulnation-sdk>=2.9.0 (which contains the gateway) and tribulnation-cli. The CLI provides the tn launcher; the engine registers its commands under tn engine, and the SDK registers tn gateway:

tn --help          # engine  Provided by tribulnation-engine
                   # gateway Provided by tribulnation-sdk
tn engine --help

tn only finds commands installed in its own Python environment. Use the tn from the environment the engine is installed in (e.g. .venv/bin/tn), or add the engine to a global one: uv tool install tribulnation-cli --with tribulnation-engine.

Quickstart

tn engine init                 # engine.toml, .env and strats/depth_listener.py
# fill in credentials in .env
tn engine start -v             # gateway + manager, dashboard on http://localhost:3121
tn engine task start depth-listener btc -p market=hl::BTC
tn engine task stop depth-listener btc

Writing a process

# strats/hedger.py
from dataclasses import dataclass

from tribulnation.engine import (
  BaseParams,
  Process,
  RestartLimit,
  RestartPolicy,
  State,
  StopReason,
  Stream,
)


class HedgerParams(BaseParams):
  market: str = 'hl::BTC'


@dataclass(frozen=True)
class Fill:
  size: float
  price: float


class Hedger(Process[HedgerParams]):
  """Hedges fills on one market."""

  Params = HedgerParams
  restart = RestartPolicy(
    on='error', delay=5.0, limit=RestartLimit(max_restarts=5, within=60)
  )
  connected = State[bool](False)
  position = State[float](0.0)
  fills = Stream[Fill]()

  async def run(self, p: Params):
    market = await self.sdk.market(p.market)
    async with market.depth_stream(queue_size=1, overflow='latest') as books:
      self.connected.set(True)
      async for book in books:
        ...

  async def shutdown(self, reason: StopReason):
    self.log.info('stopping: %s', reason)

  @Process.rpc
  async def hedge(self, size: float) -> float:
    """Hedge `size` and return the new position."""
    self.position.set(self.position() - size)
    self.fills.publish(Fill(size=size, price=0.0))
    return self.position()

1. Params. Declare a pydantic model subclassing BaseParams outside the class, and parameterize the base with it: Process[HedgerParams]. The Params = HedgerParams alias is optional; it allows Hedger.Params(...) and p: Params annotations. A process without params uses Process[BaseParams]. The params' JSON schema becomes the start form in the dashboard; raw params from the CLI or API are validated against it.

2. Name. Defaults to the lowercased class name (hedger). Override with class Maker(Process[MakerParams], name='dydx-maker').

3. run(params) runs one attempt. A fresh instance is created per attempt, so run must reconcile with the venue rather than assume a clean state. Instances are constructed with no arguments: the runtime sets self.sdk, self.log and the task binding, then calls __init__(). A process may be a @dataclass (every field with a default, or field(init=False)) or not, and may use mixins with their own __init__; one that needs arguments is rejected when it's registered. Returning normally ends the attempt with reason requested.

4. shutdown(reason) runs once after every attempt, before any restart, shielded from cancellation and bounded by shutdown_timeout (10 s by default, a class attribute). reason is one of requested (explicit stop, or run returned), error, dependency (a requirement stopped holding) or shutdown (the engine is stopping).

5. restart = RestartPolicy(...): on='never' (default), 'error' or 'always'; delay seconds between attempts; max_attempts total attempts per explicit start; limit=RestartLimit(max_restarts=N, within=T) at most N automatic restarts in any rolling T seconds. An explicit start or restart resets both budgets.

6. State[T](default) is a value that always has a current one. The owner reads it with self.position() and writes it with self.position.set(v). States reset to their default at the start of every attempt. The dashboard shows each live task's States as they change; pass ui=False to hide a large or noisy one: book = State[dict]({}, ui=False). A Derived value is computed from the States (or Derived values) its function reads, and recomputed synchronously on every set() of one. Its self is a read-only view exposing only States and Derived values; reads of anything else, like the clock, aren't tracked. It's read like a read-only State, and its age is its oldest dependency's:

   @Process.derived  # or @Process.derived(ui=False)
   def free(self) -> float:
     return self.available() - self.reserved()
   

7. Stream[T]() carries discrete events: self.fills.publish(event) delivers to current subscribers without blocking. Nothing is replayed.

8. @Process.rpc exposes a method to dependents. The owner can still call it directly as a normal method.

9. Injected attributes: self.sdk (the SDK's market interface, backed by the gateway in deployments), self.log (a logger captured into the task's log), self.tasks (the control plane, see below) and self.id (the task's id, e.g. the account a keyed service like collateral-manager/dydx serves).

Requirements and handles

A process uses another task's RPCs, States and Streams only through a declared requirement:

# strats/maker.py
from tribulnation.engine import BaseParams, Handle, Process

from .hedger import Hedger


class MakerParams(BaseParams):
  market: str
  hedger: str = 'btc'


class Maker(Process[MakerParams]):
  Params = MakerParams

  @Handle.require(Hedger, when=Hedger.connected)
  def hedger(self, p: Params) -> str:
    return p.hedger

  async def run(self, p: Params):
    position = self.hedger.get(Hedger.position, max_age=60)
    new_position = await self.hedger.call(Hedger.hedge, 0.1)
    fills = self.hedger.subscribe(Hedger.fills, queue_size=100, overflow='latest')
    async with fills as events:
      async for fill in events:
        self.log.info('hedged %s @ %s', fill.size, fill.price)

1. @Handle.require(Target, when=...) decorates a key function of the dependent's params that returns exactly one target id, or the id and the params to auto-start the target with (return p.hedger, HedgerParams(market=p.market)). It must only read params. Require a fixed number of tasks with separate declarations.
2. when is optional: a State[bool] or Derived[bool] of the target (Hedger.connected) or a synchronous, side-effect-free predicate Handle[Hedger] -> bool. A predicate that raises counts as not holding. A requirement holds while the target task is running, in the same attempt the handle was bound to, and when is true.
3. Handle use, via the decorated attribute (self.hedger):
   1. await handle.call(Target.rpc, *args): within the RPC's timeout (5 s by default; Target.rpc.with_timeout(10) for one call), raising TimeoutError past it.
   2. handle.get(Target.state, max_age=None): the current value, raising Stale if it's older than max_age seconds.
   3. handle.subscribe(Target.stream, queue_size=1000, overflow='fail'): an async context manager yielding the events. With 'fail', a subscriber more than queue_size events behind gets Overflow after its buffered events; 'latest' drops the oldest queued event instead.
   4. handle.watch(Target.state): an async context manager yielding a State's or Derived value's current value, then its value after each owner update. Latest-only: a slow consumer skips to the newest value. Use it to react to a State without polling get.
   5. handle.id, handle.status and handle.available.
4. Optional requirements, @Handle.require(Target, when=..., optional=True): planned and auto-started like any requirement, but not a precondition, and the target failing never stops the dependent. While the target isn't running with when true, handle.available is false and call, get, subscribe and watch raise Unavailable (in-flight calls, subscriptions and watches fail with it too); the handle follows the target to its next attempt. Check available, or catch Unavailable, at each use; there is no waiting API.
5. Requirements on a set of tasks, @Handle.require_all(Target, when=..., optional=...): the key function returns a set of ids, started with defaults (return set(p.hedges)), or a mapping from id to auto-start params (return {m: BookParams(market=m) for m in p.hedges}). Each id is a requirement of its own with the same when and optional. The attribute is a read-only Mapping[str, Handle[Target]] from id to handle, in no particular order (for market, book in self.books.items(): ...). An empty result requires nothing. Losing one hard entry stops the dependent like any hard requirement.
6. Only members declared on the target class type-check (Handle[Hedger]).
7. Requirement cycles between processes are rejected when a process is registered (DependencyCycle), require_all included.

Control plane

self.tasks starts, stops and inspects other tasks; actions are attributed to the calling task in the logs and the dashboard:

await self.tasks.start(Maker, 'eth', Maker.Params(market='hl::ETH', hedger='eth'))
await self.tasks.restart(Maker, 'eth')     # current params, or pass new ones
await self.tasks.stop(Maker, 'eth')        # stops its dependents first; waits for cleanup
info = self.tasks.get(Maker, 'eth')        # TaskInfo | None
everything = self.tasks.list()

start raises AlreadyRunning, UnmetDependency, ParamsConflict or DependencyCycle, all subclasses of ControlError. stop first stops every task that requires the task, dependents first, attributed like stop maker/eth by api (requires hedger/eth); restart doesn't, so dependents see dependency loss and recover. get and list give lifecycle snapshots (TaskInfo: status, attempt, stop_reason, error, ...); they grant no IPC access.

Logging

Records logged inside a task are captured into that task's log, tagged with the attempt, and streamed to the dashboard. Prefer self.log: the runtime sets it, and tribulnation.engine, to INFO. Other loggers keep their own levels.

Statuses and lifecycle

1. Waiting. Each attempt waits in waiting until its requirements hold. A wait longer than waiting_timeout (30 s by default) ends the attempt as an error, and the restart policy decides what follows.
2. Dependency loss. When a bound requirement stops holding, the dependent's run is cancelled and shutdown('dependency') runs. A fresh attempt with the same params then waits for its requirements again, independently of the restart policy and without consuming its budgets. Handles never switch to a new target attempt.
3. Auto-start. Starting a task starts the required tasks that aren't active, with the params its requirements ask for, or else with defaults alone if they're valid. Otherwise the start fails with UnmetDependency, naming every missing task. Asking for other params than an active task has, or two requirements asking for one key with different params, fails with ParamsConflict; equal params share the task. Every attempt plans again, so a required task that ended (e.g. past its restart budget) is started again by its dependents; a failed plan fails the attempt as an error.
4. Auto-stop. An auto-started task that nothing requires any more is stopped after a grace period: the process's unneeded_grace class variable (the runtime's 30 s default if unset), and at least the dependents' restart delay. Explicitly started tasks are never auto-stopped.
5. Explicit stops cascade. Stopping a task first stops, with reason 'requested', every task that requires it, directly or transitively, dependents first. Crashes, restarts, auto-stop and engine shutdown don't cascade.
6. Engine shutdown stops dependents before the tasks they require.

Running in-process

Runtime runs processes as tasks in the current event loop, against any SDK implementation; no gateway, manager or database is needed:

import asyncio

from tribulnation.engine import Runtime
from tribulnation.sdk.gateway import ProxySDK

from strats.maker import Maker


async def main():
  async with ProxySDK.at('unix:///tmp/engine-gateway.sock') as sdk:
    async with Runtime(sdk, waiting_timeout=30, auto_stop_grace=30) as runtime:
      await runtime.start(Maker, 'btc', Maker.Params(market='hl::BTC'))
      await runtime.wait()


asyncio.run(main())

Runtime has the same control methods as self.tasks (start, stop, restart, get, list), plus register, wait and shutdown. start also accepts raw params as a mapping. Leaving the async with block stops every task.

Deployment

engine.toml

[engine]
processes = ["strats.hedger:Hedger", "strats.maker:Maker"]

[daemon]
host   = "localhost"
port   = 3121
socket = "/tmp/engine-gateway.sock"   # gateway Unix socket
db     = "engine.db"                  # SQLite store

[daemon.logs]
default_limit = 500                   # rows per logs API page by default

[daemon.logs.retention]               # stored rows per task and level
info = 20000

[accounts.hl]                         # every venue used needs an account
venue       = "hyperliquid"
address     = "$HL_ADDRESS"           # or `public = true`, for market data only
private_key = "$HL_PRIVATE_KEY"

[[tasks]]
process = "maker"
id      = "btc"
args    = { market = "hl::BTC" }

1. [engine] processes are module:Class[,Class...] or path/to/module.py:Class targets, imported at startup. Required processes are registered with them.
2. Account values starting with $ are read from the environment; commands load .env from the directory of engine.toml.
3. [[tasks]] are started at startup, attributed to config. args holds the task's params.

Commands

tn engine start                       # managed: launches gateway + manager
tn engine start -d                    # background; writes .engine.pid and engine.log
tn engine stop                        # stop a background engine

tn engine start owns both subprocesses. To use a gateway you run separately, start only the manager against its socket:

tn gateway --config sdk.toml --socket /tmp/engine-gateway.sock   # SDK, reads sdk.toml
tn engine manager --config engine.toml --socket /tmp/engine-gateway.sock

tn engine gateway runs the same SDK gateway but reads accounts and the socket from engine.toml. These are alternative entry points, not commands to run together on one socket; see Spec 05.

1. No takeover: tn engine start and tn engine gateway refuse to start when a gateway is already listening on the socket, and suggest tn engine manager --socket <path> instead.
2. Shutdown order: tn engine start stops the manager first, so tasks can still reach venues during cleanup, then the gateway. Each waits up to --shutdown-timeout before being killed.

On a systemd host, install tn engine start as a user service:

tn engine service install --cwd . --memory 2G --timeout-stop-sec 20 -v
tn engine service status
tn engine service logs -f
tn engine service restart | stop | start | uninstall

CLI reference

task start and task restart take params as repeated -p key=value (--param; -a/--arg are kept as aliases); the engine validates and coerces them. task restart without -p keeps the current params. Every command takes --help.

HTTP API

Served by the manager (default http://localhost:3121), with the dashboard at /. Control routes take an optional ?actor= for attribution (default api).

1. Tasks are {process, id, key, status, params, start, attempt, stop_reason, error, traceback, started_at, stopped_at}. start counts the task's explicit starts and restarts, across manager restarts; attempt counts attempts within the current start. Logs and attempts carry both, so history is never overwritten. Live tasks also carry their ui States' current values, states: {name: {value, updated_at, truncated}}; these are never stored. Values are JSON (Decimal and datetime as strings, unknown objects as their repr); one longer than 2 KB of JSON is a cut string preview with truncated: true.
2. Errors: unknown process 404; AlreadyRunning 409; UnmetDependency 409 with {message, missing}; ParamsConflict 409 with {message, key, running, requested}; DependencyCycle 422 with {message, cycle}; invalid params 422.
3. Websocket: the first message is a snapshot (tasks, processes, health, history), followed by task_update, task_end, log, control, state and ping events, each tagged by t. A state event (key, start, attempt, name, value, updated_at, truncated) carries a State's latest value, at most every 250 ms per State; each attempt begins with its defaults.

Market picker

Type a params field as MarketId to let the dashboard browse markets instead of typing ids:

from tribulnation.engine import BaseParams, MarketId


class MakerParams(BaseParams):
  maker: MarketId
  hedger: MarketId | None = None

1. A MarketId is a plain str of the form <account>:<exchange>:<market>, e.g. hl::BTC or aster:perp:ETHUSDT. The exchange may be empty, and the market may contain colons. Other values fail validation.

2. AccountId (hl) and ExchangeId (<account>:<exchange>, e.g. hl: or dydx:perp; the exchange may be empty) work the same way for params that need only an account or an exchange.

3. In the start form, such fields get a Browse button that builds the id from the catalogue: pick an account, a market type, then a market, each with a search box. For AccountId and ExchangeId the picker stops after the account or market type.

4. Accounts are the [accounts.*] from engine.toml (the SDK has no implicit ones), hiding venues whose connector isn't installed; testnet accounts list their mainnet markets. Listings don't need the gateway.

5. The catalogue is downloaded on first use, cached and refreshed at most daily. If it can't be reached, ids can still be typed.

6. MarketSpec is a market with its contract multiplier: the asset units per unit of the market's quantity (e.g. 1000 for Hyperliquid kPEPE, which trades thousands of PEPE). The multiplier is a plain param defaulting to 1, so markets the catalogue doesn't list work too; a plain market id validates as a spec with multiplier 1. Specs are frozen and hashable:

   from tribulnation.engine import BaseParams, MarketSpec
   
   
   class HedgerParams(BaseParams):
     maker: MarketSpec
     hedges: set[MarketSpec] = set()
   

   [[tasks]]
   process = "hedger"
   id      = "main"
   args    = { maker = "hl::BTC", hedges = [{ market = "hl::kPEPE", multiplier = 1000 }] }
   

   In the start form, a MarketSpec gets the market picker plus a multiplier input; picking a market from the catalogue fills in its multiplier, which stays editable.

7. Lists, sets and dicts of any of these markers (e.g. list[MarketId], dict[str, MarketSpec]) get one picker row per item, with add and remove buttons; dict rows also have a key.

Python client

from tribulnation.engine import TradingEngine

async with TradingEngine.at(
  'unix:///tmp/engine-gateway.sock', api_url='http://localhost:3121'
) as engine:
  market = await engine.sdk.market('hl::BTC')
  await engine.task.start('maker', 'btc', {'market': 'hl::BTC'})
  await engine.task.stop('maker', 'btc')

engine.task is a TaskClient with list, get, start(process, id, params), restart(process, id, params=None) and stop(process, id, purge=False).

Upgrading from 0.1

1. Gateway moved to the SDK. tribulnation.engine.gateway is gone; import ProxySDK from tribulnation.sdk.gateway. The wire protocol is the SDK's. Upgrade the gateway and manager together.
2. Params are explicit. Declare class Maker(Process[MakerParams]); run receives the params model.
3. Declared dependencies replace self.require(...). Use @Handle.require(Target, when=Target.some_state) with a key function returning one id. Named states set with self.set_state(...) become State[bool] members passed as when. self.ipc is replaced by handle call, get and subscribe.
4. Restart budgets: the rolling limit is RestartPolicy(limit=RestartLimit(max_restarts=N, within=T)).
5. API and client fields renamed: args is now params in request bodies and task data, and attempt_count is now attempt. TaskClient.start(process, id, params) takes a mapping instead of keyword arguments. [[tasks]] args in engine.toml is unchanged.
6. Store migration is automatic. The SQLite schema is versioned with PRAGMA user_version; a 0.1 database is migrated in place on first start. Back it up first: a migrated database can't be opened by 0.1.
7. Commands: use tn engine ...; the engine command is gone. Service units installed by 0.1 call it, so reinstall them with tn engine service install.

Architecture

tn engine start
├── gateway   SDK gateway: owns venue connections and credentials (Unix socket)
└── manager   task manager
    ├── Runtime: processes run as asyncio tasks, with self.sdk = ProxySDK
    ├── REST + WebSocket API and dashboard (FastAPI / uvicorn)
    └── SQLite store: tasks, attempts and logs

Strategies never hold venue connections directly: the gateway owns them and shares them between tasks. Process declarations (tribulnation.engine.process) are separate from the runtime that runs them (tribulnation.engine.runtime); see docs/architecture.md for the module map.

Releases

GitHub Actions builds and publishes releases. PRs and pushes to main type-check the package, run its Python tests, build the dashboard using npm ci, and build/check the Python wheel and source distribution. CI verifies that the wheel contains the dashboard and uploads both distributions as workflow artifacts.

1. Create a branch named release/engine from main.
2. Bump version in lib/pyproject.toml and open a PR to main.
3. Merge after checks pass. The release workflow checks the version bump and rebuilds the exact merged commit, publishes its distributions to PyPI, then creates a v<version> tag and GitHub release with the distributions attached.

Ordinary PR merges do not publish. Re-running the release workflow skips files already uploaded to PyPI and preserves an existing GitHub release.

Configure a PyPI trusted publisher once for tribulnation-engine: owner tribulnation, repository engine, workflow release.yml, with the environment field left empty (matching the SDK setup). No PyPI API token or exchange credentials are required by CI.