code: def calibration_from_cap_floors(
      *,
      prices: types.RealTensor,
      strikes: types.RealTensor,
      expiries: types.RealTensor,
      maturities: types.RealTensor,
      daycount_fractions: types.RealTensor,
      reference_rate_fn,
      mean_reversion: Union[
          types.RealTensor, Callable[..., types.RealTensor]
      ],
      volatility: Union[
          types.RealTensor, Callable[..., types.RealTensor]
      ],
      notional: types.RealTensor = None,
      is_cap: types.BoolTensor = True,
      use_analytic_pricing: bool = True,
      num_samples: types.IntTensor = 1,
      random_type: random.RandomType = None,
      seed: types.IntTensor = None,
      skip: types.IntTensor = 0,
      time_step: types.RealTensor = None,
      optimizer_fn=None,
      mean_reversion_lower_bound=0.001,
      mean_reversion_upper_bound=0.5,
      volatility_lower_bound: types.RealTensor = 0.00001,
      volatility_upper_bound: types.RealTensor = 0.1,
      tolerance: types.RealTensor = 1e-6,
      maximum_iterations: types.IntTensor = 50,
      dtype: tf.DType = None,
      name: str = None,
  ) -> Tuple[CalibrationResult, types.BoolTensor, types.IntTensor]:
      """Calibrates the Hull-White model using the observed Cap/Floor prices.
  
      This function estimates the mean-reversion rate and volatility parameters of
      a Hull-White 1-factor model using a set of Cap/Floor prices as the target.
      The calibration is performed using least-squares optimization where
      the loss function minimizes the squared error between the observed option
      prices and the model implied prices.
  
      #### Example
      The example shows how to calibrate a Hull-White model with constant mean
      reversion rate and constant volatility.
  
      ````python
      import numpy as np
      import tensorflow.compat.v2 as tf
      import tf_quant_finance as tff
  
      # In this example, we synthetically generate some prices. Then we use our
      # calibration to back out these prices.
      dtype = tf.float64
  
      daycount_fractions = np.array([
          [0.25, 0.25, 0.25, 0.25, 0.0, 0.0, 0.0, 0.0],
          [0.25, 0.25, 0.25, 0.25, 0.0, 0.0, 0.0, 0.0],
          [0.25, 0.25, 0.25, 0.25, 0.25, 0.25, 0.25, 0.25],
          [0.25, 0.25, 0.25, 0.25, 0.25, 0.25, 0.25, 0.25],
      ])
      expiries = np.array([
          [0.0, 0.25, 0.5, 0.75, 1.0, 0.0, 0.0, 0.0],
          [0.0, 0.25, 0.5, 0.75, 1.0, 0.0, 0.0, 0.0],
          [0.0, 0.25, 0.5, 0.75, 1.0, 1.25, 1.50, 1.75],
          [0.0, 0.25, 0.5, 0.75, 1.0, 1.25, 1.50, 1.75],
      ])
      maturities = np.array([
          [0.25, 0.5, 0.75, 1.0, 0.0, 0.0, 0.0, 0.0],
          [0.25, 0.5, 0.75, 1.0, 0.0, 0.0, 0.0, 0.0],
          [0.25, 0.5, 0.75, 1.0, 1.25, 1.50, 1.75, 2.0],
          [0.25, 0.5, 0.75, 1.0, 1.25, 1.50, 1.75, 2.0],
      ])
      is_cap = np.array([True, False, True, False])
      strikes = 0.01 * np.ones_like(expiries)
  
      # Setup - generate some observed prices using the model.
      expected_mr = [0.4]
      expected_vol = [0.01]
  
      zero_rate_fn = lambda x: 0.01 * tf.ones_like(x, dtype=dtype)
      prices = tff.models.hull_white.cap_floor_price(
          strikes=strikes,
          expiries=expiries,
          maturities=maturities,
          daycount_fractions=daycount_fractions,
          reference_rate_fn=zero_rate_fn,
          notional=1.0,
          mean_reversion=expected_mr,
          volatility=expected_vol,
          is_cap=tf.expand_dims(is_cap, axis=1),
          use_analytic_pricing=True,
          dtype=dtype)
  
      # Calibrate the model.
      calibrated_model, is_converged, _ = (
          tff.models.hull_white.calibration_from_cap_floors(
              prices=tf.squeeze(prices),
              strikes=strikes,
              expiries=expiries,
              maturities=maturities,
              daycount_fractions=daycount_fractions,
              reference_rate_fn=zero_rate_fn,
              mean_reversion=[0.3],
              volatility=[0.02],
              notional=1.0,
              is_cap=tf.expand_dims(is_cap, axis=1),
              use_analytic_pricing=True,
              optimizer_fn=None,
              num_samples=1000,
              random_type=tff.math.random.RandomType.STATELESS_ANTITHETIC,
              seed=[0, 0],
              time_step=0.1,
              maximum_iterations=200,
              dtype=dtype))
  
      calibrated_mr = calibrated_model.mean_reversion.values()
      calibrated_vol = calibrated_model.volatility.values()
  
      # Running this inside a unit test passes:
      #
      # calibrated_mr, calibrated_vol = self.evaluate(
      #     [calibrated_mr, calibrated_vol])
      # self.assertTrue(is_converged)
      # self.assertAllClose(calibrated_mr, expected_mr, atol=1e-3, rtol=1e-2)
      # self.assertAllClose(calibrated_vol, expected_vol, atol=1e-3, rtol=1e-2)
  
      ````
  
      Args:
        prices: A real `Tensor` of shape [num_capfloors], holding the prices of
          cap/floors used for calibration; e.g. `prices[i]` holds the price for the
          i-th cap/floor.
        strikes: A real `Tensor` of shape [num_capfloors, num_optionlets], where the
          second axis corresponds to the strikes of the caplets or floorlets
          contained within each option; e.g. `strikes[i, j]` holds the strike price
          for the j-th caplet/floorlet of the i-th cap/floor.
        expiries: A real `Tensor` of shape [num_capfloors, num_optionlets], where
          `expiries[i, j]` holds the reset time for the j-th caplet/floorlet of the
          i-th cap/floor.
        maturities: A real `Tensor` of shape [num_capfloors, num_optionlets], where
          `maturities[i, j]` holds the maturity time (aka the end of accrual) of the
          underlying forward rate for the j-th caplet/floorlet of the i-th
          cap/floor. The payment occurs on the maturity as well.
        daycount_fractions: A real `Tensor` of shape [num_capfloors,
          num_optionlets], where `daycount_fractions[i, j]` holds the daycount
          fractions associated with the underlying forward rate of the j-th
          caplet/floorlet of the i-th cap/floor.
        reference_rate_fn: A Python callable that accepts expiry time as a real
          `Tensor` and returns a `Tensor` of either shape `input_shape` or
          `input_shape`. Returns the continuously compounded zero rate at
          the present time for the input expiry time.
        mean_reversion: A real positive scalar `Tensor` or a Python callable. The
          callable can be one of the following:
          (a) A left-continuous piecewise constant object (e.g.,
          `tff.math.piecewise.PiecewiseConstantFunc`) that has a property
          `is_piecewise_constant` set to `True`. In this case the object should
          have a method `jump_locations(self)` that returns a `Tensor` of shape
          `[num_jumps]`. The return value of `mean_reversion(t)` should return a
          `Tensor` of shape `t.shape`, `t` is a rank 1 `Tensor` of the same `dtype`
          as the output. See example in the class docstring.
          (b) A callable that accepts scalars (stands for time `t`) and returns a
          scalar `Tensor` of the same `dtype` as `strikes`.
          Corresponds to the mean reversion rate.
        volatility: A real positive `Tensor` of the same `dtype` as
          `mean_reversion` or a callable with the same specs as above.
          Corresponds to the long run price variance.
        notional: A real `Tensor` broadcast to [num_capfloors], such that
          `notional[i]` is the notional amount for the i-th cap/floor.
  
        is_cap: A boolean tensor broadcastable to [num_capfloors], such that
          `is_cap[i]` represents whether or not the i-th instrument is a cap (True)
          or floor (False).
        use_analytic_pricing: A Python boolean specifying if cap/floor pricing is
          done analytically during calibration. Analytic valuation is only supported
          for constant `mean_reversion` and piecewise constant `volatility`. If the
          input is `False`, then valuation using Monte-Carlo simulations is
          performed.
          Default value: The default value is `True`.
        num_samples: Positive scalar `int32` `Tensor`. The number of simulation
          paths during Monte-Carlo valuation of cap/floors. This input is ignored
          during analytic valuation.
          Default value: The default value is 1.
        random_type: Enum value of `RandomType`. The type of (quasi)-random number
          generator to use to generate the simulation paths. This input is relevant
          only for Monte-Carlo valuation and ignored during analytic valuation.
          Default value: `None` which maps to the standard pseudo-random numbers.
        seed: Seed for the random number generator. The seed is only relevant if
          `random_type` is one of `[STATELESS, PSEUDO, HALTON_RANDOMIZED,
          PSEUDO_ANTITHETIC, STATELESS_ANTITHETIC]`. For `PSEUDO`,
          `PSEUDO_ANTITHETIC` and `HALTON_RANDOMIZED` the seed should be an Python
          integer. For `STATELESS` and  `STATELESS_ANTITHETIC `must be supplied as
          an integer `Tensor` of shape `[2]`. This input is relevant only for
          Monte-Carlo valuation and ignored during analytic valuation.
          Default value: `None` which means no seed is set.
        skip: `int32` 0-d `Tensor`. The number of initial points of the Sobol or
          Halton sequence to skip. Used only when `random_type` is 'SOBOL',
          'HALTON', or 'HALTON_RANDOMIZED', otherwise ignored.
          Default value: `0`.
        time_step: Scalar real `Tensor`. Maximal distance between time grid points
          in Euler scheme. Relevant when Euler scheme is used for simulation. This
          input is ignored during analytic valuation.
          Default value: `None`.
        optimizer_fn: Optional Python callable which implements the algorithm used
          to minimize the objective function during calibration. It should have
          the following interface:  result =
            optimizer_fn(value_and_gradients_function, initial_position, tolerance,
            max_iterations)  `value_and_gradients_function` is a Python callable
            that accepts a point as a real `Tensor` and returns a tuple of `Tensor`s
            of real dtype containing the value of the function and its gradient at
            that point. 'initial_position' is a real `Tensor` containing the
            starting point of the optimization, 'tolerance' is a real scalar
            `Tensor` for stopping tolerance for the procedure and `max_iterations`
            specifies the maximum number of iterations.
          `optimizer_fn` should return a namedtuple containing the items: `position`
            (a tensor containing the optimal value), `converged` (a boolean
            indicating whether the optimize converged according the specified
            criteria), `failed` (a boolean indicating if the optimization resulted
            in a failure), `num_iterations` (the number of iterations used), and
            `objective_value` ( the value of the objective function at the optimal
            value). The default value for `optimizer_fn` is None and conjugate
            gradient algorithm is used.
        mean_reversion_lower_bound: An optional scalar `Tensor` specifying the lower
          limit of mean reversion rate during calibration.
          Default value: 0.001.
        mean_reversion_upper_bound: An optional scalar `Tensor` specifying the upper
          limit of mean reversion rate during calibration.
          Default value: 0.5.
        volatility_lower_bound: An optional scalar `Tensor` specifying the lower
          limit of Hull White volatility during calibration.
          Default value: 0.00001 (0.1 basis points).
        volatility_upper_bound: An optional scalar `Tensor` specifying the upper
          limit of Hull White volatility during calibration.
          Default value: 0.1.
        tolerance: Scalar `Tensor` of real dtype. The absolute tolerance for
          terminating the iterations.
          Default value: 1e-6.
        maximum_iterations: Scalar positive int32 `Tensor`. The maximum number of
          iterations during the optimization.
          Default value: 50.
        dtype: The default dtype to use when converting values to `Tensor`s.
          Default value: `None` which means that default dtypes inferred from
          `prices` is used.
        name: Python string. The name to give to the ops created by this function.
          Default value: `None` which maps to the default name
            `hw_capfloor_calibration`.
  
      Returns:
        A Tuple of three elements:
        * The first element is an instance of `CalibrationResult` whose parameters
          are calibrated to the input cap/floor prices.
        * A `Tensor` of optimization status for each batch element (whether the
          optimization algorithm succeeded in finding the optimal point based on
          the specified convergance criteria).
        * A `Tensor` containing the number of iterations performed by the
          optimization algorithm.
      """
      name = name or "hw_capfloor_calibration"
      with tf.name_scope(name):
          prices = tf.convert_to_tensor(
              prices, dtype=dtype, name="prices"
          )
          dtype = dtype or prices.dtype
  
          strikes = tf.convert_to_tensor(
              strikes, dtype=dtype, name="strikes"
          )
          expiries = tf.convert_to_tensor(
              expiries, dtype=dtype, name="expiries"
          )
          maturities = tf.convert_to_tensor(
              maturities, dtype=dtype, name="maturities"
          )
          daycount_fractions = tf.convert_to_tensor(
              daycount_fractions, dtype=dtype, name="daycount_fractions"
          )
          notional = tf.convert_to_tensor(
              notional, dtype=dtype, name="notional"
          )
          is_cap = tf.convert_to_tensor(
              is_cap, name="is_cap", dtype=tf.bool
          )
  
          if not hasattr(mean_reversion, "is_piecewise_constant"):
              mean_reversion = piecewise.PiecewiseConstantFunc(
                  jump_locations=[], values=mean_reversion, dtype=dtype
              )
          if not hasattr(volatility, "is_piecewise_constant"):
              volatility = piecewise.PiecewiseConstantFunc(
                  jump_locations=[], values=volatility, dtype=dtype
              )
  
          if optimizer_fn is None:
              optimizer_fn = optimizer.conjugate_gradient_minimize
  
          target_values = prices
          target_lb = tf.constant(0.0, dtype=dtype)
          target_ub = tf.math.reduce_max(target_values)
  
          vol_lb = tf.convert_to_tensor(
              volatility_lower_bound, dtype=dtype
          )
          vol_ub = tf.convert_to_tensor(
              volatility_upper_bound, dtype=dtype
          )
          mr_lb = tf.convert_to_tensor(
              mean_reversion_lower_bound, dtype=dtype
          )
          mr_ub = tf.convert_to_tensor(
              mean_reversion_upper_bound, dtype=dtype
          )
  
          initial_guess = tf.concat(
              [
                  _to_unconstrained(
                      mean_reversion.values(), mr_lb, mr_ub
                  ),
                  _to_unconstrained(
                      volatility.values(), vol_lb, vol_ub
                  ),
              ],
              axis=0,
          )
          num_mean_reversion = mean_reversion.values().shape.as_list()[
              0
          ]
          scaled_target = _scale(target_values, target_lb, target_ub)
  
          @make_val_and_grad_fn
          def loss_function(x):
              """Loss function for the optimization."""
              x_mr = _to_constrained(
                  x[:num_mean_reversion], mr_lb, mr_ub
              )
              x_vol = _to_constrained(
                  x[num_mean_reversion:], vol_lb, vol_ub
              )
  
              mean_reversion_param = piecewise.PiecewiseConstantFunc(
                  jump_locations=[], values=x_mr, dtype=dtype
              )
              volatility_param = piecewise.PiecewiseConstantFunc(
                  jump_locations=volatility.jump_locations(),
                  values=x_vol,
                  dtype=dtype,
              )
  
              model_values = cap_floor.cap_floor_price(
                  strikes=strikes,
                  expiries=expiries,
                  maturities=maturities,
                  daycount_fractions=daycount_fractions,
                  reference_rate_fn=reference_rate_fn,
                  mean_reversion=mean_reversion_param,
                  volatility=volatility_param,
                  notional=notional,
                  is_cap=is_cap,
                  use_analytic_pricing=use_analytic_pricing,
                  num_samples=num_samples,
                  random_type=random_type,
                  seed=seed,
                  skip=skip,
                  time_step=time_step,
                  dtype=dtype,
              )
  
              return tf.math.reduce_mean(
                  (
                      _scale(model_values, target_lb, target_ub)
                      - scaled_target
                  )
                  ** 2
              )
  
          optimization_result = optimizer_fn(
              loss_function,
              initial_position=initial_guess,
              tolerance=tolerance,
              max_iterations=maximum_iterations,
          )
          calibrated_parameters = optimization_result.position
          mean_reversion_calibrated = piecewise.PiecewiseConstantFunc(
              jump_locations=[],
              values=_to_constrained(
                  calibrated_parameters[:num_mean_reversion],
                  mr_lb,
                  mr_ub,
              ),
              dtype=dtype,
          )
          volatility_calibrated = piecewise.PiecewiseConstantFunc(
              jump_locations=volatility.jump_locations(),
              values=_to_constrained(
                  calibrated_parameters[num_mean_reversion:],
                  vol_lb,
                  vol_ub,
              ),
              dtype=dtype,
          )
  
          calibration_result = CalibrationResult(
              mean_reversion=mean_reversion_calibrated,
              volatility=volatility_calibrated,
          )
  
          return (
              calibration_result,
              optimization_result.converged,
              optimization_result.num_iterations,
          )
  
file_path: tf-quant-finance/tf_quant_finance/models/hull_white/calibration.py
imports: from typing import Callable, Tuple, Union
  import tensorflow.compat.v2
  from tf_quant_finance import types
  from tf_quant_finance import utils
  from tf_quant_finance.black_scholes import implied_vol
  from tf_quant_finance.black_scholes.implied_vol_utils import UnderlyingDistribution
  from tf_quant_finance.math import make_val_and_grad_fn
  from tf_quant_finance.math import optimizer
  from tf_quant_finance.math import piecewise
  from tf_quant_finance.math import random
  from tf_quant_finance.models.hull_white import cap_floor
  from tf_quant_finance.models.hull_white import swaption
  from tf_quant_finance.rates.analytics import swap
label_desc: Use image node_id 2 for calling a global function with example usage: calibration_from_cap_floors() and returns: calibration_result,  optimization_result,  optimization_result, tf