mineproductivity.decision¶
Auto-generated API reference for the Decision package, rendered from the package's own docstrings. For the narrative overview, dependency rules, and extension guide, see Packages -> Decision.
decision ¶
mineproductivity.decision -- the platform's prescriptive layer,
built directly on top of analytics.
Answers the question analytics deliberately does not: given a
trend, a benchmark classification, a baseline, or a confidence
interval, what should the business actually do about it? decision
consumes already-correct kpis.KPIResult/analytics.AnalyticsResult
evidence and produces recommended, ranked, explained, actionable
decisions as new, equally discoverable, equally versioned result
objects -- never recomputing a KPI value, a trend, or a benchmark
classification itself.
Phase 07.1 -- Decision Intelligence Foundation implements every
component required before concrete decision models can exist:
DecisionModel (ABC)/DecisionContext (§8); DecisionMetadata/
DecisionCategory (§30); the full DecisionResult family (§28);
Threshold (§13); DecisionPipeline/PipelineStage/ModelStage
(§9); REGISTRY/register (§32); and the full exception hierarchy.
Phase 07.2 -- Decision Rule Engine implements rule evaluation,
policy governance, and the default decision strategy: Rule/
RuleEngine/RuleEngineStage (§10); Policy/DecisionStatus
(§12); DecisionStrategy (ABC)/ThresholdDecisionStrategy (§14),
self-registered into REGISTRY at import time.
Phase 07.3 -- Decision Intelligence Analysis Layer implements the
analytical capabilities that operate on already-produced
Recommendation/DecisionResult objects: DecisionScorer/
ConfidenceScorer (§23, §24) -- the numeric weight backing ranking
and the trust weight backing how strongly a recommendation should be
acted on, the latter deriving from analytics.DataQualityScore when
present in a DecisionContext's evidence, never recomputing data
quality itself; RankingStrategy (ABC)/WeightedScoreRanking (§16)
-- orders Recommendation\ s by DecisionScore.value, descending,
self-registered into REGISTRY; ExplanationBuilder/
ExplanationStage (§17) -- walks a Recommendation's provenance
into a structured, evidence-linked Explanation; ActionPrioritizer
(§20) -- assigns each RankedRecommendation an ActionPriority
(urgency/impact/effort), a distinct question from ranking; and
RootCauseAnalyzer (§18) -- an interface-only ABC with zero concrete
subclasses by design (ADR-0007), defining the contract a future causal-
inference plugin implements. DecisionContext gained a second,
Phase-07.3 extension (recommendations, alongside Phase 07.2's
triggered_rules) for the same reason: WeightedScoreRanking/
ExplanationStage operate over a batch of already-produced
Recommendation\ s, not raw evidence (see
abstractions.DecisionContext's own docstring).
Phase 07.4 -- Decision Operational Services completes the package:
WhatIfEngine (§19) -- an interface-only ABC with zero concrete
subclasses by design, the decision-layer counterpart to
RootCauseAnalyzer, reusing events.AsOf's already-reserved
scenario field rather than inventing a second scenario concept;
ActionPlanner (§21) -- sequences prioritized actions into an
ActionPlan, respecting declared dependencies via its own narrow,
self-contained topological ordering (deliberately not
kpis.DependencyGraph, per §33's own anti-pattern entry);
AlertGenerator (§22) -- produces an Alert from a
ThresholdBreach or a high-severity Recommendation, a pure value-
object factory with no channel-delivery side effects;
RealTimeDecisionSession/BatchDecisionRunner (§25, §26) -- the
two execution modes, the live event-driven counterpart and the bounded,
scheduled-report counterpart of running one DecisionPipeline, both
composing kpis.KPIEngine/analytics.BatchAnalyticsRunner rather
than recomputing anything themselves; and DecisionAuditTrail/
DecisionAuditEntry (§27, §28) -- the append-only accountability
record every operationally-actionable DecisionPipeline run should
feed, optionally wired into both execution modes. recommendation.py
(§15, §6) holds the recommendation-generation logic with no public API
of its own, per §6's own entry for it -- ThresholdDecisionStrategy's
methods delegate Recommendation construction to it, keeping exactly
one summary-text/traceability format package-wide. This completes every
module design spec §6 enumerates -- decision is now feature-complete
per the Reference Implementation Blueprint. See the package's own
README.md for the full slice inventory and phase status.
decision depends on core, events, ontology, registry,
plugins, connectors, kpis, and analytics -- and MUST
NEVER import digital_twin, simulation, optimization,
visualization, or agents, none of which this package may see.
Everything documented here is part of the public API and can be
imported directly from mineproductivity.decision, e.g.::
from mineproductivity.decision import DecisionModel, DecisionContext
Rule ¶
Reuses core.BaseSpecification (&/|/~ composable)
exactly as events.EventFilter already does -- rule evaluation is
never a bespoke DSL in this platform.
DecisionContext ¶
DecisionContext(*, kpi_results, analytics_results, scope, event_store=None, as_of=None, triggered_rules=None, recommendations=None)
Bundles the collaborators and scope a DecisionModel needs --
the decision-layer counterpart to analytics.AnalyticsContext,
one layer up. Carries the KPIResult/AnalyticsResult evidence
already gathered, the scope the decision concerns (e.g. a specific
pit or shift), and optionally an EventStore/AsOf framing a
point-in-time or hypothetical-scenario view for future root-cause/
what-if use.
triggered_rules is populated by :class:~mineproductivity.decision.rules.RuleEngineStage
(design spec §10) as a pipeline runs, so a downstream
:class:~mineproductivity.decision.strategy.ThresholdDecisionStrategy
never has to re-run rule evaluation itself -- a necessary, minimal,
disclosed extension of this class beyond design spec §8's own
pseudocode (which shows no such field), added in Phase 07.2 because
§10's RuleEngineStage explicitly needs some way to pass its
result to the next stage, and this class is the only vehicle a
PipelineStage.process has for doing so. Defaults to None,
meaning "no RuleEngineStage has run yet" -- deliberately
not (), which is instead the legitimate, common result of a
RuleEngineStage that ran and found nothing triggered. Collapsing
both into the same falsy default would make "not yet computed" and
"computed as empty" indistinguishable to a downstream consumer
(caught and fixed during this phase's own QA review, before it
could silently defeat the "never re-run" guarantee for precisely
the most common case).
recommendations is a Phase 07.3 extension of the same shape and
for the same reason as triggered_rules: design spec §16's
WeightedScoreRanking and §17's ExplanationStage are, per
their own pseudocode, DecisionModel/PipelineStage
implementations that operate over a batch of already-produced
Recommendation\ s, not over raw KPIResult/AnalyticsResult
evidence -- yet _decide/process receive only a
DecisionContext, the one vehicle available for a caller (or an
upstream, non-terminal pipeline stage) to hand a batch across. Also
defaults to None for consistency with triggered_rules's own
field shape -- though, unlike triggered_rules, no consumer in
this phase currently branches differently on None versus ()
(:class:~mineproductivity.decision.ranking.WeightedScoreRanking
and :class:~mineproductivity.decision.explanation.ExplanationStage
both treat "not supplied" and "supplied, empty" identically, via a
plain falsy check); the distinction is preserved as a matter of
shape consistency and to leave room for a future consumer that
does need to tell the two apart, not because anything here relies
on it today.
Examples:
>>> ctx = DecisionContext(kpi_results=(), analytics_results=(), scope={"pit": "north"})
>>> ctx.scope
{'pit': 'north'}
>>> ctx.event_store is None
True
>>> ctx.triggered_rules is None
True
>>> ctx.recommendations is None
True
DecisionModel ¶
Bases: ABC
The root of every registrable decision strategy -- 'Decision-as-
object,' the direct counterpart of kpis.BaseKPI and
analytics.AnalyticsModel, one and two layers down respectively.
A concrete leaf declares meta: ClassVar[DecisionMetadata] and
implements :meth:_decide; everything else (input validation, result
envelope wrapping) is inherited.
Thread safety. Every DecisionModel subclass MUST be stateless
across :meth:decide calls -- no instance attribute is mutated by
:meth:_decide -- so a single instance is safe to share and invoke
concurrently from multiple threads.
decide ¶
Non-overridden orchestration: validates context carries at
least one KPIResult or AnalyticsResult to reason over,
then calls :meth:_decide.
AlertGenerator ¶
Produces an Alert from a ThresholdBreach (§13) or a
high-severity Recommendation (§15). A concrete, non-pluggable
utility -- alert channel delivery (email, SMS, a dashboard push)
is explicitly out of scope for this package (§4);
AlertGenerator produces the Alert value object only, never
sends it anywhere, keeping it a pure, easily-tested function with no
I/O side effects.
Examples:
>>> from datetime import datetime, timezone
>>> from mineproductivity.decision.thresholds import Threshold
>>> breach = ThresholdBreach(
... threshold=Threshold(field="value", comparator="<", limit=0.65),
... observed_value=0.58, breached_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
... )
>>> AlertGenerator().from_breach(breach).severity
'high'
>>> rec = Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="Investigate fleet availability", severity="critical", evidence=("UTIL.OEE",),
... )
>>> AlertGenerator().from_recommendation(rec).message
'Investigate fleet availability'
>>> low_severity = Recommendation(
... policy_code="X", triggered_rules=(), summary="x", severity="low", evidence=(),
... )
>>> AlertGenerator().from_recommendation(low_severity) is None
True
DecisionAuditEntry
dataclass
¶
Bases: BaseValueObject
One append-only record: what was decided, for what scope, when, and (where traceable) from which raw events.
source_event_ids is populated on a best-effort basis (§27's own
"What 'traceable' means in practice" note) -- an empty tuple is not
an error; result, context_scope, and recorded_at alone
already satisfy this package's accountability objective.
Examples:
>>> from datetime import timezone
>>> from mineproductivity.decision.result import DecisionResult
>>> entry = DecisionAuditEntry(
... recorded_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
... result=DecisionResult(model_code="STRATEGY.Threshold"),
... context_scope={"pit": "north"}, source_event_ids=(),
... )
>>> dict(entry.context_scope)
{'pit': 'north'}
DecisionAuditTrail ¶
Append-only record of every DecisionResult ever produced by
any DecisionPipeline run in this process -- the accountability
mechanism a business-decision system requires that a purely
statistical one (analytics) does not as urgently. Serializes via
core.serialization exactly as every other value object in this
platform does.
Examples:
>>> from datetime import timezone
>>> from mineproductivity.decision.result import DecisionResult
>>> trail = DecisionAuditTrail()
>>> trail.record(DecisionAuditEntry(
... recorded_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
... result=DecisionResult(model_code="STRATEGY.Threshold"),
... context_scope={"pit": "north"}, source_event_ids=(),
... ))
>>> len(trail.query())
1
>>> len(trail.query(scope={"pit": "south"}))
0
BatchDecisionRunner ¶
Runs one DecisionPipeline once over a bounded,
already-assembled DecisionContext -- the 'normal,' scheduled-
report mode, in contrast to RealTimeDecisionSession's live,
event-driven mode (§25). A thin, named wrapper over
DecisionPipeline.run (§9), exactly mirroring
analytics.BatchAnalyticsRunner's relationship to
AnalyticsPipeline (spec 06 §28).
Examples:
>>> from typing import ClassVar
>>> from mineproductivity.decision.abstractions import DecisionModel
>>> from mineproductivity.decision.metadata import DecisionCategory, DecisionMetadata
>>> from mineproductivity.decision.pipeline import ModelStage
>>> from mineproductivity.kpis import KPIResult
>>> class _Model(DecisionModel):
... meta: ClassVar[DecisionMetadata] = DecisionMetadata(
... code="STRATEGY.Doctest", category=DecisionCategory.STRATEGY, description="x",
... )
... def _decide(self, context):
... return DecisionResult(model_code="TEST")
>>> pipeline = DecisionPipeline(stages=(ModelStage(_Model()),))
>>> runner = BatchDecisionRunner(pipeline=pipeline)
>>> ctx = DecisionContext(kpi_results=(KPIResult(code="PROD.TPH", value=1.0, unit="t/h"),),
... analytics_results=(), scope={"pit": "north"})
>>> runner.run(ctx).unwrap().model_code
'TEST'
DecisionModelNotFoundError ¶
Bases: NotFoundError
REGISTRY.get(code) found no registered DecisionModel for
that code (design spec §32).
DecisionValidationError ¶
Bases: ValidationError
A DecisionMetadata, Policy, or Threshold failed
validation (design spec §30, §12, §13) -- e.g. an empty code, a
Policy with zero rules, or an empty Threshold.field -- or a
caller passed malformed arguments to a Decision method.
DecisionVersionConflictError ¶
Bases: RegistrationError
A plugin attempted to re-register an existing DecisionModel
code with materially different metadata without a version bump
(design spec §32), mirroring kpis.KPIVersionConflictError and
analytics.AnalyticsVersionConflictError.
NoApplicablePolicyError ¶
Bases: NotFoundError
Raised only where a caller explicitly requests raising behavior
instead of :meth:~mineproductivity.decision.abstractions.DecisionModel.decide's
default empty-recommendation result for a context no policy applies
to (design spec §8).
PolicyConflictError ¶
Bases: RegistrationError
A plugin or governance action attempted to re-register an
existing, Active Policy code with different rules/thresholds
without a version bump and a Superseded transition for the prior
version (design spec §12, §29) -- the Policy-layer analogue of
:class:DecisionVersionConflictError, since a Policy is a
separate governed artifact from a DecisionModel implementation
(design spec §3.6).
ExplanationBuilder ¶
Walks a Recommendation's triggered_rules/evidence and
produces its Explanation. A concrete, non-pluggable utility --
not a DecisionModel subclass, since explanation construction is
not a decision in itself.
Examples:
>>> from mineproductivity.kpis import KPIResult
>>> rec = Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="x", severity="high", evidence=("UTIL.OEE",),
... )
>>> context = DecisionContext(
... kpi_results=(KPIResult(code="UTIL.OEE", value=0.58, unit=""),),
... analytics_results=(), scope={},
... )
>>> explanation = ExplanationBuilder().build(rec, context=context)
>>> explanation.evidence_refs
('UTIL.OEE',)
>>> len(explanation.premises)
2
ExplanationStage ¶
Bases: PipelineStage
The PipelineStage wrapper composing ExplanationBuilder
into a DecisionPipeline -- attaches an Explanation to every
Recommendation in context.recommendations, in place (see
this module's own docstring for why this is non-terminal, matching
§17's own typed signature).
Examples:
>>> from mineproductivity.kpis import KPIResult
>>> rec = Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="x", severity="high", evidence=("UTIL.OEE",),
... )
>>> context = DecisionContext(
... kpi_results=(KPIResult(code="UTIL.OEE", value=0.58, unit=""),),
... analytics_results=(), scope={}, recommendations=(rec,),
... )
>>> result = ExplanationStage().process(context)
>>> result.recommendations[0].explanation is not None
True
DecisionCategory ¶
Bases: Enum
Closed enum -- adding a member is a governance-reviewed change,
mirroring analytics.AnalyticsCategory's closed-enum rule
(design spec §31).
DecisionMetadata
dataclass
¶
DecisionMetadata(code, *, name='', description, tags=frozenset(), attributes=dict(), category, version='1.0.0')
Bases: BaseMetadata
The minimal registration schema for a discoverable
DecisionModel -- as light as analytics.AnalyticsMetadata, not
as heavy as kpis.KPIMetadata, because a DecisionModel
implementation is a computational strategy, not itself the governed
business artifact (that weight belongs to Policy, design spec
§12, §3.6).
BaseMetadata.name has no default upstream and a DecisionModel's
code already serves as its identifier, so name defaults to
code (via :meth:_normalize) whenever a caller does not supply
one explicitly -- the same convention AnalyticsMetadata already
established.
Examples:
>>> meta = DecisionMetadata(
... code="STRATEGY.Threshold", category=DecisionCategory.STRATEGY,
... description="Evaluate a Policy's rules against a DecisionContext.",
... )
>>> meta.code
'STRATEGY.Threshold'
>>> meta.name
'STRATEGY.Threshold'
>>> DecisionMetadata(code="", category=DecisionCategory.STRATEGY, description="x")
Traceback (most recent call last):
...
mineproductivity.decision.exceptions.DecisionValidationError: DecisionMetadata.code must not be empty
DecisionPipeline ¶
An ordered Sequence[PipelineStage], run in order over one input
DecisionContext -- a typical pipeline gathers evidence, evaluates
rules against an active Policy, generates recommendations, ranks
them, and explains them.
A pipeline never contains strategy-specific branching (mirroring
AnalyticsPipeline's "holds no model logic" invariant one layer
down): :meth:run calls each stage's :meth:~PipelineStage.process
uniformly and has no knowledge of which concrete DecisionModel a
ModelStage wraps.
Examples:
>>> from typing import ClassVar
>>> from mineproductivity.decision.metadata import DecisionCategory, DecisionMetadata
>>> class _Model(DecisionModel):
... meta: ClassVar[DecisionMetadata] = DecisionMetadata(
... code="STRATEGY.Doctest", category=DecisionCategory.STRATEGY, description="x",
... )
... def _decide(self, context):
... return DecisionResult(model_code="TEST")
>>> from mineproductivity.kpis import KPIResult
>>> ctx = DecisionContext(kpi_results=(KPIResult(code="PROD.TPH", value=1.0, unit="t/h"),),
... analytics_results=(), scope={})
>>> pipeline = DecisionPipeline(stages=(ModelStage(_Model()),))
>>> pipeline.run(ctx).unwrap().model_code
'TEST'
>>> DecisionPipeline(stages=()).run(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 a DecisionResult) or this returns
Result.err(DecisionValidationError(...)).
ModelStage ¶
Bases: PipelineStage
A terminal stage that hands the (by now rule-evaluated) context to
one DecisionModel and yields its DecisionResult.
Examples:
>>> from typing import ClassVar
>>> from mineproductivity.decision.metadata import DecisionCategory, DecisionMetadata
>>> class _Model(DecisionModel):
... meta: ClassVar[DecisionMetadata] = DecisionMetadata(
... code="STRATEGY.Doctest", category=DecisionCategory.STRATEGY, description="x",
... )
... def _decide(self, context):
... return DecisionResult(model_code="TEST")
>>> stage = ModelStage(_Model())
>>> from mineproductivity.kpis import KPIResult
>>> ctx = DecisionContext(kpi_results=(KPIResult(code="PROD.TPH", value=1.0, unit="t/h"),),
... analytics_results=(), scope={})
>>> stage.process(ctx).model_code
'TEST'
PipelineStage ¶
Bases: ABC
One step in a DecisionPipeline. Stateless and composable -- a
new stage never requires changing DecisionPipeline itself.
Return type is deliberately DecisionContext | DecisionResult, not
just DecisionContext: :class:ModelStage -- itself a
PipelineStage -- always yields a DecisionResult, so the
honest contract every implementation (terminal or not) must satisfy
is the union of both. This mirrors
analytics.pipeline.PipelineStage's identical, already-disclosed
widening of its own design spec's abstract-method signature.
process
abstractmethod
¶
Transform one DecisionContext into another (e.g.
attaching rule-evaluation results) -- OR, for a terminal stage,
wrap the context into a DecisionResult (see
:class:ModelStage).
ActionPlanner ¶
Sequences a set of prioritized actions, respecting declared
dependencies, into an ActionPlan. Implements its own narrow
topological ordering rather than reusing kpis.DependencyGraph,
which is tightly coupled to Registry[str, type[BaseKPI]] and
KPIMetadata.dependencies -- a shape that does not fit arbitrary
action-to-action dependencies (§3.3, §21).
dependencies is an optional, caller-supplied mapping from an
action's identifying key (Recommendation.policy_code) to the
keys it depends on -- when empty (the default), :meth:plan
degrades to simply ordering actions by ActionPriority.priority_score,
so a caller with no dependency information to supply pays no
complexity cost. A dependency entry naming a key absent from
prioritized is silently ignored (the referenced action is not a
candidate this call is planning over). :meth:plan returns
Result.err on a detected cycle rather than silently dropping or
arbitrarily breaking it -- the same "fail loudly on a cycle" posture
kpis.DependencyGraph.detect_cycle already established for KPI
dependencies, applied here to this package's own, structurally
different dependency shape. For the identical reason, plan also
returns Result.err when two entries in prioritized share a
policy_code -- since the identifying key is the only handle
dependencies (and this method's own de-duplication) has for an
action, silently keeping only one would silently drop the other from
the plan (a defect caught during this phase's own QA review, before
it could ship).
Ordering uses a priority queue (heapq) over the ready frontier
rather than a plain FIFO/DFS traversal, so the emitted order always
respects priority_score (descending) among actions that are
simultaneously eligible -- O((V+E) log V), a small, disclosed
deviation from the plain O(V+E) bound a tie-break-free
topological sort achieves, required because respecting priority
order among ready actions is this method's own stated purpose, not
an afterthought. Ties break by input order (a stable-sort-shaped
guarantee, mirroring ranking.WeightedScoreRanking's own stable
tie-break discipline).
Examples:
>>> from mineproductivity.decision.result import DecisionScore, Recommendation
>>> rec_a = Recommendation(
... policy_code="A", triggered_rules=(), summary="Do A", severity="high", evidence=(),
... )
>>> rec_b = Recommendation(
... policy_code="B", triggered_rules=(), summary="Do B", severity="medium", evidence=(),
... )
>>> ranked_a = RankedRecommendation(
... recommendation=rec_a, score=DecisionScore(value=0.9, components={}), rank=1,
... )
>>> ranked_b = RankedRecommendation(
... recommendation=rec_b, score=DecisionScore(value=0.5, components={}), rank=2,
... )
>>> priority_a = ActionPriority(urgency=0.75, impact=0.9, effort=1.0)
>>> priority_b = ActionPriority(urgency=0.5, impact=0.5, effort=1.0)
>>> plan = ActionPlanner().plan(
... [(ranked_a, priority_a), (ranked_b, priority_b)], dependencies={"A": ("B",)},
... )
>>> [action.policy_code for action in plan.unwrap().ordered_actions]
['B', 'A']
DecisionStatus ¶
Bases: Enum
The Policy lifecycle -- mirrors kpis.KPIStatus exactly,
applied here to governed business artifacts rather than to
individual computed results.
Policy
dataclass
¶
Policy(code, *, version='1.0.0', status=(lambda: DecisionStatus.PROPOSED)(), rules, thresholds=dict(), strategy_code)
Bases: BaseValueObject
A named, versioned bundle of :data:~mineproductivity.decision.rules.Rule\ s,
the :class:~mineproductivity.decision.thresholds.Threshold\ s they
reference, and which DecisionStrategy/category they activate --
the business-policy equivalent of kpis.KPIMetadata's governance
weight.
Examples:
>>> from mineproductivity.core import PredicateSpecification
>>> policy = Policy(
... code="AVAIL.LowFleetAvailability",
... rules={"low_oee": PredicateSpecification(lambda ctx: True)},
... strategy_code="STRATEGY.Threshold",
... )
>>> policy.status
<DecisionStatus.PROPOSED: 'proposed'>
>>> policy.version
'1.0.0'
ActionPrioritizer ¶
Given a Sequence[RankedRecommendation], assigns each an
ActionPriority reflecting urgency/impact/effort under limited
execution capacity.
Examples:
>>> from mineproductivity.decision.result import DecisionScore, Recommendation
>>> rec = Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="x", severity="high", evidence=(),
... )
>>> ranked = RankedRecommendation(
... recommendation=rec, score=DecisionScore(value=0.7, components={}), rank=1,
... )
>>> prioritized = ActionPrioritizer().prioritize([ranked])
>>> item, priority = prioritized[0]
>>> priority.urgency, priority.impact
(0.75, 0.7)
RankingStrategy ¶
Bases: DecisionModel, ABC
Category base for ranking strategies -- orders candidate
Recommendation\ s by :class:~mineproductivity.decision.result.DecisionScore.
Distinct from :class:~mineproductivity.decision.prioritization.ActionPrioritizer
(§20): ranking answers "which recommendation is best," prioritization
answers "given limited capacity, which action happens first" --
conflating the two is a recorded anti-pattern (§33).
WeightedScoreRanking ¶
Bases: RankingStrategy
The default, concrete ranking strategy: orders Recommendation\ s
by DecisionScore.value, descending. Ties preserve input relative
order (Python's sorted() is stable).
Registered into decision.REGISTRY at import time, mirroring how
strategy.ThresholdDecisionStrategy self-registers.
Examples:
>>> low = Recommendation(
... policy_code="A", triggered_rules=("r",), summary="x", severity="low", evidence=(),
... )
>>> critical = Recommendation(
... policy_code="B", triggered_rules=("r",), summary="x", severity="critical", evidence=(),
... )
>>> ranked = WeightedScoreRanking().rank([low, critical])
>>> [item.recommendation.policy_code for item in ranked]
['B', 'A']
>>> [item.rank for item in ranked]
[1, 2]
rank ¶
Scores every Recommendation via DecisionScorer, then
orders them by DecisionScore.value, descending, assigning
each its 1-based rank position.
RealTimeDecisionSession ¶
RealTimeDecisionSession(*, bus, pipeline, kpi_engine, analytics_runner, kpi_codes, window, audit_trail=None)
A long-lived session that subscribes to an events.EventBus
and, on each new relevant envelope, refreshes the necessary
kpis.KPIEngine/analytics.BatchAnalyticsRunner outputs and
re-runs a DecisionPipeline over the refreshed
DecisionContext -- the live-operations-center counterpart of
BatchDecisionRunner (§26), mirroring
analytics.StreamingAnalyticsSession's role one layer down (spec
06 §27). Composes KPIEngine/BatchAnalyticsRunner rather than
recomputing anything itself (§3.2).
Thread safety. Mirrors analytics.StreamingAnalyticsSession:
the Event Framework's own concurrency contract makes no guarantee
about which thread a subscriber's handler runs on, so _latest is
guarded by one threading.Lock shared across every scope --
independent scopes serialize on this single lock, an intentionally
simple posture for a per-process session object, not a
high-throughput shared cache. Because a handler's own refresh work
(KPIEngine.execute, BatchAnalyticsRunner.run,
DecisionPipeline.run) happens outside the lock, two concurrent
deliveries for the same scope could otherwise finish out of order and
let an older envelope's slower handler overwrite a newer one's
already-stored result -- caught during this phase's own QA review.
_latest therefore stores each scope's envelope
event_time_utc alongside its DecisionResult and only
overwrites when the new envelope's event_time_utc is not older
than what is already stored, reusing event_time_utc as the
already-canonical "calculation basis" ordering signal
(EventEnvelope's own docstring, events spec 01) rather than
inventing a second sequencing concept.
Examples:
kpi_engine/analytics_runner are duck-typed fakes below,
mirroring analytics.AnalyticsContext's own doctest precedent
(its _FakeStore stand-in for event_store: EventStore[Any])
-- this keeps the example focused on RealTimeDecisionSession's
own refresh-and-decide wiring rather than a full KPIEngine
dependency graph.
>>> from typing import ClassVar
>>> from datetime import datetime, timezone
>>> from mineproductivity.core import Result
>>> from mineproductivity.decision.abstractions import DecisionModel
>>> from mineproductivity.decision.metadata import DecisionCategory, DecisionMetadata
>>> from mineproductivity.decision.pipeline import ModelStage
>>> 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 mineproductivity.kpis import KPIResult
>>> class _Model(DecisionModel):
... meta: ClassVar[DecisionMetadata] = DecisionMetadata(
... code="STRATEGY.Doctest", category=DecisionCategory.STRATEGY, description="x",
... )
... def _decide(self, context):
... return DecisionResult(model_code="TEST")
>>> class _FakeKPIEngine:
... def execute(self, code, *, window, scope):
... return Result.ok(KPIResult(code=code, value=1.0, unit="t/h", scope=scope))
>>> class _FakeAnalyticsRunner:
... def run(self, series):
... return Result.err("no analytics configured for this doctest")
>>> pipeline = DecisionPipeline(stages=(ModelStage(_Model()),))
>>> bus = _InMemoryEventBus()
>>> session = RealTimeDecisionSession(
... bus=bus, pipeline=pipeline, kpi_engine=_FakeKPIEngine(),
... analytics_runner=_FakeAnalyticsRunner(), kpi_codes=("PROD.TPH",), window="shift",
... )
>>> 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.latest({"equipment_id": "HT-1", "shift": "A"}).model_code
'TEST'
>>> subscription.cancel()
ActionPlan
dataclass
¶
ActionPlan(ordered_actions, priorities, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: DecisionResult
A topologically-ordered sequence of prioritized actions, respecting declared dependencies (design spec §21).
Examples:
>>> rec = Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="Investigate fleet availability", severity="high", evidence=("UTIL.OEE",),
... )
>>> plan = ActionPlan(
... ordered_actions=(rec,),
... priorities={"AVAIL.LowFleetAvailability": ActionPriority(urgency=0.8, impact=0.6, effort=0.2)},
... )
>>> len(plan.ordered_actions)
1
ActionPriority
dataclass
¶
Bases: BaseValueObject
Urgency/impact/effort under limited execution capacity (design
spec §20) -- a distinct question from ranking (§16). All three
components are retained alongside the aggregate priority_score
so an operator can audit the arithmetic, not merely trust it.
Examples:
Alert
dataclass
¶
Alert(message, severity, scope, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=(), triggered_by=None)
Bases: DecisionResult
Produced from a ThresholdBreach or a high-severity
Recommendation (design spec §22). AlertGenerator (a later
phase, out of scope here) produces this value object only -- channel
delivery is explicitly out of this package's scope.
Examples:
>>> Alert(message="Fleet availability critical", severity="critical",
... scope={"pit": "north"}).severity
'critical'
ConfidenceScore
dataclass
¶
Bases: BaseValueObject
The trust weight backing how strongly a recommendation should be
acted on (design spec §24). basis records which evidence path
produced value -- "data_quality"/"rule_strength"/
"combined" -- so a consumer can distinguish "we checked data
quality and it was good" from "we had no data-quality signal to
check."
Examples:
DecisionResult
dataclass
¶
Bases: BaseValueObject
The shared envelope every concrete Decision output composes --
mirrors analytics.AnalyticsResult's role, one layer up.
Examples:
>>> DecisionResult(model_code="STRATEGY.Threshold").model_code
'STRATEGY.Threshold'
>>> DecisionResult(warnings=("no evidence in context",)).warnings
('no evidence in context',)
DecisionScore
dataclass
¶
Bases: BaseValueObject
The numeric weight backing ranking (design spec §16, §23).
components names the contributing factors, for auditability --
an operator reviewing a ranked list can see why one recommendation
outranked another, not merely trust the aggregate.
Examples:
Explanation
dataclass
¶
Bases: BaseValueObject
A structured, evidence-linked justification -- never opaque prose
only (design spec §17). premises are human-readable statements;
evidence_refs are the exact KPIResult.code/
AnalyticsResult.model_code values a consuming system can re-fetch
to verify the claim independently.
Deliberately a plain BaseValueObject, not a DecisionResult
subclass -- it is a supporting value attached to a decision output,
not a decision output itself (§28).
Examples:
>>> Explanation(premises=("OEE below 0.65",), evidence_refs=("UTIL.OEE",)).premises
('OEE below 0.65',)
RankedRecommendation
dataclass
¶
RankedRecommendation(recommendation, score, rank, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: DecisionResult
One Recommendation ordered into a ranked sequence by
DecisionScore (design spec §16).
Examples:
>>> rec = Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="Investigate fleet availability", severity="high", evidence=("UTIL.OEE",),
... )
>>> RankedRecommendation(
... recommendation=rec, score=DecisionScore(value=0.8, components={}), rank=1,
... ).rank
1
Recommendation
dataclass
¶
Recommendation(policy_code, triggered_rules, summary, severity, evidence, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=(), explanation=None)
Bases: DecisionResult
One recommended action, produced by a DecisionStrategy (design
spec §15) -- always traceable to the specific Policy/Rule
names that produced it (policy_code, triggered_rules) and to
the specific KPIResult/AnalyticsResult evidence it was
computed from (evidence).
Examples:
>>> Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="Investigate fleet availability", severity="high",
... evidence=("UTIL.OEE",),
... ).severity
'high'
RootCauseResult
dataclass
¶
RootCauseResult(candidate_causes, confidence, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: DecisionResult
The outcome of a :class:~mineproductivity.decision.root_cause.RootCauseAnalyzer
(design spec §18) -- a future phase's interface-only extension point;
this result type is implemented now even though no producer of it
ships in this phase.
Examples:
>>> RootCauseResult(
... candidate_causes=("conveyor belt wear",),
... confidence=ConfidenceScore(value=0.6, basis="rule_strength"),
... ).candidate_causes
('conveyor belt wear',)
ThresholdBreach
dataclass
¶
Bases: BaseValueObject
Records exactly which Threshold, at what observed value, was
crossed (design spec §13, §28) -- the audit-trail-facing companion
to Threshold itself.
Examples:
>>> from datetime import datetime, timezone
>>> breach = ThresholdBreach(
... threshold=Threshold(field="value", comparator="<", limit=0.65),
... observed_value=0.58,
... breached_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
... )
>>> breach.observed_value
0.58
WhatIfResult
dataclass
¶
WhatIfResult(hypothesis, predicted_outcome, confidence, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: DecisionResult
The outcome of a :class:~mineproductivity.decision.what_if.WhatIfEngine
(design spec §19) -- a future phase's interface-only extension point;
this result type is implemented now even though no producer of it
ships in this phase.
Examples:
>>> WhatIfResult(
... hypothesis={"shift_length_hours": 10}, predicted_outcome="OEE improves by 2%",
... confidence=ConfidenceScore(value=0.4, basis="rule_strength"),
... ).predicted_outcome
'OEE improves by 2%'
RootCauseAnalyzer ¶
Bases: DecisionModel, ABC
The contract a future root-cause-analysis plugin implements. THIS MODULE SHIPS NO CONCRETE SUBCLASS -- identifying why a threshold was breached is a causal-inference problem this package's charter (§3.4) excludes.
Leaves :meth:~mineproductivity.decision.abstractions.DecisionModel._decide
abstract (inherited, unoverridden) exactly as every other category
base in this package does (DecisionStrategy, RankingStrategy)
-- only a concrete subclass decides how its own _decide relates
to _analyze, since no concrete subclass ships here to make that
decision on a future plugin's behalf.
RuleEngine ¶
Evaluates a named mapping of :data:Rule\ s against one
DecisionContext, isolating one rule's evaluation failure from
the rest.
Examples:
>>> from mineproductivity.core import PredicateSpecification
>>> from mineproductivity.kpis import KPIResult
>>> low_oee: Rule = PredicateSpecification(
... lambda ctx: any(r.value is not None and r.value < 0.65 for r in ctx.kpi_results)
... )
>>> context = DecisionContext(
... kpi_results=(KPIResult(code="UTIL.OEE", value=0.58, unit=""),),
... analytics_results=(), scope={},
... )
>>> RuleEngine().evaluate({"low_oee": low_oee}, context)
('low_oee',)
evaluate ¶
Returns the names of every rule satisfied by context, in a
stable (sorted) order. A single rule raising during
is_satisfied_by is caught, logged, and excluded from the
result -- never allowed to prevent evaluation of the remaining
rules.
RuleEngineStage ¶
Bases: PipelineStage
The PipelineStage wrapper composing RuleEngine into a
DecisionPipeline -- attaches the triggered rule names to the
DecisionContext passed downstream, so
ModelStage(ThresholdDecisionStrategy(policy=policy)) never has
to re-run rule evaluation itself.
Examples:
>>> from mineproductivity.core import PredicateSpecification
>>> from mineproductivity.decision.policy import Policy
>>> policy = Policy(
... code="TEST.Policy", rules={"always": PredicateSpecification(lambda ctx: True)},
... strategy_code="STRATEGY.Threshold",
... )
>>> stage = RuleEngineStage(policy=policy)
>>> context = DecisionContext(kpi_results=(), analytics_results=(), scope={})
>>> stage.process(context).triggered_rules
('always',)
ConfidenceScorer ¶
Computes :class:ConfidenceScore, deriving from
analytics.DataQualityScore (spec 06 §25) plus how many/which
rules corroborated a Recommendation -- never recomputing data
quality itself. Falls back to "rule_strength" basis when no
DataQualityScore is present in context.analytics_results.
This scorer's own derivation never produces a bare "data_quality"
basis: a Recommendation always carries at least one triggered
rule by construction (Policy.validate() requires at least one
rule), so rule-strength evidence is always present alongside any
DataQualityScore that may also be found -- the result is always
"rule_strength" alone or "combined". The third basis literal
("data_quality") remains a legitimate value other producers may
use; this class simply never has a reason to.
Examples:
>>> rec = Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="x", severity="high", evidence=("UTIL.OEE",),
... )
>>> context = DecisionContext(kpi_results=(), analytics_results=(), scope={})
>>> ConfidenceScorer().score(rec, context=context).basis
'rule_strength'
DecisionScorer ¶
Computes the numeric weight (:class:DecisionScore) backing
ranking (§16) -- a function of severity, policy weight (approximated
here as rule-corroboration strength, see :func:_rule_strength),
and, optionally, :class:ConfidenceScore (§24) -- never of raw
KPI/Analytics values directly re-interpreted.
Examples:
>>> from mineproductivity.decision.result import ConfidenceScore
>>> rec = Recommendation(
... policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
... summary="x", severity="critical", evidence=("UTIL.OEE",),
... )
>>> score = DecisionScorer().score(rec, confidence=ConfidenceScore(value=1.0, basis="rule_strength"))
>>> round(score.value, 4)
0.84
DecisionStrategy ¶
Bases: DecisionModel, ABC
Category base for decision-making strategies -- the pluggable
'how do we decide' abstraction, analogous to
analytics.TrendModel/BenchmarkModel one layer down.
ThresholdDecisionStrategy ¶
Bases: DecisionStrategy
The default, concrete strategy: evaluate the active Policy's
Rule\ s (via :class:~mineproductivity.decision.rules.RuleEngine)
against the DecisionContext, and produce one Recommendation
naming every triggered rule.
Registered into decision.REGISTRY at import time, mirroring how
analytics.trend/analytics.baseline/analytics.benchmarking
self-register.
severity is a constructor parameter, not a Policy/Threshold
field: design spec §12's own Policy pseudocode (code,
version, status, rules, thresholds, strategy_code)
and §13's Threshold pseudocode (field, comparator,
limit) carry no severity-bearing field anywhere, so there is
nothing to read one from. A constructor parameter models "how bad
is it when this policy's rules fire" as a property of the
policy/strategy pairing (a reviewed, deliberate choice at
configuration time), without adding an undocumented field to either
already-specified, locked value object -- the smallest compatible
resolution, disclosed here per this phase's own reconciliation rule.
Examples:
>>> from mineproductivity.core import PredicateSpecification
>>> from mineproductivity.decision.result import Recommendation
>>> from mineproductivity.decision.thresholds import Threshold
>>> from mineproductivity.kpis import KPIResult
>>> policy = Policy(
... code="AVAIL.LowFleetAvailability",
... rules={"low_oee": PredicateSpecification(
... lambda ctx: any(r.value is not None and r.value < 0.65 for r in ctx.kpi_results)
... )},
... thresholds={"low_oee": Threshold(field="value", comparator="<", limit=0.65)},
... strategy_code="STRATEGY.Threshold",
... )
>>> strategy = ThresholdDecisionStrategy(policy=policy, severity="high")
>>> context = DecisionContext(
... kpi_results=(KPIResult(code="UTIL.OEE", value=0.58, unit=""),),
... analytics_results=(), scope={"pit": "north"},
... )
>>> result = strategy.decide(context)
>>> isinstance(result, Recommendation)
True
>>> result.policy_code, result.triggered_rules, result.severity
('AVAIL.LowFleetAvailability', ('low_oee',), 'high')
>>> strategy.check_thresholds(context)[0].observed_value
0.58
check_thresholds ¶
Independently checks every Threshold in self._policy.thresholds
against context evidence, returning a ThresholdBreach for
each one whose comparator/limit the resolved observed value
satisfies. A Threshold whose field cannot be resolved
against context (see :func:_resolve_field) is silently
skipped, never raised -- "qualify, don't coerce".
Threshold
dataclass
¶
Bases: BaseValueObject
One declarative limit a Rule references, e.g.
Threshold(field="value", comparator="<", limit=0.65) checked
against the relevant KPIResult/AnalyticsResult field named by
field.
Deliberately data, not code -- a Policy can be updated to
reference a new limit value through ordinary version governance
(design spec §12) without touching any Rule's predicate logic.
Examples:
>>> Threshold(field="value", comparator="<", limit=0.65).limit
0.65
>>> Threshold(field="", comparator="<", limit=0.65)
Traceback (most recent call last):
...
mineproductivity.decision.exceptions.DecisionValidationError: Threshold.field must not be empty
WhatIfEngine ¶
Bases: DecisionModel, ABC
The contract a future what-if/scenario-simulation plugin implements. THIS MODULE SHIPS NO CONCRETE SUBCLASS -- predicting the outcome of a hypothetical change is a simulation/forecasting problem this package's charter (§3.4) excludes.
Leaves :meth:~mineproductivity.decision.abstractions.DecisionModel._decide
abstract (inherited, unoverridden) exactly as RootCauseAnalyzer
does -- only a concrete subclass decides how its own _decide
relates to _simulate, since no concrete subclass ships here to
make that decision on a future plugin's behalf.
register ¶
Register cls into :data:REGISTRY, keyed by cls.meta.code.
Raises:
| Type | Description |
|---|---|
DecisionValidationError
|
If |
DecisionVersionConflictError
|
If |