ENH: add non-deliverable forward

docs/source/i_whatsnew.rst

@@ -42,6 +42,10 @@ objects now have specific methods to allow *updates*.
        :class:`~rateslib.instruments.FloatRateNote` to allow the calculation of
        yield-to-maturity for that *Instrument* based on ``calc_mode`` similar to
        *FixedRateBonds*. (`529 <https://github.com/attack68/rateslib/pull/529>`_)
+   * - Instruments
+     - :class:`~rateslib.periods.NonDeliverableCashflow` added to allow FX forwards settled in
+       an alternate currency to be valued.
+       (`647 <https://github.com/attack68/rateslib/pull/647>`_)
    * - Splines
      - The *Spline* :meth:`~rateslib.splines.evaluate` method is enhanced to allow an x-axis
        evaluation if a :class:`~rateslib.dual.Variable` is passed, through dynamic *Dual* or *Dual2*

python/rateslib/instruments/rates/multi_currency.py

@@ -265,6 +265,71 @@ def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes:
         raise NotImplementedError("`analytic_delta` for FXExchange not defined.")
 
 
+# class NDF(Sensitivities, BaseMixin):
+#
+#     leg1: FXExchange
+#
+#     def __init__(
+#         self,
+#         settlement: datetime,
+#         pair: str,
+#         settlement_currency: str,
+#         fx_rate: DualTypes_ = NoInput(0),
+#         notional: DualTypes_ = NoInput(0),
+#         curves: Curves_ = NoInput(0),
+#     ):
+#
+#     def rate(
+#         self,
+#         curves: Curves_ = NoInput(0),
+#         solver: Solver_ = NoInput(0),
+#         fx: FX_ = NoInput(0),
+#         base: str_ = NoInput(0),
+#     ):
+#         curves_, fx_, base_ = _get_curves_fx_and_base_maybe_from_solver(
+#             self.curves,
+#             solver,
+#             curves,
+#             fx,
+#             base,
+#             self.leg1.currency,
+#         )
+#         curves_1 = _validate_curve_not_no_input(curves_[1])
+#         curves_3 = _validate_curve_not_no_input(curves_[3])
+#
+#         if isinstance(fx_, FXRates | FXForwards):
+#             imm_fx: DualTypes = fx_.rate(self.pair)
+#         elif isinstance(fx_, NoInput):
+#             raise ValueError(
+#                 "`fx` must be supplied to price FXExchange object.\n"
+#                 "Note: it can be attached to and then gotten from a Solver.",
+#             )
+#         else:
+#             imm_fx = fx_
+#
+#         _: DualTypes = forward_fx(self.settlement, curves_1, curves_3, imm_fx)
+#         return _
+#
+#     def delta(self, *args: Any, **kwargs: Any) -> DataFrame:
+#         """
+#         Calculate the delta of the *Instrument*.
+#
+#         For arguments see :meth:`Sensitivities.delta()<rateslib.instruments.Sensitivities.delta>`.
+#         """
+#         return super().delta(*args, **kwargs)
+#
+#     def gamma(self, *args: Any, **kwargs: Any) -> DataFrame:
+#         """
+#         Calculate the gamma of the *Instrument*.
+#
+#         For arguments see :meth:`Sensitivities.gamma()<rateslib.instruments.Sensitivities.gamma>`.
+#         """
+#         return super().gamma(*args, **kwargs)
+#
+#     def analytic_delta(self, *args: Any, **kwargs: Any) -> DualTypes:
+#         raise NotImplementedError("`analytic_delta` for NDF not defined.")
+
+
 class XCS(BaseDerivative):
     """
     Create a cross-currency swap (XCS) composing relevant fixed or floating *Legs*.

python/rateslib/periods.py

@@ -63,6 +63,7 @@
     _d_plus_min_u,
     _delta_type_constants,
 )
+from rateslib.instruments.utils import _validate_fx_as_forwards
 from rateslib.splines import evaluate
 
 if TYPE_CHECKING:
@@ -2756,6 +2757,193 @@ def analytic_delta(
         return 0.0
 
 
+class NonDeliverableCashflow:
+    """
+    Create a cashflow amount associated with a non-deliverable FX forward.
+
+    Parameters
+    ----------
+    notional : float, Dual, Dual2
+        The notional amount of the cashflow expressed in units of the ``reference_currency``.
+    reference_currency : str
+        The non-deliverable reference currency (3-digit code).
+    settlement_currency : str
+        The currency of the deliverable currency (3-digit code), e.g. "usd" or "eur".
+    settlement : datetime
+        The settlement date of the exchange.
+    fixing_date: datetime
+        The date on which the FX fixings will be recorded.
+    fx_rate: float, Dual, Dual2, optional
+        The pricing parameter of the period to record the entry level of the transaction.
+        The ``reference_currency`` should be the left hand side, e.g. BRLUSD, not a USDBRL rate.
+    fx_fixing: float, Dual, Dual2, optional
+        The FX fixing to determine the settlement amount.
+        The ``reference_currency`` should be the left hand side.
+
+    Notes
+    -----
+    The ``cashflow`` is defined as follows;
+
+    .. math::
+
+       C = N (f_2 - f_1)
+
+    where :math:`f_1` is the ``fx_rate``, :math:`f_2` is the ``fx_fixing`` or market forecast
+    rate at settlement. This amount is expressed in units of ``settlement_currency``.
+
+    The :meth:`~rateslib.periods.BasePeriod.npv` is defined as;
+
+    .. math::
+
+       P = Cv(m) = N (f_2 - f_1) v(m)
+
+    The :meth:`~rateslib.periods.BasePeriod.analytic_delta` is defined as;
+
+    .. math::
+
+       A = 0
+
+    Example
+    -------
+    .. ipython:: python
+
+       ndc = NonDeliverableCashflow(
+           notional=10e6,
+           reference_currency="brl",
+           settlement_currency="usd",
+           settlement=dt(2025, 6, 1),
+           fixing_date=dt(2025, 5, 29),
+           fx_rate=0.200,
+       )
+       ndc.cashflows()
+    """
+
+    def __init__(
+        self,
+        notional: DualTypes,
+        reference_currency: str,
+        settlement_currency: str,
+        settlement: datetime,
+        fixing_date: datetime,
+        fx_rate: DualTypes_ = NoInput(0),
+        fx_fixing: DualTypes_ = NoInput(0),
+    ):
+        self.notional = notional
+        self.settlement = settlement
+        self.settlement_currency = settlement_currency.lower()
+        self.reference_currency = reference_currency.lower()
+        self.pair = f"{self.reference_currency}{self.settlement_currency}"
+        self.fixing_date = fixing_date
+        self.fx_rate = fx_rate
+        self.fx_fixing = fx_fixing
+
+    def analytic_delta(
+        self,
+        curve: Curve_ = NoInput(0),
+        disc_curve: Curve_ = NoInput(0),
+        fx: FX_ = NoInput(0),
+        base: str_ = NoInput(0),
+    ) -> DualTypes:
+        """
+        Return the analytic delta of the *NonDeliverableCashflow*.
+        See
+        :meth:`BasePeriod.analytic_delta()<rateslib.periods.BasePeriod.analytic_delta>`
+        """
+        return 0.0
+
+    def npv(
+        self,
+        curve: CurveOption_ = NoInput(0),
+        disc_curve: Curve_ = NoInput(0),
+        fx: FX_ = NoInput(0),
+        base: str_ = NoInput(0),
+        local: bool = False,
+    ) -> NPV:
+        """
+        Return the NPV of the *NonDeliverableCashflow*.
+        See
+        :meth:`BasePeriod.npv()<rateslib.periods.BasePeriod.npv>`
+        """
+        disc_curve_: Curve = _disc_required_maybe_from_curve(curve, disc_curve)
+        disc_cashflow = self.cashflow(fx) * disc_curve_[self.settlement]
+        return _maybe_local(disc_cashflow, local, self.settlement_currency, fx, base)
+
+    def cashflow(self, fx: FX_) -> DualTypes:
+        """Cashflow is expressed in the settlement, i.e. deliverable currency."""
+        if isinstance(self.fx_fixing, NoInput):
+            fx_ = _validate_fx_as_forwards(fx)
+            fx_fixing: DualTypes = fx_.rate(self.pair, self.settlement)
+        else:
+            fx_fixing = self.fx_fixing
+
+        try:
+            d_value: DualTypes = self.notional * (fx_fixing - self.fx_rate)  # type: ignore[operator]
+        except TypeError as e:
+            # either fixed rate is None
+            if isinstance(self.fx_rate, NoInput):
+                raise TypeError("`fx_rate` must be set on the Period for an `npv`.")
+            else:
+                raise e
+
+        return d_value
+
+    def cashflows(
+        self,
+        curve: CurveOption_ = NoInput(0),
+        disc_curve: Curve_ = NoInput(0),
+        fx: FX_ = NoInput(0),
+        base: str_ = NoInput(0),
+    ) -> dict[str, Any]:
+        """
+        Return the cashflows of the *NonDeliverableCashflow*.
+        See
+        :meth:`BasePeriod.cashflows()<rateslib.periods.BasePeriod.cashflows>`
+        """
+        disc_curve_: Curve_ = _disc_maybe_from_curve(curve, disc_curve)
+        imm_fx_to_base, _ = _get_fx_and_base(self.settlement_currency, fx, base)
+
+        if isinstance(disc_curve_, NoInput) or not isinstance(fx, FXForwards):
+            npv, npv_fx, df, collateral, cashflow = None, None, None, None, None
+            index_val_ = None
+        else:
+            npv_: DualTypes = self.npv(curve, disc_curve_, fx)  # type: ignore[assignment]
+            npv = _dual_float(npv_)
+            index_val_ = self.rate(fx)
+
+            npv_fx = npv * _dual_float(imm_fx_to_base)
+            df, collateral = _dual_float(disc_curve_[self.settlement]), disc_curve_.collateral
+            cashflow = _dual_float(self.cashflow(fx))
+
+        rate = _float_or_none(_drb(None, self.fx_rate))
+        return {
+            defaults.headers["type"]: type(self).__name__,
+            defaults.headers["stub_type"]: f"{self.pair.upper()}",
+            defaults.headers["currency"]: self.settlement_currency.upper(),
+            # defaults.headers["a_acc_start"]: None,
+            # defaults.headers["a_acc_end"]: None,
+            defaults.headers["payment"]: self.settlement,
+            # defaults.headers["convention"]: None,
+            # defaults.headers["dcf"]: None,
+            defaults.headers["notional"]: _dual_float(self.notional),
+            defaults.headers["df"]: df,
+            defaults.headers["rate"]: rate,
+            defaults.headers["index_value"]: _float_or_none(index_val_),
+            # defaults.headers["spread"]: None,
+            defaults.headers["cashflow"]: cashflow,
+            defaults.headers["npv"]: npv,
+            defaults.headers["fx"]: _dual_float(imm_fx_to_base),
+            defaults.headers["npv_fx"]: npv_fx,
+            defaults.headers["collateral"]: collateral,
+        }
+
+    def rate(self, fx: FX_) -> DualTypes:
+        if isinstance(self.fx_fixing, NoInput):
+            fx_ = _validate_fx_as_forwards(fx)
+            return fx_.rate(self.pair, self.settlement)
+        else:
+            return self.fx_fixing
+
+
 # Licence: Creative Commons - Attribution-NonCommercial-NoDerivatives 4.0 International
 # Commercial use of this code, and/or copying and redistribution is prohibited.
 # Contact rateslib at gmail.com if this code is observed outside its intended sphere.