Your revenue ranking can change after a join, before any company reports new revenue.
You're preparing the account priorities for Friday. The enrichment job has finished, but the table puts annual revenue beside quarterly revenue. The SQL runs. That's the annoying part.
Financial statements describe different reporting windows, so their separate paths carry meaning. Quarterly collection also requires the annual-financials toggle. Flatten away those distinctions, and your scoring job can quietly promote the wrong accounts.
Request the statements your score actually needs
Start with a bounded watchlist. Disable unrelated collections so you can inspect the financial ingestion without carrying history and news through it.
Install apify-client, zod, and pg. Supply APIFY_TOKEN and the standard PG* connection environment variables.
Example input for the revenue workflow:
import { ApifyClient } from 'apify-client'; import { z } from 'zod'; import pg from 'pg'; const run_input = { mode: 'tickers', tickers: ['AAPL'], includeFinancials: true, includeQuarterlyFinancials: true, includeHistory: false, includeDividends: false, includeSplits: false, includeNews: false, includeRecommendations: false, includeAnalystPriceTargets: false, includeEarningsHistory: false, includeEarningsCalendar: false, }; This requests both cadences for storage. Your scoring policy will choose which one to use.
Keeping collection separate from scoring matters. You shouldn't need another collection run because someone wants to compare quarterly performance instead.
Wait before you read the dataset
Starting collection and replaying ingestion are different operations.
Set SOURCE_RUN_ID when you're repairing the database side. Leave it unset when you want a fresh source observation.
This cell waits for completion, then downloads the dataset through the REST endpoint:
if (!process.env.APIFY_TOKEN) { throw new Error('APIFY_TOKEN is required'); } const client = new ApifyClient({ token: process.env.APIFY_TOKEN, }); const started = process.env.SOURCE_RUN_ID ? { id: process.env.SOURCE_RUN_ID } : await client.actor( 'nimbryxo-studio/yahoo-finance-market-data-and-fundamentals-scraper', ).start(run_input); const run = await client.run(started.id).waitForFinish(); if (!run || run.status !== 'SUCCEEDED' || !run.defaultDatasetId) { throw new Error(`Source run is not publishable: ${run?.status}`); } const response = await fetch( `https://api.apify.com/v2/datasets/${run.defaultDatasetId}/items?format=json&clean=true`, { headers: { Authorization: `Bearer ${process.env.APIFY_TOKEN}`, }, }, ); if (!response.ok) { throw new Error(`Dataset download failed: ${response.status}`); } const items = z.array(z.unknown()).parse(await response.json()); The full download keeps this recipe straightforward. Its cost is memory proportional to the dataset; a large watchlist deserves paginated ingestion with the same database contract.
A successful run is permission to inspect its records. It isn't proof that every requested account has usable financials.
What belongs in the statement key?
A financial statement key needs ticker, cadence, statement kind, and fiscal period.
The period end alone can't distinguish an annual income statement from a quarterly income statement ending on that date.
Carry the source structure into explicit columns:
| Source | Stored distinction |
|---|---|
ticker | Resolved source identity |
financials.income, .balance, .cashflow | Annual cadence and statement kind |
financials.quarterly.income, .balance, .cashflow | Quarterly cadence and statement kind |
| Fiscal-period map key | Period end |
financials.currency | Reporting currency |
The flattening operation should preserve this split:
ticker annual / income / fiscal period -> statement quarterly / income / fiscal period -> statement current key: ticker + cadence + kind + period Use the returned ticker, rather than the input spelling. An alias can resolve to a different symbol, with requestedTicker retaining the original request.
That key is stable enough for refreshing source statements. It isn't a permanent company identifier. For account joins across symbol changes, corporate actions, or multiple listings, maintain a separately reviewed issuer mapping.
Currency stays outside the key deliberately. A changed reporting currency should revise the same logical period and raise a review issue, rather than silently create another statement identity.
Accept added fields, quarantine broken meanings
Optional data is omitted when the source doesn't publish it. Absence must stay distinct from zero revenue.
Use passthrough schemas because a new, unrelated field shouldn't stop account enrichment. Validate the structure you consume, then preserve statement labels in JSONB.
This cell quarantines malformed items and flattens usable periods:
const Statement = z.object({ Total_Revenue: z.unknown().optional(), }).passthrough(); const PeriodMap = z.record(z.string(), Statement); const StatementSet = z.object({ income: PeriodMap.optional(), balance: PeriodMap.optional(), cashflow: PeriodMap.optional(), }).passthrough(); const TickerItem = z.object({ recordType: z.literal('ticker'), ticker: z.string().trim().min(1), requestedTicker: z.string().optional(), scrapedAt: z.string().datetime({ offset: true }), financials: StatementSet.extend({ currency: z.string().trim().min(1).optional(), quarterly: StatementSet.optional(), periods: z.unknown().optional(), }).optional(), }).passthrough(); const ErrorItem = z.object({ recordType: z.literal('error'), }).passthrough(); type StatementRow = { item_index: number; ticker: string; cadence: 'annual' | 'quarterly'; kind: 'income' | 'balance' | 'cashflow'; period: string; reporting_currency: string; scraped_at: string; revenue: string | null; raw: Record<string, unknown>; }; type Issue = { item_index: number; code: string; detail: string; raw: unknown; }; const rows: StatementRow[] = []; const issues: Issue[] = []; const seen = new Set<string>(); function flag( item_index: number, code: string, detail: string, raw: unknown, ) { issues.push({ item_index, code, detail, raw }); } function validPeriod(value: string): boolean { return /^\d{4}-\d{2}-\d{2}$/.test(value) && Number.isFinite(Date.parse(value)) && new Date(value).toISOString().slice(0, 10) === value; } function decimal(value: unknown): string | null { if (typeof value === 'number' && Number.isSafeInteger(value)) { return String(value); } return typeof value === 'string' && /^-?\d+(?:\.\d+)?$/.test(value) ? value : null; } items.forEach((raw, item_index) => { const parsed = TickerItem.safeParse(raw); if (!parsed.success) { flag( item_index, ErrorItem.safeParse(raw).success ? 'source_error' : 'schema_rejected', JSON.stringify(parsed.error.issues), raw, ); return; } const item = parsed.data; seen.add(item.ticker); if (item.requestedTicker) seen.add(item.requestedTicker); const financials = item.financials; if (!financials?.currency) { flag(item_index, 'missing_financials_or_currency', item.ticker, raw); return; } const cadences = [ ['annual', financials], ['quarterly', financials.quarterly], ] as const; for (const [cadence, statements] of cadences) { for (const kind of ['income', 'balance', 'cashflow'] as const) { const periods = statements?.[kind]; if (!periods || Object.keys(periods).length === 0) { flag(item_index, 'missing_statement', `${cadence}/${kind}`, raw); continue; } for (const [period, statement] of Object.entries(periods)) { const detail = JSON.stringify([item.ticker, cadence, kind, period]); if (!validPeriod(period)) { flag(item_index, 'unusable_period', detail, statement); continue; } const revenue = kind === 'income' ? decimal(statement.Total_Revenue) : null; if (kind === 'income' && revenue === null) { flag(item_index, 'unusable_revenue', detail, statement); } rows.push({ item_index, ticker: item.ticker, cadence, kind, period, reporting_currency: financials.currency, scraped_at: item.scrapedAt, revenue, raw: statement, }); } } } }); for (const ticker of run_input.tickers) { if (!seen.has(ticker)) { flag(-1, 'missing_ticker', ticker, null); } } The revenue parser is intentionally conservative. Safe integer numbers and plain decimal strings become database numeric values. Formatted strings, fractional JSON numbers, and unsafe integers require review.
We'd rather leave the score absent than repair a financial amount by guessing.
A broken financial container quarantines its whole item. An unusable period label excludes that period. An unusable revenue value still preserves the statement, but leaves its typed revenue empty.
Invalid or delisted symbols produce error records instead of partial ticker records. Those records enter quarantine too, with their original payload available for diagnosis.
Strict validation would also reject harmless additions. Passthrough accepts those additions while still detecting changes that break this workflow's financial meaning.
Keep observations beside the current row
Typed columns make ranking predictable. Raw JSONB preserves statement labels and gives you somewhere to inspect revisions without adding columns for every accounting line.
They serve different jobs. JSONB alone makes every dashboard its own parser; typed columns alone discard useful evidence.
This schema keeps an immutable observation per source item and a refreshable current statement:
const statementColumns = ` run_id text NOT NULL, item_index integer NOT NULL, ticker text NOT NULL, cadence text NOT NULL, kind text NOT NULL, period date NOT NULL, reporting_currency text NOT NULL, scraped_at timestamptz NOT NULL, revenue numeric, raw jsonb NOT NULL `; const setupSql = ` CREATE TABLE IF NOT EXISTS financial_observation ( ${statementColumns}, PRIMARY KEY (run_id, item_index, ticker, cadence, kind, period) ); CREATE TABLE IF NOT EXISTS financial_current ( ${statementColumns}, PRIMARY KEY (ticker, cadence, kind, period) ); CREATE TABLE IF NOT EXISTS ingestion_issue ( run_id text NOT NULL, item_index integer NOT NULL, code text NOT NULL, detail text NOT NULL, raw jsonb NOT NULL, PRIMARY KEY (run_id, item_index, code, detail) ); `; Replaying the same completed dataset makes the observation inserts no-ops. A fresh collection run creates a separate observation, even when its values agree with the previous run.
Restatements belong in that history. Updating the current row without keeping observations would erase the distinction between “collected again” and “reported differently.”
The writer below batches inserts and upserts. Its transaction lock serializes this ingestion path so currency comparisons see a consistent predecessor.
type RankingPolicy = { cadence: 'annual' | 'quarterly'; currency: string; asOf: string; maxAgeDays: number; }; async function ingest(rankingSql: string, policy: RankingPolicy) { const pool = new pg.Pool(); try { const db = await pool.connect(); try { await db.query('BEGIN'); await db.query('SELECT pg_advisory_xact_lock(4101)'); await db.query(setupSql); await db.query(` INSERT INTO ingestion_issue SELECT $1::text, x.* FROM jsonb_to_recordset($2::jsonb) AS x( item_index integer, code text, detail text, raw jsonb ) ON CONFLICT DO NOTHING `, [run.id, JSON.stringify(issues)]); await db.query(` WITH saved AS ( INSERT INTO financial_observation SELECT $1::text, x.* FROM jsonb_to_recordset($2::jsonb) AS x( item_index integer, ticker text, cadence text, kind text, period date, reporting_currency text, scraped_at timestamptz, revenue numeric, raw jsonb ) ON CONFLICT DO NOTHING RETURNING * ), candidates AS ( SELECT DISTINCT ON (ticker, cadence, kind, period) * FROM saved ORDER BY ticker, cadence, kind, period, scraped_at DESC, run_id DESC, item_index DESC ), drift AS ( INSERT INTO ingestion_issue SELECT n.run_id, n.item_index, CASE WHEN n.reporting_currency <> c.reporting_currency THEN 'currency_changed' ELSE 'timestamp_conflict' END, jsonb_build_array(n.ticker, n.cadence, n.kind, n.period)::text, jsonb_build_object('previous', to_jsonb(c), 'incoming', to_jsonb(n)) FROM candidates n JOIN financial_current c USING (ticker, cadence, kind, period) WHERE ( n.reporting_currency <> c.reporting_currency AND (n.scraped_at, n.run_id, n.item_index) > (c.scraped_at, c.run_id, c.item_index) ) OR ( n.scraped_at = c.scraped_at AND n.raw IS DISTINCT FROM c.raw ) ON CONFLICT DO NOTHING ) INSERT INTO financial_current AS c SELECT * FROM candidates WHERE true ON CONFLICT (ticker, cadence, kind, period) DO UPDATE SET run_id = EXCLUDED.run_id, item_index = EXCLUDED.item_index, reporting_currency = EXCLUDED.reporting_currency, scraped_at = EXCLUDED.scraped_at, revenue = EXCLUDED.revenue, raw = EXCLUDED.raw WHERE (EXCLUDED.scraped_at, EXCLUDED.run_id, EXCLUDED.item_index) > (c.scraped_at, c.run_id, c.item_index) `, [run.id, JSON.stringify(rows)]); await db.query('COMMIT'); } catch (error) { await db.query('ROLLBACK'); throw error; } finally { db.release(); } const ranking = await pool.query(rankingSql, [ policy.cadence, run.id, policy.asOf, policy.currency, policy.maxAgeDays, ]); const alerts = await pool.query(` SELECT code, count(*) AS affected FROM ingestion_issue WHERE run_id = $1 GROUP BY code `, [run.id]); return { ranking: ranking.rows, alerts: alerts.rows }; } finally { await pool.end(); } } An older replay can't overwrite a newer source timestamp. Equal timestamps use run identity and item position as deterministic tie-breakers; conflicting stored statement content also creates a review issue. That ordering is bookkeeping, not evidence of which statement is financially authoritative.
The lock trades throughput for simpler correctness. At higher volume, partition the writer or introduce coordinated locks by statement key.
A database failure rolls back observations, issues, and current rows together. A crash after commit is safe to replay. Quarantined records don't erase previously accepted statements.
The worked source here's Yahoo Finance Market Data & Fundamentals Scraper. The storage contract belongs to your scoring system.
Rank a defined cohort, not every revenue value
Revenue measures activity over a period. A balance sheet measures a position at a date. Neither the statement kind nor cadence belongs in an optional dashboard filter.
Choose the latest period before checking its currency and revenue. Otherwise, an unusable recent statement can silently send the query searching backward for a more convenient answer.
Example scoring policy: annual income statements, a single reporting currency, and an explicit freshness window:
const rankingSql = ` WITH latest AS ( SELECT DISTINCT ON (ticker) ticker, period, reporting_currency, revenue FROM financial_current WHERE cadence = $1 AND kind = 'income' AND run_id = $2 AND period <= $3::date ORDER BY ticker, period DESC ) SELECT ticker, period, reporting_currency, revenue, dense_rank() OVER (ORDER BY revenue DESC) AS revenue_rank FROM latest WHERE reporting_currency = $4 AND revenue IS NOT NULL AND period >= $3::date - $5::integer ORDER BY revenue DESC, ticker `; const ingestion = await ingest(rankingSql, { cadence: 'annual', currency: 'USD', asOf: new Date().toISOString().slice(0, 10), maxAgeDays: 500, }); The provenance filter excludes accounts whose current statements weren't accepted from this source run. That makes partial ingestion visible as missing coverage instead of quietly recycling old enrichment.
This policy excludes other currencies. It doesn't convert them. Cross-currency ranking needs an explicit exchange-rate source, conversion convention, and rate date.
Annual cadence still doesn't prove identical duration or economic conditions. Fiscal year ends differ, and fiscal-calendar changes can create unusual reporting windows. Period-end keys alone can't establish duration; review those issuer-period exceptions before treating the ranking as an accounting comparison.
