Skip to content

Commit

Permalink
Chore: Make release 1.0.60
Browse files Browse the repository at this point in the history
  • Loading branch information
@martinroberson
martinroberson authored and Vanden Bon, David V [GBM Public] committed Feb 16, 2024
1 parent 5db1deb commit 42f4bd7
Show file tree
Hide file tree
Showing 12 changed files with 456 additions and 481 deletions.
309 changes: 126 additions & 183 deletions gs_quant/backtests/actions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,31 +15,38 @@
"""

from collections import namedtuple
from typing import TypeVar
import copy
from dataclasses import dataclass, field
from dataclasses_json import dataclass_json
from typing import TypeVar, Callable

from gs_quant.backtests.backtest_utils import *
from gs_quant.backtests.backtest_objects import ConstantTransactionModel, TransactionModel
from gs_quant.risk.transform import Transformer
from gs_quant.base import Priceable
from gs_quant.markets.securities import *
from gs_quant.common import RiskMeasure
from gs_quant.markets.portfolio import Portfolio
from gs_quant.markets.securities import *
from gs_quant.risk.transform import Transformer
from gs_quant.target.backtests import BacktestTradingQuantityType

action_count = 1


def default_transaction_cost(obj):
return field(default_factory=lambda: copy.copy(obj))


@dataclass_json
@dataclass
class Action(object):
def __init__(self, name: str = None):
self._needs_scaling = False
self._calc_type = CalcType.simple
self._risk = None
global action_count
if name is None:
self._name = 'Action{}'.format(action_count)
action_count += 1
else:
self._name = name
self._transaction_cost = ConstantTransactionModel(0)
_needs_scaling = False
_calc_type = CalcType.simple
_risk = None
_transaction_cost = ConstantTransactionModel(0)
name = None

def __post_init__(self):
self.set_name(self.name)

@property
def calc_type(self):
Expand All @@ -49,49 +56,52 @@ def calc_type(self):
def risk(self):
return self._risk

def set_name(self, name: str):
global action_count
if self.name is None:
self.name = 'Action{}'.format(action_count)
action_count += 1

@property
def transaction_cost(self):
return self._transaction_cost

@transaction_cost.setter
def transaction_cost(self, value):
self._transaction_cost = value


TAction = TypeVar('TAction', bound='Action')


@dataclass_json
@dataclass
class AddTradeAction(Action):
def __init__(self,
priceables: Union[Priceable, Iterable[Priceable]],
trade_duration: Union[str, dt.date, dt.timedelta] = None,
name: str = None,
transaction_cost: TransactionModel = ConstantTransactionModel(0)):
"""
create an action which adds a trade when triggered. The trades are resolved on the trigger date (state) and
last until the trade_duration if specified or for all future dates if not.
:param priceables: a priceable or a list of pricables.
:param trade_duration: an instrument attribute eg. 'expiration_date' or a date or a tenor or timedelta
if left as None the
trade will be added for all future dates
:param name: optional additional name to the priceable name
:param transaction_cost: optional a cash amount paid for each transaction, paid on both enter and exit
"""
super().__init__(name)

self._priceables = []
"""
create an action which adds a trade when triggered. The trades are resolved on the trigger date (state) and
last until the trade_duration if specified or for all future dates if not.
:param priceables: a priceable or a list of pricables.
:param trade_duration: an instrument attribute eg. 'expiration_date' or a date or a tenor or timedelta
if left as None the
trade will be added for all future dates
:param name: optional additional name to the priceable name
:param transaction_cost: optional a cash amount paid for each transaction, paid on both enter and exit
"""

priceables: Union[Priceable, Iterable[Priceable]] = None
trade_duration: Union[str, dt.date, dt.timedelta] = None
name: str = None
transaction_cost: TransactionModel = default_transaction_cost(ConstantTransactionModel())

def __post_init__(self):
self._dated_priceables = {}
self._trade_duration = trade_duration
self._transaction_cost = transaction_cost
for i, p in enumerate(make_list(priceables)):
named_priceables = []
for i, p in enumerate(make_list(self.priceables)):
if p.name is None:
self._priceables.append(p.clone(name=f'{self._name}_Priceable{i}'))
named_priceables.append(p.clone(name=f'{self.name}_Priceable{i}'))
else:
self._priceables.append(p.clone(name=f'{self._name}_{p.name}'))

@property
def priceables(self):
return self._priceables

@property
def trade_duration(self):
return self._trade_duration
named_priceables.append(p.clone(name=f'{self.name}_{p.name}'))
self.priceables = named_priceables

def set_dated_priceables(self, state, priceables):
self._dated_priceables[state] = make_list(priceables)
Expand All @@ -100,10 +110,6 @@ def set_dated_priceables(self, state, priceables):
def dated_priceables(self):
return self._dated_priceables

@property
def transaction_cost(self):
return self._transaction_cost


AddTradeActionInfo = namedtuple('AddTradeActionInfo', 'scaling')
EnterPositionQuantityScaledActionInfo = namedtuple('EnterPositionQuantityScaledActionInfo', 'not_applicable')
Expand All @@ -112,14 +118,10 @@ def transaction_cost(self):
RebalanceActionInfo = namedtuple('RebalanceActionInfo', 'not_applicable')


@dataclass_json
@dataclass
class EnterPositionQuantityScaledAction(Action):
def __init__(self,
priceables: Union[Priceable, Iterable[Priceable]],
trade_duration: Union[str, dt.date, dt.timedelta] = None,
name: str = None,
trade_quantity: float = 1,
trade_quantity_type: Union[BacktestTradingQuantityType, str] = BacktestTradingQuantityType.quantity):
"""
"""
create an action which enters trades when triggered. The trades are executed with specified quantity and
last until the trade_duration if specified, or for all future dates if not.
:param priceables: a priceable or a list of pricables.
Expand All @@ -129,160 +131,101 @@ def __init__(self,
:param trade_quantity: the amount, in units of trade_quantity_type to be traded
:param trade_quantity_type: the quantity type used to scale trade. eg. quantity for units, notional for
underlier notional
"""
super().__init__(name)
self._priceables = []
self._trade_duration = trade_duration
for i, p in enumerate(make_list(priceables)):
"""
priceables: Union[Priceable, Iterable[Priceable]] = None
trade_duration: Union[str, dt.date, dt.timedelta] = None
name: str = None
trade_quantity: float = 1
trade_quantity_type: Union[BacktestTradingQuantityType, str] = BacktestTradingQuantityType.quantity

def __post_init__(self):
named_priceables = []
for i, p in enumerate(make_list(self.priceables)):
if p.name is None:
self._priceables.append(p.clone(name=f'{self._name}_Priceable{i}'))
named_priceables.append(p.clone(name=f'{self.name}_Priceable{i}'))
else:
self._priceables.append(p.clone(name=f'{self._name}_{p.name}'))
self._trade_quantity = trade_quantity
self._trade_quantity_type = trade_quantity_type

@property
def priceables(self):
return self._priceables

@property
def trade_duration(self):
return self._trade_duration

@property
def trade_quantity(self):
return self._trade_quantity

@property
def trade_quantity_type(self):
return self._trade_quantity_type
named_priceables.append(p.clone(name=f'{self.name}_{p.name}'))
self.priceables = named_priceables


@dataclass_json
@dataclass
class ExitPositionAction(Action):
def __init__(self, name: str = None):
"""
Fully exit all held positions
:param name: optional name of the action
"""
super().__init__(name)
name: str = None


@dataclass_json
@dataclass
class ExitTradeAction(Action):
def __init__(self, priceable_names: Union[str, Iterable[str]] = None, name: str = None):
"""
Fully exit all held positions
:param priceable_names: optional string or list of strings of priceable names
:param name: optional name of the action
"""
super().__init__(name)
self._priceables_names = make_list(priceable_names)
priceable_names: Union[str, Iterable[str]] = None
name: str = None

@property
def priceable_names(self):
return self._priceables_names
def __post_init__(self):
self.priceables_names = make_list(self.priceable_names)


@dataclass_json
@dataclass
class ExitAllPositionsAction(ExitTradeAction):
def __init__(self, name: str = None):
"""
Fully exit all held positions
:param name: optional name of the action
"""
super().__init__([], name)
"""
Fully exit all held positions
"""

def __post_init__(self):
self._calc_type = CalcType.path_dependent


@dataclass_json
@dataclass
class HedgeAction(Action):
def __init__(self, risk, priceables: Priceable = None, trade_duration: str = None, name: str = None,
csa_term: str = None, scaling_parameter: str = 'notional_amount',
transaction_cost: TransactionModel = ConstantTransactionModel(0),
risk_transformation: Transformer = None):
super().__init__(name)
risk: RiskMeasure = None
priceables: Optional[Priceable] = None
trade_duration: Union[str, dt.date, dt.timedelta] = None
name: str = None
csa_term: str = None
scaling_parameter: str = 'notional_amount'
transaction_cost: TransactionModel = default_transaction_cost(ConstantTransactionModel())
risk_transformation: Transformer = None

def __post_init__(self):
self._calc_type = CalcType.semi_path_dependent
self._priceable = priceables
self._risk = risk
self._trade_duration = trade_duration
self._csa_term = csa_term
self._scaling_parameter = scaling_parameter
self._transaction_cost = transaction_cost
self._risk_transformation = risk_transformation
if isinstance(priceables, Portfolio):
trades = []
for i, priceable in enumerate(priceables):
if isinstance(self.priceables, Portfolio):
named_priceables = []
for i, priceable in enumerate(self.priceables):
if priceable.name is None:
trades.append(priceable.clone(name=f'{self._name}_Priceable{i}'))
named_priceables.append(priceable.clone(name=f'{self.name}_Priceable{i}'))
else:
trades.append(priceable.clone(name=f'{self._name}_{priceable.name}'))
self._priceable = Portfolio(trades)
named_priceables.append(priceable.clone(name=f'{self.name}_{priceable.name}'))
named_priceable = Portfolio(named_priceables)
elif isinstance(self.priceables, Priceable):
if self.priceables.name is None:
named_priceable = self.priceables.clone(name=f'{self.name}_Priceable0')
else:
named_priceable = self.priceables.clone(name=f'{self.name}_{self.priceables.name}')
else:
if priceables is not None:
if priceables.name is None:
self._priceable = priceables.clone(name=f'{self._name}_Priceable0')
else:
self._priceable = priceables.clone(name=f'{self._name}_{priceables.name}')

@property
def trade_duration(self):
return self._trade_duration
raise RuntimeError('hedge action only accepts one trade or one portfolio')
self.priceables = named_priceable

@property
def priceable(self):
return self._priceable

@property
def risk(self):
return self._risk

@property
def csa_term(self):
return self._csa_term

@property
def scaling_parameter(self):
return self._scaling_parameter

@property
def transaction_cost(self):
return self._transaction_cost

@property
def risk_transformation(self):
return self._risk_transformation
return self.priceables


@dataclass_json
@dataclass
class RebalanceAction(Action):
def __init__(self, priceable: Priceable, size_parameter, method,
transaction_cost: TransactionModel = ConstantTransactionModel(0)):
super().__init__()
priceable: Priceable = None
size_parameter: Union[str, float] = None
method: Callable = None
transaction_cost: TransactionModel = default_transaction_cost(ConstantTransactionModel())
name: str = None

def __post_init__(self):
self._calc_type = CalcType.path_dependent
self._size_parameter = size_parameter
self._method = method
self._transaction_cost = transaction_cost
if priceable.unresolved is None:
if self.priceable.unresolved is None:
raise ValueError("Please specify a resolved priceable to rebalance.")
if priceable is not None:
if priceable.name is None:
self._priceable = priceable.clone(name=f'{self._name}_Priceable0')
if self.priceable is not None:
if self.priceable.name is None:
self.priceable = self.priceable.clone(name=f'{self.name}_Priceable0')
else:
self._priceable = priceable.clone(name=f'{self._name}_{priceable.name}')

@property
def priceable(self):
return self._priceable

@property
def size_parameter(self):
return self._size_parameter

@property
def method(self):
return self._method

@property
def args(self):
return self._args

@property
def transaction_cost(self):
return self._transaction_cost
self.priceable = self.priceable.clone(name=f'{self.name}_{self.priceable.name}')

0 comments on commit 42f4bd7

Please sign in to comment.