Resolve #78: migrate domain models to Pydantic v2

docs/BASICS.md

@@ -19,6 +19,21 @@
 
 # Basics
 
+## Python persistence API
+
+`Series`, `AnalyzedSeries`, metrics, statistics, and change-point groups are Pydantic v2
+domain models. Persist an analysis with `analysis.model_dump(mode="json")`, and restore it
+with `AnalyzedSeries.model_validate(payload)`. The JSON document remains compatible with
+previous releases, including legacy string metric units and older flat change-point entries.
+
+`to_json()` and `from_json()` remain as deprecated wrappers for the current release and
+emit `DeprecationWarning`. Migrate callers before the next breaking release, when those
+wrappers and the `ChangePointSerializer` compatibility shim will be removed.
+
+The old `otava.serialization.*Model` imports also remain available for this release as
+flat-persistence compatibility models. New code should use the domain models and
+`AnalyzedSeries.model_validate()` instead.
+
 ## Listing Available Tests
 
 ```

otava/analysis.py

@@ -16,7 +16,6 @@
 # under the License.
 
 import copy
-from dataclasses import dataclass, replace
 from typing import List, Optional, Sequence, SupportsFloat, Tuple
 
 from scipy.stats import ttest_ind_from_stats
@@ -36,7 +35,6 @@
 )
 
 
-@dataclass
 class TTestStats(BaseStats):
     """
     Statistics related to the calculation of a two-sided Student's T-test.
@@ -47,8 +45,7 @@ class TTestStats(BaseStats):
     degrees_of_freedom: int = 0
 
     def copy(self):
-        # replace() preserves the subclass
-        return replace(self)
+        return self.model_copy(deep=True)
 
     def to_json(self):
         obj = super().to_json()

otava/bigquery.py

@@ -28,7 +28,7 @@
     ScalarQueryParameter = Any
 
 from otava._optional import import_optional_dependency
-from otava.change_point_divisive.base import ChangePointGroup, ChangePointSerializer
+from otava.change_point_divisive.base import ChangePointGroup
 from otava.test_config import BigQueryTestConfig
 
 
@@ -106,7 +106,7 @@ def insert_change_point(
         attributes: Dict,
         change_point_group: ChangePointGroup,
     ):
-        change_point = ChangePointSerializer(change_point_group[metric_name])
+        change_point = change_point_group[metric_name]
         kwargs = {**attributes, **{test.time_column: datetime.utcfromtimestamp(change_point_group.time)}}
         update_stmt = test.update_stmt.format(
             metric=metric_name,

otava/change_point_divisive/base.py

@@ -45,24 +45,31 @@
           .pivot()                         < - - >  .pivot()
 """
 
-from dataclasses import dataclass, fields, replace
 from datetime import datetime, timezone
 from typing import Dict, Generic, List, Optional, Sequence, SupportsFloat, TypeVar
+from warnings import warn
 
 import numpy as np
 from numpy.typing import NDArray
+from pydantic import BaseModel, ConfigDict, field_validator, model_validator
 
+JsonScalar = str | int | float | bool | None
 
-@dataclass
-class CandidateChangePoint:
+
+class DomainModel(BaseModel):
+    """Shared validation and assignment semantics for domain value objects."""
+
+    model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True)
+
+
+class CandidateChangePoint(DomainModel):
     """Candidate for a change point. The point that maximizes Q-hat function on [start:end+1] slice"""
 
     index: int
     qhat: float
 
 
-@dataclass
-class BaseStats:
+class BaseStats(DomainModel):
     """Abstract statistics class for change point. Implementation depends on the statistical test."""
 
     # The pvalue for this change point. Exact value depends on the algorithm that was used.
@@ -99,7 +106,7 @@ def calculate(left: Sequence[SupportsFloat], right: Sequence[SupportsFloat], pva
 
     def copy(self):
         """Return a copy of these statistics, preserving the concrete (sub)class."""
-        return replace(self)
+        return self.model_copy(deep=True)
 
     def forward_rel_change(self, value_if_nan=0):
         """Relative change from left to right"""
@@ -138,6 +145,7 @@ def stddev_after(self):
         return self.std_2
 
     def to_json(self):
+        warn("BaseStats.to_json() is deprecated; use model_dump(mode='json')", DeprecationWarning, stacklevel=2)
         return {
             "forward_change_percent": f"{self.forward_change_percent():-0f}",
             "magnitude": f"{self.change_magnitude():-0f}",
@@ -153,7 +161,6 @@ def to_json(self):
 GenericStats = TypeVar("GenericStats", bound=BaseStats)
 
 
-@dataclass
 class ChangePoint(CandidateChangePoint, Generic[GenericStats]):
     """
     ChangePoint class.
@@ -182,7 +189,7 @@ def copy(self):
 
         :return: A deep copy of self, recursively calls also stats.copy().
         """
-        return ChangePoint(
+        return self.__class__(
             index=self.index, qhat=self.qhat, stats=self.stats.copy(), metric=self.metric
         )
 
@@ -198,11 +205,35 @@ def from_candidate(
 
     def to_candidate(self) -> CandidateChangePoint:
         """Downgrades Change Point to a Candidate Change Point. Used to recompute stats for Weak Change Points."""
-        data = {f.name: getattr(self, f.name) for f in fields(CandidateChangePoint)}
-        return CandidateChangePoint(**data)
+        return CandidateChangePoint(index=self.index, qhat=self.qhat)
 
     def to_json(self):
-        return ChangePointSerializer(self).to_json()
+        warn("ChangePoint.to_json() is deprecated; use model_dump(mode='json')", DeprecationWarning, stacklevel=2)
+        return self.persistence_dict()
+
+    def forward_change_percent(self) -> float:
+        return self.stats.forward_change_percent()
+
+    def backward_change_percent(self) -> float:
+        return self.stats.backward_change_percent()
+
+    def magnitude(self) -> float:
+        return self.stats.change_magnitude()
+
+    def persistence_dict(self) -> dict:
+        """The historical flat change-point document used inside persisted analyses."""
+        return {
+            "metric": self.metric,
+            "index": int(self.index),
+            "qhat": self.qhat,
+            "forward_change_percent": self.forward_change_percent(),
+            "magnitude": self.magnitude(),
+            "mean_before": self.stats.mean_before(),
+            "stddev_before": self.stats.stddev_before(),
+            "mean_after": self.stats.mean_after(),
+            "stddev_after": self.stats.stddev_after(),
+            "pvalue": self.stats.pvalue,
+        }
 
 
 class ChangePointSerializer(ChangePoint):
@@ -214,10 +245,12 @@ class ChangePointSerializer(ChangePoint):
     """
 
     def __init__(self, cp: ChangePoint[GenericStats]):
-        self.stats = cp.stats
-        self.index = cp.index
-        self.qhat = cp.qhat
-        self.metric = cp.metric
+        warn(
+            "ChangePointSerializer is deprecated; use ChangePoint's model-derived values instead",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(stats=cp.stats, index=cp.index, qhat=cp.qhat, metric=cp.metric)
 
     def forward_change_percent(self) -> float:
         return self.stats.forward_rel_change() * 100.0
@@ -244,22 +277,10 @@ def pvalue(self):
         return self.stats.pvalue
 
     def to_json(self):
-        return {
-            "metric": self.metric,
-            "index": int(self.index),
-            "qhat": self.qhat,
-            "forward_change_percent": self.forward_change_percent(),
-            "magnitude": self.magnitude(),
-            "mean_before": self.mean_before(),
-            "stddev_before": self.stddev_before(),
-            "mean_after": self.mean_after(),
-            "stddev_after": self.stddev_after(),
-            "pvalue": self.pvalue(),
-        }
+        return self.persistence_dict()
 
 
-@dataclass
-class ChangePointGroup:
+class ChangePointGroup(DomainModel):
     """
     A group of change points on multiple metrics, at the same point in time.
 
@@ -275,15 +296,30 @@ class ChangePointGroup:
     :param changes: For each metric that has a change point at this time(stamp), the ChangePoint object.
     """
 
-    time: float
-    attributes: Dict[str, str]
+    time: int | float
+    attributes: Dict[str, JsonScalar]
     # ChangePointGroup.changes.keys() stores the set of metrics that were used at this ChangePointGroup.time.
     changes: Dict[str, ChangePoint]
 
+    @field_validator("time")
+    @classmethod
+    def validate_time(cls, value):
+        if isinstance(value, bool) or not isinstance(value, (int, float)):
+            raise ValueError("time must be numeric")
+        return value
+
+    @model_validator(mode="after")
+    def validate_metric_mapping(self):
+        for metric, change in self.changes.items():
+            if change.metric is None:
+                change.metric = metric
+            elif change.metric != metric:
+                raise ValueError(f"metric field is not internally consistent. {metric} != {change.metric}")
+        return self
+
     def to_json(self):
-        changes = []
-        for metric, cp in self.changes.items():
-            changes.append(cp.to_json())
+        warn("ChangePointGroup.to_json() is deprecated; use model_dump(mode='json')", DeprecationWarning, stacklevel=2)
+        changes = [cp.persistence_dict() for cp in self.changes.values()]
 
         return {
             "time": self.time,
@@ -294,8 +330,7 @@ def to_json(self):
     def copy(self):
         new_attributes = {k: v for k, v in self.attributes.items()}
         new_changes = {metric: cp.copy() for metric, cp in self.changes.items()}
-        new_obj = ChangePointGroup(time=self.time, attributes=new_attributes, changes=new_changes)
-        return new_obj
+        return ChangePointGroup(time=self.time, attributes=new_attributes, changes=new_changes)
 
     def __getitem__(self, metric):
         return self.changes[metric]

otava/change_point_divisive/significance_test.py

@@ -15,11 +15,11 @@
 # specific language governing permissions and limitations
 # under the License.
 
-from dataclasses import dataclass, replace
 from typing import List, Optional, Type
 
 import numpy as np
 from numpy.typing import NDArray
+from pydantic import field_validator
 
 from otava.change_point_divisive.base import (
     BaseStats,
@@ -30,17 +30,23 @@
 )
 
 
-@dataclass
 class PermutationStats(BaseStats):
     '''Statistics for permutation significance test'''
     permuted_qhats: NDArray
     extreme_qhat_perm: int
     n_perm: int
 
+    @field_validator("permuted_qhats", mode="before")
+    @classmethod
+    def unwrap_single_array_tuple(cls, value):
+        # Older callers occasionally passed ``np.array(...),`` by accident;
+        # dataclasses accepted it, so retain that input shape during migration.
+        if isinstance(value, tuple) and len(value) == 1 and isinstance(value[0], np.ndarray):
+            return value[0]
+        return value
+
     def copy(self):
-        c = replace(self)
-        c.permuted_qhats = self.permuted_qhats.copy()
-        return c
+        return self.model_copy(deep=True)
 
     def to_json(self):
         obj = super().to_json()

otava/postgres.py

@@ -27,7 +27,7 @@
     Connection = Any
 
 from otava._optional import import_optional_dependency
-from otava.change_point_divisive.base import ChangePointGroup, ChangePointSerializer
+from otava.change_point_divisive.base import ChangePointGroup
 from otava.test_config import PostgresTestConfig
 
 
@@ -102,7 +102,7 @@ def insert_change_point(
         change_point_group: ChangePointGroup,
     ):
         cursor = self.__get_conn().cursor()
-        change_point = ChangePointSerializer(change_point_group[metric_name])
+        change_point = change_point_group[metric_name]
         kwargs = {**attributes, **{test.time_column: datetime.utcfromtimestamp(change_point_group.time)}}
         update_stmt = test.update_stmt.format(metric=metric_name, **kwargs)
         cursor.execute(

otava/report.py

@@ -21,7 +21,7 @@
 
 from tabulate import tabulate
 
-from otava.change_point_divisive.base import ChangePoints, ChangePointSerializer
+from otava.change_point_divisive.base import ChangePoints
 from otava.series import Series
 from otava.util import format_timestamp, insert_multiple, remove_common_prefix
 
@@ -87,8 +87,7 @@ def __format_log_annotated(self, test_name: str) -> str:
                 col_width = col_widths[col_index]
                 change = [c for m, c in cpg.changes.items() if m == col_name]
                 if change:
-                    change = ChangePointSerializer(change[0])
-                    change_percent = change.forward_change_percent()
+                    change_percent = change[0].forward_change_percent()
                     separator += "·" * col_width + "  "
                     info += f"{change_percent:+.1f}%".rjust(col_width) + "  "
                 else:
@@ -121,26 +120,24 @@ def __format_change_point_group_json(cpg):
 
     @staticmethod
     def __format_change_point_json(cp):
-        cp = ChangePointSerializer(cp)
         return {
             "metric": cp.metric,
             "index": int(cp.index),
             "forward_change_percent": f"{cp.forward_change_percent():.0f}",
             "backward_change_percent": f"{cp.backward_change_percent():.0f}",
             "magnitude": f"{cp.magnitude():.6f}",
-            "mean_before": f"{cp.mean_before():.6f}",
-            "stddev_before": f"{cp.stddev_before():.6f}",
-            "mean_after": f"{cp.mean_after():.6f}",
-            "stddev_after": f"{cp.stddev_after():.6f}",
-            "pvalue": f"{cp.pvalue():.6f}",
+            "mean_before": f"{cp.stats.mean_before():.6f}",
+            "stddev_before": f"{cp.stats.stddev_before():.6f}",
+            "mean_after": f"{cp.stats.mean_after():.6f}",
+            "stddev_after": f"{cp.stats.stddev_after():.6f}",
+            "pvalue": f"{cp.stats.pvalue:.6f}",
         }
 
     def __format_regressions_only(self, test_name: str) -> str:
         output = []
         for cpg in self.__change_points:
             regressions = []
             for metric_name, cp in cpg.changes.items():
-                cp = ChangePointSerializer(cp)
                 metric = self.__series.metrics[metric_name]
                 if metric.direction * cp.forward_change_percent() < 0:
                     regressions.append(

otava/serialization.py

@@ -15,13 +15,24 @@
 # specific language governing permissions and limitations
 # under the License.
 
+"""Deprecated compatibility models for the pre-#78 persistence API."""
+
 from datetime import datetime
 from typing import Dict, List, Optional
 
 from pydantic import BaseModel, ConfigDict
 
 JsonScalar = str | int | float | bool | None
 
+__all__ = [
+    "AnalysisOptionsModel",
+    "AnalyzedSeriesModel",
+    "ChangePointGroupModel",
+    "ChangePointModel",
+    "JsonScalar",
+    "MetricModel",
+]
+
 
 class AnalysisOptionsModel(BaseModel):
     model_config = ConfigDict(extra="forbid", validate_assignment=True)