feat(instruments): add the knock-in and double-barrier call family

src/deephedging/__init__.py

@@ -54,13 +54,17 @@
     AsianCall,
     AsianPut,
     BasketCall,
+    DoubleKnockOutCall,
+    DownAndInCall,
+    DownAndOutCall,
     EuropeanCall,
     EuropeanPut,
     GeometricBasketCall,
     LookbackCall,
     LookbackPut,
     Payoff,
     SingleAssetPayoff,
+    UpAndInCall,
     UpAndOutCall,
 )
 from deephedging.market import (
@@ -109,6 +113,9 @@
     "DeepBSDESolver",
     "DefaultFeatures",
     "DiscountGenerator",
+    "DoubleKnockOutCall",
+    "DownAndInCall",
+    "DownAndOutCall",
     "Entropic",
     "FeatureMap",
     "EuropeanCall",
@@ -148,6 +155,7 @@
     "SpectralRisk",
     "TiltedGBMSimulator",
     "TrainConfig",
+    "UpAndInCall",
     "UpAndOutCall",
     "VarianceFeatures",
     "ZeroGenerator",

src/deephedging/instruments/__init__.py

@@ -1,7 +1,13 @@
 """Derivative payoffs."""
 
 from deephedging.instruments.asian import AsianCall, AsianPut
-from deephedging.instruments.barrier import UpAndOutCall
+from deephedging.instruments.barrier import (
+    DoubleKnockOutCall,
+    DownAndInCall,
+    DownAndOutCall,
+    UpAndInCall,
+    UpAndOutCall,
+)
 from deephedging.instruments.base import Payoff
 from deephedging.instruments.basket import BasketCall, GeometricBasketCall
 from deephedging.instruments.lookback import LookbackCall, LookbackPut
@@ -12,12 +18,16 @@
     "AsianCall",
     "AsianPut",
     "BasketCall",
+    "DoubleKnockOutCall",
+    "DownAndInCall",
+    "DownAndOutCall",
     "EuropeanCall",
     "EuropeanPut",
     "GeometricBasketCall",
     "LookbackCall",
     "LookbackPut",
     "Payoff",
     "SingleAssetPayoff",
+    "UpAndInCall",
     "UpAndOutCall",
 ]

src/deephedging/instruments/barrier.py

@@ -43,3 +43,156 @@ def __call__(self, paths: torch.Tensor) -> torch.Tensor:
         alive = paths.max(dim=0).values < self.barrier
         vanilla = torch.clamp(paths[-1] - self.strike, min=0.0)
         return vanilla * alive
+
+
+@dataclass(frozen=True)
+class UpAndInCall:
+    """Up-and-in call, a vanilla call that activates only past a barrier.
+
+    Pays ``max(S_T - K, 0)`` only if the path reaches the barrier and zero
+    otherwise, the knock-in complement of :class:`UpAndOutCall`. Holding both
+    at one strike and barrier reconstructs the vanilla call, the in-out
+    parity the test suite pins. The barrier is monitored discretely on the
+    rebalancing grid, inception included.
+
+    Attributes:
+        strike: Strike price K.
+        barrier: Knock-in level B, strictly above the strike.
+    """
+
+    strike: float
+    barrier: float
+
+    def __post_init__(self) -> None:
+        if self.barrier <= self.strike:
+            raise ValueError(
+                f"barrier must exceed strike, got barrier={self.barrier} strike={self.strike}"
+            )
+
+    def __call__(self, paths: torch.Tensor) -> torch.Tensor:
+        """Computes the knocked-in payoff from the full path.
+
+        Args:
+            paths: Price paths of shape ``(n_steps + 1, n_paths)``.
+
+        Returns:
+            Payoff per path of shape ``(n_paths,)``.
+        """
+        knocked_in = paths.max(dim=0).values >= self.barrier
+        vanilla = torch.clamp(paths[-1] - self.strike, min=0.0)
+        return vanilla * knocked_in
+
+
+@dataclass(frozen=True)
+class DownAndOutCall:
+    """Down-and-out call, a vanilla call knocked out at a lower barrier.
+
+    Pays ``max(S_T - K, 0)`` if the path never falls to the barrier and zero
+    otherwise. The barrier sits below the spot, so a level at or above the
+    initial spot makes the contract worthless from the start, which the
+    constructor cannot reject because the payoff never sees the spot.
+    Monitoring is discrete on the rebalancing grid, inception included.
+
+    Attributes:
+        strike: Strike price K.
+        barrier: Knock-out level B, a positive level below the spot.
+    """
+
+    strike: float
+    barrier: float
+
+    def __post_init__(self) -> None:
+        if self.barrier <= 0.0:
+            raise ValueError(f"barrier must be positive, got {self.barrier}")
+
+    def __call__(self, paths: torch.Tensor) -> torch.Tensor:
+        """Computes the knocked payoff from the full path.
+
+        Args:
+            paths: Price paths of shape ``(n_steps + 1, n_paths)``.
+
+        Returns:
+            Payoff per path of shape ``(n_paths,)``.
+        """
+        alive = paths.min(dim=0).values > self.barrier
+        vanilla = torch.clamp(paths[-1] - self.strike, min=0.0)
+        return vanilla * alive
+
+
+@dataclass(frozen=True)
+class DownAndInCall:
+    """Down-and-in call, a vanilla call that activates only below a barrier.
+
+    Pays ``max(S_T - K, 0)`` only if the path falls to the barrier, the
+    knock-in complement of :class:`DownAndOutCall`, and the two together at
+    one strike and barrier reconstruct the vanilla call. Monitoring is
+    discrete on the rebalancing grid, inception included.
+
+    Attributes:
+        strike: Strike price K.
+        barrier: Knock-in level B, a positive level below the spot.
+    """
+
+    strike: float
+    barrier: float
+
+    def __post_init__(self) -> None:
+        if self.barrier <= 0.0:
+            raise ValueError(f"barrier must be positive, got {self.barrier}")
+
+    def __call__(self, paths: torch.Tensor) -> torch.Tensor:
+        """Computes the knocked-in payoff from the full path.
+
+        Args:
+            paths: Price paths of shape ``(n_steps + 1, n_paths)``.
+
+        Returns:
+            Payoff per path of shape ``(n_paths,)``.
+        """
+        knocked_in = paths.min(dim=0).values <= self.barrier
+        vanilla = torch.clamp(paths[-1] - self.strike, min=0.0)
+        return vanilla * knocked_in
+
+
+@dataclass(frozen=True)
+class DoubleKnockOutCall:
+    """Call knocked out by either an upper or a lower barrier.
+
+    Pays ``max(S_T - K, 0)`` only while the path stays strictly inside the
+    corridor and zero once it touches either side, so the contract is worth
+    no more than the single-barrier knock-outs and cheaper still. Monitoring
+    is discrete on the rebalancing grid, inception included.
+
+    Attributes:
+        strike: Strike price K.
+        lower_barrier: Lower knock-out level, positive and below the upper.
+        upper_barrier: Upper knock-out level, above the lower.
+    """
+
+    strike: float
+    lower_barrier: float
+    upper_barrier: float
+
+    def __post_init__(self) -> None:
+        if self.lower_barrier <= 0.0:
+            raise ValueError(f"lower_barrier must be positive, got {self.lower_barrier}")
+        if self.upper_barrier <= self.lower_barrier:
+            raise ValueError(
+                f"upper_barrier must exceed lower_barrier, got upper={self.upper_barrier} "
+                f"lower={self.lower_barrier}"
+            )
+
+    def __call__(self, paths: torch.Tensor) -> torch.Tensor:
+        """Computes the corridor-knocked payoff from the full path.
+
+        Args:
+            paths: Price paths of shape ``(n_steps + 1, n_paths)``.
+
+        Returns:
+            Payoff per path of shape ``(n_paths,)``.
+        """
+        alive = (paths.min(dim=0).values > self.lower_barrier) & (
+            paths.max(dim=0).values < self.upper_barrier
+        )
+        vanilla = torch.clamp(paths[-1] - self.strike, min=0.0)
+        return vanilla * alive

tests/unit/test_barrier.py

@@ -3,7 +3,14 @@
 import pytest
 import torch
 
-from deephedging.instruments import EuropeanCall, UpAndOutCall
+from deephedging.instruments import (
+    DoubleKnockOutCall,
+    DownAndInCall,
+    DownAndOutCall,
+    EuropeanCall,
+    UpAndInCall,
+    UpAndOutCall,
+)
 from deephedging.market import GBMSimulator, NoiseSpec
 
 
@@ -44,3 +51,68 @@ def test_barrier_payoff_never_exceeds_vanilla() -> None:
 def test_barrier_below_strike_rejected() -> None:
     with pytest.raises(ValueError):
         UpAndOutCall(strike=100.0, barrier=90.0)
+
+
+def test_up_in_out_parity_reconstructs_vanilla() -> None:
+    paths = _paths()
+    vanilla = EuropeanCall(strike=100.0)(paths)
+    knocked_out = UpAndOutCall(strike=100.0, barrier=130.0)(paths)
+    knocked_in = UpAndInCall(strike=100.0, barrier=130.0)(paths)
+    assert torch.allclose(knocked_out + knocked_in, vanilla)
+
+
+def test_down_in_out_parity_reconstructs_vanilla() -> None:
+    paths = _paths()
+    vanilla = EuropeanCall(strike=100.0)(paths)
+    knocked_out = DownAndOutCall(strike=100.0, barrier=80.0)(paths)
+    knocked_in = DownAndInCall(strike=100.0, barrier=80.0)(paths)
+    assert torch.allclose(knocked_out + knocked_in, vanilla)
+
+
+def test_down_and_out_unreachable_barrier_equals_vanilla() -> None:
+    paths = _paths()
+    vanilla = EuropeanCall(strike=100.0)(paths)
+    knocked = DownAndOutCall(strike=100.0, barrier=1e-6)(paths)
+    assert torch.equal(knocked, vanilla)
+
+
+def test_double_knock_out_stays_within_single_barriers() -> None:
+    paths = _paths()
+    double = DoubleKnockOutCall(strike=100.0, lower_barrier=80.0, upper_barrier=130.0)(paths)
+    up = UpAndOutCall(strike=100.0, barrier=130.0)(paths)
+    down = DownAndOutCall(strike=100.0, barrier=80.0)(paths)
+    assert torch.all(double <= up + 1e-9)
+    assert torch.all(double <= down + 1e-9)
+
+
+def test_double_knock_out_wide_corridor_equals_vanilla() -> None:
+    paths = _paths()
+    vanilla = EuropeanCall(strike=100.0)(paths)
+    double = DoubleKnockOutCall(strike=100.0, lower_barrier=1e-6, upper_barrier=1e9)(paths)
+    assert torch.equal(double, vanilla)
+
+
+def test_corridor_pays_zero_on_either_breach() -> None:
+    paths = torch.tensor(
+        [
+            [100.0, 100.0, 100.0],
+            [105.0, 70.0, 140.0],
+            [112.0, 95.0, 110.0],
+        ]
+    )
+    payoff = DoubleKnockOutCall(strike=100.0, lower_barrier=80.0, upper_barrier=130.0)
+    result = payoff(paths)
+    assert float(result[0]) == 12.0
+    assert float(result[1]) == 0.0
+    assert float(result[2]) == 0.0
+
+
+def test_barrier_family_validation() -> None:
+    with pytest.raises(ValueError):
+        UpAndInCall(strike=100.0, barrier=90.0)
+    with pytest.raises(ValueError):
+        DownAndOutCall(strike=100.0, barrier=0.0)
+    with pytest.raises(ValueError):
+        DownAndInCall(strike=100.0, barrier=-1.0)
+    with pytest.raises(ValueError):
+        DoubleKnockOutCall(strike=100.0, lower_barrier=120.0, upper_barrier=110.0)