Merge pull request #221 from chinelo002/feat/req-docs-logging-graph-memory-187-195-199

docs+chore: clean requirements, central logging, streaming graph builder

Author: gelluisaac

REQUIREMENTS.md

@@ -0,0 +1,77 @@
+# Python requirements files
+
+Three requirements files live at the repo root. Pick the one that matches
+your environment — they are *not* meant to be combined.
+
+## Decision tree
+
+```
+Need to train models (GPU/CUDA available)?
+└─ yes → pip install -r requirements.txt
+└─ no
+   └─ Need to run training or feature jobs (CPU only)?
+      └─ yes → pip install -r requirements-cpu.txt
+      └─ no
+         └─ Just want to load Hydra config / parse dataframes /
+            run unit tests that don't touch torch?
+            └─ yes → pip install -r requirements-minimal.txt
+```
+
+## What each file ships
+
+### `requirements.txt` — full GPU training stack
+The everything-on-board file. Pulls the full GPU `torch` wheel,
+`pytorch-lightning`, `mlflow`, the feature-store stack (redis, pyarrow,
+fastparquet, networkx, click, rich), visualization (matplotlib, seaborn),
+notebooks, and dev tooling (pytest + black + flake8 + mypy).
+
+Use this on GPU CI runners and developer machines that build dashboards or
+notebooks.
+
+### `requirements-cpu.txt` — CPU-only training stack
+Same shape as `requirements.txt` but pins the **CPU-only** torch wheels
+from the official PyTorch CPU index:
+
+```
+torch>=2.0.0+cpu --index-url https://download.pytorch.org/whl/cpu
+```
+
+Drops `mlflow`, `scikit-learn` standalone (still pulled transitively via
+some libs), the feature-store stack, visualization, dev tooling, and
+notebooks — they're not needed for headless CPU jobs. Pick this when:
+
+- You're building the Docker image for production / CI.
+- You're running batch ingestion or model serving on a CPU box.
+- You want the fastest possible `pip install` for a smoke test.
+
+### `requirements-minimal.txt` — Hydra + dataframes only
+The smallest viable set: `numpy`, `pandas`, `polars`, `pyyaml`,
+`hydra-core`, `omegaconf`. Nothing else. Use it when:
+
+- You just want to import `astroml.config` and resolve a Hydra schema.
+- You're running config-only unit tests in CI.
+- You're embedding a small piece of astroml into another service and want
+  to keep the install footprint tiny.
+
+## Pin policy
+
+Where a package appears in more than one file, the lower bound is held in
+sync across all of them. The actual lower bounds in use:
+
+| package          | pin                | files                                           |
+|------------------|--------------------|-------------------------------------------------|
+| `numpy`          | `>=1.24`           | requirements.txt, -cpu.txt, -minimal.txt        |
+| `pandas`         | `>=2.0`            | requirements.txt, -cpu.txt, -minimal.txt        |
+| `polars`         | `>=1.0`            | requirements.txt, -cpu.txt, -minimal.txt        |
+| `pyyaml`         | `>=6.0`            | requirements.txt, -cpu.txt, -minimal.txt        |
+| `hydra-core`     | `>=1.3.0`          | requirements.txt, -cpu.txt, -minimal.txt        |
+| `omegaconf`      | `>=2.3.0`          | requirements.txt, -cpu.txt, -minimal.txt        |
+| `torch`          | `>=2.0.0` / `+cpu` | requirements.txt (GPU), -cpu.txt (CPU)          |
+| `torch-geometric`| `>=2.3.0`          | requirements.txt, -cpu.txt                      |
+| `sqlalchemy`     | `>=2.0`            | requirements.txt, -cpu.txt                      |
+| `psycopg2-binary`| `>=2.9`            | requirements.txt, -cpu.txt                      |
+| `aiohttp`        | `>=3.9`            | requirements.txt, -cpu.txt                      |
+| `stellar-sdk`    | `>=9.0.0`          | requirements.txt, -cpu.txt                      |
+
+If you bump one, run `grep -E "^<package>\b" requirements*.txt` to confirm
+you've bumped them in lockstep.

astroml/features/graph/snapshot.py

@@ -2,10 +2,16 @@
 
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from typing import Generator, Iterable, List, Optional, Sequence, Set, Tuple
+from typing import Generator, Iterable, Iterator, List, Optional, Sequence, Set, Tuple
 import bisect
 
 
+# Issue #199 — default chunk size for the streaming graph builder. SQLAlchemy
+# fetches rows from the DB in batches of this many; the iterator yields each
+# edge individually so callers never see a fully-materialised window list.
+DEFAULT_STREAM_CHUNK_SIZE = 5_000
+
+
 @dataclass(frozen=True)
 class Edge:
     src: str
@@ -116,6 +122,111 @@ def _parse_window_size(window: str) -> timedelta:
     raise ValueError(f"Unknown window unit '{unit}'. Use 'd', 'h', or 's'.")
 
 
+@dataclass(frozen=True)
+class SnapshotMeta:
+    """Window metadata without the edge payload — issue #199.
+
+    Yielded alongside a fresh edge iterator by
+    :func:`iter_db_snapshot_edges` so callers can decide how (or whether)
+    to buffer the edges, instead of being forced to hold a fully-built
+    ``List[Edge]`` in RAM.
+    """
+
+    index: int
+    start: datetime
+    end: datetime
+
+
+def iter_db_snapshot_edges(
+    window: str = "7d",
+    t0: Optional[datetime] = None,
+    t_now: Optional[datetime] = None,
+    step: Optional[str] = None,
+    session=None,
+    chunk_size: int = DEFAULT_STREAM_CHUNK_SIZE,
+) -> Generator[Tuple["SnapshotMeta", Iterator["Edge"]], None, None]:
+    """Streaming variant of :func:`iter_db_snapshots` — issue #199.
+
+    Each yielded ``(meta, edges)`` pair gives the window bounds plus a
+    fresh generator that pulls rows from the database in chunks of
+    ``chunk_size`` via SQLAlchemy's ``yield_per`` and converts each row
+    into an :class:`Edge` lazily. Peak memory per window is bounded by
+    ``chunk_size`` (default 5 000 edges ≈ a few MB) regardless of how
+    many edges the window actually contains.
+
+    The edge iterator MUST be drained or discarded before advancing to
+    the next ``(meta, edges)`` pair — the underlying SQLAlchemy result
+    will be reused. The function does not yield a ``nodes`` set; build
+    it incrementally if you need one.
+
+    Use this in place of :func:`iter_db_snapshots` whenever a window may
+    plausibly contain enough edges to risk OOM on the training machine.
+    """
+    from astroml.db.schema import NormalizedTransaction
+    from sqlalchemy import func as sqlfunc, select
+
+    if session is None:
+        from astroml.db.session import get_session
+        session = get_session()
+
+    win_delta = _parse_window_size(window)
+    step_delta = _parse_window_size(step) if step else win_delta
+
+    if t_now is None:
+        t_now = datetime.now(timezone.utc)
+
+    if t0 is None:
+        result = session.execute(
+            select(sqlfunc.min(NormalizedTransaction.timestamp))
+        ).scalar()
+        if result is None:
+            return  # empty DB
+        t0 = result if result.tzinfo else result.replace(tzinfo=timezone.utc)
+
+    if t_now.tzinfo is None:
+        t_now = t_now.replace(tzinfo=timezone.utc)
+    if t0.tzinfo is None:
+        t0 = t0.replace(tzinfo=timezone.utc)
+
+    window_start = t0
+    index = 0
+
+    while window_start < t_now:
+        window_end = min(window_start + win_delta, t_now)
+
+        result = session.execute(
+            select(
+                NormalizedTransaction.sender,
+                NormalizedTransaction.receiver,
+                NormalizedTransaction.timestamp,
+            )
+            .where(
+                NormalizedTransaction.timestamp >= window_start,
+                NormalizedTransaction.timestamp <= window_end,
+                NormalizedTransaction.receiver.isnot(None),
+                NormalizedTransaction.sender != NormalizedTransaction.receiver,
+            )
+            .order_by(NormalizedTransaction.timestamp)
+            .execution_options(yield_per=chunk_size, stream_results=True)
+        )
+
+        def _edges_iter(_result=result) -> Iterator[Edge]:
+            for row in _result:
+                yield Edge(
+                    src=row.sender,
+                    dst=row.receiver,
+                    timestamp=int(row.timestamp.timestamp()),
+                )
+
+        yield (
+            SnapshotMeta(index=index, start=window_start, end=window_end),
+            _edges_iter(),
+        )
+
+        window_start += step_delta
+        index += 1
+
+
 def iter_db_snapshots(
     window: str = "7d",
     t0: Optional[datetime] = None,

astroml/ingestion/enhanced_service.py

@@ -266,13 +266,16 @@ def get_all_stats(self) -> Dict[str, any]:
 # ---------------------------------------------------------------------------
 
 def _configure_logging() -> None:
-    """Configure structured logging."""
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s %(levelname)-8s [%(name)s] %(message)s",
-        datefmt="%Y-%m-%d %H:%M:%S",
-        stream=sys.stderr,
-    )
+    """Configure structured logging.
+
+    Delegates to :func:`astroml.utils.logging.configure_logging` so log
+    level (``ASTROML_LOG_LEVEL``) and format (``ASTROML_LOG_FORMAT=
+    text|json``) are consistent across every astroml entry point. See
+    issue #195.
+    """
+    from astroml.utils.logging import configure_logging
+
+    configure_logging()
 
 
 async def run_single_stream(config: EnhancedStreamConfig) -> None:

astroml/ingestion/stellar_ledger.py

@@ -132,7 +132,11 @@ async def main():
 
     args = parser.parse_args()
 
-    logging.basicConfig(level=logging.INFO)
+    # Issue #195 — central logging config (level + text/json format)
+    # via ASTROML_LOG_LEVEL / ASTROML_LOG_FORMAT env vars.
+    from astroml.utils.logging import configure_logging
+
+    configure_logging()
 
     async with StellarLedgerDownloader() as downloader:
         try:

astroml/utils/logging.py

@@ -0,0 +1,145 @@
+"""Centralized logging configuration for astroml (issue #195).
+
+Every CLI entry point and long-running service should call
+:func:`configure_logging` early in startup. It standardises:
+
+- Log level (set via ``ASTROML_LOG_LEVEL`` env var; default ``INFO``).
+- Output format (set via ``ASTROML_LOG_FORMAT`` env var: ``text`` for
+  human-readable, ``json`` for structured aggregator-friendly output;
+  default ``text``).
+
+Modules should keep using ``logging.getLogger(__name__)`` as they already
+do — calling :func:`configure_logging` once at startup is the only
+required change for the structured output to take effect everywhere.
+
+Replacement for ad-hoc ``logging.basicConfig(...)`` calls scattered
+through ingestion services.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import sys
+from typing import Optional
+
+
+_DEFAULT_LEVEL = "INFO"
+_DEFAULT_FORMAT = "text"
+_TEXT_FORMAT = "%(asctime)s %(levelname)-7s %(name)s — %(message)s"
+
+# Guard so importing this module twice (or `configure_logging` being called
+# from both a library and a CLI entry point) doesn't pile multiple
+# StreamHandlers onto the root logger.
+_CONFIGURED = False
+
+
+class _JsonFormatter(logging.Formatter):
+    """One JSON object per log record, structured for log aggregators.
+
+    Avoids a hard dependency on ``python-json-logger`` (which isn't pinned
+    in any of the requirements files). The fields are the ones aggregators
+    like Datadog / Loki / CloudWatch surface by default.
+    """
+
+    def format(self, record: logging.LogRecord) -> str:  # noqa: D401
+        payload = {
+            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+        }
+        if record.exc_info:
+            payload["exc_info"] = self.formatException(record.exc_info)
+        # Surface any structured `extra={...}` fields the caller passed.
+        for key, value in record.__dict__.items():
+            if key in payload:
+                continue
+            if key in {
+                "args",
+                "asctime",
+                "created",
+                "exc_info",
+                "exc_text",
+                "filename",
+                "funcName",
+                "levelname",
+                "levelno",
+                "lineno",
+                "module",
+                "msecs",
+                "msg",
+                "name",
+                "pathname",
+                "process",
+                "processName",
+                "relativeCreated",
+                "stack_info",
+                "thread",
+                "threadName",
+                "taskName",
+            }:
+                continue
+            # Only include serialisable values; fall back to repr().
+            try:
+                json.dumps(value)
+                payload[key] = value
+            except (TypeError, ValueError):
+                payload[key] = repr(value)
+        return json.dumps(payload, default=str)
+
+
+def configure_logging(
+    level: Optional[str] = None,
+    format: Optional[str] = None,  # noqa: A002 - matches argparse arg name
+    force: bool = False,
+) -> None:
+    """Configure the root logger.
+
+    Parameters
+    ----------
+    level:
+        Log level string (``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``,
+        ``CRITICAL``). Falls back to ``ASTROML_LOG_LEVEL`` env var, then
+        ``INFO``.
+    format:
+        Either ``"text"`` (human-readable single line) or ``"json"``
+        (one JSON object per line). Falls back to ``ASTROML_LOG_FORMAT``
+        env var, then ``text``.
+    force:
+        If True, reconfigure even if :func:`configure_logging` has been
+        called already in this process.
+    """
+    global _CONFIGURED
+    if _CONFIGURED and not force:
+        return
+
+    resolved_level = (
+        level
+        or os.environ.get("ASTROML_LOG_LEVEL")
+        or _DEFAULT_LEVEL
+    ).upper()
+    resolved_format = (
+        format
+        or os.environ.get("ASTROML_LOG_FORMAT")
+        or _DEFAULT_FORMAT
+    ).lower()
+
+    handler = logging.StreamHandler(stream=sys.stderr)
+    if resolved_format == "json":
+        handler.setFormatter(_JsonFormatter())
+    else:
+        handler.setFormatter(logging.Formatter(_TEXT_FORMAT))
+
+    root = logging.getLogger()
+    # Clear any previously installed handlers so we don't get duplicate
+    # lines when a library called `logging.basicConfig(...)` first.
+    for existing in list(root.handlers):
+        root.removeHandler(existing)
+    root.addHandler(handler)
+    root.setLevel(resolved_level)
+
+    _CONFIGURED = True
+
+
+__all__ = ["configure_logging"]