diff --git a/docs/source/autogreek.rst b/docs/source/autogreek.rst index 730f25bc..dd617c07 100644 --- a/docs/source/autogreek.rst +++ b/docs/source/autogreek.rst @@ -8,5 +8,17 @@ pfhedge.autogreek .. currentmodule:: pfhedge.autogreek +Delta +----- + .. autofunction:: delta + +Gamma +----- + .. autofunction:: gamma + +Vega +---- + +.. autofunction:: vega diff --git a/docs/source/nn.functional.rst b/docs/source/nn.functional.rst index 46c82eb3..dfa76326 100644 --- a/docs/source/nn.functional.rst +++ b/docs/source/nn.functional.rst @@ -30,3 +30,7 @@ Other Functions .. autofunction:: realized_variance .. autofunction:: realized_volatility .. autofunction:: terminal_value +.. autofunction:: ncdf +.. autofunction:: npdf +.. autofunction:: d1 +.. autofunction:: d2 diff --git a/pfhedge/_utils/parse.py b/pfhedge/_utils/parse.py index ad64690a..7c5cd7a4 100644 --- a/pfhedge/_utils/parse.py +++ b/pfhedge/_utils/parse.py @@ -1,9 +1,15 @@ +from numbers import Real from typing import Optional +from typing import Union import torch from torch import Tensor +def _as_optional_tensor(input: Optional[Union[Tensor, Real]]) -> Optional[Tensor]: + return torch.as_tensor(input) if input is not None else input + + def parse_spot( *, spot: Optional[Tensor] = None, @@ -12,6 +18,11 @@ def parse_spot( log_moneyness: Optional[Tensor] = None, **kwargs ) -> Tensor: + spot = _as_optional_tensor(spot) + strike = _as_optional_tensor(strike) + moneyness = _as_optional_tensor(moneyness) + log_moneyness = _as_optional_tensor(log_moneyness) + if spot is not None: return spot elif moneyness is not None and strike is not None: @@ -19,4 +30,15 @@ def parse_spot( elif log_moneyness is not None and strike is not None: return log_moneyness.exp() * strike else: - raise ValueError("Insufficient parameters to parse `spot`") + raise ValueError("Insufficient parameters to parse spot") + + +def parse_volatility( + *, volatility: Optional[Tensor] = None, variance: Optional[Tensor] = None, **kwargs +) -> Tensor: + if volatility is not None: + return volatility + elif variance is not None: + return variance.clamp(min=0.0).sqrt() + else: + raise ValueError("Insufficient parameters to parse volatility") diff --git a/pfhedge/_utils/typing.py b/pfhedge/_utils/typing.py new file mode 100644 index 00000000..c5d1956f --- /dev/null +++ b/pfhedge/_utils/typing.py @@ -0,0 +1,5 @@ +from typing import Union + +from torch import Tensor + +TensorOrScalar = Union[Tensor, float, int] diff --git a/pfhedge/autogreek.py b/pfhedge/autogreek.py index 97fb1512..190ecfe4 100644 --- a/pfhedge/autogreek.py +++ b/pfhedge/autogreek.py @@ -5,6 +5,7 @@ from torch import Tensor from ._utils.parse import parse_spot +from ._utils.parse import parse_volatility def delta( @@ -25,9 +26,9 @@ def delta( Args: pricer (callable): Pricing formula of a derivative. - create_graph (bool, default=False): If ``True``, graph of the derivative - will be constructed, allowing to compute higher order derivative - products. + create_graph (bool, default=False): If ``True``, + graph of the derivative will be constructed, + allowing to compute higher order derivative products. **params: Parameters passed to ``pricer``. Returns: @@ -96,7 +97,6 @@ def delta( if parameter not in signature(pricer).parameters.keys(): del params[parameter] - assert spot.requires_grad price = pricer(**params) return torch.autograd.grad( price, @@ -124,8 +124,9 @@ def gamma( Args: pricer (callable): Pricing formula of a derivative. - create_graph (bool, default=False): If ``True``, graph of the derivative will be - constructed, allowing to compute higher order derivative products. + create_graph (bool, default=False): If ``True``, + graph of the derivative will be constructed, + allowing to compute higher order derivative products. **params: Parameters passed to ``pricer``. Returns: @@ -170,3 +171,63 @@ def gamma( grad_outputs=torch.ones_like(tensor_delta), create_graph=create_graph, )[0] + + +def vega( + pricer: Callable[..., Tensor], *, create_graph: bool = False, **params +) -> Tensor: + """Computes and returns vega of a derivative using automatic differentiation. + + Vega is a differentiation of a derivative price with respect to + a variance of underlying instrument. + + Note: + The keyword argument ``**params`` should contain at least one of the + following parameters: + + - ``volatility`` + - ``variance`` + + Args: + pricer (callable): Pricing formula of a derivative. + create_graph (bool, default=False): If ``True``, + graph of the derivative will be constructed, + allowing to compute higher order derivative products. + **params: Parameters passed to ``pricer``. + + Returns: + torch.Tensor + + Examples: + + Vega of a European option can be evaluated as follows. + + >>> import pfhedge.autogreek as autogreek + >>> from pfhedge.nn import BSEuropeanOption + >>> + >>> pricer = BSEuropeanOption().price + >>> autogreek.vega( + ... pricer, + ... log_moneyness=torch.zeros(3), + ... time_to_maturity=torch.ones(3), + ... volatility=torch.tensor([0.18, 0.20, 0.22]), + ... ) + tensor([0.3973, 0.3970, 0.3965]) + """ + volatility = parse_volatility(**params).requires_grad_() + params["volatility"] = volatility + params["variance"] = volatility.pow(2) + + # Delete parameters that are not in the signature of pricer to avoid + # TypeError: got an unexpected keyword argument '' + for parameter in list(params.keys()): + if parameter not in signature(pricer).parameters.keys(): + del params[parameter] + + price = pricer(**params) + return torch.autograd.grad( + price, + inputs=volatility, + grad_outputs=torch.ones_like(price), + create_graph=create_graph, + )[0] diff --git a/pfhedge/features/__init__.py b/pfhedge/features/__init__.py index 8a48b469..6098a341 100644 --- a/pfhedge/features/__init__.py +++ b/pfhedge/features/__init__.py @@ -11,5 +11,6 @@ from .features import Moneyness from .features import PrevHedge from .features import TimeToMaturity +from .features import Variance from .features import Volatility from .features import Zeros diff --git a/pfhedge/features/_getter.py b/pfhedge/features/_getter.py index 13acb540..89279abf 100644 --- a/pfhedge/features/_getter.py +++ b/pfhedge/features/_getter.py @@ -11,6 +11,7 @@ from .features import Moneyness from .features import PrevHedge from .features import TimeToMaturity +from .features import Variance from .features import Volatility from .features import Zeros @@ -23,6 +24,7 @@ MaxMoneyness, Moneyness, PrevHedge, + Variance, Volatility, Zeros, ] diff --git a/pfhedge/features/features.py b/pfhedge/features/features.py index be74c2a3..f84070ab 100644 --- a/pfhedge/features/features.py +++ b/pfhedge/features/features.py @@ -71,6 +71,17 @@ def __getitem__(self, time_step: Optional[int]) -> Tensor: return self.derivative.ul().volatility[:, index].unsqueeze(-1) +class Variance(StateIndependentFeature): + """Variance of the underlier of the derivative.""" + + def __str__(self) -> str: + return "variance" + + def __getitem__(self, time_step: Optional[int]) -> Tensor: + index = [time_step] if isinstance(time_step, int) else ... + return self.derivative.ul().variance[:, index].unsqueeze(-1) + + class PrevHedge(Feature): """Previous holding of underlier.""" diff --git a/pfhedge/instruments/derivative/base.py b/pfhedge/instruments/derivative/base.py index d893adca..901b6443 100644 --- a/pfhedge/instruments/derivative/base.py +++ b/pfhedge/instruments/derivative/base.py @@ -13,12 +13,12 @@ from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.doc import _set_docstring from pfhedge._utils.str import _addindent +from pfhedge._utils.typing import TensorOrScalar from ..base import Instrument from ..primary.base import Primary T = TypeVar("T", bound="Derivative") -TensorOrFloat = Union[Tensor, float] class Derivative(Instrument): @@ -60,7 +60,7 @@ def device(self) -> Optional[torch.device]: return self.underlier.device def simulate( - self, n_paths: int = 1, init_state: Optional[Tuple[TensorOrFloat, ...]] = None + self, n_paths: int = 1, init_state: Optional[Tuple[TensorOrScalar, ...]] = None ) -> None: """Simulate time series associated with the underlier. diff --git a/pfhedge/instruments/primary/base.py b/pfhedge/instruments/primary/base.py index 89a7284e..c94b31da 100644 --- a/pfhedge/instruments/primary/base.py +++ b/pfhedge/instruments/primary/base.py @@ -14,11 +14,11 @@ from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.doc import _set_docstring +from pfhedge._utils.typing import TensorOrScalar from ..base import Instrument T = TypeVar("T", bound="Primary") -TensorOrFloat = Union[float, Tensor] class Primary(Instrument): @@ -52,7 +52,7 @@ def __init__(self) -> None: self.register_buffer("spot", None) @property - def default_init_state(self) -> Tuple[TensorOrFloat, ...]: + def default_init_state(self) -> Tuple[TensorOrScalar, ...]: """Returns the default initial state of simulation.""" # TODO(simaki): Remove @no_type_check once BrownianStock and HestonStock get @@ -63,7 +63,7 @@ def simulate( self, n_paths: int, time_horizon: float, - init_state: Optional[Tuple[TensorOrFloat, ...]] = None, + init_state: Optional[Tuple[TensorOrScalar, ...]] = None, ) -> None: """Simulate time series associated with the instrument and add them as buffers. diff --git a/pfhedge/instruments/primary/brownian.py b/pfhedge/instruments/primary/brownian.py index a696f6a3..1823cdc1 100644 --- a/pfhedge/instruments/primary/brownian.py +++ b/pfhedge/instruments/primary/brownian.py @@ -10,12 +10,11 @@ from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.doc import _set_docstring from pfhedge._utils.str import _format_float +from pfhedge._utils.typing import TensorOrScalar from pfhedge.stochastic import generate_geometric_brownian from .base import Primary -TensorOrFloat = Union[Tensor, float] - class BrownianStock(Primary): """A stock of which spot prices follow the geometric Brownian motion. @@ -104,7 +103,7 @@ def simulate( self, n_paths: int = 1, time_horizon: float = 20 / 250, - init_state: Optional[Tuple[TensorOrFloat]] = None, + init_state: Optional[Tuple[TensorOrScalar]] = None, ) -> None: """Simulate the spot price and add it as a buffer named ``spot``. diff --git a/pfhedge/instruments/primary/cir.py b/pfhedge/instruments/primary/cir.py index 689141fb..78b62f47 100644 --- a/pfhedge/instruments/primary/cir.py +++ b/pfhedge/instruments/primary/cir.py @@ -9,12 +9,11 @@ from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.doc import _set_docstring from pfhedge._utils.str import _format_float +from pfhedge._utils.typing import TensorOrScalar from pfhedge.stochastic import generate_cir from .base import Primary -TensorOrFloat = Union[Tensor, float] - class CIRRate(Primary): """A rate which follow the CIR process. @@ -83,7 +82,7 @@ def simulate( self, n_paths: int = 1, time_horizon: float = 20 / 250, - init_state: Optional[Tuple[TensorOrFloat, ...]] = None, + init_state: Optional[Tuple[TensorOrScalar, ...]] = None, ) -> None: """Simulate the spot rate and add it as a buffer named ``spot``. diff --git a/pfhedge/instruments/primary/heston.py b/pfhedge/instruments/primary/heston.py index d2e2491e..80ca5274 100644 --- a/pfhedge/instruments/primary/heston.py +++ b/pfhedge/instruments/primary/heston.py @@ -9,12 +9,11 @@ from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.doc import _set_docstring from pfhedge._utils.str import _format_float +from pfhedge._utils.typing import TensorOrScalar from pfhedge.stochastic import generate_heston from .base import Primary -TensorOrFloat = Union[Tensor, float] - class HestonStock(Primary): """A stock of which spot price and variance follow Heston process. @@ -104,7 +103,7 @@ def simulate( self, n_paths: int = 1, time_horizon: float = 20 / 250, - init_state: Optional[Tuple[TensorOrFloat, ...]] = None, + init_state: Optional[Tuple[TensorOrScalar, ...]] = None, ) -> None: """Simulate the spot price and add it as a buffer named ``spot``. diff --git a/pfhedge/nn/functional.py b/pfhedge/nn/functional.py index ed0ba453..e3894aef 100644 --- a/pfhedge/nn/functional.py +++ b/pfhedge/nn/functional.py @@ -5,6 +5,10 @@ import torch import torch.nn.functional as fn from torch import Tensor +from torch.distributions.normal import Normal +from torch.distributions.utils import broadcast_all + +from pfhedge._utils.typing import TensorOrScalar def european_payoff(input: Tensor, call: bool = True, strike: float = 1.0) -> Tensor: @@ -146,7 +150,7 @@ def isoelastic_utility(input: Tensor, a: float) -> Tensor: if a == 1.0: return input.log() else: - return input ** (1.0 - a) + return input.pow(1.0 - a) def topp(input: Tensor, p: float, dim: Optional[int] = None, largest: bool = True): @@ -185,7 +189,7 @@ def topp(input: Tensor, p: float, dim: Optional[int] = None, largest: bool = Tru if dim is None: return input.topk(ceil(p * input.numel()), largest=largest) else: - return input.topk(ceil(p * input.size()[dim]), dim=dim, largest=largest) + return input.topk(ceil(p * input.size(dim)), dim=dim, largest=largest) def expected_shortfall(input: Tensor, p: float, dim: Optional[int] = None) -> Tensor: @@ -265,7 +269,7 @@ def clamp( return output -def realized_variance(input: Tensor, dt: Union[Tensor, float]) -> Tensor: +def realized_variance(input: Tensor, dt: TensorOrScalar) -> Tensor: """Returns the realized variance of the price. Realized variance :math:`\\sigma^2` of the stock price :math:`S` is defined as: @@ -389,3 +393,71 @@ def terminal_value( value -= cost * unit[..., 0].abs() * spot[..., 0] return value + + +def ncdf(input: Tensor) -> Tensor: + """Returns a new tensor with the normal cumulative distribution function. + + Args: + input (torch.Tensor): The input tensor. + + Returns: + torch.Tensor + + Examples: + >>> from pfhedge.nn.functional import ncdf + >>> + >>> input = torch.tensor([-1.0, 0.0, 10.0]) + >>> ncdf(input) + tensor([0.1587, 0.5000, 1.0000]) + """ + return Normal(0.0, 1.0).cdf(input) + + +def npdf(input: Tensor) -> Tensor: + """Returns a new tensor with the normal distribution function. + + Args: + input (torch.Tensor): The input tensor. + + Returns: + torch.Tensor + + Examples: + >>> from pfhedge.nn.functional import npdf + >>> + >>> input = torch.tensor([-1.0, 0.0, 10.0]) + >>> npdf(input) + tensor([2.4197e-01, 3.9894e-01, 7.6946e-23]) + """ + return Normal(0.0, 1.0).log_prob(input).exp() + + +def d1(log_moneyness: Tensor, time_to_maturity: Tensor, volatility: Tensor) -> Tensor: + """Returns :math:`d_1` in the Black-Scholes formula. + + Args: + log_moneyness (torch.Tensor or float): Log moneyness of the underlying asset. + time_to_maturity (torch.Tensor or float): Time to maturity of the derivative. + volatility (torch.Tensor or float): Volatility of the underlying asset. + + Returns: + torch.Tensor + """ + s, t, v = broadcast_all(log_moneyness, time_to_maturity, volatility) + return (s + (v.pow(2) / 2) * t).div(v * t.sqrt()) + + +def d2(log_moneyness: Tensor, time_to_maturity: Tensor, volatility: Tensor) -> Tensor: + """Returns :math:`d_2` in the Black-Scholes formula. + + Args: + log_moneyness (torch.Tensor or float): Log moneyness of the underlying asset. + time_to_maturity (torch.Tensor or float): Time to maturity of the derivative. + volatility (torch.Tensor or float): Volatility of the underlying asset. + + Returns: + torch.Tensor + """ + s, t, v = broadcast_all(log_moneyness, time_to_maturity, volatility) + return (s - (v.pow(2) / 2) * t).div(v * t.sqrt()) diff --git a/pfhedge/nn/modules/bs/_base.py b/pfhedge/nn/modules/bs/_base.py index c2eb4e70..f821c07e 100644 --- a/pfhedge/nn/modules/bs/_base.py +++ b/pfhedge/nn/modules/bs/_base.py @@ -5,9 +5,10 @@ import torch from torch import Tensor -from torch.distributions.normal import Normal from torch.nn import Module +import pfhedge.autogreek as autogreek + class BSModuleMixin(Module): """A mixin class for Black-Scholes formula modules. @@ -34,58 +35,45 @@ def forward(self, input: Tensor) -> Tensor: """ return self.delta(*(input[..., [i]] for i in range(input.size(-1)))) - @abc.abstractmethod @no_type_check - def delta(self, *args, **kwargs) -> Tensor: - """Returns delta of the derivative. + def price(self, *args, **kwargs) -> Tensor: + """Returns price of the derivative. Returns: torch.Tensor """ - def inputs(self) -> List[str]: - """Returns the names of input features. + @no_type_check + def delta(self, **kwargs) -> Tensor: + """Returns delta of the derivative. Returns: - list + torch.Tensor """ - return list(signature(self.delta).parameters.keys()) + return autogreek.delta(self.price, **kwargs) - @property - def N(self) -> Normal: - """Returns normal distribution with zero mean and unit standard deviation.""" - return Normal(torch.tensor(0.0), torch.tensor(1.0)) + @no_type_check + def gamma(self, **kwargs) -> Tensor: + """Returns delta of the derivative. - @staticmethod - def d1( - log_moneyness: Tensor, time_to_maturity: Tensor, volatility: Tensor - ) -> Tensor: - """Returns :math:`d_1` in the Black-Scholes formula. + Returns: + torch.Tensor + """ + return autogreek.gamma(self.price, **kwargs) - Args: - log_moneyness (torch.Tensor): Log moneyness of the underlying asset. - time_to_maturity (torch.Tensor): Time to expiry of the option. - volatility (torch.Tensor): Volatility of the underlying asset. + @no_type_check + def vega(self, **kwargs) -> Tensor: + """Returns delta of the derivative. Returns: torch.Tensor """ - s, t, v = map(torch.as_tensor, (log_moneyness, time_to_maturity, volatility)) - return (s + (v ** 2 / 2) * t) / (v * t.sqrt()) - - @staticmethod - def d2( - log_moneyness: Tensor, time_to_maturity: Tensor, volatility: Tensor - ) -> Tensor: - """Returns :math:`d_2` in the Black-Scholes formula. + return autogreek.vega(self.price, **kwargs) - Args: - log_moneyness (torch.Tensor): Log moneyness of the underlying asset. - time_to_maturity (torch.Tensor): Time to expiry of the option. - volatility (torch.Tensor): Volatility of the underlying asset. + def inputs(self) -> List[str]: + """Returns the names of input features. Returns: - torch.Tensor + list """ - s, t, v = map(torch.as_tensor, (log_moneyness, time_to_maturity, volatility)) - return (s - (v ** 2 / 2) * t) / (v * t.sqrt()) + return list(signature(self.delta).parameters.keys()) diff --git a/pfhedge/nn/modules/bs/american_binary.py b/pfhedge/nn/modules/bs/american_binary.py index d2d2c204..7de408f8 100644 --- a/pfhedge/nn/modules/bs/american_binary.py +++ b/pfhedge/nn/modules/bs/american_binary.py @@ -2,11 +2,16 @@ import torch from torch import Tensor +from torch.distributions.utils import broadcast_all import pfhedge.autogreek as autogreek from pfhedge._utils.bisect import find_implied_volatility from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.str import _format_float +from pfhedge.nn.functional import d1 +from pfhedge.nn.functional import d2 +from pfhedge.nn.functional import ncdf +from pfhedge.nn.functional import npdf from ._base import BSModuleMixin @@ -114,14 +119,12 @@ def price( Returns: torch.Tensor """ - s, m, t, v = map( - torch.as_tensor, - (log_moneyness, max_log_moneyness, time_to_maturity, volatility), + s, m, t, v = broadcast_all( + log_moneyness, max_log_moneyness, time_to_maturity, volatility ) - n1 = self.N.cdf(self.d1(s, t, v) / sqrt(2.0)) - n2 = self.N.cdf(self.d2(s, t, v) / sqrt(2.0)) - + n1 = ncdf(d1(s, t, v) / sqrt(2)) + n2 = ncdf(d2(s, t, v) / sqrt(2)) p = (1 / 2) * (s.exp() * (1 + n1) + n2) return p.where(m < 0, torch.ones_like(p)) @@ -152,17 +155,13 @@ def delta( Returns: torch.Tensor """ - s, m, t, v = map( - torch.as_tensor, - (log_moneyness, max_log_moneyness, time_to_maturity, volatility), + s, m, t, v = broadcast_all( + log_moneyness, max_log_moneyness, time_to_maturity, volatility ) - d1 = self.d1(s, t, v) - d2 = self.d2(s, t, v) - c1 = self.N.cdf(d1 / sqrt(2.0)) - p1 = self.N.log_prob(d1 / sqrt(2.0)).exp() - p2 = self.N.log_prob(d2 / sqrt(2.0)).exp() - + c1 = ncdf(d1(s, t, v) / sqrt(2)) + p1 = npdf(d1(s, t, v) / sqrt(2)) + p2 = npdf(d2(s, t, v) / sqrt(2)) d = (1 + c1 + (p1 + p2)) / (2 * self.strike) return d.where(m < 0, torch.zeros_like(d)) @@ -194,8 +193,7 @@ def gamma( Returns: torch.Tensor """ - return autogreek.gamma( - self.price, + return super().gamma( strike=self.strike, log_moneyness=log_moneyness, max_log_moneyness=max_log_moneyness, diff --git a/pfhedge/nn/modules/bs/european.py b/pfhedge/nn/modules/bs/european.py index 9805d83e..f2533227 100644 --- a/pfhedge/nn/modules/bs/european.py +++ b/pfhedge/nn/modules/bs/european.py @@ -1,9 +1,14 @@ import torch from torch import Tensor +from torch.distributions.utils import broadcast_all from pfhedge._utils.bisect import find_implied_volatility from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.str import _format_float +from pfhedge.nn.functional import d1 +from pfhedge.nn.functional import d2 +from pfhedge.nn.functional import ncdf +from pfhedge.nn.functional import npdf from ._base import BSModuleMixin @@ -102,9 +107,9 @@ def delta( Returns: torch.Tensor """ - s, t, v = map(torch.as_tensor, (log_moneyness, time_to_maturity, volatility)) + s, t, v = broadcast_all(log_moneyness, time_to_maturity, volatility) - delta = self.N.cdf(self.d1(s, t, v)) + delta = ncdf(d1(s, t, v)) delta = delta - 1 if not self.call else delta # put-call parity return delta @@ -134,9 +139,9 @@ def gamma( f"{self.__class__.__name__} for a put option is not yet supported." ) - s, t, v = map(torch.as_tensor, (log_moneyness, time_to_maturity, volatility)) + s, t, v = broadcast_all(log_moneyness, time_to_maturity, volatility) price = self.strike * s.exp() - gamma = self.N.log_prob(self.d1(s, t, v)).exp() / (price * v * t.sqrt()) + gamma = npdf(d1(s, t, v)) / (price * v * t.sqrt()) return gamma @@ -160,10 +165,10 @@ def price( Returns: torch.Tensor """ - s, t, v = map(torch.as_tensor, (log_moneyness, time_to_maturity, volatility)) + s, t, v = broadcast_all(log_moneyness, time_to_maturity, volatility) - n1 = self.N.cdf(self.d1(s, t, v)) - n2 = self.N.cdf(self.d2(s, t, v)) + n1 = ncdf(d1(s, t, v)) + n2 = ncdf(d2(s, t, v)) price = self.strike * (s.exp() * n1 - n2) diff --git a/pfhedge/nn/modules/bs/european_binary.py b/pfhedge/nn/modules/bs/european_binary.py index 9aec0fa0..995af498 100644 --- a/pfhedge/nn/modules/bs/european_binary.py +++ b/pfhedge/nn/modules/bs/european_binary.py @@ -2,12 +2,16 @@ import torch from torch import Tensor +from torch.distributions.utils import broadcast_all -import pfhedge.autogreek as autogreek from pfhedge._utils.bisect import find_implied_volatility from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.doc import _set_docstring from pfhedge._utils.str import _format_float +from pfhedge.nn.functional import d1 +from pfhedge.nn.functional import d2 +from pfhedge.nn.functional import ncdf +from pfhedge.nn.functional import npdf from ._base import BSModuleMixin @@ -110,9 +114,9 @@ def price( Returns: torch.Tensor """ - s, t, v = map(torch.as_tensor, (log_moneyness, time_to_maturity, volatility)) + s, t, v = broadcast_all(log_moneyness, time_to_maturity, volatility) - price = self.N.cdf(self.d2(s, t, v)) + price = ncdf(d2(s, t, v)) price = 1.0 - price if not self.call else price # put-call parity return price @@ -137,11 +141,9 @@ def delta( Returns: torch.Tensor """ - s, t, v = map(torch.as_tensor, (log_moneyness, time_to_maturity, volatility)) + s, t, v = broadcast_all(log_moneyness, time_to_maturity, volatility) - delta = self.N.log_prob(self.d2(s, t, v)).exp() / ( - self.strike * s.exp() * v * t.sqrt() - ) + delta = npdf(d2(s, t, v)) / (self.strike * s.exp() * v * t.sqrt()) return delta def gamma( @@ -163,8 +165,7 @@ def gamma( Returns: torch.Tensor """ - return autogreek.gamma( - self.price, + return super().gamma( strike=self.strike, log_moneyness=log_moneyness, time_to_maturity=time_to_maturity, diff --git a/pfhedge/nn/modules/bs/lookback.py b/pfhedge/nn/modules/bs/lookback.py index 4150f638..75db5650 100644 --- a/pfhedge/nn/modules/bs/lookback.py +++ b/pfhedge/nn/modules/bs/lookback.py @@ -1,14 +1,17 @@ from typing import List -from typing import Optional import torch from torch import Tensor +from torch.distributions.utils import broadcast_all -import pfhedge.autogreek as autogreek from pfhedge._utils.bisect import find_implied_volatility from pfhedge._utils.doc import _set_attr_and_docstring from pfhedge._utils.doc import _set_docstring from pfhedge._utils.str import _format_float +from pfhedge.nn.functional import d1 +from pfhedge.nn.functional import d2 +from pfhedge.nn.functional import ncdf +from pfhedge.nn.functional import npdf from ._base import BSModuleMixin @@ -125,24 +128,24 @@ def price( (log_moneyness, max_log_moneyness, time_to_maturity, volatility), ) - d1 = self.d1(s, t, v) - d2 = self.d2(s, t, v) - e1 = (s - m + (v ** 2 / 2) * t) / (v * t.sqrt()) # d' in paper - e2 = (s - m - (v ** 2 / 2) * t) / (v * t.sqrt()) + d1_ = d1(s, t, v) + d2_ = d2(s, t, v) + e1 = (s - m + (v.pow(2) / 2) * t) / (v * t.sqrt()) # d' in paper + e2 = (s - m - (v.pow(2) / 2) * t) / (v * t.sqrt()) # when max moneyness < strike price_0 = self.strike * ( - s.exp() * self.N.cdf(d1) - - self.N.cdf(d2) - + s.exp() * v * t.sqrt() * (d1 * self.N.cdf(d1) + self.N.log_prob(d1).exp()) + s.exp() * ncdf(d1_) + - ncdf(d2_) + + s.exp() * v * t.sqrt() * (d1_ * ncdf(d1_) + npdf(d1_)) ) # when max moneyness >= strike price_1 = self.strike * ( - s.exp() * self.N.cdf(e1) - - m.exp() * self.N.cdf(e2) + s.exp() * ncdf(e1) + - m.exp() * ncdf(e2) + m.exp() - 1 - + s.exp() * v * t.sqrt() * (e1 * self.N.cdf(e1) + self.N.log_prob(e1).exp()) + + s.exp() * v * t.sqrt() * (e1 * ncdf(e1) + npdf(e1)) ) return torch.where(m < 0, price_0, price_1) @@ -155,7 +158,6 @@ def delta( time_to_maturity: Tensor, volatility: Tensor, create_graph: bool = False, - strike: Optional[Tensor] = None, ) -> Tensor: """Returns delta of the derivative. @@ -178,8 +180,7 @@ def delta( Returns: torch.Tensor """ - return autogreek.delta( - self.price, + return super().delta( log_moneyness=log_moneyness, max_log_moneyness=max_log_moneyness, time_to_maturity=time_to_maturity, @@ -215,8 +216,7 @@ def gamma( Returns: torch.Tensor """ - return autogreek.gamma( - self.price, + return super().gamma( strike=self.strike, log_moneyness=log_moneyness, max_log_moneyness=max_log_moneyness, diff --git a/pfhedge/nn/modules/hedger.py b/pfhedge/nn/modules/hedger.py index 49ff5a9d..cea4e2d8 100644 --- a/pfhedge/nn/modules/hedger.py +++ b/pfhedge/nn/modules/hedger.py @@ -1,3 +1,4 @@ +from typing import Callable from typing import List from typing import Optional from typing import Tuple @@ -17,6 +18,7 @@ from pfhedge._utils.lazy import has_lazy from pfhedge._utils.operations import ensemble_mean from pfhedge._utils.str import _format_float +from pfhedge._utils.typing import TensorOrScalar from pfhedge.features import FeatureList from pfhedge.features._base import Feature from pfhedge.instruments.base import Instrument @@ -27,8 +29,6 @@ from .loss import EntropicRiskMeasure from .loss import HedgeLoss -TensorOrFloat = Union[Tensor, float] - class Hedger(Module): """Module to hedge and price derivatives. @@ -320,7 +320,7 @@ def compute_pnl( derivative: Derivative, hedge: Optional[List[Instrument]] = None, n_paths: int = 1000, - init_state: Optional[Tuple[TensorOrFloat, ...]] = None, + init_state: Optional[Tuple[TensorOrScalar, ...]] = None, ) -> Tensor: """Returns the terminal portfolio value after hedging a given derivative. @@ -376,7 +376,7 @@ def compute_loss( hedge: Optional[List[Instrument]] = None, n_paths: int = 1000, n_times: int = 1, - init_state: Optional[Tuple[TensorOrFloat, ...]] = None, + init_state: Optional[Tuple[TensorOrScalar, ...]] = None, enable_grad: bool = True, ) -> Tensor: """Returns the value of the criterion for the terminal portfolio value @@ -430,16 +430,18 @@ def compute_loss( return mean_loss def _configure_optimizer( - self, derivative: Derivative, optimizer: Union[Optimizer, type] + self, + derivative: Derivative, + optimizer: Union[Optimizer, Callable[..., Optimizer]], ) -> Optimizer: - if isinstance(optimizer, type): + if not isinstance(optimizer, Optimizer): if has_lazy(self): # Run a placeholder forward to initialize lazy parameters _ = self.compute_pnl(derivative, n_paths=1) # If we use `if issubclass(optimizer, Optimizer)` here, mypy thinks that # optimizer is Optimizer rather than its subclass (e.g. Adam) # and complains that the required parameter default is missing. - if Optimizer in optimizer.__mro__: + if Optimizer in getattr(optimizer, "__mro__", []): optimizer = cast(Optimizer, optimizer(self.model.parameters())) else: raise TypeError("optimizer is not an Optimizer type") @@ -452,8 +454,8 @@ def fit( n_epochs: int = 100, n_paths: int = 1000, n_times: int = 1, - optimizer=Adam, - init_state: Optional[Tuple[TensorOrFloat, ...]] = None, + optimizer: Union[Optimizer, Callable[..., Optimizer]] = Adam, + init_state: Optional[Tuple[TensorOrScalar, ...]] = None, verbose: bool = True, validation: bool = True, ) -> Optional[List[float]]: @@ -566,7 +568,7 @@ def price( hedge: Optional[List[Instrument]] = None, n_paths: int = 1000, n_times: int = 1, - init_state: Optional[Tuple[TensorOrFloat, ...]] = None, + init_state: Optional[Tuple[TensorOrScalar, ...]] = None, enable_grad: bool = False, ) -> Tensor: """Evaluate the premium of the given derivative. diff --git a/pfhedge/nn/modules/ww.py b/pfhedge/nn/modules/ww.py index c3576420..37e95ac1 100644 --- a/pfhedge/nn/modules/ww.py +++ b/pfhedge/nn/modules/ww.py @@ -149,6 +149,6 @@ def width(self, input: Tensor) -> Tensor: spot = self.derivative.strike * input[..., [0]].exp() gamma = self.bs.gamma(*(input[..., [i]] for i in range(input.size(-1)))) - width = (cost * (3 / 2) * (gamma ** 2) * spot / self.a) ** (1 / 3) + width = (cost * (3 / 2) * gamma.pow(2) * spot / self.a).pow(1 / 3) return width diff --git a/pfhedge/stochastic/brownian.py b/pfhedge/stochastic/brownian.py index 0de2c88b..69d45f73 100644 --- a/pfhedge/stochastic/brownian.py +++ b/pfhedge/stochastic/brownian.py @@ -6,13 +6,13 @@ import torch from torch import Tensor -TensorOrFloat = Union[Tensor, float] +from pfhedge._utils.typing import TensorOrScalar def generate_brownian( n_paths: int, n_steps: int, - init_state: Union[Tuple[TensorOrFloat, ...], TensorOrFloat] = (0.0,), + init_state: Union[Tuple[TensorOrScalar, ...], TensorOrScalar] = (0.0,), sigma: float = 0.2, dt: float = 1 / 250, dtype: Optional[torch.dtype] = None, @@ -80,7 +80,7 @@ def generate_brownian( def generate_geometric_brownian( n_paths: int, n_steps: int, - init_state: Union[Tuple[TensorOrFloat, ...], TensorOrFloat] = (1.0,), + init_state: Union[Tuple[TensorOrScalar, ...], TensorOrScalar] = (1.0,), sigma: float = 0.2, dt: float = 1 / 250, dtype: Optional[torch.dtype] = None, diff --git a/pfhedge/stochastic/cir.py b/pfhedge/stochastic/cir.py index aca57a28..7027ec19 100644 --- a/pfhedge/stochastic/cir.py +++ b/pfhedge/stochastic/cir.py @@ -6,17 +6,17 @@ import torch from torch import Tensor -TensorOrFloat = Union[Tensor, float] +from pfhedge._utils.typing import TensorOrScalar def generate_cir( n_paths: int, n_steps: int, - init_state: Tuple[TensorOrFloat, ...] = (0.04,), - kappa: TensorOrFloat = 1.0, - theta: TensorOrFloat = 0.04, - sigma: TensorOrFloat = 0.2, - dt: TensorOrFloat = 1 / 250, + init_state: Tuple[TensorOrScalar, ...] = (0.04,), + kappa: TensorOrScalar = 1.0, + theta: TensorOrScalar = 0.04, + sigma: TensorOrScalar = 0.2, + dt: TensorOrScalar = 1 / 250, dtype: Optional[torch.dtype] = None, device: Optional[torch.device] = None, ) -> Tensor: @@ -103,14 +103,14 @@ def generate_cir( exp = (-kappa * dt).exp() m = theta + (v - theta) * exp s2 = v * (sigma ** 2) * exp * (1 - exp) / kappa + theta * (sigma ** 2) * ( - (1 - exp) ** 2 + (1 - exp).pow(2) ) / (2 * kappa) - psi = s2 / (m ** 2 + EPSILON) + psi = s2 / (m.pow(2) + EPSILON) # Compute V(t + dt) where psi <= PSI_CRIT: Eq(23, 27, 28) b = ((2 / psi) - 1 + (2 / psi).sqrt() * (2 / psi - 1).sqrt()).sqrt() - a = m / (1 + b ** 2) - next_0 = a * (b + randn[:, i_step]) ** 2 + a = m / (1 + b.pow(2)) + next_0 = a * (b + randn[:, i_step]).pow(2) # Compute V(t + dt) where psi > PSI_CRIT: Eq(25) u = rand[:, i_step] diff --git a/pfhedge/stochastic/heston.py b/pfhedge/stochastic/heston.py index cbb59f50..8e7ac14c 100644 --- a/pfhedge/stochastic/heston.py +++ b/pfhedge/stochastic/heston.py @@ -8,11 +8,10 @@ from torch import Tensor from pfhedge._utils.str import _addindent +from pfhedge._utils.typing import TensorOrScalar from .cir import generate_cir -TensorOrFloat = Union[Tensor, float] - class HestonTuple(namedtuple("HestonTuple", ["spot", "variance"])): @@ -33,7 +32,7 @@ def volatility(self) -> Tensor: def generate_heston( n_paths: int, n_steps: int, - init_state: Tuple[TensorOrFloat, ...] = (1.0, 0.04), + init_state: Tuple[TensorOrScalar, ...] = (1.0, 0.04), kappa: float = 1.0, theta: float = 0.04, sigma: float = 0.2, diff --git a/pyproject.toml b/pyproject.toml index 04c84e70..614afbad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "pfhedge" -version = "0.12.2" +version = "0.12.3" description = "Deep Hedging in PyTorch" authors = ["Shota Imaki "] license = "MIT" diff --git a/tests/features/test_features.py b/tests/features/test_features.py index 1f69f4d7..7cab4c1d 100644 --- a/tests/features/test_features.py +++ b/tests/features/test_features.py @@ -14,6 +14,7 @@ from pfhedge.features import Moneyness from pfhedge.features import PrevHedge from pfhedge.features import TimeToMaturity +from pfhedge.features import Variance from pfhedge.features import Volatility from pfhedge.features import Zeros from pfhedge.instruments import BrownianStock @@ -251,6 +252,71 @@ def test_is_state_dependent(self): assert not f.is_state_dependent() +class TestVariance(_TestFeature): + @pytest.mark.parametrize("sigma", [0.2, 0.1]) + def test_constant_volatility(self, sigma): + derivative = EuropeanOption(BrownianStock(sigma=sigma)) + derivative.underlier.register_buffer("spot", torch.empty(2, 3)) + + f = Variance().of(derivative) + + result = f[0] + expect = torch.full((2, 1), sigma ** 2) + expect = expect.unsqueeze(-1) + assert_close(result, expect, check_stride=False) + + result = f[1] + expect = torch.full((2, 1), sigma ** 2) + expect = expect.unsqueeze(-1) + assert_close(result, expect, check_stride=False) + + result = f[2] + expect = torch.full((2, 1), sigma ** 2) + expect = expect.unsqueeze(-1) + assert_close(result, expect, check_stride=False) + + result = f[None] + expect = torch.full((2, 3), sigma ** 2) + expect = expect.unsqueeze(-1) + assert_close(result, expect, check_stride=False) + + def test_stochastic_volatility(self): + derivative = EuropeanOption(HestonStock(dt=0.1), maturity=0.2) + derivative.simulate(n_paths=2) + variance = derivative.ul().variance + + f = Variance().of(derivative) + + result = f[0] + expect = variance[:, [0]] + expect = expect.unsqueeze(-1) + assert_close(result, expect, check_stride=False) + + result = f[1] + expect = variance[:, [1]] + expect = expect.unsqueeze(-1) + assert_close(result, expect, check_stride=False) + + result = f[None] + expect = variance + expect = expect.unsqueeze(-1) + assert_close(result, expect, check_stride=False) + + def test_str(self): + assert str(Variance()) == "variance" + + @pytest.mark.parametrize("dtype", [torch.float32, torch.float64]) + def test_dtype(self, dtype): + derivative = EuropeanOption(BrownianStock()) + self.assert_same_dtype(Variance(), derivative, dtype) + + def test_is_state_dependent(self): + derivative = EuropeanOption(BrownianStock()) + hedger = Hedger(Naked(), inputs=["empty"]) + f = Variance().of(derivative, hedger) + assert not f.is_state_dependent() + + class TestPrevHedge(_TestFeature): @pytest.mark.parametrize("volatility", [0.2, 0.1]) def test(self, volatility): diff --git a/tests/test_autogreek.py b/tests/test_autogreek.py new file mode 100644 index 00000000..8e5bfb71 --- /dev/null +++ b/tests/test_autogreek.py @@ -0,0 +1,23 @@ +import torch +from torch.testing import assert_close + +import pfhedge.autogreek as autogreek + + +def test_vega(): + def pricer(volatility, coef): + return coef * volatility.pow(2) + + torch.manual_seed(42) + volatility = torch.randn(10).exp() # make it positive + coef = torch.randn(10) + result = autogreek.vega(pricer, volatility=volatility, coef=coef) + expect = 2 * coef * volatility + assert_close(result, expect) + + torch.manual_seed(42) + variance = torch.randn(10).exp() # make it positive + coef = torch.randn(10) + result = autogreek.vega(pricer, variance=variance, coef=coef) + expect = 2 * coef * variance.sqrt() + assert_close(result, expect)