Reservoir-sample the current dataset. Returns (sample_records, k)
(
self,
mode: Literal["manual", "proportion", "mean"] = "manual",
*,
k: int | None = None,
conf_level: float = 0.95,
margin: float = 0.03,
p: float = 0.5,
sigma: float = 10.0,
)
| 104 | return math.ceil(n0 / (1 + (n0 - 1) / N)) |
| 105 | |
| 106 | def rsample( |
| 107 | self, |
| 108 | mode: Literal["manual", "proportion", "mean"] = "manual", |
| 109 | *, |
| 110 | k: int | None = None, |
| 111 | conf_level: float = 0.95, |
| 112 | margin: float = 0.03, |
| 113 | p: float = 0.5, |
| 114 | sigma: float = 10.0, |
| 115 | ) -> Tuple[List[Dict[str, Any]], int]: |
| 116 | """ |
| 117 | Reservoir-sample the current dataset. |
| 118 | |
| 119 | Returns (sample_records, k) |
| 120 | """ |
| 121 | N = self.count() |
| 122 | |
| 123 | # Decide sample size ------------------------------------------------- |
| 124 | if mode == "manual": |
| 125 | if not k or k <= 0: |
| 126 | raise ValueError("manual mode requires a positive integer k") |
| 127 | elif mode == "proportion": |
| 128 | k = self.sample_size_proportion(N, conf_level, margin, p) |
| 129 | elif mode == "mean": |
| 130 | k = self.sample_size_mean(N, conf_level, margin, sigma) |
| 131 | else: |
| 132 | raise ValueError('mode must be "manual", "proportion", or "mean"') |
| 133 | |
| 134 | self.logger.info(f"Sampling k={k} from N={N} (mode={mode})") |
| 135 | |
| 136 | # Simple reservoir algorithm ---------------------------------------- |
| 137 | reservoir: List[Dict[str, Any]] = [] |
| 138 | for t, rec in enumerate(self.fetch_stream(), start=1): |
| 139 | if t <= k: |
| 140 | reservoir.append(rec) |
| 141 | else: |
| 142 | j = random.randrange(t) |
| 143 | if j < k: # replace with probability k/t |
| 144 | reservoir[j] = rec |
| 145 | |
| 146 | return reservoir, k |
no test coverage detected