Add kalman

Author: lsbardel

docs/api/ta/kalman.md

@@ -0,0 +1,3 @@
+# Kalman Filter
+
+::: quantflow.ta.KalmanFilter

docs/api/ta/supersmoother.md

@@ -11,4 +11,4 @@ Reference:
     Ehlers, J. (2013). "Cycle Analytics for Traders"
 
 
-::: quantflow.ta.supersmoother.SuperSmoother
+::: quantflow.ta.SuperSmoother

mkdocs.yml

@@ -56,6 +56,7 @@ nav:
           - api/ta/index.md
           - OHLC: api/ta/ohlc.md
           - Paths: api/ta/paths.md
+          - Kalman Filter: api/ta/kalman.md
           - Super Smoother: api/ta/supersmoother.md
       - Utils:
           - api/utils/index.md

quantflow/ta/__init__.py

@@ -0,0 +1,5 @@
+from .supersmoother import SuperSmoother
+from .kalman import KalmanFilter
+
+
+__all__ = ["SuperSmoother", "KalmanFilter"]

quantflow/ta/kalman.py

@@ -0,0 +1,72 @@
+from typing_extensions import Annotated, Doc
+from pydantic import BaseModel, Field, PrivateAttr
+
+
+class KalmanFilter(BaseModel):
+    r"""One-dimensional Kalman filter for time series data.
+
+    This implementation uses a simple 1D state-space model:
+
+    $$
+    \begin{align}
+        x_t &= x_{t-1} + w_t, \quad w_t \sim \mathcal{N}(0, Q) \\
+        z_t &= x_t + v_t, \quad v_t \sim \mathcal{N}(0, R)
+    \end{align}
+    $$
+
+    The Kalman filter estimates the hidden state $x_t$ given noisy measurements $z_t$.
+    The ratio $Q/R$ determines the smoothing behavior.
+    """
+
+    R: float = Field(default=1.0, gt=0.0, description="Measurement noise covariance")
+    Q: float = Field(default=0.01, gt=0.0, description="Process noise covariance")
+
+    _x: float | None = PrivateAttr(default=None)  # State estimate
+    _P: float = PrivateAttr(default=1.0)  # Error covariance
+    _K: float = PrivateAttr(default=0.0)  # Kalman Gain
+
+    def value(self) -> float | None:
+        """Get the most recent smoothed value (state estimate), if available."""
+        return self._x
+
+    def update(
+        self,
+        value: Annotated[float, Doc("New noisy measurement to update the filter")],
+    ) -> float:
+        """Update the filter with a new value and return the smoothed result."""
+        # Initialize on first update
+        if self._x is None:
+            self._x = value
+            self._P = self.R
+            return value
+
+        # Prediction step
+        # x_pred = x_prev (Random walk model)
+        # P_pred = P_prev + Q
+        x_pred = self._x
+        P_pred = self._P + self.Q
+
+        # Update step
+        # K = P_pred / (P_pred + R)
+        # x_new = x_pred + K * (measurement - x_pred)
+        # P_new = (1 - K) * P_pred
+        K = P_pred / (P_pred + self.R)
+        x_new = x_pred + K * (value - x_pred)
+        P_new = (1 - K) * P_pred
+
+        # Update state
+        self._x = x_new
+        self._P = P_new
+        self._K = K
+
+        return x_new
+
+    @property
+    def error_covariance(self) -> float:
+        """Current estimated error covariance."""
+        return self._P
+
+    @property
+    def kalman_gain(self) -> float:
+        """Most recent Kalman gain."""
+        return self._K

quantflow/ta/supersmoother.py

@@ -6,12 +6,30 @@
 
 
 class SuperSmoother(BaseModel):
-    """SuperSmoother filter for time series data.
+    r"""SuperSmoother filter for time series data.
 
     This implementation uses a two-pole Butterworth filter with adaptive smoothing.
-
     The SuperSmoother filter is designed to remove high-frequency noise while
     preserving the underlying trend with minimal lag.
+    The filter is defined by the following recurrence relation:
+
+    $$
+    y_t = c_1 \frac{x_t + x_{t-1}}{2} + c_2 y_{t-1} + c_3 y_{t-2}
+    $$
+
+    where the coefficients are calculated as:
+
+    $$
+    \begin{align}
+        \lambda &= \frac{\pi \sqrt{2}}{N} \\
+        a &= \exp(-\lambda) \\
+        c_2 &= 2 a \cos(\lambda) \\
+        c_3 &= -a^2 \\
+        c_1 &= 1 - c_2 - c_3
+    \end{align}
+    $$
+
+    where $N$ is the period.
 
     ## Example