From cb390c1b8b812e45feb451ccc200794eff596159 Mon Sep 17 00:00:00 2001 From: Shay Palachy Date: Tue, 28 Apr 2026 07:56:11 +0300 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20Milestone=206=20=E2=80=94=20mechani?= =?UTF-8?q?sm=20layer=20v1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the full mechanism library in leadforge/mechanisms/: base.py — Mechanism ABC, MechanismContext, MechanismSummary, MechanismAssignment static.py — CategoricalDraw, BoundedNumericDraw, MixtureDraw influence.py — AdditiveInfluence, LogisticInfluence, SaturatingInfluence, ThresholdInfluence, InteractionTerm scores.py — LatentScore (logistic from weighted latent combination) hazards.py — ConversionHazard (daily P(convert) = base + scale*score) transitions.py — StageSequence (funnel order + terminal detection), HazardTransition (dwell-aware daily advancement hazard) counts.py — PoissonIntensity (log-linear), RecencyDecayIntensity (decay+floor) categorical.py — CategoricalInfluence, CHANNEL_QUALITY_SCORES measurement.py — NoisyProxy, NoisyCategorization, ProxyCompression policies.py — assign_mechanisms(): motif-family-aware MechanismAssignment factory with tuned score weights for all 5 v1 motif families 74 tests covering all mechanism types, serialisation roundtrips, monotonicity properties, edge-case validation, and cross-motif hazard ordering. Co-Authored-By: Claude Sonnet 4.6 --- .agent-plan.md | 41 ++- leadforge/mechanisms/base.py | 155 +++++++++ leadforge/mechanisms/categorical.py | 68 ++++ leadforge/mechanisms/counts.py | 130 +++++++ leadforge/mechanisms/hazards.py | 72 ++++ leadforge/mechanisms/influence.py | 199 +++++++++++ leadforge/mechanisms/measurement.py | 174 ++++++++++ leadforge/mechanisms/policies.py | 211 ++++++++++++ leadforge/mechanisms/scores.py | 55 +++ leadforge/mechanisms/static.py | 138 ++++++++ leadforge/mechanisms/transitions.py | 158 +++++++++ tests/mechanisms/__init__.py | 0 tests/mechanisms/test_mechanisms.py | 504 ++++++++++++++++++++++++++++ 13 files changed, 1891 insertions(+), 14 deletions(-) create mode 100644 leadforge/mechanisms/base.py create mode 100644 leadforge/mechanisms/categorical.py create mode 100644 leadforge/mechanisms/counts.py create mode 100644 leadforge/mechanisms/hazards.py create mode 100644 leadforge/mechanisms/influence.py create mode 100644 leadforge/mechanisms/measurement.py create mode 100644 leadforge/mechanisms/policies.py create mode 100644 leadforge/mechanisms/scores.py create mode 100644 leadforge/mechanisms/static.py create mode 100644 leadforge/mechanisms/transitions.py create mode 100644 tests/mechanisms/__init__.py create mode 100644 tests/mechanisms/test_mechanisms.py diff --git a/.agent-plan.md b/.agent-plan.md index 317a687..dc9e7e9 100644 --- a/.agent-plan.md +++ b/.agent-plan.md @@ -6,35 +6,48 @@ ## Current System State -**v0.3.0 in progress — Milestone 5 complete (PR open).** Population generation fully -implemented: accounts, contacts, leads with all observable fields, full latent state -(8 traits across 3 entity types), motif-family-aware bias, and FK integrity guaranteed. -358 tests passing. +**v0.3.0 in progress — Milestone 6 complete (PR open).** Full mechanism library implemented: +base ABC + context, static draws, influence transforms, latent scoring, conversion hazard, +stage transitions, count intensities, categorical influence, measurement proxies, and +motif-family-aware mechanism assignment. 437 tests passing. --- -## Active Task Breakdown — Milestone 6: Mechanism Layer v1 (v0.3.0) +## Active Task Breakdown — Milestone 7: Simulation Engine (v0.3.0) -Goal: Implement the static and dynamic mechanisms that drive simulation behavior. +Goal: Run the first real evolving world and derive conversion outcomes from events. -- [ ] **1. Base mechanism interface** (`mechanisms/base.py`) -- [ ] **2. Static mechanisms** — categorical, ordinal, bounded-numeric draws (`mechanisms/static.py`) -- [ ] **3. Transition mechanisms** — lead-stage advancement logic (`mechanisms/transitions.py`) -- [ ] **4. Score/hazard mechanisms** — latent-to-observable scoring, conversion hazard (`mechanisms/scores.py`, `mechanisms/hazards.py`) -- [ ] **5. Measurement mechanisms** — noisy proxy observation of latent traits (`mechanisms/measurement.py`) +- [ ] **1. World state** (`simulation/state.py`) — per-lead mutable state during simulation +- [ ] **2. Simulation engine** (`simulation/engine.py`) — daily step loop, 90-day horizon +- [ ] **3. Event generation** — touches/sessions derived from count mechanisms +- [ ] **4. Stage advancement** — HazardTransition drives funnel progression +- [ ] **5. Conversion derivation** — ConversionHazard fires event; sets `converted_within_90_days` --- ## Context Pointers -- Milestone 6 scope: `docs/leadforge_implementation_plan.md` §9 "Milestone 6" -- Mechanism types: `docs/leadforge_architecture_spec.md` §10 "Mechanism layer" -- Latent variables: `docs/leadforge_architecture_spec.md` §9 +- Milestone 7 scope: `docs/leadforge_implementation_plan.md` §10 "Milestone 7" +- Simulation spec: `docs/leadforge_architecture_spec.md` §11 "Simulation engine" +- Mechanism layer: `leadforge/mechanisms/` (all M6 files) --- ## Completed Phases +### Milestone 6 — Mechanism Layer v1 ✓ (v0.3.0 in PR) +- `mechanisms/base.py`: `Mechanism` ABC, `MechanismContext`, `MechanismSummary`, `MechanismAssignment` +- `mechanisms/static.py`: `CategoricalDraw`, `BoundedNumericDraw`, `MixtureDraw` +- `mechanisms/influence.py`: `AdditiveInfluence`, `LogisticInfluence`, `SaturatingInfluence`, `ThresholdInfluence`, `InteractionTerm` +- `mechanisms/scores.py`: `LatentScore` — logistic score from weighted latent combination +- `mechanisms/hazards.py`: `ConversionHazard` — daily conversion probability from latent score +- `mechanisms/transitions.py`: `StageSequence`, `HazardTransition` — funnel stage advancement +- `mechanisms/counts.py`: `PoissonIntensity`, `RecencyDecayIntensity` — touch/session counts +- `mechanisms/categorical.py`: `CategoricalInfluence`, `CHANNEL_QUALITY_SCORES` +- `mechanisms/measurement.py`: `NoisyProxy`, `NoisyCategorization`, `ProxyCompression` +- `mechanisms/policies.py`: `assign_mechanisms()` — motif-family-aware `MechanismAssignment` factory +- 74 tests; total 437 passing + ### Milestone 5 — Population Generation ✓ (v0.3.0 in PR) - `leadforge/simulation/population.py`: `build_population()` — accounts (3 latent traits), contacts (4 latent traits, conditional on account), leads (1 latent trait, FK-consistent), diff --git a/leadforge/mechanisms/base.py b/leadforge/mechanisms/base.py new file mode 100644 index 0000000..217c36a --- /dev/null +++ b/leadforge/mechanisms/base.py @@ -0,0 +1,155 @@ +"""Mechanism base classes and shared contracts. + +All mechanism implementations inherit from :class:`Mechanism` and expose a +single ``sample(context, rng)`` method. :class:`MechanismContext` is the +universal carrier of state passed into every ``sample`` call. +:class:`MechanismAssignment` holds the named mechanism instances that the +simulation engine will invoke on each time step. +""" + +from __future__ import annotations + +import json +import random +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import Any + +# --------------------------------------------------------------------------- +# Context +# --------------------------------------------------------------------------- + + +@dataclass +class MechanismContext: + """State snapshot passed to every :meth:`Mechanism.sample` call. + + Attributes: + latents: Merged latent traits for the relevant entity set + (account + contact + lead for a full lead context). + stage: Current funnel stage of the lead, or ``None`` if not + applicable (e.g. account-level mechanisms). + t: Day index within the simulation window (0-based). + extra: Mechanism-specific extra fields (e.g. ``"channel"``, + ``"rep_id"``). + """ + + latents: dict[str, float] = field(default_factory=dict) + stage: str | None = None + t: int = 0 + extra: dict[str, Any] = field(default_factory=dict) + + +# --------------------------------------------------------------------------- +# Abstract base +# --------------------------------------------------------------------------- + + +class Mechanism(ABC): + """Abstract base class for all leadforge mechanism types. + + Subclasses must implement :meth:`sample` and :meth:`to_dict`. + """ + + @property + @abstractmethod + def name(self) -> str: + """Short machine-readable identifier for this mechanism type.""" + + @abstractmethod + def sample(self, context: MechanismContext, rng: random.Random) -> Any: + """Draw one sample given *context* using *rng*. + + Args: + context: Current state snapshot. + rng: Seeded stdlib :class:`random.Random` instance. + + Returns: + A value whose type depends on the mechanism family + (float, int, str, bool, or ``None``). + """ + + @abstractmethod + def to_dict(self) -> dict[str, Any]: + """Return a JSON-serialisable representation of this mechanism.""" + + def to_json(self) -> str: + """Return a JSON string representation.""" + return json.dumps(self.to_dict(), indent=2) + + +# --------------------------------------------------------------------------- +# Mechanism summary +# --------------------------------------------------------------------------- + + +@dataclass +class MechanismSummary: + """Serialisable summary of the mechanism assignment for one world. + + Stored in ``mechanism_summary.json`` in ``research_instructor`` mode. + + Attributes: + motif_family: Name of the motif family that drove parameter choices. + conversion_hazard: Summary dict for the conversion hazard mechanism. + stage_transition: Summary dict for the stage-transition mechanism. + touch_intensity: Summary dict for the touch-count mechanism. + measurement: Summary dict for the measurement / proxy mechanism. + """ + + motif_family: str + conversion_hazard: dict[str, Any] + stage_transition: dict[str, Any] + touch_intensity: dict[str, Any] + measurement: dict[str, Any] + + def to_dict(self) -> dict[str, Any]: + return { + "motif_family": self.motif_family, + "conversion_hazard": self.conversion_hazard, + "stage_transition": self.stage_transition, + "touch_intensity": self.touch_intensity, + "measurement": self.measurement, + } + + def to_json(self) -> str: + return json.dumps(self.to_dict(), indent=2) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> MechanismSummary: + return cls( + motif_family=data["motif_family"], + conversion_hazard=data["conversion_hazard"], + stage_transition=data["stage_transition"], + touch_intensity=data["touch_intensity"], + measurement=data["measurement"], + ) + + +# --------------------------------------------------------------------------- +# Mechanism assignment +# --------------------------------------------------------------------------- + + +@dataclass +class MechanismAssignment: + """Named mechanism instances consumed by the simulation engine. + + All fields are populated by :func:`~leadforge.mechanisms.policies.assign_mechanisms`. + """ + + motif_family: str + conversion_hazard: Mechanism + stage_transition: Mechanism + touch_intensity: Mechanism + measurement: Mechanism + + def summary(self) -> MechanismSummary: + """Return a :class:`MechanismSummary` for serialisation.""" + return MechanismSummary( + motif_family=self.motif_family, + conversion_hazard=self.conversion_hazard.to_dict(), + stage_transition=self.stage_transition.to_dict(), + touch_intensity=self.touch_intensity.to_dict(), + measurement=self.measurement.to_dict(), + ) diff --git a/leadforge/mechanisms/categorical.py b/leadforge/mechanisms/categorical.py new file mode 100644 index 0000000..38a56a2 --- /dev/null +++ b/leadforge/mechanisms/categorical.py @@ -0,0 +1,68 @@ +"""Categorical influence mechanisms — channel and segment effects. + +These mechanisms map categorical context values (e.g. lead source channel, +industry segment) to a numeric influence score, allowing categorical features +to modulate latent dynamics without embedding them as continuous latents. +""" + +from __future__ import annotations + +import random +from typing import Any + +from leadforge.mechanisms.base import Mechanism, MechanismContext + + +class CategoricalInfluence(Mechanism): + """Map a categorical context key to a numeric score via a lookup table. + + Looks up ``context.extra[context_key]`` in *lookup* and returns the + corresponding float. Falls back to *default* if the value is absent or + unknown. + + Args: + context_key: The key to look up in ``context.extra``. + lookup: Mapping of category label → score in [0, 1]. + default: Score to return when the lookup key or value is absent. + """ + + def __init__( + self, + context_key: str, + lookup: dict[str, float], + default: float = 0.5, + ) -> None: + if not lookup: + raise ValueError("lookup must not be empty") + self._context_key = context_key + self._lookup = dict(lookup) + self._default = default + + @property + def name(self) -> str: + return "categorical_influence" + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + value = context.extra.get(self._context_key) + return self._lookup.get(str(value), self._default) if value is not None else self._default + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "context_key": self._context_key, + "lookup": self._lookup, + "default": self._default, + } + + +# --------------------------------------------------------------------------- +# V1 channel influence scores (used by policies.py) +# --------------------------------------------------------------------------- + +#: Default quality multipliers by lead source channel. +#: Partner referrals tend to arrive pre-qualified; outbound is colder. +CHANNEL_QUALITY_SCORES: dict[str, float] = { + "inbound_marketing": 0.55, + "sdr_outbound": 0.40, + "partner_referral": 0.70, +} diff --git a/leadforge/mechanisms/counts.py b/leadforge/mechanisms/counts.py new file mode 100644 index 0000000..eb80fb0 --- /dev/null +++ b/leadforge/mechanisms/counts.py @@ -0,0 +1,130 @@ +"""Count / event-intensity mechanisms — generate touch and session counts. + +These mechanisms answer "how many events of type X happen today for this lead?" +The simulation engine uses them to populate the ``touches`` and ``sessions`` +tables. +""" + +from __future__ import annotations + +import math +import random +from typing import Any + +from leadforge.mechanisms.base import Mechanism, MechanismContext + + +class PoissonIntensity(Mechanism): + """Poisson-distributed event count driven by latent traits. + + Expected count per day:: + + lambda = base_rate * exp(sum(weight_i * latents[key_i])) + + This is a log-linear intensity model: latent weights are on the log scale + so they multiplicatively modulate the base rate. + + Args: + base_rate: Expected daily event count when all latent keys are 0. + weights: Mapping of latent-key → log-scale weight. + """ + + def __init__(self, base_rate: float, weights: dict[str, float] | None = None) -> None: + if base_rate <= 0: + raise ValueError(f"base_rate must be positive, got {base_rate}") + self._base_rate = base_rate + self._weights: dict[str, float] = dict(weights) if weights else {} + + @property + def name(self) -> str: + return "poisson_intensity" + + def expected_count(self, latents: dict[str, float]) -> float: + """Return the expected daily event count for the given latent state.""" + log_rate = math.log(self._base_rate) + sum( + self._weights.get(k, 0.0) * latents.get(k, 0.0) for k in self._weights + ) + return math.exp(log_rate) + + def sample(self, context: MechanismContext, rng: random.Random) -> int: + """Draw a Poisson count for today.""" + lam = self.expected_count(context.latents) + # Simulate Poisson via waiting times (exact for moderate lambda). + count = 0 + p = math.exp(-lam) + cum = p + u = rng.random() + while u > cum: + count += 1 + p *= lam / count + cum += p + if count > 1000: # safety cap + break + return count + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "base_rate": self._base_rate, + "weights": self._weights, + } + + +class RecencyDecayIntensity(Mechanism): + """Poisson intensity that decays exponentially with time since lead creation. + + Models the observation that CRM activity is front-loaded: most touches + happen early in the sales cycle. + + Args: + base_rate: Expected daily count at ``t=0``. + decay_factor: Per-day multiplicative decay in (0, 1]. At day *t*, + the effective rate is ``base_rate * decay_factor ** t``. + floor_rate: Minimum daily rate (floor applied after decay). + """ + + def __init__( + self, + base_rate: float, + decay_factor: float = 0.97, + floor_rate: float = 0.01, + ) -> None: + if base_rate <= 0: + raise ValueError(f"base_rate must be positive, got {base_rate}") + if not (0.0 < decay_factor <= 1.0): + raise ValueError(f"decay_factor must be in (0, 1], got {decay_factor}") + if floor_rate < 0: + raise ValueError(f"floor_rate must be non-negative, got {floor_rate}") + self._base_rate = base_rate + self._decay = decay_factor + self._floor = floor_rate + + @property + def name(self) -> str: + return "recency_decay_intensity" + + def expected_count(self, t: int) -> float: + """Return the expected daily count at day *t*.""" + return max(self._floor, self._base_rate * (self._decay**t)) + + def sample(self, context: MechanismContext, rng: random.Random) -> int: + lam = self.expected_count(context.t) + count = 0 + p = math.exp(-lam) + cum = p + u = rng.random() + while u > cum: + count += 1 + p *= lam / count + cum += p + if count > 1000: + break + return count + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "base_rate": self._base_rate, + "decay_factor": self._decay, + "floor_rate": self._floor, + } diff --git a/leadforge/mechanisms/hazards.py b/leadforge/mechanisms/hazards.py new file mode 100644 index 0000000..a288e88 --- /dev/null +++ b/leadforge/mechanisms/hazards.py @@ -0,0 +1,72 @@ +"""Conversion hazard mechanism — daily probability of lead conversion. + +:class:`ConversionHazard` is the primary mechanism called by the simulation +engine on each day step for each active lead. It maps the merged latent state +to a daily conversion probability via a :class:`~leadforge.mechanisms.scores.LatentScore`. +""" + +from __future__ import annotations + +import random +from typing import Any + +from leadforge.mechanisms.base import Mechanism, MechanismContext +from leadforge.mechanisms.scores import LatentScore + + +class ConversionHazard(Mechanism): + """Daily conversion probability driven by latent score. + + Daily probability:: + + p_convert = clip(base_rate + scale * score, 0, max_daily_rate) + + Args: + score_mech: A :class:`~leadforge.mechanisms.scores.LatentScore` + instance that maps latents → [0, 1] score. + base_rate: Minimum daily conversion probability (intercept). + scale: Multiplier on the latent score. + max_daily_rate: Hard cap on the daily probability. + """ + + def __init__( + self, + score_mech: LatentScore, + base_rate: float = 0.005, + scale: float = 0.05, + max_daily_rate: float = 0.20, + ) -> None: + if not (0.0 <= base_rate <= 1.0): + raise ValueError(f"base_rate must be in [0, 1], got {base_rate}") + if scale < 0: + raise ValueError(f"scale must be non-negative, got {scale}") + if not (0.0 < max_daily_rate <= 1.0): + raise ValueError(f"max_daily_rate must be in (0, 1], got {max_daily_rate}") + self._score_mech = score_mech + self._base_rate = base_rate + self._scale = scale + self._max_daily_rate = max_daily_rate + + @property + def name(self) -> str: + return "conversion_hazard" + + def daily_probability(self, latents: dict[str, float]) -> float: + """Return the daily conversion probability for the given latent state.""" + score = self._score_mech.score(latents) + p = self._base_rate + self._scale * score + return max(0.0, min(self._max_daily_rate, p)) + + def sample(self, context: MechanismContext, rng: random.Random) -> bool: + """Return ``True`` if the lead converts on this day step.""" + p = self.daily_probability(context.latents) + return rng.random() < p + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "score_mech": self._score_mech.to_dict(), + "base_rate": self._base_rate, + "scale": self._scale, + "max_daily_rate": self._max_daily_rate, + } diff --git a/leadforge/mechanisms/influence.py b/leadforge/mechanisms/influence.py new file mode 100644 index 0000000..50443ed --- /dev/null +++ b/leadforge/mechanisms/influence.py @@ -0,0 +1,199 @@ +"""Influence mechanisms — latent-to-latent propagation along graph edges. + +Each mechanism maps a subset of latent traits from ``context.latents`` to a +single float output in [0, 1], representing the influenced child node's value. +""" + +from __future__ import annotations + +import math +import random +from typing import Any + +from leadforge.mechanisms.base import Mechanism, MechanismContext + + +def _weighted_sum(latents: dict[str, float], weights: dict[str, float], bias: float) -> float: + return bias + sum(weights.get(k, 0.0) * latents.get(k, 0.0) for k in weights) + + +def _sigmoid(x: float) -> float: + return 1.0 / (1.0 + math.exp(-x)) + + +class AdditiveInfluence(Mechanism): + """Weighted sum of parent latents, clipped to [0, 1]. + + Args: + weights: Mapping of latent-key → weight. + bias: Additive intercept before clipping. + """ + + def __init__(self, weights: dict[str, float], bias: float = 0.0) -> None: + self._weights = dict(weights) + self._bias = bias + + @property + def name(self) -> str: + return "additive_influence" + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + raw = _weighted_sum(context.latents, self._weights, self._bias) + return max(0.0, min(1.0, raw)) + + def to_dict(self) -> dict[str, Any]: + return {"name": self.name, "weights": self._weights, "bias": self._bias} + + +class LogisticInfluence(Mechanism): + """Logistic (sigmoid) transform of a weighted latent sum. + + Args: + weights: Mapping of latent-key → weight. + bias: Additive intercept inside the sigmoid. + temperature: Inverse scale applied to the linear combination + (higher = sharper decision boundary; default 1.0). + """ + + def __init__( + self, + weights: dict[str, float], + bias: float = 0.0, + temperature: float = 1.0, + ) -> None: + if temperature <= 0: + raise ValueError(f"temperature must be positive, got {temperature}") + self._weights = dict(weights) + self._bias = bias + self._temperature = temperature + + @property + def name(self) -> str: + return "logistic_influence" + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + raw = _weighted_sum(context.latents, self._weights, self._bias) + return _sigmoid(raw * self._temperature) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "weights": self._weights, + "bias": self._bias, + "temperature": self._temperature, + } + + +class SaturatingInfluence(Mechanism): + """Saturating (tanh) transform of a weighted latent sum, mapped to [0, 1]. + + Args: + weights: Mapping of latent-key → weight. + bias: Additive intercept inside the tanh. + scale: Pre-tanh multiplier controlling curvature. + """ + + def __init__( + self, + weights: dict[str, float], + bias: float = 0.0, + scale: float = 1.0, + ) -> None: + self._weights = dict(weights) + self._bias = bias + self._scale = scale + + @property + def name(self) -> str: + return "saturating_influence" + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + raw = _weighted_sum(context.latents, self._weights, self._bias) + return (math.tanh(raw * self._scale) + 1.0) / 2.0 + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "weights": self._weights, + "bias": self._bias, + "scale": self._scale, + } + + +class ThresholdInfluence(Mechanism): + """Hard threshold: returns 1.0 if weighted sum ≥ *threshold*, else 0.0. + + Args: + weights: Mapping of latent-key → weight. + threshold: Decision boundary. + bias: Additive intercept before threshold comparison. + """ + + def __init__( + self, + weights: dict[str, float], + threshold: float = 0.5, + bias: float = 0.0, + ) -> None: + self._weights = dict(weights) + self._threshold = threshold + self._bias = bias + + @property + def name(self) -> str: + return "threshold_influence" + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + raw = _weighted_sum(context.latents, self._weights, self._bias) + return 1.0 if raw >= self._threshold else 0.0 + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "weights": self._weights, + "threshold": self._threshold, + "bias": self._bias, + } + + +class InteractionTerm(Mechanism): + """Product of two latent keys, optionally scaled and biased. + + Captures synergy/antagonism effects (e.g. fit × authority). + + Args: + key_a: First latent key. + key_b: Second latent key. + weight: Scalar multiplier on the product. + bias: Additive term; result clipped to [0, 1]. + """ + + def __init__( + self, + key_a: str, + key_b: str, + weight: float = 1.0, + bias: float = 0.0, + ) -> None: + self._key_a = key_a + self._key_b = key_b + self._weight = weight + self._bias = bias + + @property + def name(self) -> str: + return "interaction_term" + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + a = context.latents.get(self._key_a, 0.0) + b = context.latents.get(self._key_b, 0.0) + return max(0.0, min(1.0, self._weight * a * b + self._bias)) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "key_a": self._key_a, + "key_b": self._key_b, + "weight": self._weight, + "bias": self._bias, + } diff --git a/leadforge/mechanisms/measurement.py b/leadforge/mechanisms/measurement.py new file mode 100644 index 0000000..d7b9003 --- /dev/null +++ b/leadforge/mechanisms/measurement.py @@ -0,0 +1,174 @@ +"""Measurement mechanisms — hidden truth → noisy CRM observations. + +These mechanisms transform latent float values into the imperfect proxies +that appear in CRM exports. They are applied during the rendering layer +(M9) to introduce realistic data-quality issues. +""" + +from __future__ import annotations + +import random +from typing import Any + +from leadforge.mechanisms.base import Mechanism, MechanismContext + + +class NoisyProxy(Mechanism): + """Add Gaussian noise to a latent value and optionally inject missingness. + + Reads ``context.latents[latent_key]``, adds noise, clips to [0, 1], then + returns ``None`` with probability *missing_rate*. + + Args: + latent_key: Key to read from ``context.latents``. + noise_std: Standard deviation of the Gaussian noise term. + missing_rate: Probability that the observed value is ``None`` + (simulates incomplete enrichment / data gaps). + """ + + def __init__( + self, + latent_key: str, + noise_std: float = 0.10, + missing_rate: float = 0.05, + ) -> None: + if noise_std < 0: + raise ValueError(f"noise_std must be non-negative, got {noise_std}") + if not (0.0 <= missing_rate <= 1.0): + raise ValueError(f"missing_rate must be in [0, 1], got {missing_rate}") + self._key = latent_key + self._noise_std = noise_std + self._missing_rate = missing_rate + + @property + def name(self) -> str: + return "noisy_proxy" + + def sample(self, context: MechanismContext, rng: random.Random) -> float | None: + if rng.random() < self._missing_rate: + return None + true_val = context.latents.get(self._key, 0.5) + noisy = true_val + rng.gauss(0.0, self._noise_std) if self._noise_std > 0 else true_val + return max(0.0, min(1.0, noisy)) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "latent_key": self._key, + "noise_std": self._noise_std, + "missing_rate": self._missing_rate, + } + + +class NoisyCategorization(Mechanism): + """Randomly flip a categorical value to simulate CRM data-entry noise. + + With probability *confusion_prob*, replaces the value found in + ``context.extra[context_key]`` with a uniformly drawn alternative from + *categories*. With probability ``1 - confusion_prob``, returns the + true value unchanged. + + Args: + context_key: Key to read from ``context.extra``. + categories: All valid category labels. + confusion_prob: Per-record mislabelling probability. + missing_rate: Probability the field is ``None``. + """ + + def __init__( + self, + context_key: str, + categories: list[str], + confusion_prob: float = 0.05, + missing_rate: float = 0.03, + ) -> None: + if not categories: + raise ValueError("categories must not be empty") + if not (0.0 <= confusion_prob <= 1.0): + raise ValueError(f"confusion_prob must be in [0, 1], got {confusion_prob}") + if not (0.0 <= missing_rate <= 1.0): + raise ValueError(f"missing_rate must be in [0, 1], got {missing_rate}") + self._key = context_key + self._categories = list(categories) + self._confusion_prob = confusion_prob + self._missing_rate = missing_rate + + @property + def name(self) -> str: + return "noisy_categorization" + + def sample(self, context: MechanismContext, rng: random.Random) -> str | None: + if rng.random() < self._missing_rate: + return None + true_val = context.extra.get(self._key) + if rng.random() < self._confusion_prob or true_val not in self._categories: + return rng.choice(self._categories) + return str(true_val) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "context_key": self._key, + "categories": self._categories, + "confusion_prob": self._confusion_prob, + "missing_rate": self._missing_rate, + } + + +class ProxyCompression(Mechanism): + """Compress a continuous latent to a coarse ordinal label. + + Partitions [0, 1] into bands using *thresholds* and maps each band to the + corresponding label in *labels*. Simulates CRM fields like lead score + tiers ("low" / "medium" / "high") that collapse a continuous signal. + + Args: + latent_key: Key to read from ``context.latents``. + thresholds: Strictly increasing cut-points in (0, 1). With *k* + thresholds, *k+1* labels are required. + labels: Ordered labels, one per band (lowest band first). + missing_rate: Probability the field is ``None``. + """ + + def __init__( + self, + latent_key: str, + thresholds: list[float], + labels: list[str], + missing_rate: float = 0.05, + ) -> None: + if len(labels) != len(thresholds) + 1: + raise ValueError( + f"Expected {len(thresholds) + 1} labels for {len(thresholds)} thresholds, " + f"got {len(labels)}" + ) + if thresholds != sorted(thresholds): + raise ValueError("thresholds must be strictly increasing") + if not (0.0 <= missing_rate <= 1.0): + raise ValueError(f"missing_rate must be in [0, 1], got {missing_rate}") + self._key = latent_key + self._thresholds = list(thresholds) + self._labels = list(labels) + self._missing_rate = missing_rate + + @property + def name(self) -> str: + return "proxy_compression" + + def sample(self, context: MechanismContext, rng: random.Random) -> str | None: + if rng.random() < self._missing_rate: + return None + val = context.latents.get(self._key, 0.5) + for i, threshold in enumerate(self._thresholds): + if val < threshold: + return self._labels[i] + return self._labels[-1] + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "latent_key": self._key, + "thresholds": self._thresholds, + "labels": self._labels, + "missing_rate": self._missing_rate, + } diff --git a/leadforge/mechanisms/policies.py b/leadforge/mechanisms/policies.py new file mode 100644 index 0000000..e3dec7a --- /dev/null +++ b/leadforge/mechanisms/policies.py @@ -0,0 +1,211 @@ +"""Mechanism assignment policy — wires mechanism instances to a world. + +:func:`assign_mechanisms` is the single entry point. It inspects the active +motif family and constructs a :class:`~leadforge.mechanisms.base.MechanismAssignment` +whose parameters reflect the structural bias of that world. + +Motif-family parameter tuning +------------------------------ +Each motif family tilts the mechanism parameters so the DGP is consistent +with the hidden world structure selected by the sampler: + +- **fit_dominant** — conversion hazard weighted heavily on account fit and + budget readiness; stage transition also fit-driven. +- **intent_dominant** — conversion hazard weighted on engagement propensity + and problem awareness. +- **sales_execution_sensitive** — stage transition heavily penalised by sales + friction; low base conversion rate. +- **demo_trial_mediated** — touch intensity is higher; conversion gated on + engagement. +- **buying_committee_friction** — very low base conversion rate; authority + and friction interact. +""" + +from __future__ import annotations + +import random +from typing import Any + +from leadforge.mechanisms.base import MechanismAssignment +from leadforge.mechanisms.counts import RecencyDecayIntensity +from leadforge.mechanisms.hazards import ConversionHazard +from leadforge.mechanisms.measurement import NoisyProxy +from leadforge.mechanisms.scores import LatentScore +from leadforge.mechanisms.transitions import HazardTransition + +# --------------------------------------------------------------------------- +# Motif-family parameter tables +# --------------------------------------------------------------------------- + +# Each entry: {latent_key: weight} for the LatentScore used by the +# ConversionHazard. Positive = facilitates conversion; negative = inhibits. +_CONVERSION_SCORE_WEIGHTS: dict[str, dict[str, float]] = { + "fit_dominant": { + "latent_account_fit": 2.5, + "latent_budget_readiness": 1.5, + "latent_problem_awareness": 0.5, + "latent_engagement_propensity": 0.5, + "latent_sales_friction": -0.5, + }, + "intent_dominant": { + "latent_engagement_propensity": 2.5, + "latent_problem_awareness": 1.5, + "latent_account_fit": 0.5, + "latent_sales_friction": -0.5, + }, + "sales_execution_sensitive": { + "latent_account_fit": 1.0, + "latent_engagement_propensity": 1.0, + "latent_responsiveness": 1.5, + "latent_sales_friction": -2.0, + }, + "demo_trial_mediated": { + "latent_engagement_propensity": 2.0, + "latent_problem_awareness": 1.0, + "latent_account_fit": 1.0, + "latent_sales_friction": -0.5, + }, + "buying_committee_friction": { + "latent_account_fit": 1.5, + "latent_contact_authority": 1.5, + "latent_budget_readiness": 1.0, + "latent_sales_friction": -2.5, + }, +} + +# Conversion hazard base_rate and scale per motif family. +_HAZARD_PARAMS: dict[str, dict[str, float]] = { + "fit_dominant": {"base_rate": 0.008, "scale": 0.06}, + "intent_dominant": {"base_rate": 0.010, "scale": 0.07}, + "sales_execution_sensitive": {"base_rate": 0.004, "scale": 0.05}, + "demo_trial_mediated": {"base_rate": 0.007, "scale": 0.06}, + "buying_committee_friction": {"base_rate": 0.003, "scale": 0.04}, +} + +# Stage-transition HazardTransition score weights per motif family. +_TRANSITION_SCORE_WEIGHTS: dict[str, dict[str, float]] = { + "fit_dominant": { + "latent_account_fit": 2.0, + "latent_problem_awareness": 1.0, + "latent_responsiveness": 0.5, + }, + "intent_dominant": { + "latent_engagement_propensity": 2.0, + "latent_problem_awareness": 1.5, + "latent_responsiveness": 0.5, + }, + "sales_execution_sensitive": { + "latent_responsiveness": 2.0, + "latent_engagement_propensity": 1.0, + "latent_sales_friction": -1.5, + }, + "demo_trial_mediated": { + "latent_engagement_propensity": 2.5, + "latent_account_fit": 0.5, + }, + "buying_committee_friction": { + "latent_contact_authority": 2.0, + "latent_account_fit": 1.0, + "latent_sales_friction": -2.0, + }, +} + +# Touch intensity (RecencyDecayIntensity) base_rate per motif family. +_TOUCH_BASE_RATES: dict[str, float] = { + "fit_dominant": 0.40, + "intent_dominant": 0.55, + "sales_execution_sensitive": 0.35, + "demo_trial_mediated": 0.60, + "buying_committee_friction": 0.30, +} + +# Fallback weights/params for unknown motif families. +_DEFAULT_CONVERSION_WEIGHTS: dict[str, float] = { + "latent_account_fit": 1.0, + "latent_engagement_propensity": 1.0, + "latent_sales_friction": -0.5, +} +_DEFAULT_HAZARD_PARAMS: dict[str, float] = {"base_rate": 0.006, "scale": 0.05} +_DEFAULT_TOUCH_BASE_RATE: float = 0.40 + + +# --------------------------------------------------------------------------- +# Public entry point +# --------------------------------------------------------------------------- + + +def assign_mechanisms( + motif_family: str, + rng: random.Random, +) -> MechanismAssignment: + """Build a :class:`~leadforge.mechanisms.base.MechanismAssignment` for *motif_family*. + + Parameters are tuned to the structural bias of the motif family so the + resulting simulation is consistent with the hidden world sampled by + :func:`~leadforge.structure.sampler.sample_hidden_graph`. + + Args: + motif_family: Name of the active motif family (e.g. ``"fit_dominant"``). + rng: Seeded :class:`random.Random` instance for any stochastic + parameter perturbation (currently unused but reserved for future + use so the signature is stable). + + Returns: + A fully populated :class:`~leadforge.mechanisms.base.MechanismAssignment`. + """ + conv_weights = _CONVERSION_SCORE_WEIGHTS.get(motif_family, _DEFAULT_CONVERSION_WEIGHTS) + hazard_p = _HAZARD_PARAMS.get(motif_family, _DEFAULT_HAZARD_PARAMS) + trans_weights = _TRANSITION_SCORE_WEIGHTS.get(motif_family, _DEFAULT_CONVERSION_WEIGHTS) + touch_rate = _TOUCH_BASE_RATES.get(motif_family, _DEFAULT_TOUCH_BASE_RATE) + + conversion_hazard = ConversionHazard( + score_mech=LatentScore(weights=conv_weights, bias=-1.5), + base_rate=hazard_p["base_rate"], + scale=hazard_p["scale"], + ) + + stage_transition = HazardTransition( + score_mech=LatentScore(weights=trans_weights, bias=-1.0), + base_rate=0.05, + scale=0.15, + min_dwell_days=2, + ) + + touch_intensity = RecencyDecayIntensity( + base_rate=touch_rate, + decay_factor=0.97, + floor_rate=0.02, + ) + + measurement = NoisyProxy( + latent_key="latent_account_fit", + noise_std=0.10, + missing_rate=0.05, + ) + + return MechanismAssignment( + motif_family=motif_family, + conversion_hazard=conversion_hazard, + stage_transition=stage_transition, + touch_intensity=touch_intensity, + measurement=measurement, + ) + + +def mechanism_params_for_motif(motif_family: str) -> dict[str, Any]: + """Return a plain dict of the parameter tables for *motif_family*. + + Useful for inspection, testing, and mechanism summary rendering without + constructing mechanism objects. + """ + return { + "motif_family": motif_family, + "conversion_score_weights": _CONVERSION_SCORE_WEIGHTS.get( + motif_family, _DEFAULT_CONVERSION_WEIGHTS + ), + "hazard_params": _HAZARD_PARAMS.get(motif_family, _DEFAULT_HAZARD_PARAMS), + "transition_score_weights": _TRANSITION_SCORE_WEIGHTS.get( + motif_family, _DEFAULT_CONVERSION_WEIGHTS + ), + "touch_base_rate": _TOUCH_BASE_RATES.get(motif_family, _DEFAULT_TOUCH_BASE_RATE), + } diff --git a/leadforge/mechanisms/scores.py b/leadforge/mechanisms/scores.py new file mode 100644 index 0000000..143dc68 --- /dev/null +++ b/leadforge/mechanisms/scores.py @@ -0,0 +1,55 @@ +"""Latent scoring — maps merged latent state to a scalar score in [0, 1]. + +:class:`LatentScore` is the core building block used by +:class:`~leadforge.mechanisms.hazards.ConversionHazard` and +:class:`~leadforge.mechanisms.transitions.HazardTransition` to collapse +multiple latent traits into a single predictive signal. +""" + +from __future__ import annotations + +import math +import random +from typing import Any + +from leadforge.mechanisms.base import Mechanism, MechanismContext + + +class LatentScore(Mechanism): + """Logistic score from a weighted combination of latent keys. + + Computes:: + + score = sigmoid(bias + sum(weight_i * latents[key_i])) + + Keys absent from ``context.latents`` contribute 0. + + Args: + weights: Mapping of latent-key → weight. Positive weights increase + the score; negative weights decrease it. + bias: Additive intercept (controls the base conversion propensity + before any latent influence). + """ + + def __init__(self, weights: dict[str, float], bias: float = 0.0) -> None: + if not weights: + raise ValueError("weights must not be empty") + self._weights = dict(weights) + self._bias = bias + + @property + def name(self) -> str: + return "latent_score" + + def score(self, latents: dict[str, float]) -> float: + """Return the [0, 1] score without sampling noise.""" + linear = self._bias + sum( + self._weights.get(k, 0.0) * latents.get(k, 0.0) for k in self._weights + ) + return 1.0 / (1.0 + math.exp(-linear)) + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + return self.score(context.latents) + + def to_dict(self) -> dict[str, Any]: + return {"name": self.name, "weights": self._weights, "bias": self._bias} diff --git a/leadforge/mechanisms/static.py b/leadforge/mechanisms/static.py new file mode 100644 index 0000000..db1230a --- /dev/null +++ b/leadforge/mechanisms/static.py @@ -0,0 +1,138 @@ +"""Static latent mechanisms — used for trait sampling at population time. + +These mechanisms draw a single value from a fixed distribution given only an +RNG; they do not depend on parent state in the graph. They are provided as a +library for higher-level code (e.g. future mechanism assignment passes) rather +than being called directly by the simulation engine. +""" + +from __future__ import annotations + +import random +from typing import Any + +from leadforge.mechanisms.base import Mechanism, MechanismContext + + +class CategoricalDraw(Mechanism): + """Draw one category from a weighted categorical distribution. + + Args: + categories: Ordered list of category labels. + weights: Non-negative weights parallel to *categories*; need not sum + to 1 (normalised internally). + """ + + def __init__(self, categories: list[str], weights: list[float]) -> None: + if len(categories) != len(weights): + raise ValueError("categories and weights must have the same length") + if not categories: + raise ValueError("categories must not be empty") + if any(w < 0 for w in weights): + raise ValueError("all weights must be non-negative") + total = sum(weights) + if total <= 0: + raise ValueError("weights must sum to a positive value") + self._categories = list(categories) + self._weights = [w / total for w in weights] + + @property + def name(self) -> str: + return "categorical_draw" + + def sample(self, context: MechanismContext, rng: random.Random) -> str: + return rng.choices(self._categories, weights=self._weights, k=1)[0] + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "categories": self._categories, + "weights": self._weights, + } + + +class BoundedNumericDraw(Mechanism): + """Draw a float from a clipped Gaussian in [*lo*, *hi*]. + + Args: + lo: Lower bound (inclusive). + hi: Upper bound (inclusive). + mean: Mean of the underlying Gaussian (clamped to [lo, hi]). + std: Standard deviation of the underlying Gaussian. + """ + + def __init__( + self, + lo: float = 0.0, + hi: float = 1.0, + mean: float = 0.5, + std: float = 0.2, + ) -> None: + if lo >= hi: + raise ValueError(f"lo ({lo}) must be < hi ({hi})") + if std <= 0: + raise ValueError(f"std must be positive, got {std}") + self._lo = lo + self._hi = hi + self._mean = max(lo, min(hi, mean)) + self._std = std + + @property + def name(self) -> str: + return "bounded_numeric_draw" + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + return max(self._lo, min(self._hi, rng.gauss(self._mean, self._std))) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "lo": self._lo, + "hi": self._hi, + "mean": self._mean, + "std": self._std, + } + + +class MixtureDraw(Mechanism): + """Draw from a finite mixture of :class:`BoundedNumericDraw` components. + + Args: + components: List of ``(mean, std)`` pairs, one per mixture component. + All components share the same ``[lo, hi]`` range. + mix_weights: Mixture weights (need not sum to 1). + lo: Shared lower bound. + hi: Shared upper bound. + """ + + def __init__( + self, + components: list[tuple[float, float]], + mix_weights: list[float], + lo: float = 0.0, + hi: float = 1.0, + ) -> None: + if not components: + raise ValueError("components must not be empty") + if len(components) != len(mix_weights): + raise ValueError("components and mix_weights must have the same length") + total = sum(mix_weights) + if total <= 0: + raise ValueError("mix_weights must sum to a positive value") + self._drawers = [BoundedNumericDraw(lo, hi, m, s) for m, s in components] + self._mix_weights = [w / total for w in mix_weights] + + @property + def name(self) -> str: + return "mixture_draw" + + def sample(self, context: MechanismContext, rng: random.Random) -> float: + drawer = rng.choices(self._drawers, weights=self._mix_weights, k=1)[0] + return drawer.sample(context, rng) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "components": [d.to_dict() for d in self._drawers], + "mix_weights": self._mix_weights, + } diff --git a/leadforge/mechanisms/transitions.py b/leadforge/mechanisms/transitions.py new file mode 100644 index 0000000..8c90f35 --- /dev/null +++ b/leadforge/mechanisms/transitions.py @@ -0,0 +1,158 @@ +"""Stage-transition mechanisms — advance leads through the funnel. + +:class:`HazardTransition` decides whether a lead advances on a given day. +:class:`StageSequence` defines the ordered funnel stages and resolves the +next stage name. The simulation engine calls these on each day step per lead. +""" + +from __future__ import annotations + +import random +from typing import Any + +from leadforge.mechanisms.base import Mechanism, MechanismContext +from leadforge.mechanisms.scores import LatentScore + +# Default v1 funnel stage ordering (matches narrative.yaml funnel_stages). +_DEFAULT_STAGE_ORDER = ( + "mql", + "sal", + "sql", + "demo_scheduled", + "demo_completed", + "proposal_sent", + "negotiation", + "closed_won", + "closed_lost", +) + +# Stages from which advancement is no longer possible. +_TERMINAL_STAGES = frozenset({"closed_won", "closed_lost"}) + + +class StageSequence(Mechanism): + """Ordered funnel stage registry. + + Returns the next stage name given the current one, or ``None`` if the + current stage is terminal or unknown. + + Args: + stage_order: Ordered tuple of stage names. The last stage is + terminal (no advancement). + terminal_stages: Set of stage names from which no further + advancement occurs. + """ + + def __init__( + self, + stage_order: tuple[str, ...] = _DEFAULT_STAGE_ORDER, + terminal_stages: frozenset[str] = _TERMINAL_STAGES, + ) -> None: + self._order = stage_order + self._terminal = terminal_stages + self._next: dict[str, str] = {} + for i, stage in enumerate(stage_order[:-1]): + if stage not in terminal_stages: + self._next[stage] = stage_order[i + 1] + + @property + def name(self) -> str: + return "stage_sequence" + + def next_stage(self, current: str) -> str | None: + """Return the stage after *current*, or ``None`` if terminal.""" + return self._next.get(current) + + def is_terminal(self, stage: str) -> bool: + return stage in self._terminal + + def sample(self, context: MechanismContext, rng: random.Random) -> str | None: + """Return the stage after ``context.stage``, or ``None`` if terminal.""" + if context.stage is None: + return None + return self.next_stage(context.stage) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "stage_order": list(self._order), + "terminal_stages": sorted(self._terminal), + } + + +class HazardTransition(Mechanism): + """Discrete-time hazard for stage advancement. + + On each day step, computes a transition probability from the lead's + latent score and returns ``True`` if the lead should advance. + + Daily probability:: + + p_advance = clip(base_rate + scale * score, 0, max_daily_rate) + + A minimum dwell time enforces that leads cannot skip through stages + unrealistically quickly. + + Args: + score_mech: :class:`~leadforge.mechanisms.scores.LatentScore` + mapping merged latents → [0, 1] score. + base_rate: Minimum daily advancement probability. + scale: Score multiplier. + max_daily_rate: Hard cap on daily probability. + min_dwell_days: Minimum days in the current stage before any + advancement can occur. + """ + + def __init__( + self, + score_mech: LatentScore, + base_rate: float = 0.03, + scale: float = 0.10, + max_daily_rate: float = 0.25, + min_dwell_days: int = 1, + ) -> None: + if not (0.0 <= base_rate <= 1.0): + raise ValueError(f"base_rate must be in [0, 1], got {base_rate}") + if scale < 0: + raise ValueError(f"scale must be non-negative, got {scale}") + if not (0.0 < max_daily_rate <= 1.0): + raise ValueError(f"max_daily_rate must be in (0, 1], got {max_daily_rate}") + if min_dwell_days < 0: + raise ValueError(f"min_dwell_days must be >= 0, got {min_dwell_days}") + self._score_mech = score_mech + self._base_rate = base_rate + self._scale = scale + self._max_daily_rate = max_daily_rate + self._min_dwell = min_dwell_days + + @property + def name(self) -> str: + return "hazard_transition" + + def daily_probability(self, latents: dict[str, float], dwell: int) -> float: + """Return the daily advancement probability given *dwell* days in stage.""" + if dwell < self._min_dwell: + return 0.0 + score = self._score_mech.score(latents) + p = self._base_rate + self._scale * score + return max(0.0, min(self._max_daily_rate, p)) + + def sample(self, context: MechanismContext, rng: random.Random) -> bool: + """Return ``True`` if the lead advances today. + + ``context.extra["dwell_days"]`` should carry the number of days the + lead has spent in the current stage. Defaults to 0 if absent. + """ + dwell = int(context.extra.get("dwell_days", 0)) + p = self.daily_probability(context.latents, dwell) + return rng.random() < p + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "score_mech": self._score_mech.to_dict(), + "base_rate": self._base_rate, + "scale": self._scale, + "max_daily_rate": self._max_daily_rate, + "min_dwell_days": self._min_dwell, + } diff --git a/tests/mechanisms/__init__.py b/tests/mechanisms/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/mechanisms/test_mechanisms.py b/tests/mechanisms/test_mechanisms.py new file mode 100644 index 0000000..a9b1897 --- /dev/null +++ b/tests/mechanisms/test_mechanisms.py @@ -0,0 +1,504 @@ +"""Tests for the leadforge mechanism layer (M6).""" + +from __future__ import annotations + +import json +import random + +import pytest + +from leadforge.mechanisms.base import MechanismAssignment, MechanismContext, MechanismSummary +from leadforge.mechanisms.categorical import CHANNEL_QUALITY_SCORES, CategoricalInfluence +from leadforge.mechanisms.counts import PoissonIntensity, RecencyDecayIntensity +from leadforge.mechanisms.hazards import ConversionHazard +from leadforge.mechanisms.influence import ( + AdditiveInfluence, + InteractionTerm, + LogisticInfluence, + SaturatingInfluence, + ThresholdInfluence, +) +from leadforge.mechanisms.measurement import NoisyCategorization, NoisyProxy, ProxyCompression +from leadforge.mechanisms.policies import assign_mechanisms, mechanism_params_for_motif +from leadforge.mechanisms.scores import LatentScore +from leadforge.mechanisms.static import BoundedNumericDraw, CategoricalDraw, MixtureDraw +from leadforge.mechanisms.transitions import HazardTransition, StageSequence +from leadforge.structure.motifs import MOTIF_FAMILY_NAMES + +_LATENTS = { + "latent_account_fit": 0.7, + "latent_budget_readiness": 0.6, + "latent_engagement_propensity": 0.8, + "latent_problem_awareness": 0.5, + "latent_contact_authority": 0.6, + "latent_responsiveness": 0.55, + "latent_sales_friction": 0.3, + "latent_process_maturity": 0.5, +} +_CTX = MechanismContext(latents=_LATENTS, stage="mql", t=5) + + +def _rng(seed: int = 0) -> random.Random: + return random.Random(seed) # noqa: S311 + + +# =========================================================================== +# MechanismContext +# =========================================================================== + + +def test_context_defaults() -> None: + ctx = MechanismContext() + assert ctx.latents == {} + assert ctx.stage is None + assert ctx.t == 0 + assert ctx.extra == {} + + +# =========================================================================== +# Static mechanisms +# =========================================================================== + + +def test_categorical_draw_returns_valid_category() -> None: + mech = CategoricalDraw(["a", "b", "c"], [1.0, 2.0, 1.0]) + results = {mech.sample(_CTX, _rng(i)) for i in range(50)} + assert results <= {"a", "b", "c"} + + +def test_categorical_draw_weights_normalised() -> None: + mech = CategoricalDraw(["x", "y"], [3.0, 1.0]) + assert abs(sum(mech._weights) - 1.0) < 1e-9 + + +def test_categorical_draw_empty_raises() -> None: + with pytest.raises(ValueError, match="empty"): + CategoricalDraw([], []) + + +def test_categorical_draw_mismatched_raises() -> None: + with pytest.raises(ValueError, match="same length"): + CategoricalDraw(["a"], [1.0, 2.0]) + + +def test_categorical_draw_serialise() -> None: + mech = CategoricalDraw(["a", "b"], [1.0, 1.0]) + d = mech.to_dict() + assert d["name"] == "categorical_draw" + assert set(d["categories"]) == {"a", "b"} + + +def test_bounded_numeric_draw_in_range() -> None: + mech = BoundedNumericDraw(0.0, 1.0, 0.5, 0.2) + for i in range(200): + v = mech.sample(_CTX, _rng(i)) + assert 0.0 <= v <= 1.0 + + +def test_bounded_numeric_draw_lo_ge_hi_raises() -> None: + with pytest.raises(ValueError, match="lo"): + BoundedNumericDraw(lo=1.0, hi=0.5) + + +def test_mixture_draw_in_range() -> None: + mech = MixtureDraw([(0.2, 0.1), (0.8, 0.1)], [1.0, 1.0]) + for i in range(200): + v = mech.sample(_CTX, _rng(i)) + assert 0.0 <= v <= 1.0 + + +def test_mixture_draw_serialise_roundtrip() -> None: + mech = MixtureDraw([(0.3, 0.15), (0.7, 0.15)], [2.0, 1.0]) + d = mech.to_dict() + assert d["name"] == "mixture_draw" + assert len(d["components"]) == 2 + assert abs(sum(d["mix_weights"]) - 1.0) < 1e-9 + + +# =========================================================================== +# Influence mechanisms +# =========================================================================== + + +def test_additive_influence_clips_to_unit() -> None: + mech = AdditiveInfluence({"latent_account_fit": 2.0}, bias=0.5) + v = mech.sample(_CTX, _rng()) + assert 0.0 <= v <= 1.0 + + +def test_logistic_influence_in_unit() -> None: + mech = LogisticInfluence({"latent_account_fit": 3.0}, bias=-1.0) + v = mech.sample(_CTX, _rng()) + assert 0.0 < v < 1.0 + + +def test_logistic_influence_zero_temperature_raises() -> None: + with pytest.raises(ValueError, match="temperature"): + LogisticInfluence({}, temperature=0.0) + + +def test_saturating_influence_in_unit() -> None: + mech = SaturatingInfluence({"latent_engagement_propensity": 2.0}) + v = mech.sample(_CTX, _rng()) + assert 0.0 <= v <= 1.0 + + +def test_threshold_influence_binary() -> None: + mech = ThresholdInfluence({"latent_account_fit": 1.0}, threshold=0.5) + v = mech.sample(_CTX, _rng()) + assert v in (0.0, 1.0) + + +def test_interaction_term_clips_to_unit() -> None: + mech = InteractionTerm("latent_account_fit", "latent_contact_authority", weight=2.0) + v = mech.sample(_CTX, _rng()) + assert 0.0 <= v <= 1.0 + + +def test_influence_serialise() -> None: + for mech in [ + AdditiveInfluence({"k": 1.0}), + LogisticInfluence({"k": 1.0}), + SaturatingInfluence({"k": 1.0}), + ThresholdInfluence({"k": 1.0}), + InteractionTerm("a", "b"), + ]: + d = mech.to_dict() + assert "name" in d + json.dumps(d) # must be JSON-serialisable + + +# =========================================================================== +# Latent score +# =========================================================================== + + +def test_latent_score_in_unit() -> None: + mech = LatentScore({"latent_account_fit": 2.0, "latent_sales_friction": -1.0}) + v = mech.sample(_CTX, _rng()) + assert 0.0 < v < 1.0 + + +def test_latent_score_monotone_in_positive_key() -> None: + mech = LatentScore({"latent_account_fit": 2.0}) + low_ctx = MechanismContext(latents={"latent_account_fit": 0.1}) + high_ctx = MechanismContext(latents={"latent_account_fit": 0.9}) + assert mech.score(low_ctx.latents) < mech.score(high_ctx.latents) + + +def test_latent_score_empty_weights_raises() -> None: + with pytest.raises(ValueError, match="empty"): + LatentScore({}) + + +def test_latent_score_missing_key_treated_as_zero() -> None: + mech = LatentScore({"missing_key": 1.0}, bias=0.0) + # With only the missing key contributing 0, score = sigmoid(0) = 0.5 + assert abs(mech.score({}) - 0.5) < 1e-9 + + +# =========================================================================== +# Conversion hazard +# =========================================================================== + + +def test_conversion_hazard_returns_bool() -> None: + score = LatentScore({"latent_account_fit": 2.0}, bias=-1.5) + hazard = ConversionHazard(score, base_rate=0.01, scale=0.05) + result = hazard.sample(_CTX, _rng()) + assert isinstance(result, bool) + + +def test_conversion_hazard_probability_in_range() -> None: + score = LatentScore({"latent_account_fit": 2.0}, bias=-1.5) + hazard = ConversionHazard(score, base_rate=0.01, scale=0.05) + p = hazard.daily_probability(_LATENTS) + assert 0.0 <= p <= hazard._max_daily_rate + + +def test_conversion_hazard_higher_fit_higher_prob() -> None: + score = LatentScore({"latent_account_fit": 3.0}, bias=-1.5) + hazard = ConversionHazard(score, base_rate=0.005, scale=0.08) + low = hazard.daily_probability({"latent_account_fit": 0.1}) + high = hazard.daily_probability({"latent_account_fit": 0.9}) + assert high > low + + +def test_conversion_hazard_invalid_params() -> None: + score = LatentScore({"k": 1.0}) + with pytest.raises(ValueError, match="base_rate"): + ConversionHazard(score, base_rate=1.5) + with pytest.raises(ValueError, match="scale"): + ConversionHazard(score, scale=-0.1) + + +def test_conversion_hazard_serialise() -> None: + score = LatentScore({"latent_account_fit": 1.0}) + hazard = ConversionHazard(score) + d = hazard.to_dict() + assert d["name"] == "conversion_hazard" + json.dumps(d) + + +# =========================================================================== +# Stage sequence + hazard transition +# =========================================================================== + + +def test_stage_sequence_next_stage() -> None: + seq = StageSequence() + assert seq.next_stage("mql") == "sal" + assert seq.next_stage("sal") == "sql" + assert seq.next_stage("closed_won") is None + assert seq.next_stage("closed_lost") is None + assert seq.next_stage("unknown") is None + + +def test_stage_sequence_is_terminal() -> None: + seq = StageSequence() + assert seq.is_terminal("closed_won") + assert seq.is_terminal("closed_lost") + assert not seq.is_terminal("mql") + + +def test_stage_sequence_sample_returns_next() -> None: + seq = StageSequence() + ctx = MechanismContext(stage="sql") + assert seq.sample(ctx, _rng()) == "demo_scheduled" + + +def test_hazard_transition_returns_bool() -> None: + score = LatentScore({"latent_engagement_propensity": 2.0}) + trans = HazardTransition(score, base_rate=0.05, scale=0.15) + ctx = MechanismContext(latents=_LATENTS, extra={"dwell_days": 5}) + assert isinstance(trans.sample(ctx, _rng()), bool) + + +def test_hazard_transition_min_dwell_blocks() -> None: + score = LatentScore({"latent_account_fit": 5.0}) + trans = HazardTransition(score, base_rate=0.99, scale=0.0, min_dwell_days=10) + assert trans.daily_probability(_LATENTS, dwell=3) == 0.0 + + +def test_hazard_transition_invalid_params() -> None: + score = LatentScore({"k": 1.0}) + with pytest.raises(ValueError, match="base_rate"): + HazardTransition(score, base_rate=-0.1) + with pytest.raises(ValueError, match="min_dwell"): + HazardTransition(score, min_dwell_days=-1) + + +def test_hazard_transition_serialise() -> None: + score = LatentScore({"latent_account_fit": 1.0}) + trans = HazardTransition(score) + d = trans.to_dict() + assert d["name"] == "hazard_transition" + json.dumps(d) + + +# =========================================================================== +# Count mechanisms +# =========================================================================== + + +def test_poisson_intensity_non_negative() -> None: + mech = PoissonIntensity(base_rate=0.5, weights={"latent_engagement_propensity": 0.3}) + for i in range(100): + assert mech.sample(_CTX, _rng(i)) >= 0 + + +def test_poisson_intensity_expected_count_positive() -> None: + mech = PoissonIntensity(base_rate=0.4) + assert mech.expected_count(_LATENTS) > 0 + + +def test_poisson_intensity_invalid_rate() -> None: + with pytest.raises(ValueError, match="positive"): + PoissonIntensity(base_rate=0.0) + + +def test_recency_decay_decreases_with_time() -> None: + mech = RecencyDecayIntensity(base_rate=1.0, decay_factor=0.9) + assert mech.expected_count(0) > mech.expected_count(10) > mech.expected_count(50) + + +def test_recency_decay_floor_respected() -> None: + mech = RecencyDecayIntensity(base_rate=1.0, decay_factor=0.5, floor_rate=0.05) + assert mech.expected_count(1000) >= 0.05 + + +def test_recency_decay_invalid_factor() -> None: + with pytest.raises(ValueError, match="decay_factor"): + RecencyDecayIntensity(base_rate=1.0, decay_factor=0.0) + + +# =========================================================================== +# Categorical influence +# =========================================================================== + + +def test_categorical_influence_known_key() -> None: + mech = CategoricalInfluence("channel", CHANNEL_QUALITY_SCORES) + ctx = MechanismContext(extra={"channel": "partner_referral"}) + assert mech.sample(ctx, _rng()) == pytest.approx(0.70) + + +def test_categorical_influence_missing_key_returns_default() -> None: + mech = CategoricalInfluence("channel", CHANNEL_QUALITY_SCORES, default=0.5) + ctx = MechanismContext(extra={}) + assert mech.sample(ctx, _rng()) == pytest.approx(0.5) + + +def test_categorical_influence_unknown_value_returns_default() -> None: + mech = CategoricalInfluence("channel", CHANNEL_QUALITY_SCORES, default=0.5) + ctx = MechanismContext(extra={"channel": "unknown_channel"}) + assert mech.sample(ctx, _rng()) == pytest.approx(0.5) + + +# =========================================================================== +# Measurement mechanisms +# =========================================================================== + + +def test_noisy_proxy_in_unit_or_none() -> None: + mech = NoisyProxy("latent_account_fit", noise_std=0.1, missing_rate=0.1) + results = [mech.sample(_CTX, _rng(i)) for i in range(200)] + non_none = [v for v in results if v is not None] + assert all(0.0 <= v <= 1.0 for v in non_none) + assert any(v is None for v in results) # missingness fires at 10% + + +def test_noisy_proxy_zero_noise_close_to_true() -> None: + mech = NoisyProxy("latent_account_fit", noise_std=0.0, missing_rate=0.0) + v = mech.sample(_CTX, _rng()) + assert v == pytest.approx(_LATENTS["latent_account_fit"]) + + +def test_noisy_proxy_invalid_params() -> None: + with pytest.raises(ValueError, match="noise_std"): + NoisyProxy("k", noise_std=-0.1) + with pytest.raises(ValueError, match="missing_rate"): + NoisyProxy("k", missing_rate=1.5) + + +def test_noisy_categorization_valid_category() -> None: + cats = ["low", "medium", "high"] + mech = NoisyCategorization("tier", cats, confusion_prob=0.1, missing_rate=0.0) + ctx = MechanismContext(extra={"tier": "medium"}) + results = {mech.sample(ctx, _rng(i)) for i in range(100)} + assert results <= set(cats) + + +def test_noisy_categorization_missing_fires() -> None: + mech = NoisyCategorization("tier", ["a", "b"], confusion_prob=0.0, missing_rate=1.0) + assert mech.sample(_CTX, _rng()) is None + + +def test_proxy_compression_correct_band() -> None: + mech = ProxyCompression( + "latent_account_fit", + thresholds=[0.33, 0.67], + labels=["low", "medium", "high"], + ) + low_ctx = MechanismContext(latents={"latent_account_fit": 0.1}) + mid_ctx = MechanismContext(latents={"latent_account_fit": 0.5}) + high_ctx = MechanismContext(latents={"latent_account_fit": 0.9}) + assert mech.sample(low_ctx, _rng()) == "low" + assert mech.sample(mid_ctx, _rng()) == "medium" + assert mech.sample(high_ctx, _rng()) == "high" + + +def test_proxy_compression_bad_labels_count() -> None: + with pytest.raises(ValueError, match="labels"): + ProxyCompression("k", thresholds=[0.5], labels=["a"]) + + +def test_proxy_compression_unsorted_thresholds() -> None: + with pytest.raises(ValueError, match="increasing"): + ProxyCompression("k", thresholds=[0.7, 0.3], labels=["a", "b", "c"]) + + +# =========================================================================== +# Policies / MechanismAssignment +# =========================================================================== + + +@pytest.mark.parametrize("motif", MOTIF_FAMILY_NAMES) +def test_assign_mechanisms_returns_assignment(motif: str) -> None: + assignment = assign_mechanisms(motif, _rng()) + assert isinstance(assignment, MechanismAssignment) + assert assignment.motif_family == motif + + +@pytest.mark.parametrize("motif", MOTIF_FAMILY_NAMES) +def test_assignment_mechanisms_are_callable(motif: str) -> None: + assignment = assign_mechanisms(motif, _rng()) + ctx = MechanismContext(latents=_LATENTS, stage="mql", t=3, extra={"dwell_days": 3}) + assert isinstance(assignment.conversion_hazard.sample(ctx, _rng()), bool) + assert isinstance(assignment.stage_transition.sample(ctx, _rng()), bool) + assert isinstance(assignment.touch_intensity.sample(ctx, _rng()), int) + proxy = assignment.measurement.sample(ctx, _rng()) + assert proxy is None or 0.0 <= proxy <= 1.0 + + +@pytest.mark.parametrize("motif", MOTIF_FAMILY_NAMES) +def test_assignment_summary_serialisable(motif: str) -> None: + assignment = assign_mechanisms(motif, _rng()) + summary = assignment.summary() + assert isinstance(summary, MechanismSummary) + d = summary.to_dict() + json.dumps(d) # must not raise + assert d["motif_family"] == motif + + +@pytest.mark.parametrize("motif", MOTIF_FAMILY_NAMES) +def test_assignment_summary_roundtrip(motif: str) -> None: + assignment = assign_mechanisms(motif, _rng()) + summary = assignment.summary() + restored = MechanismSummary.from_dict(summary.to_dict()) + assert restored.motif_family == motif + assert restored.conversion_hazard == summary.conversion_hazard + + +def test_assign_mechanisms_deterministic() -> None: + a1 = assign_mechanisms("fit_dominant", random.Random(7)) # noqa: S311 + a2 = assign_mechanisms("fit_dominant", random.Random(7)) # noqa: S311 + assert a1.summary().to_dict() == a2.summary().to_dict() + + +def test_assign_unknown_motif_falls_back_gracefully() -> None: + assignment = assign_mechanisms("nonexistent_motif", _rng()) + assert assignment.motif_family == "nonexistent_motif" + ctx = MechanismContext(latents=_LATENTS, extra={"dwell_days": 5}) + assert isinstance(assignment.conversion_hazard.sample(ctx, _rng()), bool) + + +def test_mechanism_params_for_motif_contains_expected_keys() -> None: + params = mechanism_params_for_motif("fit_dominant") + assert "conversion_score_weights" in params + assert "hazard_params" in params + assert "transition_score_weights" in params + assert "touch_base_rate" in params + + +# =========================================================================== +# Fit-dominant vs buying-committee-friction hazard ordering +# =========================================================================== + + +def test_fit_dominant_higher_conversion_rate_than_friction() -> None: + """Across a range of high-fit latent states, fit_dominant worlds should + have higher daily conversion probability than buying_committee_friction.""" + high_fit_latents = dict(_LATENTS) + high_fit_latents["latent_account_fit"] = 0.9 + high_fit_latents["latent_sales_friction"] = 0.2 + + fit_p = [] + fric_p = [] + for seed in range(20): + fit_asgn = assign_mechanisms("fit_dominant", random.Random(seed)) # noqa: S311 + fric_asgn = assign_mechanisms("buying_committee_friction", random.Random(seed)) # noqa: S311 + fit_p.append(fit_asgn.conversion_hazard.daily_probability(high_fit_latents)) + fric_p.append(fric_asgn.conversion_hazard.daily_probability(high_fit_latents)) + + assert sum(fit_p) / len(fit_p) > sum(fric_p) / len(fric_p) From 5bdc40da2f5e3f77a06bbc1fc6981232f70d1357 Mon Sep 17 00:00:00 2001 From: Shay Palachy Date: Tue, 28 Apr 2026 09:17:31 +0300 Subject: [PATCH 2/2] fix: address Copilot round-1 review on PR #11 - MixtureDraw: guard against negative mix_weights (COPILOT-1) - ProxyCompression: strict adjacent-pair increase + (0,1) range checks (COPILOT-2) - LatentScore.score: numerically stable two-branch sigmoid (COPILOT-3) - influence._sigmoid: numerically stable two-branch sigmoid (COPILOT-4) - Tests: overflow, duplicate/boundary threshold, negative weight cases Co-Authored-By: Claude Sonnet 4.6 --- leadforge/mechanisms/influence.py | 6 +++- leadforge/mechanisms/measurement.py | 7 ++-- leadforge/mechanisms/scores.py | 5 ++- leadforge/mechanisms/static.py | 2 ++ tests/mechanisms/test_mechanisms.py | 53 +++++++++++++++++++++++++++++ 5 files changed, 69 insertions(+), 4 deletions(-) diff --git a/leadforge/mechanisms/influence.py b/leadforge/mechanisms/influence.py index 50443ed..e9dd54f 100644 --- a/leadforge/mechanisms/influence.py +++ b/leadforge/mechanisms/influence.py @@ -18,7 +18,11 @@ def _weighted_sum(latents: dict[str, float], weights: dict[str, float], bias: fl def _sigmoid(x: float) -> float: - return 1.0 / (1.0 + math.exp(-x)) + """Numerically stable sigmoid that avoids overflow for large |x|.""" + if x >= 0: + return 1.0 / (1.0 + math.exp(-x)) + ex = math.exp(x) + return ex / (1.0 + ex) class AdditiveInfluence(Mechanism): diff --git a/leadforge/mechanisms/measurement.py b/leadforge/mechanisms/measurement.py index d7b9003..55b16e0 100644 --- a/leadforge/mechanisms/measurement.py +++ b/leadforge/mechanisms/measurement.py @@ -142,8 +142,11 @@ def __init__( f"Expected {len(thresholds) + 1} labels for {len(thresholds)} thresholds, " f"got {len(labels)}" ) - if thresholds != sorted(thresholds): - raise ValueError("thresholds must be strictly increasing") + for i in range(len(thresholds) - 1): + if thresholds[i] >= thresholds[i + 1]: + raise ValueError("thresholds must be strictly increasing") + if any(not (0.0 < t < 1.0) for t in thresholds): + raise ValueError("all thresholds must be in (0, 1)") if not (0.0 <= missing_rate <= 1.0): raise ValueError(f"missing_rate must be in [0, 1], got {missing_rate}") self._key = latent_key diff --git a/leadforge/mechanisms/scores.py b/leadforge/mechanisms/scores.py index 143dc68..8d5ad48 100644 --- a/leadforge/mechanisms/scores.py +++ b/leadforge/mechanisms/scores.py @@ -46,7 +46,10 @@ def score(self, latents: dict[str, float]) -> float: linear = self._bias + sum( self._weights.get(k, 0.0) * latents.get(k, 0.0) for k in self._weights ) - return 1.0 / (1.0 + math.exp(-linear)) + if linear >= 0: + return 1.0 / (1.0 + math.exp(-linear)) + ex = math.exp(linear) + return ex / (1.0 + ex) def sample(self, context: MechanismContext, rng: random.Random) -> float: return self.score(context.latents) diff --git a/leadforge/mechanisms/static.py b/leadforge/mechanisms/static.py index db1230a..6c1817f 100644 --- a/leadforge/mechanisms/static.py +++ b/leadforge/mechanisms/static.py @@ -116,6 +116,8 @@ def __init__( raise ValueError("components must not be empty") if len(components) != len(mix_weights): raise ValueError("components and mix_weights must have the same length") + if any(w < 0 for w in mix_weights): + raise ValueError("all mix_weights must be non-negative") total = sum(mix_weights) if total <= 0: raise ValueError("mix_weights must sum to a positive value") diff --git a/tests/mechanisms/test_mechanisms.py b/tests/mechanisms/test_mechanisms.py index a9b1897..8acf753 100644 --- a/tests/mechanisms/test_mechanisms.py +++ b/tests/mechanisms/test_mechanisms.py @@ -115,6 +115,11 @@ def test_mixture_draw_serialise_roundtrip() -> None: assert abs(sum(d["mix_weights"]) - 1.0) < 1e-9 +def test_mixture_draw_negative_weight_raises() -> None: + with pytest.raises(ValueError, match="non-negative"): + MixtureDraw([(0.3, 0.1), (0.7, 0.1)], [1.0, -0.5]) + + # =========================================================================== # Influence mechanisms # =========================================================================== @@ -418,6 +423,54 @@ def test_proxy_compression_unsorted_thresholds() -> None: ProxyCompression("k", thresholds=[0.7, 0.3], labels=["a", "b", "c"]) +def test_proxy_compression_duplicate_thresholds_raises() -> None: + with pytest.raises(ValueError, match="increasing"): + ProxyCompression("k", thresholds=[0.5, 0.5], labels=["a", "b", "c"]) + + +def test_proxy_compression_threshold_at_zero_raises() -> None: + with pytest.raises(ValueError, match=r"\(0, 1\)"): + ProxyCompression("k", thresholds=[0.0, 0.5], labels=["a", "b", "c"]) + + +def test_proxy_compression_threshold_at_one_raises() -> None: + with pytest.raises(ValueError, match=r"\(0, 1\)"): + ProxyCompression("k", thresholds=[0.5, 1.0], labels=["a", "b", "c"]) + + +# =========================================================================== +# Numeric stability — sigmoid / score +# =========================================================================== + + +def test_latent_score_extreme_positive_no_overflow() -> None: + mech = LatentScore({"k": 1.0}, bias=0.0) + ctx = MechanismContext(latents={"k": 1_000.0}) + v = mech.sample(ctx, _rng()) + assert v == pytest.approx(1.0, abs=1e-6) + + +def test_latent_score_extreme_negative_no_overflow() -> None: + mech = LatentScore({"k": 1.0}, bias=0.0) + ctx = MechanismContext(latents={"k": -1_000.0}) + v = mech.sample(ctx, _rng()) + assert v == pytest.approx(0.0, abs=1e-6) + + +def test_logistic_influence_extreme_positive_no_overflow() -> None: + mech = LogisticInfluence({"k": 1.0}, bias=0.0) + ctx = MechanismContext(latents={"k": 1_000.0}) + v = mech.sample(ctx, _rng()) + assert v == pytest.approx(1.0, abs=1e-6) + + +def test_logistic_influence_extreme_negative_no_overflow() -> None: + mech = LogisticInfluence({"k": 1.0}, bias=0.0) + ctx = MechanismContext(latents={"k": -1_000.0}) + v = mech.sample(ctx, _rng()) + assert v == pytest.approx(0.0, abs=1e-6) + + # =========================================================================== # Policies / MechanismAssignment # ===========================================================================