Spaces:
Running
Running
File size: 2,653 Bytes
ad91e86 | 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 | """Leakage-safe classical baselines for current-state prediction."""
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
def _normalize(counts: np.ndarray) -> np.ndarray:
return counts / counts.sum(axis=-1, keepdims=True).clip(min=1e-12)
@dataclass
class ClassicalBaselines:
state_count: int
category_count: int
smoothing: float = 0.25
def fit(
self,
*,
target_category: np.ndarray,
target_state: np.ndarray,
last_state: np.ndarray,
weekday: np.ndarray,
time_sin_cos: np.ndarray,
) -> None:
self.object_counts = np.full(
(self.category_count, self.state_count), self.smoothing, dtype=np.float64
)
self.markov_counts = np.full(
(self.category_count, self.state_count, self.state_count),
self.smoothing,
dtype=np.float64,
)
self.time_counts = np.full(
(self.category_count, 7, 4, self.state_count), self.smoothing, dtype=np.float64
)
hour = _time_quarter(time_sin_cos)
np.add.at(self.object_counts, (target_category, target_state), 1.0)
np.add.at(self.markov_counts, (target_category, last_state, target_state), 1.0)
np.add.at(self.time_counts, (target_category, weekday, hour, target_state), 1.0)
self.object_probabilities = _normalize(self.object_counts)
self.markov_probabilities = _normalize(self.markov_counts)
self.time_probabilities = _normalize(self.time_counts)
def predict(
self,
method: str,
*,
target_category: np.ndarray,
last_state: np.ndarray,
weekday: np.ndarray,
time_sin_cos: np.ndarray,
) -> np.ndarray:
if method == "last_seen":
probabilities = np.zeros((len(last_state), self.state_count), dtype=np.float64)
probabilities[np.arange(len(last_state)), last_state] = 1.0
return probabilities
if method == "object_frequency":
return self.object_probabilities[target_category]
if method == "markov":
return self.markov_probabilities[target_category, last_state]
if method == "time_frequency":
quarter = _time_quarter(time_sin_cos)
return self.time_probabilities[target_category, weekday, quarter]
raise ValueError(method)
def _time_quarter(time_sin_cos: np.ndarray) -> np.ndarray:
angle = np.arctan2(time_sin_cos[:, 0], time_sin_cos[:, 1])
fraction = np.mod(angle, 2 * np.pi) / (2 * np.pi)
return np.floor(fraction * 4).astype(np.int64).clip(0, 3)
|