From 7156ffc81758f9747f9b0c88d395933dfcfac78d Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Sat, 18 Jul 2026 12:50:54 +0300 Subject: [PATCH 1/3] fix: share one Alpha Vantage client across alerts to avoid rate limit --- apps/alerts/src/alerts/sp500-close.test.ts | 7 +- apps/alerts/src/alerts/sp500-close.ts | 10 +- apps/alerts/src/alerts/sp500-drawdown.test.ts | 7 +- apps/alerts/src/alerts/sp500-drawdown.ts | 9 +- apps/alerts/src/alerts/types.ts | 7 ++ apps/alerts/src/scheduled.test.ts | 16 +++ apps/alerts/src/scheduled.ts | 9 +- .../alpha-vantage/src/alpha-vantage.test.ts | 60 ++++++++++++ packages/alpha-vantage/src/alpha-vantage.ts | 97 +++++++++++++------ 9 files changed, 178 insertions(+), 44 deletions(-) diff --git a/apps/alerts/src/alerts/sp500-close.test.ts b/apps/alerts/src/alerts/sp500-close.test.ts index 45dbe61..f557bb1 100644 --- a/apps/alerts/src/alerts/sp500-close.test.ts +++ b/apps/alerts/src/alerts/sp500-close.test.ts @@ -1,3 +1,4 @@ +import { AlphaVantageClient } from "@repo/alpha-vantage"; import type { DailyBar, DailyTimeSeries } from "@repo/alpha-vantage/types"; import type { Logger } from "@repo/logger"; import { beforeEach, describe, expect, it, vi } from "vitest"; @@ -68,7 +69,11 @@ const env = { ALPHA_VANTAGE_API_KEY: "test-key", }; -const run = () => sp500Close.run({ env, logger: createMockLogger() }); +const run = () => { + const logger = createMockLogger(); + const alphaVantage = new AlphaVantageClient("test-key", logger); + return sp500Close.run({ env, logger, alphaVantage }); +}; describe("sp500Close", () => { beforeEach(() => { diff --git a/apps/alerts/src/alerts/sp500-close.ts b/apps/alerts/src/alerts/sp500-close.ts index 7836a41..59ccadd 100644 --- a/apps/alerts/src/alerts/sp500-close.ts +++ b/apps/alerts/src/alerts/sp500-close.ts @@ -1,5 +1,3 @@ -import { AlphaVantageClient } from "@repo/alpha-vantage"; - import { fetchSp500, SP500_SYMBOL } from "./sp500"; import type { Alert } from "./types"; @@ -10,11 +8,11 @@ import type { Alert } from "./types"; const sp500Close: Alert = { name: "sp500-close", cron: "0 8 * * *", // reuses the existing daily 08:00 UTC trigger - run: async ({ env, logger }) => { - const client = new AlphaVantageClient(env.ALPHA_VANTAGE_API_KEY, logger); + run: async ({ logger, alphaVantage }) => { // Free daily `compact` closes + the free weekly series for the ATH baseline - // (daily `full` history is a premium feature). See `fetchSp500`. - const { bars, priorHigh } = await fetchSp500(client); + // (daily `full` history is a premium feature). See `fetchSp500`. The client + // is shared across alerts so the fetch de-duplicates within the run. + const { bars, priorHigh } = await fetchSp500(alphaVantage); if (bars.length === 0) { logger.warn("no SPY bars returned", { symbol: SP500_SYMBOL }); return null; diff --git a/apps/alerts/src/alerts/sp500-drawdown.test.ts b/apps/alerts/src/alerts/sp500-drawdown.test.ts index fe3f194..7927bae 100644 --- a/apps/alerts/src/alerts/sp500-drawdown.test.ts +++ b/apps/alerts/src/alerts/sp500-drawdown.test.ts @@ -1,3 +1,4 @@ +import { AlphaVantageClient } from "@repo/alpha-vantage"; import type { DailyBar, DailyTimeSeries } from "@repo/alpha-vantage/types"; import type { Logger } from "@repo/logger"; import { beforeEach, describe, expect, it, vi } from "vitest"; @@ -74,7 +75,11 @@ const env = { ALPHA_VANTAGE_API_KEY: "test-key", }; -const run = () => sp500Drawdown.run({ env, logger: createMockLogger() }); +const run = () => { + const logger = createMockLogger(); + const alphaVantage = new AlphaVantageClient("test-key", logger); + return sp500Drawdown.run({ env, logger, alphaVantage }); +}; describe("sp500Drawdown", () => { beforeEach(() => { diff --git a/apps/alerts/src/alerts/sp500-drawdown.ts b/apps/alerts/src/alerts/sp500-drawdown.ts index 6ad2c85..eefbfc7 100644 --- a/apps/alerts/src/alerts/sp500-drawdown.ts +++ b/apps/alerts/src/alerts/sp500-drawdown.ts @@ -1,4 +1,3 @@ -import { AlphaVantageClient } from "@repo/alpha-vantage"; import type { DailyBar } from "@repo/alpha-vantage/types"; import { fetchSp500 } from "./sp500"; @@ -67,11 +66,11 @@ const formatPercent = (fraction: number): string => const sp500Drawdown: Alert = { name: "sp500-drawdown", cron: "0 8 * * *", // reuses the existing daily 08:00 UTC trigger - run: async ({ env, logger }) => { - const client = new AlphaVantageClient(env.ALPHA_VANTAGE_API_KEY, logger); + run: async ({ logger, alphaVantage }) => { // Free daily `compact` closes + the free weekly series for the ATH baseline - // (daily `full` history is a premium feature). See `fetchSp500`. - const { bars, priorHigh } = await fetchSp500(client); + // (daily `full` history is a premium feature). See `fetchSp500`. The client + // is shared across alerts so the fetch de-duplicates within the run. + const { bars, priorHigh } = await fetchSp500(alphaVantage); if (bars.length < 2) { logger.warn("need at least 2 SPY bars for drawdown", { diff --git a/apps/alerts/src/alerts/types.ts b/apps/alerts/src/alerts/types.ts index d5721e4..5eecdf8 100644 --- a/apps/alerts/src/alerts/types.ts +++ b/apps/alerts/src/alerts/types.ts @@ -1,3 +1,4 @@ +import type { AlphaVantageClient } from "@repo/alpha-vantage"; import type { Logger } from "@repo/logger"; import type { Env } from "../types/env"; @@ -5,6 +6,12 @@ import type { Env } from "../types/env"; type AlertContext = { env: Env; logger: Logger; + /** + * Shared Alpha Vantage client for the run. Constructed once by the scheduled + * handler so alerts fetching the same series de-duplicate to one HTTP request + * (see `AlphaVantageClient`), staying under the free-tier burst limit. + */ + alphaVantage: AlphaVantageClient; }; type Alert = { diff --git a/apps/alerts/src/scheduled.test.ts b/apps/alerts/src/scheduled.test.ts index 8d74b38..778387e 100644 --- a/apps/alerts/src/scheduled.test.ts +++ b/apps/alerts/src/scheduled.test.ts @@ -27,8 +27,15 @@ const spySeries = { ], }; +// Every `new AlphaVantageClient(...)` pushes here so the test can assert the +// handler builds exactly one client and shares it across the due alerts. +const clientInstances: object[] = []; + vi.mock("@repo/alpha-vantage", () => ({ AlphaVantageClient: class { + constructor() { + clientInstances.push(this); + } getDailyTimeSeries = () => Promise.resolve(spySeries); getWeeklyTimeSeries = () => Promise.resolve(spySeries); }, @@ -52,6 +59,7 @@ describe("handleScheduled", () => { beforeEach(() => { sendMessage.mockReset(); sendMessage.mockResolvedValue({ ok: true }); + clientInstances.length = 0; }); it("sends the alert messages when the daily cron fires", async () => { @@ -71,6 +79,14 @@ describe("handleScheduled", () => { } }); + it("builds one Alpha Vantage client and shares it across the due alerts", async () => { + await handleScheduled(makeEvent("0 8 * * *"), env, ctx); + + // A single shared client means both SPY alerts de-duplicate their fetches + // instead of racing the free-tier burst limit with separate clients. + expect(clientInstances).toHaveLength(1); + }); + it("does not send anything when no alert matches the fired cron", async () => { await handleScheduled(makeEvent("*/5 * * * *"), env, ctx); diff --git a/apps/alerts/src/scheduled.ts b/apps/alerts/src/scheduled.ts index fa45c4d..a21470c 100644 --- a/apps/alerts/src/scheduled.ts +++ b/apps/alerts/src/scheduled.ts @@ -1,3 +1,4 @@ +import { AlphaVantageClient } from "@repo/alpha-vantage"; import { Logger } from "@repo/logger"; import { TelegramService } from "@repo/telegram"; @@ -14,6 +15,12 @@ const handleScheduled = async ( const parsed = parseEnv(env); const telegram = new TelegramService(parsed.TELEGRAM_BOT_TOKEN, logger); const chatId = Number(parsed.ALLOWED_CHAT_ID); + // One client for the whole run so alerts fetching the same series share a + // single request instead of racing the free-tier burst limit. + const alphaVantage = new AlphaVantageClient( + parsed.ALPHA_VANTAGE_API_KEY, + logger + ); const due = alerts.filter((alert) => alert.cron === event.cron); if (due.length === 0) { @@ -25,7 +32,7 @@ const handleScheduled = async ( const results = await Promise.allSettled( due.map(async (alert) => { - const message = await alert.run({ env: parsed, logger }); + const message = await alert.run({ env: parsed, logger, alphaVantage }); if (message === null) { logger.info("alert produced no message", { alert: alert.name }); return; diff --git a/packages/alpha-vantage/src/alpha-vantage.test.ts b/packages/alpha-vantage/src/alpha-vantage.test.ts index e9a3bfb..52f3fdb 100644 --- a/packages/alpha-vantage/src/alpha-vantage.test.ts +++ b/packages/alpha-vantage/src/alpha-vantage.test.ts @@ -183,4 +183,64 @@ describe("AlphaVantageClient", () => { ); }); }); + + describe("request de-duplication", () => { + it("collapses concurrent identical requests into a single fetch", async () => { + mockFetch.mockResolvedValue(createJsonResponse(dailySuccessBody)); + + const [a, b] = await Promise.all([ + client.getDailyTimeSeries("SPY"), + client.getDailyTimeSeries("SPY"), + ]); + + expect(mockFetch).toHaveBeenCalledTimes(1); + expect(a).toBe(b); + }); + + it("treats omitted and explicit compact output size as one request", async () => { + mockFetch.mockResolvedValue(createJsonResponse(dailySuccessBody)); + + await Promise.all([ + client.getDailyTimeSeries("SPY"), + client.getDailyTimeSeries("SPY", { outputSize: "compact" }), + ]); + + expect(mockFetch).toHaveBeenCalledTimes(1); + }); + + it("caches daily and weekly for the same symbol separately", async () => { + // A fresh Response per call (bodies are single-read) with the shape the + // called endpoint expects. + mockFetch.mockImplementation((url: unknown) => + Promise.resolve( + createJsonResponse( + String(url).includes("TIME_SERIES_WEEKLY") + ? weeklySuccessBody + : dailySuccessBody + ) + ) + ); + + await Promise.all([ + client.getDailyTimeSeries("SPY"), + client.getWeeklyTimeSeries("SPY"), + ]); + + expect(mockFetch).toHaveBeenCalledTimes(2); + }); + + it("evicts a failed request so a later identical call retries", async () => { + mockFetch.mockRejectedValueOnce(new Error("network down")); + mockFetch.mockResolvedValueOnce(createJsonResponse(dailySuccessBody)); + + await expect(client.getDailyTimeSeries("SPY")).rejects.toThrow( + "network down" + ); + + const series = await client.getDailyTimeSeries("SPY"); + + expect(series.symbol).toBe("SPY"); + expect(mockFetch).toHaveBeenCalledTimes(2); + }); + }); }); diff --git a/packages/alpha-vantage/src/alpha-vantage.ts b/packages/alpha-vantage/src/alpha-vantage.ts index e5ced72..639cf3a 100644 --- a/packages/alpha-vantage/src/alpha-vantage.ts +++ b/packages/alpha-vantage/src/alpha-vantage.ts @@ -38,10 +38,18 @@ class AlphaVantageError extends Error { * reuse it. Note: the underlying `HttpClient` applies no timeout or retry * (matching how `@repo/telegram` uses it) — add `AbortSignal.timeout` here if * that becomes necessary. + * + * Identical requests are de-duplicated for the instance's lifetime: concurrent + * and repeat calls with the same parameters share a single in-flight (or + * already-resolved) HTTP request. This keeps the free tier's 1-request-per-second + * burst limit safe when several callers (e.g. multiple alerts) fetch the same + * series in the same run. A failed request is evicted so a later call can retry. */ class AlphaVantageClient { private readonly client: HttpClient; private readonly apiKey: string; + /** In-flight/resolved requests keyed by endpoint + params (see `memoize`). */ + private readonly cache = new Map>(); constructor(apiKey: string, logger: Logger) { this.apiKey = apiKey; @@ -54,29 +62,25 @@ class AlphaVantageClient { * * @throws {AlphaVantageError} on an API-level error (bad key/symbol, rate limit). */ - async getDailyTimeSeries( + getDailyTimeSeries( symbol: string, options: DailyTimeSeriesOptions = {} ): Promise { - const raw = await this.client.get("/query", { - schema: rawResponseSchema, - // Passed as `query` (not baked into the path) so the API key stays out of logs. - query: { - function: "TIME_SERIES_DAILY", - symbol, - outputsize: options.outputSize ?? "compact", - apikey: this.apiKey, - }, - }); - - const apiError = alphaVantageErrorSchema.parse(raw); - const errorMessage = - apiError["Error Message"] ?? apiError.Note ?? apiError.Information; - if (errorMessage !== undefined) { - throw new AlphaVantageError(errorMessage); - } + const outputSize = options.outputSize ?? "compact"; + return this.memoize(`daily|${symbol}|${outputSize}`, async () => { + const raw = await this.client.get("/query", { + schema: rawResponseSchema, + // Passed as `query` (not baked into the path) so the API key stays out of logs. + query: { + function: "TIME_SERIES_DAILY", + symbol, + outputsize: outputSize, + apikey: this.apiKey, + }, + }); - return dailyResponseSchema.parse(raw); + return dailyResponseSchema.parse(this.assertNoApiError(raw)); + }); } /** @@ -87,25 +91,58 @@ class AlphaVantageClient { * * @throws {AlphaVantageError} on an API-level error (bad key/symbol, rate limit). */ - async getWeeklyTimeSeries(symbol: string): Promise { - const raw = await this.client.get("/query", { - schema: rawResponseSchema, - // Passed as `query` (not baked into the path) so the API key stays out of logs. - query: { - function: "TIME_SERIES_WEEKLY", - symbol, - apikey: this.apiKey, - }, + getWeeklyTimeSeries(symbol: string): Promise { + return this.memoize(`weekly|${symbol}`, async () => { + const raw = await this.client.get("/query", { + schema: rawResponseSchema, + // Passed as `query` (not baked into the path) so the API key stays out of logs. + query: { + function: "TIME_SERIES_WEEKLY", + symbol, + apikey: this.apiKey, + }, + }); + + return weeklyResponseSchema.parse(this.assertNoApiError(raw)); }); + } + /** + * Return the cached promise for `key`, or start `fetcher`, cache its promise + * (so concurrent callers share one HTTP request), and evict on rejection so a + * failed request isn't cached — a later identical call retries. + */ + private memoize( + key: string, + fetcher: () => Promise + ): Promise { + const cached = this.cache.get(key); + if (cached !== undefined) { + return cached; + } + + const promise = fetcher().catch((error: unknown) => { + this.cache.delete(key); + throw error; + }); + this.cache.set(key, promise); + return promise; + } + + /** + * Alpha Vantage signals errors (bad key/symbol, exhausted rate limit) in an + * HTTP 200 body. Throw on those; otherwise return the raw body for parsing. + * + * @throws {AlphaVantageError} when the body carries an API-level error. + */ + private assertNoApiError(raw: unknown): unknown { const apiError = alphaVantageErrorSchema.parse(raw); const errorMessage = apiError["Error Message"] ?? apiError.Note ?? apiError.Information; if (errorMessage !== undefined) { throw new AlphaVantageError(errorMessage); } - - return weeklyResponseSchema.parse(raw); + return raw; } } From 8c1dc31fe89c77823e93ec4a5ccd1acdbc1e419f Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Sat, 18 Jul 2026 12:58:23 +0300 Subject: [PATCH 2/3] fix: de-duplicate only in-flight requests to avoid stale reads --- .../alpha-vantage/src/alpha-vantage.test.ts | 18 +++++++- packages/alpha-vantage/src/alpha-vantage.ts | 44 +++++++++++-------- 2 files changed, 42 insertions(+), 20 deletions(-) diff --git a/packages/alpha-vantage/src/alpha-vantage.test.ts b/packages/alpha-vantage/src/alpha-vantage.test.ts index 52f3fdb..9bd28a9 100644 --- a/packages/alpha-vantage/src/alpha-vantage.test.ts +++ b/packages/alpha-vantage/src/alpha-vantage.test.ts @@ -208,7 +208,7 @@ describe("AlphaVantageClient", () => { expect(mockFetch).toHaveBeenCalledTimes(1); }); - it("caches daily and weekly for the same symbol separately", async () => { + it("de-duplicates daily and weekly concurrently as separate requests", async () => { // A fresh Response per call (bodies are single-read) with the shape the // called endpoint expects. mockFetch.mockImplementation((url: unknown) => @@ -229,7 +229,21 @@ describe("AlphaVantageClient", () => { expect(mockFetch).toHaveBeenCalledTimes(2); }); - it("evicts a failed request so a later identical call retries", async () => { + it("does not cache results — a call after the first settles refetches", async () => { + // Fresh Response per call so both sequential reads succeed. + mockFetch.mockImplementation(() => + Promise.resolve(createJsonResponse(dailySuccessBody)) + ); + + await client.getDailyTimeSeries("SPY"); + await client.getDailyTimeSeries("SPY"); + + // In-flight de-dup only: sequential calls each hit the network so the data + // stays fresh for a reused client. + expect(mockFetch).toHaveBeenCalledTimes(2); + }); + + it("does not let a failed request poison a later identical call", async () => { mockFetch.mockRejectedValueOnce(new Error("network down")); mockFetch.mockResolvedValueOnce(createJsonResponse(dailySuccessBody)); diff --git a/packages/alpha-vantage/src/alpha-vantage.ts b/packages/alpha-vantage/src/alpha-vantage.ts index 639cf3a..152e355 100644 --- a/packages/alpha-vantage/src/alpha-vantage.ts +++ b/packages/alpha-vantage/src/alpha-vantage.ts @@ -39,17 +39,18 @@ class AlphaVantageError extends Error { * (matching how `@repo/telegram` uses it) — add `AbortSignal.timeout` here if * that becomes necessary. * - * Identical requests are de-duplicated for the instance's lifetime: concurrent - * and repeat calls with the same parameters share a single in-flight (or - * already-resolved) HTTP request. This keeps the free tier's 1-request-per-second - * burst limit safe when several callers (e.g. multiple alerts) fetch the same - * series in the same run. A failed request is evicted so a later call can retry. + * Concurrent identical requests are de-duplicated: while a request with the + * same parameters is in flight, other callers share it instead of issuing their + * own. This keeps the free tier's 1-request-per-second burst limit safe when + * several callers (e.g. multiple alerts) fetch the same series in the same run. + * The entry is dropped once the request settles, so a client reused across + * cycles still fetches fresh data and a failure never poisons later calls. */ class AlphaVantageClient { private readonly client: HttpClient; private readonly apiKey: string; - /** In-flight/resolved requests keyed by endpoint + params (see `memoize`). */ - private readonly cache = new Map>(); + /** In-flight requests keyed by endpoint + params (see `memoize`). */ + private readonly inFlight = new Map>(); constructor(apiKey: string, logger: Logger) { this.apiKey = apiKey; @@ -108,24 +109,31 @@ class AlphaVantageClient { } /** - * Return the cached promise for `key`, or start `fetcher`, cache its promise - * (so concurrent callers share one HTTP request), and evict on rejection so a - * failed request isn't cached — a later identical call retries. + * Share an in-flight request: return the pending promise for `key` if one + * exists, otherwise start `fetcher` and register its promise so concurrent + * callers reuse it. The entry is removed once the request settles (success or + * failure), so this de-duplicates bursts without caching results — later calls + * fetch fresh data, and a failure never poisons a subsequent call. */ private memoize( key: string, fetcher: () => Promise ): Promise { - const cached = this.cache.get(key); - if (cached !== undefined) { - return cached; + const pending = this.inFlight.get(key); + if (pending !== undefined) { + return pending; } - const promise = fetcher().catch((error: unknown) => { - this.cache.delete(key); - throw error; - }); - this.cache.set(key, promise); + const promise = fetcher(); + this.inFlight.set(key, promise); + const evict = () => { + if (this.inFlight.get(key) === promise) { + this.inFlight.delete(key); + } + }; + // Settle in both directions; the rejection handler here keeps the eviction + // chain from surfacing as an unhandled rejection (callers await `promise`). + promise.then(evict, evict); return promise; } From 3a17ed061a6f4f547c9fcb90b0ad47f77ff07d0f Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:12:09 +0300 Subject: [PATCH 3/3] fix: address PR review feedback --- apps/alerts/src/scheduled.ts | 13 +-- .../alpha-vantage/src/alpha-vantage.test.ts | 34 ++++++- packages/alpha-vantage/src/alpha-vantage.ts | 93 ++++++++++++++++--- 3 files changed, 119 insertions(+), 21 deletions(-) diff --git a/apps/alerts/src/scheduled.ts b/apps/alerts/src/scheduled.ts index a21470c..7cfea72 100644 --- a/apps/alerts/src/scheduled.ts +++ b/apps/alerts/src/scheduled.ts @@ -15,12 +15,6 @@ const handleScheduled = async ( const parsed = parseEnv(env); const telegram = new TelegramService(parsed.TELEGRAM_BOT_TOKEN, logger); const chatId = Number(parsed.ALLOWED_CHAT_ID); - // One client for the whole run so alerts fetching the same series share a - // single request instead of racing the free-tier burst limit. - const alphaVantage = new AlphaVantageClient( - parsed.ALPHA_VANTAGE_API_KEY, - logger - ); const due = alerts.filter((alert) => alert.cron === event.cron); if (due.length === 0) { @@ -30,6 +24,13 @@ const handleScheduled = async ( logger.info("running alerts", { cron: event.cron, count: due.length }); + // One client for the whole run so alerts fetching the same series share a + // single request instead of racing the free-tier burst limit. + const alphaVantage = new AlphaVantageClient( + parsed.ALPHA_VANTAGE_API_KEY, + logger + ); + const results = await Promise.allSettled( due.map(async (alert) => { const message = await alert.run({ env: parsed, logger, alphaVantage }); diff --git a/packages/alpha-vantage/src/alpha-vantage.test.ts b/packages/alpha-vantage/src/alpha-vantage.test.ts index 9bd28a9..5c7fda7 100644 --- a/packages/alpha-vantage/src/alpha-vantage.test.ts +++ b/packages/alpha-vantage/src/alpha-vantage.test.ts @@ -76,7 +76,11 @@ describe("AlphaVantageClient", () => { beforeEach(() => { mockFetch.mockReset(); - client = new AlphaVantageClient("test-key", createMockLogger()); + // Throttling off by default so most tests stay fast; the serialization test + // opts back in with its own interval. + client = new AlphaVantageClient("test-key", createMockLogger(), { + minRequestIntervalMs: 0, + }); }); describe("getDailyTimeSeries", () => { @@ -256,5 +260,33 @@ describe("AlphaVantageClient", () => { expect(series.symbol).toBe("SPY"); expect(mockFetch).toHaveBeenCalledTimes(2); }); + + it("serializes distinct requests spaced by the configured interval", async () => { + const interval = 50; + const throttled = new AlphaVantageClient("test-key", createMockLogger(), { + minRequestIntervalMs: interval, + }); + const starts: number[] = []; + mockFetch.mockImplementation((url: unknown) => { + starts.push(Date.now()); + return Promise.resolve( + createJsonResponse( + String(url).includes("TIME_SERIES_WEEKLY") + ? weeklySuccessBody + : dailySuccessBody + ) + ); + }); + + // Fired in the same tick — without throttling both would hit the network + // simultaneously and risk the per-second burst limit. + await Promise.all([ + throttled.getDailyTimeSeries("SPY"), + throttled.getWeeklyTimeSeries("SPY"), + ]); + + expect(starts).toHaveLength(2); + expect(starts[1] - starts[0]).toBeGreaterThanOrEqual(interval - 5); + }); }); }); diff --git a/packages/alpha-vantage/src/alpha-vantage.ts b/packages/alpha-vantage/src/alpha-vantage.ts index 152e355..c5753af 100644 --- a/packages/alpha-vantage/src/alpha-vantage.ts +++ b/packages/alpha-vantage/src/alpha-vantage.ts @@ -11,9 +11,25 @@ import { const BASE_URL = "https://www.alphavantage.co"; +/** + * Minimum spacing between request starts, honoring Alpha Vantage's free-tier + * "1 request per second" burst limit with a little margin. + */ +const DEFAULT_MIN_REQUEST_INTERVAL_MS = 1100; + /** Any object; the shape is validated per-endpoint after error detection. */ const rawResponseSchema = z.record(z.string(), z.unknown()); +const delay = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * De-duplication key for a request. `JSON.stringify` keeps the parts + * unambiguous, so a symbol containing the separator can't collide with a + * different request. + */ +const keyOf = (...parts: string[]): string => JSON.stringify(parts); + type DailyTimeSeriesOptions = { /** * `compact` returns the latest ~100 data points (default); `full` returns @@ -22,6 +38,15 @@ type DailyTimeSeriesOptions = { outputSize?: "compact" | "full"; }; +type AlphaVantageClientOptions = { + /** + * Minimum milliseconds between request starts (see + * `DEFAULT_MIN_REQUEST_INTERVAL_MS`). Set to `0` to disable throttling, e.g. + * in tests. + */ + minRequestIntervalMs?: number; +}; + /** * Thrown when Alpha Vantage returns an application-level error in an HTTP 200 * body — an invalid API key, unknown symbol, or an exhausted rate limit. @@ -39,22 +64,36 @@ class AlphaVantageError extends Error { * (matching how `@repo/telegram` uses it) — add `AbortSignal.timeout` here if * that becomes necessary. * - * Concurrent identical requests are de-duplicated: while a request with the - * same parameters is in flight, other callers share it instead of issuing their - * own. This keeps the free tier's 1-request-per-second burst limit safe when - * several callers (e.g. multiple alerts) fetch the same series in the same run. - * The entry is dropped once the request settles, so a client reused across - * cycles still fetches fresh data and a failure never poisons later calls. + * The client keeps the free tier's "1 request per second" burst limit safe for + * every caller that shares it (e.g. all alerts in a run) in two ways: + * - Concurrent *identical* requests are de-duplicated: while one is in flight, + * other callers share it rather than issuing their own. The entry is dropped + * once the request settles, so a reused client still fetches fresh data and a + * failure never poisons later calls. + * - *Distinct* requests are serialized and spaced at least + * `minRequestIntervalMs` apart, so a burst of different series (e.g. daily + + * weekly) can't fire in the same instant. */ class AlphaVantageClient { private readonly client: HttpClient; private readonly apiKey: string; + private readonly minRequestIntervalMs: number; /** In-flight requests keyed by endpoint + params (see `memoize`). */ private readonly inFlight = new Map>(); + /** Tail of the serialized request queue (see `schedule`). */ + private queue: Promise = Promise.resolve(); + /** `Date.now()` when the most recent queued request started. */ + private lastRequestStart = 0; - constructor(apiKey: string, logger: Logger) { + constructor( + apiKey: string, + logger: Logger, + options: AlphaVantageClientOptions = {} + ) { this.apiKey = apiKey; this.client = new HttpClient({ logger, baseUrl: BASE_URL }); + this.minRequestIntervalMs = + options.minRequestIntervalMs ?? DEFAULT_MIN_REQUEST_INTERVAL_MS; } /** @@ -68,7 +107,7 @@ class AlphaVantageClient { options: DailyTimeSeriesOptions = {} ): Promise { const outputSize = options.outputSize ?? "compact"; - return this.memoize(`daily|${symbol}|${outputSize}`, async () => { + return this.memoize(keyOf("daily", symbol, outputSize), async () => { const raw = await this.client.get("/query", { schema: rawResponseSchema, // Passed as `query` (not baked into the path) so the API key stays out of logs. @@ -93,7 +132,7 @@ class AlphaVantageClient { * @throws {AlphaVantageError} on an API-level error (bad key/symbol, rate limit). */ getWeeklyTimeSeries(symbol: string): Promise { - return this.memoize(`weekly|${symbol}`, async () => { + return this.memoize(keyOf("weekly", symbol), async () => { const raw = await this.client.get("/query", { schema: rawResponseSchema, // Passed as `query` (not baked into the path) so the API key stays out of logs. @@ -110,10 +149,11 @@ class AlphaVantageClient { /** * Share an in-flight request: return the pending promise for `key` if one - * exists, otherwise start `fetcher` and register its promise so concurrent - * callers reuse it. The entry is removed once the request settles (success or - * failure), so this de-duplicates bursts without caching results — later calls - * fetch fresh data, and a failure never poisons a subsequent call. + * exists, otherwise start `fetcher` (via `schedule`) and register its promise + * so concurrent callers reuse it. The entry is removed once the request + * settles (success or failure), so this de-duplicates bursts without caching + * results — later calls fetch fresh data, and a failure never poisons a + * subsequent call. */ private memoize( key: string, @@ -124,7 +164,7 @@ class AlphaVantageClient { return pending; } - const promise = fetcher(); + const promise = this.schedule(fetcher); this.inFlight.set(key, promise); const evict = () => { if (this.inFlight.get(key) === promise) { @@ -137,6 +177,31 @@ class AlphaVantageClient { return promise; } + /** + * Run `fetcher` on the serial queue, starting it no sooner than + * `minRequestIntervalMs` after the previous queued request began. This keeps + * distinct requests from bursting past the free-tier per-second limit. The + * queue chain swallows settlement so one failing request can't stall the rest. + */ + private schedule( + fetcher: () => Promise + ): Promise { + const run = this.queue.then(async () => { + const wait = + this.minRequestIntervalMs - (Date.now() - this.lastRequestStart); + if (wait > 0) { + await delay(wait); + } + this.lastRequestStart = Date.now(); + return fetcher(); + }); + this.queue = run.then( + () => undefined, + () => undefined + ); + return run; + } + /** * Alpha Vantage signals errors (bad key/symbol, exhausted rate limit) in an * HTTP 200 body. Throw on those; otherwise return the raw body for parsing.