Skip to content

Commit 4b7dc5f

Browse files
FullstackGJJFullstackGJJ
authored
doc(fi): added tests and examples (#20)
1 parent 7f9d2f2 commit 4b7dc5f

File tree

4 files changed

+208
-1
lines changed

4 files changed

+208
-1
lines changed

‎.gitignore‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ bld/
2323
[Bb]in/
2424
[Oo]bj/
2525
[Ll]og/
26+
*.tar.gz
2627

2728
# Visual Studio 2015 cache/options directory
2829
.vs/
@@ -260,4 +261,10 @@ paket-files/
260261

261262
# Python Tools for Visual Studio (PTVS)
262263
__pycache__/
263-
*.pyc
264+
*.pyc
265+
266+
# VSCode
267+
.vscode
268+
269+
# Python Environments
270+
env/

‎examples/fi_example.py‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
import json
2+
import sys
3+
import time
4+
import pandas as pd
5+
import uuid
6+
import os
7+
8+
from fds.analyticsapi.engines.api.fi_calculations_api import FICalculationsApi
9+
from fds.analyticsapi.engines.api_client import ApiClient
10+
from fds.analyticsapi.engines.configuration import Configuration
11+
from fds.analyticsapi.engines.models.fi_calculation_parameters import FICalculationParameters
12+
from fds.analyticsapi.engines.models.fi_security import FISecurity
13+
from fds.analyticsapi.engines.models.fi_job_settings import FIJobSettings
14+
from fds.analyticsapi.engines.stach_extensions import StachExtensions
15+
from fds.protobuf.stach.Package_pb2 import Package
16+
17+
from google.protobuf import json_format
18+
from urllib3 import Retry
19+
20+
host = "https://api.factset.com"
21+
username = os.environ["ANALYTICS_API_USERNAME_SERIAL"]
22+
password = os.environ["ANALYTICS_API_PASSWORD"]
23+
24+
25+
def main():
26+
config = Configuration()
27+
config.host = host
28+
config.username = username
29+
config.password = password
30+
# add proxy and/or disable ssl verification according to your development environment
31+
# config.proxy = "<proxyUrl>"
32+
config.verify_ssl = False
33+
34+
# Setting configuration to retry api calls on http status codes of 429 and 503.
35+
config.retries = Retry(total=3, status=3, status_forcelist=frozenset([429, 503]), backoff_factor=2,
36+
raise_on_status=False)
37+
38+
api_client = ApiClient(config)
39+
40+
calculations = ["Security Type",
41+
"Security Name",
42+
"Run Status",
43+
"Elapse Time (seconds)",
44+
"Calc From Method",
45+
"Option Pricing Model",
46+
"Yield Curve Date",
47+
"Settlement Date",
48+
"Discount Curve",
49+
"Price",
50+
"Yield to No Call",
51+
"OAS",
52+
"Effective Duration",
53+
"Effective Convexity"]
54+
55+
security1 = FISecurity("Price", 100.285, 10000.0, "912828ZG8", "20201202", "UST")
56+
security2 = FISecurity("Price", 101.138, 200000.0, "US037833AR12", "20201203", "UST")
57+
58+
securities = [security1, security2]
59+
60+
jobSettings = FIJobSettings("20201201")
61+
62+
fi_calculation_parameters = FICalculationParameters(securities, calculations, jobSettings)
63+
64+
print(fi_calculation_parameters)
65+
66+
fi_calculations_api = FICalculationsApi(api_client)
67+
run_calculation_response = fi_calculations_api.run_fi_calculation_with_http_info(
68+
fi_calculation_parameters=fi_calculation_parameters)
69+
70+
if run_calculation_response[1] != 202 and run_calculation_response[1] != 201:
71+
print_error(run_calculation_response)
72+
sys.exit()
73+
74+
if run_calculation_response[1] == 201:
75+
print_result(run_calculation_response[0])
76+
sys.exit()
77+
78+
calculation_id = run_calculation_response[2].get("location").split("/")[-1]
79+
print("Calculation Id: " + calculation_id)
80+
81+
status_response = fi_calculations_api.get_fi_calculation_by_id_with_http_info(calculation_id)
82+
while status_response[1] == 202:
83+
max_age = '5'
84+
age_value = status_response[2].get("cache-control")
85+
if age_value is not None:
86+
max_age = age_value.replace("max-age=", "")
87+
print('Sleeping: ' + max_age)
88+
time.sleep(int(max_age))
89+
status_response = fi_calculations_api.get_fi_calculation_by_id_with_http_info(calculation_id)
90+
91+
if status_response[1] != 200:
92+
print_error(status_response)
93+
sys.exit()
94+
95+
print_result(status_response[0])
96+
97+
98+
def print_result(response):
99+
# converting the data to Package object
100+
result = json_format.Parse(json.dumps(response), Package())
101+
# print(MessageToJson(result)) # To print the result object as a JSON
102+
# print(MessageToDict(result)) # To print the result object as a Dictionary
103+
tables = StachExtensions.convert_to_table_format(result) # To convert result to 2D tables.
104+
print(tables[0]) # Prints the result in 2D table format.
105+
# generate_excel(result) # Uncomment this line to get the result in table format exported to excel file.
106+
107+
108+
def generate_excel(package):
109+
for table in StachExtensions.convert_to_table_format(package):
110+
writer = pd.ExcelWriter(str(uuid.uuid1()) + ".xlsx")
111+
table.to_excel(excel_writer=writer)
112+
writer.save()
113+
writer.close()
114+
115+
116+
def print_error(response):
117+
print("Calculation Failed!!!")
118+
print("Status Code: " + str(response[1]))
119+
print("Request Key: " + response[2].get("x-datadirect-request-key"))
120+
print(response[0])
121+
122+
123+
if __name__ == '__main__':
124+
main()

‎examples/requirements.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
fds.protobuf.stach <2.0.0

‎tests/test_fi_calculations_api.py‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
import time
2+
import json
3+
import unittest
4+
import sys
5+
import uuid
6+
7+
from fds.analyticsapi.engines.api.fi_calculations_api import FICalculationsApi
8+
from fds.analyticsapi.engines.api_client import ApiClient
9+
from fds.analyticsapi.engines.configuration import Configuration
10+
from fds.analyticsapi.engines.models.fi_calculation_parameters import FICalculationParameters
11+
from fds.analyticsapi.engines.models.fi_security import FISecurity
12+
from fds.analyticsapi.engines.models.fi_job_settings import FIJobSettings
13+
from fds.analyticsapi.engines.stach_extensions import StachExtensions
14+
from fds.protobuf.stach.Package_pb2 import Package
15+
16+
from google.protobuf import json_format
17+
from google.protobuf.json_format import MessageToJson
18+
from google.protobuf.json_format import MessageToDict
19+
20+
import common_parameters
21+
from common_functions import CommonFunctions
22+
23+
24+
class TestFICalculationsApi(unittest.TestCase):
25+
def setUp(self):
26+
self.api_client = CommonFunctions.build_api_client()
27+
self.calculations_api = FICalculationsApi(self.api_client)
28+
self.run_response = self.run_calculation()
29+
self.status_response = None
30+
31+
def run_calculation(self):
32+
calculations = ["Security Type",
33+
"Security Name",
34+
"Run Status",
35+
"Elapse Time (seconds)",
36+
"Calc From Method",
37+
"Option Pricing Model",
38+
"Yield Curve Date",
39+
"Settlement Date",
40+
"Discount Curve",
41+
"Price",
42+
"Yield to No Call",
43+
"OAS",
44+
"Effective Duration",
45+
"Effective Convexity"]
46+
47+
security1 = FISecurity("Price", 100.285, 10000.0, "912828ZG8", "20201202", "UST")
48+
security2 = FISecurity("Price", 101.138, 200000.0, "US037833AR12", "20201203", "UST")
49+
50+
securities = [security1, security2]
51+
52+
jobSettings = FIJobSettings("20201201")
53+
54+
fi_calculation_parameters = FICalculationParameters(securities, calculations, jobSettings)
55+
return self.calculations_api.run_fi_calculation_with_http_info(
56+
fi_calculation_parameters=fi_calculation_parameters)
57+
58+
def test_calculation_success(self):
59+
if self.run_response[1] == 202:
60+
calculation_id = self.run_response[2].get("location").split("/")[-1]
61+
self.run_response = self.calculations_api.get_fi_calculation_by_id_with_http_info(calculation_id)
62+
while self.run_response[1] == 202:
63+
age_value = self.run_response[2].get("cache-control")
64+
if age_value is not None:
65+
max_age = age_value.replace("max-age=", "")
66+
time.sleep(int(max_age))
67+
else:
68+
time.sleep(5)
69+
self.run_response = self.calculations_api.get_fi_calculation_by_id_with_http_info(calculation_id)
70+
71+
self.assertTrue(self.run_response[1] == 200 or self.run_response[1] == 201, "Calculation should be completed")
72+
73+
74+
if __name__ == '__main__':
75+
unittest.main()

0 commit comments

Comments
 (0)