Skip to content

Cost Primitives as Athena-Compatible Ops

Status & scope

  • Module: lens/athena/ops/lens/
  • Milestone: Phase 2

Purpose

Restructure the 10 scoring primitives and 6 transforms as LensCommand subclasses that mirror athena's Command interface. This enables drop-in integration when migrating to the platform — swap import path, same math.

Key insight: 5 lens specs × 10 primitives × 6 transforms = 100s of use cases from YAML alone. No new Python code needed per scenario.

Alignment With Athena Ops

Athena Op Lens Scoring Op primitive_id Same Math?
Scale(minmax) LinearScale LS-01 Yes: (v - min) / (max - min) * range
Scale(standard) MinMaxNormalize LS-06 Yes: standard scaler
Bin StepFunction LS-09 Yes: piecewise constant from ranges
Encode(categorical) CategoricalMatch LT-05 Yes: category → numeric
TimeSeries.ema Decay(exponential) LS-08 Yes: EMA = exponential decay
— (lens-specific) ThresholdGate LS-02 Binary 0/inf gate
— (lens-specific) ExponentialRamp LS-03 Penalty approaching limit
— (lens-specific) MultiplicativePenalty LS-04 base × factor
— (lens-specific) InverseLinear LS-05 1 - v/ref
— (lens-specific) ClampedRatio LS-06 v / max ∈ [0,1]
— (lens-specific) WeightedComposite LS-07 Multi-score aggregation
— (lens-specific) Passthrough LS-10 Identity

Public API

Command Base Class

class LensCommand:
    """Lightweight Command — plain Python, same interface as athena.ops.command.Command."""

    primitive_id: str      # e.g. "LS-01"
    primitive_name: str    # e.g. "linear_scale"

    def __init__(self, parameters: dict, data: Any = None):
        self.parameters = parameters
        self.data = data
        self.result = None

    def execute(self) -> float:
        """Run the computation. Subclasses override."""
        raise NotImplementedError

    def validate(self) -> list[str]:
        """Optional param validation. Returns error messages."""
        return []

Registry

LENS_SCORING_OPS: dict[str, type[LensCommand]]   # primitive_name → class
LENS_TRANSFORM_OPS: dict[str, type[LensCommand]]  # transform_name → class
LENS_OPS: dict[str, type[LensCommand]]            # union of both

Resolution (updated binding.py)

def resolve_cost_model(cost_model, ...) -> CostFn:
    # 1. Legacy cost_registry (backward compat)
    # 2. LENS_SCORING_OPS (new: Command subclass)
    # 3. COST_PRIMITIVES (backward compat: pure functions)
    # 4. Raise

Scoring Ops (10 LensCommand subclasses)

LS-01: LinearScale

  • Math: (v - min) / (max - min) * max_cost, optionally inverted
  • Params: max_cost (1.0), max_value (1.0), min_value (0.0), invert (False)
  • Athena equivalent: Scale(minmax)

LS-02: ThresholdGate

  • Math: 0 if pass, inf if fail
  • Params: threshold (0.0), direction ("below")

LS-03: ExponentialRamp

  • Math: (base^(v/limit * 10) - 1) * scale, inf at limit
  • Params: limit (1.0), base (2.0), scale (1.0)

LS-04: MultiplicativePenalty

  • Math: base_cost * value * multiplier
  • Params: base_cost (1.0), multiplier (1.0)

LS-05: InverseLinear

  • Math: max(0, 1 - value/reference)
  • Params: reference (1.0)

LS-06: ClampedRatio

  • Math: clamp(value / max_value, 0, 1)
  • Params: max_value (1.0)

LS-07: WeightedComposite

  • Math: weighted average of sub_scores
  • Params: sub_scores (list of {value, weight})

LS-08: Decay

  • Math: linear: max(0, 1 - v/max) or exponential: e^(-0.693 * v/half_life)
  • Params: mode ("linear"), linear_max (1.0), half_life (1.0)
  • Athena equivalent: TimeSeries.exponential_moving_average

LS-09: StepFunction

  • Math: piecewise constant from ranges
  • Params: steps (list of {min, max, cost}), default (0.0)
  • Athena equivalent: Bin

LS-10: Passthrough

  • Math: return value as-is
  • Params: none

Transform Ops (6 LensCommand subclasses)

LT-01: ExtractField

  • Math: data[field] cast to float
  • Params: field, default (0.0)

LT-02: ExtractRatio

  • Math: data[numerator] / data[denominator]
  • Params: numerator, denominator, default (0.0)

LT-03: MinMaxNormalize

  • Math: (value - min) / (max - min) ∈ [0, 1]
  • Params: min (0.0), max (1.0)
  • Athena equivalent: Scale(standard)

LT-04: Invert

  • Math: 1.0 - value
  • Params: field (optional)

LT-05: CategoricalMatch

  • Math: 1.0 if field value ∈ values, else 0.0
  • Params: field, values (list)
  • Athena equivalent: Encode(categorical)

LT-06: Identity

  • Math: pass through as float
  • Params: none

Backward Compatibility

The existing pure functions in lens/primitives/cost_primitives.py and lens/primitives/transforms.py remain as-is. Each LensCommand.execute() delegates to the same math. The COST_PRIMITIVES and TRANSFORMS dicts stay for backward compat — existing code that imports them continues to work.

Resolution order in binding.py: 1. Legacy cost_registry (domain cost_models.py dicts) 2. LENS_SCORING_OPS (new Command subclasses) 3. COST_PRIMITIVES (legacy pure functions — backward compat) 4. Raise

Test Fixtures

Test Input Expected File
Each scoring op produces same output as pure function Same value + params Identical float tests/test_primitives.py
Each transform op produces same output as pure function Same data + params Identical float tests/test_transforms.py
LENS_OPS registry has all 16 entries — 10 scoring + 6 transforms tests/test_primitives.py
Command.validate() catches bad params Missing required param Error list tests/test_primitives.py
Declarative YAML → Command dispatch YAML spec with op names Same scores as hardcoded tests/test_integration.py
Backward compat: old imports still work from lens.primitives import COST_PRIMITIVES No ImportError tests/test_primitives.py

File Layout

lens/athena/
  __init__.py              ← Package marker
  ops/
    __init__.py            ← Package marker
    command.py             ← LensCommand base class (plain Python, no Dask)
    lens/
      __init__.py          ← LENS_OPS registry dict (union of scoring + transforms)
      scoring.py           ← 10 scoring LensCommand subclasses
      transforms.py        ← 6 transform LensCommand subclasses

Integration Points

  • Athena migration (TASK-A01): Move lens/athena/ops/lens/ → atlas-fl-athena/athena/ops/lens/. Change LensCommand parent to real Command. Add Dask support. Register in athena OPS dict.
  • Cortex (TASK-C01): cortex/tools/lens.py replaces its COST_MODELS dict with LENS_OPS registry. Same dispatch: layer.cost_model → LENS_OPS[name].
  • Beacon: No change — Beacon renders results, doesn't compute.

DO NOT

  • Import from athena directly — this is a local shim with the same interface
  • Add Dask, Redis, or any infrastructure deps — plain Python only
  • Duplicate the math — each Command.execute() calls the existing pure function
  • Remove the pure functions — they stay for backward compat and readability
  • Create domain-specific ops here — domain cost_models.py files keep their own registries

Depends on: component.prism.universal-lens-parser

Realizes: product.lens

Required by: component.prism.mobility-surface, component.prism.observation-engine, component.prism.operational-lifecycle, component.prism.scoring-engine, component.prism.temporal-engine, component.prism.threat-engine, component.prism.traversal-engine