Initial Commit

6 files changed, 611 insertions, 0 deletions

diff --git a/utils/__init__.py b/utils/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/utils/__init__.py
diff --git a/utils/engine.py b/utils/engine.py
new file mode 100644
index 0000000..e8d2b1f
--- /dev/null
+++ b/utils/engine.py
@@ -0,0 +1,106 @@
+# author : S. Mandalia
+#          shivesh.mandalia@outlook.com
+#
+# date   : March 19, 2020
+
+"""
+Pricing engine for exotic options.
+"""
+
+import math
+from statistics import mean, stdev
+from dataclasses import dataclass
+from typing import List
+
+from utils.path import PathGenerator
+from utils.payoff import BasePayoff
+
+
+__all__ = ['MCResult', 'PricingEngine']
+
+
+@dataclass
+class MCResult:
+    """
+    Price of option along with its MC error.
+    """
+    price: float
+    stderr: float
+
+
+@dataclass
+class PricingEngine:
+    """
+    Class for generating underlying prices using MC techniques.
+
+    Attributes
+    ----------
+    payoff : Payoff object for calculating the options payoff.
+    path : PathGenerator object for generating the evolution of the underlying.
+
+    Methods
+    ----------
+    price()
+
+    Examples
+    ----------
+    >>> from utils.engine import PricingEngine
+    >>> from utils.path import PathGenerator
+    >>> from utils.payoff import AsianArithmeticPayOff
+    >>> path = PathGenerator(S=100., r=0.1, div=0.01, vol=0.3)
+    >>> payoff = AsianArithmeticPayOff(option_right='Call', K=110)
+    >>> engine = PricingEngine(payoff=payoff, path=path)
+    >>> print(engine.price(T=range(4)))
+    2.1462567745518335
+
+    """
+    payoff: BasePayoff
+    path: PathGenerator
+
+    def price(self, T: List[float], ntrials: int = 1E4,
+              antithetic: bool = True) -> MCResult:
+        """
+        Price the option using MC techniques.
+
+        Parameters
+        ----------
+        T : Set of times {t1, t2, ..., tn} in years.
+        ntrials : Number of trials to simulate.
+        antithetic : Use antithetic variates technique.
+
+        Returns
+        ----------
+        MCResult : Price of the option.
+
+        """
+        if ntrials < len(T):
+            raise AssertionError('Number of trials cannot be less than the '
+                                 'number of setting dates!')
+
+        # Generation start
+        ntrials = int(ntrials // len(T))
+        payoffs = [0] * ntrials
+        for idx in range(ntrials):
+            # Generate a random path
+            if not antithetic:
+                spot_prices = self.path.generate(T)
+            else:
+                prices_tuple = self.path.generate_antithetic(T)
+                spot_prices, a_spot_prices = prices_tuple
+
+            # Calculate the payoff
+            payoff = self.payoff.calculate(spot_prices)
+            if antithetic:
+                a_po = self.payoff.calculate(a_spot_prices)
+                payoff = (payoff + a_po) / 2
+            payoffs[idx] = payoff
+
+        # Discount to current time
+        df = math.exp(-self.path.net_r * (T[-1] - T[0]))
+        dis_payoffs = [x * df for x in payoffs]
+
+        # Payoff expectation and standard error
+        exp_payoff = mean(dis_payoffs)
+        stderr = stdev(dis_payoffs, exp_payoff) / math.sqrt(ntrials)
+
+        return MCResult(exp_payoff, stderr)
diff --git a/utils/enums.py b/utils/enums.py
new file mode 100644
index 0000000..6a607ea
--- /dev/null
+++ b/utils/enums.py
@@ -0,0 +1,40 @@
+# author : S. Mandalia
+#          shivesh.mandalia@outlook.com
+#
+# date   : March 19, 2020
+
+"""
+Enumeration utility classes.
+"""
+
+from enum import Enum, auto
+
+__all__ = ['OptionRight', 'BarrierUpDown', 'BarrierInOut']
+
+
+class PPEnum(Enum):
+    """Enum with prettier printing."""
+
+    def __repr__(self) -> str:
+        return super().__repr__().split('.')[1].split(':')[0]
+
+    def __str__(self) -> str:
+        return super().__str__().split('.')[1]
+
+
+class OptionRight(PPEnum):
+    """Right of an option."""
+    Call = auto()
+    Put = auto()
+
+
+class BarrierUpDown(PPEnum):
+    """Up or down type barrier option."""
+    Up = auto()
+    Down = auto()
+
+
+class BarrierInOut(PPEnum):
+    """In or out type barrier option."""
+    In = auto()
+    Out = auto()
diff --git a/utils/misc.py b/utils/misc.py
new file mode 100644
index 0000000..708d759
--- /dev/null
+++ b/utils/misc.py
@@ -0,0 +1,65 @@
+# author : S. Mandalia
+#          shivesh.mandalia@outlook.com
+#
+# date   : March 19, 2020
+
+"""
+Miscellaneous utility methods.
+"""
+
+
+__all__ = ['is_num', 'is_pos']
+
+
+def is_num(val: (int, float)) -> bool:
+    """
+    Check if the input value is a non-infinite number.
+
+    Parameters
+    ----------
+    val : Value to check.
+
+    Returns
+    ----------
+    is_num : Whether it is a non-infinite number.
+
+    Examples
+    ----------
+    >>> from utils.misc import is_num
+    >>> print(is_num(10))
+    True
+    >>> print(is_num(None))
+    False
+
+    """
+    if not isinstance(val, (int, float)):
+        return False
+    return True
+
+
+def is_pos(val: (int, float)) -> bool:
+    """
+    Check if the input value is a non-infinite positive number.
+
+    Parameters
+    ----------
+    val : Value to check.
+
+    Returns
+    ----------
+    is_pos : Whether it is a non-infinite positive number.
+
+    Examples
+    ----------
+    >>> from utils.misc import is_pos
+    >>> print(is_pos(10))
+    True
+    >>> print(is_pos(-10))
+    False
+
+    """
+    if not is_num(val):
+        return False
+    if not val >= 0:
+        return False
+    return True
diff --git a/utils/path.py b/utils/path.py
new file mode 100644
index 0000000..6855b7f
--- /dev/null
+++ b/utils/path.py
@@ -0,0 +1,135 @@
+# author : S. Mandalia
+#          shivesh.mandalia@outlook.com
+#
+# date   : March 19, 2020
+
+"""
+Path generator for underlying.
+"""
+
+import math
+import random
+from copy import deepcopy
+from dataclasses import dataclass
+from typing import List, Tuple
+
+
+__all__ = ['PathGenerator']
+
+
+@dataclass
+class PathGenerator:
+    """
+    Class for generating underlying prices using MC techniques.
+
+    Attributes
+    ----------
+    S : Spot price.
+    r : Risk-free interest rate.
+    div : Dividend yield.
+    vol : Volatility.
+    net_r : Net risk free rate.
+
+    Methods
+    ----------
+    generate(T)
+        Generate a random path {S_t1, S_t2, ..., S_tn}.
+    generate_antithetic(T)
+        Generate a random plus antithetic path
+        [{S_t1, S_t2, ..., S_tn}, {S'_t1, S'_t2, ..., S'_tn}].
+
+    Examples
+    ----------
+    >>> from utils.path import PathGenerator
+    >>> path = PathGenerator(S=100., r=0.1, div=0.01, vol=0.3)
+    >>> print(path.generate(T=range(4)))
+    [100.0, 91.11981160354563, 94.87596210593794, 117.44132223235353]
+    >>> print(path.generate(T=range(4)))
+    [100.0, 68.73668230722738, 71.43490333826567, 70.70833180133955]
+
+    """
+    S: float
+    r: float
+    div: float
+    vol: float
+
+    @property
+    def net_r(self) -> float:
+        """Net risk free rate."""
+        return self.r - self.div
+
+    def generate(self, T: List[float]) -> List[float]:
+        """
+        Generate a random path {S_t1, S_t2, ..., S_tn}.
+
+        Parameters
+        ----------
+        T : Set of times {t1, t2, ..., tn} in years.
+
+        Returns
+        ----------
+        spot_prices : Set of prices for the underlying {S_t1, S_t2, ..., S_tn}.
+
+        """
+        # Calculate dt time differences
+        dts = [T[idx + 1] - T[idx] for idx in range(len(T) - 1)]
+
+        spot_prices = [0] * len(T)
+        spot_prices[0] = self.S
+        for idx, dt in enumerate(dts):
+            # Calculate the drift e^{(r - (1/2) σ²) Δt}
+            drift = math.exp((self.net_r - (1/2) * self.vol**2) * dt)
+
+            # Calculate the volatility term e^{σ √{Δt} N(0, 1)}
+            rdm_gauss = random.gauss(0, 1)
+            vol_term = math.exp(self.vol * math.sqrt(dt) * rdm_gauss)
+
+            # Calculate next spot price
+            S_t = spot_prices[idx] * drift * vol_term
+            spot_prices[idx + 1] = S_t
+        return spot_prices
+
+    def generate_antithetic(self, T: List[float]) -> Tuple[List[float],
+                                                           List[float]]:
+        """
+        Generate a random plus antithetic path
+        [{S_t1, S_t2, ..., S_tn}, {S'_t1, S'_t2, ..., S'_tn}].
+
+        Parameters
+        ----------
+        T : Set of times {t1, t2, ..., tn} in years.
+
+        Returns
+        ----------
+        prices_tuple : Set of prices for the underlying
+                       [{S_t1, S_t2, ..., S_tn}, {S'_t1, S'_t2, ..., S'_tn}].
+
+        """
+        # Calculate dt time differences
+        dts = [T[idx + 1] - T[idx] for idx in range(len(T) - 1)]
+
+        # Create data structures
+        spot_prices = [0] * len(T)
+        spot_prices[0] = self.S
+
+        a_spot_prices = deepcopy(spot_prices)
+
+        for idx, dt in enumerate(dts):
+            # Calculate the drift e^{(r - (1/2) σ²) Δt}
+            drift = math.exp((self.net_r - (1/2) * self.vol**2) * dt)
+
+            # Calculate the volatility term e^{σ √{Δt} N(0, 1)}
+            rdm_gauss = random.gauss(0, 1)
+            a_gauss = -rdm_gauss
+            vol_term = math.exp(self.vol * math.sqrt(dt) * rdm_gauss)
+            a_vol_term = math.exp(self.vol * math.sqrt(dt) * a_gauss)
+
+            # Calculate next spot price
+            S_t = spot_prices[idx] * drift * vol_term
+            a_S_t = a_spot_prices[idx] * drift * a_vol_term
+
+            # Add to data structure
+            spot_prices[idx + 1] = S_t
+            a_spot_prices[idx + 1] = a_S_t
+
+        return (spot_prices, a_spot_prices)
diff --git a/utils/payoff.py b/utils/payoff.py
new file mode 100644
index 0000000..fffbdba
--- /dev/null
+++ b/utils/payoff.py
@@ -0,0 +1,265 @@
+# author : S. Mandalia
+#          shivesh.mandalia@outlook.com
+#
+# date   : March 19, 2020
+
+"""
+Payoff of an option.
+"""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import List
+
+from utils.enums import OptionRight, BarrierUpDown, BarrierInOut
+
+
+__all__ = ['BasePayoff', 'VanillaPayOff', 'AsianArithmeticPayOff',
+           'DiscreteBarrierPayOff']
+
+
+@dataclass
+class BasePayoff(ABC):
+    """Base class for calculating the payoff."""
+    K: float
+    option_right: (str, OptionRight)
+
+    @property
+    def option_right(self) -> OptionRight:
+        """Right of the option."""
+        return self._option_right
+
+    @option_right.setter
+    def option_right(self, val: (str, OptionRight)) -> None:
+        """Set the option_right of the option."""
+        if isinstance(val, str):
+            if not hasattr(OptionRight, val):
+                or_names = [x.name for x in OptionRight]
+                raise ValueError(f'Invalid str {val}, expected {or_names}')
+            self._option_right = OptionRight[val]
+        elif isinstance(val, OptionRight):
+            self._option_right = val
+        else:
+            raise TypeError(
+                f'Expected str or OptionRight, instead got type {type(val)}!'
+            )
+
+    def _calculate_call(self, S: float) -> float:
+        """Call option."""
+        return max(S - self.K, 0.)
+
+    def _calculate_put(self, S: float) -> float:
+        """Put option."""
+        return max(self.K - S, 0.)
+
+    @abstractmethod
+    def calculate(self, S: float) -> float:
+        """Calulate the payoff for a given spot."""
+
+
+class VanillaPayOff(BasePayoff):
+    """
+    Class for calculating the payoff of a vanilla option.
+
+    Attributes
+    ----------
+    option_right : Right of the option.
+    K : Strike price.
+
+    Methods
+    ----------
+    calculate(S)
+        Calulate the payoff given a spot price.
+
+    Examples
+    ----------
+    >>> from utils.payoff import VanillaPayOff
+    >>> payoff = VanillaPayOff(option_right='Call', K=150.)
+    >>> print(payoff.calculate(160.))
+    10.0
+
+    """
+
+    def calculate(self, S: (float, List[float])) -> float:
+        """
+        Calulate the payoff given a spot price.
+
+        Parameters
+        ----------
+        S : Spot price or list of spot prices.
+
+        Returns
+        ----------
+        payoff : Payoff.
+
+        Notes
+        ----------
+        If a list is given as input, the final entry will be taken to evaluate.
+
+        """
+        if not isinstance(S, float):
+            S = S[-1]
+        if self.option_right == OptionRight.Call:
+            payoff = self._calculate_call(S)
+        else:
+            payoff = self._calculate_put(S)
+        return payoff
+
+
+class AsianArithmeticPayOff(BasePayoff):
+    """
+    Class for calculating the payoff of an arithmetic Asian option.
+
+    Attributes
+    ----------
+    option_right : Right of the option.
+    K : Strike price.
+
+    Methods
+    ----------
+    calculate(S)
+        Calulate the payoff given a set of prices for the underlying.
+
+    Examples
+    ----------
+    >>> from utils.payoff import AsianArithmeticPayOff
+    >>> payoff = AsianArithmeticPayOff(option_right='Call', K=150)
+    >>> print(payoff.calculate([140, 150, 160, 170, 180]))
+    10.0
+
+    """
+
+    def calculate(self, S: List[float]) -> float:
+        """
+        Calulate the payoff given a set of prices for the underlying.
+
+        Parameters
+        ----------
+        S : Set of prices for the underlying {S_t1, S_t2, ..., S_tn}.
+
+        Returns
+        ----------
+        payoff : Payoff.
+
+        """
+        avg_sum = sum(S) / len(S)
+        if self.option_right == OptionRight.Call:
+            payoff = self._calculate_call(avg_sum)
+        else:
+            payoff = self._calculate_put(avg_sum)
+        return payoff
+
+
+@dataclass(init=False)
+class DiscreteBarrierPayOff(BasePayoff):
+    """
+    Class for calculating the payoff of a discrete barrier European style
+    option.
+
+    Attributes
+    ----------
+    option_right : Right of the option.
+    K : Strike price.
+    B : Barrier price.
+    barrier_updown : Up or down type barrier option.
+    barrier_inout : In or out type barrier option.
+
+    Methods
+    ----------
+    calculate(S)
+        Calulate the payoff given a set of prices for the underlying.
+
+    Examples
+    ----------
+    >>> from utils.payoff import DiscreteBarrierPayOff
+    >>> payoff = DiscreteBarrierPayOff(option_right='Call', K=100, B=90,
+                                       barrier_updown='Down', barrier_inout='Out')
+    >>> print(payoff.calculate([100., 110., 120.]))
+    20.0
+    >>> print(payoff.calculate([100., 110., 120., 80., 110.]))
+    0.0
+
+    """
+    B: float
+    barrier_updown: (str, BarrierUpDown)
+    barrier_inout: (str, BarrierInOut)
+
+    def __init__(self, option_right: (str, OptionRight), K: float, B: float,
+                 barrier_updown: (str, BarrierUpDown),
+                 barrier_inout: (str, BarrierInOut)):
+        super().__init__(K, option_right)
+        self.B = B
+        self.barrier_updown = barrier_updown
+        self.barrier_inout = barrier_inout
+
+    @property
+    def barrier_updown(self) -> BarrierUpDown:
+        """Up or down type barrier option."""
+        return self._barrier_updown
+
+    @barrier_updown.setter
+    def barrier_updown(self, val: (str, BarrierUpDown)) -> None:
+        """Set either up or down type barrier option."""
+        if isinstance(val, str):
+            if not hasattr(BarrierUpDown, val):
+                or_names = [x.name for x in BarrierUpDown]
+                raise ValueError(f'Invalid str {val}, expected {or_names}')
+            self._barrier_updown = BarrierUpDown[val]
+        elif isinstance(val, BarrierUpDown):
+            self._barrier_updown = val
+        else:
+            raise TypeError(
+                f'Expected str or BarrierUpDown, instead got type {type(val)}!'
+            )
+
+    @property
+    def barrier_inout(self) -> BarrierInOut:
+        """Up or down type barrier option."""
+        return self._barrier_inout
+
+    @barrier_inout.setter
+    def barrier_inout(self, val: (str, BarrierInOut)) -> None:
+        """Set either up or down type barrier option."""
+        if isinstance(val, str):
+            if not hasattr(BarrierInOut, val):
+                or_names = [x.name for x in BarrierInOut]
+                raise ValueError(f'Invalid str {val}, expected {or_names}')
+            self._barrier_inout = BarrierInOut[val]
+        elif isinstance(val, BarrierInOut):
+            self._barrier_inout = val
+        else:
+            raise TypeError(
+                f'Expected str or BarrierInOut, instead got type {type(val)}!'
+            )
+
+    def calculate(self, S: List[float]) -> float:
+        """
+        Calulate the payoff given a set of prices for the underlying.
+
+        Parameters
+        ----------
+        S : Set of prices for the underlying {S_t1, S_t2, ..., S_tn}.
+
+        Returns
+        ----------
+        payoff : Payoff.
+
+        """
+        # Calculate the heavyside
+        if self.barrier_updown == BarrierUpDown.Up:
+            H = [1 if self.B - x > 0 else 0 for x in S]
+        else:
+            H = [1 if x - self.B > 0 else 0 for x in S]
+
+        # Calculate whether it has been activated
+        if self.barrier_inout == BarrierInOut.In:
+            activation = 1 if min(H) == 0 else 0
+        else:
+            activation = min(H)
+
+        # Calculate payoff using final price
+        if self.option_right == OptionRight.Call:
+            payoff = activation * self._calculate_call(S[-1])
+        else:
+            payoff = activation * self._calculate_put(S[-1])
+        return payoff