
========================================
File: README.md
========================================

# mocaco

[![CI](https://github.com/emalvarol/mocaco/actions/workflows/ci.yml/badge.svg)](https://github.com/emalvarol/mocaco/actions/workflows/ci.yml)
![Python Versions](https://img.shields.io/badge/python-3.12%20|%203.13%20|%203.14-blue)
![License: LGPL](https://img.shields.io/badge/license-LGPL-blue)

Convergence analysis for Monte Carlo series using Polars DataFrames.

<p align="center">
  <img src="assets/Logo_v1.png" alt="mocaco logo" width="400"/>
</p>

The library provides a simple API for applying different convergence and
stopping criteria to Monte Carlo simulation results.

## Quickstart

```python
import polars as pl
import mocaco as mcc

# Prepare your Monte Carlo samples
df = pl.DataFrame({"it": range(1000), "value": your_samples})

# Create a normalized sample frame
samples = mcc.samples(df, it_col="it", target_col="value")

# Run a convergence criterion
result = mcc.convergence.clt_absolute(samples, threshold=0.01)

# Inspect results
print(result.is_converged)  # True/False
print(result.estimate)  # Final cumulative mean
print(result.error)  # Final CLT-based absolute error
print(result.n)  # Number of samples
# and more...
```

## API Overview

| Function | Description |
|----------|-------------|
| `mcc.samples(df, it_col, target_col)` | Create a normalized sample frame from a Polars DataFrame |
| `mcc.convergence(samples, method="...")` | Generic invocation of any registered criterion |
| `mcc.convergence.clt_absolute(samples, ...)` | Typed invocation of a specific criterion |
| `mcc.methods()` | List all registered convergence criteria |
| `mcc.describe("clt_absolute")` | Get metadata and parameter info for a criterion |

## Available Methods

- [`clt_absolute`](docs/methods/clt_absolute.md) — CLT-based absolute error threshold

## Documentation

Full documentation is available in the `docs/` directory or can be built with:

```bash
mkdocs serve
```

## Installation (Developers)

### Environment

Create the virtual environment and install the project in editable mode with `uv`:

```bash
uv venv
uv sync
uv pip install -e .
```

If it is needed to remove an old environment:

```bash
Remove-Item -Recurse -Force .venv
```

### ruff
```bash
uv run ruff check --fix . && uv run ruff format .
```

### mypy
```bash
uv run mypy src/
```


========================================
File: src/mocaco/api.py
========================================



from __future__ import annotations

from typing import TYPE_CHECKING, Any

from . import methods as _methods  # noqa: F401
from .protocols import SampleFrame
from .registry import registry
from .wrappers import EvalFrequencyWrapper

if TYPE_CHECKING:
    from collections.abc import Callable

    import polars as pl

    from .result import ConvergenceResult


def samples(
    df: pl.DataFrame,
    *,
    target_col: str,
    it_col: str | None = None,
) -> SampleFrame:
    
    return SampleFrame(
        df=df,
        target_col=target_col,
        it_col=it_col,
    )


class Convergence:
    

    def __call__(
        self,
        samples: SampleFrame,
        *,
        method: str,
        **kwargs: Any,
    ) -> ConvergenceResult:
        
        criterion = registry.get(method)
        eval_frequency = kwargs.pop("eval_frequency", None)
        params = criterion.params_type(**kwargs)

        if eval_frequency is not None:
            if not getattr(criterion, "supports_eval_frequency", False):
                raise TypeError(f"Criterion '{criterion.name}' does not support eval_frequency")
            wrapper = EvalFrequencyWrapper(criterion, eval_frequency=eval_frequency)
            return wrapper.run(samples=samples, params=params)

        return criterion.run(
            samples=samples,
            params=params,
        )

    def __getattr__(self, name: str) -> Callable[..., ConvergenceResult]:
        
        criterion = registry.get(name)

        def _method(samples: SampleFrame, **kwargs: Any) -> ConvergenceResult:
            eval_frequency = kwargs.pop("eval_frequency", None)
            params = criterion.params_type(**kwargs)

            if eval_frequency is not None:
                if not getattr(criterion, "supports_eval_frequency", False):
                    raise TypeError(f"Criterion '{criterion.name}' does not support eval_frequency")
                wrapper = EvalFrequencyWrapper(criterion, eval_frequency=eval_frequency)
                return wrapper.run(samples=samples, params=params)

            return criterion.run(samples=samples, params=params)

        return _method


convergence = Convergence()


def methods() -> tuple[str, ...]:
    
    return registry.names()


def describe(name: str) -> None:
    
    return registry.describe(name)


========================================
File: src/mocaco/protocols.py
========================================



from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, ClassVar, Protocol

import polars as pl

if TYPE_CHECKING:
    from collections.abc import Sequence

    from .result import ConvergenceResult

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class SampleFrame:
    

    df: pl.DataFrame
    target_col: str
    it_col: str | None = None

    def __post_init__(self) -> None:
        
        # 1. Resolve and normalize it_col name
        if self.it_col is None:
            it_col_name = "it"
            new_df = self.df.with_columns(pl.arange(1, len(self.df) + 1).alias(it_col_name))
            object.__setattr__(self, "df", new_df)
            object.__setattr__(self, "it_col", it_col_name)
            logger.info("it_col was None — auto-created column 'it' with sequential indices.")
        else:
            it_col_name = self.it_col

        # 2. Validate structural integrity (it_col_name is guaranteed to be str here)
        missing = [col for col in (it_col_name, self.target_col) if col not in self.df.columns]
        if missing:
            raise ValueError(
                f"Columns not found in DataFrame: {missing}. Available columns: {self.df.columns}"
            )

        if self.df.is_empty():
            raise ValueError("The sample DataFrame cannot be empty.")

        # 3. Validate Data Types (it_col and target_col must be numeric)
        it_dtype = self.df.get_column(it_col_name).dtype
        if not it_dtype.is_numeric():
            raise TypeError(
                f"Iteration column '{it_col_name}' must be numeric (int or float), got {it_dtype}."
            )

        target = self.df.get_column(self.target_col)
        if not target.dtype.is_numeric():
            raise TypeError(
                f"Target column '{self.target_col}' must be numeric (int or float), got {target.dtype}."
            )

        # 4. Input validation and edge case handling (Null, NaN, Inf)
        if target.is_null().any():
            raise ValueError(f"Target column '{self.target_col}' contains Null (missing) values.")

        if target.dtype in (pl.Float32, pl.Float64):
            if target.is_nan().any():
                raise ValueError(f"Target column '{self.target_col}' contains NaN values.")
            if target.is_infinite().any():
                raise ValueError(
                    f"Target column '{self.target_col}' contains Infinite (Inf) values."
                )


class Criterion(Protocol):
    

    name: str
    supports_eval_frequency: bool
    description: str
    params_type: type[Any]

    assumptions: ClassVar[Sequence[str]]
    limitations: ClassVar[Sequence[str]]
    result_interpretation: str
    example_usage: str
    references: ClassVar[Sequence[str]]

    def run(
        self,
        samples: SampleFrame,
        params: Any,
    ) -> ConvergenceResult: ...


========================================
File: src/mocaco/registry.py
========================================



from __future__ import annotations

from typing import TYPE_CHECKING, Any, TypeVar, overload

from rich.console import Console
from rich.panel import Panel
from rich.table import Table

if TYPE_CHECKING:
    from collections.abc import Callable

    from .protocols import Criterion

C = TypeVar("C", bound=type)

console = Console()


class CriterionRegistry:
    def __init__(self) -> None:
        self._criteria: dict[str, Criterion] = {}

    # Overload 1: Decorator usage (no criterion provided)
    @overload
    def register(self, criterion: None = None, *, overwrite: bool = False) -> Callable[[C], C]: ...

    # Overload 2: Programmatic usage (criterion instance provided)
    @overload
    def register(self, criterion: Criterion, *, overwrite: bool = False) -> Criterion: ...

    def register(
        self,
        criterion: Criterion | None = None,
        *,
        overwrite: bool = False,
    ) -> Any:
        # 1. Programmatic use: registry.register(MyCriterion())
        if criterion is not None:
            name = criterion.name

            if not name:
                raise ValueError("A criterion must define a non-empty name.")

            if name in self._criteria and not overwrite:
                raise ValueError(f"Criterion {name!r} is already registered.")

            self._criteria[name] = criterion
            return criterion

        # 2. Decorator use: @registry.register()
        def wrapper(cls: C) -> C:
            instance = cls()
            self.register(instance, overwrite=overwrite)
            return cls

        return wrapper

    def get(self, name: str) -> Criterion:
        try:
            return self._criteria[name]
        except KeyError as exc:
            available = ", ".join(self.names())
            raise ValueError(
                f"Unknown convergence criterion: {name!r}. Available criteria: {available}"
            ) from exc

    def names(self) -> tuple[str, ...]:
        return tuple(self._criteria.keys())

    def describe(self, name: str) -> None:
        # 1. Fetch the criterion from the registry
        try:
            criterion = registry.get(name)
        except ValueError:
            console.print(f"[bold red]Error:[/bold red] Method '{name}' not found in registry.")
            return

        # 2. Print the Method Name and Main Description in a Panel
        console.print()
        console.print(
            Panel(
                f"{criterion.description}",
                title=f"Method: [bold cyan]{name}[/bold cyan]",
                title_align="left",
                border_style="cyan",
                expand=False,
            )
        )

        # 3. Create a Table for the Parameters
        table = Table(
            title="Parameters",
            title_style="bold magenta",
            title_justify="left",
            show_header=True,
            header_style="bold black on white",
        )

        table.add_column("Name", style="bold cyan", no_wrap=True)
        table.add_column("Type", style="green")
        table.add_column("Required", justify="center")
        table.add_column("Default", style="yellow")
        table.add_column("Description", style="dim")

        # 4. Introspect the Pydantic V2 model fields
        params_type = criterion.params_type

        for field_name, field_info in params_type.model_fields.items():
            # Get type name cleanly
            field_type = getattr(field_info.annotation, "__name__", str(field_info.annotation))

            is_required = field_info.is_required()
            req_icon = "[green]✔[/green]" if is_required else "[red]✘[/red]"
            default_val = "-" if is_required else repr(field_info.default)
            desc = field_info.description or "No description provided."

            table.add_row(field_name, field_type, req_icon, default_val, desc)

        # 5. Render the table
        console.print(table)
        console.print()


registry = CriterionRegistry()


========================================
File: src/mocaco/result.py
========================================



from __future__ import annotations

from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any

import matplotlib.pyplot as plt
import polars as pl
from rich.console import Console
from rich.panel import Panel
from rich.table import Table

if TYPE_CHECKING:
    import pandas as pd


class ResultPlotter:
    

    def __init__(self, result: ConvergenceResult) -> None:
        self.result = result

    def evolution(
        self,
        *,
        show: bool = False,
    ) -> Any:
        
        df = self.result.data

        # Ensure required columns exist
        missing = [
            col
            for col in (
                self.result.estimate_col,
                self.result.error_col,
                self.result.n_col,
                self.result.is_converged_col,
            )
            if col not in self.result.data.columns
        ]
        if missing:
            raise ValueError(f"Columns not found in diagnostic data for plotting: {missing}")

        # Extract arrays
        n_vals = df.get_column(self.result.n_col).to_numpy()
        estimate_vals = df.get_column(self.result.estimate_col).to_numpy()
        error_vals = df.get_column(self.result.error_col).to_numpy()

        fig, ax = plt.subplots(figsize=(8, 5))

        # Plot estimate line
        ax.plot(n_vals, estimate_vals, label="Estimate", color="#1f77b4", linewidth=2)

        # Plot error band
        ax.fill_between(
            n_vals,
            estimate_vals - error_vals,
            estimate_vals + error_vals,
            color="#1f77b4",
            alpha=0.2,
            label="Error Band",
        )

        # Plot vertical line for convergence (first True occurrence)
        converged_df = df.filter(pl.col(self.result.is_converged_col))
        if not converged_df.is_empty():
            first_converged_n = converged_df.get_column(self.result.n_col)[0]
            ax.axvline(
                x=first_converged_n,
                color="red",
                linestyle="--",
                label=f"Converged at n={first_converged_n}",
            )

        ax.set_xlabel("Number of samples (n)")
        ax.set_ylabel("Estimate")
        ax.set_title(f"Convergence Evolution ({self.result.method})")
        ax.legend()
        ax.grid(True, linestyle=":", alpha=0.7)

        if show:
            plt.show()
            plt.close(fig)

        return ax


@dataclass(frozen=True)
class ConvergenceResult:
    

    # Common outputs to all criterions
    method: str
    n: int
    n_col: str
    estimate: float | None
    estimate_col: str
    error: float | None
    error_col: str
    is_converged: bool | None
    is_converged_col: str

    # Build by Wrapper or by the criterion default
    data: pl.DataFrame

    # Method-specific outputs
    diagnostics: dict[str, Any] = field(default_factory=dict)

    @property
    def plot(self) -> ResultPlotter:
        
        return ResultPlotter(self)

    def to_polars(self) -> pl.DataFrame:
        
        return self.data

    def to_pandas(self) -> pd.DataFrame:
        
        return self.data.to_pandas()

    def summary(self) -> None:
        
        console = Console()

        # Define status colors and icons
        status_color = "green" if self.is_converged else "yellow"
        status_icon = "✔" if self.is_converged else "⚠"
        status_text = "Converged" if self.is_converged else "Not Converged"

        # Create the inner layout table
        table = Table(show_header=False, box=None, padding=(0, 2))
        table.add_row("[bold]Method:[/bold]", f"[cyan]{self.method}[/cyan]")
        table.add_row(
            "[bold]Status:[/bold]", f"[{status_color}]{status_icon} {status_text}[/{status_color}]"
        )
        table.add_row("[bold]Samples (n):[/bold]", str(self.n))
        est_str = f"{self.estimate:.6g}" if self.estimate is not None else "-"
        table.add_row("[bold]Estimate:[/bold]", est_str)
        err_str = f"{self.error:.6g}" if self.error is not None else "-"
        table.add_row("[bold]Error:[/bold]", err_str)

        # Append diagnostics dynamically if they exist
        if self.diagnostics:
            table.add_row("", "")  # Spacer
            table.add_row("[bold italic]Diagnostics:[/bold italic]", "")
            for key, value in self.diagnostics.items():
                # Format floats cleanly, otherwise just cast to str
                val_str = f"{value:.6g}" if isinstance(value, float) else str(value)
                table.add_row(f"  {key}:", val_str)

        # Wrap in a stylized panel
        panel = Panel(
            table,
            title="[bold]Convergence Summary[/bold]",
            title_align="left",
            border_style=status_color,
            expand=False,
        )

        console.print()
        console.print(panel)
        console.print()

    def __str__(self) -> str:
        
        return str(self.data)

    def __repr__(self) -> str:
        
        return (
            f"ConvergenceResult("
            f"method={self.method!r}, "
            f"n={self.n!r}"
            f"estimate={self.estimate!r}, "
            f"error={self.error!r}, "
            f"is_converged={self.is_converged!r}, "
            f")"
        )


========================================
File: src/mocaco/wrappers.py
========================================



from __future__ import annotations

from typing import TYPE_CHECKING, Any

import polars as pl

from .protocols import SampleFrame
from .result import ConvergenceResult

if TYPE_CHECKING:
    from collections.abc import Sequence

    from .protocols import Criterion


class CriterionWrapper:
    

    def __init__(self, criterion: Criterion) -> None:
        self._criterion = criterion

    @property
    def name(self) -> str:
        return self._criterion.name

    @property
    def description(self) -> str:
        return self._criterion.description

    @property
    def params_type(self) -> type[Any]:
        return self._criterion.params_type

    @property
    def assumptions(self) -> Sequence[str]:
        return self._criterion.assumptions

    @property
    def limitations(self) -> Sequence[str]:
        return self._criterion.limitations

    @property
    def result_interpretation(self) -> str:
        return self._criterion.result_interpretation

    @property
    def example_usage(self) -> str:
        return self._criterion.example_usage

    @property
    def references(self) -> Sequence[str]:
        return self._criterion.references

    def run(
        self,
        samples: SampleFrame,
        params: Any,
    ) -> ConvergenceResult:
        return self._criterion.run(samples, params)


class EvalFrequencyWrapper(CriterionWrapper):
    

    def __init__(self, criterion: Criterion, eval_frequency: int) -> None:
        super().__init__(criterion)
        if eval_frequency <= 0:
            raise ValueError(f"eval_frequency must be a positive integer, got {eval_frequency}")
        self._eval_frequency = eval_frequency

    @property
    def eval_frequency(self) -> int:
        return self._eval_frequency

    def run(
        self,
        samples: SampleFrame,
        params: Any,
    ) -> ConvergenceResult:
        total_rows = len(samples.df)

        if self._eval_frequency >= total_rows:
            result = self._criterion.run(samples, params)
            return ConvergenceResult(
                method=result.method,
                n=result.n,
                n_col=result.n_col,
                estimate=result.estimate,
                estimate_col=result.estimate_col,
                error=result.error,
                error_col=result.error_col,
                is_converged=result.is_converged,
                is_converged_col=result.is_converged_col,
                data=result.data,
                diagnostics={
                    **result.diagnostics,
                    "eval_frequency": self._eval_frequency,
                    "n_evaluations": 1,
                },
            )

        eval_points = list(range(self._eval_frequency, total_rows + 1, self._eval_frequency))
        if eval_points[-1] != total_rows:
            eval_points.append(total_rows)

        results: list[ConvergenceResult] = []
        for point in eval_points:
            sliced_df = samples.df.head(point)
            sliced_samples = SampleFrame(
                df=sliced_df,
                target_col=samples.target_col,
                it_col=samples.it_col,
            )
            result = self._criterion.run(sliced_samples, params)
            results.append(result)

        data_frames = [r.data for r in results]
        combined_data = pl.concat(data_frames, how="align")

        final = results[-1]
        return ConvergenceResult(
            method=final.method,
            n=final.n,
            n_col=final.n_col,
            estimate=final.estimate,
            estimate_col=final.estimate_col,
            error=final.error,
            error_col=final.error_col,
            is_converged=final.is_converged,
            is_converged_col=final.is_converged_col,
            data=combined_data,
            diagnostics={
                **final.diagnostics,
                "eval_frequency": self._eval_frequency,
                "n_evaluations": len(results),
            },
        )


========================================
File: src/mocaco/__init__.py
========================================



from .api import convergence, describe, methods, samples

__all__ = [
    "convergence",
    "describe",
    "methods",
    "samples",
]


========================================
File: src/mocaco/methods/clt_absolute.py
========================================



# Mandatory basic modules
from __future__ import annotations

from typing import TYPE_CHECKING

# Basic modules used to build the criterion (According contributor needs)
import polars as pl
from pydantic import BaseModel, Field
from scipy.stats import norm

# Modules needed from mocaco
from mocaco.registry import registry
from mocaco.result import ConvergenceResult

if TYPE_CHECKING:
    from mocaco.protocols import SampleFrame


# 1. Criterion input parameters
class CLTAbsoluteParams(BaseModel):
    threshold: float = Field(
        ...,
        gt=0,
        description="Maximum accepted absolute Monte Carlo error.",
    )

    confidence_level: float = Field(
        0.95,
        gt=0,
        lt=1,
        description="Confidence level used for the CLT margin of error.",
    )


# 2. Criterion logic (calculous)
@registry.register()
class CLTAbsoluteCriterion:
    name = "clt_absolute"

    supports_eval_frequency = True

    description = (
        "Central Limit Theorem based convergence criterion "
        "using an absolute Monte Carlo error threshold."
    )

    params_type = CLTAbsoluteParams

    assumptions = (
        "Samples are independent and identically distributed (i.i.d.).",
        "The target quantity has finite variance.",
        "The sample size is sufficiently large for the CLT to provide a normal approximation. The deafult stb_window = 30 aims to support this assumption.",
    )

    limitations = (
        "This is a probabilistic stopping criterion, not a proof of convergence.",
        "For small sample sizes, the normal approximation may be inaccurate.",
        "Does not detect systematic bias or non-stationarity in the sampling process.",
        "The stability window introduces a lag in detecting convergence.",
    )

    result_interpretation = (
        "is_converged is True when the CLT-based absolute error has stayed below "
        "the threshold for at least stb_window consecutive iterations. "
        "estimate is the cumulative mean at the final sample. "
        "error is the CLT-based absolute error (z_score * SEM) at the final sample. "
        "n is the total number of samples evaluated."
    )

    example_usage = 

    references = ("https://en.wikipedia.org/wiki/Central_limit_theorem",)

    def run(
        self,
        samples: SampleFrame,
        params: CLTAbsoluteParams,
    ) -> ConvergenceResult:

        target = pl.col(samples.target_col).cast(pl.Float64)

        agg_df = samples.df.select(
            [pl.len().alias("n"), target.mean().alias("mean"), target.std().alias("std")]
        )

        final = agg_df.row(0, named=True)
        n = final["n"]
        mean = final["mean"]
        std = final["std"] if final["std"] is not None else 0.0

        sem = std / (n**0.5) if n > 0 else 0.0
        z_score = float(norm.ppf(1.0 - (1.0 - params.confidence_level) / 2.0))
        abs_error = sem * z_score

        is_converged = abs_error <= params.threshold

        final_df = pl.DataFrame(
            {
                "n": [n],
                "mean": [mean],
                "std": [std],
                "sem": [sem],
                "abs_error": [abs_error],
                "is_converged": [is_converged],
            }
        )

        return ConvergenceResult(
            method=self.name,
            n=int(n),
            n_col="n",
            estimate=float(mean) if mean is not None else None,
            estimate_col="mean",
            error=float(abs_error),
            error_col="abs_error",
            is_converged=is_converged,
            is_converged_col="is_converged",
            data=final_df,
            diagnostics={
                "threshold": params.threshold,
                "confidence_level": params.confidence_level,
                "z_score": z_score,
            },
        )


========================================
File: src/mocaco/methods/__init__.py
========================================



import importlib
import pkgutil

for _finder, module_name, _ispkg in pkgutil.iter_modules(__path__):
    importlib.import_module(f"{__name__}.{module_name}")

