From 1c38596422b00289e9baa96c32d9d3468a7cbac9 Mon Sep 17 00:00:00 2001 From: dev-belly Date: Fri, 2 Oct 2026 17:45:40 +0200 Subject: [PATCH] Handle finite large feature values and overflow-safe rolling aggregates --- docs/DATA_CONTRACT.md | 4 ++-- docs/ROLLING.md | 8 ++++++-- src/pitbridge/core.py | 21 +++++++++++++++++--- src/pitbridge/rolling.py | 43 +++++++++++++++++++++++++++++----------- tests/test_rolling.py | 29 +++++++++++++++++++++++++++ tests/test_temporal.py | 14 +++++++++++++ 6 files changed, 100 insertions(+), 19 deletions(-) diff --git a/docs/DATA_CONTRACT.md b/docs/DATA_CONTRACT.md index 6c9c70e..469b12a 100644 --- a/docs/DATA_CONTRACT.md +++ b/docs/DATA_CONTRACT.md @@ -12,10 +12,10 @@ The JSON root requires `observations`, `decisions` and `specs` arrays. Optional | `published_at` | Timezone-aware instant, no earlier than event time. | | `ingested_at` | Timezone-aware instant, no earlier than event time. | | `revision` | Integer in `[1, 2^63-1]`; booleans are rejected. | -| `value` | Finite number for an active record; null for a tombstone. | +| `value` | Number representable as finite binary64 for an active record; null for a tombstone. Integers are converted to binary64 for computation, including values outside SQLite's signed-int64 range. | | `deleted` | Boolean, default false. | -The logical version key `(entity_id, source, feature, event_at, revision)` must also be unique. Value units are governed upstream: the demo uses CNY for invoice revenue and bank net inflow, kWh for utility consumption. Unit conversion and numeric aggregation are outside this implementation. +The logical version key `(entity_id, source, feature, event_at, revision)` must also be unique. Value units are governed upstream: the demo uses CNY for invoice revenue and bank net inflow, kWh for utility consumption. Binary64 conversion can round integers above `2^53`; original input values remain in `inputs.json`. This is a feature engine, not an exact monetary ledger. Unit conversion is outside scope; [rolling aggregates](ROLLING.md) have a separate contract. ## Decisions and specs diff --git a/docs/ROLLING.md b/docs/ROLLING.md index 35385ca..6da8640 100644 --- a/docs/ROLLING.md +++ b/docs/ROLLING.md @@ -37,11 +37,15 @@ window durations are rounded to microseconds. Both boundaries are inclusive: 4. Aggregate active observations in the window, exporting their identities, revisions and availability timestamps in `members.csv`. -The SQLite temporal join uses a custom `math.fsum` aggregate. A separate +The SQLite temporal join uses custom compensated sum and mean aggregates. A separate Python enumerator independently selects contributors. These paths share compensated binary64 arithmetic, not their temporal selection logic. This reduces cancellation error but is not exact decimal accounting; -non-finite outputs are rejected. Ordinary floating-point summation can +non-finite outputs are rejected. If `math.fsum` overflows while accumulating, +an exact rational sum of the already-converted binary64 inputs checks the final +sum or mean. Thus two `1e308` observations have a valid mean of `1e308`, +although their sum is rejected. This fallback also preserves finite sums after +large cancellation. Ordinary floating-point summation can lose small terms when large values cancel; see the [SQLite aggregate documentation](https://www.sqlite.org/lang_aggfunc.html). diff --git a/src/pitbridge/core.py b/src/pitbridge/core.py index f4c7041..f2275c6 100644 --- a/src/pitbridge/core.py +++ b/src/pitbridge/core.py @@ -49,8 +49,15 @@ def __post_init__(self): if self.deleted: if self.value is not None: raise ValueError("a tombstone must have value=null") - elif isinstance(self.value, bool) or not isinstance(self.value, (int, float)) or not math.isfinite(self.value): - raise ValueError("active observations require a finite numeric value") + else: + if isinstance(self.value, bool) or not isinstance(self.value, (int, float)): + raise ValueError("active observations require a finite numeric value") + try: + finite = math.isfinite(float(self.value)) + except OverflowError: + finite = False + if not finite: + raise ValueError("active observations require a finite binary64 numeric value") if self.published_at < self.event_at or self.ingested_at < self.event_at: raise ValueError("publication and ingestion cannot precede the observation event") @@ -122,6 +129,14 @@ def _instant(value: str) -> int: return (delta.days * 86400 + delta.seconds) * 1_000_000 + delta.microseconds +def _sql_observation(row: Observation) -> tuple: + fields = asdict(row) + # Python integers bind as SQLite INTEGER even when the column is REAL. + # Match the reference engine's binary64 value contract explicitly. + fields["value"] = None if row.deleted else float(row.value) + return tuple(fields.values()) + (row.available_at, _instant(row.event_at), _instant(row.available_at)) + + SQL = """ WITH eligible AS ( SELECT d.decision_id, o.*, @@ -179,7 +194,7 @@ def build_snapshot(observations, decisions, specs) -> list[Snapshot]: CREATE TABLE specs(source TEXT,feature TEXT,max_age_us INTEGER); CREATE INDEX temporal_lookup ON observations(entity_id,source,feature,event_us,available_us); """) - db.executemany("INSERT INTO observations VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", [tuple(asdict(r).values()) + (r.available_at, _instant(r.event_at), _instant(r.available_at)) for r in obs]) + db.executemany("INSERT INTO observations VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", [_sql_observation(r) for r in obs]) db.executemany("INSERT INTO decisions VALUES (?,?,?,?)", [(r.decision_id, r.entity_id, r.decision_at, _instant(r.decision_at)) for r in dec]) db.executemany("INSERT INTO specs VALUES (?,?,?)", [(r.source, r.feature, None if r.max_age_days is None else round(r.max_age_days * 86400 * 1e6)) for r in features]) return [Snapshot(*row) for row in db.execute(SQL)] diff --git a/src/pitbridge/rolling.py b/src/pitbridge/rolling.py index 67afb8a..015c4d5 100644 --- a/src/pitbridge/rolling.py +++ b/src/pitbridge/rolling.py @@ -1,10 +1,11 @@ """Availability-aware rolling aggregates with every contributing revision retained.""" from dataclasses import asdict, dataclass +from fractions import Fraction import math import sqlite3 -from .core import Decision, FeatureSpec, Observation, _identifier, _instant, validate +from .core import Decision, FeatureSpec, Observation, _identifier, _instant, _sql_observation, validate @dataclass(frozen=True) @@ -71,8 +72,26 @@ def validate_rolling(observations, decisions, specs): return observations, decisions, sorted(specs, key=lambda spec: spec.name) +def _aggregate(values, *, mean=False): + try: + value = math.fsum(values) + except OverflowError: + # The partial sum can overflow even when the final sum or mean fits. + # Exact ratios of the binary64 inputs are a rare, bounded fallback; + # this does not turn upstream decimal amounts into exact accounting. + exact = sum((Fraction.from_float(value) for value in values), Fraction()) + if mean: + exact /= len(values) + try: + return float(exact) + except OverflowError as error: + raise ValueError("rolling aggregate is outside the finite binary64 range") from error + return value / len(values) if mean else value + + class _PreciseSum: """Compensated binary64 summation; this is not exact decimal accounting.""" + mean = False def __init__(self): self.values = [] @@ -82,11 +101,15 @@ def step(self, value): def finalize(self): try: - return math.fsum(self.values) if self.values else None - except OverflowError: + return _aggregate(self.values, mean=self.mean) if self.values else None + except ValueError: return math.inf # Rejected by the finite-output check below. +class _PreciseMean(_PreciseSum): + mean = True + + CTE = """ WITH eligible AS ( SELECT d.decision_id, o.*, @@ -111,6 +134,7 @@ def build_rolling(observations, decisions, specs): observations, decisions, specs = validate_rolling(observations, decisions, specs) with sqlite3.connect(":memory:") as db: db.create_aggregate("precise_sum", 1, _PreciseSum) + db.create_aggregate("precise_mean", 1, _PreciseMean) db.executescript(""" CREATE TABLE observations(record_id TEXT,entity_id TEXT,source TEXT,feature TEXT, event_at TEXT,published_at TEXT,ingested_at TEXT,revision INTEGER,value REAL, @@ -121,7 +145,7 @@ def build_rolling(observations, decisions, specs): CREATE INDEX temporal_lookup ON observations(entity_id,source,feature,event_us,available_us); """) db.executemany("INSERT INTO observations VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", [ - tuple(asdict(row).values()) + (row.available_at, _instant(row.event_at), _instant(row.available_at)) + _sql_observation(row) for row in observations ]) db.executemany("INSERT INTO decisions VALUES (?,?,?,?)", [ @@ -136,8 +160,8 @@ def build_rolling(observations, decisions, specs): s.window_days,s.aggregation, CASE WHEN count(a.record_id)=0 THEN NULL WHEN s.aggregation='count' THEN count(a.record_id) - WHEN s.aggregation='sum' THEN precise_sum(CASE WHEN s.aggregation='count' THEN NULL ELSE a.value END) - ELSE precise_sum(CASE WHEN s.aggregation='count' THEN NULL ELSE a.value END)/count(a.record_id) END, + WHEN s.aggregation='sum' THEN precise_sum(CASE WHEN s.aggregation='sum' THEN a.value END) + ELSE precise_mean(CASE WHEN s.aggregation='mean' THEN a.value END) END, count(a.record_id), CASE WHEN count(a.record_id)>0 THEN 'selected' WHEN NOT EXISTS (SELECT 1 FROM observations o @@ -188,12 +212,7 @@ def reference_rolling(observations, decisions, specs): if spec.aggregation == "count": value = len(active) else: - try: - value = math.fsum(float(row.value) for row in active) - except OverflowError as error: - raise ValueError("rolling aggregate is outside the finite binary64 range") from error - if spec.aggregation == "mean": - value /= len(active) + value = _aggregate([float(row.value) for row in active], mean=spec.aggregation == "mean") features.append(RollingFeature(decision.decision_id, decision.entity_id, decision.decision_at, spec.name, spec.source, spec.feature, spec.window_days, spec.aggregation, value, len(active), status)) diff --git a/tests/test_rolling.py b/tests/test_rolling.py index 761ad7d..12c120a 100644 --- a/tests/test_rolling.py +++ b/tests/test_rolling.py @@ -26,6 +26,35 @@ def spec(name="cash_30d", days=30, aggregation="sum"): class RollingTests(unittest.TestCase): + def test_large_integer_values_match_the_temporal_oracle(self): + rows = [observation("large", "2026-01-01T00:00:00Z", 10**20)] + data = (rows, [Decision("D", "A", "2026-01-03T00:00:00Z")], + [spec("sum"), spec("mean", aggregation="mean"), spec("count", aggregation="count")]) + actual = build_rolling(*data) + self.assertEqual(actual, reference_rolling(*data)) + self.assertEqual({row.name: row.value for row in actual[0]}, + {"sum": 1e20, "mean": 1e20, "count": 1}) + + def test_finite_mean_survives_unrepresentable_intermediate_sum(self): + for value in (1e308, -1e308): + rows = [observation(str(i), f"2026-01-0{i+1}T00:00:00Z", value) for i in range(2)] + data = (rows, [Decision("D", "A", "2026-01-04T00:00:00Z")], + [spec(aggregation="mean")]) + with self.subTest(value=value): + actual = build_rolling(*data) + self.assertEqual(actual[0][0].value, value) + self.assertEqual(actual, reference_rolling(*data)) + + def test_finite_sum_survives_intermediate_overflow_and_cancellation(self): + for values in ((1e308, 1e308, -1e308), (-1e308, 1e308, 1e308)): + rows = [observation(str(i), f"2026-01-0{i+1}T00:00:00Z", value) + for i, value in enumerate(values)] + data = (rows, [Decision("D", "A", "2026-01-04T00:00:00Z")], [spec()]) + with self.subTest(values=values): + actual = build_rolling(*data) + self.assertEqual(actual[0][0].value, 1e308) + self.assertEqual(actual, reference_rolling(*data)) + def test_hand_auditable_demo_sums_and_members(self): data = decode_rolling(rolling_demo_inputs()) features, members = build_rolling(*data) diff --git a/tests/test_temporal.py b/tests/test_temporal.py index 5c23ee7..46901a2 100644 --- a/tests/test_temporal.py +++ b/tests/test_temporal.py @@ -19,6 +19,20 @@ def snapshot(records, at="2026-01-05T00:00:00Z", age=None, entity="A"): class TemporalTests(unittest.TestCase): + def test_large_integer_values_use_binary64_in_both_engines(self): + for value in (10**20, -(10**20), 10**308): + with self.subTest(value=value): + rows = [observation(value=value)] + decisions = [Decision("D", "A", "2026-01-05T00:00:00Z")] + specs = [FeatureSpec("tax", "revenue")] + actual = build_snapshot(rows, decisions, specs) + self.assertEqual(actual, reference_snapshot(rows, decisions, specs)) + self.assertEqual(actual[0].value, float(value)) + + def test_integer_outside_binary64_range_is_a_contract_error(self): + with self.assertRaisesRegex(ValueError, "finite"): + observation(value=10**400) + def test_historical_revision_cannot_change_old_decision(self): revision = observation(record_id="r2", revision=2, value=90, published_at="2026-01-10T00:00:00Z", ingested_at="2026-01-10T00:00:00Z") self.assertEqual(snapshot([observation(), revision]).value, 10)