Metrics Module

This module contains all the network statistics (metrics) that can be used with ERGM models.

MetricsCollection

A collection of metrics that handles the calculation of network statistics.

MetricsCollection

A collection of metrics for ERGM models.

This class manages multiple metrics, handles feature calculations across samples, prepares data for MPLE optimization, and automatically detects and removes collinear features.

Parameters:

Name	Type	Description	Default
`metrics`	`Sequence[Metric]`	Sequence of Metric instances to include in the model.	required
`is_directed`	`bool`	Whether the network is directed.	required
`n_nodes`	`int`	Number of nodes in the network.	required
`fix_collinearity`	`bool`	If True, automatically detect and remove collinear features. Default is True.	`True`
`collinearity_fixer_sample_size`	`int`	Number of random networks to sample for collinearity detection. Default is 1000.	`1000`
`is_collinearity_distributed`	`bool`	If True, distribute collinearity fixing computation. Default is False.	`False`
`mask`	`ndarray`	Boolean mask indicating which edges to consider (1D flattened). Default is None.	`None`
`**kwargs`		Additional keyword arguments for collinearity fixer configuration.	`{}`

Source code in pyERGM/metrics.py

class MetricsCollection:
    """
    A collection of metrics for ERGM models.

    This class manages multiple metrics, handles feature calculations across samples,
    prepares data for MPLE optimization, and automatically detects and removes
    collinear features.

    Parameters
    ----------
    metrics : Sequence[Metric]
        Sequence of Metric instances to include in the model.
    is_directed : bool
        Whether the network is directed.
    n_nodes : int
        Number of nodes in the network.
    fix_collinearity : bool, optional
        If True, automatically detect and remove collinear features. Default is True.
    collinearity_fixer_sample_size : int, optional
        Number of random networks to sample for collinearity detection. Default is 1000.
    is_collinearity_distributed : bool, optional
        If True, distribute collinearity fixing computation. Default is False.
    mask : np.ndarray, optional
        Boolean mask indicating which edges to consider (1D flattened). Default is None.
    **kwargs
        Additional keyword arguments for collinearity fixer configuration.
    """

    @staticmethod
    def _copy_metrics(metrics: Sequence[Metric]) -> tuple:
        """
        Initialize metrics by creating deep copies to avoid modifying the originals.

        Parameters
        ----------
        metrics : Sequence[Metric]
            The metrics to initialize.

        Returns
        -------
        tuple
            A tuple of copied metrics.
        """
        return tuple([deepcopy(metric) for metric in metrics])

    def __init__(
            self,
            metrics: Sequence[Metric],
            is_directed: bool,
            n_nodes: int,
            fix_collinearity=True,
            collinearity_fixer_sample_size=1000,
            is_collinearity_distributed=False,
            mask: npt.NDArray[bool] | None = None,
            **kwargs
    ):
        logger.debug("Initializing MetricsCollection")
        self.metrics = MetricsCollection._copy_metrics(metrics)

        for m in self.metrics:
            m._n_nodes = n_nodes
            m.initialize_indices_to_ignore()

        metric_names = [(type(x).__name__, x._name) for x in self.metrics if x._name is not None]

        if len(set(metric_names)) < len(metric_names):
            raise ValueError(f"Got at least 2 metrics with the same name - {metric_names}")

        self._has_dyadic_dependent_metrics = any([not x._is_dyadic_independent for x in self.metrics])

        self.is_directed = is_directed
        self.mask = None
        if mask is not None:
            if mask.ndim != 1:
                raise ValueError(f"MetricsCollection expects flattened masks (1D). Got {mask.ndim}D")
            self.mask = mask

        if OptimizationScheme.MPLE != self.choose_optimization_scheme() and self.mask is not None:
            raise NotImplementedError("Masking is currently supported only for edge independent models.")

        for x in self.metrics:
            if (x._is_directed is not None) and (x._is_directed != self.is_directed):
                model_is_directed_str = "a directed" if self.is_directed else "an undirected"
                metric_is_directed_str = "a directed" if x._is_directed else "an undirected"
                raise ValueError(f"Trying to initialize {model_is_directed_str} model with {metric_is_directed_str} "
                                 f"metric `{str(x)}`!")
            if self.mask is not None and not x.does_support_mask:
                raise ValueError(f"Trying to initialize a masked model with metric `{str(x)}` which does not support "
                                 f"masks!")
        self.n_nodes = n_nodes

        if is_collinearity_distributed and self._has_dyadic_dependent_metrics:
            raise ValueError(
                "Distributed optimization is only supported for dyadic-independent models. "
                "This model contains dyadic-dependent metrics: "
                f"{[str(m) for m in self.metrics if not m._is_dyadic_independent]}"
            )

        # Returns the number of features that are being calculated. Since a single metric might return more than one
        # feature, the length of the statistics vector might be larger than the amount of metrics. Since it also depends
        # on the network size, n is a mandatory parameters. That's why we're using the get_effective_feature_count
        # function
        self.num_of_features = self.calc_num_of_features()
        self.features_per_metric = np.array([metric.get_effective_feature_count() for metric in self.metrics])

        if fix_collinearity:
            logger.debug("Starting collinearity fix")
            self.collinearity_fixer(sample_size=collinearity_fixer_sample_size,
                                    is_distributed=is_collinearity_distributed,
                                    num_samples_per_job=kwargs.get('num_samples_per_job_collinearity_fixer', 5),
                                    ratio_threshold=kwargs.get('ratio_threshold_collinearity_fixer', 5e-6),
                                    nonzero_thr=kwargs.get('nonzero_threshold_collinearity_fixer', 0.1),
                                    )

        self.num_of_metrics = len(self.metrics)
        self.metric_names = tuple([str(metric) for metric in self.metrics])

    def _delete_metric(self, metric: Metric):
        self.metrics = tuple([m for m in self.metrics if m != metric])

    def calc_num_of_features(self):
        return sum([metric.get_effective_feature_count() for metric in self.metrics])

    def get_metric(self, metric_name: str) -> Metric:
        """
        Get a metric instance
        """
        return self.metrics[self.metric_names.index(metric_name)]

    def get_metric_by_feat_idx(self, idx: int) -> Metric:
        cum_sum_num_feats = 0
        for m in self.metrics:
            cum_sum_num_feats += m.get_effective_feature_count()
            if cum_sum_num_feats > idx:
                return m
        raise ValueError(
            f"Feature index {idx} out of bounds. "
            f"Valid range: [0, {cum_sum_num_feats - 1}]"
        )

    def get_feature_idx_within_metric(self, idx: int):
        cum_sum_num_feats = 0
        for m_idx in range(len(self.metrics)):
            next_met_num_feats = self.metrics[m_idx].get_effective_feature_count()
            if cum_sum_num_feats + next_met_num_feats > idx:
                # We want to return the index in the "full" array, namely regardless of ignored features. So, in case
                # there are indices that are ignored, we must take the returned index from the array of non-ignored
                # indices, to compensate for the ones ignored by the indexing of MetricCollection (which holds only
                # effective features, after ignoring).
                # For example - if we ignore the degree of the first node, and now want to ignore the degree of the
                # fifth node as well, we must return 4 for the metric to correctly ignore it. But it is the fourth
                # feature corresponding to the metric in the MetricCollection vector of features (not fifth, because the
                # first is missing, not returned by the Metric as it's ignored). By returning the fourth element from
                # the list of non-ignored indices, which is [1, 2, ..., n], we solve the problem.

                effective_idx_within_metric = idx - cum_sum_num_feats
                if hasattr(self.metrics[m_idx], "_indices_to_ignore"):
                    non_ignored_indices = np.where(~self.metrics[m_idx]._indices_to_ignore)[0]
                    # Return the index of the feature with relation to the whole set of features (without ignoring).
                    return non_ignored_indices[effective_idx_within_metric]
                else:
                    return effective_idx_within_metric
            else:
                cum_sum_num_feats += next_met_num_feats

    def _metric_feature_iter(self):
        """Iterate over metrics with feature index tracking.

        Yields:
            (metric, start_idx, n_features) for each metric in the collection

        This helper eliminates duplication in the common pattern:
            feature_idx = 0
            for metric in self.metrics:
                n_features = metric._get_effective_feature_count()
                ...process metric...
                feature_idx += n_features
        """
        feature_idx = 0
        for metric in self.metrics:
            n_features = metric.get_effective_feature_count()
            yield metric, feature_idx, n_features
            feature_idx += n_features

    def calc_statistics_for_binomial_tensor_local(self, tensor_size, p=0.5):
        sample = generate_binomial_tensor(self.n_nodes, tensor_size, p=p)

        # Symmetrize samples if not directed
        if not self.is_directed:
            sample = np.round((sample + sample.transpose(1, 0, 2)) / 2)

        # Make sure the main diagonal is 0
        sample[np.arange(self.n_nodes, dtype=int), np.arange(self.n_nodes, dtype=int), :] = 0

        # Calculate the features of the sample
        return self.calculate_sample_statistics(sample)

    def calc_statistics_for_binomial_tensor_distributed(self, tensor_size, p=0.5, num_samples_per_job=1):
        # Note - if tensor_size % num_samples_per_job != 0 the sample size will be larger than tensor_size specified by
        # the user (it is (tensor_size // num_samples_per_job + 1) * num_samples_per_job)
        out_dir_path = (Path.cwd().parent / "OptimizationIntermediateCalculations").resolve()
        data_path = (out_dir_path / "data").resolve()
        os.makedirs(data_path, exist_ok=True)

        with open((data_path / 'metric_collection.pkl').resolve(), 'wb') as f:
            pickle.dump(self, f)  # type: ignore[arg-type]

        cmd_line_for_bsub = (f'python ./sample_statistics_distributed_calcs.py '
                             f'--out_dir_path {out_dir_path} '
                             f'--num_samples_per_job {num_samples_per_job} '
                             f'--p {p}')

        num_jobs = int(np.ceil(tensor_size / num_samples_per_job))
        logger.debug(f"Sending {num_jobs} children job array for collinearity fixer")
        job_array_ids, children_logs_dir = run_distributed_children_jobs(out_dir_path, cmd_line_for_bsub,
                                                                         "distributed_binomial_tensor_statistics.sh",
                                                                         num_jobs, "sample_stats")

        # Wait for all jobs to finish.
        sample_stats_path = (out_dir_path / "sample_statistics").resolve()
        os.makedirs(sample_stats_path, exist_ok=True)
        wait_for_distributed_children_outputs(num_jobs, [sample_stats_path], job_array_ids, "sample_stats",
                                              children_logs_dir)

        # Clean current scripts and data
        shutil.rmtree(data_path)
        shutil.rmtree((out_dir_path / "scripts").resolve())

        # Aggregate results
        whole_sample_statistics = cat_children_jobs_outputs(num_jobs, sample_stats_path, axis=1)

        # clean outputs
        shutil.rmtree(sample_stats_path)
        return whole_sample_statistics

    def remove_feature_by_idx(self, idx: int):
        metric_of_feat = self.get_metric_by_feat_idx(idx)
        is_trimmable = metric_of_feat.get_effective_feature_count() > 1
        if not is_trimmable:
            logger.info(f"Removing the metric {str(metric_of_feat)} from the collection")
            self._delete_metric(metric=metric_of_feat)
        else:
            idx_to_delete_within_metric = self.get_feature_idx_within_metric(idx)
            logger.info(f"Removing the {idx_to_delete_within_metric} feature of {str(metric_of_feat)}")
            metric_of_feat.update_indices_to_ignore([idx_to_delete_within_metric])
        self.num_of_features = self.calc_num_of_features()

        # Re-calculate the number of features per metric after deleting a feature.
        self.features_per_metric = np.array([metric.get_effective_feature_count() for metric in self.metrics])

    def collinearity_fixer(self, sample_size=1000, nonzero_thr=10 ** -1, ratio_threshold=10 ** -6,
                           eigenvec_thr=10 ** -4, is_distributed=False, **kwargs):
        """
        Find collinearity between metrics in the collection.

        Currently, this is a naive version that only handles the very simple cases.
        """
        is_linearly_dependent = True

        # Sample networks from a maximum entropy distribution, for avoiding edge cases (such as a feature is 0 for
        # all networks in the sample).
        if not is_distributed:
            sample_features = self.calc_statistics_for_binomial_tensor_local(sample_size)
        else:
            logger.debug("Starting distributed collinearity fixing")
            sample_features = self.calc_statistics_for_binomial_tensor_distributed(sample_size,
                                                                                   num_samples_per_job=kwargs.get(
                                                                                       "num_samples_per_job", 5))
        while is_linearly_dependent:
            features_cov_mat = sample_features @ sample_features.T

            # Determine whether the matrix of features is invertible. If not - this means there is a non-trivial vector,
            # that when multiplied by the matrix gives the 0 vector. Namely, there is a single set of coefficients that
            # defines a non-trivial linear combination that equals 0, for *all* the sampled feature vectors. This means
            # the features are linearly dependent.
            eigen_vals, eigen_vecs = np.linalg.eigh(features_cov_mat)

            minimal_non_zero_eigen_val = np.min(np.abs(eigen_vals[np.abs(eigen_vals) > nonzero_thr]))
            small_eigen_vals_indices = np.where(np.abs(eigen_vals) / minimal_non_zero_eigen_val < ratio_threshold)[0]

            if small_eigen_vals_indices.size == 0:
                is_linearly_dependent = False
            else:
                logger.info("Collinearity detected, identifying features to remove")

                # For each linear dependency (corresponding to an eigen vector with a low value), mark the indices of
                # features that are involved (identified by a non-zero coefficient in the eigen vector).
                dependent_features_flags = np.zeros((small_eigen_vals_indices.size, self.num_of_features))
                for i in range(small_eigen_vals_indices.size):
                    dependent_features_flags[
                        i, np.where(np.abs(eigen_vecs[:, small_eigen_vals_indices[i]]) > eigenvec_thr)[0]] = 1

                # Calculate the fraction of dependencies each feature is involved in.
                fraction_of_dependencies_involved = dependent_features_flags.mean(axis=0)

                # Sort the features (their indices) by the fraction of dependencies they are involved in (remove first
                # features that are involved in more dependencies). Break ties by the original order of the features in
                # the array (for the consistency of sorting. E.g, if we need to get rid of degree features, always
                # remove them by the order of nodes).
                removal_order = np.lexsort((np.arange(self.num_of_features), -fraction_of_dependencies_involved))

                # Iterate the metrics to find one with multiple features, namely effective number of features that is
                # larger than 1 ('trimmable'). We prefer to trim metrics rather than totally eliminate them from the
                # collection.
                i = 0
                cur_metric = self.get_metric_by_feat_idx(removal_order[i])
                is_trimmable = cur_metric.get_effective_feature_count() > 1
                while (not is_trimmable and i < removal_order.size - 1 and fraction_of_dependencies_involved[
                    removal_order[i]] > 0):
                    i += 1
                    cur_metric = self.get_metric_by_feat_idx(removal_order[i])
                    is_trimmable = cur_metric.get_effective_feature_count() > 1

                # If a trimmable metric was not found (i.e., all features that are involved in the dependency are of
                # metrics with an effective number of features of 1), totally remove the metric that is involved in most
                # dependencies.
                if not is_trimmable:
                    i = 0
                self.remove_feature_by_idx(removal_order[i])

                sample_features = np.delete(sample_features, removal_order[i], axis=0)

    def calculate_statistics(self, W: np.ndarray):
        """
        Calculate the statistics of a graph, formally written as g(y).

        Parameters
        ----------
        W : np.ndarray
            An N x N connectivity matrix.

        Returns
        -------
        statistics : np.ndarray
            An array of statistics
        """
        return self.calculate_sample_statistics(W)[:, 0]

    def calc_change_scores(self, current_network: np.ndarray, indices: tuple[int, int]) -> np.ndarray:
        """
        Calculates the vector of change scores, namely g(net_2) - g(net_1)

        NOTE - this function assumes that the size current_network and self.n_nodes are the same, and doesn't validate
        it, due to runtime considerations.
        """
        change_scores = np.zeros(self.num_of_features)

        for metric, start_idx, n_features in self._metric_feature_iter():
            change_scores[start_idx:start_idx + n_features] = (
                metric.calc_change_score(current_network, indices)
            )

        return change_scores

    def calculate_sample_statistics(
            self,
            networks_sample: np.ndarray,
    ) -> np.ndarray:
        """
        Calculate the statistics over a sample of networks

        Parameters
        ----------
        networks_sample
            The networks sample - an array of n X n X sample_size
        Returns
        -------
        an array of the statistics vector per sample (num_features X sample_size)
        """
        networks_sample = expand_net_dims(networks_sample)
        if networks_sample.shape[0] != self.n_nodes or networks_sample.shape[1] != self.n_nodes:
            raise ValueError(
                f"Got a sample of networks of size {networks_sample.shape[0]}, but Metrics expect size {self.n_nodes}")

        num_of_samples = networks_sample.shape[2]
        features_of_net_samples = np.zeros((self.num_of_features, num_of_samples))

        for metric, start_idx, n_features in self._metric_feature_iter():
            calc_for_sample_kwargs = {'networks_sample': networks_sample}
            if self.mask is not None:
                calc_for_sample_kwargs |= {'mask': self.mask}
            features = metric.calculate_for_sample(**calc_for_sample_kwargs)

            features_of_net_samples[start_idx:start_idx + n_features] = features

        return features_of_net_samples

    def _get_global_mask(self) -> npt.NDArray[bool]:
        """Return the global mask for edge indices, or a mask of all True if no mask is set."""
        full_net_size = self.n_nodes * self.n_nodes - self.n_nodes
        if not self.is_directed:
            full_net_size //= 2
        return self.mask if self.mask is not None else np.ones(full_net_size, dtype=bool)

    @staticmethod
    def _validate_edge_indices_lims(
            edge_indices_lims: tuple[int, int] | None,
            global_mask: npt.NDArray[bool],
    ) -> tuple[int, int]:
        """Validate and return edge index limits, defaulting to full range if None."""
        if edge_indices_lims is None:
            return 0, global_mask.sum()
        if (
                edge_indices_lims[0] < 0 or edge_indices_lims[1] < 0 or
                edge_indices_lims[0] > global_mask.sum() or edge_indices_lims[1] > global_mask.sum() or
                edge_indices_lims[0] >= edge_indices_lims[1]
        ):
            raise ValueError(
                'edge_indices_lims out of bounds. expected strictly monotonic increasing limits between 0 '
                'and the number of considered edges (the size of the dataset for MPLE optimization), which is '
                f'{global_mask.sum()}. Got {edge_indices_lims}')
        return edge_indices_lims

    @staticmethod
    def _get_masked_indices_slice(
            edge_indices_lims: tuple[int, int],
            global_mask: npt.NDArray[bool],
    ) -> npt.NDArray[np.intp]:
        """Get the actual array indices for a slice within the masked edge space."""
        return np.where(global_mask)[0][edge_indices_lims[0]:edge_indices_lims[1]]

    def slice_flat_array_by_edge_indices(
            self,
            flat_array: npt.NDArray,
            edge_indices_lims: tuple[int, int] | None = None,
    ) -> npt.NDArray:
        """
        Slice a flattened edge array using the same indexing logic as MPLE data chunks.

        Parameters
        ----------
        flat_array : np.ndarray
            A flattened array of edge values (e.g., weights) with length matching
            the full network edge count.
        edge_indices_lims : tuple[int, int] or None
            The (start, end) indices within the masked edge space. If None, returns all masked edges.

        Returns
        -------
        np.ndarray
            The sliced portion of the array.
        """
        global_mask = self._get_global_mask()
        edge_indices_lims = self._validate_edge_indices_lims(edge_indices_lims, global_mask)
        indices = self._get_masked_indices_slice(edge_indices_lims, global_mask)
        return flat_array[indices]

    def _get_mple_data_chunk_mask(
            self,
            edge_indices_lims: tuple[int, int] | None,
    ) -> npt.NDArray[bool]:
        full_net_size = self.n_nodes * self.n_nodes - self.n_nodes
        if not self.is_directed:
            full_net_size //= 2
        data_chunk_mask = np.zeros(full_net_size, dtype=bool)
        global_mask = self._get_global_mask()
        edge_indices_lims = self._validate_edge_indices_lims(edge_indices_lims, global_mask)
        # First constrain to the "universe" of edge indices to consider by the global mask, then slice according to the
        # limits within the subset of considered edges.
        data_chunk_mask[self._get_masked_indices_slice(edge_indices_lims, global_mask)] = True
        return data_chunk_mask

    def prepare_mple_regressors(
            self,
            observed_network: np.ndarray | None = None,
            edge_indices_lims: tuple[int, int] | None = None,
    ) -> np.ndarray:
        data_chunk_mask = self._get_mple_data_chunk_mask(edge_indices_lims)
        Xs = np.zeros((data_chunk_mask.sum(), self.num_of_features))

        feature_idx = 0
        for metric in self.metrics:
            n_features_from_metric = metric.get_effective_feature_count()
            metric.calculate_mple_regressors(
                Xs_out=Xs,
                feature_col_indices=np.arange(feature_idx, feature_idx + n_features_from_metric, dtype=int),
                observed_network=observed_network,
                edge_indices_mask=data_chunk_mask,
            )
            feature_idx += n_features_from_metric
        return Xs

    def prepare_mple_labels(
            self,
            observed_networks: np.ndarray,
            edge_indices_lims: tuple[int, int] | None = None,
    ) -> np.ndarray:
        observed_networks = expand_net_dims(observed_networks)
        chunk_mask = self._get_mple_data_chunk_mask(edge_indices_lims)
        ys = np.zeros((chunk_mask.sum(), 1))
        num_nets = observed_networks.shape[-1]
        for net_idx in range(num_nets):
            net = observed_networks[..., net_idx]
            ys += flatten_square_matrix_to_edge_list(net, self.is_directed)[chunk_mask].reshape(-1, 1)
        return ys / num_nets

    def prepare_mple_reciprocity_regressors(self):
        Xs = np.zeros(((self.n_nodes ** 2 - self.n_nodes) // 2, 4, self.calc_num_of_features()))
        idx = 0
        zeros_net = np.zeros((self.n_nodes, self.n_nodes))
        for i in range(self.n_nodes - 1):
            for j in range(i + 1, self.n_nodes):
                change_score_i_j = self.calc_change_scores(zeros_net, (i, j))
                net_with_i_j = zeros_net.copy()
                net_with_i_j[i, j] = 1
                Xs[idx, DyadStateIdx.UPPER_IDX.value] = change_score_i_j
                Xs[idx, DyadStateIdx.LOWER_IDX.value] = self.calc_change_scores(zeros_net, (j, i))
                Xs[idx, DyadStateIdx.RECIPROCAL_IDX.value] = (
                        self.calc_change_scores(net_with_i_j, (j, i)) + change_score_i_j
                )

                idx += 1
        return Xs

    @staticmethod
    def prepare_mple_reciprocity_labels(observed_networks: np.ndarray):
        ys = convert_connectivity_to_dyad_states(observed_networks[..., 0])
        num_nets = observed_networks.shape[-1]
        for i in range(1, num_nets):
            ys += convert_connectivity_to_dyad_states(observed_networks[..., i])
        return ys / num_nets

    def choose_optimization_scheme(self) -> OptimizationScheme:
        """
        Automatically select the appropriate optimization scheme for the model.

        Returns
        -------
        OptimizationScheme
            One of MPLE, MPLE_RECIPROCITY, or MCMLE depending on the metrics.

        Notes
        -----
        - MPLE: Maximum Pseudo-Likelihood Estimation, for dyadic independent metrics
        - MPLE_RECIPROCITY: Extended MPLE for models with only reciprocity dependence
        - MCMLE: Monte Carlo Maximum Likelihood Estimation, for complex dependencies
        """
        if not self._has_dyadic_dependent_metrics:
            return OptimizationScheme.MPLE
        # The only edge dependence comes from reciprocal edges
        if not any(
                [not x._is_dyadic_independent for x in self.metrics if
                 str(x) not in ['total_reciprocity', 'reciprocity']]):
            return OptimizationScheme.MPLE_RECIPROCITY
        return OptimizationScheme.MCMLE

    def find_unnamed_duplicate_metrics(self) -> Sequence[tuple[Metric, str]]:
        """
        Find metrics that are duplicated (same class) and don't have user-provided names.

        Returns
        -------
        Sequence[tuple[Metric, str]]
            List of (metric, class_name) tuples for metrics needing disambiguation.
        """
        class_counts = Counter(type(m).__name__ for m in self.metrics)
        unnamed_duplicates = []
        for metric in self.metrics:
            class_name = type(metric).__name__
            if class_counts[class_name] > 1 and metric._name is None:
                unnamed_duplicates.append((metric, class_name))
        return unnamed_duplicates

    def get_parameter_names(self):
        """
        Returns the names of the parameters of the metrics in the collection.

        If multiple metrics of the same class exist and don't have user-provided
        names, each gets a unique random suffix appended with double underscore
        (e.g., param_name__x7k3a2). Metrics with user-provided names are not
        given random suffixes since the name already disambiguates them.
        """
        # Find metrics needing disambiguation (duplicates without user-provided names)
        unnamed_duplicates = self.find_unnamed_duplicate_metrics()

        # Group by class name for ID generation
        class_instances = {}
        for metric, class_name in unnamed_duplicates:
            class_instances.setdefault(class_name, []).append(metric)

        # Pre-generate unique IDs for each instance needing disambiguation
        metric_ids = {}
        for class_name, instances in class_instances.items():
            used_ids = set()
            for metric in instances:
                metric_id = generate_short_id()
                while metric_id in used_ids:
                    metric_id = generate_short_id()
                used_ids.add(metric_id)
                metric_ids[id(metric)] = metric_id

        # Collect parameter names with disambiguation
        parameter_names = []
        for metric in self.metrics:
            metric_id = metric_ids.get(id(metric))
            for name in metric._get_metric_names():
                if metric_id is not None:
                    name = f"{name}__{metric_id}"
                parameter_names.append(name)

        return tuple(parameter_names)

    def get_ignored_features(self):
        """
        Get the names of features that have been ignored due to collinearity.

        Returns
        -------
        tuple
            Names of ignored features across all metrics in the collection.
        """
        parameter_names = tuple()

        for metric in self.metrics:
            parameter_names += metric._get_ignored_features()

        return parameter_names

    def validate_supports_observed_bootstrap(self):
        for metric in self.metrics:
            if not hasattr(metric, "calculate_bootstrapped_features"):
                raise ValueError(
                    f"metric {metric.metric_name} does not have a calculate_bootstrapped_features method, the "
                    f"model doesn't support observed_bootstrap as a convergence criterion.")

    def bootstrap_observed_features(
            self,
            observed_network: np.ndarray,
            num_subsamples: int = 1000,
            splitting_method: DataBootstrapSplittingMethod = DataBootstrapSplittingMethod.UNIFORM
    ):
        self.validate_supports_observed_bootstrap()
        if observed_network.ndim == 3:
            observed_network = observed_network[..., 0]
        observed_net_size = observed_network.shape[0]
        second_half_size = observed_net_size // 2
        first_half_size = observed_net_size - second_half_size
        first_halves = np.zeros((first_half_size, first_half_size, num_subsamples))
        second_halves = np.zeros((second_half_size, second_half_size, num_subsamples))
        first_halves_indices = np.zeros((first_half_size, num_subsamples), dtype=int)
        second_halves_indices = np.zeros((second_half_size, num_subsamples), dtype=int)
        for i in range(num_subsamples):
            first_half_indices, second_half_indices = split_network_for_bootstrapping(
                observed_net_size, first_half_size, splitting_method=splitting_method
            )
            first_halves[:, :, i] = observed_network[first_half_indices, first_half_indices.T]
            second_halves[:, :, i] = observed_network[second_half_indices, second_half_indices.T]
            first_halves_indices[:, i] = first_half_indices[:, 0]
            second_halves_indices[:, i] = second_half_indices[:, 0]

        bootstrapped_features = np.zeros((self.num_of_features, num_subsamples))

        for metric, start_idx, n_features in self._metric_feature_iter():
            cur_metric_bootstrapped_features = metric.calculate_bootstrapped_features(
                first_halves, second_halves, first_halves_indices, second_halves_indices
            )
            bootstrapped_features[start_idx:start_idx + n_features] = cur_metric_bootstrapped_features

        return bootstrapped_features

init

__init__(metrics: Sequence[Metric], is_directed: bool, n_nodes: int, fix_collinearity=True, collinearity_fixer_sample_size=1000, is_collinearity_distributed=False, mask: npt.NDArray[bool] | None = None, **kwargs)

Source code in pyERGM/metrics.py

def __init__(
        self,
        metrics: Sequence[Metric],
        is_directed: bool,
        n_nodes: int,
        fix_collinearity=True,
        collinearity_fixer_sample_size=1000,
        is_collinearity_distributed=False,
        mask: npt.NDArray[bool] | None = None,
        **kwargs
):
    logger.debug("Initializing MetricsCollection")
    self.metrics = MetricsCollection._copy_metrics(metrics)

    for m in self.metrics:
        m._n_nodes = n_nodes
        m.initialize_indices_to_ignore()

    metric_names = [(type(x).__name__, x._name) for x in self.metrics if x._name is not None]

    if len(set(metric_names)) < len(metric_names):
        raise ValueError(f"Got at least 2 metrics with the same name - {metric_names}")

    self._has_dyadic_dependent_metrics = any([not x._is_dyadic_independent for x in self.metrics])

    self.is_directed = is_directed
    self.mask = None
    if mask is not None:
        if mask.ndim != 1:
            raise ValueError(f"MetricsCollection expects flattened masks (1D). Got {mask.ndim}D")
        self.mask = mask

    if OptimizationScheme.MPLE != self.choose_optimization_scheme() and self.mask is not None:
        raise NotImplementedError("Masking is currently supported only for edge independent models.")

    for x in self.metrics:
        if (x._is_directed is not None) and (x._is_directed != self.is_directed):
            model_is_directed_str = "a directed" if self.is_directed else "an undirected"
            metric_is_directed_str = "a directed" if x._is_directed else "an undirected"
            raise ValueError(f"Trying to initialize {model_is_directed_str} model with {metric_is_directed_str} "
                             f"metric `{str(x)}`!")
        if self.mask is not None and not x.does_support_mask:
            raise ValueError(f"Trying to initialize a masked model with metric `{str(x)}` which does not support "
                             f"masks!")
    self.n_nodes = n_nodes

    if is_collinearity_distributed and self._has_dyadic_dependent_metrics:
        raise ValueError(
            "Distributed optimization is only supported for dyadic-independent models. "
            "This model contains dyadic-dependent metrics: "
            f"{[str(m) for m in self.metrics if not m._is_dyadic_independent]}"
        )

    # Returns the number of features that are being calculated. Since a single metric might return more than one
    # feature, the length of the statistics vector might be larger than the amount of metrics. Since it also depends
    # on the network size, n is a mandatory parameters. That's why we're using the get_effective_feature_count
    # function
    self.num_of_features = self.calc_num_of_features()
    self.features_per_metric = np.array([metric.get_effective_feature_count() for metric in self.metrics])

    if fix_collinearity:
        logger.debug("Starting collinearity fix")
        self.collinearity_fixer(sample_size=collinearity_fixer_sample_size,
                                is_distributed=is_collinearity_distributed,
                                num_samples_per_job=kwargs.get('num_samples_per_job_collinearity_fixer', 5),
                                ratio_threshold=kwargs.get('ratio_threshold_collinearity_fixer', 5e-6),
                                nonzero_thr=kwargs.get('nonzero_threshold_collinearity_fixer', 0.1),
                                )

    self.num_of_metrics = len(self.metrics)
    self.metric_names = tuple([str(metric) for metric in self.metrics])

calculate_statistics

calculate_statistics(W: np.ndarray)

Calculate the statistics of a graph, formally written as g(y).

Parameters:

Name	Type	Description	Default
`W`	`ndarray`	An N x N connectivity matrix.	required

Returns:

Name	Type	Description
`statistics`	`ndarray`	An array of statistics

Source code in pyERGM/metrics.py

def calculate_statistics(self, W: np.ndarray):
    """
    Calculate the statistics of a graph, formally written as g(y).

    Parameters
    ----------
    W : np.ndarray
        An N x N connectivity matrix.

    Returns
    -------
    statistics : np.ndarray
        An array of statistics
    """
    return self.calculate_sample_statistics(W)[:, 0]

calculate_sample_statistics

calculate_sample_statistics(networks_sample: np.ndarray) -> np.ndarray

Calculate the statistics over a sample of networks

Parameters:

Name	Type	Description	Default
`networks_sample`	`ndarray`	The networks sample - an array of n X n X sample_size	required

Returns:

Type	Description
`an array of the statistics vector per sample (num_features X sample_size)`

Source code in pyERGM/metrics.py

def calculate_sample_statistics(
        self,
        networks_sample: np.ndarray,
) -> np.ndarray:
    """
    Calculate the statistics over a sample of networks

    Parameters
    ----------
    networks_sample
        The networks sample - an array of n X n X sample_size
    Returns
    -------
    an array of the statistics vector per sample (num_features X sample_size)
    """
    networks_sample = expand_net_dims(networks_sample)
    if networks_sample.shape[0] != self.n_nodes or networks_sample.shape[1] != self.n_nodes:
        raise ValueError(
            f"Got a sample of networks of size {networks_sample.shape[0]}, but Metrics expect size {self.n_nodes}")

    num_of_samples = networks_sample.shape[2]
    features_of_net_samples = np.zeros((self.num_of_features, num_of_samples))

    for metric, start_idx, n_features in self._metric_feature_iter():
        calc_for_sample_kwargs = {'networks_sample': networks_sample}
        if self.mask is not None:
            calc_for_sample_kwargs |= {'mask': self.mask}
        features = metric.calculate_for_sample(**calc_for_sample_kwargs)

        features_of_net_samples[start_idx:start_idx + n_features] = features

    return features_of_net_samples

calc_num_of_features

calc_num_of_features()

Source code in pyERGM/metrics.py

def calc_num_of_features(self):
    return sum([metric.get_effective_feature_count() for metric in self.metrics])

get_metric

get_metric(metric_name: str) -> Metric

Get a metric instance

Source code in pyERGM/metrics.py

def get_metric(self, metric_name: str) -> Metric:
    """
    Get a metric instance
    """
    return self.metrics[self.metric_names.index(metric_name)]

get_parameter_names

get_parameter_names()

Returns the names of the parameters of the metrics in the collection.

If multiple metrics of the same class exist and don't have user-provided names, each gets a unique random suffix appended with double underscore (e.g., param_name__x7k3a2). Metrics with user-provided names are not given random suffixes since the name already disambiguates them.

Source code in pyERGM/metrics.py

def get_parameter_names(self):
    """
    Returns the names of the parameters of the metrics in the collection.

    If multiple metrics of the same class exist and don't have user-provided
    names, each gets a unique random suffix appended with double underscore
    (e.g., param_name__x7k3a2). Metrics with user-provided names are not
    given random suffixes since the name already disambiguates them.
    """
    # Find metrics needing disambiguation (duplicates without user-provided names)
    unnamed_duplicates = self.find_unnamed_duplicate_metrics()

    # Group by class name for ID generation
    class_instances = {}
    for metric, class_name in unnamed_duplicates:
        class_instances.setdefault(class_name, []).append(metric)

    # Pre-generate unique IDs for each instance needing disambiguation
    metric_ids = {}
    for class_name, instances in class_instances.items():
        used_ids = set()
        for metric in instances:
            metric_id = generate_short_id()
            while metric_id in used_ids:
                metric_id = generate_short_id()
            used_ids.add(metric_id)
            metric_ids[id(metric)] = metric_id

    # Collect parameter names with disambiguation
    parameter_names = []
    for metric in self.metrics:
        metric_id = metric_ids.get(id(metric))
        for name in metric._get_metric_names():
            if metric_id is not None:
                name = f"{name}__{metric_id}"
            parameter_names.append(name)

    return tuple(parameter_names)

get_ignored_features

get_ignored_features()

Get the names of features that have been ignored due to collinearity.

Returns:

Type	Description
`tuple`	Names of ignored features across all metrics in the collection.

Source code in pyERGM/metrics.py

def get_ignored_features(self):
    """
    Get the names of features that have been ignored due to collinearity.

    Returns
    -------
    tuple
        Names of ignored features across all metrics in the collection.
    """
    parameter_names = tuple()

    for metric in self.metrics:
        parameter_names += metric._get_ignored_features()

    return parameter_names

choose_optimization_scheme

choose_optimization_scheme() -> OptimizationScheme

Automatically select the appropriate optimization scheme for the model.

Returns:

Type	Description
`OptimizationScheme`	One of MPLE, MPLE_RECIPROCITY, or MCMLE depending on the metrics.

Notes

MPLE: Maximum Pseudo-Likelihood Estimation, for dyadic independent metrics
MPLE_RECIPROCITY: Extended MPLE for models with only reciprocity dependence
MCMLE: Monte Carlo Maximum Likelihood Estimation, for complex dependencies

Source code in pyERGM/metrics.py

def choose_optimization_scheme(self) -> OptimizationScheme:
    """
    Automatically select the appropriate optimization scheme for the model.

    Returns
    -------
    OptimizationScheme
        One of MPLE, MPLE_RECIPROCITY, or MCMLE depending on the metrics.

    Notes
    -----
    - MPLE: Maximum Pseudo-Likelihood Estimation, for dyadic independent metrics
    - MPLE_RECIPROCITY: Extended MPLE for models with only reciprocity dependence
    - MCMLE: Monte Carlo Maximum Likelihood Estimation, for complex dependencies
    """
    if not self._has_dyadic_dependent_metrics:
        return OptimizationScheme.MPLE
    # The only edge dependence comes from reciprocal edges
    if not any(
            [not x._is_dyadic_independent for x in self.metrics if
             str(x) not in ['total_reciprocity', 'reciprocity']]):
        return OptimizationScheme.MPLE_RECIPROCITY
    return OptimizationScheme.MCMLE

Base Metric Class

Metric

Bases: ABC

Abstract base class for all network metrics in the ERGM framework.

This class defines the interface for computing statistics on networks, including methods for calculating metrics, change scores, and handling sampling operations. All concrete metric implementations must inherit from this class.

Source code in pyERGM/metrics.py

class Metric(ABC):
    """
    Abstract base class for all network metrics in the ERGM framework.

    This class defines the interface for computing statistics on networks, including
    methods for calculating metrics, change scores, and handling sampling operations.
    All concrete metric implementations must inherit from this class.

    """

    def __init__(self, name: str | None = None):
        # Each metric either expects directed or undirected graphs. This field should be initialized in the constructor
        # and should not change.
        self._name = name  # optional user-provided label for disambiguation
        self._is_directed = None
        self._is_dyadic_independent = True
        self._n_nodes = None
        self._indices_to_ignore = None
        self.does_support_mask = False

    def initialize_indices_to_ignore(self):
        """
        Initialize the array tracking which feature indices should be ignored.

        This method creates a boolean array to track features that should be excluded
        from calculations, typically to avoid multicollinearity issues.
        """
        self._indices_to_ignore = np.array([False] * self._get_total_feature_count())

        if hasattr(self, "_indices_from_user"):
            if self._indices_from_user is not None:
                self.update_indices_to_ignore(self._indices_from_user)

    def _handle_indices_to_ignore(self, res, axis=0):
        if self._indices_to_ignore is None:
            return res

        if axis > 1:
            raise ValueError("Axis should be 0 or 1")
        if axis == 1:
            return res[:, ~self._indices_to_ignore]
        return res[~self._indices_to_ignore]

    def calculate(self, input: np.ndarray):
        """
        Calculate metric statistic for a single network.

        Parameters
        ----------
        input : np.ndarray
            A network with shape (n, n).

        Returns
        -------
        np.ndarray or scalar
            The metric statistic(s) for the network.
        """
        result = self.calculate_for_sample(expand_net_dims(input))
        # Single-feature metrics return (k,), multi-feature return (m, k)
        if result.ndim == 1:
            return result[0]
        return result[..., 0]

    def get_effective_feature_count(self):
        """
        How many features does this metric produce. Defaults to 1.
        """
        if self._indices_to_ignore is None:
            return self._get_total_feature_count()
        return np.sum(~self._indices_to_ignore)

    def _get_total_feature_count(self):
        """
        How many features does this metric produce, including the ignored ones. Defaults to 1
        """
        return 1

    def _get_name_prefix(self):
        """
        Returns the name prefix for parameter names (label if set, else empty).
        """
        return f"{self._name}_" if self._name is not None else ""

    @property
    def metric_name(self):
        return self._get_name_prefix() + str(self)

    def update_indices_to_ignore(self, indices_to_ignore):
        """
        Mark specific feature indices to be ignored in metric calculations.

        Parameters
        ----------
        indices_to_ignore : array-like
            Indices of features that should be excluded from calculations.
        """
        self._indices_to_ignore[indices_to_ignore] = True

    def calc_change_score(self, current_network: np.ndarray, indices: tuple):
        """
        The default naive way to calculate the change score (namely, the difference in statistics) of a pair of
        networks.

        The newly proposed network is created by flipping the edge denoted by `indices`

        Returns
        -------
        statistic of proposed_network minus statistic of current_network.
        """
        i, j = indices
        proposed_network = current_network.copy()
        proposed_network[i, j] = 1 - proposed_network[i, j]

        if not self._is_directed:
            proposed_network[j, i] = 1 - proposed_network[j, i]

        proposed_network_stat = self.calculate(proposed_network)
        current_network_stat = self.calculate(current_network)
        return proposed_network_stat - current_network_stat

    @abstractmethod
    def calculate_for_sample(self, networks_sample: np.ndarray):
        """
        Calculate metric statistics for a sample of networks.

        Parameters
        ----------
        networks_sample : np.ndarray
            A collection of networks with shape (n, n, sample_size).

        Returns
        -------
        np.ndarray
            Array of statistics. Shape is (sample_size,) for scalar metrics,
            or (num_features, sample_size) for vector metrics.
        """
        pass

    def calculate_bootstrapped_features(
        self,
        first_halves_to_use: np.ndarray,
        second_halves_to_use: np.ndarray,
        first_halves_indices: np.ndarray,
        second_halves_indices: np.ndarray,
    ) -> np.ndarray:
        """
        Calculate bootstrapped features for observed network analysis.

        This method is used for bootstrap-based confidence interval estimation
        and is only supported by specific metric implementations.

        Raises
        ------
        NotImplementedError
            If the metric does not support bootstrapping.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support bootstrapping"
        )

    def calculate_mple_regressors(
            self,
            Xs_out: np.ndarray,
            feature_col_indices: npt.NDArray[np.int64],
            edge_indices_mask: npt.NDArray[bool],
            observed_network: np.ndarray,
    ) -> None:
        """
        Calculate the design matrix (regressors) for Maximum Pseudo-Likelihood Estimation.

        This method populates the design matrix for MPLE optimization by computing
        change scores for each potential edge in the network.

        Parameters
        ----------
        Xs_out : np.ndarray
            Output array to populate with regressor values. Modified in-place.
        feature_col_indices : np.ndarray
            Column indices in Xs_out corresponding to this metric's features.
        edge_indices_mask : np.ndarray
            Boolean mask indicating which edges to include in the calculation.
        observed_network : np.ndarray
            The observed network adjacency matrix.
        """
        edge_idx_in_full_xs = 0
        edge_idx_in_masked_xs = 0

        if self._is_directed:
            for i in range(self._n_nodes):
                for j in range(self._n_nodes):
                    if i == j:
                        continue
                    if not edge_indices_mask[edge_idx_in_full_xs]:
                        edge_idx_in_full_xs += 1
                        continue
                    indices = (i, j)
                    observed_edge_off = observed_network.copy()
                    observed_edge_off[i, j] = 0

                    Xs_out[edge_idx_in_masked_xs, feature_col_indices] = self.calc_change_score(
                        observed_edge_off, indices
                    )
                    edge_idx_in_full_xs += 1
                    edge_idx_in_masked_xs += 1

        else:
            for i in range(self._n_nodes - 1):
                for j in range(i + 1, self._n_nodes):
                    if not edge_indices_mask[edge_idx_in_full_xs]:
                        edge_idx_in_full_xs += 1
                        continue

                    indices = (i, j)

                    observed_edge_off = observed_network.copy()
                    observed_edge_off[i, j] = 0
                    observed_edge_off[j, i] = 0

                    Xs_out[edge_idx_in_masked_xs, feature_col_indices] = self.calc_change_score(
                        observed_edge_off, indices
                    )
                    edge_idx_in_full_xs += 1
                    edge_idx_in_masked_xs += 1

    def _get_metric_names(self):
        """
        Get the names of the features that this metric produces. Defaults to the name of the metric if the metric
        creates a single feature, and multiple names if the metric creates multiple features.

        Returns
        -------
        parameter_names: tuple
            A tuple of strings, each string is the name of a parameter that this metric produces.
        """
        total_n_features = self._get_total_feature_count()
        if total_n_features == 1:

            return (self.metric_name,)
        else:
            parameter_names = ()
            for i in range(total_n_features):
                if self._indices_to_ignore is not None and self._indices_to_ignore[i]:
                    continue
                parameter_names += (f"{self.metric_name}_{i + 1}",)
            return parameter_names

    def _get_ignored_features(self):
        if self._indices_to_ignore is None or not np.any(self._indices_to_ignore):
            return tuple()

        ignored_features = ()
        for i in range(self._get_total_feature_count()):
            if self._indices_to_ignore is not None and self._indices_to_ignore[i]:
                ignored_features += (f"{str(self)}_{i + 1}",)

        return ignored_features

    def _calc_bootstrapped_scalar_feature(self, first_halves_to_use: np.ndarray,
                                          first_halves_indices: np.ndarray[int], second_halves_indices: np.ndarray[int],
                                          max_feature_val_calculator: Callable):
        num_nodes_in_observed = first_halves_indices.shape[0] + second_halves_indices.shape[0]
        num_nodes_in_first_half = first_halves_indices.shape[0]
        max_feature_in_observed = max_feature_val_calculator(num_nodes_in_observed)
        max_feature_in_first_half = max_feature_val_calculator(num_nodes_in_first_half)
        return self.calculate_for_sample(
            first_halves_to_use) * max_feature_in_observed / max_feature_in_first_half

init

__init__(name: str | None = None)

Source code in pyERGM/metrics.py

def __init__(self, name: str | None = None):
    # Each metric either expects directed or undirected graphs. This field should be initialized in the constructor
    # and should not change.
    self._name = name  # optional user-provided label for disambiguation
    self._is_directed = None
    self._is_dyadic_independent = True
    self._n_nodes = None
    self._indices_to_ignore = None
    self.does_support_mask = False

calculate

calculate(input: np.ndarray)

Calculate metric statistic for a single network.

Parameters:

Name	Type	Description	Default
`input`	`ndarray`	A network with shape (n, n).	required

Returns:

Type	Description
`ndarray or scalar`	The metric statistic(s) for the network.

Source code in pyERGM/metrics.py

def calculate(self, input: np.ndarray):
    """
    Calculate metric statistic for a single network.

    Parameters
    ----------
    input : np.ndarray
        A network with shape (n, n).

    Returns
    -------
    np.ndarray or scalar
        The metric statistic(s) for the network.
    """
    result = self.calculate_for_sample(expand_net_dims(input))
    # Single-feature metrics return (k,), multi-feature return (m, k)
    if result.ndim == 1:
        return result[0]
    return result[..., 0]

calculate_for_sample `abstractmethod`

calculate_for_sample(networks_sample: np.ndarray)

Calculate metric statistics for a sample of networks.

Parameters:

Name	Type	Description	Default
`networks_sample`	`ndarray`	A collection of networks with shape (n, n, sample_size).	required

Returns:

Type	Description
`ndarray`	Array of statistics. Shape is (sample_size,) for scalar metrics, or (num_features, sample_size) for vector metrics.

Source code in pyERGM/metrics.py

@abstractmethod
def calculate_for_sample(self, networks_sample: np.ndarray):
    """
    Calculate metric statistics for a sample of networks.

    Parameters
    ----------
    networks_sample : np.ndarray
        A collection of networks with shape (n, n, sample_size).

    Returns
    -------
    np.ndarray
        Array of statistics. Shape is (sample_size,) for scalar metrics,
        or (num_features, sample_size) for vector metrics.
    """
    pass

calc_change_score

calc_change_score(current_network: np.ndarray, indices: tuple)

The default naive way to calculate the change score (namely, the difference in statistics) of a pair of networks.

The newly proposed network is created by flipping the edge denoted by indices

Returns:

Type	Description
`statistic of proposed_network minus statistic of current_network.`

Source code in pyERGM/metrics.py

def calc_change_score(self, current_network: np.ndarray, indices: tuple):
    """
    The default naive way to calculate the change score (namely, the difference in statistics) of a pair of
    networks.

    The newly proposed network is created by flipping the edge denoted by `indices`

    Returns
    -------
    statistic of proposed_network minus statistic of current_network.
    """
    i, j = indices
    proposed_network = current_network.copy()
    proposed_network[i, j] = 1 - proposed_network[i, j]

    if not self._is_directed:
        proposed_network[j, i] = 1 - proposed_network[j, i]

    proposed_network_stat = self.calculate(proposed_network)
    current_network_stat = self.calculate(current_network)
    return proposed_network_stat - current_network_stat

Edge Count Metrics

NumberOfEdgesUndirected

Bases: NumberOfEdges

Metric for counting edges in an undirected network.

In an undirected network, each edge is counted once (even though it appears twice in the adjacency matrix due to symmetry).

Source code in pyERGM/metrics.py

class NumberOfEdgesUndirected(NumberOfEdges):
    """
    Metric for counting edges in an undirected network.

    In an undirected network, each edge is counted once (even though it appears
    twice in the adjacency matrix due to symmetry).
    """

    def __str__(self):
        return "num_edges_undirected"

    def __init__(self, name: str | None = None):
        super().__init__(name=name)
        self._is_directed = False

    @staticmethod
    def _get_num_edges_in_mat_factor() -> int:
        return 2

NumberOfEdgesDirected

Bases: NumberOfEdges

Metric for counting edges in a directed network.

In a directed network, each directed edge (i -> j) is counted separately from its reverse (j -> i).

Source code in pyERGM/metrics.py

class NumberOfEdgesDirected(NumberOfEdges):
    """
    Metric for counting edges in a directed network.

    In a directed network, each directed edge (i -> j) is counted separately
    from its reverse (j -> i).
    """

    def __str__(self):
        return "num_edges_directed"

    def __init__(self, name: str | None = None):
        super().__init__(name=name)
        self._is_directed = True

    @staticmethod
    def _get_num_edges_in_mat_factor() -> int:
        return 1

Triangle Metrics

NumberOfTrianglesUndirected

Bases: NumberOfTriangles

Metric for counting triangles in an undirected network.

A triangle is a set of three nodes where each pair is connected by an edge. This is a measure of network clustering and transitivity.

Notes

The count is computed using matrix multiplication: tr(W^3) / 6, where W is the adjacency matrix.

Source code in pyERGM/metrics.py

class NumberOfTrianglesUndirected(NumberOfTriangles):
    """
    Metric for counting triangles in an undirected network.

    A triangle is a set of three nodes where each pair is connected by an edge.
    This is a measure of network clustering and transitivity.

    Notes
    -----
    The count is computed using matrix multiplication: tr(W^3) / 6, where W
    is the adjacency matrix.
    """

    def __str__(self):
        return "num_triangles"

    def __init__(self, name: str | None = None):
        super().__init__(name=name)
        self._is_directed = False

    @staticmethod
    def _get_triangle_divisor() -> int:
        return 2

NumberOfTrianglesDirected

Bases: NumberOfTriangles

Metric for counting directed 3-cycles in a directed network.

A directed 3-cycle is a set of three nodes i, j, k where edges form a closed directed path: i→j→k→i. This is a measure of network clustering in directed networks.

Notes

The count is computed using matrix multiplication: tr(W^3) / 3, where W is the adjacency matrix. Each directed 3-cycle is counted exactly 3 times in the trace (once for each possible starting node in the cycle).

Source code in pyERGM/metrics.py

class NumberOfTrianglesDirected(NumberOfTriangles):
    """
    Metric for counting directed 3-cycles in a directed network.

    A directed 3-cycle is a set of three nodes i, j, k where edges form
    a closed directed path: i→j→k→i. This is a measure of network
    clustering in directed networks.

    Notes
    -----
    The count is computed using matrix multiplication: tr(W^3) / 3, where W
    is the adjacency matrix. Each directed 3-cycle is counted exactly 3 times
    in the trace (once for each possible starting node in the cycle).
    """

    def __str__(self):
        return "num_triangles_directed"

    def __init__(self, name: str | None = None):
        super().__init__(name=name)
        self._is_directed = True

    @staticmethod
    def _get_triangle_divisor() -> int:
        return 1

Degree Metrics

InDegree

Bases: BaseDegreeVector

Calculate the in-degree of each node in a directed graph.

In-degree is the number of incoming edges to a node. This metric produces a feature vector of length n (number of nodes), where each element represents the in-degree of the corresponding node.

Parameters:

Name	Type	Description	Default
`indices_from_user`	`array - like`	Indices of nodes whose in-degrees should be excluded from the feature vector to avoid multicollinearity.	`None`
`name`	`str`	Optional name for this metric instance to avoid conflicts with other metrics of the same type.	`None`

Source code in pyERGM/metrics.py

class InDegree(BaseDegreeVector):
    """
    Calculate the in-degree of each node in a directed graph.

    In-degree is the number of incoming edges to a node. This metric produces
    a feature vector of length n (number of nodes), where each element represents
    the in-degree of the corresponding node.

    Parameters
    ----------
    indices_from_user : array-like, optional
        Indices of nodes whose in-degrees should be excluded from the feature vector
        to avoid multicollinearity.
    name : str, optional
        Optional name for this metric instance to avoid conflicts with other metrics of the same type.
    """

    def __str__(self):
        return "indegree"

    def __init__(self, indices_from_user=None, name: str | None = None):
        super().__init__(
            is_directed=True,
            summation_axis=BaseDegreeVector.SummationAxis.ROWS,
            indices_from_user=indices_from_user,
            name=name,
        )

    @classmethod
    def _create_fresh(cls) -> "InDegree":
        """Create a fresh instance for internal use (bootstrap calculations)."""
        return cls()

    def _get_change_score_indices_from_summation_axis(
            self,
            edge_indices: tuple[int, int],
    ) -> tuple[int, ...]:
        # In degree - summing over rows, the statistic of the second node (target in edge) changes.
        return (edge_indices[1],)

    def calculate_mple_regressors(
            self,
            Xs_out: np.ndarray,
            feature_col_indices: npt.NDArray[np.int64],
            edge_indices_mask: npt.NDArray[bool],
            observed_network: np.ndarray,
    ) -> None:
        num_neurons = num_edges_to_num_nodes(edge_indices_mask.size, is_directed=True)
        in_deg_xs = np.tile(np.eye(num_neurons), (num_neurons, 1))[~np.eye(num_neurons, dtype=bool).flatten()]
        in_deg_xs = self._handle_indices_to_ignore(in_deg_xs, axis=1)
        # Looping rather than setting with the full mask directly is negligible for small arrays but much faster for
        # large arrays due too under-the-hood copies of large arrays.
        for masked_idx, non_masked_idx in enumerate(np.where(edge_indices_mask)[0]):
            Xs_out[masked_idx, feature_col_indices] = in_deg_xs[non_masked_idx]

OutDegree

Bases: BaseDegreeVector

Calculate the out-degree of each node in a directed graph.

Out-degree is the number of outgoing edges from a node. This metric produces a feature vector of length n (number of nodes), where each element represents the out-degree of the corresponding node.

Parameters:

Name	Type	Description	Default
`indices_from_user`	`array - like`	Indices of nodes whose out-degrees should be excluded from the feature vector to avoid multicollinearity.	`None`
`name`	`str`	Optional name for this metric instance to avoid conflicts with other metrics of the same type.	`None`

Source code in pyERGM/metrics.py

class OutDegree(BaseDegreeVector):
    """
    Calculate the out-degree of each node in a directed graph.

    Out-degree is the number of outgoing edges from a node. This metric produces
    a feature vector of length n (number of nodes), where each element represents
    the out-degree of the corresponding node.

    Parameters
    ----------
    indices_from_user : array-like, optional
        Indices of nodes whose out-degrees should be excluded from the feature vector
        to avoid multicollinearity.
    name : str, optional
        Optional name for this metric instance to avoid conflicts with other metrics of the same type.
    """

    def __str__(self):
        return "outdegree"

    def __init__(self, indices_from_user=None, name: str | None = None):
        super().__init__(
            is_directed=True,
            summation_axis=BaseDegreeVector.SummationAxis.COLUMNS,
            indices_from_user=indices_from_user,
            name=name,
        )

    @classmethod
    def _create_fresh(cls) -> "OutDegree":
        """Create a fresh instance for internal use (bootstrap calculations)."""
        return cls()

    def _get_change_score_indices_from_summation_axis(
            self,
            edge_indices: tuple[int, int],
    ) -> tuple[int, ...]:
        # Out degree - summing over columns, the statistic of the first node (source in edge) changes.
        return (edge_indices[0],)

    def calculate_mple_regressors(
            self,
            Xs_out: np.ndarray,
            feature_col_indices: npt.NDArray[np.int64],
            edge_indices_mask: npt.NDArray[bool],
            observed_network: np.ndarray,
    ) -> None:
        num_neurons = num_edges_to_num_nodes(edge_indices_mask.size, is_directed=True)
        out_deg_xs = np.repeat(np.eye(num_neurons), num_neurons - 1, axis=0)
        out_deg_xs = self._handle_indices_to_ignore(out_deg_xs, axis=1)
        # See comment in InDegree.calculate_mple_regressors on loop performance.
        for masked_idx, non_masked_idx in enumerate(np.where(edge_indices_mask)[0]):
            Xs_out[masked_idx, feature_col_indices] = out_deg_xs[non_masked_idx]

UndirectedDegree

Bases: BaseDegreeVector

Calculate the degree of each node in an undirected graph.

Degree is the number of edges connected to a node. This metric produces a feature vector of length n (number of nodes), where each element represents the degree of the corresponding node.

Parameters:

Name	Type	Description	Default
`indices_from_user`	`array - like`	Indices of nodes whose degrees should be excluded from the feature vector to avoid multicollinearity.	`None`
`name`	`str`	Optional name for this metric instance to avoid conflicts with other metrics of the same type.	`None`

Source code in pyERGM/metrics.py

class UndirectedDegree(BaseDegreeVector):
    """
    Calculate the degree of each node in an undirected graph.

    Degree is the number of edges connected to a node. This metric produces
    a feature vector of length n (number of nodes), where each element represents
    the degree of the corresponding node.

    Parameters
    ----------
    indices_from_user : array-like, optional
        Indices of nodes whose degrees should be excluded from the feature vector
        to avoid multicollinearity.
    name : str, optional
        Optional name for this metric instance to avoid conflicts with other metrics of the same type.
    """

    def __str__(self):
        return "undirected_degree"

    def __init__(self, indices_from_user=None, name: str | None = None):
        super().__init__(
            is_directed=False,
            summation_axis=BaseDegreeVector.SummationAxis.ROWS,  # it doesn't matter over which axis to sum
            indices_from_user=indices_from_user,
            name=name,
        )

    @classmethod
    def _create_fresh(cls) -> "UndirectedDegree":
        """Create a fresh instance for internal use (bootstrap calculations)."""
        return cls()

    def _get_change_score_indices_from_summation_axis(
            self,
            edge_indices: tuple[int, int],
    ) -> tuple[int, ...]:
        # Undirected degree - the statistic of both nodes changes.
        return edge_indices

    def calculate_mple_regressors(
            self,
            Xs_out: np.ndarray,
            feature_col_indices: npt.NDArray[np.int64],
            edge_indices_mask: npt.NDArray[bool],
            observed_network: np.ndarray,
    ) -> None:
        num_node_pairs = edge_indices_mask.size
        num_nodes = num_edges_to_num_nodes(num_node_pairs, is_directed=False)
        deg_xs = np.zeros((num_node_pairs, num_nodes))
        row_selector = np.arange(num_node_pairs, dtype=int)
        i_indices, j_indices = np.triu_indices(num_nodes, k=1)
        deg_xs[row_selector, i_indices] = 1
        deg_xs[row_selector, j_indices] = 1
        deg_xs = self._handle_indices_to_ignore(deg_xs, axis=1)
        # See comment in InDegree.calculate_mple_regressors on loop performance.
        for masked_idx, non_masked_idx in enumerate(np.where(edge_indices_mask)[0]):
            Xs_out[masked_idx, feature_col_indices] = deg_xs[non_masked_idx]

Reciprocity Metrics

Reciprocity

Bases: Metric

Calculate reciprocity indicators for all node pairs in a directed graph.

This metric produces a feature vector of size n-choose-2, where each element indicates whether a pair of nodes (i, j) has reciprocal connections, i.e., both i -> j and j -> i edges exist. Formally: y_{i,j} * y_{j,i} for all pairs.

Returns:

Type	Description
`The metric returns a vector where 1 indicates reciprocal connection and 0 otherwise.`

Source code in pyERGM/metrics.py

class Reciprocity(Metric):
    """
    Calculate reciprocity indicators for all node pairs in a directed graph.

    This metric produces a feature vector of size n-choose-2, where each element
    indicates whether a pair of nodes (i, j) has reciprocal connections, i.e.,
    both i -> j and j -> i edges exist. Formally: y_{i,j} * y_{j,i} for all pairs.

    Returns
    -------
    The metric returns a vector where 1 indicates reciprocal connection and 0 otherwise.
    """

    def __str__(self):
        return "reciprocity"

    def __init__(self, name: str | None = None):
        super().__init__(name=name)
        self._is_directed = True
        self._is_dyadic_independent = False

    def calculate_for_sample(self, networks_sample: np.ndarray):
        n = networks_sample.shape[0]
        # Element-wise multiply each network with its transpose
        # networks_sample is (n, n, sample_size), transpose swaps axes 0 and 1
        reciprocal = networks_sample * np.transpose(networks_sample, (1, 0, 2))
        # Extract upper triangular elements for each sample
        triu_indices = np.triu_indices(n, 1)
        return reciprocal[triu_indices[0], triu_indices[1], :]

    def _get_total_feature_count(self):
        # n choose 2
        return self._n_nodes * (self._n_nodes - 1) // 2

    def calc_change_score(self, current_network: np.ndarray, indices: tuple):
        # Note: we intentionally initialize the whole matrix and return np.triu_indices() by the end (rather than
        # initializing an array of zeros of size n choose 2) to ensure compliance with the indexing returned by
        # the calculate method.
        i, j = indices
        all_changes = np.zeros(current_network.shape)
        min_idx = min(indices)
        max_idx = max(indices)

        if current_network[j, i] and not current_network[i, j]:
            all_changes[min_idx, max_idx] = 1
        elif current_network[j, i] and current_network[i, j]:
            all_changes[min_idx, max_idx] = -1
        return all_changes[np.triu_indices(all_changes.shape[0], 1)]

TotalReciprocity

Bases: Metric

Calculate the total number of reciprocal connections in a directed network.

This metric counts the number of node pairs (i, j) where both i -> j and j -> i edges exist. Unlike the Reciprocity metric which returns a vector for each pair, this returns a single scalar value representing the total count.

Returns:

Type	Description
`float`	The total number of reciprocal dyads in the network.

Source code in pyERGM/metrics.py

class TotalReciprocity(Metric):
    """
    Calculate the total number of reciprocal connections in a directed network.

    This metric counts the number of node pairs (i, j) where both i -> j and j -> i
    edges exist. Unlike the Reciprocity metric which returns a vector for each pair,
    this returns a single scalar value representing the total count.

    Returns
    -------
    float
        The total number of reciprocal dyads in the network.
    """

    def __str__(self):
        return "total_reciprocity"

    def __init__(self, name: str | None = None):
        super().__init__(name=name)
        self._is_directed = True
        self._is_dyadic_independent = False

    @staticmethod
    @njit
    def calc_change_score(current_network: np.ndarray, indices: tuple):
        i, j = indices

        if current_network[j, i] and not current_network[i, j]:
            return 1
        elif current_network[j, i] and current_network[i, j]:
            return -1
        else:
            return 0

    def calculate_for_sample(self, networks_sample: np.ndarray):
        return np.einsum("ijk,jik->k", networks_sample, networks_sample) / 2

    def calculate_bootstrapped_features(self, first_halves_to_use: np.ndarray,
                                        second_halves_to_use: np.ndarray,
                                        first_halves_indices: np.ndarray[int], second_halves_indices: np.ndarray[int]):
        """
        Calculates the bootstrapped number of reciprocal dyads, by counting such pairs in the sampled subnetworks, and
        normalizing by network size (i.e., calculating the fraction of existing reciprocal dyads out of all possible
        ones in sampled subnetworks, and multiplying by the number of possible reciprocal dyads in the full observed
        network).
        Parameters
        ----------
        first_halves_to_use
            Multiple samples of subnetworks of an observed network, representing the connectivity between half of the
            nodes in the large network.
        second_halves_to_use
            The subnetworks formed by the complementary set of nodes of the large network for each sample.
        first_halves_indices
            The indices of the nodes in the first half of the large network for each sample, according to the ordering
            of the nodes in the large network.
        second_halves_indices
            The indices of the nodes in the second half of the large network for each sample, according to the ordering

        Returns
        -------
        Properly normalized statistics of subnetworks of an observed network.
        """
        return self._calc_bootstrapped_scalar_feature(first_halves_to_use, first_halves_indices, second_halves_indices,
                                                      lambda n: n * (n - 1) / 2)

calculate_bootstrapped_features

calculate_bootstrapped_features(first_halves_to_use: np.ndarray, second_halves_to_use: np.ndarray, first_halves_indices: np.ndarray[int], second_halves_indices: np.ndarray[int])

Calculates the bootstrapped number of reciprocal dyads, by counting such pairs in the sampled subnetworks, and normalizing by network size (i.e., calculating the fraction of existing reciprocal dyads out of all possible ones in sampled subnetworks, and multiplying by the number of possible reciprocal dyads in the full observed network).

Parameters:

Name	Type	Description	Default
`first_halves_to_use`	`ndarray`	Multiple samples of subnetworks of an observed network, representing the connectivity between half of the nodes in the large network.	required
`second_halves_to_use`	`ndarray`	The subnetworks formed by the complementary set of nodes of the large network for each sample.	required
`first_halves_indices`	`ndarray[int]`	The indices of the nodes in the first half of the large network for each sample, according to the ordering of the nodes in the large network.	required
`second_halves_indices`	`ndarray[int]`	The indices of the nodes in the second half of the large network for each sample, according to the ordering	required

Returns:

Type	Description
`Properly normalized statistics of subnetworks of an observed network.`

Source code in pyERGM/metrics.py

def calculate_bootstrapped_features(self, first_halves_to_use: np.ndarray,
                                    second_halves_to_use: np.ndarray,
                                    first_halves_indices: np.ndarray[int], second_halves_indices: np.ndarray[int]):
    """
    Calculates the bootstrapped number of reciprocal dyads, by counting such pairs in the sampled subnetworks, and
    normalizing by network size (i.e., calculating the fraction of existing reciprocal dyads out of all possible
    ones in sampled subnetworks, and multiplying by the number of possible reciprocal dyads in the full observed
    network).
    Parameters
    ----------
    first_halves_to_use
        Multiple samples of subnetworks of an observed network, representing the connectivity between half of the
        nodes in the large network.
    second_halves_to_use
        The subnetworks formed by the complementary set of nodes of the large network for each sample.
    first_halves_indices
        The indices of the nodes in the first half of the large network for each sample, according to the ordering
        of the nodes in the large network.
    second_halves_indices
        The indices of the nodes in the second half of the large network for each sample, according to the ordering

    Returns
    -------
    Properly normalized statistics of subnetworks of an observed network.
    """
    return self._calc_bootstrapped_scalar_feature(first_halves_to_use, first_halves_indices, second_halves_indices,
                                                  lambda n: n * (n - 1) / 2)

Type-Based Metrics

NumberOfEdgesTypesUndirected

Bases: NumberOfEdgesTypes

Source code in pyERGM/metrics.py

class NumberOfEdgesTypesUndirected(NumberOfEdgesTypes):
    def __str__(self):
        return "num_edges_between_types_undirected"

    @staticmethod
    def _get_num_edges_in_mat_factor():
        return 2

    def __init__(self, exogenous_attr: Sequence[Any], indices_from_user=None, name: str | None = None):
        super().__init__(exogenous_attr, indices_from_user, name=name)
        self._is_directed = False

    def _get_total_feature_count(self):
        num_unique_types = len(self.unique_types)
        return num_unique_types * (num_unique_types + 1) // 2

    def _get_sorted_canonical_type_pairs(self):
        return [p for p in self.sorted_type_pairs if p == tuple(sorted(p))]

    def _calc_type_pairs_indices(self):
        self._sorted_type_pairs_indices = {}

        sorted_canonical_type_paris = self._get_sorted_canonical_type_pairs()
        for i, (a, b) in enumerate(sorted_canonical_type_paris):
            self._sorted_type_pairs_indices[(a, b)] = i
            if a != b:
                self._sorted_type_pairs_indices[(b, a)] = i

    def _get_flattened_edge_type_idx_assignment(self):
        # Reducing 1 because the type indexing in self._edge_type_idx_assignment is 1-based (0 is reserved for
        # non-exising edges).
        return self._edge_type_idx_assignment[
            np.triu_indices(self._edge_type_idx_assignment.shape[0], k=1)].flatten() - 1

    def _get_metric_names(self):
        parameter_names = tuple()

        sorted_canonical_type_pairs = self._get_sorted_canonical_type_pairs()
        for i in range(self._get_total_feature_count()):
            if self._indices_to_ignore is not None and self._indices_to_ignore[i]:
                continue
            type_pair = str(sorted_canonical_type_pairs[i][0]) + "__" + str(sorted_canonical_type_pairs[i][1])
            parameter_names += (f"{self.metric_name}_{type_pair}",)

        return parameter_names

    def _get_ignored_features(self):
        if self._indices_to_ignore is None or not np.any(self._indices_to_ignore):
            return tuple()

        ignored_features = ()
        sorted_canonical_type_pairs = self._get_sorted_canonical_type_pairs()
        for i in range(self._get_total_feature_count()):
            if self._indices_to_ignore is not None and self._indices_to_ignore[i]:
                type_pair = str(sorted_canonical_type_pairs[i][0]) + "__" + str(sorted_canonical_type_pairs[i][1])
                ignored_features += (f"{self.metric_name}_{type_pair}",)

        return ignored_features

NumberOfEdgesTypesDirected

Bases: NumberOfEdgesTypes

A metric that counts how many edges exist between different node types in a directed graph. For example - A graph with n nodes, with an exogenous attribute type=[A, B] assigned to each node. The metric counts the number of edges between nodes of type A->A, A->B, B->A, B->B of a given graph, yielding len(type)**2 features.

Parameters:

Name	Type	Description	Default
`exogenous_attr`	`Sequence[Any]`	A sequence of attributes assigned to each node in a graph with n nodes.	required
`indices_from_user`	`list`	List of indices to ignore in the metric calculation.	`None`

Source code in pyERGM/metrics.py

class NumberOfEdgesTypesDirected(NumberOfEdgesTypes):
    """
    A metric that counts how many edges exist between different node types in a directed graph.
    For example -
        A graph with `n` nodes, with an exogenous attribute `type=[A, B]` assigned to each node.
        The metric counts the number of edges between nodes of type A->A, A->B, B->A, B->B of a given graph,
        yielding len(type)**2 features.

    Parameters
    ----------
    exogenous_attr : Sequence[Any]
        A sequence of attributes assigned to each node in a graph with n nodes.

    indices_from_user : list, optional
        List of indices to ignore in the metric calculation.
    """

    def __str__(self):
        return "num_edges_between_types_directed"

    def __init__(self, exogenous_attr: Sequence[Any], indices_from_user=None, name: str | None = None):
        super().__init__(exogenous_attr, indices_from_user, name=name)
        self._is_directed = True

    @staticmethod
    def _get_num_edges_in_mat_factor():
        return 1

    def _get_sorted_canonical_type_pairs(self):
        return self.sorted_type_pairs

    def _calc_type_pairs_indices(self):
        self._sorted_type_pairs_indices = {pair: i for i, pair in enumerate(self.sorted_type_pairs)}

    def _get_total_feature_count(self):
        return len(self.unique_types) ** 2

    def _get_flattened_edge_type_idx_assignment(self):
        # Reducing 1 because the type indexing in self._edge_type_idx_assignment is 1-based (0 is reserved for
        # non-exising edges).
        return flatten_square_matrix_to_edge_list(self._edge_type_idx_assignment, is_directed=True) - 1

    def _get_metric_names(self):
        parameter_names = tuple()

        for i in range(self._get_total_feature_count()):
            if self._indices_to_ignore is not None and self._indices_to_ignore[i]:
                continue
            type_pair = str(self.sorted_type_pairs[i][0]) + "__" + str(self.sorted_type_pairs[i][1])
            parameter_names += (f"{self.metric_name}_{type_pair}",)

        return parameter_names

    def _get_ignored_features(self):
        if self._indices_to_ignore is None or not np.any(self._indices_to_ignore):
            return tuple()

        ignored_features = ()
        for i in range(self._get_total_feature_count()):
            if self._indices_to_ignore is not None and self._indices_to_ignore[i]:
                type_pair = str(self.sorted_type_pairs[i][0]) + "__" + str(self.sorted_type_pairs[i][1])
                ignored_features += (f"{self.metric_name}_{type_pair}",)

        return ignored_features

Node Attribute Metrics

NodeAttrSum

Bases: ExWeightNumEdges

Source code in pyERGM/metrics.py

class NodeAttrSum(ExWeightNumEdges):
    def __init__(self, exogenous_attr: Sequence[Any], is_directed: bool, name: str | None = None):
        super().__init__(exogenous_attr, name=name)
        self._is_directed = is_directed

    def _calc_edge_weights(self):
        num_nodes = len(self.exogenous_attr)
        self.edge_weights = np.zeros((self._get_num_weight_mats(), num_nodes, num_nodes))
        for i in range(num_nodes):
            for j in range(num_nodes):
                if i == j:
                    continue
                self.edge_weights[0, i, j] = self.exogenous_attr[i] + self.exogenous_attr[j]

    def _get_num_weight_mats(self):
        return 1

    def __str__(self):
        return "node_attribute_sum"

NodeAttrSumOut

Bases: ExWeightNumEdges

Source code in pyERGM/metrics.py

class NodeAttrSumOut(ExWeightNumEdges):
    def __init__(self, exogenous_attr: Sequence[Any], name: str | None = None):
        super().__init__(exogenous_attr, name=name)
        self._is_directed = True

    def _calc_edge_weights(self):
        num_nodes = len(self.exogenous_attr)
        self.edge_weights = np.zeros((self._get_num_weight_mats(), num_nodes, num_nodes))
        for i in range(num_nodes):
            self.edge_weights[0, i, :] = self.exogenous_attr[i] * np.ones(num_nodes)
            self.edge_weights[0, i, i] = 0

    def _get_num_weight_mats(self):
        return 1

    def __str__(self):
        return "node_attribute_sum_out"

NodeAttrSumIn

Bases: ExWeightNumEdges

Source code in pyERGM/metrics.py

class NodeAttrSumIn(ExWeightNumEdges):
    def __init__(self, exogenous_attr: Sequence[Any], name: str | None = None):
        super().__init__(exogenous_attr, name=name)
        self._is_directed = True

    def _calc_edge_weights(self):
        num_nodes = len(self.exogenous_attr)
        self.edge_weights = np.zeros((self._get_num_weight_mats(), num_nodes, num_nodes))
        for j in range(num_nodes):
            self.edge_weights[0, :, j] = self.exogenous_attr[j] * np.ones(num_nodes)
            self.edge_weights[0, j, j] = 0

    def _get_num_weight_mats(self):
        return 1

    def __str__(self):
        return "node_attribute_in"

SumDistancesConnectedNeurons

Bases: ExWeightNumEdges

Sum of Euclidean distances between all connected node pairs.

This metric weights each edge by the Euclidean distance between the spatial positions of the connected nodes. Useful for modeling spatial effects in networks.

Parameters:

Name	Type	Description	Default
`exogenous_attr`	`pd.DataFrame, pd.Series, np.ndarray, list, or tuple`	Spatial coordinates for each node. If 1D, interpreted as positions on a line. If 2D, each row represents a node and columns represent coordinate dimensions.	required
`is_directed`	`bool`	Whether the network is directed.	required

Source code in pyERGM/metrics.py

class SumDistancesConnectedNeurons(ExWeightNumEdges):
    """
    Sum of Euclidean distances between all connected node pairs.

    This metric weights each edge by the Euclidean distance between the spatial
    positions of the connected nodes. Useful for modeling spatial effects in networks.

    Parameters
    ----------
    exogenous_attr : pd.DataFrame, pd.Series, np.ndarray, list, or tuple
        Spatial coordinates for each node. If 1D, interpreted as positions on a line.
        If 2D, each row represents a node and columns represent coordinate dimensions.
    is_directed : bool
        Whether the network is directed.
    """

    def __init__(self, exogenous_attr: pd.DataFrame | pd.Series | np.ndarray | list | tuple, is_directed: bool,
                 name: str | None = None):
        if isinstance(exogenous_attr, (pd.DataFrame, pd.Series, list, tuple)):
            exogenous_attr = np.array(exogenous_attr)
        elif not isinstance(exogenous_attr, np.ndarray):
            raise ValueError(
                f"Unsupported type of positions: {type(exogenous_attr)}. Supported types are pd.DataFrame, "
                f"pd.Series, list, tuple and np.ndarray.")
        if len(exogenous_attr.shape) == 1:
            exogenous_attr = exogenous_attr.reshape(-1, 1)

        super().__init__(exogenous_attr, name=name)
        self._is_directed = is_directed

    def _calc_edge_weights(self):
        num_nodes = len(self.exogenous_attr)
        self.edge_weights = np.reshape(squareform(pdist(self.exogenous_attr, metric='euclidean')),
                                       (self._get_num_weight_mats(), num_nodes, num_nodes))

    def calc_change_score(self, current_network: np.ndarray, indices: tuple):
        sign = -1 if current_network[indices[0], indices[1]] else 1
        return sign * self.edge_weights[:, indices[0], indices[1]]

    def _get_num_weight_mats(self):
        return 1

    def __str__(self):
        return "sum_distances_connected_neurons"

Metrics Module

MetricsCollection

MetricsCollection

__init__

calculate_statistics

calculate_sample_statistics

calc_num_of_features

get_metric

get_parameter_names

get_ignored_features

choose_optimization_scheme

Base Metric Class

Metric

__init__

calculate

calculate_for_sample abstractmethod

calc_change_score

Edge Count Metrics

NumberOfEdgesUndirected

NumberOfEdgesUndirected

NumberOfEdgesDirected

NumberOfEdgesDirected

Triangle Metrics

NumberOfTrianglesUndirected

NumberOfTrianglesUndirected

NumberOfTrianglesDirected

NumberOfTrianglesDirected

Degree Metrics

InDegree

InDegree

OutDegree

OutDegree

UndirectedDegree

UndirectedDegree

Reciprocity Metrics

Reciprocity

Reciprocity

TotalReciprocity

TotalReciprocity

calculate_bootstrapped_features

Type-Based Metrics

NumberOfEdgesTypesUndirected

NumberOfEdgesTypesUndirected

NumberOfEdgesTypesDirected

NumberOfEdgesTypesDirected

Node Attribute Metrics

NodeAttrSum

NodeAttrSum

NodeAttrSumOut

NodeAttrSumOut

NodeAttrSumIn

NodeAttrSumIn

SumDistancesConnectedNeurons

SumDistancesConnectedNeurons

init

init

calculate_for_sample `abstractmethod`