markets changes, bug fixes, tests (#188)

gs_quant/analytics/core/processor.py

@@ -14,6 +14,7 @@
 under the License.
 """
 import asyncio
+import functools
 import logging
 import uuid
 from abc import ABCMeta, abstractmethod
@@ -57,6 +58,13 @@ class DataQueryInfo:
  data: Series = None
 
 
+@dataclass
+class MeasureQueryInfo:
+ attr: str
+ processor: 'BaseProcessor'
+ entity: Entity
+
+
 DateOrDatetimeOrRDate = Union[DateOrDatetime, RelativeDate]
 
 
@@ -70,6 +78,7 @@ def __init__(self, **kwargs):
  self.children_data: Dict[str, ProcessorResult] = {}
  self.data_cell = None
  self.last_value = kwargs.get('last_value', False)
+ self.measure_processor = kwargs.get('measure_processor', False)
 
  @abstractmethod
  def process(self, *args):
@@ -86,7 +95,7 @@ def __handle_date_range(self,
  result,
  rdate_entity_map: Dict[str, date]):
  """
- Applies a date/datetime mask on the result using the start/end parameters on a processoor
+ Applies a date/datetime mask on the result using the start/end parameters on a processor
  :param result:
  :param rdate_entity_map: map of entity, rule, base_date to date value
  :return: None
@@ -125,23 +134,32 @@ async def update(self,
  attribute: str,
  result: ProcessorResult,
  rdate_entity_map: Dict[str, date],
- pool: ProcessPoolExecutor = None):
+ pool: ProcessPoolExecutor = None,
+ query_info: Union[DataQueryInfo, MeasureQueryInfo] = None):
  """ Handle the update of a single coordinate and recalculate the value
 
  :param attribute: Attribute alinging to data coordinate in the processor
  :param result: Processor result including success and series from data query
  """
- self.__handle_date_range(result, rdate_entity_map)
+ if not self.measure_processor:
+ self.__handle_date_range(result, rdate_entity_map)
  self.children_data[attribute] = result
 
  if isinstance(result, ProcessorResult):
  if result.success:
  try:
  if pool:
- value = await asyncio.get_running_loop().run_in_executor(pool, self.process)
+ if self.measure_processor:
+ value = await asyncio.get_running_loop()\
+ .run_in_executor(pool, functools.partial(self.process, query_info.entity))
+ else:
+ value = await asyncio.get_running_loop().run_in_executor(pool, self.process)
  self.value = value
  else:
- self.process()
+ if self.measure_processor:
+ self.process(query_info.entity)
+ else:
+ self.process()
  self.post_process()
  except Exception as e:
  self.value = ProcessorResult(False,
@@ -168,7 +186,7 @@ def __add_required_rdates(self, entity: Entity, rdate_entity_map: Dict[str, Set[
  def build_graph(self,
  entity: Entity,
  cell,
- queries: List[DataQueryInfo],
+ queries: List[Union[DataQueryInfo, MeasureQueryInfo]],
  rdate_entity_map: Dict[str, Set[Tuple]],
  overrides: Optional[List]):
  """ Generates the nested cell graph and keeps a map of leaf data queries to processors"""
@@ -177,6 +195,9 @@ def build_graph(self,
 
  attributes = self.__dict__
 
+ if self.measure_processor:
+ queries.append(MeasureQueryInfo(attr='a', processor=self, entity=entity))
+
  for attr_name, child in self.children.items():
  if isinstance(child, DataCoordinate):
  # Override coordinate dimensions
@@ -212,14 +233,15 @@ async def calculate(self,
  attribute: str,
  result: ProcessorResult,
  rdate_entity_map: Dict[str, date],
- pool: ProcessPoolExecutor = None):
+ pool: ProcessPoolExecutor = None,
+ query_info: Union[DataQueryInfo, MeasureQueryInfo] = None):
  """ Sets the result on the processor and recursively calls parent to set and calculate value
 
  :param attribute: Attribute alinging to data coordinate in the processor
  :param result: Processor result including success and series from data query
  """
  # update the result
- await self.update(attribute, result, rdate_entity_map, pool)
+ await self.update(attribute, result, rdate_entity_map, pool, query_info)
 
  # if there is a parent, traverse up and recompute
  if self.parent:

gs_quant/analytics/core/query_helpers.py

@@ -21,6 +21,7 @@
 
 from pandas import DataFrame, to_datetime
 
+from gs_quant.analytics.core.processor import MeasureQueryInfo
 from gs_quant.analytics.core.processor_result import ProcessorResult
 from gs_quant.data import DataFrequency
 from gs_quant.session import GsSession
@@ -31,6 +32,8 @@
 def aggregate_queries(query_infos):
  mappings = defaultdict(dict) # DataSet -> start/end
  for query_info in query_infos:
+ if isinstance(query_info, MeasureQueryInfo):
+ continue
  query = query_info.query
  coordinate = query.coordinate
  dataset_id = coordinate.dataset_id

gs_quant/analytics/datagrid/data_cell.py

@@ -16,13 +16,13 @@
 
 import copy
 import uuid
-from typing import List, Optional, Dict, Set, Tuple
+from typing import List, Optional, Dict, Set, Tuple, Union
 
 from pandas import Series
 
 from gs_quant.analytics.common import DATA_CELL_NOT_CALCULATED
 from gs_quant.analytics.core import BaseProcessor
-from gs_quant.analytics.core.processor import DataQueryInfo
+from gs_quant.analytics.core.processor import DataQueryInfo, MeasureQueryInfo
 from gs_quant.analytics.core.processor_result import ProcessorResult
 from gs_quant.analytics.datagrid import Override
 from gs_quant.analytics.datagrid.utils import get_utc_now
@@ -61,7 +61,8 @@ def __init__(self,
  # Store the cell data queries
  self.data_queries: List[DataQueryInfo] = []
 
- def build_cell_graph(self, all_queries: List[DataQueryInfo], rdate_entity_map: Dict[str, Set[Tuple]]) -> None:
+ def build_cell_graph(self, all_queries: List[Union[DataQueryInfo, MeasureQueryInfo]],
+ rdate_entity_map: Dict[str, Set[Tuple]]) -> None:
  """ Generate and store the cell graph and data queries
 
  This can be modified to return the data queries rather than store on the cell

gs_quant/analytics/datagrid/datagrid.py

@@ -29,7 +29,7 @@
 from gs_quant.analytics.common import DATAGRID_HELP_MSG
 from gs_quant.analytics.common.helpers import resolve_entities, get_entity_rdate_key, get_entity_rdate_key_from_rdate, \
  get_rdate_cache_key
-from gs_quant.analytics.core.processor import DataQueryInfo
+from gs_quant.analytics.core.processor import DataQueryInfo, MeasureQueryInfo
 from gs_quant.analytics.core.processor_result import ProcessorResult
 from gs_quant.analytics.core.query_helpers import aggregate_queries, fetch_query, build_query_string, valid_dimensions
 from gs_quant.analytics.datagrid.data_cell import DataCell
@@ -125,7 +125,7 @@ def __init__(self,
  # store the graph, data queries to leaf processors and results
  self._primary_column_index: int = kwargs.get('primary_column_index', 0)
  self._cells: List[DataCell] = []
- self._data_queries: List[DataQueryInfo] = []
+ self._data_queries: List[Union[DataQueryInfo, MeasureQueryInfo]] = []
  self._entity_cells: List[DataCell] = []
  self._coord_processor_cells: List[DataCell] = []
  self._value_cells: List[DataCell] = []
@@ -152,7 +152,7 @@ def initialize(self) -> None:
 
  Upon providing data to a leaf, the leaf processor is calculated and propagated up the graph to the cell level.
  """
- all_queries: List[DataQueryInfo] = []
+ all_queries: List[Union[DataQueryInfo, MeasureQueryInfo]] = []
  entity_cells: List[DataCell] = []
  current_row_group = None
 
@@ -201,6 +201,8 @@ def initialize(self) -> None:
  cell.processor.children['a'].set_dimensions(data_overrides[-1].dimensions)
 
  self._coord_processor_cells.append(cell)
+ elif column_processor.measure_processor:
+ all_queries.append(MeasureQueryInfo(attr='', entity=entity, processor=column_processor))
  else:
  # append the required queries to the map
  cell.build_cell_graph(all_queries, self.rdate_entity_map)
@@ -342,7 +344,8 @@ def _resolve_queries(self, availability_cache: Dict = None) -> None:
 
  for query in self._data_queries:
  entity = query.entity
- if isinstance(entity, str): # If we were unable to fetch entity (404/403)
+ if isinstance(entity, str) or isinstance(query, MeasureQueryInfo):
+ # If we were unable to fetch entity (404/403) or if we're processing a measure processor
  continue
  query = query.query
  coord = query.coordinate
@@ -397,7 +400,13 @@ def _fetch_queries(self):
  query_info.data = Series()
 
  for query_info in self._data_queries:
- if query_info.data is None or len(query_info.data) == 0:
+ if isinstance(query_info, MeasureQueryInfo):
+ asyncio.get_event_loop().run_until_complete(
+ query_info.processor.calculate(query_info.attr,
+ ProcessorResult(True, None),
+ self.rule_cache,
+ query_info=query_info))
+ elif query_info.data is None or len(query_info.data) == 0:
  asyncio.get_event_loop().run_until_complete(
  query_info.processor.calculate(query_info.attr,
  ProcessorResult(False,

gs_quant/analytics/processors/__init__.py

@@ -16,9 +16,9 @@
 
 from .analysis_processors import DiffProcessor
 from .econometrics_processors import SharpeRatioProcessor, VolatilityProcessor, CorrelationProcessor, ChangeProcessor, \
- ReturnsProcessor, BetaProcessor
+ ReturnsProcessor, BetaProcessor, FXImpliedCorrProcessor
 from .special_processors import EntityProcessor, CoordinateProcessor
-from .statistics_processors import PercentilesProcessor, PercentileProcessor, \
+from .statistics_processors import PercentilesProcessor, PercentileProcessor, StdMoveProcessor, \
  CovarianceProcessor, ZscoresProcessor, MeanProcessor, VarianceProcessor, SumProcessor, StdDevProcessor
 from .utility_processors import LastProcessor, AppendProcessor, AdditionProcessor, SubtractionProcessor, \
  MultiplicationProcessor, DivisionProcessor, MinProcessor, MaxProcessor, NthLastProcessor

gs_quant/analytics/processors/econometrics_processors.py

@@ -21,12 +21,14 @@
 from gs_quant.analytics.core.processor import BaseProcessor, DataCoordinateOrProcessor, DataQueryInfo, \
  DateOrDatetimeOrRDate
 from gs_quant.analytics.core.processor_result import ProcessorResult
+from gs_quant.analytics.processors.special_processors import MeasureProcessor
 from gs_quant.data.coordinate import DataCoordinate
 from gs_quant.data.query import DataQuery
 from gs_quant.entities.entity import Entity
-from gs_quant.markets.securities import Stock
+from gs_quant.markets.securities import Stock, Cross
 from gs_quant.target.common import Currency
-from gs_quant.timeseries import correlation, Window, SeriesType, DataMeasure, DataFrequency
+from gs_quant.timeseries import correlation, Window, SeriesType, DataMeasure, DataFrequency, DataContext, \
+ fx_implied_correlation
 from gs_quant.timeseries import excess_returns_pure
 from gs_quant.timeseries.econometrics import get_ratio_pure, SharpeAssets, change, returns
 from gs_quant.timeseries.econometrics import volatility, Returns, beta
@@ -349,3 +351,47 @@ def process(self):
 
  def get_plot_expression(self):
  pass
+
+
+class FXImpliedCorrProcessor(MeasureProcessor):
+
+ def __init__(self,
+ *,
+ cross2: Entity = None,
+ tenor: str = '3m',
+ start: Optional[DateOrDatetimeOrRDate] = None,
+ end: Optional[DateOrDatetimeOrRDate] = None,
+ **kwargs):
+ """ CorrelationProcessor
+
+ :param a: DataCoordinate or BaseProcessor for the series
+ :param benchmark: benchmark to compare price series
+ :param start: start date or time used in the underlying data query
+ :param end: end date or time used in the underlying data query
+ :param w: Window, int, or str: size of window and ramp up to use. e.g. Window(22, 10) where 22 is the window
+ size and 10 the ramp up value. If w is a string, it should be a relative date like '1m', '1d', etc.
+ Window size defaults to length of series.
+ :param type_: type of both input series: prices or returns
+ """
+ super().__init__(**kwargs, )
+ self.cross2: Entity = cross2
+ self.tenor: str = tenor
+ # datetime
+ self.start = start
+ self.end = end
+
+ def process(self, cross1: Entity) -> ProcessorResult:
+ if isinstance(cross1, Cross) and isinstance(self.cross2, Cross):
+ try:
+ with DataContext(self.start, self.end):
+ result = fx_implied_correlation(cross1, self.cross2, self.tenor)
+ self.value = ProcessorResult(True, result)
+ except Exception as e:
+ self.value = ProcessorResult(False, str(e))
+ else:
+ self.value = ProcessorResult(False, "Processor does not have valid crosses as inputs")
+
+ return self.value
+
+ def get_plot_expression(self):
+ pass

gs_quant/analytics/processors/special_processors.py

@@ -95,3 +95,14 @@ def update(self, attribute: str, result: ProcessorResult):
 
  def get_plot_expression(self):
  pass
+
+
+class MeasureProcessor(BaseProcessor):
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs, measure_processor=True)
+
+ def process(self, *args):
+ pass
+
+ def get_plot_expression(self):
+ pass

gs_quant/analytics/processors/statistics_processors.py

@@ -20,6 +20,7 @@
 
 from gs_quant.analytics.core.processor import BaseProcessor, DataCoordinateOrProcessor, DateOrDatetimeOrRDate
 from gs_quant.analytics.core.processor_result import ProcessorResult
+from gs_quant.timeseries import returns
 from gs_quant.timeseries.statistics import percentiles, percentile, Window, mean, sum_, std, var, cov, zscores
 
 
@@ -440,3 +441,53 @@ def process(self):
 
  def get_plot_expression(self):
  pass
+
+
+class StdMoveProcessor(BaseProcessor):
+ def __init__(self,
+ a: DataCoordinateOrProcessor,
+ *,
+ start: Optional[DateOrDatetimeOrRDate] = None,
+ end: Optional[DateOrDatetimeOrRDate] = None,
+ w: Union[Window, int] = None,
+ **kwargs):
+ """ StdMoveProcessor: Returns normalized by std deviation of series a
+
+ :param a: DataCoordinate or BaseProcessor for the first series
+ :param start: start date or time used in the underlying data query
+ :param end: end date or time used in the underlying data query
+ :param w: Window or int: size of window and ramp up to use. e.g. Window(22, 10) where 22 is the window size
+ and 10 the ramp up value. If w is a string, it should be a relative date like '1m', '1d', etc.
+ Window size defaults to length of series.
+ """
+ super().__init__(**kwargs)
+ self.children['a'] = a
+
+ self.start = start
+ self.end = end
+ self.w = w
+
+ def process(self):
+ a_data = self.children_data.get('a')
+ if isinstance(a_data, ProcessorResult):
+ if a_data.success:
+ data_series = a_data.data
+ change_pd = data_series.tail(2)
+ change = returns(change_pd).iloc[-1]
+
+ # Pass in all values except last value (which is last value)
+ returns_series = returns(data_series.head(-1))
+ std_result = std(returns_series, w=Window(None, 0) if self.w is None else self.w).iloc[-1]
+
+ if change is not None and std_result != 0:
+ self.value = ProcessorResult(True, pd.Series([change / std_result]))
+ else:
+ self.value = ProcessorResult(False, "StdMoveProcessor returns a NaN")
+ else:
+ self.value = ProcessorResult(False, "StdMoveProcessor does not have 'a' series values yet")
+ else:
+ self.value = ProcessorResult(False, "StdMoveProcessor does not have 'a' series yet")
+ return self.value
+
+ def get_plot_expression(self):
+ pass

gs_quant/api/gs/data.py

@@ -135,6 +135,8 @@ class QueryType(Enum):
  USD_OIS = "Usd Ois"
  NON_USD_OIS = "Non Usd Ois"
  SETTLEMENT_PRICE = "Settlement Price"
+ THEMATIC_EXPOSURE = "Thematic Exposure"
+ THEMATIC_BETA = "Thematic Beta"
 
 
 class GsDataApi(DataApi):