from enum import IntEnum
from derivitec.market_data import EquityVolSurface
from derivitec.market_data import FxVolSurface
from derivitec.market_data import EquityVolIndexVolSurface
from derivitec.core.dict_builder_abstract import DictBuilderAbstract
[docs]
class RiskEngineType(DictBuilderAbstract):
_dotnet_namespace = ""
[docs]
class RiskResult(RiskEngineType):
[docs]
def __init__(self,
bond_issuer=None,
commodity_id=None,
contract=None,
currency=None,
curve_name=None,
equity_asset=None,
error_message=None,
fx_pair=None,
measure_id=None,
rate_asset_id=None,
scenario_key=None,
tenor=None,
tradable_id=None,
unit_value=None,
unit_value_currency=None,
unit_value_is_valid=None):
"""
:param bond_issuer: The bond issuer
:param commodity_id: The commodity id
:param contract: The contract
:param currency: The currency
:param curve_name: The curve name
:param equity_asset: The equity asset
:param error_message: The error message
:param fx_pair: The fx pair
:param measure_id: The risk measure id
:param rate_asset_id: The rate asset id
:param scenario_key: The scenario key
:param tenor: The tenor
:param tradable_id: The tradable id
:param unit_value: The unit value
:param unit_value_currency: The unit value currency
:param unit_value_is_valid: The unit value is valid
"""
self.bond_issuer = bond_issuer
self.commodity_id = commodity_id
self.contract = contract
self.currency = currency
self.curve_name = curve_name
self.equity_asset = equity_asset
self.error_message = error_message
self.fx_pair = fx_pair
self.measure_id = measure_id
self.rate_asset_id = rate_asset_id
self.scenario_key = scenario_key
self.tenor = tenor
self.tradable_id = tradable_id
self.unit_value = unit_value
self.unit_value_currency = unit_value_currency
self.unit_value_is_valid = unit_value_is_valid
def json_params(self):
params = [
'bond_issuer',
'commodity_id',
'contract',
'currency',
'curve_name',
'equity_asset',
'error_message',
'fx_pair',
'measure_id',
'rate_asset_id',
'scenario_key',
'tenor',
'tradable_id',
'unit_value',
'unit_value_currency',
'unit_value_is_valid',
]
return params
[docs]
class RiskMeasure(RiskEngineType):
[docs]
def __init__(self,
risk_measure_id_stored=None):
"""
:param risk_measure_id_stored: The risk measure id stored
"""
self.risk_measure_id_stored = risk_measure_id_stored
def json_params(self):
params = [
'risk_measure_id_stored',
]
return params
[docs]
class RiskScenario(RiskEngineType):
[docs]
def __init__(self,
scenario_description=None,
scenario_name=None):
"""
:param scenario_description: The scenario description
:param scenario_name: The scenario name
"""
self.scenario_description = scenario_description
self.scenario_name = scenario_name
def json_params(self):
params = [
'scenario_description',
'scenario_name',
]
return params
[docs]
class RiskCalculationResult(RiskEngineType):
[docs]
def __init__(self,
error_messages=None,
market_data_error_summary=None,
risk_results=None,
error_messages_by_security_id=None,
**kwargs):
"""
:param error_messages: The error messages
:param market_data_error_summary: The market data error summary
:param risk_results: The risk results
:param error_messages_by_security_id: Collects error messages that can be attributed to individual securities
"""
self.error_messages = error_messages
self.market_data_error_summary = market_data_error_summary
self.risk_results = risk_results
self.error_messages_by_security_id = error_messages_by_security_id
def json_params(self):
params = [
'error_messages',
'market_data_error_summary',
'risk_results',
'error_messages_by_security_id',
]
return params
[docs]
class CompositeScenario(RiskScenario):
[docs]
def __init__(self,
scenario_description=None,
scenario_name=None,
scenarios=None):
"""
:param scenario_description: The scenario description
:param scenario_name: The scenario name
:param scenarios: The scenarios
"""
self.scenarios = scenarios
super(CompositeScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'scenarios',
]
params += super(CompositeScenario, self).json_params()
return params
[docs]
class EqSpotScenario(RiskScenario):
[docs]
def __init__(self,
scenario_description=None,
scenario_name=None,
shock_size=0,
shocks_by_equity_asset=None,
shocks_by_country=None,
shocks_by_region=None,
vol_surface_spot_model=EquityVolSurface.SpotModel.StickyDelta,
sticky_smile_ratio=0.0,
bump_cash_dividends=False,
equity_vol_index_vol_surface_spot_model_id=None,
vol_surface_spot_model_definitions=None):
"""
:param scenario_description: The scenario description
:param scenario_name: The scenario name
:param shock_size: The shock size
:param shocks_by_equity_asset: The shocks by equity asset (overrides the shocks by country)
:param shocks_by_country: The shocks by country (overrides the shocks by region)
:param shocks_by_region: The shocks by region (overrides the global shock)
:param vol_surface_spot_model: The vol surface spot model
:param sticky_smile_ratio: The sticky smile ratio
:param bump_cash_dividends: Whether the equity spot shock should also bump cash dividends by the same amount
:param equity_vol_index_vol_surface_spot_model_id: Exact user-defined model_id used when uploading
volatility-index spot-model betas (for example "ExampleVixBetas"). Definitions are selected for
the scenario market-data date and portfolio volatility indices using the market-data provider
priority. None, empty or whitespace disables database loading; inline definitions still apply.
:param vol_surface_spot_model_definitions: Optional list of VolSurfaceSpotModelDefinition objects.
Each replaces the entire stored definition for its resolved volatility-index asset or adds
a new asset. Duplicate assets are rejected. The inline model_id is a label and need not match
the base ID. Definitions are scenario-only: nothing is uploaded or saved. None or [] adds no overrides.
"""
self.shock_size = shock_size
self.shocks_by_equity_asset = shocks_by_equity_asset
self.shocks_by_country = shocks_by_country
self.shocks_by_region = shocks_by_region
self.vol_surface_spot_model = vol_surface_spot_model
self.sticky_smile_ratio = sticky_smile_ratio
self.bump_cash_dividends = bump_cash_dividends
self.equity_vol_index_vol_surface_spot_model_id = equity_vol_index_vol_surface_spot_model_id
self.vol_surface_spot_model_definitions = vol_surface_spot_model_definitions
super(EqSpotScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'shock_size',
'shocks_by_equity_asset',
'shocks_by_country',
'shocks_by_region',
'vol_surface_spot_model',
'sticky_smile_ratio',
'bump_cash_dividends',
'equity_vol_index_vol_surface_spot_model_id',
'vol_surface_spot_model_definitions'
]
params += super(EqSpotScenario, self).json_params()
return params
def to_dict(self):
param_dict = super(EqSpotScenario, self).to_dict()
# shocks_by_country and shocks_by_region are str -> float maps, which
# DictBuilderAbstract does not automatically convert to the API's
# key/value-list convention (only the InstrumentIdentifier -> float
# equity-asset map is handled there). Convert them manually here, the
# same way MarketDataOverride.to_dict() handles fx_spot_overrides_by_currency.
for param in ('shocks_by_country', 'shocks_by_region'):
if param in param_dict.keys():
shocks = param_dict[param]
if isinstance(shocks, dict):
shocks_list = []
for k, v in shocks.items():
shocks_list.append({'key': k, 'value': v})
param_dict.update({param: shocks_list})
return param_dict
[docs]
class EqVolScenario(RiskScenario):
[docs]
def __init__(self,
bumped_vol_floor=0,
scenario_description=None,
scenario_name=None,
shock_size=0,
use_atmf_rescale_factor=False,
use_proportional_vol_bump=False):
"""
:param bumped_vol_floor: ATMF volatilities are floored at this value
:param scenario_description: The scenario description
:param scenario_name: The scenario name
:param shock_size: The shock size
:param use_atmf_rescale_factor: Rescales skew and smile parameters to make vol surface shock more parallel
:param use_proportional_vol_bumps: Set to true to specify proportional shocks.
"""
self.bumped_vol_floor = bumped_vol_floor
self.shock_size = shock_size
self.use_atmf_rescale_factor = use_atmf_rescale_factor
self.use_proportional_vol_bump = use_proportional_vol_bump
super(EqVolScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'shock_size',
'bumped_vol_floor',
'use_atmf_rescale_factor',
'use_proportional_vol_bump'
]
params += super(EqVolScenario, self).json_params()
return params
[docs]
class DividendIndexVolScenario(RiskScenario):
[docs]
def __init__(self,
shock_size=0.01,
use_proportional_vol_bump=False,
use_parallel_shift=True,
bumped_vol_floor=-.0,
vol_shock_func=None,
shocks_by_dividend_index=None,
scenario_name='DividendIndexVolScenario',
scenario_description=''):
"""
:param shock_size: The shock size
:param use_proportional_vol_bump: Set to true to specify proportional shocks.
:param use_parallel_shift: Set to true to specify parallel shifts.
:param bumped_vol_floor: ATMF volatilities are floored at this value
:param vol_shock_func: The shock function
:param shocks_by_dividend_index: The shocks by dividend index
:param scenario_name: The scenario name
:param scenario_description: The scenario description
"""
self.shock_size = shock_size
self.use_proportional_vol_bump = use_proportional_vol_bump
self.use_parallel_shift = use_parallel_shift
self.bumped_vol_floor = bumped_vol_floor
self.vol_shock_func = vol_shock_func
self.shocks_by_dividend_index = shocks_by_dividend_index
super(DividendIndexVolScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'shock_size',
'use_proportional_vol_bump',
'use_parallel_shift',
'bumped_vol_floor',
'vol_shock_func',
'shocks_by_dividend_index',
]
params += super(DividendIndexVolScenario, self).json_params()
return params
[docs]
class EqVolSurfaceScenario(RiskScenario):
[docs]
def __init__(self,
global_shock=None,
shocks_by_asset=None,
shock_equity_vol_indicies=False,
scenario_name=None,
scenario_description=None):
"""
:param global_shock: The global shock of type EqVolSurfaceBump
:param shocks_by_asset: The shocks by asset of type Dict<InstrumentIdentifier, EqVolSurfaceBump>
:param shock_equity_vol_indicies: The shock equity vol indicies flag
:param scenario_name: The scenario name
"""
self.global_shock = global_shock
self.shocks_by_asset = shocks_by_asset
self.shock_equity_vol_indicies = shock_equity_vol_indicies
super(EqVolSurfaceScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'global_shock',
'shocks_by_asset',
'shock_equity_vol_indicies'
]
params += super(EqVolSurfaceScenario, self).json_params()
return params
[docs]
class EqVolIndexVolSurfaceScenario(RiskScenario):
[docs]
def __init__(self,
global_shock=None,
shocks_by_vol_index=None,
scenario_name=None,
scenario_description=None):
"""
:param global_shock: The global shock of type EqVolIndexVolSurfaceBump
:param shocks_by_vol_index: The shocks by vol index of type
Dict<InstrumentIdentifier, EqVolIndexVolSurfaceBump>
:param scenario_name: The scenario name
:param scenario_description: The scenario description
"""
self.global_shock = global_shock
self.shocks_by_vol_index = shocks_by_vol_index
super(EqVolIndexVolSurfaceScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'global_shock',
'shocks_by_vol_index',
]
params += super(EqVolIndexVolSurfaceScenario, self).json_params()
return params
[docs]
class EquityVolIndexScenario(RiskScenario):
[docs]
def __init__(self,
shocks_by_vol_index=None,
vol_surface_spot_model=None,
vol_indices_to_remove=None,
scenario_name=None,
scenario_description=None):
"""
:param shocks_by_vol_index: Dict<InstrumentIdentifier, EqVolSurfaceBump>
:param vol_surface_spot_model: The vol surface spot model
:param vol_indices_to_remove: List of vol indices to remove
:param scenario_name: The scenario name
:param scenario_description: The scenario description
"""
self.shocks_by_vol_index = shocks_by_vol_index
self.vol_surface_spot_model = vol_surface_spot_model
self.vol_indices_to_remove = vol_indices_to_remove
super(EquityVolIndexScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name
)
def json_params(self):
params = [
'shocks_by_vol_index',
'vol_surface_spot_model',
'vol_indices_to_remove'
]
params += super(EquityVolIndexScenario, self).json_params()
return params
[docs]
class ThetaScenario(RiskScenario):
[docs]
def __init__(self,
roll_days=0,
scenario_description=None,
scenario_name=None):
"""
:param roll_days: The number of days the pricing date is rolled forward
:param scenario_description: The scenario description
:param scenario_name: The scenario name
"""
self.roll_days = roll_days
super(ThetaScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'roll_days',
]
params += super(ThetaScenario, self).json_params()
return params
[docs]
class EqRiskMeasure(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
vol_surface_spot_model=EquityVolSurface.SpotModel.StickyDelta,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param vol_surface_spot_model: The vol surface spot model
:param use_contract_currency: The use contract currency flag
"""
self.vol_surface_spot_model = vol_surface_spot_model
self.use_contract_currency = use_contract_currency
super(EqRiskMeasure, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['vol_surface_spot_model', 'use_contract_currency']
params += super(EqRiskMeasure, self).json_params()
return params
[docs]
class FxRiskMeasure(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta):
"""
:param risk_measure_id_stored: The risk measure id stored
:param vol_surface_spot_model: The vol surface spot model
"""
self.vol_surface_spot_model = vol_surface_spot_model
super(FxRiskMeasure, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['vol_surface_spot_model']
params += super(FxRiskMeasure, self).json_params()
return params
[docs]
class IrRiskMeasure(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None):
"""
:param risk_measure_id_stored: The risk measure id stored
"""
super(IrRiskMeasure, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = []
params += super(IrRiskMeasure, self).json_params()
return params
[docs]
class CreditRiskMeasure(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None):
"""
:param risk_measure_id_stored: The risk measure id stored
"""
super(CreditRiskMeasure, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = []
params += super(CreditRiskMeasure, self).json_params()
return params
[docs]
class IRParallelShockScenario(RiskScenario):
[docs]
def __init__(self,
global_shock=None,
scenario_description=None,
scenario_name=None):
"""
:param global_shock: The shock size
:param scenario_description: The scenario description
:param scenario_name: The scenario name
"""
self.global_shock = global_shock
super(IRParallelShockScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'global_shock',
]
params += super(IRParallelShockScenario, self).json_params()
return params
[docs]
class IRShockScenario(RiskScenario):
[docs]
def __init__(self,
global_shock=None,
shocks_by_currency=None,
bump_discount_curves=True,
bump_bond_yield_curves=True,
bump_equity_assets=True,
bump_equity_vol_indices=True,
bump_rate_assets=True,
bump_fx_crosses=True,
bump_commodities=True,
scenario_description=None,
scenario_name=None):
"""
:param global_shock: The shock for all currencies
:param shocks_by_currency: The shocks by currency
:param bump_discount_curves: The bump discount curve flag
:param bump_bond_yield_curves: The bump bond yield curve flag
:param bump_equity_assets: The bump equity assets flag
:param bump_equity_vol_indices: The bump equity vol indices flag
:param bump_rate_assets: The bump rate assets flag
:param bump_fx_crosses: The bump fx crosses flag
:param bump_commodities: The bump commodities flag
:param scenario_description: The scenario description
:param scenario_name: The scenario name
"""
self.global_shock = global_shock
self.shocks_by_currency = shocks_by_currency
self.bump_discount_curves = bump_discount_curves
self.bump_bond_yield_curves = bump_bond_yield_curves
self.bump_equity_assets = bump_equity_assets
self.bump_equity_vol_indices = bump_equity_vol_indices
self.bump_rate_assets = bump_rate_assets
self.bump_fx_crosses = bump_fx_crosses
self.bump_commodities = bump_commodities
super(IRShockScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'global_shock',
'shocks_by_currency',
'bump_discount_curves',
'bump_bond_yield_curves',
'bump_equity_assets',
'bump_equity_vol_indices',
'bump_rate_assets',
'bump_fx_crosses',
'bump_commodities',
]
params += super(IRShockScenario, self).json_params()
return params
[docs]
class CmdtyRiskMeasure(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None):
"""
:param risk_measure_id_stored: The risk measure id stored
"""
super(CmdtyRiskMeasure, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = []
params += super(CmdtyRiskMeasure, self).json_params()
return params
[docs]
class FxSpotScenario(RiskScenario):
[docs]
def __init__(self,
currency=None,
scenario_description=None,
scenario_name=None,
shock_size=None,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta):
"""
:param currency: The currency
:param scenario_description: The scenario description
:param scenario_name: The scenario name
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
"""
self.currency = currency
self.shock_size = shock_size
self.vol_surface_spot_model = vol_surface_spot_model
super(FxSpotScenario, self).__init__(
scenario_description=scenario_description,
scenario_name=scenario_name)
def json_params(self):
params = [
'currency',
'shock_size',
'vol_surface_spot_model'
]
params += super(FxSpotScenario, self).json_params()
return params
[docs]
class EqGamma(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
vol_surface_spot_model=EquityVolSurface.SpotModel.StickyDelta,
sticky_smile_ratio=0.0,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
:param sticky_smile_ratio: The sticky smile ratio
:param use_contract_currency: The use contract currency flag
"""
self.shock_size = shock_size
self.sticky_smile_ratio = sticky_smile_ratio
super(EqGamma, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model,
use_contract_currency=use_contract_currency)
def json_params(self):
params = ['shock_size', 'sticky_smile_ratio']
params += super(EqGamma, self).json_params()
return params
[docs]
class IrParallelShock(IrRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
"""
self.shock_size = shock_size
super(IrParallelShock, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
]
params += super(IrParallelShock, self).json_params()
return params
[docs]
class IrVega(IrRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
"""
self.shock_size = shock_size
super(IrVega, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
]
params += super(IrVega, self).json_params()
return params
[docs]
class IrGamma(IrRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.0001):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
"""
self.shock_size = shock_size
super(IrGamma, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
]
params += super(IrGamma, self).json_params()
return params
[docs]
class IrDelta(IrRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.0001,
tenors=None,
use_proportional_shocks=None,
use_forward_rate_bumps=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param tenors: The tenors
:param use_proportional_shocks: The use proportional shocks flag
:param use_forward_rate_bumps: The use forward rate bumps flag
"""
self.shock_size = shock_size
self.tenors = tenors
self.use_proportional_shocks = use_proportional_shocks
self.use_forward_rate_bumps = use_forward_rate_bumps
super(IrDelta, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'tenors',
'use_proportional_shocks',
'use_forward_rate_bumps'
]
params += super(IrDelta, self).json_params()
return params
[docs]
class BorrowCurveDelta(IrRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.0001,
tenors=None,
use_proportional_shocks=None,
use_forward_rate_bumps=None,
use_contract_currency=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param tenors: The tenors
:param use_proportional_shocks: The use proportional shocks flag
:param use_forward_rate_bumps: The use forward rate bumps flag
:param use_contract_currency: The use contract currency flag
"""
self.shock_size = shock_size
self.tenors = tenors
self.use_proportional_shocks = use_proportional_shocks
self.use_forward_rate_bumps = use_forward_rate_bumps
self.use_contract_currency = use_contract_currency
super(BorrowCurveDelta, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'tenors',
'use_proportional_shocks',
'use_forward_rate_bumps',
'use_contract_currency'
]
params += super(BorrowCurveDelta, self).json_params()
return params
[docs]
class BorrowCurveShock(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.1,
use_proportional_shocks=None,
use_forward_rate_bumps=None,
use_contract_currency=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_proportional_shocks: The use proportional shocks flag
:param use_forward_rate_bumps: The use forward rate bumps flag
:param use_contract_currency: The use contract currency flag
"""
self.shock_size = shock_size
self.use_proportional_shocks = use_proportional_shocks
self.use_forward_rate_bumps = use_forward_rate_bumps
self.use_contract_currency = use_contract_currency
super(BorrowCurveShock, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_proportional_shocks',
'use_forward_rate_bumps',
'use_contract_currency'
]
params += super(BorrowCurveShock, self).json_params()
return params
[docs]
class BorrowCurveScenario(RiskScenario):
[docs]
def __init__(self,
global_shock=None,
shocks_by_equity_asset=None,
scenario_name="BorrowCurveScenario",
scenario_description="BorrowCurveScenario"):
"""
:param scenario_name: The scenario name
:param scenario_description: The scenario description
:param global_shock: The global shock
:param shocks_by_equity_asset: The shocks by equity asset
"""
self.scenario_name = scenario_name
self.scenario_description = scenario_description
self.global_shock = global_shock
self.shocks_by_equity_asset = shocks_by_equity_asset
super(BorrowCurveScenario, self).__init__(scenario_name=scenario_name,
scenario_description=scenario_description)
def json_params(self):
params = [
'global_shock',
'shocks_by_equity_asset'
]
params += super(BorrowCurveScenario, self).json_params()
return params
[docs]
class DividendDelta(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
is_all_in_rate_shock=False,
tenors=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param is_all_in_rate_shock: The is all in rate shock
:param tenors: The tenors
"""
self.shock_size = shock_size
self.is_all_in_rate_shock = is_all_in_rate_shock
self.tenors = tenors
super(DividendDelta, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'is_all_in_rate_shock',
'tenors'
]
params += super(DividendDelta, self).json_params()
return params
[docs]
class DividendShock(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.1,
all_in_rate_shock_size=0.01):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param all_in_rate_shock_size: The all in rate shock size
"""
self.shock_size = shock_size
self.all_in_rate_shock_size = all_in_rate_shock_size
super(DividendShock, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'all_in_rate_shock_size'
]
params += super(DividendShock, self).json_params()
return params
[docs]
class DividendScenario(RiskScenario):
[docs]
def __init__(self,
global_shock=None,
shocks_by_equity_asset=None,
scenario_name="DividendScenario",
scenario_description="DividendScenario"):
"""
:param scenario_name: The scenario name
:param scenario_description: The scenario description
:param global_shock: The global shock
:param shocks_by_equity_asset: The shocks by equity asset
"""
self.scenario_name = scenario_name
self.scenario_description = scenario_description
self.global_shock = global_shock
self.shocks_by_equity_asset = shocks_by_equity_asset
super(DividendScenario, self).__init__(scenario_name=scenario_name, scenario_description=scenario_description)
def json_params(self):
params = [
'global_shock',
'shocks_by_equity_asset'
]
params += super(DividendScenario, self).json_params()
return params
[docs]
class SpotThetaModel(IntEnum):
Fixed = 0
ForwardCurve = 1
[docs]
class VolThetaModel(IntEnum):
VolFixedByTenor = 0
VolFixedByDate = 1
[docs]
class Theta(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
roll_days=None,
use_contract_currency=False,
spot_theta_model=SpotThetaModel.Fixed,
volatility_theta_model=VolThetaModel.VolFixedByTenor):
"""
:param risk_measure_id_stored: The risk measure id stored
:param roll_days: The roll days
:param use_contract_currency: If true the measure uses the contract currency instead of the report currency for calculations
:param spot_theta_model: The spot theta model
:param volatility_theta_model: The vol theta model
"""
self.roll_days = roll_days
self.use_contract_currency = use_contract_currency
self.spot_theta_model = spot_theta_model
self.volatility_theta_model = volatility_theta_model
super(Theta, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'roll_days',
'use_contract_currency',
'spot_theta_model',
'volatility_theta_model'
]
params += super(Theta, self).json_params()
return params
[docs]
class FxSpotVolShock(FxRiskMeasure):
[docs]
def __init__(self,
calculate_price_changes=None,
include_reporting_currency_shock=None,
risk_measure_id_stored=None,
spot_shock_size=None,
use_implied_vol_bumps=None,
use_proportional_vol_bumps=None,
vol_shock_size=None,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta):
"""
:param calculate_price_changes: The calculate price changes
:param include_reporting_currency_shock: The include reporting currency shock
:param risk_measure_id_stored: The risk measure id stored
:param spot_shock_size: The spot shock size
:param use_implied_vol_bumps: The use implied vol bumps
:param use_proportional_vol_bumps: The use proportional vol bumps
:param vol_shock_size: The vol shock size
:param vol_surface_spot_model: The vol surface spot model
"""
self.calculate_price_changes = calculate_price_changes
self.include_reporting_currency_shock = include_reporting_currency_shock
self.spot_shock_size = spot_shock_size
self.use_implied_vol_bumps = use_implied_vol_bumps
self.use_proportional_vol_bumps = use_proportional_vol_bumps
self.vol_shock_size = vol_shock_size
super(FxSpotVolShock, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = [
'calculate_price_changes',
'include_reporting_currency_shock',
'spot_shock_size',
'use_implied_vol_bumps',
'use_proportional_vol_bumps',
'vol_shock_size',
]
params += super(FxSpotVolShock, self).json_params()
return params
[docs]
class Npv(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
return_futures_price=False,
return_percent_price=False,
use_contract_currency=False,
return_futures_value=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param return_futures_price: If true the measure returns the futures price instead of NPV for all futures contracts
:param return_percent_price: If true the measure returns 100 * Price / Notional for all contracts
:param use_contract_currency: If true the measure uses the contract currency instead of the report currency for calculations
:param return_futures_value: If true the measure returns the futures value instead of NPV for all futures contracts
"""
self.return_futures_price = return_futures_price
self.return_percent_price = return_percent_price
self.use_contract_currency = use_contract_currency
self.return_futures_value = return_futures_value
super(Npv, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ["return_futures_price",
"return_percent_price",
"use_contract_currency",
"return_futures_value"]
params += super(Npv, self).json_params()
return params
[docs]
class CmdtyGamma(CmdtyRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
"""
self.shock_size = shock_size
self.use_contract_currency = use_contract_currency
super(CmdtyGamma, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_contract_currency'
]
params += super(CmdtyGamma, self).json_params()
return params
[docs]
class Rho(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.0001,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_contract_currency: If true the measure uses the contract currency instead of the report currency for calculations
"""
self.shock_size = shock_size
self.use_contract_currency = use_contract_currency
super(Rho, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_contract_currency'
]
params += super(Rho, self).json_params()
return params
[docs]
class FxVega(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
use_implied_vol_bumps=False,
use_inverse_currency_shock=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_implied_vol_bumps: If set to true vol bumps for surfaces without explicit reporting currency dependency
will be implied from bumps to vol surfaces involving the reporting currency.
:param use_inverse_currency_shock: The use inverse currency shock flag
"""
self.shock_size = shock_size
self.use_implied_vol_bumps = use_implied_vol_bumps
self.use_inverse_currency_shock = use_inverse_currency_shock
super(FxVega, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_implied_vol_bumps',
'use_inverse_currency_shock'
]
params += super(FxVega, self).json_params()
return params
[docs]
class EqDeltaShares(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
vol_surface_spot_model=EquityVolSurface.SpotModel.StickyDelta):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
"""
self.shock_size = shock_size
super(EqDeltaShares, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = ['shock_size']
params += super(EqDeltaShares, self).json_params()
return params
[docs]
class EqVolIndexDeltaShares(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
vol_surface_spot_model=EquityVolIndexVolSurface.SpotModel.StickyDelta):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
"""
self.shock_size = shock_size
self.vol_surface_spot_model = vol_surface_spot_model
super(EqVolIndexDeltaShares, self).__init__(risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['shock_size', 'vol_surface_spot_model']
params += super(EqVolIndexDeltaShares, self).json_params()
return params
[docs]
class EqVolIndexGammaShares(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
vol_surface_spot_model=EquityVolIndexVolSurface.SpotModel.StickyDelta):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
"""
self.shock_size = shock_size
self.vol_surface_spot_model = vol_surface_spot_model
super(EqVolIndexGammaShares, self).__init__(risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['shock_size', 'vol_surface_spot_model']
params += super(EqVolIndexGammaShares, self).json_params()
return params
[docs]
class EqVolIndexVega(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_contract_currency: If true the risk is reported in the contract currency; otherwise in the reporting currency
"""
self.shock_size = shock_size
self.use_contract_currency = use_contract_currency
super(EqVolIndexVega, self).__init__(risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['shock_size', 'use_contract_currency']
params += super(EqVolIndexVega, self).json_params()
return params
[docs]
class EqFairVolDelta(RiskMeasure):
[docs]
def __init__(self, risk_measure_id_stored=None, shock_size=0.01):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
"""
self.shock_size = shock_size
super(EqFairVolDelta, self).__init__(risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['shock_size']
params += super(EqFairVolDelta, self).json_params()
return params
[docs]
class EqFairVolGamma(RiskMeasure):
[docs]
def __init__(self, risk_measure_id_stored=None, shock_size=0.01):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
"""
self.shock_size = shock_size
super(EqFairVolGamma, self).__init__(risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['shock_size']
params += super(EqFairVolGamma, self).json_params()
return params
[docs]
class EqRealisedVol(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
lookback_window=30,
use_percentage_points=True,
decay_period=0):
"""
:param risk_measure_id_stored: The risk measure id stored
:param lookback_window: The length of the look back window (in calendar days)
:param use_percentage_points: If set true realised vols are expressed in percentage points, otherwise they are expressed in currency units.
:param decay_period: If >0 exponential decay weights are used, e.g., yesterday's return weight is exp(-1/decayPeriod).
"""
self.lookback_window = lookback_window
self.use_percentage_points = use_percentage_points
self.decay_period = decay_period
super(EqRealisedVol, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['lookback_window',
'use_percentage_points',
'decay_period']
params += super(EqRealisedVol, self).json_params()
return params
[docs]
class CmdtyRealisedVol(CmdtyRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
lookback_window=30,
use_percentage_points=True,
decay_period=0):
"""
:param risk_measure_id_stored: The risk measure id stored
:param lookback_window: The length of the look back window (in business calendar days)
:param use_percentage_points: If set true realised vols are expressed in percentage points, otherwise they are expressed in currency units.
:param decay_period: If >0 exponential decay weights are used, e.g., yesterday's return weight is exp(-1/decayPeriod).
"""
self.lookback_window = lookback_window
self.use_percentage_points = use_percentage_points
self.decay_period = decay_period
super(CmdtyRealisedVol, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['lookback_window',
'use_percentage_points',
'decay_period']
params += super(CmdtyRealisedVol, self).json_params()
return params
[docs]
class FxRealisedVol(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
lookback_window=30,
use_percentage_points=True,
decay_period=0):
"""
:param risk_measure_id_stored: The risk measure id stored
:param lookback_window: The length of the look back window (in business calendar days)
:param use_percentage_points: If set true realised vols are expressed in percentage points, otherwise they are expressed in currency units.
:param decay_period: If >0 exponential decay weights are used, e.g., yesterday's return weight is exp(-1/decayPeriod).
"""
self.lookback_window = lookback_window
self.use_percentage_points = use_percentage_points
self.decay_period = decay_period
super(FxRealisedVol, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = ['lookback_window',
'use_percentage_points',
'decay_period']
params += super(FxRealisedVol, self).json_params()
return params
[docs]
class FxGamma(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta,
use_inverse_currency_shock=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
:param use_inverse_currency_shock: The use inverse currency shock flag
"""
self.shock_size = shock_size
self.use_inverse_currency_shock = use_inverse_currency_shock
super(FxGamma, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = [
'shock_size',
'use_inverse_currency_shock'
]
params += super(FxGamma, self).json_params()
return params
[docs]
class CmdtyDelta(CmdtyRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_contract_currency: If true the measure uses the contract currency instead of the report currency for calculations
"""
self.shock_size = shock_size
self.use_contract_currency = use_contract_currency
super(CmdtyDelta, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_contract_currency'
]
params += super(CmdtyDelta, self).json_params()
return params
[docs]
class EqDelta(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
vol_surface_spot_model=EquityVolSurface.SpotModel.StickyDelta,
sticky_smile_ratio=0.0,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
:param sticky_smile_ratio: The sticky smile ratio
:param use_contract_currency: The use contract currency flag
"""
self.shock_size = shock_size
self.sticky_smile_ratio = sticky_smile_ratio
super(EqDelta, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model,
use_contract_currency=use_contract_currency)
def json_params(self):
params = [
'shock_size',
'sticky_smile_ratio'
]
params += super(EqDelta, self).json_params()
return params
[docs]
class EqCorrelationDelta(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_contract_currency: The use contract currency flag
"""
self.shock_size = shock_size
self.use_contract_currency = use_contract_currency
super(EqCorrelationDelta, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_contract_currency'
]
params += super(EqCorrelationDelta, self).json_params()
return params
[docs]
class FxDelta(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta,
use_inverse_currency_shock=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
:param use_inverse_currency_shock: The use inverse currency shock flag
"""
self.shock_size = shock_size
self.use_inverse_currency_shock = use_inverse_currency_shock
super(FxDelta, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = [
'shock_size',
'use_inverse_currency_shock'
]
params += super(FxDelta, self).json_params()
return params
[docs]
class FxDeltaByCross(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
:param use_contract_currency: The use contract currency flag
"""
self.shock_size = shock_size
self.use_contract_currency = use_contract_currency
super(FxDeltaByCross, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = [
'shock_size',
'use_contract_currency'
]
params += super(FxDeltaByCross, self).json_params()
return params
[docs]
class EqSpotShock(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
vol_surface_spot_model=EquityVolSurface.SpotModel.StickyDelta,
sticky_smile_ratio=0.0):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
:param sticky_smile_ratio: The sticky smile ratio
"""
self.shock_size = shock_size
self.sticky_smile_ratio = sticky_smile_ratio
super(EqSpotShock, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = [
'shock_size',
'sticky_smile_ratio'
]
params += super(EqSpotShock, self).json_params()
return params
[docs]
class EqVega(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
use_proportional_vol_bumps=False,
use_parallel_shift=True,
shock_equity_vol_indices=True,
use_contract_currency=False,
tenors=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock_size
:param use_proportional_vol_bumps: The use_proportional_vol_bumps
:param use_parallel_shift: The use_parallel_shift
:param shock_equity_vol_indices: The shock_equity_vol_indices
:param use_contract_currency: The use_contract_currency
:param tenors: The tenors
"""
self.shock_size = shock_size
self.use_proportional_vol_bumps = use_proportional_vol_bumps
self.use_parallel_shift = use_parallel_shift
self.shock_equity_vol_indices = shock_equity_vol_indices
self.tenors = tenors
super(EqVega, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=EquityVolSurface.SpotModel.StickyDelta,
use_contract_currency=use_contract_currency)
def json_params(self):
params = [
'shock_size',
'use_proportional_vol_bumps',
'use_parallel_shift',
'shock_equity_vol_indices',
'tenors'
]
params += super(EqVega, self).json_params()
return params
[docs]
class FxSpotShock(FxRiskMeasure):
[docs]
def __init__(self,
calculate_price_changes=None,
include_reporting_currency_shock=None,
risk_measure_id_stored=None,
shock_size=None,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta,
use_inverse_currency_shock=False):
"""
:param calculate_price_changes: The calculate price changes
:param include_reporting_currency_shock: The include reporting currency shock
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
:param use_inverse_currency_shock: The use inverse currency shock flag
"""
self.calculate_price_changes = calculate_price_changes
self.include_reporting_currency_shock = include_reporting_currency_shock
self.shock_size = shock_size
self.use_inverse_currency_shock = use_inverse_currency_shock
super(FxSpotShock, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = [
'calculate_price_changes',
'include_reporting_currency_shock',
'shock_size',
'use_inverse_currency_shock'
]
params += super(FxSpotShock, self).json_params()
return params
[docs]
class FxVolShock(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
deltas=None,
shocks=None,
use_proportional_shocks=None,
vol_shock_func=None,
tenors=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:type risk_measure_id_stored: str
:param deltas: The deltas
:type deltas: list of str
:param shocks: The shocks
:type shocks: list of float
:param use_proportional_shocks: The use_proportional_shocks flag
:type use_proportional_shocks: bool
:param vol_shock_func: The vol shock func
:type vol_shock_func: :class:`derivitec.risk_engine.types.VolShockFuncShocksByStrike`
:param tenors: The tenors
:type tenors: list of str
"""
self.risk_measure_id_stored = risk_measure_id_stored
self.deltas = deltas
self.shocks = shocks
self.use_proportional_shocks = use_proportional_shocks
self.vol_shock_func = vol_shock_func
self.tenors = tenors
super(FxVolShock, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta)
def json_params(self):
params = [
'deltas',
'shocks',
'use_proportional_shocks',
'vol_shock_func',
'tenors'
]
params += super(FxVolShock, self).json_params()
return params
[docs]
class CmdtyVega(CmdtyRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
"""
self.shock_size = shock_size
super(CmdtyVega, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
]
params += super(CmdtyVega, self).json_params()
return params
[docs]
class ScenarioImpact(RiskMeasure):
[docs]
def __init__(self,
calculate_price_changes=True,
risk_measure_id_stored=None,
scale_factor=1.0,
scenario=None):
"""
:param calculate_price_changes: If true the measure calculates PNLs, otherwise NPVs
:param risk_measure_id_stored: The risk measure id stored
:param scale_factor: This scale factor is applied to the end result
:param scenario: The scenario whose PNL or NPV impact we want to calculate
"""
self.calculate_price_changes = calculate_price_changes
self.scale_factor = scale_factor
self.scenario = scenario
super(ScenarioImpact, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'calculate_price_changes',
'scale_factor',
'scenario',
]
params += super(ScenarioImpact, self).json_params()
return params
[docs]
class CmdtySpotScenario(RiskScenario):
[docs]
def __init__(self,
shock_size=None,
scenario_name=None,
scenario_description=None):
"""
:param shock_size: The shock size
:param scenario_name: The scenario name
:param scenario_description: The scenario description
"""
self.shock_size = shock_size
super(CmdtySpotScenario, self).__init__(
scenario_name=scenario_name,
scenario_description=scenario_description)
def json_params(self):
params = [
'shock_size',
]
params += super(CmdtySpotScenario, self).json_params()
return params
[docs]
class CmdtyVolScenario(RiskScenario):
[docs]
def __init__(self,
shock_size=None,
use_proportional_vol_bump=None,
scenario_name=None,
scenario_description=None):
"""
:param shock_size: The shock size
:param use_proportional_vol_bump: The use proportional vol bump flag
:param scenario_name: The scenario name
:param scenario_description: The scenario description
"""
self.shock_size = shock_size
self.use_proportional_vol_bump = use_proportional_vol_bump
super(CmdtyVolScenario, self).__init__(
scenario_name=scenario_name,
scenario_description=scenario_description)
def json_params(self):
params = [
'shock_size',
'use_proportional_vol_bump'
]
params += super(CmdtyVolScenario, self).json_params()
return params
[docs]
class CreditScenario(RiskScenario):
[docs]
def __init__(self,
global_shock=None,
shocks_by_issuer_identifier=None,
use_proportional_shocks=None,
credit_spread_floor=None,
use_forward_rate_bumps=None,
tenors=None,
reference_issuer_identifier=None,
reference_currency=None,
reference_debt_type=None,
scenario_name=None,
scenario_description=None):
"""
:param global_shock: The global shock size
:param shocks_by_issuer_identifier: shocks by issuer identifier
:param use_proportional_shocks: The use proportional vol bump flag
:param credit_spread_floor: The credit spread floor
:param use_forward_rate_bumps: The use forward rate bumps
:param tenors: The tenors
:param reference_issuer_identifier: The reference issuer identifier
:param reference_currency: The reference currency
:param reference_debt_type: The reference debt type
:param scenario_name: The scenario name
:param scenario_description: The scenario description
"""
self.global_shock = global_shock
self.shocks_by_issuer_identifier = shocks_by_issuer_identifier
self.use_proportional_shocks = use_proportional_shocks
self.credit_spread_floor = credit_spread_floor
self.use_forward_rate_bumps = use_forward_rate_bumps
self.tenors = tenors
self.reference_issuer_identifier = reference_issuer_identifier
self.reference_currency = reference_currency
self.reference_debt_type = reference_debt_type
super(CreditScenario, self).__init__(
scenario_name=scenario_name,
scenario_description=scenario_description)
def json_params(self):
params = [
'global_shock',
'shocks_by_issuer_identifier',
'use_proportional_shocks',
'credit_spread_floor',
'use_forward_rate_bumps',
'tenors',
'reference_issuer_identifier',
'reference_currency',
'reference_debt_type',
]
params += super(CreditScenario, self).json_params()
return params
[docs]
class FxVolSurfaceScenario(RiskScenario):
[docs]
def __init__(self,
currency_to_bump=None,
vol_shock=None,
use_proportional_vol_bumps=None,
counter_currency=None,
vol_shock_funcs_by_fx_pair=None,
scenario_name=None,
scenario_description=None):
"""
:param currency_to_bump: The currency to bump
:param vol_shock: The vol shock
:param use_proportional_vol_bumps: The use proportional vol bump flag
:param counter_currency: The counter currency
:param vol_shock_funcs_by_fx_pair: The vol_shock_funcs_by_fx_pair
:param scenario_name: The scenario name
:param scenario_description: The scenario description
"""
self.currency_to_bump = currency_to_bump
self.vol_shock = vol_shock
self.use_proportional_vol_bumps = use_proportional_vol_bumps
self.counter_currency = counter_currency
self.vol_shock_funcs_by_fx_pair = vol_shock_funcs_by_fx_pair
super(FxVolSurfaceScenario, self).__init__(
scenario_name=scenario_name,
scenario_description=scenario_description)
def json_params(self):
params = [
'currency_to_bump',
'vol_shock',
'use_proportional_vol_bumps',
'counter_currency',
'vol_shock_funcs_by_fx_pair'
]
params += super(FxVolSurfaceScenario, self).json_params()
return params
[docs]
class JumpToDefaultScenario(RiskScenario):
[docs]
def __init__(self,
recovery_rates_by_issuer_identifier=None,
scenario_name=None,
scenario_description=None):
"""
:param recovery_rates_by_issuer_identifier: The recovery rates by issuer identifier
:param scenario_name: The scenario name
:param scenario_description: The scenario description
"""
self.recovery_rates_by_issuer_identifier = recovery_rates_by_issuer_identifier
super(JumpToDefaultScenario, self).__init__(
scenario_name=scenario_name,
scenario_description=scenario_description)
def json_params(self):
params = [
'recovery_rates_by_issuer_identifier',
]
params += super(JumpToDefaultScenario, self).json_params()
return params
[docs]
class CreditParallelShock(CreditRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_contract_currency: If true the measure uses the contract currency instead of the report currency for calculations
"""
self.shock_size = shock_size
self.use_contract_currency = use_contract_currency
super(CreditParallelShock, self).__init__(risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_contract_currency'
]
params += super(CreditParallelShock, self).json_params()
return params
[docs]
class CreditDelta(CreditRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
is_spread_based=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param is_spread_based: The is spread based flag
"""
self.shock_size = shock_size
self.is_spread_based = is_spread_based
super(CreditDelta, self).__init__(risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'is_spread_based'
]
params += super(CreditDelta, self).json_params()
return params
[docs]
class CreditGamma(CreditRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
is_spread_based=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param is_spread_based: The is spread based flag
"""
self.shock_size = shock_size
self.is_spread_based = is_spread_based
super(CreditGamma, self).__init__(risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'is_spread_based'
]
params += super(CreditGamma, self).json_params()
return params
[docs]
class VarStatistics(RiskEngineType):
[docs]
def __init__(self,
expected_shortfall=None,
expected_shortfall_pct=None,
id=None,
marginal_var=None,
marginal_var_pct=None,
mean=None,
message=None,
name=None,
std_dev=None,
std_dev_pct=None,
valid=None,
var=None,
var_pct=None,
worst_day=None,
worst_day_index=None,
worst_loss=None):
"""
:param expected_shortfall: The expected shortfall
:param expected_shortfall_pct: The expected shortfall pct
:param id: The id
:param marginal_var: The marginal var
:param marginal_var_pct: The marginal var pct
:param mean: The mean
:param message: The message
:param name: The name
:param std_dev: The std dev
:param std_dev_pct: The std dev pct
:param valid: The valid flag
:param var: The var
:param var_pct: The var pct
:param worst_day: The worst day
:param worst_day_index: The worst day index
:param worst_loss: The worst loss
"""
self.expected_shortfall = expected_shortfall
self.expected_shortfall_pct = expected_shortfall_pct
self.id = id
self.marginal_var = marginal_var
self.marginal_var_pct = marginal_var_pct
self.mean = mean
self.message = message
self.name = name
self.std_dev = std_dev
self.std_dev_pct = std_dev_pct
self.valid = valid
self.var = var
self.var_pct = var_pct
self.worst_day = worst_day
self.worst_day_index = worst_day_index
self.worst_loss = worst_loss
def json_params(self):
params = [
'expected_shortfall',
'expected_shortfall_pct',
'id',
'marginal_var',
'marginal_var_pct',
'mean',
'message',
'name',
'std_dev',
'std_dev_pct',
'valid',
'var',
'var_pct',
'worst_day',
'worst_day_index',
'worst_loss'
]
return params
[docs]
class PriceHistory(RiskEngineType):
[docs]
def __init__(self,
observation_dates=None,
final_prices=None,
final_prices_valid=None,
initial_price=None,
initial_price_valid=None,
message=None):
"""
:param observation_dates: The observation dates
:param final_prices: The final prices
:param final_prices_valid: The final prices valid
:param initial_price: The initial price
:param initial_price_valid: The initial price valid
:param message: The message
"""
self.observation_dates = observation_dates
self.final_prices = final_prices
self.final_prices_valid = final_prices_valid
self.initial_price = initial_price
self.initial_price_valid = initial_price_valid
self.message = message
def json_params(self):
params = [
'observation_dates',
'final_prices',
'final_prices_valid',
'initial_price',
'initial_price_valid',
'message',
]
return params
[docs]
class VarResult(RiskEngineType):
[docs]
def __init__(self,
price_histories_by_tradable_id=None,
scenario_price_histories_by_tradable_id=None):
"""
:param price_histories_by_tradable_id: The price histories by tradable id
:param scenario_price_histories_by_tradable_id: The scenario price histories by tradable id
"""
self.price_histories_by_tradable_id = price_histories_by_tradable_id
self.scenario_price_histories_by_tradable_id = scenario_price_histories_by_tradable_id
def json_params(self):
params = [
'price_histories_by_tradable_id',
'scenario_price_histories_by_tradable_id',
]
return params
[docs]
class EqVanna(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
vol_surface_spot_model=EquityVolSurface.SpotModel.StickyDelta,
sticky_smile_ratio=0.0,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
"""
self.sticky_smile_ratio = sticky_smile_ratio
super(EqVanna, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
use_contract_currency=use_contract_currency,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = ['sticky_smile_ratio']
params += super(EqVanna, self).json_params()
return params
[docs]
class EqVolGamma(EqRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None):
"""
:param risk_measure_id_stored: The risk measure id stored
"""
super(EqVolGamma, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = []
params += super(EqVolGamma, self).json_params()
return params
[docs]
class FxVanna(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
use_implied_vol_bumps=False,
use_inverse_currency_shock=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_implied_vol_bumps: The use implied vol bumps flag
:param use_inverse_currency_shock: The use inverse currency shock
"""
self.shock_size = shock_size
self.use_implied_vol_bumps = use_implied_vol_bumps
self.use_inverse_currency_shock = use_inverse_currency_shock
super(FxVanna, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_implied_vol_bumps',
'use_inverse_currency_shock'
]
params += super(FxVanna, self).json_params()
return params
[docs]
class FxVolGamma(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=None,
use_implied_vol_bumps=False,
use_inverse_currency_shock=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param use_implied_vol_bumps: The use implied vol bumps flag
:param use_inverse_currency_shock: The use inverse currency shock flag
"""
self.shock_size = shock_size
self.use_implied_vol_bumps = use_implied_vol_bumps
self.use_inverse_currency_shock = use_inverse_currency_shock
super(FxVolGamma, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'shock_size',
'use_implied_vol_bumps',
'use_inverse_currency_shock'
]
params += super(FxVolGamma, self).json_params()
return params
[docs]
class FxGammaByCross(FxRiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None,
shock_size=0.01,
vol_surface_spot_model=FxVolSurface.SpotModel.StickyDelta,
use_contract_currency=False):
"""
:param risk_measure_id_stored: The risk measure id stored
:param shock_size: The shock size
:param vol_surface_spot_model: The vol surface spot model
:param use_contract_currency: The use contract currency flag
"""
self.shock_size = shock_size
self.use_contract_currency = use_contract_currency
super(FxGammaByCross, self).__init__(
risk_measure_id_stored=risk_measure_id_stored,
vol_surface_spot_model=vol_surface_spot_model)
def json_params(self):
params = [
'shock_size',
'use_contract_currency'
]
params += super(FxGammaByCross, self).json_params()
return params
[docs]
class ReferenceVol(RiskMeasure):
[docs]
def __init__(self,
risk_measure_id_stored=None):
"""
:param risk_measure_id_stored: The risk measure id stored
"""
super(ReferenceVol, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = []
params += super(ReferenceVol, self).json_params()
return params
[docs]
class PricingDetail(RiskMeasure):
class Type(IntEnum):
UnderlyingPrice = 0
ReferenceVol = 1,
FuturesPrice = 2,
[docs]
def __init__(self,
risk_measure_id_stored=None,
pricing_detail_type=None):
"""
:param risk_measure_id_stored: The risk measure id stored
:param pricing_detail_type: The pricing detail type to return
"""
self.pricing_detail_type = pricing_detail_type
super(PricingDetail, self).__init__(
risk_measure_id_stored=risk_measure_id_stored)
def json_params(self):
params = [
'pricing_detail_type'
]
params += super(PricingDetail, self).json_params()
return params
[docs]
class VolShockFuncShocksByStrike(RiskEngineType):
[docs]
def __init__(self,
shock_by_strike_func=None,
maturity_beta_func=None,
strike_func_expects_delta=None,
use_proportional_shocks=None):
"""
:param shock_by_strike_func: The shock_by_strike_func
:param maturity_beta_func: The maturity_beta_func
:param strike_func_expects_delta: The strike_func_expects_delta
:param use_proportional_shocks: The use_proportional_shocks
"""
self.shock_by_strike_func = shock_by_strike_func
self.maturity_beta_func = maturity_beta_func
self.strike_func_expects_delta = strike_func_expects_delta
self.use_proportional_shocks = use_proportional_shocks
def json_params(self):
params = [
'shock_by_strike_func',
'maturity_beta_func',
'strike_func_expects_delta',
'use_proportional_shocks'
]
return params
[docs]
class VolShockFuncBetaAdj(RiskEngineType):
[docs]
def __init__(self, vol_beta_func=None, maturity_beta_func=None):
"""
:param vol_beta_func: The vol_beta_func
:param maturity_beta_func: The maturity_beta_func
"""
self.vol_beta_func = vol_beta_func
self.maturity_beta_func = maturity_beta_func
def json_params(self):
params = [
'vol_beta_func',
'maturity_beta_func',
]
return params
[docs]
class HistoricalScenario(RiskScenario):
[docs]
def __init__(self,
scenario_name=None,
scenario_description=None,
start_date=None,
end_date=None):
"""
:param scenario_name: The scenario name
:param scenario_description: The scenario description
:param start_date: The start date
:param end_date: The end date
"""
self.start_date = start_date
self.end_date = end_date
super(HistoricalScenario, self).__init__(
scenario_name=scenario_name,
scenario_description=scenario_description)
def json_params(self):
params = [
'start_date',
'end_date',
]
params += super(HistoricalScenario, self).json_params()
return params