From 651415d99b2ea4503b066b70113cad168b3e69bc Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 9 Sep 2026 08:24:40 +0000 Subject: [PATCH] fix(adv): bound the ingest folder and keep the newest filing per CRD MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `IngestAdvSnapshotTask` joined `SEC_RAW_DATA_FOLDER` with an unvalidated `folder`, and supplying it skipped `advArchiveFolder`'s check on `snapshot` too. The task is registered, so both inputs arrive from `sec-base task run`, the console form and the MCP tool surface: any local `*.csv` could be read into `adv_row`, and a chosen snapshot label wiped a real one through the `deleteSearch` that runs first. It now resolves and prefix-checks the directory the way `BootstrapDownloadTask` does, and validates `snapshot` whether or not `folder` overrides where the archive is read from. `adv_adviser` is keyed `(snapshot, crd_number)` while the base member is one row per filing, so the cumulative archive's thirteen years of amendments all collided on one key and the survivor was decided by CSV position rather than by `date_submitted` — leaving `sec show advisers --min-aum` filtering on an arbitrary historical AUM. Base rows are now reduced to one per CRD, newest `date_submitted` winning with `filing_id` as tie-break and an undated filing never displacing a dated one, and the reported `advisers:` count is what landed rather than what was pushed. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01T8DU24G9GWuRUJncc5TJFZ --- src/task/adv/IngestAdvSnapshotTask.test.ts | 90 +++++++++++++++++++++- src/task/adv/IngestAdvSnapshotTask.ts | 86 +++++++++++++++++---- 2 files changed, 161 insertions(+), 15 deletions(-) diff --git a/src/task/adv/IngestAdvSnapshotTask.test.ts b/src/task/adv/IngestAdvSnapshotTask.test.ts index f2d03b39..84c37c82 100644 --- a/src/task/adv/IngestAdvSnapshotTask.test.ts +++ b/src/task/adv/IngestAdvSnapshotTask.test.ts @@ -6,7 +6,7 @@ import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { join, relative } from "node:path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { globalServiceRegistry } from "workglow"; import { SEC_RAW_DATA_FOLDER } from "../../config/tokens"; @@ -144,6 +144,94 @@ describe("IngestAdvSnapshotTask", () => { expect((await rows.query({ snapshot: "2026-06" })) ?? []).toHaveLength(3); }); + it("keeps the newest base filing per CRD, not whichever the file listed last", async () => { + // One adviser, two of its annual amendments, newest FIRST — so a pass that + // takes the last row it sees keeps the 2013 description of a firm that has + // filed every year since. The cumulative archive stamps thirteen years of + // filings with one snapshot label, so every one of them collides here. + extract("2011-2024", { + "IA_ADV_Base_A.csv": [ + "FilingID,DateSubmitted,1A,1B1,1D,1E1,1F1-City,1F1-State,1F1-Country,5F2c", + '900002,6/30/2024,"Acme Capital Management, LP",Acme Capital,801-12345,110001,Boston,MA,United States,"3,000,000,000"', + '900001,4/1/2013,Acme Capital LLC,Acme,801-12345,110001,Providence,RI,United States,"90,000,000"', + ].join("\n"), + }); + + const out = await new IngestAdvSnapshotTask().run({ snapshot: "2011-2024" }); + + const advisers = globalServiceRegistry.get(ADV_ADVISER_REPOSITORY_TOKEN); + const landed = (await advisers.query({ snapshot: "2011-2024" })) ?? []; + expect(landed).toHaveLength(1); + // The reported count is rows that landed, not rows pushed at the key. + expect(out.advisers).toBe(1); + + const acme = landed[0]!; + expect(acme.date_submitted).toBe("2024-06-30"); + expect(acme.filing_id).toBe("900002"); + expect(acme.regulatory_aum).toBe(3_000_000_000); + expect(acme.main_office_state).toBe("MA"); + }); + + it("orders undated filings behind dated ones rather than by position", async () => { + extract("2011-2024", { + "IA_ADV_Base_A.csv": [ + "FilingID,DateSubmitted,1A,1B1,1D,1E1,1F1-City,1F1-State,1F1-Country,5F2c", + '900010,5/1/2020,Gamma Advisers LLC,Gamma,801-22222,110003,Denver,CO,United States,"5,000,000"', + "900011,,Gamma Advisers LLC,Gamma,801-22222,110003,Nowhere,ZZ,United States,", + ].join("\n"), + }); + + await new IngestAdvSnapshotTask().run({ snapshot: "2011-2024" }); + + const advisers = globalServiceRegistry.get(ADV_ADVISER_REPOSITORY_TOKEN); + const gamma = await advisers.get({ snapshot: "2011-2024", crd_number: "110003" }); + expect(gamma?.date_submitted).toBe("2020-05-01"); + expect(gamma?.main_office_state).toBe("CO"); + }); + + it("refuses a folder that resolves outside SEC_RAW_DATA_FOLDER", async () => { + const outside = mkdtempSync(join(tmpdir(), "sec-adv-outside-")); + writeFileSync(join(outside, "Private.csv"), "secret\nvalue\n"); + try { + // The same shape `BootstrapDownloadTask` guards: the task is registered, + // so this input arrives from `sec-base task run`, the console form and + // the MCP tool surface, not only from the sync leaf. + await expect( + new IngestAdvSnapshotTask().run({ + snapshot: "2026-06", + folder: relative(dir, outside), + }) + ).rejects.toThrow(/SEC_RAW_DATA_FOLDER/); + + const rows = globalServiceRegistry.get(ADV_ROW_REPOSITORY_TOKEN); + expect((await rows.getAll()) ?? []).toHaveLength(0); + } finally { + rmSync(outside, { recursive: true, force: true }); + } + }); + + it("still validates the snapshot label when folder is given", async () => { + // `snapshot` is stamped on every row AND names the rows dropped before the + // read, so a label that never passed `advArchiveFolder` can wipe a real one. + await expect( + new IngestAdvSnapshotTask().run({ + snapshot: "not-a-period", + folder: advArchiveFolder("2026-06"), + }) + ).rejects.toThrow(/YYYY-MM/); + }); + + it("does not drop another snapshot's rows when the read is refused", async () => { + await new IngestAdvSnapshotTask().run({ snapshot: "2026-06" }); + + await expect( + new IngestAdvSnapshotTask().run({ snapshot: "2026-06", folder: "../.." }) + ).rejects.toThrow(/SEC_RAW_DATA_FOLDER/); + + const rows = globalServiceRegistry.get(ADV_ROW_REPOSITORY_TOKEN); + expect((await rows.query({ snapshot: "2026-06" })) ?? []).toHaveLength(3); + }); + it("says what to run when the archive was never downloaded", async () => { await expect(new IngestAdvSnapshotTask().run({ snapshot: "2026-07" })).rejects.toThrow( /sec update adv --period 2026-07/ diff --git a/src/task/adv/IngestAdvSnapshotTask.ts b/src/task/adv/IngestAdvSnapshotTask.ts index 99e0edad..5f9b37b8 100644 --- a/src/task/adv/IngestAdvSnapshotTask.ts +++ b/src/task/adv/IngestAdvSnapshotTask.ts @@ -5,7 +5,7 @@ */ import { readdir, readFile } from "node:fs/promises"; -import { extname, join } from "node:path"; +import { extname, join, resolve, sep } from "node:path"; import { Type } from "typebox"; import type { IExecuteContext } from "workglow"; import { globalServiceRegistry, Task, TaskAbortedError } from "workglow"; @@ -89,6 +89,40 @@ function toNumber(value: string | undefined): number | null { return Number.isFinite(parsed) ? parsed : null; } +/** + * Orders two `filing_id`s, ascending, with a missing one lowest. + * + * Numerically where both parse, because ADV writes the id unpadded and a + * lexical comparison puts `900010` below `900002`. + */ +function compareFilingId(a: string | null, b: string | null): number { + if (a === b) return 0; + if (a === null) return -1; + if (b === null) return 1; + const na = Number(a); + const nb = Number(b); + if (Number.isFinite(na) && Number.isFinite(nb) && na !== nb) return na < nb ? -1 : 1; + return a < b ? -1 : 1; +} + +/** + * Whether `candidate` is the later of two base filings for one CRD. + * + * `date_submitted` decides, and `filing_id` breaks a tie. An undated filing + * loses to a dated one: it cannot be ordered against it, and a header the + * archive left undated must not displace one that says when it was filed. + */ +function isLaterFiling(candidate: AdvAdviser, current: AdvAdviser): boolean { + const a = candidate.date_submitted; + const b = current.date_submitted; + if (a !== b) { + if (a === null) return false; + if (b === null) return true; + return a > b; + } + return compareFilingId(candidate.filing_id, current.filing_id) > 0; +} + /** ISO date, or null — ADV dates arrive in several spellings across archives. */ function toIsoDate(value: string | undefined): string | null { if (value === undefined) return null; @@ -102,7 +136,9 @@ function toIsoDate(value: string | undefined): string | null { /** * Lands one extracted Form ADV archive: every member as `adv_row`, and the - * base-filing member additionally as typed `adv_adviser` rows. + * base-filing member additionally as typed `adv_adviser` rows — one per CRD, + * carrying that adviser's latest filing in the snapshot. Per-filing history + * stays in `adv_row`, which keeps every base row verbatim. * * Nothing here knows the SEC's column set. The two headline tables people * filter on get columns; everything else stays queryable as JSON. A member the @@ -145,8 +181,17 @@ export class IngestAdvSnapshotTask extends Task< context: IExecuteContext ): Promise> { const root = globalServiceRegistry.get(SEC_RAW_DATA_FOLDER); - const folder = input.folder ?? advArchiveFolder(input.snapshot); - const dir = join(root, folder); + // Unconditionally, so `folder` cannot smuggle past it: `snapshot` is + // stamped on every row and names the rows dropped before the read. + const snapshotFolder = advArchiveFolder(input.snapshot); + const folder = input.folder ?? snapshotFolder; + const dir = resolve(root, folder); + const safeBase = resolve(root) + sep; + if (!dir.startsWith(safeBase)) { + throw new Error( + `Invalid folder "${folder}": must resolve to a subdirectory of SEC_RAW_DATA_FOLDER` + ); + } const members = (await readMembers(dir)).sort(); if (members.length === 0) { @@ -168,7 +213,11 @@ export class IngestAdvSnapshotTask extends Task< } let rowTotal = 0; - let adviserTotal = 0; + // Keyed by CRD because `adv_adviser` is, and the base member is one row per + // FILING: the cumulative archive carries every amendment an adviser filed + // between 2011 and 2024 under one snapshot label. Resolved here rather than + // left to the write, whose last-write-wins is CSV position. + const advisers = new Map(); for (const [position, member] of members.entries()) { if (context.signal?.aborted) throw new TaskAbortedError(); @@ -197,11 +246,10 @@ export class IngestAdvSnapshotTask extends Task< if (!isBaseFilingMember(table)) continue; const era = isEraMember(table); - const advisers: AdvAdviser[] = []; for (const row of rows) { const crd = advField(row, "1E1", "CRD Number", "crd_number"); if (crd === undefined) continue; - advisers.push({ + const adviser: AdvAdviser = { snapshot: input.snapshot, crd_number: crd, sec_file_number: advField(row, "1D", "SEC File Number") ?? null, @@ -214,16 +262,26 @@ export class IngestAdvSnapshotTask extends Task< regulatory_aum: toNumber(advField(row, "5F2c", "5F(2)(c)")), filing_id: advField(row, "FilingID") ?? null, date_submitted: toIsoDate(advField(row, "DateSubmitted", "Date Submitted")), - }); + }; + const held = advisers.get(crd); + if (held === undefined || isLaterFiling(adviser, held)) advisers.set(crd, adviser); } - if (!dryRun) { - for (let i = 0; i < advisers.length; i += WRITE_BATCH) { - await adviserRepo.putBulk(advisers.slice(i, i + WRITE_BATCH)); - } + } + + // After every member, because the IA and ERA base members are one adviser + // set split in two and either can hold the later filing for a CRD. + const landedAdvisers = [...advisers.values()]; + if (!dryRun) { + for (let i = 0; i < landedAdvisers.length; i += WRITE_BATCH) { + await adviserRepo.putBulk(landedAdvisers.slice(i, i + WRITE_BATCH)); } - adviserTotal += advisers.length; } - return { success: true, tables: members.length, rows: rowTotal, advisers: adviserTotal }; + return { + success: true, + tables: members.length, + rows: rowTotal, + advisers: landedAdvisers.length, + }; } }