docstr: Tests path properties for 1-dimentional Ito process.
  
  We construct the following Ito process.
  
  ````
  dX = mu * sqrt(t) * dt + (a * t + b) dW
  ````
  
  For this process expected value at time t is x_0 + 2/3 * mu * t^1.5 .
  Args:
    use_batch: Test parameter to specify if we are testing the batch of Euler
      sampling.
    watch_params: Triggers custom for loop.
    supply_normal_draws: Supply normal draws.
    random_type: `RandomType` of the sampled normal draws.
docstr_tok: ["Tests","path","properties","for","1-dimentional","Ito","process",".","We","construct","the","following","Ito","process",".","``","``","dX","=","mu","*","sqrt","(","t",")","*","dt","+","(","a","*","t","+","b",")","dW","``","``","For","this","process","expected","value","at","time","t","is","x_0","+","2\/3","*","mu","*","t^1.5",".","Args",":","use_batch",":","Test","parameter","to","specify","if","we","are","testing","the","batch","of","Euler","sampling",".","watch_params",":","Triggers","custom","for","loop",".","supply_normal_draws",":","Supply","normal","draws",".","random_type",":","`","RandomType","`","of","the","sampled","normal","draws","."]
code: def test_sample_paths_1d(
      self, use_batch, watch_params, supply_normal_draws, random_type
  ):
      """Tests path properties for 1-dimentional Ito process.
  
      We construct the following Ito process.
  
      ````
      dX = mu * sqrt(t) * dt + (a * t + b) dW
      ````
  
      For this process expected value at time t is x_0 + 2/3 * mu * t^1.5 .
      Args:
        use_batch: Test parameter to specify if we are testing the batch of Euler
          sampling.
        watch_params: Triggers custom for loop.
        supply_normal_draws: Supply normal draws.
        random_type: `RandomType` of the sampled normal draws.
      """
      dtype = tf.float64
      mu = 0.2
      a = 0.4
      b = 0.33
  
      def drift_fn(t, x):
          drift = mu * tf.sqrt(t) * tf.ones_like(x, dtype=t.dtype)
          return drift
  
      def vol_fn(t, x):
          del x
          if not use_batch:
              return (a * t + b) * tf.ones([1, 1], dtype=t.dtype)
          else:
              return (a * t + b) * tf.ones([2, 1, 1, 1], dtype=t.dtype)
  
      times = np.array([0.0, 0.1, 0.21, 0.32, 0.43, 0.55])
      num_samples = 10000
  
      if supply_normal_draws:
          # Use antithetic sampling
          normal_draws = tf.random.stateless_normal(
              shape=[2, 5000, 55, 1], seed=[1, 42], dtype=dtype
          )
          normal_draws = tf.concat(
              [normal_draws, -normal_draws], axis=1
          )
      else:
          normal_draws = None
  
      if use_batch:
          # x0.shape = [2, 1, 1]
          x0 = np.array([[[0.1]], [[0.1]]])
      else:
          x0 = np.array([0.1])
      paths = self.evaluate(
          euler_sampling.sample(
              dim=1,
              drift_fn=drift_fn,
              volatility_fn=vol_fn,
              times=times,
              num_samples=num_samples,
              initial_state=x0,
              random_type=random_type,
              normal_draws=normal_draws,
              watch_params=watch_params,
              time_step=0.01,
              seed=[1, 42],
              dtype=dtype,
          )
      )
      paths_no_zero = self.evaluate(
          euler_sampling.sample(
              dim=1,
              drift_fn=drift_fn,
              volatility_fn=vol_fn,
              times=times[1:],
              num_samples=num_samples,
              initial_state=x0,
              random_type=random_type,
              normal_draws=normal_draws,
              time_step=0.01,
              seed=[1, 42],
              dtype=dtype,
          )
      )
  
      with self.subTest("CorrectShape"):
          if not use_batch:
              self.assertAllClose(
                  paths.shape, (num_samples, 6, 1), atol=0
              )
          else:
              self.assertAllClose(
                  paths.shape, (2, num_samples, 6, 1), atol=0
              )
      if not use_batch:
          means = np.mean(paths, axis=0).reshape(-1)
      else:
          means = np.mean(paths, axis=1).reshape([2, 1, 6])
      expected_means = x0 + (2.0 / 3.0) * mu * np.power(times, 1.5)
      with self.subTest("ExpectedResult"):
          self.assertAllClose(
              means, expected_means, rtol=1e-2, atol=1e-2
          )
      if not use_batch:
          with self.subTest("IncludeInitialState"):
              self.assertAllClose(paths[:, 1:, :], paths_no_zero)
  
file_path: tf-quant-finance/tf_quant_finance/models/euler_sampling_test.py
imports: 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 3 for calling the EulerSamplingTest obj's underlying member method code with example usage: obj.test_sample_paths_1d(use_batch, watch_params, supply_normal_draws, random_type) and returns: drift, unknown, unknown