Spaces:
Sleeping
Sleeping
File size: 3,403 Bytes
761d149 9641d1d feb1b1c 9641d1d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 |
from __future__ import annotations
import hashlib
from datetime import date
from typing import TYPE_CHECKING, Final
import numpy as np
from redstack.ports.rng import EntropyDisabledError
#: Width (bytes) of the sha256 prefix folded into a numpy seed (64-bit).
_SUBSEED_BYTES: Final[int] = 8
def _derive_subseed(seed: int, label: str) -> int:
"""Deterministically fold ``(seed, label)`` into a stable 64-bit sub-seed."""
digest = hashlib.sha256(f"{seed}:{label}".encode("utf-8")).digest()
return int.from_bytes(digest[:_SUBSEED_BYTES], "big")
class OfflineEntropy:
"""Seeded, labeled, reproducible entropy for the offline pipeline."""
__slots__ = ("_seed", "_as_of")
def __init__(self, seed: int, as_of: date) -> None:
"""Bind the run seed and the fixed reference date.
Args:
seed: The run seed from config.
as_of: The fixed reference date from config (the only clock).
"""
self._seed: Final[int] = seed
self._as_of: Final[date] = as_of
@property
def seed(self) -> int:
"""The run seed."""
return self._seed
def as_of(self) -> date:
"""The fixed reference date from config."""
return self._as_of
def derive(self, label: str) -> int:
"""Return a stable sub-seed deterministically derived from ``(seed, label)``."""
return _derive_subseed(self._seed, label)
def numpy_generator(self, label: str) -> np.random.Generator:
"""Return a seeded PCG64 ``Generator`` for the named substream."""
return np.random.default_rng(self.derive(label))
class OnlineEntropy:
"""RNG-disabled entropy for the online pipeline: ``as_of`` only.
Any attempt to draw randomness raises :class:`EntropyDisabledError`,
enforcing the online RNG-free guarantee (ties break by ``candidate_id``).
"""
__slots__ = ("_seed", "_as_of")
def __init__(self, seed: int, as_of: date) -> None:
"""Bind the recorded seed (provenance only) and the fixed reference date."""
self._seed: Final[int] = seed
self._as_of: Final[date] = as_of
@property
def seed(self) -> int:
"""The recorded run seed (audit/provenance; never used to draw randomness)."""
return self._seed
def as_of(self) -> date:
"""The fixed reference date from config."""
return self._as_of
def derive(self, label: str) -> int:
"""Always raise: online randomness is forbidden.
Raises:
EntropyDisabledError: online RNG is disabled.
"""
raise EntropyDisabledError(
f"online RNG is disabled; derive({label!r}) is not permitted"
)
def numpy_generator(self, label: str) -> np.random.Generator:
"""Always raise: online randomness is forbidden.
Raises:
EntropyDisabledError: online RNG is disabled.
"""
raise EntropyDisabledError(
f"online RNG is disabled; numpy_generator({label!r}) is not permitted"
)
if TYPE_CHECKING:
from redstack.ports.rng import DeterministicEntropyPort
# Compile-time structural conformance to the frozen port surface.
_OFFLINE_CONFORMANCE: type[DeterministicEntropyPort] = OfflineEntropy
_ONLINE_CONFORMANCE: type[DeterministicEntropyPort] = OnlineEntropy
__all__: tuple[str, ...] = ("OfflineEntropy", "OnlineEntropy") |