feat(stats,probability): bias/EV assessment + coverage map (remix pt.1)

backend/src/backend/app/services/ev_service.py

@@ -0,0 +1,108 @@
+"""Expected-value analysis for generated combinations (EV-15).
+
+Lever A from the number-generation-remix design: surface the *payout-if-win*
+expectation of a ticket so the UI can be honest about odds. The app models a
+single jackpot tier (Draw.jackpot) with parimutuel split by winners, so the only
+data-driven lever is the current jackpot magnitude relative to ticket cost and
+historical mean winners. Without per-combination sales data (lever B is
+out-of-scope), every combination has the same expected value; `combination_ev`
+still accepts an `expected_winners` so popularity-differentiated ranking can be
+added later without an API change.
+"""
+
+from __future__ import annotations
+
+import math
+from decimal import ROUND_HALF_UP, Decimal
+
+from sqlalchemy.orm import Session
+
+from backend.app.repositories.lottery_repository import LotteryRepository
+from backend.app.repositories.stat_payload_repository import StatPayloadRepository
+from backend.app.services.errors import NotFoundError
+
+
+class EVService:
+    """Compute expected value of a ticket / combination for a lottery."""
+
+    def __init__(self, session: Session) -> None:
+        self._session = session
+        self._lotteries = LotteryRepository(session)
+        self._payloads = StatPayloadRepository(session)
+
+    def _resolve_lottery(self, *, lottery_code: str | None, lottery_id: int | None):
+        """Resolve the lottery from ``code`` or ``id``; 404-style when absent."""
+        lottery = None
+        if lottery_code is not None:
+            lottery = self._lotteries.get_by_code(lottery_code)
+        elif lottery_id is not None:
+            lottery = self._lotteries.get(lottery_id)
+        if lottery is None:
+            raise NotFoundError("lottery does not exist")
+        return lottery
+
+    def combinations_count(
+        self, *, lottery_code: str | None = None, lottery_id: int | None = None
+    ) -> int:
+        """Number of possible combinations C(universe, numbers_to_select)."""
+        lottery = self._resolve_lottery(lottery_code=lottery_code, lottery_id=lottery_id)
+        universe = lottery.max_number - lottery.min_number + 1
+        return math.comb(universe, lottery.numbers_to_select)
+
+    def _latest_jackpot_and_avg_winners(self, lottery_id: int) -> tuple[Decimal, Decimal]:
+        jackpots: list[Decimal] = []
+        winners: list[Decimal] = []
+        for _draw_number, _numbers, jackpot, winner in self._payloads.iter_draws(lottery_id):
+            if jackpot is not None:
+                jackpots.append(Decimal(str(jackpot)))
+            if winner is not None:
+                winners.append(Decimal(str(winner)))
+        latest = jackpots[-1] if jackpots else Decimal(0)
+        avg = (sum(winners) / Decimal(len(winners))) if winners else Decimal(1)
+        return latest, avg
+
+    @staticmethod
+    def combination_ev(
+        combo,
+        jackpot: Decimal | float | str,
+        ticket_cost: Decimal | float | str,
+        combinations: int,
+        expected_winners: Decimal | float | str = 1,
+    ) -> Decimal:
+        """Expected value of playing one combination.
+
+        ``EV = (jackpot / expected_winners) / combinations - ticket_cost``.
+        The ``combo`` argument is accepted for API symmetry (per-combination
+        winner estimates can differentiate EV later) but is not yet used.
+        """
+        jackpot_d = Decimal(str(jackpot))
+        cost = Decimal(str(ticket_cost))
+        combos = Decimal(combinations)
+        winners = Decimal(str(expected_winners)) if expected_winners else Decimal(1)
+        expected_share = jackpot_d / winners
+        ev = expected_share / combos - cost
+        return ev.quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
+
+    def estimate_ticket_ev(
+        self,
+        ticket_cost: Decimal | float | str,
+        *,
+        lottery_code: str | None = None,
+        lottery_id: int | None = None,
+    ) -> Decimal:
+        """EV of a single ticket using the latest jackpot and historical mean winners."""
+        lottery = self._resolve_lottery(lottery_code=lottery_code, lottery_id=lottery_id)
+        jackpot, avg_winners = self._latest_jackpot_and_avg_winners(lottery.id)
+        combos = self.combinations_count(lottery_id=lottery.id)
+        return self.combination_ev(None, jackpot, ticket_cost, combos, avg_winners)
+
+    def is_high_ev_window(
+        self,
+        ticket_cost: Decimal | float | str,
+        *,
+        lottery_code: str | None = None,
+        lottery_id: int | None = None,
+    ) -> bool:
+        """True when a ticket's EV is positive (rare; only on large rollovers)."""
+        ev = self.estimate_ticket_ev(ticket_cost, lottery_code=lottery_code, lottery_id=lottery_id)
+        return ev > 0

backend/src/backend/app/services/probability_service.py

@@ -49,6 +49,7 @@
     SnapshotNotFoundError,
     ValidationError,
 )
+from backend.app.statistics.engine import frequency
 
 # Supported model bundles and scopes (mirrors F3/F4).
 PROB_MODEL_SET_CORE: str = "core"
@@ -216,6 +217,50 @@ def get_active(
             )
         return snapshot
 
+    # --- coverage map (PM-08) -------------------------------------------------
+
+    def coverage_map(
+        self,
+        *,
+        lottery_code: str | None = None,
+        lottery_id: int | None = None,
+        z_threshold: float = 1.5,
+    ) -> dict[int, str]:
+        """Classify each number COLD/NORMAL/HOT from historical draw frequency.
+
+        Uses the empirical appearance count vs the binomial expectation
+        (``numbers_to_select / universe`` per draw). Cold numbers are
+        under-represented and eligible for a coverage boost by the generator.
+        """
+        lottery = self._resolve_lottery(lottery_code=lottery_code, lottery_id=lottery_id)
+        draws = [d.numbers for d in self._draw_reader.iter_draws(lottery.id)]
+        counts = frequency(draws)
+        return _classify_coverage(
+            counts,
+            lottery.min_number,
+            lottery.max_number,
+            lottery.numbers_to_select,
+            z_threshold,
+        )
+
+    def cold_boost_weights(
+        self,
+        *,
+        lottery_code: str | None = None,
+        lottery_id: int | None = None,
+        z_threshold: float = 1.5,
+        boost: Decimal | float | str = "1.5",
+    ) -> dict[int, Decimal]:
+        """Weights for the generator: cold numbers get ``boost``, others 1.0 (PM-08)."""
+        coverage = self.coverage_map(
+            lottery_code=lottery_code, lottery_id=lottery_id, z_threshold=z_threshold
+        )
+        boost_d = Decimal(str(boost))
+        return {
+            number: (boost_d if status == "cold" else Decimal(1))
+            for number, status in coverage.items()
+        }
+
     def read_values(
         self,
         *,
@@ -536,6 +581,42 @@ def _checksum(rows: Iterable[ProbValue]) -> str:
     return hashlib.sha256(canonical).hexdigest()
 
 
+def _classify_coverage(
+    counts: dict[int, int],
+    min_number: int,
+    max_number: int,
+    numbers_to_select: int,
+    z_threshold: float = 1.5,
+) -> dict[int, str]:
+    """Classify each number COLD/NORMAL/HOT from empirical vs binomial expectation.
+
+    A number appearing far less than ``total_draws * p`` (with ``p =
+    numbers_to_select / universe``) is COLD; far more is HOT; otherwise NORMAL.
+    Classification is a pure, deterministic function of the draw counts.
+    """
+    universe = max_number - min_number + 1
+    total_draws = sum(counts.values()) // numbers_to_select if counts else 0
+    if total_draws == 0 or universe == 0:
+        return {n: "normal" for n in range(min_number, max_number + 1)}
+    expected = total_draws * numbers_to_select / universe
+    p = numbers_to_select / universe
+    std = (expected * (1 - p)) ** 0.5
+    result: dict[int, str] = {}
+    for number in range(min_number, max_number + 1):
+        if std == 0:
+            result[number] = "normal"
+            continue
+        observed = counts.get(number, 0)
+        z = (observed - expected) / std
+        if z < -z_threshold:
+            result[number] = "cold"
+        elif z > z_threshold:
+            result[number] = "hot"
+        else:
+            result[number] = "normal"
+    return result
+
+
 __all__ = [
     "PROB_MODEL_SET_CORE",
     "SCOPE_FULL",

backend/src/backend/app/services/statistics_service.py

@@ -42,11 +42,15 @@
 )
 from backend.app.statistics.checksum import stat_checksum
 from backend.app.statistics.engine import (
+    BiasReport,
     entropy_base2,
     frequency,
     null_aware_average,
     positional_frequency,
 )
+from backend.app.statistics.engine import (
+    bias_report as engine_bias_report,
+)
 from backend.app.statistics.engine import (
     gaps as engine_gaps,
 )
@@ -218,6 +222,21 @@ def read_scalars(
 
     # --- resolution / validation ---------------------------------------------
 
+    def bias_report(
+        self, *, lottery_code: str | None = None, lottery_id: int | None = None
+    ) -> BiasReport:
+        """Return a fairness/bias diagnostic over the lottery's draw history (STE-14).
+
+        Recomputes frequencies and the per-draw sequence from stored draws and
+        returns a `BiasReport` (chi-square, runs z, outliers, fair/anomalous).
+        """
+        lottery = self._resolve_lottery(lottery_code=lottery_code, lottery_id=lottery_id)
+        draws: list[list[int]] = [
+            numbers for _, numbers, _, _ in self._payloads.iter_draws(lottery.id)
+        ]
+        counts = frequency(draws)
+        return engine_bias_report(counts, draws, lottery.min_number, lottery.max_number)
+
     def _persist_new(self, lottery, metric_set: str, payload: dict) -> StatSnapshot:
         """Atomically write a NEW version and its payload, retiring the old active.