# This code is part of Qiskit.
#
# (C) Copyright IBM 2020, 2023.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""CVaRMeasurement class."""
from typing import Callable, Optional, Tuple, Union, cast, Dict
import numpy as np
from qiskit.circuit import ParameterExpression
from qiskit.opflow.exceptions import OpflowError
from qiskit.opflow.list_ops import ListOp, SummedOp, TensoredOp
from qiskit.opflow.operator_base import OperatorBase
from qiskit.opflow.primitive_ops import PauliOp, PauliSumOp
from qiskit.opflow.state_fns.circuit_state_fn import CircuitStateFn
from qiskit.opflow.state_fns.dict_state_fn import DictStateFn
from qiskit.opflow.state_fns.operator_state_fn import OperatorStateFn
from qiskit.opflow.state_fns.state_fn import StateFn
from qiskit.opflow.state_fns.vector_state_fn import VectorStateFn
from qiskit.quantum_info import Statevector
from qiskit.utils.deprecation import deprecate_func
[문서]class CVaRMeasurement(OperatorStateFn):
r"""Deprecated: A specialized measurement class to compute CVaR expectation values.
See https://arxiv.org/pdf/1907.04769.pdf for further details.
Used in :class:`~qiskit.opflow.CVaRExpectation`, see there for more details.
"""
primitive: OperatorBase
# TODO allow normalization somehow?
@deprecate_func(
since="0.24.0",
additional_msg="For code migration guidelines, visit https://qisk.it/opflow_migration.",
)
def __init__(
self,
primitive: OperatorBase = None,
alpha: float = 1.0,
coeff: Union[complex, ParameterExpression] = 1.0,
) -> None:
"""
Args:
primitive: The ``OperatorBase`` which defines the diagonal operator
measurement.
coeff: A coefficient by which to multiply the state function
alpha: A real-valued parameter between 0 and 1 which specifies the
fraction of observed samples to include when computing the
objective value. alpha = 1 corresponds to a standard observable
expectation value. alpha = 0 corresponds to only using the single
sample with the lowest energy. alpha = 0.5 corresponds to ranking each
observation by lowest energy and using the best
Raises:
ValueError: TODO remove that this raises an error
ValueError: If alpha is not in [0, 1].
OpflowError: If the primitive is not diagonal.
"""
if primitive is None:
raise ValueError
if not 0 <= alpha <= 1:
raise ValueError("The parameter alpha must be in [0, 1].")
self._alpha = alpha
if not _check_is_diagonal(primitive):
raise OpflowError(
"Input operator to CVaRMeasurement must be diagonal, but is not:", str(primitive)
)
super().__init__(primitive, coeff=coeff, is_measurement=True)
@property
def alpha(self) -> float:
"""A real-valued parameter between 0 and 1 which specifies the
fraction of observed samples to include when computing the
objective value. alpha = 1 corresponds to a standard observable
expectation value. alpha = 0 corresponds to only using the single
sample with the lowest energy. alpha = 0.5 corresponds to ranking each
observation by lowest energy and using the best half.
Returns:
The parameter alpha which was given at initialization
"""
return self._alpha
@property
def settings(self) -> Dict:
"""Return settings."""
return {"primitive": self._primitive, "coeff": self._coeff, "alpha": self._alpha}
[문서] def add(self, other: OperatorBase) -> SummedOp:
return SummedOp([self, other])
[문서] def adjoint(self):
"""The adjoint of a CVaRMeasurement is not defined.
Returns:
Does not return anything, raises an error.
Raises:
OpflowError: The adjoint of a CVaRMeasurement is not defined.
"""
raise OpflowError("Adjoint of a CVaR measurement not defined")
[문서] def mul(self, scalar: Union[complex, ParameterExpression]) -> "CVaRMeasurement":
if not isinstance(scalar, (int, float, complex, ParameterExpression)):
raise ValueError(
"Operators can only be scalar multiplied by float or complex, not "
"{} of type {}.".format(scalar, type(scalar))
)
return self.__class__(self.primitive, coeff=self.coeff * scalar, alpha=self._alpha)
[문서] def tensor(self, other: OperatorBase) -> Union["OperatorStateFn", TensoredOp]:
if isinstance(other, OperatorStateFn):
return OperatorStateFn(
self.primitive.tensor(other.primitive), coeff=self.coeff * other.coeff
)
return TensoredOp([self, other])
[문서] def to_density_matrix(self, massive: bool = False):
"""Not defined."""
raise NotImplementedError
[문서] def to_matrix_op(self, massive: bool = False):
"""Not defined."""
raise NotImplementedError
[문서] def to_matrix(self, massive: bool = False):
"""Not defined."""
raise NotImplementedError
[문서] def to_circuit_op(self):
"""Not defined."""
raise NotImplementedError
def __str__(self) -> str:
return f"CVaRMeasurement({str(self.primitive)}) * {self.coeff}"
[문서] def eval(
self, front: Union[str, dict, np.ndarray, OperatorBase, Statevector] = None
) -> complex:
r"""
Given the energies of each sampled measurement outcome (H_i) as well as the
sampling probability of each measurement outcome (p_i, we can compute the
CVaR as H_j + 1/α*(sum_i<j p_i*(H_i - H_j)). Note that index j corresponds
to the measurement outcome such that only some of the samples with
measurement outcome j will be used in computing CVaR. Note also that the
sampling probabilities serve as an alternative to knowing the counts of each
observation.
This computation is broken up into two subroutines. One which evaluates each
measurement outcome and determines the sampling probabilities of each. And one
which carries out the above calculation. The computation is split up this way
to enable a straightforward calculation of the variance of this estimator.
Args:
front: A StateFn or primitive which specifies the results of evaluating
a quantum state.
Returns:
The CVaR of the diagonal observable specified by self.primitive and
the sampled quantum state described by the inputs
(energies, probabilities). For index j (described above), the CVaR
is computed as H_j + 1/α*(sum_i<j p_i*(H_i - H_j))
"""
energies, probabilities = self.get_outcome_energies_probabilities(front)
return self.compute_cvar(energies, probabilities)
[문서] def eval_variance(
self, front: Optional[Union[str, dict, np.ndarray, OperatorBase]] = None
) -> complex:
r"""
Given the energies of each sampled measurement outcome (H_i) as well as the
sampling probability of each measurement outcome (p_i, we can compute the
variance of the CVaR estimator as
H_j^2 + 1/α * (sum_i<j p_i*(H_i^2 - H_j^2)).
This follows from the definition that Var[X] = E[X^2] - E[X]^2.
In this case, X = E[<bi|H|bi>], where H is the diagonal observable and bi
corresponds to measurement outcome i. Given this, E[X^2] = E[<bi|H|bi>^2]
Args:
front: A StateFn or primitive which specifies the results of evaluating
a quantum state.
Returns:
The Var[CVaR] of the diagonal observable specified by self.primitive
and the sampled quantum state described by the inputs
(energies, probabilities). For index j (described above), the CVaR
is computed as H_j^2 + 1/α*(sum_i<j p_i*(H_i^2 - H_j^2))
"""
energies, probabilities = self.get_outcome_energies_probabilities(front)
sq_energies = [energy**2 for energy in energies]
return self.compute_cvar(sq_energies, probabilities) - self.eval(front) ** 2
[문서] def get_outcome_energies_probabilities(
self, front: Optional[Union[str, dict, np.ndarray, OperatorBase, Statevector]] = None
) -> Tuple[list, list]:
r"""
In order to compute the CVaR of an observable expectation, we require
the energies of each sampled measurement outcome as well as the sampling
probability of each measurement outcome. Note that the counts for each
measurement outcome will also suffice (and this is often how the CVaR
is presented).
Args:
front: A StateFn or a primitive which defines a StateFn.
This input holds the results of a sampled/simulated circuit.
Returns:
Two lists of equal length. `energies` contains the energy of each
unique measurement outcome computed against the diagonal observable
stored in self.primitive. `probabilities` contains the corresponding
sampling probability for each measurement outcome in `energies`.
Raises:
ValueError: front isn't a DictStateFn or VectorStateFn
"""
if isinstance(front, CircuitStateFn):
front = cast(StateFn, front.eval())
# Standardize the inputs to a dict
if isinstance(front, DictStateFn):
data = front.primitive
elif isinstance(front, VectorStateFn):
vec = front.primitive.data
# Determine how many bits are needed
key_len = int(np.ceil(np.log2(len(vec))))
# Convert the vector primitive into a dict. The formatting here ensures
# that the proper number of leading `0` characters are added.
data = {format(index, "0" + str(key_len) + "b"): val for index, val in enumerate(vec)}
else:
raise ValueError("Unsupported input to CVaRMeasurement.eval:", type(front))
obs = self.primitive
outcomes = list(data.items())
# add energy evaluation
for i, outcome in enumerate(outcomes):
key = outcome[0]
outcomes[i] += (obs.eval(key).adjoint().eval(key),) # type: ignore
# Sort each observation based on it's energy
outcomes = sorted(outcomes, key=lambda x: x[2]) # type: ignore
# Here probabilities are the (root) probabilities of
# observing each state. energies are the expectation
# values of each state with the provided Hamiltonian.
_, root_probabilities, energies = zip(*outcomes)
# Square the dict values
# (since CircuitSampler takes the root...)
probabilities = [p_i * np.conj(p_i) for p_i in root_probabilities]
return list(energies), probabilities
[문서] def compute_cvar(self, energies: list, probabilities: list) -> complex:
r"""
Given the energies of each sampled measurement outcome (H_i) as well as the
sampling probability of each measurement outcome (p_i, we can compute the
CVaR. Note that the sampling probabilities serve as an alternative to knowing
the counts of each observation and that the input energies are assumed to be
sorted in increasing order.
Consider the outcome with index j, such that only some of the samples with
measurement outcome j will be used in computing CVaR. The CVaR calculation
can then be separated into two parts. First we sum each of the energies for
outcomes i < j, weighted by the probability of observing that outcome (i.e
the normalized counts). Second, we add the energy for outcome j, weighted by
the difference (α - \sum_i<j p_i)
Args:
energies: A list containing the energies (H_i) of each sample measurement
outcome, sorted in increasing order.
probabilities: The sampling probabilities (p_i) for each corresponding
measurement outcome.
Returns:
The CVaR of the diagonal observable specified by self.primitive and
the sampled quantum state described by the inputs
(energies, probabilities). For index j (described above), the CVaR
is computed as H_j + 1/α * (sum_i<j p_i*(H_i - H_j))
Raises:
ValueError: front isn't a DictStateFn or VectorStateFn
"""
alpha = self._alpha
# Determine j, the index of the measurement outcome such
# that only some samples with this outcome will be used to
# compute the CVaR.
j = 0
running_total = 0
for i, p_i in enumerate(probabilities):
running_total += p_i
j = i
if running_total > alpha:
break
h_j = energies[j]
cvar = alpha * h_j
if alpha == 0 or j == 0:
return self.coeff * h_j
energies = energies[:j]
probabilities = probabilities[:j]
# Let H_i be the energy associated with outcome i
# and let the outcomes be sorted by ascending energy.
# Let p_i be the probability of observing outcome i.
# CVaR = H_j + 1/α*(sum_i<j p_i*(H_i - H_j))
for h_i, p_i in zip(energies, probabilities):
cvar += p_i * (h_i - h_j)
return self.coeff * cvar / alpha
[문서] def traverse(
self, convert_fn: Callable, coeff: Optional[Union[complex, ParameterExpression]] = None
) -> OperatorBase:
r"""
Apply the convert_fn to the internal primitive if the primitive is an Operator (as in
the case of ``OperatorStateFn``). Otherwise do nothing. Used by converters.
Args:
convert_fn: The function to apply to the internal OperatorBase.
coeff: A coefficient to multiply by after applying convert_fn.
If it is None, self.coeff is used instead.
Returns:
The converted StateFn.
"""
if coeff is None:
coeff = self.coeff
if isinstance(self.primitive, OperatorBase):
return self.__class__(convert_fn(self.primitive), coeff=coeff, alpha=self._alpha)
return self
[문서] def sample(self, shots: int = 1024, massive: bool = False, reverse_endianness: bool = False):
raise NotImplementedError
def _check_is_diagonal(operator: OperatorBase) -> bool:
"""Check whether ``operator`` is diagonal.
Args:
operator: The operator to check for diagonality.
Returns:
True, if the operator is diagonal, False otherwise.
Raises:
OpflowError: If the operator is not diagonal.
"""
if isinstance(operator, PauliOp):
# every X component must be False
return not np.any(operator.primitive.x)
# For sums (PauliSumOp and SummedOp), we cover the case of sums of diagonal paulis, but don't
# raise since there might be summand canceling the non-diagonal parts. That case is checked
# in the inefficient matrix check at the bottom.
if isinstance(operator, PauliSumOp):
if not np.any(operator.primitive.paulis.x):
return True
elif isinstance(operator, SummedOp):
if all(isinstance(op, PauliOp) and not np.any(op.primitive.x) for op in operator.oplist):
return True
elif isinstance(operator, ListOp):
return all(operator.traverse(_check_is_diagonal))
# cannot efficiently check if a operator is diagonal, converting to matrix
matrix = operator.to_matrix()
return np.all(matrix == np.diag(np.diagonal(matrix)))