Calculations Transform

Calculated Signals allow you to create new signals by applying custom Python logic to existing signals. This is useful for performing complex math, multi-signal aggregations, or custom transformations.

Jupyter Notebook tutorials

• Create a calculated signal from raw signals
• Create a ROC calculation for a signal

Create a calculated model

This endpoint is used to define a calculated model by providing a Python script and specifying the aggregation logic.

Method	Path
POST	/api/1.3/accounts/{account_id}/flows

Request Body Fields

Field	Type	Required	Description
name	string	Yes	A descriptive name for the flow.
flowType	string	Yes	Must be set to MODELSETUP.
description	string	No	A short summary of the flow's purpose.
spec	object	Yes	Specifications for the model.
spec.workspace	string	Yes	The ID of the workspace where the model will reside.
spec.modelName	string	No	Custom name for the model. Defaults to {{workspaceName}}/{{modelType}}/M[{{index}}].
spec.modelType	string	Yes	Must be set to CALCULATED.
spec.modelDetails	object	Yes	Configuration details for the script and execution.

Model Details Object

Field	Type	Description
statistic	string	Aggregation to use for input signals (e.g., mean, max, min, std, count, raw_points_count).
script	string	Python script as a string. Must define input_schema, output_schema, and a calculate(args) function.
valueType	string	Input signal value type: Numeric or Categorical.
evaluationWindow	string	ISO8601 duration defining the data window for script execution (e.g., PT10M).

Usage Examples

cURLPython

curl -X POST "https://app3.falkonry.ai/api/1.3/accounts/{ACCOUNT_ID}/flows" \
     -H "Authorization: Bearer <token>" \
     -H "Content-Type: application/json" \
     -d '{
      "name": "Create Sum Model",
      "flowType": "MODELSETUP",
      "spec": {
        "workspace": "ws_12345",
        "modelName": "sum_model",
        "modelType": "CALCULATED",
        "modelDetails": {
          "statistic": "mean",
          "script": "input_schema = [{"name": "sensor1", "valueType": "Numeric"}]\noutput_schema = [{"name": "sum_sensor1", "valueType": "Numeric"}]\n\ndef calculate(args):\n    total = 0\n    for k, values in args.items():\n        total = sum(values)\n    return {"sum_sensor1": total}",
          "valueType": "Numeric",
          "evaluationWindow": "PT10M"
        }
      }
    }'

    import requests

    URL = "https://app3.falkonry.ai/api/1.3/accounts/{ACCOUNT_ID}/flows"

    # The Python logic to be executed in the Falkonry engine
    # Note: input_schema and output_schema must be defined in the script
    python_script = """
    input_schema = [{"name": "sensor_a", "valueType": "Numeric"}]
    output_schema = [{"name": "normalized_a", "valueType": "Numeric"}]

    def calculate(args):
        # args is a dict where keys match input_schema names
        values = args.get("sensor_a", [])
        mean_val = sum(values) / len(values) if values else 0

        # Return a dict matching output_schema
        return {"normalized_a": [v - mean_val for v in values]}
    """

    payload = {
        "name": "Setup Normalized Signal Flow",
        "flowType": "MODELSETUP",
        "spec": {
            "workspace": "ws_12345",
            "modelName": "normalization_model",
            "modelType": "CALCULATED",
            "modelDetails": {
                "statistic": "mean",
                "script": python_script,
                "valueType": "Numeric",
                "evaluationWindow": "PT10M"
            }
        }
    }

    response = requests.post(URL, json=payload, headers=HEADERS)
    print(response.json())

---

Evaluate a calculated model

Apply a Calculated Model to historical data to generate the derived signal.

Method	Path
POST	/api/1.3/accounts/{account_id}/flows

Parameters

Field	Type	Required	Description
name	string	Yes	Name for the evaluation task.
flowType	string	Yes	Must be set to CALCEVAL.
spec.model	string	Yes	The ID of the calculated model.
spec.timeRange	object	Yes	startTime and endTime in ISO8601.
spec.inputsignals	array	Yes	Mapping of signal IDs to schema names.
spec.outputsignalPrefix	string	Conditional*	Prefix for new signals. Required if outputsignals is empty.
spec.outputsignals	array	Conditional*	Map to existing signal IDs. Required if outputsignalPrefix is empty.
spec.assessmentRate	string	No	ISO duration (e.g., PT5M). Overrides model default.

Usage Examples

Option 1: New Output Signals (Prefix)Option 2: Existing Output Signals

{
  "name": "Eval Calc Run 1",
  "flowType": "CALCEVAL",
  "spec": {
    "model": "calc_model_id",
    "workspace": "ws_id",
    "timeRange": {
      "startTime": "2026-01-01T00:00:00Z",
      "endTime": "2026-01-02T00:00:00Z"
    },
    "inputsignals": [
      { "signal": "cs_id_001", "name": "sensor1" }
    ],
    "outputsignalPrefix": "calculated/test1"
  }
}

{
  "name": "Eval Calc Run 2",
  "flowType": "CALCEVAL",
  "spec": {
    "model": "calc_model_id",
    "workspace": "ws_id",
    "timeRange": {
      "startTime": "2026-01-02T00:00:00Z",
      "endTime": "2026-01-03T00:00:00Z"
    },
    "inputsignals": [
      { "signal": "cs_id_001", "name": "sensor1" }
    ],
    "outputsignals": [
      { "signal": "existing_out_id", "name": "sum_sensor1" }
    ]
  }
}

---

Note

Upon completion, an Eval object is created in the workspace referencing the input/output signalsets and the model used. If using a prefix, new Connected Sources are named as {{prefix}}/{{output_schema_name}}.