Cashflow Interface and Models (#40)

KSafran · web-flow · commit 79ccd2a3138f · 2025-04-23T09:51:12.000-04:00
* working cashflow setup

* add `get`, skipping name gathering step

* Automated version bump to 0.0.22

* fix type hints

* fix model name recovery

* fix ruff errors

* clean up cashflow model code

* add some cashflow unit tests

---------

Co-authored-by: KSafran &lt;KSafran@users.noreply.github.com&gt;
diff --git a/ledger_analytics/__about__.py b/ledger_analytics/__about__.py
@@ -1 +1 @@
-__version__ = '0.0.21'
+__version__ = '0.0.22'
diff --git a/ledger_analytics/__init__.py b/ledger_analytics/__init__.py
@@ -1,9 +1,10 @@
 from .__about__ import __version__
 from .api import AnalyticsClient
 from .autofit import AutofitControl
+from .cashflow import CashflowModel
 from .development import GMCL, ChainLadder, ManualATA, MeyersCRC, TraditionalChainLadder
 from .forecast import AR1, SSM, TraditionalGCC
-from .interface import ModelInterface, TriangleInterface
+from .interface import CashflowInterface, ModelInterface, TriangleInterface
 from .model import DevelopmentModel, ForecastModel, TailModel
 from .requester import Requester
 from .tail import ClassicalPowerTransformTail, GeneralizedBondy, Sherman
diff --git a/ledger_analytics/api.py b/ledger_analytics/api.py
@@ -4,7 +4,7 @@
 from abc import ABC
 from collections import namedtuple
 
-from .interface import ModelInterface, TriangleInterface
+from .interface import CashflowInterface, ModelInterface, TriangleInterface
 from .requester import Requester
 
 DEFAULT_HOST = "https://api.ldgr.app/analytics/"
@@ -78,6 +78,11 @@ def __init__(
             "forecast_model", self.host, self._requester, self.asynchronous
         )
     )
+    cashflow_model = property(
+        lambda self: CashflowInterface(
+            "cashflow_model", self.host, self._requester, self.asynchronous
+        )
+    )
 
     def test_endpoint(self) -> str:
         self._requester.get(self.host + "triangle")
diff --git a/ledger_analytics/cashflow.py b/ledger_analytics/cashflow.py
@@ -0,0 +1,208 @@
+from __future__ import annotations
+
+import time
+from typing import Dict
+
+from requests import Response
+from rich.console import Console
+
+from .config import JSONDict, ValidationConfig
+from .interface import CashflowInterface, TriangleInterface
+from .requester import Requester
+from .triangle import Triangle
+
+
+class CashflowModel(CashflowInterface):
+    def __init__(
+        self,
+        id: str,
+        name: str,
+        dev_model_name: str,
+        tail_model_name: str,
+        model_class: str,
+        endpoint: str,
+        requester: Requester,
+        asynchronous: bool = False,
+    ) -> None:
+        super().__init__(model_class, endpoint, requester, asynchronous)
+
+        self._endpoint = endpoint
+        self._id = id
+        self._name = name
+        self._dev_model_name = dev_model_name
+        self._tail_model_name = tail_model_name
+        self._model_class = model_class
+        self._fit_response: Response | None = None
+        self._predict_response: Response | None = None
+        self._get_response: Response | None = None
+
+    id = property(lambda self: self._id)
+    name = property(lambda self: self._name)
+    dev_model_name = property(lambda self: self._dev_model_name)
+    tail_model_name = property(lambda self: self._tail_model_name)
+    model_class = property(lambda self: self._model_class)
+    endpoint = property(lambda self: self._endpoint)
+    fit_response = property(lambda self: self._fit_response)
+    predict_response = property(lambda self: self._predict_response)
+    get_response = property(lambda self: self._get_response)
+    delete_response = property(lambda self: self._delete_response)
+
+    @classmethod
+    def get(
+        cls,
+        id: str,
+        name: str,
+        dev_model_name: str,
+        tail_model_name: str,
+        model_class: str,
+        endpoint: str,
+        requester: Requester,
+        asynchronous: bool = False,
+    ) -> CashflowModel:
+        console = Console()
+        with console.status("Retrieving...", spinner="bouncingBar") as _:
+            console.log(f"Getting model '{name}' with ID '{id}'")
+            get_response = requester.get(endpoint, stream=True)
+
+        self = cls(
+            id,
+            name,
+            dev_model_name,
+            tail_model_name,
+            model_class,
+            endpoint,
+            requester,
+            asynchronous,
+        )
+        self._get_response = get_response
+        return self
+
+    @classmethod
+    def fit_from_interface(
+        cls,
+        name: str,
+        dev_model_name: str,
+        tail_model_name: str,
+        model_class: str,
+        endpoint: str,
+        requester: Requester,
+        asynchronous: bool = False,
+    ) -> CashflowModel:
+        """This method fits a new model and constructs a CashflowModel instance.
+        It's intended to be used from the `ModelInterface` class mainly,
+        and in the future will likely be superseded by having separate
+        `create` and `fit` API endpoints.
+        """
+
+        post_data = {
+            "development_model_name": dev_model_name,
+            "tail_model_name": tail_model_name,
+            "name": name,
+            "model_config": {},
+        }
+        fit_response = requester.post(endpoint, data=post_data)
+        id = fit_response.json()["model"]["id"]
+        self = cls(
+            id=id,
+            name=name,
+            dev_model_name=dev_model_name,
+            tail_model_name=tail_model_name,
+            model_class=model_class,
+            endpoint=endpoint + f"/{id}",
+            requester=requester,
+            asynchronous=asynchronous,
+        )
+
+        self._fit_response = fit_response
+
+        return self
+
+    def predict(
+        self,
+        triangle: str | Triangle,
+        config: JSONDict | None = None,
+        initial_loss_triangle: Triangle | str | None = None,
+        prediction_name: str | None = None,
+        timeout: int = 300,
+    ) -> Triangle:
+        triangle_name = triangle if isinstance(triangle, str) else triangle.name
+        config = {
+            "triangle_name": triangle_name,
+            "predict_config": self.PredictConfig(**(config or {})).__dict__,
+        }
+        if prediction_name:
+            config["prediction_name"] = prediction_name
+
+        if isinstance(initial_loss_triangle, Triangle):
+            config["predict_config"]["initial_loss_name"] = initial_loss_triangle.name
+        elif isinstance(initial_loss_triangle, str):
+            config["predict_config"]["initial_loss_name"] = initial_loss_triangle
+
+        url = self.endpoint + "/predict"
+        self._predict_response = self._requester.post(url, data=config)
+
+        if self._asynchronous:
+            return self
+
+        task_id = self.predict_response.json()["modal_task"]["id"]
+        task_response = self._poll_remote_task(
+            task_id=task_id,
+            task_name=f"Predicting from model '{self.name}' on triangle '{triangle_name}'",
+            timeout=timeout,
+        )
+        if task_response.get("status") != "success":
+            raise ValueError(f"Task failed: {task_response['error']}")
+        triangle_id = self.predict_response.json()["predictions"]
+        triangle = TriangleInterface(
+            host=self.endpoint.replace(f"{self.model_class_slug}/{self.id}", ""),
+            requester=self._requester,
+        ).get(id=triangle_id)
+        return triangle
+
+    def delete(self) -> CashflowModel:
+        self._delete_response = self._requester.delete(self.endpoint)
+        return self
+
+    def _poll(self, task_id: str) -> JSONDict:
+        endpoint = self.endpoint.replace(
+            f"{self.model_class_slug}/{self.id}", f"tasks/{task_id}"
+        )
+        return self._requester.get(endpoint)
+
+    def _poll_remote_task(
+        self, task_id: str, task_name: str = "", timeout: int = 300
+    ) -> dict:
+        start = time.time()
+        status = ["CREATED"]
+        console = Console()
+        with console.status("Working...", spinner="bouncingBar") as _:
+            while time.time() - start < timeout:
+                task = self._poll(task_id).json()
+                modal_status = (
+                    "FINISHED" if task["task_response"] is not None else "PENDING"
+                )
+                status.append(modal_status)
+                if status[-1] != status[-2]:
+                    console.log(f"{task_name}: {status[-1]}")
+                if status[-1].lower() == "finished":
+                    return task["task_response"]
+            raise TimeoutError(f"Task '{task}' timed out")
+
+    class PredictConfig(ValidationConfig):
+        """Cashflow model configuration class.
+
+        Attributes:
+            use_bf: Whether or not to use Bornhuetter-Ferguson method to adjust reserve estimates.
+            use_reverse_bf: Whether or not to use the Reverse B-F method to adjust reserve
+                estimates.
+            gamma: Gamma parameter in the Reverse B-F method.
+            min_reserve: Minimum reserve amounts as a function of development lag.
+            seed: Seed to use for model sampling. Defaults to ``None``, but it is highly recommended
+                to set.
+        """
+
+        use_bf: bool = True
+        use_reverse_bf: bool = True
+        gamma: float = 0.7
+        min_reserve: Dict[float, float] | None
+        seed: int | None = None
diff --git a/ledger_analytics/development.py b/ledger_analytics/development.py
@@ -71,6 +71,8 @@ class Config(ValidationConfig):
                 industry-informed priors for best results.
             autofit_override: override the MCMC autofitting procedure arguments. See the documentation
                 for a fully description of options in the User Guide.
+            sigma_volume: Boolean indicating whether to use a volume parameter in the variance
+                function.
             prior_only: should a prior predictive simulation be run?
             seed: Seed to use for model sampling. Defaults to ``None``, but it is highly recommended
                 to set.
@@ -85,6 +87,7 @@ class Config(ValidationConfig):
         informed_priors_version: str | None = None
         use_multivariate: bool = False
         autofit_override: dict[str, float | int | None] = None
+        sigma_volume: bool = False
         prior_only: bool = False
         seed: int | None = None
 
diff --git a/ledger_analytics/interface.py b/ledger_analytics/interface.py
@@ -207,3 +207,111 @@ def _get_details_from_id_name(
             )
             raise ValueError(f"No model found with {name_or_id}.")
         return models[0]
+
+
+class CashflowInterface(metaclass=ModelRegistry):
+    """The CashflowInterface class allows basic CRUD operations
+    on for Cashflow endpoints and objects."""
+
+    def __init__(
+        self,
+        model_class: str,
+        host: str,
+        requester: Requester,
+        asynchronous: bool = False,
+    ) -> None:
+        self._model_class = model_class
+        self._host = host
+        self._endpoint = host + self.model_class_slug
+        self._requester = requester
+        self._asynchronous = asynchronous
+
+    model_class = property(lambda self: self._model_class)
+    endpoint = property(lambda self: self._endpoint)
+
+    def create(
+        self,
+        dev_model: str | "DevelopmentModel",
+        tail_model: str | "TailModel",
+        name: str,
+    ):
+        dev_model_name = dev_model if isinstance(dev_model, str) else dev_model.name
+        tail_model_name = tail_model if isinstance(tail_model, str) else tail_model.name
+        return ModelRegistry.REGISTRY["cashflow_model"].fit_from_interface(
+            name=name,
+            dev_model_name=dev_model_name,
+            tail_model_name=tail_model_name,
+            model_class=self.model_class,
+            endpoint=self.endpoint,
+            requester=self._requester,
+            asynchronous=self._asynchronous,
+        )
+
+    def get(self, name: str | None = None, id: str | None = None):
+        model_obj = self._get_details_from_id_name(name, id)
+        endpoint = self.endpoint + f"/{model_obj['id']}"
+        dev_interface = ModelInterface(
+            "development-model", self._host, self._requester, self._asynchronous
+        )
+        dev_model_name = dev_interface.get(id=model_obj["development_model"]).name
+        tail_interface = ModelInterface(
+            "tail-model", self._host, self._requester, self._asynchronous
+        )
+        tail_model_name = tail_interface.get(id=model_obj["tail_model"]).name
+        return ModelRegistry.REGISTRY["cashflow_model"].get(
+            id=model_obj["id"],
+            name=model_obj["name"],
+            dev_model_name=dev_model_name,
+            tail_model_name=tail_model_name,
+            model_class=self.model_class,
+            endpoint=endpoint,
+            requester=self._requester,
+            asynchronous=self._asynchronous,
+        )
+
+    def predict(
+        self,
+        triangle: str | Triangle,
+        config: JSONDict | None = None,
+        initial_loss_triangle: str | Triangle | None = None,
+        timeout: int = 300,
+        name: str | None = None,
+        id: str | None = None,
+    ):
+        model = self.get(name, id)
+        return model.predict(
+            triangle,
+            config=config,
+            initial_loss_triangle=initial_loss_triangle,
+            timeout=timeout,
+        )
+
+    def delete(self, name: str | None = None, id: str | None = None) -> None:
+        model = self.get(name, id)
+        return model.delete()
+
+    def list(self) -> list[JSONDict]:
+        return self._requester.get(self.endpoint, stream=True).json()
+
+    def list_model_types(self) -> list[JSONDict]:
+        url = self.endpoint + "-type"
+        return self._requester.get(url).json()
+
+    @property
+    def model_class_slug(self):
+        return self.model_class.replace("_", "-")
+
+    def _get_details_from_id_name(
+        self, model_name: str | None = None, model_id: str | None = None
+    ) -> str:
+        models = [
+            result
+            for result in self.list().get("results")
+            if result.get("name") == model_name or result.get("id") == model_id
+        ]
+        if not len(models):
+            name_or_id = (
+                f"name '{model_name}'" if model_id is None else f"ID '{model_id}'"
+            )
+            raise ValueError(f"No model found with {name_or_id}.")
+        return models[0]
diff --git a/ledger_analytics/model.py b/ledger_analytics/model.py
@@ -110,8 +110,6 @@ def fit_from_interface(
             "model_config": cls.Config(**config).__dict__,
         }
         fit_response = requester.post(endpoint, data=config)
-        if not fit_response.ok:
-            fit_response.raise_for_status()
         id = fit_response.json()["model"]["id"]
         self = cls(
             id=id,
@@ -162,8 +160,6 @@ def predict(
 
         url = self.endpoint + "/predict"
         self._predict_response = self._requester.post(url, data=config)
-        if not self._predict_response.ok:
-            self._predict_response.raise_for_status()
 
         if self._asynchronous:
             return self
diff --git a/test/unit/test_api.py b/test/unit/test_api.py

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-__version__ = '0.0.21'`
	`1`	`+__version__ = '0.0.22'`