Handle finite feature values and overflow-safe rolling aggregates

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
 

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).
 

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)]

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))

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)

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)