analytics

analytics

Aggregated statistics engine for the Marianne Dashboard.

Computes dashboard-wide analytics from daemon job data via DaemonStateAdapter. All methods use a TTL-based cache to avoid hammering IPC on concurrent page loads.

Aggregation logic lives in module-level pure functions so it can be tested independently of the cache and IPC layers.

Attributes

Classes

DaemonAnalytics

DaemonAnalytics(adapter, cache_ttl=10.0)

Compute aggregate statistics from daemon job data.

Parameters

adapter: A StateBackend (typically DaemonStateAdapter) used to fetch live job data. cache_ttl: Default time-to-live in seconds for cached results (default 10.0).

Source code in src/marianne/dashboard/services/analytics.py

def __init__(self, adapter: StateBackend, cache_ttl: float = 10.0) -> None:
    self._adapter = adapter
    self._cache_ttl = cache_ttl
    self._cache: dict[str, tuple[float, Any]] = {}

Functions

get_stats async

get_stats()

Dashboard stats: total, running, completed, failed, success_rate, total_spend, throughput_sheets_per_hour.

Source code in src/marianne/dashboard/services/analytics.py

async def get_stats(self) -> dict[str, Any]:
    """Dashboard stats: total, running, completed, failed, success_rate,
    total_spend, throughput_sheets_per_hour.
    """
    return await self._compute("stats", _aggregate_stats)

cost_rollup async

cost_rollup()

Cost breakdown by job, total spend, avg cost per job.

Source code in src/marianne/dashboard/services/analytics.py

async def cost_rollup(self) -> dict[str, Any]:
    """Cost breakdown by job, total spend, avg cost per job."""
    return await self._compute("cost_rollup", _aggregate_costs)

validation_stats async

validation_stats()

Validation pass rates by rule type, overall pass rate.

Source code in src/marianne/dashboard/services/analytics.py

async def validation_stats(self) -> dict[str, Any]:
    """Validation pass rates by rule type, overall pass rate."""
    return await self._compute("validation_stats", _aggregate_validations)

error_breakdown async

error_breakdown()

Error counts by category: transient, rate_limit, permanent.

Source code in src/marianne/dashboard/services/analytics.py

async def error_breakdown(self) -> dict[str, Any]:
    """Error counts by category: transient, rate_limit, permanent."""
    return await self._compute("error_breakdown", _aggregate_errors)

duration_stats async

duration_stats()

Avg sheet duration, total job durations, slowest sheets.

Source code in src/marianne/dashboard/services/analytics.py

async def duration_stats(self) -> dict[str, Any]:
    """Avg sheet duration, total job durations, slowest sheets."""
    return await self._compute("duration_stats", _aggregate_durations)