docstr: Monte Carlo and PDE pricing of a basket option.
docstr_tok: ["Monte","Carlo","and","PDE","pricing","of","a","basket","option","."]
code: def test_backward_pde_vs_mc_coeffs_with_batching_2d(self):
      """Monte Carlo and PDE pricing of a basket option."""
      num_grid_points = 256  # Number of grid points
      dtype = tf.float64
  
      # Process dimensionality
      dim = 2
  
      # Build a uniform grid
      s_min = 0
      s_max = 200
      grid = tff.math.pde.grids.uniform_grid(
          minimums=dim * [s_min],
          maximums=dim * [s_max],
          sizes=dim * [num_grid_points],
          dtype=dtype,
      )
      # Specify volatilities, interest rates and strikes for the options
      strikes = tf.constant([[50], [100], [90]], dtype)
      volatilities = tf.constant(
          [[0.3, 0.1], [0.15, 0.1], [0.1, 0.1]], dtype
      )
      rates = tf.constant([[0.01], [0.03], [0.01]], dtype)
      expiries = 1.0
  
      # Batch size of the process
      batch_size = 3
  
      def drift_fn(t, x):
          del t
          # `x` is expected to be of shape [batch_size] + sample_shape + [dim]
          # We need to expand rank of rates to [batch_size] + extra_rank * [1] + [1]
          expand_rank = x.shape.rank - 2
          rates_expand = tf.reshape(
              rates, [batch_size] + (expand_rank) * [1] + [1]
          )
          # Output is of shape [batch_size] + sample_shape + [dim]
          return rates_expand * x
  
      def vol_fn(t, x):
          del t
          # `x` is expected to be of shape [batch_size] + sample_shape + [dim]
          # As before, need to expand rank of volatilities to
          # `[batch_size] + extra_rank * [1] + [1]`
          expand_rank = x.shape.rank - 2
          volatilities_expand = tf.reshape(
              volatilities, [batch_size] + (expand_rank) * [1] + [dim]
          )
          # Output is of shape [batch_size] + sample_shape + [dim, dim]
          return tf.expand_dims(
              volatilities_expand * x, axis=-1
          ) * tf.eye(
              dim, batch_shape=x.shape.as_list()[:-1], dtype=x.dtype
          )
  
      process = tff.models.GenericItoProcess(
          dim=dim, drift_fn=drift_fn, volatility_fn=vol_fn, dtype=dtype
      )
  
      # Define discounting function
      def discounting(t, x):
          del t, x
          # `x` is of shape [batch_size, d1, d2]. Broadcast rates to shape
          # [batch_size, 1, 1]
          return tf.expand_dims(rates, axis=-1)
  
      x, y = tf.meshgrid(*grid, indexing="ij")
      # Basket option payoff
      # Shape [batch_size, num_grid_points, num_grid_points]
      final_value_grid = tf.broadcast_to(
          tf.nn.relu((x + y) / 2 - tf.reshape(strikes, [-1, 1, 1])),
          [batch_size] + dim * [num_grid_points],
      )
  
      @tff.math.pde.boundary_conditions.dirichlet
      def upper_boundary_fn(t, coord_grid):
          del coord_grid
          return (s_max + grid[0]) / 2 - strikes * tf.exp(
              -rates * (expiries - t)
          )
  
      # Compute basket option prices with a PDE
      # Shape (batch_size, num_grid_points, num_grid_points)
      pde_prices = process.fd_solver_backward(
          start_time=expiries,
          end_time=0,
          time_step=0.1,
          coord_grid=grid,
          values_grid=final_value_grid,
          discounting=discounting,
          boundary_condtions=[
              (None, upper_boundary_fn),
              (None, upper_boundary_fn),
          ],
      )[0]
      with self.subTest("PDEOutputShape"):
          self.assertAllEqual(
              pde_prices.shape.as_list(),
              [batch_size, num_grid_points, num_grid_points],
          )
      # Get prices for some spots
      loc_1 = 95
      loc_2 = 123
      loc_3 = 155
      spots = tf.stack(
          [[grid[0][loc_1]], [grid[0][loc_2]], [grid[0][loc_3]]], axis=0
      )
      inds = tf.stack([loc_1, loc_2, loc_3])
      inds = tf.stack([[0, 1, 2], inds, inds], axis=1)
      # Shape [batch_size]
      pde_prices = tf.gather_nd(pde_prices, inds)
  
      # Monte Carlo prices
      x0 = tf.expand_dims(spots, axis=-2) + np.zeros(
          [batch_size, 1, dim]
      )
  
      times_grid = None
      time_step = 0.01
  
      num_samples = 10000
      normal_draws = None
      paths = process.sample_paths(
          [expiries],
          num_samples=num_samples,
          initial_state=x0,
          time_step=time_step,
          times_grid=times_grid,
          normal_draws=normal_draws,
          random_type=tff.math.random.RandomType.SOBOL,
          seed=[1, 42],
      )
      mc_prices = tf.reduce_mean(
          tf.nn.relu(
              (paths[..., 0] + paths[..., 1]) / 2
              - tf.reshape(strikes, [-1, 1, 1])
          ),
          -2,
      )
      # Discounted prices
      # Shape [batch_size, 1]
      mc_prices *= tf.exp(-rates * expiries)
  
      with self.subTest("Prices"):
          self.assertAllClose(
              mc_prices[..., 0], pde_prices, rtol=0.1, atol=0.1
          )
  
imports: import os
  from unittest import mock
  from absl.testing import parameterized
  import numpy
  import tensorflow.compat.v2
  import tf_quant_finance
  from tensorflow.python.framework import test_util
label_desc: Use image node_id 7 for calling the GenericItoProcessTest obj's underlying member method code with example usage: obj.test_backward_pde_vs_mc_coeffs_with_batching_2d() and returns: unknown, unknown, tf, unknown