mineproductivity.analytics¶
Auto-generated API reference for the Analytics package, rendered from the package's own docstrings. For the narrative overview, dependency rules, and extension guide, see Packages -> Analytics.
analytics ¶
mineproductivity.analytics -- the platform's statistical and
analytical computation layer, built directly on top of kpis.
Answers the question the KPI Engine deliberately does not: given a
number (or a series of numbers), what does it mean? Analytics consumes
already-correct inputs (raw event rows via events.EventStore, and
already-computed kpis.KPIResult objects) and produces analytical
judgments about them -- trends, benchmarks, baselines, distributions,
confidence intervals, data-quality scores -- as new, equally
discoverable, equally versioned result objects.
Implements the Analytics Foundation (entities, value objects, metadata
classes, enums, exception hierarchy), the Metric Pipelines &
Aggregation, and the Statistical Primitives portions of
docs/architecture/06_Analytics_Engine_Design_Specification.md:
AnalyticsPipeline/PipelineStage/ModelStage compose ordered
stages over a TimeSeries; AggregationEngine group-and-reduces
kpis.KPIResult series, correctly delegating back to
kpis.KPIEngine wherever KPIMetadata.aggregation requires it
rather than averaging already-computed values; describe,
percentile, histogram, distribution, and
confidence_interval are the stateless, deterministic statistical
primitives every concrete AnalyticsModel calls internally, requiring
no third-party numerical dependency; rolling_mean/rolling_std/
rolling_apply are the sliding-window counterparts, reusing
statistics.py's own mean/standard-deviation computation rather than
duplicating it, and representing "not yet enough data" as an absent
point rather than a sentinel value; TrendModel/LinearTrendModel
fit a deterministic ordinary-least-squares line over a TimeSeries,
reusing statistics.py's own mean computation, and characterize the
observed window only -- never extrapolating beyond it (that is
forecasting, a separate, interface-only concern); BaselineModel/
RollingBaselineModel compute a trailing-window [mean - k*std,
mean + k*std] historical-norm band, consuming rolling_mean/
rolling_std directly rather than reimplementing moving-average
logic; BenchmarkModel/BandBenchmarkModel classify a
kpis.KPIResult's value against its own KPIMetadata.benchmark_bands,
read via Registry.metadata_for -- never a second, parallel band
schema -- respecting KPIMetadata.direction (inverting the comparison
for LOWER_IS_BETTER, comparing distance-from-target for
TARGET_IS_BEST); DataQualityScorer/MissingDataPolicy/
DataQualityStage grade a set of rows against required columns into a
completeness/validity/overall_score triple and compose
directly into an AnalyticsPipeline as a PipelineStage that gates
the pipeline below a caller-configurable minimum score; ForecastingModel
is an interface-only contract (_forecast(series, *, horizon, context)
-> ForecastResult) -- this package ships zero concrete subclasses of
it, deliberately: choosing a forecasting algorithm is a modeling
decision outside this package's charter (ADR-0006), left for a future,
independently-versioned plugin to implement against this stable
interface; AnomalyDetector is likewise an interface-only contract
(_detect(series, *, baseline, context) -> Sequence[AnomalyFlag])
built to be composed with the primitives this package already ships
(describe, Baseline, rolling_std) rather than a new
statistical foundation -- this package ships zero concrete subclasses of
it either, for the same ADR-0006 reasoning as ForecastingModel;
OutlierDetector completes the set of three interface-only contracts
(_detect(series, *, distribution, context) -> Sequence[OutlierFlag])
-- distinct from AnomalyDetector in scope (a static
DistributionSummary reference, mandatory, vs. a temporal
Baseline reference, optional), ships zero concrete subclasses for
the same reason. The three execution modes (§27-§29) are also
implemented: BatchAnalyticsRunner is a thin, named wrapper over
AnalyticsPipeline.run for the bounded, retrospective-report mode;
StreamingAnalyticsSession subscribes to an events.EventBus and
incrementally updates one or more IncrementalAccumulator\ s as new
envelopes arrive, without ever re-scanning the full EventStore, one
threading.Lock per accumulator key serializing concurrent updates
to that key; IncrementalAccumulator implements Welford's
online algorithm for O(1)-update, O(1)-memory streaming mean/variance --
the one genuinely new numerical primitive in this package, since
statistics.py's batch functions all require a materialized
Sequence[float] and would defeat the whole point of a streaming
accumulator if reused here instead. REGISTRY/register (§32-33)
specialize registry.Registry exactly as kpis._registry does one
layer down; LinearTrendModel, RollingBaselineModel, and
BandBenchmarkModel are all @register-decorated and discoverable
via REGISTRY at import time. AggregationEngine.reduce() is fully
implemented: it groups a plain numeric TimeSeries by scope field(s)
and reuses describe() verbatim per group, since
describe()'s own _DEFAULT_PERCENTILES guarantees a 50th
percentile (median) is always present, resolving the gap that
previously deferred this method (see its own docstring for the full
explanation). Every module described in
docs/architecture/06_Analytics_Engine_Design_Specification.md §6 is
now implemented; see the package's own README.md for the full slice
inventory.
analytics depends on core, ontology, events, registry,
plugins, connectors, and kpis -- and MUST NEVER import
decision, digital_twin, optimization, simulation,
agents, or visualization, none of which this package may see.
Everything documented here is part of the public API and can be
imported directly from mineproductivity.analytics, e.g.::
from mineproductivity.analytics import AnalyticsModel, TimeSeries
AnalyticsContext ¶
Bundles the collaborators an AnalyticsModel may need -- the
analytics-layer counterpart to KPIEngine's own constructor
bundle. Every field is optional except event_store because some
models (e.g. a pure statistical summary over an already-fetched
TimeSeries) need nothing else.
Examples:
>>> class _FakeStore: ...
>>> ctx = AnalyticsContext(event_store=_FakeStore())
>>> ctx.kpi_engine is None
True
AnalyticsModel ¶
Bases: ABC
The root of every registrable analytics strategy -- 'Analytics-
as-object,' the direct one-layer-up counterpart of kpis.BaseKPI.
A concrete leaf declares meta: ClassVar[AnalyticsMetadata] and
implements :meth:_analyze; everything else (input validation,
result envelope wrapping) is inherited, mirroring BaseKPI/
_compute exactly.
AnalyticsModel is deliberately not split into category families
the way kpis.BaseKPI is split by mining-domain namespace --
Analytics organizes by analytical function (trend, baseline,
benchmark, forecasting, anomaly, outlier), not by mining domain
namespace, so every category shares this one abstraction.
Thread safety. Every AnalyticsModel subclass MUST be
stateless across :meth:analyze calls -- no instance attribute is
mutated by :meth:_analyze -- so a single instance is safe to share
and invoke concurrently from multiple threads.
analyze ¶
Non-overridden orchestration: checks series has the
minimum number of observations this model declares needing
(meta.min_observations), then calls :meth:_analyze.
AggregationEngine ¶
Group-and-reduce over raw numeric series or over KPIResult
series, delegating correctly to kpis.KPIEngine whenever the
underlying KPI's Aggregation semantics require it.
backend is accepted and stored for constructor-shape consistency
with kpis.KPIEngine (design spec §36's "accepts an optional
ExecutionBackend, falls back to a plain-Python reference path when
none is supplied"), but neither :meth:reduce nor
:meth:reduce_kpi_results reads it: reduce_kpi_results delegates
to its own engine parameter's already-configured backend instead,
and reduce deliberately never needs one --
kpis.backends.ExecutionBackend.group_and_aggregate is keyed by
kpis.Aggregation (ADDITIVE/RATIO/AVERAGE/...), which
has no "median" member and is inherently a KPI aggregation
vocabulary; reduce is explicitly "no KPI semantics involved"
(see its own docstring), so bridging its sum/mean/median
vocabulary onto Aggregation would force an artificial, partly
impossible correspondence rather than genuine reuse. reduce is
therefore always the plain-Python reference path, satisfying §36's
fallback requirement unconditionally rather than conditionally.
reduce ¶
Pure statistical reduction over a series of plain numeric
observations -- no KPI semantics involved. Safe for any input
that is not itself a KPIResult series.
Groups series.points by the scope values named in by.by
(a point missing any of those scope keys is excluded from every
group -- "qualify, don't coerce": it cannot be classified, so it
is silently left out rather than mis-grouped under a fabricated
key), then reuses :func:~mineproductivity.analytics.statistics.describe
verbatim over each group's own sub-series. No new numerical
computation is introduced.
This was previously deferred because no existing
AnalyticsResult subtype appeared to expose reduction's
three named values individually. Re-examined for this phase:
describe()'s own _DEFAULT_PERCENTILES constant is fixed
at (50, 90, 99) (not caller-configurable), so
StatisticalSummary.percentiles[50] is guaranteed, not
merely likely, present for every non-empty group -- the earlier
concern about relying on an "undocumented," possibly-absent key
no longer applies once that guarantee is stated explicitly
(as it now is, here). Each of reduction's three values is
therefore always exactly derivable from the one returned
StatisticalSummary per group:
"mean"--summary.meandirectly."median"--summary.percentiles[50]."sum"--summary.mean * summary.n(an algebraic identity, not a second summation pass over the group).
The full StatisticalSummary is returned rather than a bare
float so a caller reading more than the one field they asked
for is never worse off, and so this method needs no result
shape narrower than the one statistics.py already provides
-- avoiding both a new result type and a second, parallel
"just the one number" abstraction next to the descriptive one
this package already has.
reduce_kpi_results ¶
Group-and-reduce over KPIResult observations, respecting
KPIMetadata.aggregation (kpis spec §10.2, §19's "RATIO-never-
averaged" rule) one layer up from where kpis itself enforces
it.
For ADDITIVE/CUMULATIVE-aggregation KPIs, this is a
direct sum. For every other aggregation kind (RATIO,
AVERAGE, WEIGHTED_AVERAGE, ROLLING, DERIVED),
this method does not average results' already-computed
.value\ s -- it re-invokes
engine.execute(code, window=window, scope=combined_scope)
over the union scope, exactly reusing kpis' own engine-level
ratio-correctness guarantee instead of re-deriving (and risking
getting wrong) the same rule a second time (§34).
A window keyword-only parameter is added beyond the design
spec's own illustrative signature: KPIEngine.execute requires
one and no other parameter here can supply it -- a necessary,
minimal, disclosed correction (the same kind already applied to
TimeSeries.from_kpi_results' timestamps parameter in the
Analytics Foundation phase).
Raises:
| Type | Description |
|---|---|
AnalyticsValidationError
|
If |
GroupBySpec ¶
AnomalyDetector ¶
Bases: AnalyticsModel, ABC
Contract for a future anomaly-detection plugin. THIS MODULE SHIPS
NO CONCRETE SUBCLASS, for the same reason as ForecastingModel
(§16): "is this point anomalous" is a modeling choice this package
does not make on the implementer's behalf. A future detector is
expected to be built on top of the primitives this package DOES ship
(describe, Baseline, rolling_std), not on a new
statistical foundation.
Leaves :meth:~mineproductivity.analytics.abstractions.AnalyticsModel._analyze
abstract (inherited, unoverridden) exactly as every other category
base in this package does (TrendModel, BaselineModel,
BenchmarkModel, ForecastingModel) -- only a concrete
subclass decides how its own _analyze relates to _detect,
since no concrete subclass ships here to make that decision on a
future plugin's behalf.
BaselineModel ¶
Bases: AnalyticsModel, ABC
Category base for self-referential historical-norm computation --
distinct from BenchmarkModel (§13), which compares against an
externally published target/band. A Baseline answers "is this
normal for this asset/site, historically," not "how does this
compare to the industry."
RollingBaselineModel ¶
Bases: BaselineModel
The default, concrete baseline strategy: trailing-window mean and
standard deviation, forming a [mean - k*std, mean + k*std] band.
Registered into analytics.REGISTRY at import time (§32-33),
mirroring how kpis.standard_library self-registers.
Consumes :func:~mineproductivity.analytics.rolling.rolling_mean/
:func:~mineproductivity.analytics.rolling.rolling_std directly for
the trailing-window computation -- no moving-average logic is
reimplemented here. The returned :class:~mineproductivity.analytics.result.Baseline
reflects the trailing window ending at the series' most recent
observation (the "current" historical norm); rolling.py already
represents "not yet enough trailing history" as an absent point
rather than a sentinel, so if no point in the series has a full
window (e.g. spec.periods exceeds the series length beyond what
spec.min_periods tolerates), this model falls back to a
whole-series baseline via :mod:mineproductivity.analytics.statistics's
own _mean/_population_stdev helpers, with a warning attached
-- never raises, matching the "qualify, don't coerce" rule already
governing every AnalyticsModel.
Examples:
>>> from datetime import datetime, timedelta, timezone
>>> from mineproductivity.analytics.timeseries import TimeSeriesPoint
>>> from mineproductivity.events.store import _InMemoryEventStore
>>> start = datetime(2026, 1, 1, tzinfo=timezone.utc)
>>> series = TimeSeries(points=tuple(
... TimeSeriesPoint(timestamp=start + timedelta(days=i), value=10.0)
... for i in range(5)
... ))
>>> model = RollingBaselineModel(spec=RollingSpec(periods=3, min_periods=3))
>>> context = AnalyticsContext(event_store=_InMemoryEventStore())
>>> result = model.analyze(series, context=context)
>>> result.mean, result.lower, result.upper
(10.0, 10.0, 10.0)
BatchAnalyticsRunner ¶
Runs one AnalyticsPipeline once over a bounded TimeSeries
and returns a single AnalyticsResult -- the 'normal,'
retrospective-report mode, in contrast to
StreamingAnalyticsSession's live, unbounded mode (§27).
Examples:
>>> from typing import ClassVar
>>> from mineproductivity.analytics.abstractions import AnalyticsModel
>>> from mineproductivity.analytics.metadata import AnalyticsCategory, AnalyticsMetadata
>>> from mineproductivity.analytics.pipeline import ModelStage
>>> from mineproductivity.events.store import _InMemoryEventStore
>>> class _Model(AnalyticsModel):
... meta: ClassVar[AnalyticsMetadata] = AnalyticsMetadata(
... code="TREND.Doctest", category=AnalyticsCategory.TREND,
... description="x", min_observations=0,
... )
... def _analyze(self, series, *, context):
... return AnalyticsResult(model_code="TEST")
>>> pipeline = AnalyticsPipeline(stages=(ModelStage(_Model()),))
>>> context = AnalyticsContext(event_store=_InMemoryEventStore())
>>> runner = BatchAnalyticsRunner(pipeline=pipeline, context=context)
>>> result = runner.run(TimeSeries(points=()))
>>> result.unwrap().model_code
'TEST'
BandBenchmarkModel ¶
Bases: BenchmarkModel
The default, concrete benchmarking strategy: classify a
KPIResult's value against its own KPIMetadata.benchmark_bands,
respecting KPIMetadata.direction -- HIGHER_IS_BETTER and
LOWER_IS_BETTER invert the comparison; TARGET_IS_BEST compares
distance-from-target rather than raw magnitude (design spec §13).
Registered into analytics.REGISTRY at import time (§32-33),
mirroring how kpis.standard_library self-registers.
:meth:benchmark is the primary entry point, matching the design
spec's own signature exactly: it operates on a single KPIResult
plus an explicit registry, not a TimeSeries, since benchmarking
is a point-in-time classification rather than a series computation.
_analyze (the AnalyticsModel contract every concrete model
must satisfy) is a thin adapter over the same classification logic:
it benchmarks the series' most recent point against the KPI code this
instance was constructed with, via the module-level kpis.REGISTRY
-- a bare TimeSeries carries no KPI code of its own (scope
only), so the constructor must supply the one otherwise-unavailable
piece of information, the same "necessary, minimal, disclosed"
correction already applied to TimeSeries.from_kpi_results'
timestamps parameter and AggregationEngine.reduce_kpi_results'
window parameter in earlier Analytics phases.
Examples:
>>> from mineproductivity.registry import Registry
>>> registry: Registry[str, type] = Registry(name="doctest")
>>> meta = KPIMetadata(
... name="PROD.TPH", code="PROD.TPH", official_name="Tonnes Per Hour",
... business_purpose="x", operational_question="x", business_meaning="x",
... formula="x", unit="t/h", dimensions=("Shift",), required_events=("CYCLE",),
... direction=Direction.HIGHER_IS_BETTER,
... benchmark_bands={"top_quartile": ">1200", "below_average": "<1000"},
... )
>>> _ = registry.register("PROD.TPH", object, metadata=meta)
>>> model = BandBenchmarkModel(kpi_code="PROD.TPH")
>>> result = KPIResult(code="PROD.TPH", value=1250.0, unit="t/h")
>>> classified = model.benchmark(result, registry=registry)
>>> classified.band
'top_quartile'
benchmark ¶
Classify result against its own registered
KPIMetadata.benchmark_bands/direction.
Returns a plain AnalyticsResult (never raising) rather than
the more specific BenchmarkResult when classification is
genuinely impossible -- no registered metadata for result.code,
or result.value is None (legitimately uncomputable, kpis spec
§10.1) -- since BenchmarkResult.value/.direction are
mandatory fields with no honest value to put there. This is the
same "qualify, don't coerce" widening AnalyticsModel.analyze
itself performs for insufficient-data series.
BenchmarkModel ¶
Bases: AnalyticsModel, ABC
Category base for benchmarking strategies -- classify a value
against an externally published target/band, distinct from
BaselineModel (§15)'s self-referential historical norm.
AnalyticsModelNotFoundError ¶
AnalyticsValidationError ¶
Bases: ValidationError
An AnalyticsMetadata, RollingSpec, TimeSeries, or other
Analytics value object failed validation -- e.g. an empty code,
a RollingSpec with neither (or both) of time_window/periods
set -- or a caller passed malformed arguments to an Analytics method,
e.g. AggregationEngine.reduce_kpi_results given an empty or
mixed-code sequence of KPIResult\ s.
AnalyticsVersionConflictError ¶
Bases: RegistrationError
A plugin attempted to re-register an existing AnalyticsModel
code with materially different metadata without a version bump,
mirroring kpis.KPIVersionConflictError.
InsufficientDataError ¶
Bases: ValidationError
Raised only where a caller explicitly requests raising behavior
instead of :meth:~mineproductivity.analytics.abstractions.AnalyticsModel.analyze's
default warning-carrying result for a series shorter than
AnalyticsMetadata.min_observations.
ForecastingModel ¶
Bases: AnalyticsModel, ABC
The contract a future forecasting plugin implements. THIS MODULE SHIPS NO CONCRETE SUBCLASS -- choosing a forecasting algorithm is a modeling decision this package deliberately does not make (§3.4). Defining the contract now lets a future, independently-versioned plugin register against a stable interface without waiting for a future revision of this specification.
Leaves :meth:~mineproductivity.analytics.abstractions.AnalyticsModel._analyze
abstract (inherited, unoverridden) exactly as every other category
base in this package does (TrendModel, BaselineModel,
BenchmarkModel) -- only a concrete subclass decides how its own
_analyze relates to _forecast, since no concrete subclass
ships here to make that decision on a future plugin's behalf.
IncrementalAccumulator ¶
Welford's online algorithm for numerically-stable, streaming mean
and variance: O(1) update per new observation, O(1) memory,
regardless of how many observations have been seen. This is the
algorithmic primitive both StreamingAnalyticsSession (§27) and,
optionally, BatchAnalyticsRunner (§28, for very large row counts
that should not be held in memory at once) use to avoid an O(n)
re-scan on every update. A well-known, deterministic, closed-form
numerical method -- not a statistical model and not machine
learning.
Thread safety. Unlike every AnalyticsModel subclass (which
MUST be stateless), IncrementalAccumulator is deliberately,
visibly mutable -- update() changes internal running-mean/
running-variance state. It is therefore not safe to share a
single instance across threads without external synchronization.
StreamingAnalyticsSession (§27) owns exactly one
IncrementalAccumulator per tracked key and is responsible for
serializing concurrent update() calls to the same key; this
class does not synchronize itself.
Examples:
>>> accumulator = IncrementalAccumulator()
>>> for value in (2.0, 4.0, 4.0, 4.0, 5.0, 5.0, 7.0, 9.0):
... accumulator.update(value)
>>> summary = accumulator.snapshot()
>>> summary.n, summary.mean
(8, 5.0)
>>> round(summary.std, 4)
2.0
update ¶
Fold value into the running mean/variance/min/max in O(1)
time -- never re-reads any prior observation.
snapshot ¶
The current StatisticalSummary, computed in O(1) time
from the running state -- never re-scans prior observations.
percentiles is always empty (see module docstring).
Raises:
| Type | Description |
|---|---|
AnalyticsValidationError
|
If no observation has been recorded yet -- mirrors
|
AnalyticsCategory ¶
Bases: Enum
Closed enum -- adding a member is a governance-reviewed change,
mirroring kpis.Aggregation's closed-enum rule.
AnalyticsMetadata
dataclass
¶
AnalyticsMetadata(code, *, name='', description, tags=frozenset(), attributes=dict(), category, min_observations=2, version='1.0.0')
Bases: BaseMetadata
The minimal registration schema for a discoverable AnalyticsModel
-- deliberately not a copy of kpis.KPIMetadata's 29-field
governance schema: an AnalyticsModel is a computational strategy,
not an audited business metric.
BaseMetadata.name has no default upstream and an AnalyticsModel's
code (e.g. "TREND.Linear") already serves as its identifier, so
name defaults to code (via :meth:_normalize) whenever a caller
does not supply one explicitly.
Examples:
>>> meta = AnalyticsMetadata(
... code="TREND.Linear", category=AnalyticsCategory.TREND,
... description="Ordinary least squares linear trend fit.",
... )
>>> meta.code
'TREND.Linear'
>>> meta.name
'TREND.Linear'
>>> meta.min_observations
2
>>> AnalyticsMetadata(code="", category=AnalyticsCategory.TREND, description="x")
Traceback (most recent call last):
...
mineproductivity.analytics.exceptions.AnalyticsValidationError: AnalyticsMetadata.code must not be empty
OutlierDetector ¶
Bases: AnalyticsModel, ABC
Contract for a future outlier-detection plugin -- distinct from
AnomalyDetector (§18) in scope: an outlier is a single
observation unusual relative to a static distribution (e.g. via IQR
or z-score against DistributionSummary, §23); an anomaly is a
point unusual relative to a temporal baseline (§15). The two
interfaces are kept separate because the reference data they compare
against differs structurally (a distribution vs. a rolling
baseline), even though a future implementation could satisfy both.
THIS MODULE SHIPS NO CONCRETE SUBCLASS, for the same reason as
ForecastingModel/AnomalyDetector.
Leaves :meth:~mineproductivity.analytics.abstractions.AnalyticsModel._analyze
abstract (inherited, unoverridden) exactly as every other category
base in this package does (TrendModel, BaselineModel,
BenchmarkModel, ForecastingModel, AnomalyDetector) --
only a concrete subclass decides how its own _analyze relates to
_detect, since no concrete subclass ships here to make that
decision on a future plugin's behalf.
AnalyticsPipeline ¶
An ordered Sequence[PipelineStage], run in order over one input
TimeSeries. Mirrors the shape of the KPI Cookbook's "Putting
Everything Together" narrative -- fetch, clean, aggregate, analyze --
generalized to any analytical question.
A pipeline never contains model-specific branching (mirroring
KPIEngine's own "holds no metric logic" invariant one layer up):
:meth:run calls each stage's :meth:~PipelineStage.process
uniformly and has no knowledge of which concrete AnalyticsModel a
ModelStage wraps.
Examples:
>>> from typing import ClassVar
>>> from mineproductivity.analytics.metadata import AnalyticsCategory, AnalyticsMetadata
>>> class _FakeStore: ...
>>> class _Model(AnalyticsModel):
... meta: ClassVar[AnalyticsMetadata] = AnalyticsMetadata(
... code="TREND.Doctest", category=AnalyticsCategory.TREND,
... description="x", min_observations=0,
... )
... def _analyze(self, series, *, context):
... return AnalyticsResult(model_code="TEST")
>>> pipeline = AnalyticsPipeline(stages=(ModelStage(_Model()),))
>>> ctx = AnalyticsContext(event_store=_FakeStore())
>>> series = TimeSeries(points=())
>>> pipeline.run(series, context=ctx).unwrap().model_code
'TEST'
>>> AnalyticsPipeline(stages=()).run(series, context=ctx).is_err
True
run ¶
Runs every stage in order; a non-terminal stage's output feeds
the next stage's input. The last stage MUST be a ModelStage
(or otherwise yield an AnalyticsResult) or this returns
Result.err(AnalyticsValidationError(...)).
ModelStage ¶
Bases: PipelineStage
A terminal stage that hands the (by now cleaned/aggregated)
series to one AnalyticsModel and yields its AnalyticsResult.
Examples:
>>> from typing import ClassVar
>>> from mineproductivity.analytics.metadata import AnalyticsCategory, AnalyticsMetadata
>>> class _FakeStore: ...
>>> class _Model(AnalyticsModel):
... meta: ClassVar[AnalyticsMetadata] = AnalyticsMetadata(
... code="TREND.Doctest", category=AnalyticsCategory.TREND,
... description="x", min_observations=0,
... )
... def _analyze(self, series, *, context):
... return AnalyticsResult(model_code="TEST")
>>> stage = ModelStage(_Model())
>>> ctx = AnalyticsContext(event_store=_FakeStore())
>>> series = TimeSeries(points=())
>>> stage.process(series, context=ctx).model_code
'TEST'
PipelineStage ¶
Bases: ABC
One step in an AnalyticsPipeline. Stateless and composable --
a new stage never requires changing AnalyticsPipeline itself.
Return type is deliberately TimeSeries | AnalyticsResult, not
just TimeSeries: :class:ModelStage -- itself a PipelineStage
-- always yields an AnalyticsResult, so the honest contract every
implementation (terminal or not) must satisfy is the union of both.
This widens, rather than contradicts, the design spec's own
abstract-method signature, which showed only TimeSeries while
describing ModelStage as yielding an AnalyticsResult in the
very next paragraph -- a necessary, minimal, disclosed correction.
process
abstractmethod
¶
Transform one TimeSeries into another (e.g. missing-data
handling, aggregation) -- OR, for a terminal stage, wrap the
series into an AnalyticsResult (see :class:ModelStage).
DataQualityScorer ¶
Produces a :class:~mineproductivity.analytics.result.DataQualityScore
for a set of rows against a set of required columns -- the
Analytics-layer counterpart of BaseKPI._required_columns()'s
missing-column check, generalized into a graded score rather than a
binary present/absent warning.
Examples:
>>> scorer = DataQualityScorer()
>>> rows = [{"payload_t": 220.0, "operating_h": 8.0}, {"payload_t": None, "operating_h": 7.5}]
>>> score = scorer.score(rows, required_columns=("payload_t", "operating_h"))
>>> score.completeness
0.75
>>> score.overall_score
0.75
DataQualityStage ¶
DataQualityStage(*, required_columns, missing_data_policy=MissingDataPolicy.FLAG_ONLY, min_score=0.0)
Bases: PipelineStage
The PipelineStage wrapper composing DataQualityScorer (and
a MissingDataPolicy) into an AnalyticsPipeline -- runs before
any statistical or model stage.
Below min_score, :meth:process returns the computed
DataQualityScore itself instead of a TimeSeries, gating the
pipeline via AnalyticsPipeline.run's existing non-terminal-result
rule (see module docstring). At or above min_score, it returns a
TimeSeries cleaned per missing_data_policy. The default
min_score=0.0 never gates (overall_score is always >= 0),
matching design spec §9's own worked example, which passes
DataQualityStage straight through to a following ModelStage.
Examples:
>>> from datetime import datetime, timezone
>>> from mineproductivity.events.store import _InMemoryEventStore
>>> points = (
... TimeSeriesPoint(timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), value=1.0,
... scope={"pit": "north"}),
... TimeSeriesPoint(timestamp=datetime(2026, 1, 2, tzinfo=timezone.utc), value=2.0, scope={}),
... )
>>> stage = DataQualityStage(required_columns=("pit",), missing_data_policy=MissingDataPolicy.FORWARD_FILL)
>>> context = AnalyticsContext(event_store=_InMemoryEventStore())
>>> cleaned = stage.process(TimeSeries(points=points), context=context)
>>> isinstance(cleaned, TimeSeries)
True
>>> cleaned.points[1].scope["pit"]
'north'
MissingDataPolicy ¶
Bases: Enum
Closed enum, mirroring the closed-enum-change-requires-governance
rule already established for kpis.Aggregation. All four policies
are deterministic, closed-form operations.
AnalyticsResult
dataclass
¶
Bases: BaseValueObject
The shared envelope every concrete Analytics result composes.
Mirrors KPIResult's role as the one result shape -- except
Analytics genuinely has more than one kind of output, so this is a
shared base rather than kpis' single concrete type.
Examples:
>>> AnalyticsResult(model_code="TREND.Linear").model_code
'TREND.Linear'
>>> AnalyticsResult(warnings=("insufficient data",)).warnings
('insufficient data',)
AnomalyFlag
dataclass
¶
Bases: BaseValueObject
One flagged observation within a series -- deliberately a plain
BaseValueObject, not an :class:AnalyticsResult subclass, since
it represents one point, not a summary result about a series.
Examples:
>>> AnomalyFlag(timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc),
... observed_value=500.0, expected_value=100.0, severity="high").severity
'high'
Baseline
dataclass
¶
Baseline(mean, std, lower, upper, spec, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
A self-referential historical-norm band -- distinct from
:class:BenchmarkResult, which compares against an externally
published target/band.
Examples:
>>> b = Baseline(mean=100.0, std=5.0, lower=90.0, upper=110.0,
... spec=RollingSpec(periods=14))
>>> b.lower, b.upper
(90.0, 110.0)
BenchmarkResult
dataclass
¶
BenchmarkResult(kpi_code, value, band, direction, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
Classification of a KPIResult against its own
KPIMetadata.benchmark_bands/direction.
Examples:
>>> BenchmarkResult(kpi_code="PROD.TPH", value=1200.0,
... band="top_quartile", direction=Direction.HIGHER_IS_BETTER).band
'top_quartile'
ConfidenceInterval
dataclass
¶
ConfidenceInterval(lower, upper, confidence, method, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
A closed-form (normal or Student's t) interval around a
sample mean.
Examples:
>>> ci = ConfidenceInterval(lower=9.5, upper=10.5, confidence=0.95, method="t")
>>> ci.confidence
0.95
DataQualityScore
dataclass
¶
DataQualityScore(completeness, validity, overall_score, reasons, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
A graded completeness/validity judgment over a set of rows
against a set of required columns -- the single number a pipeline
gates on is overall_score.
Examples:
>>> q = DataQualityScore(completeness=1.0, validity=0.9,
... overall_score=0.9, reasons=("2 rows out of range",))
>>> q.overall_score
0.9
DistributionSummary
dataclass
¶
DistributionSummary(mean, std, skewness, kurtosis, percentiles, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
A superset of :class:StatisticalSummary that adds shape
descriptors (skewness, kurtosis) that describe()
deliberately omits, keeping describe() cheap.
Examples:
>>> d = DistributionSummary(mean=1.0, std=0.5, skewness=0.1,
... kurtosis=3.0, percentiles={50: 1.0})
>>> d.skewness
0.1
ForecastResult
dataclass
¶
ForecastResult(horizon, predicted, intervals, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
horizon future points, each with a point estimate and an
uncertainty band. No producer of this result type ships in this
release -- :class:~mineproductivity.analytics.abstractions.AnalyticsModel
subclasses that produce it are a future forecasting plugin's job.
Examples:
>>> f = ForecastResult(horizon=1, predicted=(105.0,),
... intervals=(ConfidenceInterval(lower=100.0, upper=110.0,
... confidence=0.95, method="normal"),))
>>> f.horizon
1
Histogram
dataclass
¶
Histogram(bin_edges, counts, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
A binned frequency count over a sequence of values.
Examples:
>>> h = Histogram(bin_edges=(0.0, 1.0, 2.0), counts=(3, 5))
>>> len(h.counts) == len(h.bin_edges) - 1
True
OutlierFlag
dataclass
¶
Bases: BaseValueObject
One observation unusual relative to a static distribution --
deliberately a plain BaseValueObject, not an
:class:AnalyticsResult subclass, for the same reason as
:class:AnomalyFlag.
Examples:
StatisticalSummary
dataclass
¶
StatisticalSummary(n, mean, std, minimum, maximum, percentiles, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
Descriptive statistics over a
:class:~mineproductivity.analytics.timeseries.TimeSeries -- the
Analytics-layer equivalent of a spreadsheet's "Descriptive
Statistics" tool.
Examples:
>>> s = StatisticalSummary(n=3, mean=2.0, std=1.0, minimum=1.0,
... maximum=3.0, percentiles={50: 2.0})
>>> s.percentiles[50]
2.0
TrendResult
dataclass
¶
TrendResult(slope, intercept, r_squared, direction, window, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: AnalyticsResult
The outcome of a trend-fitting model -- characterizes an observed series' direction and fit quality; never a prediction of a future value (that is forecasting, a separate concern).
Examples:
>>> TrendResult(slope=1.5, intercept=0.0, r_squared=0.98,
... direction="increasing", window=RollingSpec(periods=7)).direction
'increasing'
StreamingAnalyticsSession ¶
A long-lived session that subscribes to an events.EventBus
and incrementally updates one or more IncrementalAccumulator\ s
as new envelopes arrive, without ever re-scanning the full
historical EventStore. The live-operations-center counterpart of
BatchAnalyticsRunner (§28).
A session is long-lived but restartable-from-cold: if a session's
process restarts, its IncrementalAccumulator\ s start from zero
rather than attempting to replay history through the bus -- a caller
that needs a warm-started session is expected to seed the
accumulator from a BatchAnalyticsRunner pass over the relevant
window first (§28), then call :meth:start to continue
incrementally. This class performs no such seeding automatically.
Thread safety. The Event Framework's own concurrency contract
guarantees EventBus.publish() is only called after the
corresponding append() has confirmed durability, but makes no
guarantee about which thread a subscriber's handler runs on, nor
that a single subscriber never receives two concurrent calls. This
class does not assume single-threaded delivery: it serializes
concurrent update() calls per accumulator key via one
threading.Lock per key (mirroring kpis.ResultCache's own
per-key-write concurrency contract) -- independent keys have no
shared state and update fully in parallel.
Examples:
>>> from mineproductivity.events.bus import _InMemoryEventBus
>>> from mineproductivity.events.canonical import CycleEvent
>>> from mineproductivity.events.envelope import EventEnvelope, EventMetadata
>>> from mineproductivity.events.identifier import EventID
>>> from mineproductivity.events.versioning import EventVersion
>>> from datetime import datetime, timezone
>>> bus = _InMemoryEventBus()
>>> session = StreamingAnalyticsSession(
... bus=bus, accumulators={"payload_t": IncrementalAccumulator()}
... )
>>> subscription = session.start()
>>> now = datetime(2026, 1, 1, tzinfo=timezone.utc)
>>> envelope = EventEnvelope(
... event_id=EventID.generate(), version=EventVersion(),
... payload=CycleEvent(equipment_id="HT-1", shift_id="A", queue_min=1.0,
... spot_min=0.5, load_min=2.0, haul_min=8.0, dump_min=1.0,
... return_min=6.0, payload_t=220.0),
... event_time_utc=now, processing_time_utc=now, ingestion_time_utc=now,
... metadata=EventMetadata(name="cycle", source_system="test"),
... )
>>> bus.publish(envelope)
>>> session.snapshot("payload_t").mean
220.0
>>> subscription.cancel()
TimeSeries
dataclass
¶
Bases: BaseValueObject
An ordered (by timestamp) sequence of :class:TimeSeriesPoint\ s
-- the one series shape every statistical primitive, rolling
function, and category model in this package operates on; there is
no second, parallel "just a list of floats" convention anywhere in
the public API.
Examples:
>>> ts = TimeSeries(points=(
... TimeSeriesPoint(timestamp=datetime(2026, 1, 2), value=2.0),
... TimeSeriesPoint(timestamp=datetime(2026, 1, 1), value=1.0),
... ))
>>> len(ts)
2
>>> ts.values()
(1.0, 2.0)
values ¶
The plain numeric values, in timestamp order -- the shape
several statistical primitives (percentile, histogram,
distribution) need, since they have no meaningful use for
the timestamp dimension.
from_kpi_results
classmethod
¶
Wrap a Sequence[KPIResult] into a TimeSeries.
KPIResult itself carries no timestamp field, so the caller
assembling a TimeSeries from KPI results must supply
timestamps alongside the results being wrapped, one per
result, in the same order.
from_event_query
classmethod
¶
Wrap the result of EventStore.query(query) into a
TimeSeries, reading value_field off each envelope's
payload -- for direct statistical description of raw event data
without going through a KPI at all. Requests only value_field
(plus the equipment_id/shift_id fields every
BaseEvent already carries, used to populate each point's
scope), never the full envelope payload.
TimeSeriesPoint
dataclass
¶
Bases: BaseValueObject
One observation: a timestamp and a value, plus the originating
scope (mirrors KPIResult.scope).
Examples:
>>> TimeSeriesPoint(timestamp=datetime(2026, 1, 1), value=1200.0, scope={"pit": "north"}).value
1200.0
LinearTrendModel ¶
Bases: TrendModel
The default, concrete trend strategy: ordinary-least-squares
linear fit over a TimeSeries' (timestamp, value) pairs. Fully
deterministic and closed-form -- no forecasting, no extrapolation
beyond the observed window.
Registered into analytics.REGISTRY at import time (§32-33),
mirroring how kpis.standard_library self-registers.
Examples:
>>> from datetime import datetime, timedelta, timezone
>>> from mineproductivity.analytics.timeseries import TimeSeriesPoint
>>> from mineproductivity.events.store import _InMemoryEventStore
>>> start = datetime(2026, 1, 1, tzinfo=timezone.utc)
>>> series = TimeSeries(points=tuple(
... TimeSeriesPoint(timestamp=start + timedelta(days=i), value=float(i))
... for i in range(5)
... ))
>>> context = AnalyticsContext(event_store=_InMemoryEventStore())
>>> result = LinearTrendModel().analyze(series, context=context)
>>> result.direction
'increasing'
>>> round(result.r_squared, 4)
1.0
TrendModel ¶
RollingSpec
dataclass
¶
Bases: BaseValueObject
Either a time-based rolling window (delegates to
:class:~mineproductivity.kpis.RollingWindow directly) or a
count-based window over the last periods observations of a
:class:~mineproductivity.analytics.timeseries.TimeSeries, regardless
of how much wall-clock time they span -- which matters for
irregularly-sampled series (e.g. shift-level KPIResult s, which do
not arrive at a fixed cadence). Exactly one of time_window/
periods is set.
Examples:
>>> RollingSpec(periods=7).periods
7
>>> RollingSpec(time_window=None, periods=None)
Traceback (most recent call last):
...
mineproductivity.analytics.exceptions.AnalyticsValidationError: RollingSpec requires exactly one of time_window or periods
register ¶
Register cls into :data:REGISTRY, keyed by cls.meta.code.
Raises:
| Type | Description |
|---|---|
AnalyticsValidationError
|
If |
AnalyticsVersionConflictError
|
If |
rolling_apply ¶
Apply fn over a sliding spec-shaped window ending at each
observation in series, in timestamp order. Points before
spec.min_periods observations are available are omitted from
the result entirely -- represented as an absent point rather than a
sentinel value (e.g. NaN), so a caller cannot mistake "not yet
enough data" for a computed zero (mirroring kpis.KPIResult's
"never silently returns zero" rule).
Examples:
>>> from datetime import datetime, timezone
>>> points = tuple(
... TimeSeriesPoint(timestamp=datetime(2026, 1, i, tzinfo=timezone.utc), value=float(i))
... for i in range(1, 5)
... )
>>> series = TimeSeries(points=points)
>>> spec = RollingSpec(periods=2, min_periods=2)
>>> result = rolling_apply(series, spec, sum)
>>> len(result)
3
>>> result.values()
(3.0, 5.0, 7.0)
rolling_mean ¶
The rolling arithmetic mean over spec's sliding window.
Examples:
rolling_std ¶
The rolling population standard deviation (ddof=0, matching
:func:~mineproductivity.analytics.statistics.describe's own
convention) over spec's sliding window.
Examples:
>>> from datetime import datetime, timezone
>>> points = tuple(
... TimeSeriesPoint(timestamp=datetime(2026, 1, i, tzinfo=timezone.utc), value=v)
... for i, v in enumerate([1.0, 2.0, 3.0, 4.0], start=1)
... )
>>> result = rolling_std(TimeSeries(points=points), RollingSpec(periods=2, min_periods=2))
>>> result.values()
(0.5, 0.5, 0.5)
confidence_interval ¶
Closed-form only: normal-approximation or Student's t-interval around the sample mean. Resampling-based methods (e.g. bootstrap) are a documented future extension (§37), not shipped now, to keep the reference implementation's dependency footprint and determinism guarantees simple.
Both methods use the sample standard deviation (Bessel's
correction, ddof=1) to estimate the standard error, the standard
convention for confidence intervals -- distinct from, and not to be
confused with, :func:describe/:func:distribution's own
population-std (ddof=0) convention for plain descriptive
statistics. Requires at least 2 observations for either method: a
single observation cannot support a sample-based standard-error
estimate.
Examples:
describe ¶
The Analytics-layer equivalent of a spreadsheet's "Descriptive
Statistics" tool, over any TimeSeries -- raw event values or
KPIResult series alike.
Examples:
>>> from datetime import datetime, timezone
>>> from mineproductivity.analytics.timeseries import TimeSeriesPoint
>>> series = TimeSeries(points=(
... TimeSeriesPoint(timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), value=1.0),
... TimeSeriesPoint(timestamp=datetime(2026, 1, 2, tzinfo=timezone.utc), value=2.0),
... TimeSeriesPoint(timestamp=datetime(2026, 1, 3, tzinfo=timezone.utc), value=3.0),
... ))
>>> summary = describe(series)
>>> summary.n, summary.mean
(3, 2.0)
distribution ¶
A superset of :func:describe that adds shape descriptors
(skewness, kurtosis) that describe() deliberately omits,
keeping describe() cheap and distribution() the more
complete, slightly more expensive call for callers who explicitly
need distribution shape.
Examples:
histogram ¶
bins as an int requests that many equal-width bins spanning
[min(values), max(values)]; bins as a Sequence[float]
requests caller-supplied edges (e.g. the site's own published
benchmark bands, §13, when a histogram is being drawn against them).
Examples:
percentile ¶
Linear-interpolation percentile (the same convention NumPy's
default numpy.percentile and pandas' default .quantile use),
named explicitly rather than left to a caller to discover which
interpolation convention a third-party library defaults to.
q is on the 0-100 percentile scale (q=50 is the median),
matching StatisticalSummary.percentiles' own integer keys.
Examples: