Exposed modeling functions

A useful subset of modeling functions have been exposed. To access them, import via

from valsys.modeling.service import X

For example, from valsys.modeling.service import tag_model.

The exposed functions are listed out below.

If a number of imports are expected to be used, in order to not explode the number of imports, it may be useful to import and alias the entire modeling service via

import valsys.modeling.service as Modeling

In which case functions are used via

Modeling.tag_model(...)
Modeling.filter_user_models(...)

in which we used a placeholder for the correct function arguments.

Model operations

Model pulling

pull_model(model_id)

Pull a model by its ID.

Parameters:

Name	Type	Description	Default
model_id	str	the ID of the required model.	required

Returns:

Type	Description
Model	The Model object for the model.

Source code in valsys/modeling/service.py

def pull_model(model_id: str) -> Model:
    """Pull a model by its ID.

    Args:
        model_id: the ID of the required model.

    Returns:
        The `Model` object for the model.
    """
    client = new_client()

    resp = client.get(
        url=VSURL.PULL_MODEL,
        headers={Headers.MODEL_ID: model_id},
    )
    check_success(resp, 'pull model')
    return Model.from_json(resp.get(Resp.DATA).get(Resp.MODEL))

Model tagging

tag_model(model_id: str, tags: List[str])

Tag the model with model_id with the list of tags.

Note that this removes any existing tags; if you wanted to append tags, use the append_tags function.

update: turns on dynamic updates

Parameters:

Name	Type	Description	Default
model_id	str	ID of the model to add tags to	required
tags	List[str]	List of tags to add to the model	required
auth_token	str	Optional authentication token	None

Source code in valsys/modeling/service.py

def tag_model(model_id: str, tags: List[str], auth_token: str = None):
    """Tag the model with `model_id` with the list of `tags`.

    Note that this removes any existing tags;
    if you wanted to append tags, use the `append_tags` function.

    update: turns on dynamic updates

    Args:
        model_id: ID of the model to add tags to
        tags: List of tags to add to the model
        auth_token: Optional authentication token
    """

    client = new_client(auth_token)
    payload = {
        Headers.MODEL_ID: model_id,
        Headers.TAGS: tags,
        Headers.UPDATES: True,
        Headers.ROLL_FORWARD: True,
    }
    try:
        return client.post(
            url=VSURL.MODELING_MODEL_PROPERTIES,
            data=payload,
        )
    except ModelingServicePostException as err:
        raise TagModelException(
            f'error tagging model via call {VSURL.MODELING_MODEL_PROPERTIES}; got {err.status_code}; message={err.data}'
        )

Model sharing

share_model(model_id: str, email: str, permission: str)

Share model to another user.

Parameters:

Name	Type	Description	Default
model_id	str	ID of the model to share	required
email	str	The email address of the user to share the model with	required
permission	str	The permissions to give to the user	required

Source code in valsys/modeling/service.py

def share_model(model_id: str,
                email: str,
                permission: str,
                auth_token: str = None):
    """Share model to another user.

    Args:
         model_id: ID of the model to share
         email: The email address of the user to share the model with
         permission: The permissions to give to the user
    """

    client = new_client(auth_token)
    permissions = Permissions(permission)

    try:
        client.post(
            url=VSURL.USERS_SHARE_MODEL,
            headers={
                "email": email,
                Headers.MODEL_ID: model_id,
            },
            data=permissions.jsonify(),
        )
    except ModelingServicePostException as err:
        raise ShareModelException(f"failed to share models {str(err)}")

The model is shared to the user with specified permissions; the allowed values are

• permission = view
• permission = edit
• permission = fullAccess

Any other permission value will result in a NotImplementedError exception being thrown.

The allowed permissions and the correct strings can be found via

from valsys.modeling.models import Permissions

So, for example, Permissions.VIEW could be provided to the share_model function call.

If you attempt to share the model with a user that dosent exist, a ShareModelException will be thrown.

Model deleting

delete_models(model_ids: List[str])

Delete the specified models

Parameters:

Name	Type	Description	Default
model_ids	List[str]	List of model IDs to be deleted.	required

Source code in valsys/modeling/service.py

def delete_models(model_ids: List[str]):
    """ Delete the specified models

    Args:
        model_ids: List of model IDs to be deleted.
    """
    client = new_client()
    url = VSURL.USERS_MODELS
    payload = {Headers.MODELS: model_ids}
    resp = client.delete(url=url, data=payload)
    check_success(resp, 'deleting models')
    return resp

Model searching/filtering

filter_user_models

Search for a set of models, using the provided set of filters for the using user

Parameters:

Name	Type	Description	Default
filter_on	List[str]	List of strings of properties to filter on; allowed: Name, Ticker, Geography, Industry.	None
filter_term	str	Will match according the props in the filter_on list.	''
model_type	str	Options are user, shared, both.	'user'
max_date	str	Maximum creation date of the model (required format: YYYY-MM-DDTHH:MM:DD.SSSZ)	tomorrow()
min_date	str	Minimum creation date of the model (required format: YYYY-MM-DDTHH:MM:DD.SSSZ)	'2002-01-01T00:00:00.000Z'
geo_filters	List[str]	The geographies to include in the search	None
ind_filters	List[str]	The industries to include in the search	None
tags	List[str]	List of tags to filter on	None
tag_filter_type	str	How to combine the tags to search over; options are and and or.	''
pagination	int	Page number of results	1
fields	List[str]	Fields to return per model	None

Returns:

Type	Description
List[ModelDetailInformationWithFields]	List of matching model information objects.

Source code in valsys/modeling/service.py

def filter_user_models(
        tags: List[str] = None,
        model_type: str = 'user',
        max_date: str = tomorrow(),
        min_date: str = "2002-01-01T00:00:00.000Z",
        tag_filter_type: str = '',
        geo_filters: List[str] = None,
        ind_filters: List[str] = None,
        filter_on: List[str] = None,
        filter_term: str = '',
        pagination: int = 1,
        fields: List[str] = None) -> List[ModelDetailInformationWithFields]:
    """Search for a set of models, using the provided set of filters for the using user

    Args:
        filter_on: List of strings of properties to filter on; allowed: `Name`, `Ticker`, `Geography`, `Industry`.
        filter_term: Will match according the props in the `filter_on` list.  
        model_type: Options are `user`, `shared`, `both`. 
        max_date: Maximum creation date of the model (required format: YYYY-MM-DDTHH:MM:DD.SSSZ)
        min_date: Minimum creation date of the model (required format: YYYY-MM-DDTHH:MM:DD.SSSZ)
        geo_filters: The geographies to include in the search
        ind_filters: The industries to include in the search
        tags: List of tags to filter on
        tag_filter_type: How to combine the tags to search over; options are `and` and `or`.
        pagination: Page number of results
        fields: Fields to return per model
    Returns:
        List of matching model information objects.
    """
    filters = ModelsFilter(
        max_date=max_date,
        min_date=min_date,
        tag_filter_type=tag_filter_type,
        model_type=model_type,
        geo_filters=geo_filters,
        ind_filters=ind_filters,
        tag_filters=tags,
        predicate=filter_term,
    )
    filters.set_filter_on(filter_on)
    url = VSURL.USERS_FILTER_HISTORY

    if fields is not None:
        filters.add_fields(fields)
        url = VSURL.USERS_FILTER_HISTORY_FIELDS

    headers = {
        Headers.PAGINATION: str(pagination),
    }
    client = new_client()
    try:
        payload = filters.jsonify()
        resp = client.post(url=url, headers=headers, data=payload)
    except ModelingServicePostException as err:
        raise err
    try:
        return [
            ModelDetailInformationWithFields.from_json(j)
            for j in resp.get(Resp.DATA).get(Resp.MODELS)
        ]
    except (TypeError, AttributeError):
        return []

Get model information

pull_model_information(model_id: str)                

Pulls the model information for the model_id.

Parameters:

Name	Type	Description	Default
model_id	str	the ID of the required model.	required

Returns:

Type	Description
ModelInformation	The ModelInformation object for the model.

Source code in valsys/modeling/service.py

def pull_model_information(model_id: str) -> ModelInformation:
    """Pulls the model information for the `model_id`.

    Args:
        model_id: the ID of the required model.

    Returns:
        The `ModelInformation` object for the model.
    """
    client = new_client()
    try:
        resp = client.get(
            url=VSURL.MODEL_INFO,
            headers={Headers.MODEL_IDS: model_id},
        )
        if resp.get('status') == Vars.SUCCESS:
            if resp["data"]["models"]:
                if len(resp["data"]["models"]) > 0:
                    cases = resp["data"]["models"][0][Resp.MODEL]
            else:
                raise PullModelInformationException(
                    f"could not pull model info for model={model_id}; no models returned"
                )
        else:
            raise PullModelInformationException(
                f"could not pull model info for model={model_id}; status={resp.get('status')}"
            )

    except (ModelingServiceGetException, Exception) as err:
        raise PullModelInformationException(
            f"could not pull model info for model={model_id}")
    return ModelInformation.from_json(model_id, cases)

This function returns a ModelInformation object, whose structure is

class ModelInformation:
    uid: str # the uid of the model
    tags: List[str] # tags on the model
    cases: List[CaseInformation] # list of case information inside the model

in which a CaseInformation object has the structure

class CaseInformation:
    uid: str # the uid of the case
    case: str # the name of the case

Get model case

pull_case(case_id: str)

Retreive a Case by its uid.

Parameters:

Name	Type	Description	Default
case_id	str	the case's UID	required

Returns:

Type	Description
Case	The appropriate Case object.

Source code in valsys/modeling/service.py

def pull_case(case_id: str) -> Case:
    """Retreive a `Case` by its uid.

    Args:
        case_id: the case's UID

    Returns:
        The appropriate `Case` object.
    """
    client = new_client()
    resp = client.get(
        url=VSURL.CASE,
        headers={
            Headers.CASE_ID: case_id,
        },
    )
    return Case.from_json(resp["data"]["case"])

Recalculate model

recalculate_model(model_id: str)

Recalculates the model.

Parameters:

Name	Type	Description	Default
model_id	str	The ID of the model to be recalculated.	required

Returns:

Type	Description
List[Fact]	List of Facts updated during the recalculation process.

Source code in valsys/modeling/service.py

def recalculate_model(model_id: str) -> List[Fact]:
    """Recalculates the model.

    Args:
        model_id: The ID of the model to be recalculated.

    Returns:
        List of Facts updated during the recalculation process.
    """
    client = new_client()
    payload = {Headers.MODEL_ID: model_id, 'update': True}

    try:
        resp = client.post(url=VSURL.RECALC_MODEL, data=payload)
    except ModelingServicePostException as err:
        raise RecalculateModelException(
            f"error recalculating model: {str(err)}")

    check_success(resp,
                  'recalculating model',
                  exception=RecalculateModelException)
    return facts_list(resp.get(Resp.DATA).get(Resp.FACTS))

Dynamic updates

dynamic_updates()                      

Requests dynamic updates are executed.

Source code in valsys/modeling/service.py

def dynamic_updates():
    """Requests dynamic updates are executed."""
    client = new_socket_client()

    resp = client.get(url=VSURL.SCK_ORCHESTRATOR,
                      data={
                          "action": ModelingActions.DYNAMIC_UPDATES,
                          "username": API_USERNAME,
                          "password": API_PASSWORD
                      })
    return resp

Module operations

Add child module

add_child_module(parent_module_id: str, name: str, model_id: str, case_id: str) 

Add a new module to the parent module.

Parameters:

Name	Type	Description	Default
parent_module_id	str	The moduleID of the parent	required
name	str	The name of the new module	required
model_id	str	The ID of the model into which the module is to be inserted	required
case_id	str	The caseID of the module.	required

Returns:

Type	Description
Module	The newly constructed Module object.

Source code in valsys/modeling/service.py

def add_child_module(parent_module_id: str, name: str, model_id: str,
                     case_id: str) -> Module:
    """Add a new module to the parent module.

    Args:
        parent_module_id: The moduleID of the parent
        name: The name of the new module
        model_id: The ID of the model into which the module is to be inserted
        case_id: The caseID of the module.

    Returns:
        The newly constructed `Module` object.
    """
    payload = {
        Headers.CASE_ID: case_id,
        Headers.MODEL_ID: model_id,
        Headers.NAME: name,
        Headers.PARENT_MODULE_ID: parent_module_id,
    }
    client = new_client()
    resp = client.post(
        url=VSURL.ADD_MODULE,
        data=payload,
    )
    try:
        child_modules = resp["data"]["module"]['edges']["childModules"]
    except KeyError:
        raise AddChildModuleException(
            f"Error adding child module: unexpected data structure")
    for module in child_modules:
        if module["name"] == name:
            return Module.from_json(module)
    raise AddChildModuleException(
        f"Error adding child module: could not find module with name {name}")

Delete module

remove_module(model_id: str, module_id: str)

Removes the specified module from the model.

Parameters:

Name	Type	Description	Default
model_id	str	The ID of the model.	required
module_id	str	The ID of the module to be removed.	required

Source code in valsys/modeling/service.py

def remove_module(model_id: str, module_id: str):
    """Removes the specified module from the model.

    Args:
        model_id: The ID of the model.
        module_id: The ID of the module to be removed.
    """

    client = new_client()
    try:
        rm = client.post(
            url=VSURL.DELETE_MODULE,
            data={
                Headers.MODEL_ID: model_id,
                Headers.MODULE_ID: module_id,
            },
        )
    except ModelingServicePostException as err:
        raise RemoveModuleException(f'error removing module: {str(err)}')
    return rm.get('status') == Vars.SUCCESS

Rename module

rename_module(model_id: str, module_id: str, new_module_name: str) 

Rename the module.

Parameters:

Name	Type	Description	Default
model_id	str	the ID of the model	required
module_id	str	the ID of the module to be renamed	required
new_module_name	str	the new name of the module.	required

Returns:

Type	Description
Module	The new renamed module object.

Source code in valsys/modeling/service.py

def rename_module(model_id: str, module_id: str,
                  new_module_name: str) -> Module:
    """Rename the module.

    Args:
        model_id: the ID of the model
        module_id: the ID of the module to be renamed
        new_module_name: the new name of the module.

    Returns:
        The new renamed module object.
    """
    client = new_client()

    r = client.post(
        url=VSURL.RENAME_MODULE,
        data={
            Headers.MODEL_ID: model_id,
            Headers.MODULE_ID: module_id,
            Headers.NAME: new_module_name
        },
    )
    check_success(r, 'adding column')
    return module_from_resp(r)

Line item operations

Add line item

add_line_item(case_id: str, model_id: str, module_id: str, name: str, order: int)                      

Add a line item to an existing module.

Parameters:

Name	Type	Description	Default
case_id	str	The caseID of the model	required
model_id	str	The modelID	required
module_id	str	The ID of the module for the new line item	required
name	str	Name of the line item	required
order	int	Order of the line item in the module	required

Returns:

Type	Description
LineItem	The newly created LineItem object.

Source code in valsys/modeling/service.py

def add_line_item(case_id: str, model_id: str, module_id: str, name: str,
                  order: int) -> LineItem:
    """Add a line item to an existing module.

    Args:
        case_id: The caseID of the model
        model_id: The modelID
        module_id: The ID of the module for the new line item
        name: Name of the line item
        order: Order of the line item in the module

    Returns:
        The newly created `LineItem` object.
    """

    client = new_client()
    try:
        resp = client.post(
            url=VSURL.ADD_ITEM,
            data={
                Headers.CASE_ID: case_id,
                Headers.MODEL_ID: model_id,
                Headers.LINE_ITEM_NAME: name,
                Headers.ORDER: order,
                Headers.MODULE_ID: module_id,
            },
        )

    except (ModelingServicePostException, Exception) as err:
        logger.exception(err)
        raise AddLineItemException(
            f"error adding line item to model={model_id} module={module_id}; {str(err)}"
        )
    try:
        line_items = resp["data"]["module"]['edges'][Resp.LINE_ITEMS]
    except KeyError as err:
        raise AddLineItemException(
            "error adding line item: invalid data structure")

    for l in line_items:
        if l['name'] == name:
            return LineItem.from_json(l)

    raise AddLineItemException(
        f"error adding line item: cannot find module with name {name}")

Delete line item

delete_line_item(model_id: str, module_id: str, line_item_id: str)                    

Delete a line item from an existing module.

Parameters:

Name	Type	Description	Default
model_id	str	The modelID	required
module_id	str	The ID of the module containing the line item	required
line_item_id	str	The ID of the line item to be deleted.	required

Returns:

Type	Description
Module	The Module without the deleted line item.

Source code in valsys/modeling/service.py

def delete_line_item(model_id: str, module_id: str,
                     line_item_id: str) -> Module:
    """Delete a line item from an existing module.

    Args:
        model_id: The modelID
        module_id: The ID of the module containing the line item
        line_item_id: The ID of the line item to be deleted.

    Returns:
        The `Module` without the deleted line item.
    """
    client = new_client()

    resp = client.post(
        url=VSURL.DELETE_ITEM,
        data={
            Headers.MODEL_ID: model_id,
            Headers.LINE_ITEM_ID: line_item_id,
            Headers.MODULE_ID: module_id,
        },
    )
    check_success(resp, 'delete line item')
    return module_from_resp(resp)

Tag a line item

tag_line_item(model_id: str, line_item_id: str, tags: List[str])                    

Tag a line item.

Note that this replaces any existing tags on the line item.

Parameters:

Name	Type	Description	Default
model_id	str	The ID of the model containing the line item	required
line_item_id	str	The ID of the line item	required
tags	List[str]	The tags to give to the line item	required

Returns:

Type	Description
LineItem	LineItem from the backend, containing updated tags.

Source code in valsys/modeling/service.py

def tag_line_item(model_id: str, line_item_id: str,
                  tags: List[str]) -> LineItem:
    """Tag a line item.

    Note that this replaces any existing tags on the line item.

    Args:
        model_id: The ID of the model containing the line item
        line_item_id: The ID of the line item
        tags: The tags to give to the line item

    Returns:
        LineItem from the backend, containing updated tags.

    """
    client = new_client()
    payload = {
        Headers.MODEL_ID: model_id,
        Headers.LINE_ITEM_ID: line_item_id,
        Headers.TAGS: tags
    }
    try:
        ait = client.post(
            url=VSURL.ADD_ITEM_TAGS,
            data=payload,
        )
    except ModelingServicePostException as err:
        raise TagLineItemException(
            f"error tagging line item: payload={payload} err={str(err)}")
    return LineItem.from_json(ait.get(Resp.DATA).get(Resp.LINE_ITEM))

Edit line items

edit_line_items(model_id: str, line_items: List[LineItem])                    

Edit line items

The passed in line items will be used to update the line items.

Parameters:

Name	Type	Description	Default
model_id	str	the ID of the model containing the line items	required
line_items	List[LineItem]	List of LineItems that will be updated.	required

Source code in valsys/modeling/service.py

def edit_line_items(model_id: str,
                    line_items: List[LineItem]) -> List[LineItem]:
    """Edit line items

    The passed in line items will be used to update the line items.

    Args:
        model_id: the ID of the model containing the line items
        line_items: List of `LineItem`s that will be updated.

    """
    client = new_client()
    payload = {
        Headers.MODEL_ID: model_id,
        Headers.LINE_ITEMS: [li.jsonify() for li in line_items],
    }
    r = client.post(url=VSURL.EDIT_LINE_ITEMS, data=payload)
    check_success(r, 'line item editing')
    return line_items_list(r.get(Resp.DATA).get(Resp.LINE_ITEMS))

Fact operations

Edit formula

edit_formula(case_id: str, model_id: str, facts: List[Fact])                      

Edit the formula on the supplied facts.

Parameters:

Name	Type	Description	Default
case_id	str	The caseID for where the facts live.	required
model_id	str	The modelID for where the facts live.	required
facts	List[Fact]	The list of facts whose formulae are to be edited.	required

Returns:

Type	Description
List[Fact]	List of Facts modified by the edit.

Source code in valsys/modeling/service.py

def edit_formula(case_id: str, model_id: str, facts: List[Fact]) -> List[Fact]:
    """Edit the formula on the supplied facts.

    Args:
        case_id: The caseID for where the facts live.
        model_id: The modelID for where the facts live.
        facts: The list of facts whose formulae are to be edited.


    Returns:
        List of `Fact`s modified by the edit.
    """
    return edit_facts(url=VSURL.EDIT_FORMULA,
                      case_id=case_id,
                      model_id=model_id,
                      facts=facts)

Model groups

Get model groups

pull_model_groups()

Pulls model groups.

Returns a list of ModelGroup objects under the groups attribute. Each ModelGroup has a uid, name, user_id, model_ids

Returns:

Type	Description
ModelGroups	ModelGroups

Source code in valsys/modeling/service.py

def pull_model_groups() -> ModelGroups:
    """Pulls model groups.

    Returns a list of `ModelGroup` objects under the `groups` attribute. 
    Each `ModelGroup` has a `uid`, `name`, `user_id`, `model_ids`

    Returns:
        ModelGroups
    """
    client = new_client()
    try:
        g = client.get(url=VSURL.USERS_GROUPS)
    except ModelingServiceGetException as err:
        raise PullModelGroupsException(
            f"error pulling model groups: {str(err)}")
    return ModelGroups.from_json(g.get(Resp.DATA))

Add new model group

new_model_groups(group_name: str, model_ids: List[str]) 

Add a new model group.

Parameters:

Name	Type	Description	Default
group_name	str	The name of the new model group	required
model_ids	List[str]	The IDs of the models to go into the group	required

Returns:

Type	Description
ModelGroups	ModelGroups

Source code in valsys/modeling/service.py

def new_model_groups(group_name: str, model_ids: List[str]) -> ModelGroups:
    """Add a new model group.

    Args:
        group_name: The name of the new model group
        model_ids: The IDs of the models to go into the group

    Returns:
        ModelGroups
    """
    client = new_client()
    try:
        g = client.post(url=VSURL.USERS_GROUP,
                        data={
                            Headers.NAME: group_name,
                            'modelIDs': model_ids
                        })
    except ModelingServicePostException as err:
        raise NewModelGroupsException(
            f"error adding new model groups: {str(err)}")
    return ModelGroups.from_json(g.get(Resp.DATA))

Update model groups

update_model_groups(uid: str, name: str, model_ids: List[str])

Updates the models groups.

Parameters:

Name	Type	Description	Default
uid	str	The UID of the model	required
name	str	The name of the model group	required
model_ids	List[str]	The IDs	required

Returns:

Type	Description
ModelGroups	ModelGroups

Source code in valsys/modeling/service.py

def update_model_groups(uid: str, name: str,
                        model_ids: List[str]) -> ModelGroups:
    """Updates the models groups.

    Args:
        uid: The UID of the model
        name: The name of the model group
        model_ids: The IDs

    Returns:
        ModelGroups
    """
    client = new_client()
    try:
        g = client.post(url=VSURL.USERS_UPDATE_GROUP,
                        data={
                            'name': name,
                            'uid': uid,
                            'modelIDs': model_ids
                        })
    except ModelingServicePostException as err:
        raise UpdateModelGroupsException(str(err))

    return ModelGroups.from_json(g.get(Resp.DATA))