t9 telegram: single-id allowlist delivery — notify-once via claim-send-commit, thin bot api client, lifespan-owned poller

.env.example

@@ -20,3 +20,11 @@ MASSIVE_RATE_PER_MIN=5
 ANTHROPIC_API_KEY=your-anthropic-key
 # One short call per anomaly; the cheap/fast tier is plenty.
 ANTHROPIC_MODEL=claude-haiku-4-5-20251001
+
+# --- Telegram (delivery) — token from @BotFather ----------------------------
+TELEGRAM_BOT_TOKEN=your-bot-token
+TELEGRAM_BASE_URL=https://api.telegram.org
+# Your numeric user id (message @userinfobot to find it). This single id is
+# both the alert destination and the allowlist — everyone else is ignored.
+TELEGRAM_CHAT_ID=123456789
+TELEGRAM_RATE_PER_MIN=30

packages/recommend/src/hodlin_recommend/config.py

@@ -33,3 +33,10 @@ class Settings(BaseSettings):
     # Anthropic — the single explanation call per anomaly (T7).
     anthropic_api_key: str
     anthropic_model: str
+
+    # Telegram — delivery (T9). ``chat_id`` is both the destination and the
+    # single-ID allowlist: inbound messages from anyone else are dropped.
+    telegram_bot_token: str
+    telegram_base_url: str
+    telegram_chat_id: int
+    telegram_rate_per_min: float

packages/recommend/src/hodlin_recommend/connectors/base.py

@@ -116,18 +116,39 @@ async def request_json(
     source: str,
     url: str,
     params: Mapping[str, str] | None = None,
+    method: str = "GET",
+    json_body: Any | None = None,
+    http_timeout: httpx.Timeout | None = None,
     rate: RateLimiter | None = None,
     retry: RetryPolicy = DEFAULT_RETRY,
+    secrets: tuple[str, ...] = (),
 ) -> Any:
-    """GET ``url`` and return parsed JSON, applying the shared rate-limit + retry
-    policy and wrapping any HTTP failure as ``SourceUnavailable``."""
+    """Call ``url`` and return parsed JSON, applying the shared rate-limit +
+    retry policy and wrapping any HTTP failure as ``SourceUnavailable``.
+
+    ``method``/``json_body`` exist for POST-style APIs (Telegram);
+    ``http_timeout`` overrides the client default per request (a long poll
+    must outlive it). Note retries make non-idempotent POSTs at-least-once:
+    a reply lost on the wire is retried even if the server acted — callers
+    choose semantics.
+
+    ``secrets`` are redacted from the wrapped error message: httpx errors
+    quote the full URL, which carries API keys (query params) or the bot
+    token (path) — and ``SourceUnavailable`` text ends up *persisted* in
+    ``ingest_runs.detail``, so the secret must be scrubbed at the wrap.
+    """
+    request_timeout = http_timeout if http_timeout is not None else httpx.USE_CLIENT_DEFAULT
 
     async def _once() -> Any:
         if rate is not None:
             async with rate:
-                response = await client.get(url, params=params)
+                response = await client.request(
+                    method, url, params=params, json=json_body, timeout=request_timeout
+                )
         else:
-            response = await client.get(url, params=params)
+            response = await client.request(
+                method, url, params=params, json=json_body, timeout=request_timeout
+            )
         response.raise_for_status()
         # parse_float=Decimal keeps money exact: JSON numbers never become floats
         # (which the domain models reject), so precision survives from the wire.
@@ -143,6 +164,13 @@ async def _once() -> Any:
             with attempt:
                 return await _once()
     except (httpx.HTTPError, json.JSONDecodeError) as exc:
-        raise SourceUnavailable(source, exc) from exc
+        raise SourceUnavailable(source, _redact(str(exc), secrets)) from exc
     # Unreachable: the loop either returns or reraises, but satisfies the type.
     raise SourceUnavailable(source, "retries exhausted")
+
+
+def _redact(text: str, secrets: tuple[str, ...]) -> str:
+    for secret in secrets:
+        if secret:
+            text = text.replace(secret, "***")
+    return text

packages/recommend/src/hodlin_recommend/connectors/finnhub.py

@@ -68,6 +68,7 @@ async def get_news(self, symbol: str, since: datetime) -> list[NewsItem]:
             params=params,
             rate=self._rate,
             retry=self._retry,
+            secrets=(self._api_key,),  # httpx errors quote the URL, key included
         )
         if not isinstance(payload, list):
             raise SourceUnavailable(self.source, "expected a JSON array of articles")

packages/recommend/src/hodlin_recommend/connectors/massive.py

@@ -85,6 +85,7 @@ async def get_candles(
             params=params,
             rate=self._rate,
             retry=self._retry,
+            secrets=(self._api_key,),  # httpx errors quote the URL, key included
         )
         bars = payload.get("bars") if isinstance(payload, Mapping) else None
         if not isinstance(bars, list):

packages/recommend/src/hodlin_recommend/delivery/__init__.py

@@ -0,0 +1 @@
+"""Telegram delivery (T9): outbound anomaly alerts + a thin inbound poller."""

packages/recommend/src/hodlin_recommend/delivery/formatting.py

@@ -0,0 +1,42 @@
+"""Message formatting — pure functions, and the escaping boundary (T9).
+
+Messages use Telegram HTML parse mode, so every dynamic field is escaped
+*here*, at the one place text becomes markup. The reasoning is LLM prose over
+untrusted headlines (T7): it may quote anything, and none of it may become
+tags. What stays literal markup is only what this module writes itself.
+"""
+
+from html import escape
+
+from hodlin_recommend.domain.models import Anomaly, Explanation
+
+
+def format_anomaly(anomaly: Anomaly, explanation: Explanation) -> str:
+    """One alert: the move in numbers, then the LLM's why, then lineage."""
+    arrow = (
+        "\N{CHART WITH UPWARDS TREND}"
+        if anomaly.direction == "up"
+        else ("\N{CHART WITH DOWNWARDS TREND}")
+    )
+    news_cited = sum(1 for ref in explanation.evidence if ref.kind == "news")
+    lines = [
+        f"{arrow} <b>{escape(anomaly.symbol)}</b> {escape(anomaly.interval)} "
+        f"bar {anomaly.bar_ts:%Y-%m-%d %H:%M} UTC",
+        f"move {anomaly.return_pct:+f}% \N{MIDDLE DOT} z-score {anomaly.z_score} "
+        f"\N{MIDDLE DOT} baseline {anomaly.window} bars",
+        "",
+        escape(explanation.reasoning),
+        "",
+        f"<i>{news_cited} news source(s) cited \N{MIDDLE DOT} "
+        f"{escape(explanation.model_version)}</i>",
+    ]
+    return "\n".join(lines)
+
+
+def format_status(latest: tuple[Anomaly, Explanation] | None) -> str:
+    """The reply to any allowlisted inbound message: the newest explained
+    anomaly, or an honest 'nothing yet'."""
+    if latest is None:
+        return "No explained anomalies yet \N{EM DASH} you'll be notified here when one lands."
+    anomaly, explanation = latest
+    return "Latest anomaly:\n\n" + format_anomaly(anomaly, explanation)

packages/recommend/src/hodlin_recommend/delivery/poller.py

@@ -0,0 +1,90 @@
+"""Inbound long-poll loop (T9): thin by design, default-deny by design.
+
+The poller reads raw updates and enforces the single-ID allowlist where
+messages *enter* the system: an update whose chat or sender isn't the one
+allowlisted ID is dropped without a reply (strangers learn nothing, not even
+that the bot exists). An allowlisted message gets the latest explained
+anomaly back — the reply text comes through an injected async callable so the
+poller itself never touches the database and unit tests never need one.
+
+Runs as a background task owned by the app lifespan; cancellation is the stop
+signal — and the *only* thing that stops it. Any other failure (Telegram
+down, a DB blip while building the reply, a malformed update) is absorbed
+with a backoff or a skip: a background task that dies silently is a bot that
+looks alive and answers no one until the next restart.
+"""
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import Any, Protocol
+
+from hodlin_recommend.delivery.formatting import format_status
+from hodlin_recommend.delivery.telegram import Messenger, UpdateSource
+from hodlin_recommend.store.db import SessionFactory
+from hodlin_recommend.store.repositories import AnomalyRepository
+
+# Tuning, not secrets (D17): how long to sit out a Telegram outage before the
+# next poll attempt.
+ERROR_BACKOFF_S = 5.0
+
+
+class TelegramAPI(Messenger, UpdateSource, Protocol):
+    """Both halves — what the poller needs (read updates, send replies)."""
+
+
+class UpdatePoller:
+    def __init__(
+        self,
+        api: TelegramAPI,
+        *,
+        allowed_chat_id: int,
+        reply_text: Callable[[], Awaitable[str]],
+    ) -> None:
+        self._api = api
+        self._allowed = allowed_chat_id
+        self._reply_text = reply_text
+
+    async def run(self) -> None:
+        """Poll forever; the owner cancels this task to stop it. ``offset``
+        acknowledges processed updates so Telegram never redelivers them.
+
+        ``except Exception`` (never ``CancelledError``) is deliberate at both
+        levels: a failed poll backs off and retries; a poisoned update is
+        acknowledged and skipped rather than redelivered into the same crash
+        forever. Replies are best-effort — the user can just ask again.
+        """
+        offset: int | None = None
+        while True:
+            try:
+                updates = await self._api.get_updates(offset)
+            except Exception:  # the loop must outlive any outage
+                await asyncio.sleep(ERROR_BACKOFF_S)
+                continue
+            for update in updates:
+                update_id = update.get("update_id")
+                if isinstance(update_id, int):
+                    offset = update_id + 1  # acknowledge first: never re-crash on it
+                try:
+                    await self._handle(update)
+                except Exception:  # one bad update must not kill the loop
+                    continue
+
+    async def _handle(self, update: dict[str, Any]) -> None:
+        message = update.get("message") or {}
+        chat_id = (message.get("chat") or {}).get("id")
+        sender_id = (message.get("from") or {}).get("id")
+        if chat_id != self._allowed or sender_id != self._allowed:
+            return  # default-deny: no reply, no error, no acknowledgement
+        await self._api.send(self._allowed, await self._reply_text())
+
+
+def latest_anomaly_reply(session_factory: SessionFactory) -> Callable[[], Awaitable[str]]:
+    """The production reply builder: a fresh session per inbound message,
+    the newest explained anomaly formatted, or an honest 'nothing yet'."""
+
+    async def reply() -> str:
+        async with session_factory() as session:
+            latest = await AnomalyRepository(session).latest_explained()
+        return format_status(latest)
+
+    return reply

packages/recommend/src/hodlin_recommend/delivery/telegram.py

@@ -0,0 +1,115 @@
+"""Thin Telegram Bot API client behind Protocol seams (T9).
+
+Two endpoints are all delivery needs — ``sendMessage`` out, ``getUpdates``
+in — so this is a hand-rolled client over the shared connector resilience
+(rate limit + retry + ``SourceUnavailable``), not a bot framework: Telegram
+is a sink that fails exactly like a source, and the jobs already know how to
+treat a dead source. ``Messenger`` is the outbound seam the notify job mocks;
+``UpdateSource`` is the inbound seam the poller mocks; ``TelegramClient`` is
+the one concrete implementing both.
+
+Delivery semantics are at-least-once by construction: the shared retry can
+repeat a ``sendMessage`` whose reply was lost after the server acted, and the
+notify job commits its claim only after a successful send. A rare duplicate
+alert beats a silently missing one.
+"""
+
+from typing import Any, Protocol, runtime_checkable
+
+import httpx
+
+from hodlin_recommend.connectors.base import (
+    DEFAULT_RETRY,
+    RateLimiter,
+    RetryPolicy,
+    SourceUnavailable,
+    request_json,
+)
+
+# Tuning, not secrets (D17): how long Telegram holds a getUpdates open. The
+# HTTP read timeout must outlive it or every idle poll "times out".
+LONG_POLL_S = 50
+
+
+@runtime_checkable
+class Messenger(Protocol):
+    """Outbound half — everything the notify job needs."""
+
+    async def send(self, chat_id: int, text: str) -> None: ...
+
+
+@runtime_checkable
+class UpdateSource(Protocol):
+    """Inbound half — everything the poller reads. Returns raw Bot API update
+    dicts; interpreting them (and enforcing the allowlist) is the poller's job."""
+
+    async def get_updates(self, offset: int | None) -> list[dict[str, Any]]: ...
+
+
+class TelegramClient:
+    """The one concrete: both halves over the HTTP Bot API."""
+
+    source = "telegram"
+
+    def __init__(
+        self,
+        client: httpx.AsyncClient,
+        *,
+        token: str,
+        base_url: str,
+        rate: RateLimiter,
+        retry: RetryPolicy = DEFAULT_RETRY,
+    ) -> None:
+        self._client = client
+        # Telegram puts the secret in the URL *path*, and httpx error text
+        # quotes the URL — so the token rides every wrapped error unless
+        # request_json redacts it (its text lands in ingest_runs.detail).
+        self._token = token
+        self._base = f"{base_url.rstrip('/')}/bot{token}"
+        self._rate = rate
+        self._retry = retry
+
+    async def _call(
+        self,
+        api_method: str,
+        payload: dict[str, Any],
+        *,
+        http_timeout: httpx.Timeout | None = None,
+    ) -> Any:
+        data = await request_json(
+            self._client,
+            source=self.source,
+            url=f"{self._base}/{api_method}",
+            method="POST",
+            json_body=payload,
+            http_timeout=http_timeout,
+            rate=self._rate,
+            retry=self._retry,
+            secrets=(self._token,),
+        )
+        # Telegram can answer HTTP 200 with ok=false; that's still a failure.
+        if not isinstance(data, dict) or not data.get("ok"):
+            raise SourceUnavailable(self.source, f"API answered not-ok: {data!r}")
+        return data.get("result")
+
+    async def send(self, chat_id: int, text: str) -> None:
+        await self._call(
+            "sendMessage",
+            {
+                "chat_id": chat_id,
+                "text": text,
+                # HTML parse mode pairs with formatting.py escaping every
+                # dynamic field — untrusted text can't become markup.
+                "parse_mode": "HTML",
+                "disable_web_page_preview": True,
+            },
+        )
+
+    async def get_updates(self, offset: int | None) -> list[dict[str, Any]]:
+        payload: dict[str, Any] = {"timeout": LONG_POLL_S, "allowed_updates": ["message"]}
+        if offset is not None:
+            payload["offset"] = offset
+        result = await self._call(
+            "getUpdates", payload, http_timeout=httpx.Timeout(LONG_POLL_S + 10)
+        )
+        return list(result) if isinstance(result, list) else []