code: def test_given_non_trivial_graph_with_nonlinear_relationships_when_attribute_anomaly_scores_with_it_score_then_returns_qualitatively_correct_results():
      num_training_samples = 10000
  
      # Defining ground truth model to avoid SCM learning issues and, hence, to focus on the anomaly attribution
      # algorithm.
      class MyNonLinearModel(PredictionModel):
          def fit(self, X: np.ndarray, Y: np.ndarray) -> None:
              pass
  
          def predict(self, X: np.ndarray) -> np.ndarray:
              return (np.sum(X, axis=1) ** 2).reshape(-1, 1)
  
          def clone(self):
              pass
  
      X0 = np.random.normal(0, 1, num_training_samples)
      X1 = MyNonLinearModel().predict(
          X0.reshape(-1, 1)
      ) + np.random.normal(
          0, 1, (num_training_samples, 1)
      )  # X0**2 + N
      X2 = np.random.uniform(-5, 5, num_training_samples)
      X3 = MyNonLinearModel().predict(
          np.column_stack([X1, X2])
      ) + np.random.normal(
          0, 1, (num_training_samples, 1)
      )  # (X1 + X2)**2 + N
  
      training_data = pd.DataFrame(
          {
              "X0": X0,
              "X1": X1.reshape(-1),
              "X2": X2,
              "X3": X3.reshape(-1),
          }
      )
      causal_model = InvertibleStructuralCausalModel(
          nx.DiGraph([("X0", "X1"), ("X1", "X3"), ("X2", "X3")])
      )
      causal_model.set_causal_mechanism(
          "X1", AdditiveNoiseModel(MyNonLinearModel())
      )
      causal_model.set_causal_mechanism(
          "X3", AdditiveNoiseModel(MyNonLinearModel())
      )
      auto.assign_causal_mechanisms(
          causal_model,
          training_data,
          auto.AssignmentQuality.GOOD,
          override_models=False,
      )
  
      fit(causal_model, training_data)
  
      anomaly_data = pd.DataFrame(
          {
              "X0": np.array([10, 10, 0]),
              "X1": np.array([100, 110, 0]),
              "X2": np.array([0, 0, 10]),
              "X3": np.array([10000, 12100, 110]),
          }
      )
  
      scores = attribute_anomalies(
          causal_model,
          "X3",
          anomaly_data,
          anomaly_scorer=MedianCDFQuantileScorer(),
          num_distribution_samples=num_training_samples,
          attribute_mean_deviation=False,
      )
  
      def total_anomaly_score(training_data, anomaly_sample):
          distribution_samples = training_data["X3"].to_numpy()
          anomaly_scorer = MedianCDFQuantileScorer()
          anomaly_scorer.fit(distribution_samples)
          return -np.log(
              _relative_frequency(
                  anomaly_scorer.score(distribution_samples)
                  >= anomaly_scorer.score(anomaly_sample)
              )
          )
  
      # 1. X0 is the root cause (+ 10 to the noise)
      # 2. X0 and X1 are the root causes (+ 10 to both noise)
      # 3. X2 and X3 are both root causes (+ 10 to both noises)
      assert scores["X0"][0] > 8  # Anomalous, but the only root cause.
      assert scores["X0"][1] > 4  # Anomalous, but together with X1.
      assert scores["X0"][2] < 1  # Not anomalous.
  
      assert scores["X1"][0] < 1  # Not anomalous given upstream nodes.
      assert scores["X1"][1] > 1  # Anomalous, but together with X0.
      assert scores["X1"][2] < 1  # Not anomalous given upstream nodes.
  
      assert scores["X2"][0] < 1  # Not anomalous.
      assert scores["X2"][1] < 1  # Not anomalous.
      assert scores["X2"][2] > 1  # Anomalous, but together with X3.
  
      assert scores["X3"][0] < 1  # Not anomalous given upstream nodes.
      assert scores["X3"][1] < 1  # Not anomalous given upstream nodes.
      assert scores["X3"][2] > 1  # Anomalous, but together with X1.
  
      # The sum of the scores should add up to the anomaly score of the target (here, X3).
      assert scores["X0"][0] + scores["X1"][0] + scores["X2"][
          0
      ] + scores["X3"][0] == approx(
          total_anomaly_score(
              training_data, anomaly_data["X3"].to_numpy()[0]
          ),
          abs=0.4,
      )
      assert scores["X0"][1] + scores["X1"][1] + scores["X2"][
          1
      ] + scores["X3"][1] == approx(
          total_anomaly_score(
              training_data, anomaly_data["X3"].to_numpy()[1]
          ),
          abs=0.4,
      )
      assert scores["X0"][2] + scores["X1"][2] + scores["X2"][
          2
      ] + scores["X3"][2] == approx(
          total_anomaly_score(
              training_data, anomaly_data["X3"].to_numpy()[2]
          ),
          abs=0.6,
      )
  
file_path: dowhy/tests/gcm/test_anomaly_attribution.py
imports: import networkx
  import numpy
  import pandas
  from flaky import flaky
  from pytest import approx
  from dowhy.gcm import AdditiveNoiseModel, InverseDensityScorer, InvertibleStructuralCausalModel, MedianCDFQuantileScorer, attribute_anomalies, auto, fit
  from dowhy.gcm.anomaly import _relative_frequency, attribute_anomaly_scores
  from dowhy.gcm.density_estimators import GaussianMixtureDensityEstimator
  from dowhy.gcm.ml import PredictionModel
label_desc: Use image node_id 5 for calling a global function with example usage: test_given_non_trivial_graph_with_nonlinear_relationships_when_attribute_anomaly_scores_with_it_score_then_returns_qualitatively_correct_results() and returns: unknown, unknown