AGG-131: score feature rows with HH/CG LightGBM boosters

README.md

@@ -33,6 +33,16 @@ replica for this (iin, phone) pair» — callers must handle it, not treat it as
 an error. The API never returns `[]` for a not-found lookup (that would be
 ambiguous with an iin-only lookup that happens to have zero rows).
 
+When HH/CG LightGBM boosters are configured (see [Scoring](#scoring) below),
+each row is enriched in place with two extra float fields:
+
+| field      | meaning                                                         |
+| ---------- | --------------------------------------------------------------- |
+| `HH_score` | raw `booster.predict()` output of the HH model (PD, `0..1`)     |
+| `CG_score` | raw `booster.predict()` output of the CG model (PD, `0..1`)     |
+
+If a model is not configured, its `*_score` field is omitted from the row.
+
 Status codes:
 
 | code | meaning                                                                      |
@@ -110,6 +120,8 @@ All settings are env vars with the `KZ_SCORING_` prefix; defaults are in
 | `KZ_SCORING_POLL_INTERVAL_MS`                    | `100`                                                                                    |
 | `KZ_SCORING_MAX_CONCURRENT_LOOKUPS`              | `10`                                                                                     |
 | `KZ_SCORING_SALT_CACHE_TTL_SECONDS`              | `300`                                                                                    |
+| `KZ_SCORING_HH_MODEL_PATH`                       | `""` (empty → HH scoring disabled; file path to a LightGBM booster txt otherwise)        |
+| `KZ_SCORING_CG_MODEL_PATH`                       | `""` (empty → CG scoring disabled; file path to a LightGBM booster txt otherwise)        |
 | `KZ_SCORING_LOG_LEVEL`                           | `INFO`                                                                                   |
 | `KZ_SCORING_HOST` / `KZ_SCORING_PORT`            | `0.0.0.0` / `8000`                                                                       |
 
@@ -156,6 +168,29 @@ The Argo Application that wires the chart into the Beeline cluster lives in
 `argo/overlays/vaultee/` and is set up separately by DevOps once the chart
 is green.
 
+## Scoring
+
+The service can score each feature row with two LightGBM boosters (HH and CG,
+`hh_beeline_bezzalog_2026_05_04_model` / `cg_beeline_bezzalog_2026_05_04_model`
+delivered by the modeling team on `aggregion/kz-scoring` issue AGG-131).
+
+- Boosters are loaded once at process startup from
+  `KZ_SCORING_HH_MODEL_PATH` and `KZ_SCORING_CG_MODEL_PATH`.
+- Feature vectors are projected onto `booster.feature_name()` in order; row keys
+  missing from the response become `NaN` (LightGBM handles them as native
+  missing values at tree splits).
+- `booster.predict(...)` output — raw PD in `[0, 1]` — is written back on the
+  row as `HH_score` / `CG_score`.
+- Leaving a path empty disables that model; scoring never blocks the lookup —
+  if the model file is missing at startup, the process fails fast with a
+  `FileNotFoundError`.
+- Startup logs record `scoring: loaded hh model from … (N features)` so a bad
+  model file is easy to catch in the deployment logs.
+
+In-cluster, mount the model files via the chart's `models.volume` (any k8s
+volume type — PVC, ConfigMap, Secret) and point
+`config.hh_model_path` / `config.cg_model_path` at files under its `mountPath`.
+
 ## Not in scope
 
 - AuthN / TLS — handled by external ingress.

chart/templates/configmap.yaml

@@ -22,4 +22,6 @@ data:
   KZ_SCORING_POLL_INTERVAL_MS: {{ .Values.config.poll_interval_ms | quote }}
   KZ_SCORING_MAX_CONCURRENT_LOOKUPS: {{ .Values.config.max_concurrent_lookups | quote }}
   KZ_SCORING_SALT_CACHE_TTL_SECONDS: {{ .Values.config.salt_cache_ttl_seconds | quote }}
+  KZ_SCORING_HH_MODEL_PATH: {{ .Values.config.hh_model_path | quote }}
+  KZ_SCORING_CG_MODEL_PATH: {{ .Values.config.cg_model_path | quote }}
   KZ_SCORING_LOG_LEVEL: {{ .Values.config.log_level | quote }}

chart/templates/deployment.yaml

@@ -60,6 +60,18 @@ spec:
           {{- end }}
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
+          {{- if .Values.models.volume.enabled }}
+          volumeMounts:
+            - name: models
+              mountPath: {{ .Values.models.volume.mountPath | quote }}
+              readOnly: {{ .Values.models.volume.readOnly }}
+          {{- end }}
+      {{- if .Values.models.volume.enabled }}
+      volumes:
+        - name: models
+          {{- $vol := omit .Values.models.volume "enabled" "mountPath" "readOnly" }}
+          {{- toYaml $vol | nindent 10 }}
+      {{- end }}
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}

chart/values.yaml

@@ -90,5 +90,40 @@ config:
   max_concurrent_lookups: 10
   salt_cache_ttl_seconds: 300
   log_level: "INFO"
+  # Paths inside the container where LightGBM booster txt files are mounted.
+  # Leave empty to disable scoring (endpoints return raw feature rows only).
+  # If you enable ``models.volume`` below, defaults line up with its mountPath.
+  hh_model_path: ""
+  cg_model_path: ""
+
+# Optional volume that materialises HH/CG LightGBM model files inside the pod.
+# The volume is mounted at ``mountPath`` on the app container; put your
+# ``config.hh_model_path`` / ``config.cg_model_path`` values under that path.
+#
+# Example — mount from an existing PVC that DevOps populates out of band:
+#   models:
+#     volume:
+#       enabled: true
+#       mountPath: /models
+#       persistentVolumeClaim:
+#         claimName: kz-scoring-models
+#   config:
+#     hh_model_path: /models/hh_beeline_bezzalog_2026_05_04_model.txt
+#     cg_model_path: /models/cg_beeline_bezzalog_2026_05_04_model.txt
+#
+# Example — mount from a ConfigMap or Secret; use the same keys the k8s
+# volume type accepts (``configMap``, ``secret``, ``persistentVolumeClaim``,
+# ``hostPath``, ``emptyDir``, …). Exactly one source key should be set.
+models:
+  volume:
+    enabled: false
+    mountPath: /models
+    readOnly: true
+    # persistentVolumeClaim:
+    #   claimName: kz-scoring-models
+    # configMap:
+    #   name: kz-scoring-models
+    # secret:
+    #   secretName: kz-scoring-models
 
 extraEnv: []

pyproject.toml

@@ -9,6 +9,8 @@ dependencies = [
   "httpx==0.27.2",
   "pydantic==2.9.2",
   "pydantic-settings==2.6.1",
+  "lightgbm==4.5.0",
+  "numpy==2.1.3",
 ]
 
 [project.optional-dependencies]

requirements.txt

@@ -3,3 +3,5 @@ uvicorn[standard]==0.32.1
 httpx==0.27.2
 pydantic==2.9.2
 pydantic-settings==2.6.1
+lightgbm==4.5.0
+numpy==2.1.3

src/kz_scoring_api/app.py

@@ -18,6 +18,7 @@
     PipelineUnavailableError,
     VaulteePipelinesClient,
 )
+from .scoring import ScoringService
 from .secrets import VaulteeSecretsClient
 
 logger = logging.getLogger(__name__)
@@ -98,7 +99,12 @@ async def lifespan(app: FastAPI):
             ttl_seconds=settings.salt_cache_ttl_seconds,
             http=http,
         )
-        app.state.lookup = LookupService(settings, pipelines, secrets)
+        scoring = ScoringService.from_paths(
+            hh_model_path=settings.hh_model_path,
+            cg_model_path=settings.cg_model_path,
+        )
+        app.state.lookup = LookupService(settings, pipelines, secrets, scoring=scoring)
+        app.state.scoring = scoring
         app.state.http = http
         try:
             yield

src/kz_scoring_api/config.py

@@ -59,6 +59,9 @@ def _tenant_id_not_empty(cls, v: str) -> str:
 
     salt_cache_ttl_seconds: float = Field(default=300.0)
 
+    hh_model_path: str = Field(default="")
+    cg_model_path: str = Field(default="")
+
     # Static shared-secret gate for /single and /multi. Empty = disabled
     # (matches original unauthenticated behaviour). When set, callers must
     # pass the same value in the X-API-Key request header. /healthz always

src/kz_scoring_api/lookup.py

@@ -11,6 +11,7 @@
     PipelineUnavailableError,
     VaulteePipelinesClient,
 )
+from .scoring import ScoringService
 from .secrets import VaulteeSecretsClient
 from .tsv import parse_tsv
 
@@ -30,10 +31,12 @@ def __init__(
         settings: Settings,
         pipelines: VaulteePipelinesClient,
         secrets: VaulteeSecretsClient,
+        scoring: ScoringService | None = None,
     ) -> None:
         self._settings = settings
         self._pipelines = pipelines
         self._secrets = secrets
+        self._scoring = scoring
         self._sem = asyncio.Semaphore(max(1, settings.max_concurrent_lookups))
 
     async def _salt_pkb(self) -> bytes:
@@ -94,6 +97,8 @@ async def _run_one(self, iin: str, phone: str | None) -> LookupResult:
         )
         payload = await self._pipelines.fetch_result(run_id)
         rows = parse_tsv(payload)
+        if self._scoring is not None:
+            self._scoring.score_rows(rows)
         return self._shape(rows, has_phone=phone is not None)
 
     @staticmethod

src/kz_scoring_api/scoring.py

@@ -0,0 +1,155 @@
+from __future__ import annotations
+
+import logging
+import math
+from pathlib import Path
+from typing import Any, Protocol
+
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+_MISSING = float("nan")
+
+_MISSING_TOKENS = frozenset({"", "null", "none", "nan", "na", "n/a"})
+
+
+def _to_float(value: Any) -> float:
+    """Coerce a heterogeneous cell value to float, mapping missing/garbage to NaN.
+
+    LightGBM handles NaN as an explicit missing value at tree splits, so
+    unparseable / blank cells propagate as NaN into ``predict`` rather than
+    getting silently zeroed.
+    """
+    if value is None:
+        return _MISSING
+    if isinstance(value, bool):
+        return float(value)
+    if isinstance(value, int | float):
+        f = float(value)
+        return f if not math.isnan(f) else _MISSING
+    if isinstance(value, str):
+        s = value.strip()
+        if s.lower() in _MISSING_TOKENS:
+            return _MISSING
+        try:
+            return float(s)
+        except ValueError:
+            return _MISSING
+    return _MISSING
+
+
+class SupportsPredict(Protocol):
+    def predict(self, data: Any) -> Any: ...
+    def feature_name(self) -> list[str]: ...
+
+
+class BoosterScorer:
+    """Wraps a LightGBM booster and scores a single feature-row dict.
+
+    The row dict is projected onto ``booster.feature_name()`` in-order; missing
+    keys become NaN. Extra keys in the row are ignored.
+    """
+
+    def __init__(self, booster: SupportsPredict, name: str) -> None:
+        self._booster = booster
+        self._name = name
+        self._feature_names: list[str] = list(booster.feature_name())
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def feature_names(self) -> list[str]:
+        return self._feature_names
+
+    def score(self, row: dict[str, Any]) -> float:
+        vec = [_to_float(row.get(feat)) for feat in self._feature_names]
+        arr = np.asarray([vec], dtype=np.float64)
+        preds = self._booster.predict(arr)
+        return float(np.asarray(preds).ravel()[0])
+
+
+class ScoringService:
+    """Attaches HH_score / CG_score to each feature-row.
+
+    The two LightGBM boosters are loaded once at process start (see
+    :meth:`from_paths`). Scoring is a plain in-place mutation of the row
+    dicts so it composes with the existing TSV → JSON pipeline without
+    changing the response schema shape.
+
+    Either model can be omitted by leaving its path empty; the corresponding
+    ``*_score`` field is then omitted from the response (the service falls
+    back to a pass-through).
+    """
+
+    HH_KEY = "HH_score"
+    CG_KEY = "CG_score"
+
+    def __init__(
+        self,
+        hh_scorer: BoosterScorer | None = None,
+        cg_scorer: BoosterScorer | None = None,
+    ) -> None:
+        self._hh = hh_scorer
+        self._cg = cg_scorer
+
+    @classmethod
+    def from_paths(
+        cls,
+        hh_model_path: str | None,
+        cg_model_path: str | None,
+    ) -> ScoringService:
+        return cls(
+            hh_scorer=cls._load(hh_model_path, "hh"),
+            cg_scorer=cls._load(cg_model_path, "cg"),
+        )
+
+    @staticmethod
+    def _load(model_path: str | None, name: str) -> BoosterScorer | None:
+        if not model_path:
+            logger.info(
+                "scoring: %s model path is not configured; %s_score will be omitted",
+                name,
+                name,
+            )
+            return None
+        path = Path(model_path)
+        if not path.exists():
+            raise FileNotFoundError(
+                f"scoring: {name} model file not found at {model_path}"
+            )
+        import lightgbm as lgb
+
+        booster = lgb.Booster(model_file=str(path))
+        scorer = BoosterScorer(booster, name)
+        logger.info(
+            "scoring: loaded %s model from %s (%d features)",
+            name,
+            model_path,
+            len(scorer.feature_names),
+        )
+        return scorer
+
+    @property
+    def enabled(self) -> bool:
+        return self._hh is not None or self._cg is not None
+
+    @property
+    def hh(self) -> BoosterScorer | None:
+        return self._hh
+
+    @property
+    def cg(self) -> BoosterScorer | None:
+        return self._cg
+
+    def score_rows(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        if not rows or not self.enabled:
+            return rows
+        for row in rows:
+            if self._hh is not None:
+                row[self.HH_KEY] = self._hh.score(row)
+            if self._cg is not None:
+                row[self.CG_KEY] = self._cg.score(row)
+        return rows