text: # Copyright 2019 Google LLC
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     https://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  # Lint as: python2, python3
  """Tests for bond_curve."""
  
  from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  
  import numpy as np
  import tensorflow as tf
  
  from tf_quant_finance.rates.hagan_west import bond_curve
  from tf_quant_finance.rates.hagan_west import monotone_convex
  
  from tensorflow.python.framework import test_util  # pylint: disable=g-direct-tensorflow-import
  
  
  @test_util.run_all_in_graph_and_eager_modes
  class BondCurveTest(tf.test.TestCase):
  
    def test_cashflow_times_cashflow_before_settelment_error(self):
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        with self.assertRaises(tf.errors.InvalidArgumentError):
          self.evaluate(
              bond_curve.bond_curve(
                  bond_cashflows=[
                      np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
                      np.array([30.0, 30.0, 30.0, 1030.0], dtype=dtype)
                  ],
                  bond_cashflow_times=[
                      np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
                      np.array([0.5, 1.0, 1.5, 2.0], dtype=dtype)
                  ],
                  present_values=np.array([999.0, 1022.0], dtype=dtype),
                  present_values_settlement_times=np.array([0.25, 0.25],
                                                           dtype=dtype),
                  validate_args=True,
                  dtype=dtype))
  
    def test_cashflow_times_are_strongly_ordered_error(self):
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        with self.assertRaises(tf.errors.InvalidArgumentError):
          self.evaluate(
              bond_curve.bond_curve(
                  bond_cashflows=[
                      np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
                      np.array([30.0, 30.0, 30.0, 1030.0], dtype=dtype)
                  ],
                  bond_cashflow_times=[
                      np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
                      np.array([0.5, 1.0, 1.5, 1.5], dtype=dtype)
                  ],
                  present_values=np.array([999.0, 1022.0], dtype=dtype),
                  validate_args=True,
                  dtype=dtype))
  
    def test_final_cashflow_is_the_largest_error(self):
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        with self.assertRaises(tf.errors.InvalidArgumentError):
          self.evaluate(
              bond_curve.bond_curve(
                  bond_cashflows=[
                      np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
                      np.array([30.0, 30.0, 30.0, 3.0], dtype=dtype)
                  ],
                  bond_cashflow_times=[
                      np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
                      np.array([0.5, 1.0, 1.5, 2.0], dtype=dtype)
                  ],
                  present_values=np.array([999.0, 1022.0], dtype=dtype),
                  validate_args=True,
                  dtype=dtype))
  
    def test_correctness(self):
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        cashflows = [
            # 1 year bond with 5% three monthly coupon.
            np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
            # 2 year bond with 6% semi-annual coupon.
            np.array([30, 30, 30, 1030], dtype=dtype),
            # 3 year bond with 8% semi-annual coupon.
            np.array([40, 40, 40, 40, 40, 1040], dtype=dtype),
            # 4 year bond with 3% semi-annual coupon.
            np.array([15, 15, 15, 15, 15, 15, 15, 1015], dtype=dtype)
        ]
        cashflow_times = [
            np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0], dtype=dtype)
        ]
        pvs = np.array([
            999.68155223943393, 1022.322872470043, 1093.9894418810143,
            934.20885689015677
        ],
                       dtype=dtype)
        results = self.evaluate(
            bond_curve.bond_curve(
                cashflows, cashflow_times, pvs, validate_args=True, dtype=dtype))
  
        np.testing.assert_allclose(results.times, [1.0, 2.0, 3.0, 4.0])
  
        self.assertTrue(results.converged)
        self.assertFalse(results.failed)
        expected_discount_rates = np.array([5.0, 4.75, 4.53333333, 4.775],
                                           dtype=dtype) / 100
        expected_discount_factors = np.exp(-expected_discount_rates *
                                           [1.0, 2.0, 3.0, 4.0])
        np.testing.assert_allclose(
            results.discount_rates, expected_discount_rates, atol=1e-6)
        np.testing.assert_allclose(
            results.discount_factors, expected_discount_factors, atol=1e-6)
  
    def test_unstable(self):
      """Demonstrates the instability of Hagan West for extreme cases."""
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        cashflows = [
            # 1 year bond with 5% three monthly coupon.
            np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
            # 2 year bond with 6% semi-annual coupon.
            np.array([30, 30, 30, 1030], dtype=dtype),
            # 3 year bond with 8% semi-annual coupon.
            np.array([40, 40, 40, 40, 40, 1040], dtype=dtype),
            # 4 year bond with 3% semi-annual coupon.
            np.array([15, 15, 15, 15, 15, 15, 15, 1015], dtype=dtype)
        ]
        cashflow_times = [
            np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0], dtype=dtype)
        ]
        # Computed with discount rates of [5.0, 4.75, 4.53333333, 4.775]
        # which are 100 times the values in the previous test case.
        pvs = np.array([
            11.561316110080888, 2.6491572753698067, 3.4340789041846866,
            1.28732090544209
        ],
                       dtype=dtype)
        true_discount_rates = np.array([5.0, 4.75, 4.53333333, 4.775],
                                       dtype=dtype)
        # Check failure with default initial rates.
        results_default = self.evaluate(
            bond_curve.bond_curve(
                cashflows,
                cashflow_times,
                pvs,
                maximum_iterations=100,
                validate_args=True,
                dtype=dtype))
        self.assertFalse(results_default.converged)
        self.assertTrue(results_default.failed)
        self.assertFalse(np.isnan(results_default.discount_rates[0]))
        self.assertTrue(np.isnan(results_default.discount_rates[1]))
  
        # It even fails if we underestimate the result even marginally.
        # However the behaviour is different if we start above the true values.
        # See next test.
        results_close = self.evaluate(
            bond_curve.bond_curve(
                cashflows,
                cashflow_times,
                pvs,
                initial_discount_rates=true_discount_rates * 0.9999,
                maximum_iterations=100))
        self.assertFalse(results_close.converged)
        self.assertTrue(results_close.failed)
        self.assertFalse(np.isnan(results_close.discount_rates[0]))
        self.assertFalse(np.isnan(results_close.discount_rates[1]))
        self.assertTrue(np.isnan(results_close.discount_rates[2]))
  
    def test_non_convex(self):
      """Demonstrates the nonconvexity of Hagan West for extreme cases."""
      # This is the same example as the previous one but with different starting
      # point.
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        cashflows = [
            # 1 year bond with 5% three monthly coupon.
            np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
            # 2 year bond with 6% semi-annual coupon.
            np.array([30, 30, 30, 1030], dtype=dtype),
            # 3 year bond with 8% semi-annual coupon.
            np.array([40, 40, 40, 40, 40, 1040], dtype=dtype),
            # 4 year bond with 3% semi-annual coupon.
            np.array([15, 15, 15, 15, 15, 15, 15, 1015], dtype=dtype)
        ]
        cashflow_times = [
            np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0], dtype=dtype)
        ]
        # Computed with discount rates of [5.0, 4.75, 4.53333333, 4.775]
        # which are 100 times the values in the previous test case.
        pvs = np.array([
            11.561316110080888, 2.6491572753698067, 3.4340789041846866,
            1.28732090544209
        ],
                       dtype=dtype)
        true_discount_rates = np.array([5.0, 4.75, 4.53333333, 4.775],
                                       dtype=dtype)
        initial_rates = true_discount_rates * 1.01
        # Check failure with default initial rates.
        results = self.evaluate(
            bond_curve.bond_curve(
                cashflows,
                cashflow_times,
                pvs,
                initial_discount_rates=initial_rates,
                maximum_iterations=100,
                validate_args=True,
                dtype=dtype))
        self.assertTrue(results.converged)
        self.assertFalse(results.failed)
        # It converges to a different set of rates.
        np.testing.assert_allclose(
            results.discount_rates,
            [4.9610328, 4.17662715, 2.84038942, 2.38737021],
            atol=1e-6)
  
        # However, the actual bond prices with the returned rates are indeed
        # correct.
        implied_pvs = self.evaluate(
            _compute_pv(cashflows, cashflow_times, results.discount_rates,
                        np.array([1.0, 2.0, 3.0, 4.0], dtype=dtype)))
  
        np.testing.assert_allclose(implied_pvs, pvs, rtol=1e-5)
  
    def test_flat_curve(self):
      """Checks that flat curves work."""
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        cashflows = [
            # 1 year bond with 5% three monthly coupon.
            np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
            # 2 year bond with 6% semi-annual coupon.
            np.array([30, 30, 30, 1030], dtype=dtype),
            # 3 year bond with 8% semi-annual coupon.
            np.array([40, 40, 40, 40, 40, 1040], dtype=dtype),
            # 4 year bond with 3% semi-annual coupon.
            np.array([15, 15, 15, 15, 15, 15, 15, 1015], dtype=dtype)
        ]
        cashflow_times = [
            np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0], dtype=dtype)
        ]
        # Computed with a flat curve of 15%.
        pvs = np.array([906.27355957, 840.6517334, 823.73626709, 635.7076416],
                       dtype=dtype)
        true_discount_rates = np.array([0.15] * 4, dtype=dtype)
        results = self.evaluate(
            bond_curve.bond_curve(
                cashflows, cashflow_times, pvs, validate_args=True, dtype=dtype))
        self.assertTrue(results.converged)
        self.assertFalse(results.failed)
        np.testing.assert_allclose(
            results.discount_rates, true_discount_rates, atol=1e-6)
  
    def test_negative_rates(self):
      """Checks that method works even if the actual rates are negative."""
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        cashflows = [
            # 1 year bond with 5% three monthly coupon.
            np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
            # 2 year bond with 6% semi-annual coupon.
            np.array([30, 30, 30, 1030], dtype=dtype),
            # 3 year bond with 8% semi-annual coupon.
            np.array([40, 40, 40, 40, 40, 1040], dtype=dtype),
            # 4 year bond with 3% semi-annual coupon.
            np.array([15, 15, 15, 15, 15, 15, 15, 1015], dtype=dtype)
        ]
        cashflow_times = [
            np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0], dtype=dtype)
        ]
        pvs = np.array(
            [1029.54933442, 1097.95320227, 1268.65376174, 1249.84175959],
            dtype=dtype)
        true_discount_rates = np.array([0.02, 0.01, -0.01, -0.03], dtype=dtype)
        results = self.evaluate(
            bond_curve.bond_curve(
                cashflows, cashflow_times, pvs, validate_args=True, dtype=dtype))
        self.assertTrue(results.converged)
        self.assertFalse(results.failed)
        np.testing.assert_allclose(
            results.discount_rates, true_discount_rates, atol=1e-4)
  
    def test_negative_forwards(self):
      """Checks that method works if the rates are positive by fwds are not."""
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        true_discount_rates = np.array([0.12, 0.09, 0.02, 0.01, 0.01318182],
                                       dtype=dtype)
        # Note the implied forward rates for this rate curve are:
        # [0.12, 0.06, -0.05, -0.01, 0.02]
        cashflows = [
            np.array([1.2, 10.], dtype=dtype),
            np.array([1.1, 2.2, 1.4, 15.5], dtype=dtype),
            np.array([1.22, 0.45, 2.83, 96.0], dtype=dtype),
            np.array([12.33, 9.84, 1.15, 11.87, 0.66, 104.55], dtype=dtype),
            np.array([5.84, 0.23, 5.23, 114.95], dtype=dtype)
        ]
        cashflow_times = [
            np.array([0.15, 0.25], dtype=dtype),
            np.array([0.1, 0.2, 0.4, 0.5], dtype=dtype),
            np.array([0.22, 0.45, 0.93, 1.0], dtype=dtype),
            np.array([0.33, 0.84, 0.92, 1.22, 1.45, 1.5], dtype=dtype),
            np.array([0.43, 0.77, 1.3, 2.2], dtype=dtype)
        ]
        pvs = np.array(
            [10.88135262, 19.39268844, 98.48426722, 137.91938533, 122.63546542],
            dtype=dtype)
        results = self.evaluate(
            bond_curve.bond_curve(
                cashflows, cashflow_times, pvs, validate_args=True, dtype=dtype))
        self.assertTrue(results.converged)
        self.assertFalse(results.failed)
        self.assertEqual(results.iterations, 6)
        np.testing.assert_allclose(
            results.discount_rates, true_discount_rates, atol=1e-6)
        np.testing.assert_allclose(
            results.times, [0.25, 0.5, 1., 1.5, 2.2], atol=1e-6)
  
    def test_zero_coupon_raises(self):
      dtypes = [np.float64, np.float32]
      for dtype in dtypes:
        cashflows = [
            # 1 year bond with 5% three monthly coupon.
            np.array([12.5, 12.5, 12.5, 1012.5], dtype=dtype),
            # 2 year bond with no coupons. This should cause an error.
            np.array([1030], dtype=dtype),
            # 3 year bond with 8% semi-annual coupon.
            np.array([40, 40, 40, 40, 40, 1040], dtype=dtype),
            # 4 year bond with 3% semi-annual coupon.
            np.array([15, 15, 15, 15, 15, 15, 15, 1015], dtype=dtype)
        ]
        cashflow_times = [
            np.array([0.25, 0.5, 0.75, 1.0], dtype=dtype),
            np.array([2.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0], dtype=dtype),
            np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0], dtype=dtype)
        ]
        pvs = np.array([
            999.68155223943393, 1022.322872470043, 1093.9894418810143,
            934.20885689015677
        ],
                       dtype=dtype)
        with self.assertRaises(tf.errors.InvalidArgumentError):
          self.evaluate(
              bond_curve.bond_curve(
                  cashflows, cashflow_times, pvs, validate_args=True,
                  dtype=dtype))
  
  
  def _compute_pv(cashflows, cashflow_times, reference_rates, reference_times):
    times = tf.concat(cashflow_times, axis=0)
    groups = tf.concat([
        tf.zeros_like(cashflow, dtype=tf.int32) + i
        for i, cashflow in enumerate(cashflows)
    ],
                       axis=0)
    rates = monotone_convex.interpolate_yields(
        times, reference_times, yields=reference_rates)
    discounts = tf.math.exp(-times * rates)
    cashflows = tf.concat(cashflows, axis=0)
    return tf.math.segment_sum(discounts * cashflows, groups)
  
  
  if __name__ == '__main__':
    tf.test.main()
  