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..7cfea72 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"; @@ -23,9 +24,16 @@ 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 }); + 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..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", () => { @@ -183,4 +187,106 @@ 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("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) => + 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("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)); + + await expect(client.getDailyTimeSeries("SPY")).rejects.toThrow( + "network down" + ); + + const series = await client.getDailyTimeSeries("SPY"); + + 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 e5ced72..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. @@ -38,14 +63,37 @@ 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. + * + * 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; } /** @@ -54,29 +102,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(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. + query: { + function: "TIME_SERIES_DAILY", + symbol, + outputsize: outputSize, + apikey: this.apiKey, + }, + }); - return dailyResponseSchema.parse(raw); + return dailyResponseSchema.parse(this.assertNoApiError(raw)); + }); } /** @@ -87,25 +131,91 @@ 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(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. + query: { + function: "TIME_SERIES_WEEKLY", + symbol, + apikey: this.apiKey, + }, + }); + + return weeklyResponseSchema.parse(this.assertNoApiError(raw)); }); + } + + /** + * Share an in-flight request: return the pending promise for `key` if one + * 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, + fetcher: () => Promise + ): Promise { + const pending = this.inFlight.get(key); + if (pending !== undefined) { + return pending; + } + const promise = this.schedule(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; + } + + /** + * 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. + * + * @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; } }