News Collector Revamp

Architecture Overview

Article row carries state SMALLINT (IntEnum 0-4, -1 failed). Five stages hand off via asyncio.Queue (instant). DB poll is fallback for crash recovery. Worker pool per stage with backlog cap so queue tasks always have capacity.

Execution Order

Phases 1-3 are independently deployable. Phase 4 builds on all three. Phase 5 builds on all four. Phase 6 builds on Phase 5. Phase 1 migration reserves schema for Phases 4 and 6.

Supporting Layers

Files Changed Per Phase

Key Constraints

• Python 3.13, async/await throughout
• Plain functions accepting asyncpg.Pool, no classes for DB
• TOML config, backward compatible
• No new features — refactor only (phases 1-3)

Phase 1 — DB Layer

asyncpg replacing SQLAlchemy · 01-DB-LAYER.md

Current State

lib/article/databases.py defines five classes wrapping SQLAlchemy Core:

SQLAlchemy Core used purely as query builder — no ORM, no relationships, no migrations.

Target State

Plain async functions in lib/src/lib/news/db/, accepting asyncpg.pool.Pool as first argument. No classes.

Articles Schema

CREATE TABLE articles (
    id         SERIAL PRIMARY KEY,
    url        TEXT UNIQUE NOT NULL,
    created_at TIMESTAMP NOT NULL,
    updated_at TIMESTAMP NOT NULL DEFAULT now(),
    state      SMALLINT NOT NULL DEFAULT 0,
    kind       SMALLINT NOT NULL DEFAULT 0,    -- reserved phase 4
    eval       SMALLINT NOT NULL DEFAULT 0,    -- reserved phase 4
    data       JSONB
);
CREATE INDEX idx_articles_state
    ON articles (state, created_at);

In-flight claim: state = -(stage + 10). E.g. embedding = -11, enriching = -12, scoring = -13.

Other Tables

Pattern: Direct asyncpg

# lib/news/db/articles.py
from asyncpg import Pool

async def insert(pool: Pool, url: str, data: dict) -> int:
    async with pool.acquire() as conn:
        return await conn.fetchval(
            """INSERT INTO articles (url, created_at, data)
               VALUES ($1, now(), $2)
               RETURNING id""",
            url, data,
        )

No ORM, no query builder, no class wrapping. Parameterized queries via positional $1, $2 placeholders (asyncpg native).

Function Signatures

All functions take pool: asyncpg.Pool as first argument.

Migration Path

1. Create lib/news/db/ with asyncpg functions
2. Create pool in main.py: pool = await asyncpg.create_pool(dsn)
3. Replace ArticleDatabase calls with db.articles.insert(pool, ...)
4. Replace ListsDatabase (sync) with async equivalents + TTL cache
5. Remove SQLAlchemy engine creation from main
6. Keep lib/article/databases.py untouched (used by server/)

Phase 2 — LLM Layer

thin client + routing · 02-LLM-LAYER.md

Current State

news-collector/llms.py (195 lines): ChatClient wrapping LiteLLM Router with LeakyBucket rate limiting, multi-key load balancing, 3 retries. Supports Gemini, OpenRouter, OpenAI, Anthropic — all via LiteLLM.

What Changes

1. Add claude-serve endpoint (self-hosted Claude via HTTP)
2. Add codex-serve endpoint (self-hosted Codex via HTTP)
3. Routing: claude-serve (preferred) -> codex-serve -> LiteLLM (fallback)
4. Keep LiteLLM as base for 100+ cloud providers

Routing

Config

[endpoints]
claude_url = "http://localhost:8080/v1/chat/completions"
codex_url  = "http://localhost:8081/v1/chat/completions"

Both optional. If neither is set, behavior is identical to today (LiteLLM only).

Model Uses in Pipeline

All use ChatClient.query(messages=[...]). No tool-use, no streaming. Simple request/response.

Rate Limiting

Global LeakyBucket stays. Self-hosted endpoints can have separate buckets:

[endpoints]
claude_bucket_burst = 10.0
claude_bucket_rate  = 1e9   # 1 req/sec = 60 RPM

What Stays

• create_chat_client() factory function
• ModelConfiguration dataclass
• Provider key loading from secrets
• All prompt generation in models.py

Phase 3 — Pipeline

staged workers · 03-PIPELINE.md

Problem

Current on_article() holds a global asyncio.Lock for the entire chain — DB insert, Weaviate insert, all model evals, DB update, selector, notification. One article blocks all 30+ sources. The LLM call alone takes seconds.

Design

Article row carries a state SMALLINT column. Processing split into stages. Each stage has a loop that picks articles at state = stage - 1, does work, advances to state = stage.

class State(IntEnum):
    INGESTED  = 0   # row inserted, deduped
    EMBEDDED  = 1   # Weaviate vectors inserted
    ENRICHED  = 2   # keywords, symbols, novelty, context
    SCORED    = 3   # LLM relevance + translation
    DONE      = 4   # selectors run, notified
    FAILED    = -1  # terminal, logged

Stage Loop Pattern

All stages run the same loop:

1. Try queue.get() with poll_timeout (e.g. 5s)
2. If got ID: claim row, process, advance state, push to next queue. Drain queue non-blocking for more.
3. If timeout: poll DB for unclaimed rows at state = stage - 1, process (capped at max_backlog concurrency).
4. Loop.

Workers are asyncio.create_task() behind two semaphores: sem(max_workers) for all work, backlog_sem(max_backlog) nested for DB-sourced work only.

DB Primitives

async def claim_one(pool, article_id, pick_state):
    """Queue fast path. Tx + row lock."""
    claim = -(pick_state + 11)
    async with pool.acquire() as conn:
        async with conn.transaction():
            return await conn.fetchrow(
                """UPDATE articles SET state = $3
                   WHERE id = $1 AND state = $2
                   RETURNING id, data""",
                article_id, int(pick_state), claim,
            )

async def pick_batch(pool, pick_state, limit):
    """DB recovery path. Tx + SKIP LOCKED."""
    claim = -(pick_state + 11)
    async with pool.acquire() as conn:
        async with conn.transaction():
            return await conn.fetch(
                """UPDATE articles SET state = $3
                   WHERE id IN (
                       SELECT id FROM articles
                       WHERE state = $1
                       ORDER BY created_at LIMIT $2
                       FOR UPDATE SKIP LOCKED
                   )
                   RETURNING id, data""",
                int(pick_state), limit, claim,
            )

Config

[pipeline.ingest]
max_workers = 4
max_backlog = 2
poll_timeout = 5.0

[pipeline.embed]
max_workers = 4
max_backlog = 2
poll_timeout = 5.0

[pipeline.enrich]
max_workers = 10
max_backlog = 5
poll_timeout = 5.0

[pipeline.score]
max_workers = 4
max_backlog = 2
poll_timeout = 5.0

[pipeline.notify]
max_workers = 10
max_backlog = 5
poll_timeout = 5.0

Wiring in main.py

embed_q  = asyncio.Queue()
enrich_q = asyncio.Queue()
score_q  = asyncio.Queue()
notify_q = asyncio.Queue()

await asyncio.gather(
    ingest_loop(pool, sources, embed_q, ingest_cfg),
    embed_loop(pool, embed_q, enrich_q, embed_cfg),
    enrich_loop(pool, enrich_q, score_q, enrich_cfg),
    score_loop(pool, score_q, notify_q, score_cfg),
    notify_loop(pool, notify_q, None, notify_cfg),
)

What Changes

Error Handling

Failed processing -> state = -1 (FAILED). A recovery loop resets stale in-flight states (negative, older than N minutes) back to positive for retry. Permanently failed rows stay at -1.

Phase 4 — Heterogeneous Data

articles vs metrics · 05-HETEROGENEOUS.md

The Problem

The pipeline treats everything as isolated Article rows. But data has two shapes:

Two Flows

Context via Linked Graph

select_context() models.py:45-67 follows article.links types_.py:146 by DB lookup. Threads already linked by URL chains. No explicit thread_id needed.

Metrics Table

CREATE TABLE metrics (
    article_id INTEGER NOT NULL REFERENCES articles(id),
    mtime      TIMESTAMP NOT NULL DEFAULT now(),
    price      DOUBLE PRECISION,
    volume     DOUBLE PRECISION,
    market_cap DOUBLE PRECISION,
    impressions DOUBLE PRECISION,
    likes       DOUBLE PRECISION,
    retweets    DOUBLE PRECISION,
    replies     DOUBLE PRECISION,
    PRIMARY KEY (article_id, mtime)
);
CREATE INDEX idx_metrics_article_time
    ON metrics (article_id, mtime DESC);

Metrics as Pipeline Context

Context assembly happens in ENRICH, evaluation in SCORE.

Metric context generators format 5min interval snapshots + latest point + delta:

Metrics Ingest

@dataclass
class Metric:
    article_id: int
    price: float | None = None
    volume: float | None = None
    impressions: float | None = None
    # ... etc

Re-evaluation Flow

Significance thresholds configurable per metric type. Max K evals per article prevents infinite loops.

Hyperliquid / Polymarket Changes

• First appearance of ticker -> create article -> full pipeline run
• Subsequent fills/trades -> metric rows on that article
• No new articles per trade. Ticker article is the persistent entity.
• Metric crosses threshold -> re-evaluation

article_history Table

CREATE TABLE article_history (
    id         SERIAL PRIMARY KEY,
    article_id INTEGER NOT NULL REFERENCES articles(id),
    eval       SMALLINT NOT NULL,
    data       JSONB NOT NULL,
    state      SMALLINT NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT now()
);
CREATE INDEX idx_article_history_article
    ON article_history (article_id, eval);

On re-eval: snapshot current row into article_history, update article with new data + eval incremented. Articles table always has the latest version.

Full Schema Diagram

Impact on Earlier Phases

Phase 5 — Agent Access

market analyst API · 06-AGENT-ACCESS.md

What Already Exists

The collector is a complete market intelligence engine. It just has no API.

Five Endpoints

MCP Tools

For AI agents using tool-use (MCP protocol), each endpoint maps to a tool definition:

Implementation

New module: news-collector/api.py. Endpoints added to the existing aiohttp web server (port 49199). Handlers pull pool, vectors, models from app state.

Phase 6 — Steering Layer

agent-mutable config · 07-STEERING.md

Problem

Phase 5 gives the agent read-only access. But configuration is split across three places: Google Sheets (word lists, projects), TOML (static thresholds), and DB config table (LLM prompts). The agent can observe but not steer.

Design

DB becomes single source of truth for all runtime-mutable config. MCP tools for CRUD. Every mutation audit-logged.

Projects Schema

CREATE TABLE projects (
    id         SERIAL PRIMARY KEY,
    symbol     TEXT NOT NULL,
    name       TEXT NOT NULL,
    twitter    TEXT,
    telegram   TEXT,
    discord    TEXT,
    website    TEXT,
    meta       JSONB DEFAULT '{}'::jsonb,
    created_at TIMESTAMP NOT NULL DEFAULT now(),
    updated_at TIMESTAMP NOT NULL DEFAULT now(),
    UNIQUE (symbol, name)
);
CREATE INDEX idx_projects_symbol ON projects (symbol);

Audit Log Schema

CREATE TABLE audit_log (
    id         SERIAL PRIMARY KEY,
    created_at TIMESTAMP NOT NULL DEFAULT now(),
    actor      TEXT NOT NULL,
    action     TEXT NOT NULL,
    target     TEXT NOT NULL,
    detail     JSONB
);
CREATE INDEX idx_audit_log_time
    ON audit_log (created_at DESC);

MCP Tools

Steering tools added to same mcp.py as Phase 5 read-only tools. Four groups:

Plus optional import_from_sheets for one-time migration from Sheets (additive only, never deletes from DB).

What the Agent Can Steer

GoogleSheetsUpdater Role Change

Impact on Earlier Phases

Implementation

Open Questions

04-OPEN-QUESTIONS.md