Data pipelines & the lake

Parent: Architecture · Covers roadmap phases 1 (portfolio import), 2 (lake + prices), and the data half of 6 (news). Status: as built — all three phases shipped in felix-app; as-built notes are inline.

1. Scope and goals

This piece is everything that gets data into the system and shaped for consumers (analytics engine, daily agent, chat agent, web app):

• Prices pipeline (shared): EOD OHLCV, corporate actions, adjusted series, reference data, for a universe derived from what users actually hold and watch.
• News pipeline (shared): raw articles → deduped, clustered "stories" with one cheap-LLM synthesis pass each. Agents only ever see stories.
• Portfolio ingestion (per-user): broker data behind a BrokerConnector contract, CSV import first, producing normalized transactions, tax lots, positions, and cash.
• Lake foundations and orchestration shared by all three.

Goals: rebuildable from raw at all times; idempotent re-runs; downstream consumers never touch paths or providers directly; honest data-freshness signals; LLM cost concentrated in one amortized place (story synthesis).

Non-goals (v1): intraday bars (EOD app; delayed quotes are fetched live on the ticker page with a short cache and never stored), options/derivatives, multi-currency (USD only; schema carries currency for later), crypto, direct index levels (ETF proxies instead), streaming/event-driven ingestion.

2. Lake foundations

2.1 Zones and layout

Object storage (MinIO in dev; R2 or S3 in prod), tenant-first prefixes so isolation is a prefix rule:

• market/raw/{source}/{dataset}/date=YYYY-MM-DD/… — immutable, append-only, exactly-as-fetched payloads (gzipped JSON/CSV). Never rewritten, never deleted. This is the audit and replay layer; a provider swap or schema change never loses history.
• market/curated/{table}/… — normalized, typed, deduplicated Parquet (zstd), Hive-style partitioning. Deterministically rebuildable from raw.
• users/{user_id}/raw/imports/{import_id}/… — original uploaded broker files, immutable.
• users/{user_id}/curated/{table}/… — per-user Parquet (portfolio snapshots time series).

Writes to curated happen at partition granularity with overwrite semantics: re-running a day replaces that day's partition, which is what makes every pipeline idempotent.

Supabase Storage was considered for this layer when Supabase became the app DB + auth (decision) and rejected: the lake stays on R2/MinIO, where free egress matters for analytics re-scans of Parquet.

2.2 Conventions

• Timestamps stored UTC; trading_date is the exchange-local (NYSE) calendar date and is the partition key for market tables.
• Tickers uppercase, US-listed equities and ETFs v1.
• Additive schema changes are fine in Parquet; breaking changes create a new table directory (prices_eod_v2) mapped inside the lake module, with raw enabling a full rebuild.
• Small daily files are acceptable at this scale; a monthly compaction job is a known future need, not a v1 feature.

2.3 The lake module (single point of access)

One Python module in the brain owns paths, partitioning, and DuckDB connections, and exposes logical tables: prices(tickers, range, adjusted=True/False), corporate_actions(...), trading_days(...), tickers(), news_stories(...), portfolio_snapshots(user, range). Everything downstream — analytics, agents, API — calls these. Per-user functions take a user context and refuse cross-user paths; agent tool implementations call the same module, so the model can never address another user's prefix. This module is also the seam for later upgrades (MotherDuck, Iceberg + Athena, ClickHouse) without touching consumers.

2.4 Lake vs Postgres (dual-homing rules)

• Lake = history and analytics: full price history, article archive, portfolio snapshot time series.
• Postgres = interactive app reads and anything mutated by app/agents: current positions, actions, chat, story records, latest ticker snapshot.
• Dual-homed tables and their authority: news_stories (Postgres authoritative, mirrored to lake for replay/history), prices (lake authoritative, Postgres holds only a latest-close snapshot per ticker), portfolio (Postgres authoritative for current state, lake holds the daily snapshot series).

3. Prices pipeline (shared)

3.1 Universe: fetch only what matters

A universe table (Postgres) drives every fetch. Rows are added automatically, tagged with reasons, and deactivated when no reason remains:

Reason sources: all users' holdings, watchlists, model-portfolio constituents, the TLH equivalence table (replacement candidates need price history before they can be recommended — easy to miss), and a static benchmark set: SPY, QQQ, DIA, IWM, VT, VXUS, AGG, BIL plus the 11 SPDR sector ETFs (XLK, XLF, XLV, XLE, XLI, XLY, XLP, XLU, XLB, XLRE, XLC). Sector ETFs double as index proxies v1. A ticker gaining its first reason triggers an automatic backfill from history_from.

Universe growth is sub-linear in users (holdings overlap heavily), which is why this layer stays shared and cheap.

3.2 Provider abstraction

MarketDataProvider interface with capabilities: daily_bars(tickers, range), corporate_actions(tickers, range), reference(tickers), optional fundamentals(tickers). Raw payloads always land in the raw zone, so switching providers means re-normalizing, not re-acquiring history.

Recommendation: bootstrap on yfinance (free, no key, fine for personal dev but fragile and ToS-gray) and move to Tiingo ($30/mo individual Power plan, verified in the cost model — solid EOD + corporate actions + news; fundamentals is an add-on) as the paid primary; Polygon and EODHD are credible alternates. As built (phases 2–8): prices stayed on yfinance through the entire v1 build — the interface held, and Tiingo remains the paid upgrade path behind MarketDataProvider, not a v1 requirement. News went to Finnhub instead (§4.1).

3.3 Curated tables

Postgres side: ticker_snapshot (ticker, lastclose, prev_close, day_change_pct, 52w high/low, updated_at), upserted nightly for the app. _As built: ticker_snapshot also carries as_of_date (the price date, distinct from updated_at), and both TickerRef and the curated tickers table gained a nullable sector (best-effort from yfinance .info, phase 6) — it feeds the news §4.2 sector clause.

Adjustment design decision: store unadjusted prices plus a nightly-rebuilt adjustment_factors table; the lake module serves adjusted series by joining on read. Rationale: new splits/dividends change all historical adjusted values — storing adjusted prices would mean rewriting history on every corporate action, while a factors rebuild is a tiny deterministic job and curated partitions stay append-only. Provider-adjusted closes are kept in raw and cross-checked as a quality gate.

3.4 Nightly flow (trading days only, ~6:30pm ET after vendor EOD settles)

1. Gate on trading_days.
2. Resolve universe → per-ticker fetch ranges from watermarks (new tickers get history_from).
3. Fetch bars → raw zone → normalize → quality gates → publish partitions.
4. Re-scan corporate actions for a trailing 30-day window (late postings) → rebuild adjustment_factors.
5. Upsert ticker_snapshot in Postgres.
6. Record run + watermarks; emit a prices-freshness fact.

Quality gates (block publish for offending tickers only, quarantining rows rather than failing the run): completeness vs universe × trading day; duplicate keys; non-positive prices or negative volume; single-day move over 40% with no corporate action; sampled adjusted-close cross-check vs provider. Everything logged to pipeline_runs with row counts. As built (phase 2): the sampled adjusted-close cross-check is not yet wired as a gate — provider adj_close is captured on every bar and kept in raw, so adding it later is additive.

Late vendor data: an automatic late pass (~9:30pm ET) re-fetches only the tickers that failed completeness or were quarantined in the main run and republishes their partitions; anything still missing after the late pass surfaces in Admin for a manual re-run, with the staleness and quarantine facts keeping downstream language honest in the meantime. As built (phase 2): the late pass is not scheduled — the run's partial status plus the pipeline_runs.error note carry the signal until it is.

Backfill is the same pipeline with an explicit date range and a per-provider rate-limit budget.

4. News pipeline (shared)

4.1 Stages

News sources

API + RSS

fetch→

raw/news/{source}

watermarked

normalize→

Normalize + dedupe

canonical URL hash

→

curated/news_articles

Embed title + snippet

pgvector

news_embeddings

cluster 72h→

Story synthesis

cheap model · house key

Postgres

news_stories

curated/news_stories

lake mirror

Sources. NewsSource interface: fetch(since) → articles {external_id, url, title, snippet/body, published_at, source_name, tickers?}. Start with one ticker-tagged API source (Tiingo News or Finnhub — pre-tagged tickers materially improve clustering and story tagging) plus an optional curated RSS list. Cadence v1: three fetches per trading day (pre-open, midday, post-close); the design supports 30-minute polling by config when chat freshness warrants it. As built (phase 6): the live source is Finnhub on the free tier (a free FINNHUB_API_KEY is the only news secret; 60/min throttle in the client), with a fake source powering all tests; no RSS list shipped. The three-fetch cadence is documented and CLI-driven — no scheduler yet.

Normalize. Strip HTML and tracking params → canonical URL; article_id = hash of canonical URL (exact-dupe key across sources); English-only filter; snippet capped at ~1,000 chars in curated (full body stays in raw where the source permits). → curated/news_articles partitioned by published date. As built (phase 6): the English filter is a dependency-free marker heuristic (positive English markers gated by non-English markers — hardened after a live DE/IT/NL leak), not langdetect; curated news_articles rows are write-once per article_id (first-writer-wins across sources and runs); raw keys embed the fetch day — provenance of when, deliberately.

Embed. Title + snippet through the gateway's embeddings endpoint (swappable via an env-driven EMBED_MODEL; curated text is retained so re-embedding under a new model is a batch job, not a re-fetch). Stored in pgvector (Supabase Postgres) with the model name and dimension recorded per row. As built (phase 6): the model is gemini/gemini-embedding-001 at 1536 dimensions (L2-normalized) on the existing free Gemini key — not the text-embedding-3-small default this doc first proposed. Article vectors live in news_embeddings (vector(1536), HNSW cosine index); the story centroid lives on news_stories with no index of its own.

Cluster into stories. Incremental greedy clustering over a 72-hour rolling window: each new article is compared to active story centroids (cosine similarity, plus a small bonus for ticker-set overlap); join the best story above ~0.82 and update its running-mean centroid, else open a new story. Story size capped. Mapping lives in Postgres (news_stories, news_story_articles). Alternative considered: nightly batch clustering (e.g., HDBSCAN) — rejected for v1 because re-clustering churns story IDs, which breaks citations from insights and chat; incremental keeps IDs stable. As built (phase 6): constants are 0.82 similarity / +0.04 ticker-overlap bonus / story cap 30, discriminated by a 30-story hand-labeled golden (the suite fails at both 0.5 and 0.95). The 72-hour window is anchored to published_at, not wall clock, so replays are deterministic. Already-clustered articles are skipped before scoring, so a double-run produces zero churn. At the size cap, a new article joins the next-best under-cap story above threshold; a sibling story opens only when none qualifies.

Synthesize (the one LLM pass). Triggered when a story's article count crosses thresholds (1, 3, 8) or its first held-ticker article arrives — not on every syndicated duplicate. Input: up to 10 title+snippet pairs favoring source diversity. Structured output, validated: headline, 3–5 summary bullets, tickers[], sectors[], event_type (earnings, guidance, M&A, macro_rates, regulatory, analyst, product, other), importance 1–5, sentiment_by_ticker. Validation failure → one retry → fallback story (top article's title, importance 2, flagged unsynthesized). Runs on the house key with the cheap model tier — this is the amortized "grouped highlights" layer.

Importance rubric (given to the model, used by consumers as filter thresholds): 5 = market-moving macro or major event on a widely held name; 4 = significant sector/single-name development; 3 = notable but routine (typical earnings, analyst moves); 2 = minor/derivative coverage; 1 = noise or listicles.

As built (phase 6), two synthesis refinements: the fallback shell applies to the first synthesis only — a re-synthesis failure preserves the last good content and advances bookkeeping only, so a story never regresses from synthesized to shell; and the first-held-ticker trigger is a single latch (synthesized_held) — a second, different held ticker arriving does not re-trigger.

4.2 Consumption contracts

• Daily agent: stories since its last run where (importance ≥ 3 and tickers intersect holdings) or (importance ≥ 4 and sectors intersect portfolio sectors) or importance = 5. Thresholds are config.
• Market recap job: top stories by importance in the last 24h.
• Chat: semantic search tool over story embeddings with date/ticker filters.
• Ticker page: stories by ticker, newest first.

Retention: raw and curated articles kept indefinitely (cheap); pgvector embeddings pruned to ~18 months (archived to lake); story records kept indefinitely (small). As built: the pruning job hasn't landed — it's a documented TODO in the stories module with the archive seam noted.

Cost envelope at ~150 articles/day: embeddings well under a cent; ~40 story syntheses/day on a cheap model ≈ $0.05–0.15/day, shared across all users.

5. Portfolio ingestion (per-user)

5.1 Connector contract

BrokerConnector.sync(account, since?) → SyncResult where the result may contain any subset of: a positions snapshot, broker-reported tax lots, transactions, cash balances, plus an as_of timestamp. Downstream normalization and reconciliation handle whatever subset arrives. v1 implements CSVImportConnector; SnapTrade/Plaid later implement the same contract and nothing downstream changes — including the phase-5 action verifier, which only ever reads normalized transactions.

5.2 CSV import flow

App upload

positions · transactions · lots CSV

→

users/{id}/raw/imports

parse→

Parse

account mapping profile

→

Stage + validate

validation report

commit→

Postgres

transactions · lots · cash

→

Reconcile

vs broker snapshot

Discrepancy → fact

app + daily agent

Nightly valuation

curated/portfolio_snapshots

• Every broker exports differently, so parsing is driven by mapping profiles: per broker + export-type configs declaring column mappings, date formats, and action-string translations (e.g. "YOU BOUGHT" → buy, "REINVESTMENT" → dividend_reinvest). Ship presets for Fidelity, Schwab, Vanguard, IBKR; support custom profiles. As built (phase 1): a single documented generic profile shipped with the profile mechanism; broker presets land once open question 2 picks the order. Unknown action strings fail the import with an explicit list — never guessed.
• Idempotency: imports bookkeeping table keyed by file SHA-256 (re-uploading a file is a no-op); transaction rows deduped by a per-account source-row hash, so overlapping export windows are safe.
• Staged-then-commit: a validation report (unparsed rows, unknown tickers, unknown actions) is shown before rows land; partial commits are not allowed.
• Upload hardening: files are capped (20 MB default) and parsing is bounded (row cap and a per-file parse timeout); anything outside the limits fails before staging with an explicit error — imports are interactive, not batch.
• Uploaded files land under users/{id}/raw/imports on R2/MinIO — Supabase Storage was considered and rejected here too (decision): one object store, one credential set, one lake module. As built (phase 1): the importer is CLI-first (brain import) and records file name + SHA-256 in imports; landing the raw file in the object store arrives with the lake module in phase 2 — the hash-keyed idempotency is already in place, so the move is additive.

5.3 Normalized model (Postgres, all rows user-scoped)

Lot logic: broker-provided lots win when available; otherwise lots are derived FIFO from transaction history. Splits arrive as split_adjust transactions — the broker feed states them directly (Plaid maps Fidelity's cash/deposit "DISTRIBUTION" rows to split_adjust with zero cash impact; a CSV SPLIT action maps the same way), with qty = the shares the split added — and FIFO derivation scales the open lots by (open + added) / open at that point: quantities restate, dollar basis and open dates (holding periods) do not. A nightly step that emits split_adjust from corporate_actions for histories whose feed carries no split rows remains future work. Dividend reinvestments create real lots — they matter for wash-sale windows.

Reconciliation policy: after each import and nightly, compare derived positions against the latest broker snapshot. Quantity mismatches create reconciliations rows surfaced in the app and as a fact to the daily agent ("MSFT: derived 100 vs broker 90 — transactions likely missing"); cost-basis mismatches adopt the broker value and log the delta. Nothing is silently auto-corrected.

Synthetic rows and the cash seed (as built, phase 2). A snapshot connector (Plaid) returns only a bounded window of history, so the ledger needs rows standing for everything before it: one opening lot per holding for the shares the window does not explain, plus the cash seed above. These are marked source = 'synthetic' and obey the opposite rule to observed broker rows — a broker row is an immutable fact, deduped by its source-row hash on re-land, while synthetic rows are derived and are replaced wholesale on every sync. Deduping them instead froze them at their first landing: an account that later gained a holding got a new opening lot with no matching seed correction, and its reported cash drifted by exactly that lot — live-reproduced on a Fidelity taxable account reporting −$16,541 against a broker-stated $19,301. The seed is computed by the repository, not the connector, because reconciling requires the whole stored ledger — the union of every window ever landed — and a connector only ever sees its own.

5.4 Snapshots and freshness

Nightly valuation (positions × EOD closes) appends to users/{id}/curated/portfolio_snapshots (trading_date, account, ticker, qty, close, value, cost_basis, unrealized gain, plus cash rows) — the time series behind performance charts and "what happened to my portfolio" deltas, without bloating Postgres. last_synced_at staleness beyond a configurable window (default 7 days) becomes a freshness fact: the daily agent nudges for a re-import and the briefing labels valuations "as of {date}".

As built (staleness fix, 2026-08): a stored snapshot row is a frozen derivation of its two inputs, so a retroactive correction to either leaves it asserting the old one. Both drift in practice — the lake's close for a past day changes (backfill, quarantine fix), or positions_current is repaired (the lot-split repair that rewrote share counts). The latter produced a live phantom -11.1% daily return: one day valued pre-repair share counts, the next valued repaired ones, and the flow-aware return read the correction as a market move. A default brain snapshot run therefore also re-derives stale days: it compares each stored row's own close and qty against the current lake and positions_current, and replays the days that disagree. Because replay rebuilds a day from CURRENT positions, it is guarded — a day preceding any recorded quantity-changing transaction is never rewritten (that difference is genuine history, not staleness); such days are logged for an explicit --from/--to replay instead. Auto-repair is additionally bounded to a 30-trading-day lookback; the explicit range replay is unguarded and unbounded, since an operator naming days is asking for exactly those. What the guard cannot repair, it discloses: an unexplained day-over-day quantity jump surviving in the series becomes a snapshot_vintage_seam data gap (doc 02 §2.8), because a period return chained across that boundary is meaningless — live-observed, a +47.7% MTD on a portfolio that moved -0.3% on the day.

6. Orchestration and operations

• Runner: one brain CLI with subcommands (prices, news, import, snapshot, daily composite), cron-scheduled v1. Every pipeline takes an explicit date range for backfills. Migration note: these jobs map one-to-one onto Dagster assets when the time comes.
• Nightly chain (trading days): trading-day gate → prices → corporate actions → factors → quality gates → ticker snapshot → lot split adjustments (§5.3) → per-user valuation snapshots → analytics engine (phase 3) → daily agent (phase 4). News runs independently on its own cadence; the daily agent reads whatever stories exist at run time.
• Bookkeeping (Postgres): pipeline_runs (pipeline, params, status, started/finished, rows in/out, error, resulting watermark). As built (phase 2): there is no separate watermarks table — watermarks derive from lake state (the max published partition date per ticker/source), which can't drift from what's actually published. Freshness facts for prices, news, and each user's portfolio flow into the fact pack so insights are honest about staleness.
• Failure policy: per-source/per-ticker isolation (one failure never kills a run); in-run retries with backoff; partial publishes only at partition granularity with quarantine tables; failures surface in an app admin view (push/email later).
• Testing: golden fixtures — canned provider payloads and sample broker CSVs covering splits, reinvestments, transfers, and partial sells — drive normalization/lot/reconciliation tests; the clustering threshold is tuned against a small hand-labeled article set; story synthesis gets an eval set of story → expected tickers/event_type.

7. Multi-user scaling notes

Nothing here changes shape with more users: the universe grows sub-linearly, news costs are flat, per-user valuation is linear but trivial. The upgrade levers, in order: cron → Dagster, MinIO → R2/S3, add Parquet compaction, pgvector → dedicated vector store only if embeddings volume demands it. All behind existing seams (lake module, connector interface, gateway).

8. Open questions

Most of these were answered by build decisions; resolutions are recorded inline.

1. Market data provider — Resolved (phases 2–8): yfinance carried the entire v1 build; Tiingo remains the paid upgrade path behind MarketDataProvider when yfinance's fragility bites. (The original Tiingo rationale partly rested on it bundling news — moot now that news is Finnhub free.)
2. Which brokers' CSVs first? — Still open. Phase 1 shipped the mapping-profile mechanism with a single documented generic profile; broker presets (Fidelity, Schwab, Vanguard, IBKR) land once this picks the order.
3. Backfill depth — Resolved by default: the 5-year default stands (history_from, universe-driven); nothing in the build needed deeper history.
4. News sources — Resolved (phase 6): Finnhub on the free tier as the one tagged API source, a fake source for all tests, no RSS list — $0.
5. Watchlist scope — Resolved (phase 6): nothing beyond holdings + benchmarks is pre-priced; user watchlists shipped (watchlist_items) and a watched ticker enters the universe with reason watchlist and gets priced automatically.