StreamManager

Defined in dccd.daemon.stream_manager

class StreamManager(config, health=None)

Bases: object

Manage real-time WebSocket collection jobs.

Starts one background thread per (exchange, pair) (or per (exchange, pair, channel) for Bitfinex/Bitmex). Each thread runs indefinitely and is automatically restarted after a crash. A SyncService instance pushes data to remotes periodically.

Parameters:

configCollectorConfig

Daemon configuration (stream_jobs + storage).

running_keys()

Return the keys of all currently alive stream threads.

start()

Start the sync service and all stream threads.

start_one(job, pair, channels)

Start a single stream thread for (pair, channels) if not running.

Parameters:

jobStreamJob

Stream job configuration (only exchange/time_step are used).

pairstr

Trading pair in 'CRYPTO/FIAT' format.

channelslist of str

Channels for this thread.

Returns:

str

The thread key (running or freshly started).

status(config=None)

Return the configured stream tasks with their running state.

Parameters:

configConfig, optional

Configuration to read stream_jobs from. Defaults to the config captured at construction; callers (e.g. the web UI) pass a freshly-loaded config so stream jobs added at runtime appear here without restarting the process.

Returns:

list of dict

One entry per (pair, channels) task across all configured stream_jobs, with key, exchange, pair, channels, time_step and running fields.

stop()

Signal all streams and the sync service to stop.

stop_one(key)

Signal a single stream thread (by key) to stop.

The thread and stop-event slots are freed immediately so a subsequent start_one for the same key always spawns a fresh thread instead of reusing one that is still winding down. The downloader entry is left in place; the dying thread removes its own (identity-guarded) in _run_once.

Parameters:

keystr

Thread key as returned by start_one / status.

Returns:

bool

True if a matching running thread was found and signalled.