beta.py 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119
  1. # mypy: allow-untyped-defs
  2. from typing import Optional, Union
  3. import torch
  4. from torch import Tensor
  5. from torch.distributions import constraints
  6. from torch.distributions.dirichlet import Dirichlet
  7. from torch.distributions.exp_family import ExponentialFamily
  8. from torch.distributions.utils import broadcast_all
  9. from torch.types import _Number, _size
  10. __all__ = ["Beta"]
  11. class Beta(ExponentialFamily):
  12. r"""
  13. Beta distribution parameterized by :attr:`concentration1` and :attr:`concentration0`.
  14. Example::
  15. >>> # xdoctest: +IGNORE_WANT("non-deterministic")
  16. >>> m = Beta(torch.tensor([0.5]), torch.tensor([0.5]))
  17. >>> m.sample() # Beta distributed with concentration concentration1 and concentration0
  18. tensor([ 0.1046])
  19. Args:
  20. concentration1 (float or Tensor): 1st concentration parameter of the distribution
  21. (often referred to as alpha)
  22. concentration0 (float or Tensor): 2nd concentration parameter of the distribution
  23. (often referred to as beta)
  24. """
  25. # pyrefly: ignore [bad-override]
  26. arg_constraints = {
  27. "concentration1": constraints.positive,
  28. "concentration0": constraints.positive,
  29. }
  30. support = constraints.unit_interval
  31. has_rsample = True
  32. def __init__(
  33. self,
  34. concentration1: Union[Tensor, float],
  35. concentration0: Union[Tensor, float],
  36. validate_args: Optional[bool] = None,
  37. ) -> None:
  38. if isinstance(concentration1, _Number) and isinstance(concentration0, _Number):
  39. concentration1_concentration0 = torch.tensor(
  40. [float(concentration1), float(concentration0)]
  41. )
  42. else:
  43. concentration1, concentration0 = broadcast_all(
  44. concentration1, concentration0
  45. )
  46. concentration1_concentration0 = torch.stack(
  47. [concentration1, concentration0], -1
  48. )
  49. self._dirichlet = Dirichlet(
  50. concentration1_concentration0, validate_args=validate_args
  51. )
  52. super().__init__(self._dirichlet._batch_shape, validate_args=validate_args)
  53. def expand(self, batch_shape, _instance=None):
  54. new = self._get_checked_instance(Beta, _instance)
  55. batch_shape = torch.Size(batch_shape)
  56. new._dirichlet = self._dirichlet.expand(batch_shape)
  57. super(Beta, new).__init__(batch_shape, validate_args=False)
  58. new._validate_args = self._validate_args
  59. return new
  60. @property
  61. def mean(self) -> Tensor:
  62. return self.concentration1 / (self.concentration1 + self.concentration0)
  63. @property
  64. def mode(self) -> Tensor:
  65. return self._dirichlet.mode[..., 0]
  66. @property
  67. def variance(self) -> Tensor:
  68. total = self.concentration1 + self.concentration0
  69. return self.concentration1 * self.concentration0 / (total.pow(2) * (total + 1))
  70. def rsample(self, sample_shape: _size = ()) -> Tensor:
  71. return self._dirichlet.rsample(sample_shape).select(-1, 0)
  72. def log_prob(self, value):
  73. if self._validate_args:
  74. self._validate_sample(value)
  75. heads_tails = torch.stack([value, 1.0 - value], -1)
  76. return self._dirichlet.log_prob(heads_tails)
  77. def entropy(self):
  78. return self._dirichlet.entropy()
  79. @property
  80. def concentration1(self) -> Tensor:
  81. result = self._dirichlet.concentration[..., 0]
  82. if isinstance(result, _Number):
  83. return torch.tensor([result])
  84. else:
  85. return result
  86. @property
  87. def concentration0(self) -> Tensor:
  88. result = self._dirichlet.concentration[..., 1]
  89. if isinstance(result, _Number):
  90. return torch.tensor([result])
  91. else:
  92. return result
  93. @property
  94. def _natural_params(self) -> tuple[Tensor, Tensor]:
  95. return (self.concentration1, self.concentration0)
  96. # pyrefly: ignore [bad-override]
  97. def _log_normalizer(self, x, y):
  98. return torch.lgamma(x) + torch.lgamma(y) - torch.lgamma(x + y)