analyticsapi-engines-python-sdk/examples/vault_engine_multiple_unit_example.py at master · factset/analyticsapi-engines-python-sdk · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
import time
import os
import uuid
import pandas as pd
import urllib3

from fds.analyticsapi.engines import ApiException
from fds.analyticsapi.engines.api.vault_calculations_api import VaultCalculationsApi
from fds.analyticsapi.engines.api.components_api import ComponentsApi
from fds.analyticsapi.engines.api.configurations_api import ConfigurationsApi
from fds.analyticsapi.engines.api_client import ApiClient
from fds.analyticsapi.engines.configuration import Configuration
from fds.analyticsapi.engines.model.component_summary import ComponentSummary
from fds.analyticsapi.engines.model.vault_calculation_parameters_root import VaultCalculationParametersRoot
from fds.analyticsapi.engines.model.vault_calculation_parameters import VaultCalculationParameters
from fds.analyticsapi.engines.model.vault_date_parameters import VaultDateParameters
from fds.analyticsapi.engines.model.vault_identifier import VaultIdentifier
from fds.protobuf.stach.extensions.StachVersion import StachVersion
from fds.protobuf.stach.extensions.StachExtensionFactory import StachExtensionFactory
from fds.protobuf.stach.extensions.v2.StachUtilities import StachUtilities

from urllib3 import Retry
urllib3.disable_warnings()

host = os.environ['FACTSET_HOST']
fds_username = os.environ['FACTSET_USERNAME']
fds_api_key = os.environ['FACTSET_API_KEY']

def main():
    config = Configuration()
    config.host = host
    config.username = fds_username
    config.password = fds_api_key
    config.discard_unknown_keys = True
    # add proxy and/or disable ssl verification according to your development environment
    # config.proxy = "<proxyUrl>"
    config.verify_ssl = False

    # Setting configuration to retry api calls on http status codes of 429 and 503.
    config.retries = Retry(total=3, status=3, status_forcelist=frozenset([429, 503]), backoff_factor=2,
                           raise_on_status=False)

    api_client = ApiClient(config)

    components_api = ComponentsApi(api_client)

    try:
        vault_document_name = "Client:/aapi/VAULT_QA_PI_DEFAULT_LOCKED"
        vault_component_total_returns = "Total Returns"
        vault_component_performance = "Performance Over Time"
        vault_component_category = "Performance / Performance Relative Dates"
        vault_default_account = "CLIENT:/BISAM/REPOSITORY/QA/SMALL_PORT.ACCT"
        vault_startdate = "20180101"
        vault_enddate = "20180329"
        frequency = "Monthly"
        # uncomment the below code line to setup cache control; max-stale=0 will be a fresh adhoc run and the max-stale value is in seconds.
        # Results are by default cached for 12 hours; Setting max-stale=300 will fetch a cached result which is 5 minutes older.
        # cache_control = "max-stale=0"
        get_components_response = components_api.get_vault_components(vault_document_name)
        component_total_returns_id = [id for id in list(get_components_response[0].data.keys()) if get_components_response[0].data[id].name == vault_component_total_returns and get_components_response[0].data[id].category == vault_component_category][0]
        print("Vault Total Returns Component Id: " + component_total_returns_id)


        component_performance_id = [id for id in list(get_components_response[0].data.keys()) if get_components_response[0].data[id].name == vault_component_performance and get_components_response[0].data[id].category == vault_component_category][0]
        print("Vault Performance Over Time Component Id: " +
              component_performance_id)

        vault_account_identifier = VaultIdentifier(vault_default_account)
        vault_dates = VaultDateParameters(
            startdate=vault_startdate, enddate=vault_enddate, frequency=frequency)

        configurations_api = ConfigurationsApi(api_client)
        get_vault_configurations_response = configurations_api.get_vault_configurations(
            vault_default_account)
        configuration_id = list(get_vault_configurations_response[0].data.keys())[0]

        vault_calculation_parameters = {
            "total_returns": VaultCalculationParameters(componentid=component_total_returns_id, account=vault_account_identifier, dates=vault_dates, configid=configuration_id),
            "performance_over_time": VaultCalculationParameters(componentid=component_performance_id, account=vault_account_identifier, dates=vault_dates, configid=configuration_id)}

        vault_calculation_parameters_root = VaultCalculationParametersRoot(
            data=vault_calculation_parameters)

        vault_calculations_api = VaultCalculationsApi(api_client)

        post_and_calculate_response = vault_calculations_api.post_and_calculate(
            vault_calculation_parameters_root=vault_calculation_parameters_root)
        # comment the above line and uncomment the below line to run the request with the cache_control header defined earlier
        # post_and_calculate_response = vault_calculations_api.post_and_calculate(vault_calculation_parameters_root=vault_calculation_parameters_root, cache_control=cache_control)
        if post_and_calculate_response[1] == 202 or post_and_calculate_response[1] == 200:
            calculation_id = post_and_calculate_response[0].data.calculationid
            print("Calculation Id: " + calculation_id)

            status_response = vault_calculations_api.get_calculation_status_by_id(id=calculation_id)

            while status_response[1] == 202 and (status_response[0].data.status in ("Queued", "Executing")):
                max_age = '5'
                age_value = status_response[2].get("cache-control")
                if age_value is not None:
                    max_age = age_value.replace("max-age=", "")
                print('Sleeping: ' + max_age)
                time.sleep(int(max_age))
                status_response = vault_calculations_api.get_calculation_status_by_id(calculation_id)

            for (calculation_unit_id, calculation_unit) in status_response[0].data.units.items():
                if calculation_unit.status == "Success":
                    print("Calculation Unit Id: " +
                          calculation_unit_id + " Succeeded!!!")
                    result_response = vault_calculations_api.get_calculation_unit_result_by_id(id=calculation_id,
                                                                                               unit_id=calculation_unit_id)
                    output_calculation_result(result_response[0]['data'])
                else:
                    print("Calculation Unit Id:" +
                          calculation_unit_id + " Failed!!!")
                    print("Error message : " + str(calculation_unit.errors))
        else:
            print("Calculation creation failed")
            print("Error status : " + str(post_and_calculate_response[1]))
            print("Error message : " + str(post_and_calculate_response[0]))

    except ApiException as e:
        print("Api exception Encountered")
        print(e)
        exit()


def output_calculation_result(result):
    print("Calculation Result")
    metadata_list = []
    stachBuilder = StachExtensionFactory.get_row_organized_builder(StachVersion.V2)
    stachExtension = stachBuilder.set_package(result).build()
    dataFramesList = stachExtension.convert_to_dataframe()
    MetaData = stachExtension.get_metadata()
    print(dataFramesList)
    print('\n MetaData:')
    for metadataItem in MetaData:
        metadata_dict = {}
        for keyName in metadataItem:
            appendedMetadata = ','.join(str(StachUtilities.get_value(x)) for x in metadataItem[keyName])
            metadata_dict[keyName] = appendedMetadata
        metadata_list.append(metadata_dict)

    metadata_df = pd.DataFrame(metadata_list)
    # Set display options for better readability
    pd.set_option('display.max_rows', None)
    pd.set_option('display.max_columns', None)
    pd.set_option('display.width', None)
    print(metadata_df.to_string(index=False, header=True, justify='left', col_space=20))
    # generate_excel(dataFramesList)  # Uncomment this line to get the result in table format exported to excel file.


def generate_excel(data_frames_list):
    for dataFrame in data_frames_list:
        writer = pd.ExcelWriter(  # pylint: disable=abstract-class-instantiated
            str(uuid.uuid1()) + ".xlsx")
        dataFrame.to_excel(excel_writer=writer)
        writer.save()
        writer.close()


if __name__ == '__main__':
    main()