feat(math): add finite stochastic processes domain with sigma algebras, conditional expectation, filtrations, and Doob martingales (#1806)

src/jacobian/catalog/builtins.py

@@ -86,6 +86,12 @@
 )
 from jacobian.math.finite_sets._admission import ADMISSIONS as FINITE_SETS_ADMISSIONS
 from jacobian.math.finite_sets._tools import TOOLS as FINITE_SETS_TOOLS
+from jacobian.math.finite_stochastic_processes._admission import (
+    ADMISSIONS as FINITE_STOCHASTIC_PROCESSES_ADMISSIONS,
+)
+from jacobian.math.finite_stochastic_processes._tools import (
+    TOOLS as FINITE_STOCHASTIC_PROCESSES_TOOLS,
+)
 from jacobian.math.finite_state_transducers._admission import (
     ADMISSIONS as FINITE_STATE_TRANSDUCERS_ADMISSIONS,
 )
@@ -292,6 +298,7 @@
     *DIOPHANTINE_APPROXIMATION_TOOLS,
     *COMBINATORICS_TOOLS,
     *FINITE_SETS_TOOLS,
+    *FINITE_STOCHASTIC_PROCESSES_TOOLS,
     *FINITE_FIELDS_TOOLS,
     *LOGIC_TOOLS,
     *SEQUENCES_TOOLS,
@@ -368,6 +375,7 @@
     *FINITE_GAME_THEORY_ADMISSIONS,
     *FINITE_METRIC_SPACES_ADMISSIONS,
     *FINITE_SETS_ADMISSIONS,
+    *FINITE_STOCHASTIC_PROCESSES_ADMISSIONS,
     *FINITE_STATE_TRANSDUCERS_ADMISSIONS,
     *FINITE_TOPOLOGY_ADMISSIONS,
     *FORMAL_POWER_SERIES_ADMISSIONS,

src/jacobian/math/finite_stochastic_processes/__init__.py

@@ -0,0 +1,25 @@
+"""Supported native finite stochastic process API."""
+
+from jacobian.math.finite_stochastic_processes.operations import (
+    conditional_expectation,
+    doob_martingale,
+    filtration_natural,
+    sigma_algebra_from_observation,
+    sigma_algebra_join,
+)
+from jacobian.math.finite_stochastic_processes.values import (
+    FiniteProbabilitySpace,
+    FiniteRandomVariable,
+    FiniteSigmaAlgebra,
+)
+
+__all__ = [
+    "FiniteProbabilitySpace",
+    "FiniteRandomVariable",
+    "FiniteSigmaAlgebra",
+    "conditional_expectation",
+    "doob_martingale",
+    "filtration_natural",
+    "sigma_algebra_from_observation",
+    "sigma_algebra_join",
+]

src/jacobian/math/finite_stochastic_processes/_admission.py

@@ -0,0 +1,33 @@
+"""Owner-local admission decisions for built-in math operations."""
+
+from __future__ import annotations
+
+from jacobian.catalog.admission import AdmissionDecision, OperationAdmission
+
+ADMISSIONS: tuple[OperationAdmission, ...] = (
+    OperationAdmission(
+        "probability.finite_sigma_algebra.from_observation.compute",
+        AdmissionDecision.KEEP,
+        "exact sigma algebra construction from an observation map with equal-value fibers",
+    ),
+    OperationAdmission(
+        "probability.finite_sigma_algebra.join.compute",
+        AdmissionDecision.KEEP,
+        "exact join of two finite sigma algebras as the finest common refinement",
+    ),
+    OperationAdmission(
+        "probability.conditional_expectation.finite.compute",
+        AdmissionDecision.KEEP,
+        "exact block-constant conditional expectation with probability-weighted block averages",
+    ),
+    OperationAdmission(
+        "probability.filtration.natural.compute",
+        AdmissionDecision.KEEP,
+        "exact natural filtration F_t = sigma(Y_0, ..., Y_t) with monotone refinement",
+    ),
+    OperationAdmission(
+        "probability.process.doob_martingale.compute",
+        AdmissionDecision.KEEP,
+        "exact Doob martingale M_t = E[payoff | F_t] with rational-valued conditional expectations",
+    ),
+)

src/jacobian/math/finite_stochastic_processes/_models.py

@@ -0,0 +1,71 @@
+"""Typed wire contracts for finite stochastic process operations."""
+
+from __future__ import annotations
+
+from pydantic import Field
+
+from jacobian._models import StrictModel
+from jacobian.math.finite_stochastic_processes.values import (
+    FiniteProbabilitySpace,
+    FiniteRandomVariable,
+    FiniteSigmaAlgebra,
+)
+
+
+class FromObservationRequest(StrictModel):
+    """Construct a sigma algebra from an observation map."""
+
+    space: FiniteProbabilitySpace
+    observation: tuple[str, ...] = Field(min_length=1)
+
+
+class JoinRequest(StrictModel):
+    """Compute the join of two sigma algebras."""
+
+    sigma1: FiniteSigmaAlgebra
+    sigma2: FiniteSigmaAlgebra
+
+
+class ConditionalExpectationRequest(StrictModel):
+    """Compute E[X | G]."""
+
+    rv: FiniteRandomVariable
+    sigma: FiniteSigmaAlgebra
+
+
+class FiltrationRequest(StrictModel):
+    """Compute the natural filtration of observations."""
+
+    space: FiniteProbabilitySpace
+    observations: tuple[tuple[str, ...], ...] = Field(default=())
+
+
+class DoobMartingaleRequest(StrictModel):
+    """Compute the Doob martingale of a payoff process."""
+
+    space: FiniteProbabilitySpace
+    observations: tuple[tuple[str, ...], ...] = Field(default=())
+    payoff: tuple[str, ...] = Field(min_length=1)
+
+
+class FiltrationResult(StrictModel):
+    """The natural filtration as a tuple of sigma algebra dicts."""
+
+    sigmas: tuple[dict[str, object], ...] = Field(default=())
+
+
+class DoobMartingaleResult(StrictModel):
+    """The Doob martingale as a tuple of rational-string value tuples."""
+
+    martingale: tuple[tuple[str, ...], ...] = Field(default=())
+
+
+__all__ = [
+    "ConditionalExpectationRequest",
+    "DoobMartingaleRequest",
+    "DoobMartingaleResult",
+    "FiltrationRequest",
+    "FiltrationResult",
+    "FromObservationRequest",
+    "JoinRequest",
+]

src/jacobian/math/finite_stochastic_processes/_operations.py

@@ -0,0 +1,64 @@
+"""Domain adapter for finite stochastic process operations."""
+
+from __future__ import annotations
+
+from jacobian.math.finite_stochastic_processes._models import (
+    ConditionalExpectationRequest,
+    DoobMartingaleRequest,
+    DoobMartingaleResult,
+    FiltrationRequest,
+    FiltrationResult,
+    FromObservationRequest,
+    JoinRequest,
+)
+from jacobian.math.finite_stochastic_processes.operations import (
+    conditional_expectation,
+    doob_martingale,
+    filtration_natural,
+    sigma_algebra_from_observation,
+)
+from jacobian.math.finite_stochastic_processes.values import (
+    FiniteRandomVariable,
+    FiniteSigmaAlgebra,
+)
+
+__all__ = [
+    "compute_conditional_expectation",
+    "compute_doob_martingale",
+    "compute_filtration",
+    "compute_join",
+    "compute_sigma_from_observation",
+]
+
+
+def compute_sigma_from_observation(request: FromObservationRequest) -> FiniteSigmaAlgebra:
+    return sigma_algebra_from_observation(request.space, request.observation)
+
+
+def compute_join(request: JoinRequest) -> FiniteSigmaAlgebra:
+    from jacobian.math.finite_stochastic_processes.operations import (
+        sigma_algebra_join_correct,
+    )
+    return sigma_algebra_join_correct(request.sigma1, request.sigma2)
+
+
+def compute_conditional_expectation(
+    request: ConditionalExpectationRequest,
+) -> FiniteRandomVariable:
+    return conditional_expectation(request.rv, request.sigma)
+
+
+def compute_filtration(request: FiltrationRequest) -> FiltrationResult:
+    sigmas = filtration_natural(request.space, request.observations)
+    return FiltrationResult(
+        sigmas=tuple(
+            {"blocks": s.blocks} for s in sigmas
+        )
+    )
+
+
+def compute_doob_martingale(
+    request: DoobMartingaleRequest,
+    ) -> DoobMartingaleResult:
+    result = doob_martingale(request.space, request.observations, request.payoff)
+    return DoobMartingaleResult(martingale=result)

src/jacobian/math/finite_stochastic_processes/_tools.py

@@ -0,0 +1,170 @@
+"""Finite stochastic process operation declarations."""
+
+from collections.abc import Callable
+from typing import Any
+
+from jacobian._models import StrictModel
+from jacobian.catalog._examples import example
+from jacobian.catalog.models import MathTool, OperationExample
+from jacobian.math.finite_stochastic_processes._models import (
+    ConditionalExpectationRequest,
+    DoobMartingaleRequest,
+    DoobMartingaleResult,
+    FiltrationRequest,
+    FiltrationResult,
+    FromObservationRequest,
+    JoinRequest,
+)
+from jacobian.math.finite_stochastic_processes._operations import (
+    compute_conditional_expectation,
+    compute_doob_martingale,
+    compute_filtration,
+    compute_join,
+    compute_sigma_from_observation,
+)
+from jacobian.math.finite_stochastic_processes.values import (
+    FiniteRandomVariable,
+    FiniteSigmaAlgebra,
+)
+
+
+def _op[
+    RequestT: StrictModel,
+    ResultT: StrictModel,
+](
+    operation_id: str,
+    title: str,
+    description: str,
+    request_model: type[RequestT],
+    result_model: type[ResultT],
+    operation: Callable[[RequestT], ResultT],
+    *tags: str,
+    examples: tuple[OperationExample, ...] = (),
+    version: str = "1",
+) -> MathTool[RequestT, ResultT]:
+    return MathTool(
+        operation_id=operation_id,
+        version=version,
+        title=title,
+        description=description,
+        request_type=request_model,
+        result_type=result_model,
+        run=operation,
+        tags=tags,
+        examples=examples,
+    )
+
+
+# A fair coin: samples {H, T}, masses 1/2, 1/2.
+_SPACE = {"samples": ["H", "T"], "masses": ["1/2", "1/2"]}
+
+FINIT_STOCHASTIC_PROCESS_OPERATIONS: tuple[MathTool[Any, Any], ...] = (
+    _op(
+        "probability.finite_sigma_algebra.from_observation.compute",
+        "Construct a sigma algebra from an observation map",
+        "Return the sigma algebra generated by an observation map Y: Omega -> "
+        "labels, whose blocks are equal-value fibers.",
+        FromObservationRequest,
+        FiniteSigmaAlgebra,
+        compute_sigma_from_observation,
+        "stochastic-process",
+        "sigma-algebra",
+        "exact",
+        examples=(
+            example(
+                "coin_observation",
+                "Sigma algebra from observing a fair coin.",
+                {"space": _SPACE, "observation": ["heads", "tails"]},
+            ),
+        ),
+    ),
+    _op(
+        "probability.finite_sigma_algebra.join.compute",
+        "Compute the join of two sigma algebras",
+        "Return the least sigma algebra containing both input sigma algebras. "
+        "Under the partition representation, this is the finest partition "
+        "that refines both input partitions.",
+        JoinRequest,
+        FiniteSigmaAlgebra,
+        compute_join,
+        "stochastic-process",
+        "sigma-algebra",
+        "exact",
+        examples=(
+            example(
+                "trivial_join",
+                "Join of two trivial sigma algebras on a fair coin.",
+                {
+                    "sigma1": {"space": _SPACE, "blocks": [["H", "T"]]},
+                    "sigma2": {"space": _SPACE, "blocks": [["H", "T"]]},
+                },
+            ),
+        ),
+    ),
+    _op(
+        "probability.conditional_expectation.finite.compute",
+        "Compute E[X | G] as a block-constant random variable",
+        "Return the conditional expectation of a random variable with respect "
+        "to a finite sigma algebra. On each block, the value is the "
+        "probability-weighted average of X over the samples in that block.",
+        ConditionalExpectationRequest,
+        FiniteRandomVariable,
+        compute_conditional_expectation,
+        "stochastic-process",
+        "conditional-expectation",
+        "exact",
+        examples=(
+            example(
+                "coin_conditional_expectation",
+                "Conditional expectation of a coin payoff on the trivial sigma algebra.",
+                {
+                    "rv": {"space": _SPACE, "values": ["1", "0"]},
+                    "sigma": {"space": _SPACE, "blocks": [["H", "T"]]},
+                },
+            ),
+        ),
+    ),
+    _op(
+        "probability.filtration.natural.compute",
+        "Compute the natural filtration of observations",
+        "Return the natural filtration F_t = sigma(Y_0, ..., Y_t) for each "
+        "time t. Each F_t is a finite sigma algebra, and F_t refines F_{t-1}.",
+        FiltrationRequest,
+        FiltrationResult,
+        compute_filtration,
+        "stochastic-process",
+        "filtration",
+        "exact",
+        examples=(
+            example(
+                "single_step_filtration",
+                "Natural filtration of a single coin observation.",
+                {"space": _SPACE, "observations": [["heads", "tails"]]},
+            ),
+        ),
+    ),
+    _op(
+        "probability.process.doob_martingale.compute",
+        "Compute the Doob martingale M_t = E[payoff | F_t]",
+        "Return the Doob martingale of a payoff random variable with respect "
+        "to the natural filtration of observations. The result is one tuple "
+        "of rational strings per time step.",
+        DoobMartingaleRequest,
+        DoobMartingaleResult,
+        compute_doob_martingale,
+        "stochastic-process",
+        "doob-martingale",
+        "exact",
+        examples=(
+            example(
+                "coin_doob_martingale",
+                "Doob martingale of a coin payoff process.",
+                {"space": _SPACE, "observations": [["heads", "tails"]], "payoff": ["1", "0"]},
+            ),
+        ),
+    ),
+)
+
+TOOLS = FINIT_STOCHASTIC_PROCESS_OPERATIONS
+
+__all__ = ["TOOLS"]