Runtime Data Acquistion

[fubuloubu]

Overview

It would be very useful for Silverback to automate the collection of runtime datapoints deemed important by the app creator for the purposes of monitoring the operation of their bots, for either live monitoring or historical analysis purposes (when properly stored). TaskIQ defines an ability to return a TaskiqResult data item from the execution of a task, so that the original task kicker can read and handle that data according to it's own needs.

Currently, the Runner class does this handling of the results, and with #45 it can also support updating a database, but makes no restrictions on the types of data that can be returned and stored. The Silverback Platform will have a feature for displaying different timeseries metrics and events that occur with each bot, so we need a way to allow users to specify data items they wish to populate into the platform.

Specification

Restrict the types of data that a Silverback app task can return to the following:

silverback/types.py:

from datetime import datetime
from decimal import Decimal
from typing import Annotated, Literal

from pydantic import BaseModel, Field


class Datapoint(BaseModel):
    type: str  # discriminator

    # default value ensures we don't have to set this manually
    time: Annotated[datetime, Field(default_factory=lambda: datetime.now(timezone.utc))]


class ScalarDatapoint(Datapoint):
    type: Literal["scalar"]

    # supported scalar value types:
    data: bool | int | float | Decimal


# This is what a Silverback app task must return to integrate properly with our data acq system
Result = dict[str, Datapoint]
# Otherwise, log a warning and ignore the return value

A user can return one of the basic types supported by ScalarDatapoint.data directly from their task, and Silverback task runner will convert those data items into an instance of the ScalarDatapoint class for proper integration with the platform without having to import and use a more cumbersome type.

---

For future support of new Datapoint subclasses, the user will have to import that type in order to use it effectively. For example, let's say we design a new Datapoint type that supports graphing candlestick chart datapoints, then the model for that datapoint might look something like this:

class CandlestickDatapoint(Datapoint):
    type: Literal["candlestick"]

    high: float
    low: float
    open: float
    close: float

    # datapoints can also have computed properties
    @property
    def closed_higher(self) -> bool:
        return self.open < self.close

---

The data structure will be appropriately serialized to work with the data acquisition system to be stored as a complex timeseries data point, which can be graphed either on the supporting chart type (which makes use of all the parameters by default) OR can be mapped to another chart type by exposing one or more of the inner data items (e.g. .high, .low, etc.) to what that chart expects.

If the user changes the str label applied to the result of a specific datapoint (the key in the Result dictionary), that will change the labeled dataset where the datapoint is associated. It is not recommended to change the class of the datapoint type associated with a specific label, but a user can make this change (even though it may cause inconsistent results when displaying the timeseries data chart).

A user can also selectively decide to return a specific labeled datapoint as part of a set, all that will mean is that sampling frequency of that data item is inconsistent, and updates in the chart display may display poorly as a result.

Dependencies

Somewhat related to #39

[linear]

SBK-427 Runtime Signal Data Acquistion

[mikeshultz]

This all sounds good and seems pretty clear.

Only thing I wonder is if we should either error or otherwise handle other return types for backwards compatibility and flexibility. Like maybe we could coerce that return to the default key/tag/label in the wanted return type. For example:

if not is_dict_of_datapoints(return_value):
    return_value = {"default": BlobDatapoint(data=return_value)}

Then that would get ignored in the blackbox metrics displays but be available via the UI to the user. I still think blob returns could be powerful for users.

[fubuloubu]

This all sounds good and seems pretty clear.

Only thing I wonder is if we should either error or otherwise handle other return types for backwards compatibility and flexibility. Like maybe we could coerce that return to the default key/tag/label in the wanted return type. For example:

if not is_dict_of_datapoints(return_value):
    return_value = {"default": BlobDatapoint(data=return_value)}

Then that would get ignored in the blackbox metrics displays but be available via the UI to the user. I still think blob returns could be powerful for users.

Wanted to mention I'm doing a version of this, but only if it's a valid numeric type that we can actually display, not for any random return value type

My implementation so far is to support simple numeric datapoints, or you can subclass BaseDatapoint for a more powerful model type (we can serialize it for storage as <task_name>.<key_name> to pull for timeseries data)