release-0.9.17, batching logic, bug fixes (#230)

Author: martinroberson

gs_quant/api/risk.py

@@ -159,8 +159,13 @@ def execute_requests(outstanding_requests: queue.Queue,
                     cls.shutdown_queue_listener(responses, loop=loop)
 
         async def run_async():
+            def num_risk_jobs(request: RiskRequest):
+                # size of calculation job
+                return len(request.pricing_and_market_data_as_of) * len(request.positions)
+
             def num_risk_keys(request: RiskRequest):
-                return len(request.pricing_and_market_data_as_of) * len(request.positions) * len(request.measures)
+                # total number of risk calculations
+                return num_risk_jobs(request) * len(request.measures)
 
             is_async = not requests[0].wait_for_results
             loop = asyncio.get_event_loop()
@@ -179,7 +184,7 @@ def num_risk_keys(request: RiskRequest):
                 # If async we need a task to handle result subscription
                 results_handler = loop.create_task(cls.get_results(responses, raw_results, timeout=timeout))
 
-            expected = sum(num_risk_keys(r) for r in requests)
+            expected = sum(num_risk_jobs(r) for r in requests)
             received = 0
             chunk_size = min(max_concurrent, expected)
             result_thread = None
@@ -198,7 +203,7 @@ def num_risk_keys(request: RiskRequest):
                     while requests and dispatch_risk_keys < chunk_size:
                         dispatch_request = requests.pop()
                         dispatch_requests.append(dispatch_request)
-                        dispatch_risk_keys += num_risk_keys(dispatch_request)
+                        dispatch_risk_keys += num_risk_jobs(dispatch_request)
 
                     cls.enqueue(outstanding_requests, dispatch_requests, loop=loop)
 
@@ -209,14 +214,15 @@ def num_risk_keys(request: RiskRequest):
                     break
 
                 # Enable as many new requests as we've received results, to keep the outstanding number constant
-                chunk_received = sum(num_risk_keys(request) for request, _ in completed)
-                chunk_size = min(chunk_received, expected - received)
+                risk_jobs_received = sum(num_risk_jobs(request) for request, _ in completed)
+                chunk_size = min(risk_jobs_received, expected - received)
 
                 if progress_bar:
-                    progress_bar.update(chunk_received)
+                    risk_calcs_received = sum(num_risk_keys(request) for request, _ in completed)
+                    progress_bar.update(risk_calcs_received)
                     progress_bar.refresh()
 
-                received += chunk_received
+                received += risk_jobs_received
 
                 # Handle the results
                 if unprocessed_results is not None:

gs_quant/datetime/time.py

@@ -59,6 +59,7 @@ def __exit__(self, *args):
 
 
 class Tracer:
+    __version = 0
     __stack_depth = 0
     __stack = []
 
@@ -70,10 +71,13 @@ def __init__(self, label: str = 'Execution', print_on_exit: bool = False, thresh
     def __enter__(self):
         self.__start = dt.datetime.now()
         self.__index = len(Tracer.__stack)
+        self.__version = Tracer.__version
         Tracer.__stack.append([dt.datetime.now(), 0.0, self.__label, Tracer.__stack_depth])
         Tracer.__stack_depth += 1
 
     def __exit__(self, *args):
+        if self.__version != Tracer.__version:
+            return
         self.__elapsed = dt.datetime.now() - self.__start
         elapsed_sec = self.__elapsed.seconds + self.__elapsed.microseconds / 1000000
         Tracer.__stack[self.__index][1] = elapsed_sec
@@ -84,6 +88,10 @@ def __exit__(self, *args):
 
     @staticmethod
     def reset():
+        if Tracer.__stack_depth != 0:
+            _logger.warning('Attempted to reset unfinished Trace!')
+            Tracer.__version += 1
+
         Tracer.__stack_depth = 0
         Tracer.__stack = []
 

gs_quant/markets/securities.py

@@ -231,6 +231,7 @@ class SecurityIdentifier(EntityIdentifier):
     PRIMEID = "primeId"
     BBG = "bbg"
     ASSET_ID = "assetId"
+    ANY = "identifiers"
 
 
 class ReturnType(Enum):
@@ -820,8 +821,9 @@ class Bond(Asset):
     def __init__(self,
                  id_: str,
                  name: str,
+                 asset_class: AssetClass = AssetClass.Credit,
                  entity: Optional[Dict] = None):
-        Asset.__init__(self, id_, AssetClass.Credit, name, entity=entity)
+        Asset.__init__(self, id_, asset_class, name, entity=entity)
 
     def get_type(self) -> AssetType:
         return AssetType.BOND
@@ -1141,7 +1143,7 @@ def __gs_asset_to_asset(cls, gs_asset: GsAsset) -> Asset:
             return CommodityPowerAggregatedNodes(gs_asset.id, gs_asset.name, entity=asset_entity)
 
         if asset_type in (GsAssetType.Bond.value,):
-            return Bond(gs_asset.id, gs_asset.name, entity=asset_entity)
+            return Bond(gs_asset.id, gs_asset.name, gs_asset.assetClass or AssetClass.Credit, entity=asset_entity)
 
         if asset_type in (GsAssetType.Commodity.value,):
             return Commodity(gs_asset.id, gs_asset.name, entity=asset_entity)
@@ -1504,7 +1506,7 @@ def get_asset_id_type(type_: SecurityIdentifier):
 
         assert cls._source == SecurityMasterSource.SECURITY_MASTER
         params = {
-            'identifiers': list(ids),
+            input_type.value: list(ids),
             'toIdentifiers': [identifier.value for identifier in output_types],
             'compact': True
         }
@@ -1519,7 +1521,6 @@ def get_asset_id_type(type_: SecurityIdentifier):
         if end_date is not None:
             params['endDate'] = end_date
         r = _get_with_retries('/markets/securities/map', params)
-        # TODO: pass input_type along (after updating endpoint implementation)
 
         results = r['results']
         if isinstance(results, dict):

gs_quant/test/markets/test_securities.py

@@ -885,6 +885,94 @@ def test_map_identifiers_empty(mocker):
     assert actual == {}
 
 
+def test_secmaster_map_identifiers_with_passed_input_types(mocker):
+    start = str(dt.date(2021, 10, 11))
+    end = str(dt.date(2021, 10, 12))
+
+    def mock_mapping_service_response_by_input_type(*args, **kwargs):
+        '''
+        Mocks Secmaster api's response json based on payload's input_type, output_type, and ids provided
+        '''
+        input_type = None
+        for enum in SecurityIdentifier:
+            if enum.value in kwargs['payload']:
+                input_type = enum.value
+                break
+        output_types = kwargs['payload']['toIdentifiers']
+
+        mock_output = {'results': []}
+        for id in kwargs['payload'][input_type]:
+            for output_type in output_types:
+                row = {
+                    "outputType": output_type,
+                    "outputValue": "mock output for " + id,
+                    "startDate": start,
+                    "endDate": end,
+                    "input": id
+                }
+                if output_type in (SecurityIdentifier.BBID, SecurityIdentifier.BBG, SecurityIdentifier.BCID):
+                    row['exchange'] = 'mock-exchange'
+                    row['compositeExchange'] = 'mock-comp'
+                mock_output['results'].append(row)
+        return mock_output
+
+    mocker.patch.object(GsSession.current, '_get',
+                        side_effect=mock_mapping_service_response_by_input_type)
+
+    with SecMasterContext():
+        mock_any_ids = ["mock-any-1", "mock-any-2"]
+        any_to_cusip_results = SecurityMaster.map_identifiers(input_type=SecurityIdentifier.ANY, ids=mock_any_ids,
+                                                              output_types=[SecurityIdentifier.CUSIP])
+        assert start in any_to_cusip_results.keys()
+        for input_id in mock_any_ids:
+            assert input_id in any_to_cusip_results[start].keys()
+            assert SecurityIdentifier.CUSIP.value in any_to_cusip_results[start][input_id].keys()
+        assert any_to_cusip_results == {
+            "2021-10-11": {
+                "mock-any-1": {
+                    "cusip": "mock output for mock-any-1"
+                },
+                "mock-any-2": {
+                    "cusip": "mock output for mock-any-2"
+                }
+            },
+            "2021-10-12": {
+                "mock-any-1": {
+                    "cusip": "mock output for mock-any-1"
+                },
+                "mock-any-2": {
+                    "cusip": "mock output for mock-any-2"
+                }
+            }
+        }
+
+        mock_cusip_ids = ["mock-cusip-input1", "mock-cusip-input2"]
+        cusip_to_isin_result = SecurityMaster.map_identifiers(input_type=SecurityIdentifier.CUSIP, ids=mock_cusip_ids,
+                                                              output_types=[SecurityIdentifier.ISIN])
+        assert start in cusip_to_isin_result.keys()
+        for cusip_input_id in mock_cusip_ids:
+            assert cusip_input_id in cusip_to_isin_result[start].keys()
+            assert SecurityIdentifier.ISIN.value in cusip_to_isin_result[start][cusip_input_id].keys()
+        assert cusip_to_isin_result == {
+            "2021-10-11": {
+                "mock-cusip-input1": {
+                    "isin": "mock output for mock-cusip-input1"
+                },
+                "mock-cusip-input2": {
+                    "isin": "mock output for mock-cusip-input2"
+                }
+            },
+            "2021-10-12": {
+                "mock-cusip-input1": {
+                    "isin": "mock output for mock-cusip-input1"
+                },
+                "mock-cusip-input2": {
+                    "isin": "mock output for mock-cusip-input2"
+                }
+            }
+        }
+
+
 def test_map_identifiers_asset_service(mocker):
     response = {'AAPL UN': ['AAPL.N'], 'GS UN': ['GS.N']}
     mocker.patch.object(GsAssetApi, 'map_identifiers', side_effect=lambda *arg, **kwargs: response)