compaction: single read of deltas, in-memory slicing; gc of orphan curated files
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
4 changed files +77 −28
modified
docs/STATUS.md
+36 −2
@@ -37,5 +37,39 @@ Legend: ✅ done and verified · 🟡 done, not yet verified in production · | |||||
| 37 | - ✅ Assistant endpoint implemented (Anthropic SDK, model proposes SQL only; FinData validates/executes/cites versions) — **pending configuration** until `ANTHROPIC_API_KEY` is set and a capability/cost test is recorded here; OpenAI provider not implemented | 37 | - ✅ Assistant endpoint implemented (Anthropic SDK, model proposes SQL only; FinData validates/executes/cites versions) — **pending configuration** until `ANTHROPIC_API_KEY` is set and a capability/cost test is recorded here; OpenAI provider not implemented | ||
| 38 | - ⬜ Security review pass on production | 38 | - ⬜ Security review pass on production | ||
| 39 | 39 | ||||
| 40 | −## Production (to be filled with real outputs) | 40 | +## Production evidence (2026-11-01, all outputs real) | ||
| 41 | −(see bottom of this file after deployment) | 41 | + | ||
| 42 | +**Deployment**: `mld deploy findata --node M3U96a` → PM2 `findata-api` (0.0.0.0:8600 behind a source-network guard) + `findata-worker`; Caddy route `https://findata.spboucher.ai → 10.67.0.10:8600`; node dir adopted as git clone (`mld adopt`, HEAD = origin/main). Public checks: `/api/v1/health` 200, console `/` 200, `/login/` 200, `/api/docs` 200; headers HSTS, CSP `default-src 'self'`, `X-Frame-Options: DENY`, `X-Request-ID`. | ||||
| 43 | +Gotcha fixed during deploy: the API must not bind 127.0.0.1 only (Caddy reaches the node over WireGuard), and `--proxy-headers` must be off so the guard sees the transport peer (X-Forwarded-For is read only when the peer is the 10.67.0.1 gateway). | ||||
| 44 | + | ||||
| 45 | +**Owner login**: created (`findata create-owner`), credentials in the laptop secrets folder; `POST /auth/login` 200 → `/auth/me` owner; wrong password 401; POST without CSRF 403; logout 200. TOTP not yet enrolled (owner action in Settings). | ||||
| 46 | + | ||||
| 47 | +**Console (headless Chromium against production)**: redirect to `/login/`, login, Overview rendered at 375 px (scrollWidth 375, no overflow) and 1280 px dark; 0 console/page errors; license banner present; Bars and Datasets pages render (datasets show `unsupported`/`pending_configuration` statuses). | ||||
| 48 | + | ||||
| 49 | +**Real ingestion runs (worker logs / job results)** | ||||
| 50 | +| Job | Result | | ||||
| 51 | +|---|---| | ||||
| 52 | +| metadata | 16 313 stock + 5 100 ETF listings → 21 243 instruments (13 209 active, 8 069 `-DELISTED`); 6 131 split rows, 243 209 dividend rows | | ||||
| 53 | +| incremental stock day 1min UNADJUSTED | vendor last_update 2026-10-30; archive 15 759 934 B, 6 515 files; 1 337 408 bars published (session 2026-10-30), version 1 | | ||||
| 54 | +| derive 5m (session) | 412 938 bars | | ||||
| 55 | +| fundamentals AAPL,MSFT,NVDA,JPM,XOM | 24 790 facts (quarterly + annual), earnings events; 7 s | | ||||
| 56 | +| backfill Z (period=full, ticker_range=Z, 1min UNADJUSTED) | 57 symbols, 24 815 063 rows, 471 base files, 67 s; 10 570 rows skipped because 2026-10-30 already lives in the delta file (no duplicates); 1 556 anomalies flagged (price_jump 1 480, outside_session 130 — flagged, not altered) | | ||||
| 57 | +| lake after backfill | `market_bars.stock.1m.raw` v58: 472 files, **26 152 471 rows**, 331.5 MB; sessions 2000-01-04 → 2026-10-30; 6 521 symbols (6 515 one session + 57 Z full history) | | ||||
| 58 | +| backup #1 | pg_dump 3.0 MB zstd + rsync to M4M64a, remote file count verified (ran before the backfill; a second backup follows the compaction) | | ||||
| 59 | + | ||||
| 60 | +Lake: 506.9 GB free, curated 0.54 GB, raw 0.02 GB (full Z archive deleted after publication, sha256 kept). | ||||
| 61 | + | ||||
| 62 | +**SDK against production** (`findata_sdk`, API key scoped read:market/read:fundamentals/query/export): health ok (SMART Verified, worker running); AAPL 1m 900 rows (first bar 08:00Z = 04:00 ET open, close 351.6499) in 0.30 s; 5m 192 rows with volume conserved; `split_div` adjustment applied at query time; fundamentals as_of 2025-01-01 → 156 rows, 536 hidden (latest period 2024-09-28 filed 2024-11-01) vs now → period 2026-06-27; SQL top-volume query 0.08 s; export of 5 symbols → Parquet 4 508 rows / 99 KB in 1.15 s. | ||||
| 63 | + | ||||
| 64 | +**Load benchmark (laptop → public URL over the tunnel)** | ||||
| 65 | +| Request | n / conc | p50 | p95 | rps | | ||||
| 66 | +|---|---|---|---|---| | ||||
| 67 | +| bars 1m one session (1 000 rows) | 80 / 8 | 232 ms | 800 ms | 28.7 | | ||||
| 68 | +| bars 1m 31-day span 2019 (multi-year base file) | 20 / 4 | 1 572 ms | 1 687 ms | 2.9 | | ||||
| 69 | +| bars 5m 180-day span | 20 / 4 | 2 214 ms | 2 957 ms | 1.7 | | ||||
| 70 | +| SQL per-year aggregate on one symbol (26 M-row lake) | 6 / 2 | 1 567 ms | 1 786 ms | 1.3 | | ||||
| 71 | +| SQL count(*) over the whole lake | 3 / 1 | 985 ms | — | — | | ||||
| 72 | +| catalog (Postgres) | 100 / 16 | 101 ms | 209 ms | 135 | | ||||
| 73 | +Interpretation: file selection is by `lake_files` date range, so a 31-day 1m request scans the symbol's yearly base file (≈ 100 k rows) then paginates — acceptable; compaction of the 5m deltas (6 748 session files from the Z backfill) into base files was queued to fix the derived dataset's file count. | ||||
| 74 | + | ||||
| 75 | +**Compaction / backup #2 / restore drill / quality report**: see "Final checks" below. | ||||
modified
services/ingest/src/findata_ingest/lake.py
+14 −0
@@ -183,6 +183,20 @@ class Lake: | |||||
| 183 | shutil.move(str(src), str(dst)) | 183 | shutil.move(str(src), str(dst)) | ||
| 184 | return {"version": version, "snapshot_id": snapshot_id, "file_count": file_count, "row_count": row_count, "bytes": nbytes} | 184 | return {"version": version, "snapshot_id": snapshot_id, "file_count": file_count, "row_count": row_count, "bytes": nbytes} | ||
| 185 | 185 | ||||
| 186 | + def gc_orphans(self, older_than_hours: int = 2) -> int: | ||||
| 187 | + """Delete curated files that are not registered in lake_files (left by a killed job before publication).""" | ||||
| 188 | + from findata_core.db import connection as _conn | ||||
| 189 | + with _conn() as conn: | ||||
| 190 | + known = {r["path"] for r in conn.execute("SELECT path FROM lake_files").fetchall()} | ||||
| 191 | + cutoff = datetime.now(UTC).timestamp() - older_than_hours * 3600 | ||||
| 192 | + n = 0 | ||||
| 193 | + for p in (self.root / "curated").rglob("*.parquet"): | ||||
| 194 | + rel = str(p.relative_to(self.root)) | ||||
| 195 | + if rel not in known and p.stat().st_mtime < cutoff: | ||||
| 196 | + p.unlink() | ||||
| 197 | + n += 1 | ||||
| 198 | + return n | ||||
| 199 | + | ||||
| 186 | def gc_retired(self, older_than_days: int = 7) -> int: | 200 | def gc_retired(self, older_than_days: int = 7) -> int: | ||
| 187 | cutoff = datetime.now(UTC).timestamp() - older_than_days * 86400 | 201 | cutoff = datetime.now(UTC).timestamp() - older_than_days * 86400 | ||
| 188 | n = 0 | 202 | n = 0 | ||
modified
services/ingest/src/findata_ingest/tasks/compaction.py
+25 −25
@@ -25,33 +25,33 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 25 | if not deltas: | 25 | if not deltas: | ||
| 26 | return {"compacted_files": 0} | 26 | return {"compacted_files": 0} | ||
| 27 | paths = [str(lake.abs(d["path"])) for d in deltas] | 27 | paths = [str(lake.abs(d["path"])) for d in deltas] | ||
| 28 | − con = duckdb.connect() | 28 | + con = duckdb.connect(config={"threads": 8, "memory_limit": "16GB"}) | ||
| 29 | − con.execute("SET threads TO 8") | 29 | + # read every delta once (sorted), then slice in memory per (symbol, year) — avoids thousands of file scans | ||
| 30 | − syms = [r[0] for r in con.sql("SELECT DISTINCT symbol_as_reported FROM read_parquet($p, hive_partitioning=false) ORDER BY 1", params={"p": paths}).fetchall()] | 30 | + all_rows = con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false) ORDER BY symbol_as_reported, session_date, timestamp_utc", | ||
| 31 | − years = sorted({d["date_min"].year for d in deltas}) | 31 | + params={"p": paths}).to_arrow_table().cast(BARS_SCHEMA) | ||
| 32 | + import polars as pl | ||||
| 33 | + df = pl.from_arrow(all_rows).with_columns(pl.col("session_date").dt.year().alias("_y")) | ||||
| 32 | written, retire = [], [d["path"] for d in deltas] | 34 | written, retire = [], [d["path"] for d in deltas] | ||
| 33 | rows = 0 | 35 | rows = 0 | ||
| 34 | − for s in syms: | 36 | + groups = df.partition_by(["symbol_as_reported", "_y"], as_dict=True, maintain_order=True) | ||
| 35 | − for y in years: | 37 | + log.info("compaction.loaded", delta_files=len(deltas), rows=df.height, groups=len(groups)) | ||
| 36 | − new = con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false) WHERE symbol_as_reported=$s AND year(session_date)=$y", params={"p": paths, "s": s, "y": y}).to_arrow_table() | 38 | + for (s, y), new_df in groups.items(): | ||
| 37 | − if new.num_rows == 0: | 39 | + new = new_df.drop("_y").to_arrow().cast(BARS_SCHEMA) | ||
| 38 | − continue | 40 | + with connection() as conn: | ||
| 39 | − with connection() as conn: | 41 | + old = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='base' AND version_removed IS NULL AND symbol_min=%s AND symbol_max=%s AND date_min>=%s AND date_max<=%s", | ||
| 40 | − old = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='base' AND version_removed IS NULL AND symbol_min=%s AND symbol_max=%s AND date_min>=%s AND date_max<=%s", | 42 | + (dataset_id, s, s, f"{y}-01-01", f"{y}-12-31")).fetchall() | ||
| 41 | − (dataset_id, s, s, f"{y}-01-01", f"{y}-12-31")).fetchall() | 43 | + parts = [new] | ||
| 42 | − parts = [new.cast(BARS_SCHEMA)] | 44 | + for o in old: | ||
| 43 | − for o in old: | 45 | + parts.append(con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false)", params={"p": [str(lake.abs(o["path"]))]}).to_arrow_table().cast(BARS_SCHEMA)) | ||
| 44 | − t = con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false)", params={"p": [str(lake.abs(o["path"]))]}).to_arrow_table() | 46 | + retire.append(o["path"]) | ||
| 45 | − parts.append(t.cast(BARS_SCHEMA)) | 47 | + merged = pa.concat_tables(parts) | ||
| 46 | − retire.append(o["path"]) | 48 | + con.register("m", merged) | ||
| 47 | − merged = pa.concat_tables(parts) | 49 | + dup = con.sql("SELECT count(*) AS n FROM (SELECT instrument_id, timestamp_utc FROM m GROUP BY 1,2 HAVING count(*)>1)").fetchall()[0][0] | ||
| 48 | − con.register("m", merged) | 50 | + con.unregister("m") | ||
| 49 | − dup = con.sql("SELECT count(*) AS n FROM (SELECT instrument_id, timestamp_utc FROM m GROUP BY 1,2 HAVING count(*)>1)").fetchall()[0][0] | 51 | + if dup: | ||
| 50 | − con.unregister("m") | 52 | + raise RuntimeError(f"compaction invariant violated: {dup} duplicate (instrument_id, timestamp_utc) for {s} {y}") | ||
| 51 | − if dup: | 53 | + written.append(lake.write_table(dataset_id, merged, "base", {"symbol": s, "year": str(y)})) | ||
| 52 | − raise RuntimeError(f"compaction invariant violated: {dup} duplicate (instrument_id, timestamp_utc) for {s} {y}") | 54 | + rows += merged.num_rows | ||
| 53 | − written.append(lake.write_table(dataset_id, merged, "base", {"symbol": s, "year": str(y)})) | ||||
| 54 | − rows += merged.num_rows | ||||
| 55 | pub = lake.publish(dataset_id, written, sorted(set(retire)), job.get("run_id"), note=f"compaction {len(deltas)} delta files") | 55 | pub = lake.publish(dataset_id, written, sorted(set(retire)), job.get("run_id"), note=f"compaction {len(deltas)} delta files") | ||
| 56 | log.info("compaction.done", deltas=len(deltas), base_files=len(written), rows=rows, **pub) | 56 | log.info("compaction.done", deltas=len(deltas), base_files=len(written), rows=rows, **pub) | ||
| 57 | return {"compacted_files": len(deltas), "base_files_written": len(written), "rows": rows, **pub} | 57 | return {"compacted_files": len(deltas), "base_files_written": len(written), "rows": rows, **pub} | ||
modified
services/ingest/src/findata_ingest/tasks/quality.py
+2 −1
@@ -26,6 +26,7 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 26 | 26 | ||||
| 27 | def gc(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | 27 | def gc(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | ||
| 28 | n = lake.gc_retired(7) | 28 | n = lake.gc_retired(7) | ||
| 29 | + orphans = lake.gc_orphans(2) | ||||
| 29 | removed = 0 | 30 | removed = 0 | ||
| 30 | with transaction() as conn: | 31 | with transaction() as conn: | ||
| 31 | rows = conn.execute("UPDATE exports SET status='expired' WHERE status='ready' AND expires_at < now() RETURNING path").fetchall() | 32 | rows = conn.execute("UPDATE exports SET status='expired' WHERE status='ready' AND expires_at < now() RETURNING path").fetchall() | ||
@@ -36,4 +37,4 @@ def gc(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 36 | p.unlink() | 37 | p.unlink() | ||
| 37 | removed += 1 | 38 | removed += 1 | ||
| 38 | conn.execute("DELETE FROM sessions WHERE expires_at < now() - interval '7 days'") | 39 | conn.execute("DELETE FROM sessions WHERE expires_at < now() - interval '7 days'") | ||
| 39 | − return {"retired_files_deleted": n, "exports_expired": len(rows), "export_files_removed": removed} | 40 | + return {"retired_files_deleted": n, "orphan_files_deleted": orphans, "exports_expired": len(rows), "export_files_removed": removed} | ||
| 40 | 41 | ||||