Transformation Object

This page describes how to programmatically retrieve and apply risk factor transformations to scenarios. Transformations derive new risk factors from existing ones using mathematical expressions—useful for creating lagged or leading variables, computing differences, or combining factors.

For an overview of transformation concepts and expression syntax, see Risk Factor Transformations.

Fields

The TransformationConfig object contains the following data:

Field	Description	Data Type
OutputVariables	Names of variables produced by transformation	List of Strings
ExpressionString	The transformation expression	String
DependentFactorIds	IDs of factors this transformation depends on	List of Integers

Retrieving Transformation Configurations

Before applying transformations, retrieve the available transformation configurations from the API. These configurations define the transformation expressions and their dependencies.

Go Example

Use the GetTransformationConfigs function to retrieve all available transformation configurations.

import (
    log "github.com/sirupsen/logrus"
    sdk "frg.com/streamr/sdk"
)

transConfig, err := sdkClient.GetTransformationConfigs()
if err != nil {
    log.Errorln(err)
    return
}

for key, value := range transConfig {
    log.Printf("Transformation '%v': %v\n", key, value)
}

Returns:

• A map where the key is the transformation name and the value is the TransformationConfig object.
• An error if the gRPC call fails.

Python Example

Use the get_transformation_configs function to retrieve all available transformation configurations.

from sdk import transformation

trans_config = transformation.get_transformation_configs()

for key, value in trans_config.items():
    print(f"Transformation '{key}': {value}")

Returns:

• A dictionary where the key is the transformation name and the value is the TransformationConfig object.
• A GRPCError will be raised if the gRPC call fails.

Transforming Scenarios

Once you have the transformation configurations, you can use them to transform a set of scenarios for a specific as-of-date. The transformation engine will automatically resolve the order of operations based on the dependencies defined in the configurations.

Go Example

Use the TransformScenarios function to apply the transformations.

scens, err := u.GetScenarios()
if err != nil {
    log.Errorln(err)
    return
}

transConfig, err := sdkClient.GetTransformationConfigs()
if err != nil {
    log.Errorln(err)
    return
}

result, err := sdkClient.TransformScenarios(&sdk.TransformScenariosOpts{
    Scenarios:             scens,
    TransformationConfigs: transConfig,
    AsOfDate:              "2005-03-31",
})
if err != nil {
    log.Errorln(err)
    return
}
for _, w := range result.Warnings {
    log.Warnf("[%s] %s", w.GetCode(), w.GetMessage())
}

// The transformed scenarios can now be used like regular scenarios
for _, scen := range result.Scenarios {
    scenarioName := scen.Name
    log.Printf("Scenario: %s\n", scenarioName)
    for _, d := range scen.Dates {
        log.Println(d.Date)
        for name, value := range d.Scen {
            if name == "Transformed_Factor" {
                log.Println("Found new factor:", name, value)
            }
        }
    }
}

Parameters:

The TransformScenarios function accepts a TransformScenariosOpts struct with the following fields:

• Scenarios ([]*sdk.VORScen): The scenarios to be transformed.
• TransformationConfigs (map[string]*transformation.TransformationConfig): The transformation configurations retrieved from GetTransformationConfigs.
• AsOfDate (string): The date for which the transformations should be calculated, in "YYYY-MM-DD" format.

Returns:

• A *sdk.TransformResult containing:
  • Scenarios ([]*sdk.VORScen): The transformed scenarios.
  • Warnings ([]*sdk.TransformationWarning): Non-fatal warnings (e.g., unresolved transformations due to missing inputs). Each warning has a Code and Message.
• An error if the transformation fails.

Python Example

Use the transform_scenarios function to apply the transformations.

from sdk import scenario, transformation

runID = handle.options["RunID"]
scen = scenario.get_scenarios(runID, "")

trans_config = transformation.get_transformation_configs()

result = transformation.transform_scenarios(scen, trans_config, "2005-03-31")

for w in result.warnings:
    print(f"Warning [{w.code}]: {w.message}")

# The transformed scenarios can now be used like a regular scenario dictionary
for scenarioName, s in result.scenarios.items():
    for horiz in s:
        for name, value in horiz.items():
            if name == "Another_Transformed_Factor":
                print(f"Found new factor: {name} = {value}")

Parameters:

The transform_scenarios function accepts the following arguments:

• scenarios (dict): The dictionary of scenarios, as returned by scenario.get_scenarios.
• transformation_configs (dict): The dictionary of transformation configurations, as returned by get_transformation_configs.
• as_of_date (str): The date for which the transformations should be calculated, in "YYYY-MM-DD" format.

Returns:

• A TransformResult containing:
  • scenarios (dict): The transformed scenarios, in the same format as scenario.get_scenarios.
  • warnings (list[TransformationWarning]): Non-fatal warnings (e.g., unresolved transformations due to missing inputs). Each warning has a code and message.
• A GRPCError will be raised if the transformation fails.