Enhance cashflow calculation with error handling and add JSON export endpoint

main.py

@@ -3,6 +3,7 @@ from urllib.parse import urlencode
 from fastapi import FastAPI, Query, Request, Response
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import PlainTextResponse, RedirectResponse
+import orjson
 import pandas as pd
 from pydantic import Field, model_validator
 from capacity_factors import get_solar_capacity_factor
@@ -78,14 +79,31 @@ class CashflowParams(SolarPVAssumptions):
 
 @app.get("/solarpv/cashflow.csv", response_class=PlainTextResponse)
 def get_cashflow(params: Annotated[CashflowParams, Query()]) -> str:
-    cashflow = calculate_cashflow_for_renewable_project(
-        params, tariff=params.tariff, return_model=True
-    )[0]
-    if params.transpose:
-        cashflow = cashflow.to_pandas().T
-        cashflow.columns = cashflow.loc["Period"].astype(int).astype(str)
-        return cashflow.drop(["Period"]).to_csv(float_format="%.3f")
+    try:
+        cashflow = calculate_cashflow_for_renewable_project(
+            params, tariff=params.tariff, return_model=True, errors="ignore"
+        )[0]
+        if params.transpose:
+            cashflow = cashflow.to_pandas().T
+            cashflow.columns = cashflow.loc["Period"].astype(int).astype(str)
+            return cashflow.drop(["Period"]).to_csv(float_format="%.3f")
+    except Exception as e:
+        return str(e)
     return cashflow.write_csv(float_precision=3)
 
+@app.get("/solarpv/cashflow.json")
+def get_cashflow_json(params: Annotated[CashflowParams, Query()]) -> Dict:
+    try:
+        cashflow, equity_irr, tariff, adjusted_assumptions = calculate_cashflow_for_renewable_project(
+            params, tariff=params.tariff, return_model=True, errors="ignore"
+        )
+    except Exception as e:
+        return {"error": str(e)}
+    return {
+        "cashflow": cashflow.to_pandas().to_dict(orient="records"),
+        "equity_irr": equity_irr,
+        "tariff": tariff,
+        "assumptions": adjusted_assumptions,
+    }
 
 app = gr.mount_gradio_app(app, interface, path="/")

model.py

@@ -1,4 +1,4 @@
-from typing import Annotated, Iterable, Optional, Tuple
+from typing import Annotated, Iterable, Literal, Optional, Tuple
 import numpy as np
 import polars as pl
 from pyxirr import irr, npv
@@ -10,9 +10,9 @@ from schema import SolarPVAssumptions
 
 
 def calculate_cashflow_for_renewable_project(
-    assumptions: SolarPVAssumptions, tariff: float | Iterable, return_model=False
+    assumptions: SolarPVAssumptions, tariff: float | Iterable, return_model=False, errors: Literal["raise", "ignore"] = "raise"
 ) -> (
-    Annotated[Optional[float], "Post-tax equity IRR - Cost of equity"]
+    Annotated[float | None, "Post-tax equity IRR - Cost of equity"]
     | Tuple[
         Annotated[pl.DataFrame, "Cashflow model"],
         Annotated[float | None, "Post-tax equity IRR"],
@@ -28,8 +28,11 @@ def calculate_cashflow_for_renewable_project(
         return_model (bool, optional): Whether to return the model. Defaults to False.
 
     Returns:
-        Optional[float]: The post-tax equity IRR - cost of equity
-        Tuple[pl.DataFrame, float, float, SolarPVAssumptions]: The model, post-tax equity IRR, breakeven tariff, and assumptions
+        float: post-tax equity IRR - cost of equity (if return_model is False)
+        pl.DataFrame: Cashflow model (if return_model is True)
+        float: Post-tax equity IRR (if return_model is True)
+        float: Breakeven tariff (if return_model is True)
+        SolarPVAssumptions: Assumptions (if return_model is True)
     """
 
     # Tariff must be a number
@@ -97,12 +100,11 @@ def calculate_cashflow_for_renewable_project(
                 .otherwise(pl.col("CFADS_mn") / assumptions.dscr),
             )
         )
-        assumptions.debt_pct_of_capital_cost = pyxirr.npv(
+        # Calculate debt % of capital cost by calculating NPV of debt service, and then dividing by capital cost
+        assumptions.debt_pct_of_capital_cost = min(1, pyxirr.npv(
             assumptions.cost_of_debt,
             model.select("Target_Debt_Service_mn").__array__()[0:assumptions.loan_tenor_years+1, 0],
-        ) / (assumptions.capital_cost / 1000)
-        # print(assumptions.debt_pct_of_capital_cost)
-        # assumptions.equity_pct_of_capital_cost = 1 - assumptions.debt_pct_of_capital_cost
+        ) / (assumptions.capital_cost / 1000))
         assert (
             assumptions.debt_pct_of_capital_cost
             + assumptions.equity_pct_of_capital_cost
@@ -240,18 +242,28 @@ def calculate_cashflow_for_renewable_project(
         )
     )
 
+    ## Do some sanity checks
+    # Check that the debt outstanding at the end of the loan period is zero
+    assert (
+        (model["Debt_Outstanding_EoP_mn"].slice(assumptions.loan_tenor_years,) < 0.0001).all() # type: ignore
+    ), f"Debt outstanding at the end of the loan period is not zero: {model['Debt_Outstanding_EoP_mn'].slice(assumptions.loan_tenor_years,)}" # type: ignore
+
+
     # Calculate Post-Tax Equity IRR
     try:
         post_tax_equity_irr = irr(model["Post_Tax_Net_Equity_Cashflow_mn"].to_numpy())
     except pyxirr.InvalidPaymentsError as e:
-        if assumptions.debt_pct_of_capital_cost == 1:
-            raise AssertionError(
-                "The project is fully financed by debt so equity IRR is infinite."
-            )
+        if errors == "ignore":
+            post_tax_equity_irr = None
         else:
-            raise AssertionError(
-                f"The power tariff is too low so the project never breaks even. Please increase it from {tariff}."
-            )
+            if assumptions.debt_pct_of_capital_cost == 1:
+                raise AssertionError(
+                    "The project is fully financed by debt so equity IRR is infinite."
+                )
+            else:
+                raise AssertionError(
+                    f"The power tariff is too low so the project never breaks even. Please increase it from {tariff}."
+                )
 
     if return_model:
         return model, post_tax_equity_irr, tariff, assumptions

schema.py

@@ -37,7 +37,7 @@ class SolarPVAssumptions(BaseModel):
         ),
     ] = None
     cost_of_debt: Annotated[
-        float,
+        Optional[float],
         Field(
             ge=0,
             le=0.5,
@@ -46,7 +46,7 @@ class SolarPVAssumptions(BaseModel):
         ),
     ] = 0.05
     cost_of_equity: Annotated[
-        float,
+        Optional[float],
         Field(
             ge=0,
             le=0.5,
@@ -117,6 +117,11 @@ class SolarPVAssumptions(BaseModel):
                 raise ValueError("Debt and equity percentages must sum to 1")
         return self
 
+    @model_validator(mode="before")
+    @classmethod
+    def remove_none_values(cls, values):
+        return {k: v for k, v in values.items() if v is not None}
+
     @computed_field
     @property
     def capital_cost(