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, + }; } }