# Data Pipeline Architecture — Stonks Oracle

This document describes the end-to-end data pipeline from external data sources through signal processing to trade execution. The pipeline is queue-driven, with Redis lists connecting each stage and PostgreSQL/MinIO providing durable storage at every step.

All queue names follow the convention `stonks:queue:<name>` (see `services/shared/redis_keys.py`). Dead-letter queues mirror the pattern as `stonks:dlq:<name>`.

## Pipeline Overview

```mermaid
flowchart TB
    %% ── External Data Sources ─────────────────────────────────────
    subgraph sources ["External Data Sources"]
        direction LR
        polygon["Polygon.io<br/><i>News, Market Bars,<br/>Grouped Daily</i>"]
        sec["SEC EDGAR<br/><i>10-K, 10-Q Filings</i>"]
        macro_src["Macro News APIs<br/><i>Geopolitical &amp;<br/>Economic Events</i>"]
        market_src["Market Data API<br/><i>Intraday Bars,<br/>Grouped Daily</i>"]
    end

    %% ── Scheduler ─────────────────────────────────────────────────
    scheduler["<b>Scheduler</b><br/><i>services.scheduler.app</i><br/>Cadence polling, rate limiting,<br/>backoff, stale recovery,<br/>periodic aggregation,<br/>report scheduling"]

    sources -.->|"API polling<br/>on cadence"| scheduler

    %% ── Ingestion Queue ───────────────────────────────────────────
    q_ingestion[["stonks:queue:ingestion"]]
    scheduler -->|"rpush job<br/>(company, macro,<br/>global market)"| q_ingestion

    %% ── Ingestion Worker ──────────────────────────────────────────
    ingestion["<b>Ingestion</b><br/><i>services.ingestion.worker</i><br/>Adapter dispatch, dedupe,<br/>raw artifact upload"]

    q_ingestion -->|"lpop"| ingestion

    %% ── Raw Storage ───────────────────────────────────────────────
    minio_raw[("MinIO<br/><i>Raw Artifacts</i><br/>JSON / HTML")]
    pg_docs[("PostgreSQL<br/><i>documents,<br/>ingestion_runs</i>")]
    redis_dedupe[("Redis<br/><i>Dedupe Markers</i><br/>stonks:dedupe:*")]

    ingestion -->|"upload raw payload"| minio_raw
    ingestion -->|"persist metadata"| pg_docs
    ingestion -->|"set content hash"| redis_dedupe

    %% ── Parsing Queue ─────────────────────────────────────────────
    q_parsing[["stonks:queue:parsing"]]
    ingestion -->|"rpush<br/>(news, filings,<br/>web_scrape, macro)"| q_parsing

    %% ── Parser Worker ─────────────────────────────────────────────
    parser["<b>Parser</b><br/><i>services.parser.worker</i><br/>HTML parsing, quality scoring,<br/>company mention detection"]

    q_parsing -->|"lpop"| parser

    minio_norm[("MinIO<br/><i>Normalized Text</i><br/><i>Parser Output JSON</i>")]
    parser -->|"upload normalized text<br/>+ structured output"| minio_norm
    parser -->|"update document status,<br/>insert mentions"| pg_docs
```

## Three Signal Layers

The parser routes documents into two extraction paths based on `document_type`. All three signal layers converge at the aggregation stage through the shared `WeightedSignal` abstraction.

```mermaid
flowchart TB
    %% ── Parser Output ─────────────────────────────────────────────
    parser(("Parser"))

    %% ── Extraction Queues ─────────────────────────────────────────
    q_extraction[["stonks:queue:extraction"]]
    q_macro[["stonks:queue:macro_classification"]]

    parser -->|"rpush<br/>(standard docs)"| q_extraction
    parser -->|"rpush<br/>(macro_event docs)"| q_macro

    %% ── Scheduler Recovery ────────────────────────────────────────
    scheduler_recovery(("Scheduler<br/><i>stale recovery &amp;<br/>failed retry</i>"))
    scheduler_recovery -.->|"re-enqueue orphaned<br/>parsed docs"| q_extraction
    scheduler_recovery -.->|"re-enqueue orphaned<br/>macro docs"| q_macro

    %% ── Extractor Worker ──────────────────────────────────────────
    subgraph extractor_svc ["Extractor Service"]
        direction TB
        ext_main["<b>Extractor</b><br/><i>services.extractor.main</i><br/>Alternates between queues<br/>(2 extraction : 1 macro)<br/>Token budget enforcement"]
    end

    q_extraction -->|"lpop"| ext_main
    q_macro -->|"lpop"| ext_main

    %% ── Ollama LLM ───────────────────────────────────────────────
    ollama["<b>Ollama / vLLM</b><br/><i>LLM Inference</i><br/>document-extractor agent<br/>event-classifier agent"]
    ext_main <-->|"HTTP /api/generate<br/>(AgentConfigResolver<br/>selects model + variant)"| ollama

    %% ── Signal Layer 1: Company ───────────────────────────────────
    subgraph layer1 ["Layer 1 — Company Signals"]
        direction LR
        di["document_intelligence<br/>document_impact_records"]
    end

    ext_main -->|"persist extraction<br/>(standard docs)"| di

    %% ── Signal Layer 2: Macro ─────────────────────────────────────
    subgraph layer2 ["Layer 2 — Macro Signals"]
        direction LR
        ge["global_events"]
        mir["macro_impact_records<br/><i>per-company interpolation<br/>via exposure profiles</i>"]
        ge --> mir
    end

    ext_main -->|"classify &amp; persist<br/>(macro_event docs)"| ge
    ext_main -->|"compute_macro_impact<br/>for all tracked companies"| mir

    %% ── Aggregation Queue ─────────────────────────────────────────
    q_agg[["stonks:queue:aggregation"]]
    ext_main -->|"rpush<br/>(per ticker)"| q_agg

    %% ── Scheduler Periodic Aggregation ────────────────────────────
    scheduler_agg(("Scheduler<br/><i>periodic aggregation<br/>every ~15 min</i>"))
    scheduler_agg -.->|"rpush all<br/>active tickers"| q_agg

    %% ── Aggregation Worker ────────────────────────────────────────
    aggregation["<b>Aggregation</b><br/><i>services.aggregation.main</i><br/>Trend windows, scoring,<br/>contradiction detection"]

    q_agg -->|"lpop"| aggregation

    %% ── Signal Layer 3: Competitive ──────────────────────────────
    subgraph layer3 ["Layer 3 — Competitive Signals"]
        direction LR
        pm["pattern_matcher<br/><i>historical patterns</i>"]
        sp["signal_propagation<br/><i>cross-company signals</i>"]
        csr["competitive_signal_records"]
        pm --> sp --> csr
    end

    aggregation -->|"trigger_signal_propagation<br/>(when competitive_enabled)"| layer3

    %% ── All layers merge ──────────────────────────────────────────
    pg_trends[("PostgreSQL<br/><i>trend_windows,<br/>trend_history,<br/>trend_projections</i>")]

    di -->|"WeightedSignal"| aggregation
    mir -->|"WeightedSignal"| aggregation
    csr -->|"WeightedSignal"| aggregation
    aggregation -->|"persist trend summaries"| pg_trends
```

## Recommendation → Trading → Broker

The recommendation worker consumes from the recommendation queue. The trading engine does **not** consume from a queue — it polls the `recommendations` table in PostgreSQL on a configurable interval, evaluates each recommendation through its decision pipeline, and pushes "act" decisions to the broker queue.

```mermaid
flowchart TB
    %% ── Recommendation Queue ──────────────────────────────────────
    q_rec[["stonks:queue:recommendation"]]
    aggregation(("Aggregation")) -->|"rpush<br/>(ticker + window,<br/>dedup 5 min TTL)"| q_rec

    %% ── Recommendation Worker ─────────────────────────────────────
    recommendation["<b>Recommendation</b><br/><i>services.recommendation.main</i><br/>Eligibility, suppression,<br/>thesis generation"]

    q_rec -->|"lpop"| recommendation

    ollama_thesis["<b>Ollama / vLLM</b><br/><i>thesis-rewriter agent</i><br/>(AgentConfigResolver<br/>selects model + variant)"]
    recommendation <-->|"rewrite thesis<br/>(trading-eligible only)"| ollama_thesis

    pg_recs[("PostgreSQL<br/><i>recommendations,<br/>recommendation_evidence,<br/>risk_evaluations</i>")]
    recommendation -->|"persist recommendation<br/>+ evidence + risk eval"| pg_recs

    %% ── Lake Publication (inline) ─────────────────────────────────
    minio_rec_lake[("MinIO<br/><i>Lakehouse</i><br/>recommendation facts")]
    recommendation -->|"publish_recommendation_facts<br/>(Parquet)"| minio_rec_lake

    %% ── Trading Engine ────────────────────────────────────────────
    subgraph trading_loop ["Trading Engine Decision Loop"]
        direction TB
        poll["Poll recommendations<br/><i>action IN (buy, sell)<br/>mode IN (paper, live)<br/>generated_at &gt; last_poll</i>"]
        dedup_check["Redis dedup check<br/><i>stonks:dedupe:trading:*</i>"]
        evaluate["evaluate_recommendation<br/><i>Circuit breaker check<br/>Trading window check<br/>Confidence gate<br/>Sector exposure check<br/>Correlation check<br/>Earnings blackout<br/>Max positions check</i>"]
        size["Position sizing<br/><i>Kelly criterion,<br/>risk tier limits,<br/>micro-trade support</i>"]
        decide{{"Decision"}}
        poll --> dedup_check --> evaluate --> size --> decide
    end

    pg_recs -->|"SELECT recent<br/>recommendations"| poll

    %% ── Broker Queue ──────────────────────────────────────────────
    q_broker[["stonks:queue:broker_orders"]]
    decide -->|"act → rpush<br/>order job"| q_broker
    decide -->|"skip → persist<br/>decision only"| pg_decisions

    pg_decisions[("PostgreSQL<br/><i>trading_decisions</i>")]

    %% ── Manual Override ───────────────────────────────────────────
    trading_api(("Trading API<br/><i>POST /override/order</i>"))
    trading_api -->|"rpush<br/>manual order"| q_broker

    %% ── Broker Adapter ────────────────────────────────────────────
    broker["<b>Broker Adapter</b><br/><i>services.adapters.broker_service</i><br/>Idempotency, risk evaluation,<br/>approval gate, order submission,<br/>fill tracking, position sync"]

    q_broker -->|"lpop"| broker

    %% ── Risk Engine ───────────────────────────────────────────────
    risk["<b>Risk Engine</b><br/><i>services.risk.app</i><br/>evaluate_order()<br/>Position limits, sector exposure,<br/>daily loss caps, approval workflow"]
    broker -->|"evaluate order<br/>(inline call)"| risk

    %% ── Alpaca ────────────────────────────────────────────────────
    alpaca["<b>Alpaca</b><br/><i>Paper Trading API</i><br/>Order submission,<br/>position sync,<br/>account state"]
    broker <-->|"submit order /<br/>sync positions /<br/>sync order status"| alpaca

    pg_orders[("PostgreSQL<br/><i>orders, order_events,<br/>positions,<br/>portfolio_snapshots,<br/>broker_accounts</i>")]
    broker -->|"persist order,<br/>events, positions"| pg_orders

    %% ── Lake Publication (broker inline) ──────────────────────────
    minio_broker_lake[("MinIO<br/><i>Lakehouse</i><br/>order + fill + position facts")]
    broker -->|"publish_trade_order<br/>publish_trade_fill<br/>publish_positions_daily_batch<br/>(Parquet)"| minio_broker_lake

    %% ── Notifications ─────────────────────────────────────────────
    subgraph notifications ["Notifications"]
        direction LR
        sns["AWS SNS<br/><i>SMS alerts</i>"]
        gmail["Gmail SMTP<br/><i>Email alerts</i>"]
    end

    trading_loop -->|"circuit breaker trips,<br/>order fills,<br/>stop-loss triggers"| notifications
```

## Analytical Branch — Lake Publisher

The lake publisher runs as a separate worker, consuming from its own queue and writing partitioned Parquet fact tables to MinIO for analytical queries. Some services (broker adapter, recommendation worker) also publish facts directly to MinIO inline, bypassing the queue.

```mermaid
flowchart LR
    %% ── Lake Publish Queue ────────────────────────────────────────
    q_lake[["stonks:queue:lake_publish"]]

    various(("Upstream Services<br/><i>via enqueue_lake_job()</i>"))
    various -->|"rpush job<br/>(job_type + entity_id)"| q_lake

    %% ── Lake Publisher Worker ─────────────────────────────────────
    lake["<b>Lake Publisher</b><br/><i>services.lake_publisher.jobs</i><br/>Transforms operational data<br/>into analytical facts<br/><i>15 job types supported</i>"]

    q_lake -->|"lpop"| lake

    pg_source[("PostgreSQL<br/><i>Operational Tables</i><br/>documents, extractions,<br/>orders, positions, events,<br/>global_events, macro_impacts,<br/>competitive_signals")]
    lake -->|"query source data"| pg_source

    %% ── MinIO Parquet ─────────────────────────────────────────────
    minio_lake[("MinIO<br/><i>Lakehouse Bucket</i><br/>Partitioned Parquet<br/>/year=/month=/day=")]
    lake -->|"write Parquet files"| minio_lake

    %% ── Inline Publishers ─────────────────────────────────────────
    inline(("Inline Publishers<br/><i>broker adapter,<br/>recommendation worker</i>"))
    inline -->|"publish_* functions<br/>(direct Parquet write)"| minio_lake

    %% ── Trino ─────────────────────────────────────────────────────
    trino["<b>Trino</b><br/><i>SQL Query Engine</i><br/>Hive connector → MinIO"]
    minio_lake -->|"read via<br/>Hive Metastore"| trino

    hive["<b>Hive Metastore</b><br/><i>Schema catalog</i>"]
    trino <-->|"table metadata"| hive
    hive -->|"location refs"| minio_lake

    %% ── Visualization ─────────────────────────────────────────────
    superset["<b>Superset</b><br/><i>Dashboards &amp;<br/>SQL Lab</i>"]
    dashboard["<b>React Dashboard</b><br/><i>frontend</i><br/>Charts, portfolio,<br/>recommendations"]
    query_api["<b>Query API</b><br/><i>services.api.app</i>"]

    trino --> superset
    trino --> query_api
    query_api --> dashboard
```

## Report Generation

The scheduler manages report generation as a sub-loop, enqueuing daily and weekly report jobs to a dedicated queue and consuming them inline.

```mermaid
flowchart LR
    scheduler["<b>Scheduler</b><br/><i>report schedule check</i><br/>daily @ 16:30 ET<br/>weekly @ Saturday"]

    q_report[["stonks:queue:report_generation"]]
    scheduler -->|"rpush<br/>(daily/weekly)"| q_report

    scheduler_consumer["<b>Scheduler</b><br/><i>report consumer loop</i><br/>pops up to 5 jobs/cycle"]
    q_report -->|"lpop"| scheduler_consumer

    generator["<b>Report Generator</b><br/><i>services.reporting.generator</i>"]
    scheduler_consumer -->|"process_report_job()"| generator

    pg_reports[("PostgreSQL<br/><i>trading_reports</i>")]
    generator -->|"persist report"| pg_reports
```

## Complete Queue Topology

| Queue | Full Key | Producer(s) | Consumer |
|-------|----------|-------------|----------|
| Ingestion | `stonks:queue:ingestion` | Scheduler (company, macro, global market sources) | Ingestion Worker |
| Parsing | `stonks:queue:parsing` | Ingestion Worker (news, filings, web_scrape, macro) | Parser Worker |
| Extraction | `stonks:queue:extraction` | Parser (standard docs), Scheduler (stale recovery) | Extractor Worker |
| Macro Classification | `stonks:queue:macro_classification` | Parser (macro_event docs), Scheduler (stale/failed recovery) | Extractor Worker |
| Aggregation | `stonks:queue:aggregation` | Extractor Worker (per ticker), Scheduler (periodic, all tickers) | Aggregation Worker |
| Recommendation | `stonks:queue:recommendation` | Aggregation Worker (ticker + window, 5 min dedup TTL) | Recommendation Worker |
| Broker Orders | `stonks:queue:broker_orders` | Trading Engine (act decisions), Trading API (manual overrides) | Broker Adapter |
| Lake Publish | `stonks:queue:lake_publish` | Various services (via `enqueue_lake_job()`) | Lake Publisher |
| Report Generation | `stonks:queue:report_generation` | Scheduler (daily/weekly triggers) | Scheduler (inline consumer) |

Dead-letter queues follow the pattern `stonks:dlq:<queue_name>` and are populated when a job exhausts its retry budget.

## Data Store Summary

| Store | Role | Key Tables / Buckets |
|-------|------|---------------------|
| **PostgreSQL** | Structured operational data | `documents`, `document_intelligence`, `document_impact_records`, `document_company_mentions`, `global_events`, `macro_impact_records`, `exposure_profiles`, `competitive_signal_records`, `competitor_relationships`, `trend_windows`, `trend_history`, `trend_projections`, `recommendations`, `recommendation_evidence`, `risk_evaluations`, `orders`, `order_events`, `positions`, `portfolio_snapshots`, `trading_decisions`, `circuit_breaker_events`, `reserve_pool_ledger`, `risk_tier_history`, `broker_accounts`, `ingestion_runs`, `sources`, `companies`, `company_aliases`, `ai_agents`, `agent_variants`, `agent_performance_log`, `risk_configs`, `trading_reports` |
| **Redis** | Queues, dedup markers, rate limits, circuit breaker state, pipeline toggle | `stonks:queue:*` (9 queues), `stonks:dedupe:*`, `stonks:dedupe:trading:*`, `stonks:ratelimit:*`, `stonks:trading:circuit_breaker:*`, `stonks:trading:notification_rate:*`, `stonks:order_idempotency:*`, `stonks:lock:*`, `stonks:cache:*`, `stonks:retry:*`, `stonks:rec_dedup:*`, `stonks:pipeline:enabled`, `stonks:dlq:*` |
| **MinIO** | Object storage for raw artifacts, normalized text, and analytical Parquet files | Raw artifacts bucket, normalized text bucket, parser output bucket, lakehouse bucket (partitioned Parquet: documents, extractions, market bars/quotes, orders, fills, positions, PnL, global events, macro impacts, trend projections, competitive signals, competitor relationships, recommendations) |

## External Integration Points

| Integration | Service | Protocol | Purpose |
|-------------|---------|----------|---------|
| **Polygon.io** | Ingestion (via PolygonNewsAdapter, PolygonMarketAdapter) | HTTPS REST | News articles, market bars, grouped daily data, intraday bars |
| **SEC EDGAR** | Ingestion (via SECEdgarAdapter) | HTTPS REST | 10-K, 10-Q filings |
| **Macro News** | Ingestion (via MacroNewsAdapter) | HTTPS REST | Geopolitical and economic event articles |
| **Ollama / vLLM** | Extractor, Recommendation | HTTP `/api/generate` | LLM inference for document extraction (document-extractor agent), event classification (event-classifier agent), thesis rewriting (thesis-rewriter agent). Model and variant selected via `AgentConfigResolver` with 60s TTL cache. |
| **Alpaca** | Broker Adapter | HTTPS REST | Paper/live trading: order submission, position sync, account state, order status polling |
| **AWS SNS** | Trading Engine (notifications) | boto3 SDK | SMS alerts for circuit breaker trips, order fills, stop-loss triggers |
| **Gmail** | Trading Engine (notifications) | SMTP (port 587 STARTTLS) | Email alerts for trading events |
| **Trino** | Query API, Superset | HTTP | SQL queries over lakehouse Parquet files via Hive Metastore |

## Pipeline Toggle

The pipeline can be paused globally via the Redis key `stonks:pipeline:enabled`. When set to `"0"`, all queue workers (ingestion, parser, extractor, aggregation, recommendation, broker adapter, lake publisher) enter a sleep loop and stop processing jobs. The scheduler also skips scheduling cycles when the toggle is off. The toggle can be set via the Query API's pipeline control endpoints.

Setting `PIPELINE_DEFAULT_OFF=true` on the scheduler initializes the toggle to OFF on first boot, useful for staged deployments where you want to verify infrastructure before enabling the pipeline.