From 74dea0511346b38a9a88488d3fe63132d0a74917 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Sun, 8 Nov 2026 14:20:05 +0300 Subject: [PATCH] feat(market-index): store MOEX index closes with confirmed coverage (WealthLedger/main#54) --- .../migration.sql | 15 +++ prisma/schema.prisma | 18 +++ src/integrations/moex/moex.client.spec.ts | 67 ++++++++++ src/integrations/moex/moex.client.ts | 62 +++++++++ .../market-index/market-index.module.ts | 13 ++ .../market-index/market-index.repository.ts | 52 ++++++++ .../market-index/market-index.service.spec.ts | 121 ++++++++++++++++++ .../market-index/market-index.service.ts | 115 +++++++++++++++++ 8 files changed, 463 insertions(+) create mode 100644 prisma/migrations/20261107214500_add_market_index_history/migration.sql create mode 100644 src/modules/investments/market-index/market-index.module.ts create mode 100644 src/modules/investments/market-index/market-index.repository.ts create mode 100644 src/modules/investments/market-index/market-index.service.spec.ts create mode 100644 src/modules/investments/market-index/market-index.service.ts diff --git a/prisma/migrations/20261107214500_add_market_index_history/migration.sql b/prisma/migrations/20261107214500_add_market_index_history/migration.sql new file mode 100644 index 0000000..daa1001 --- /dev/null +++ b/prisma/migrations/20261107214500_add_market_index_history/migration.sql @@ -0,0 +1,15 @@ +CREATE TABLE "MarketIndexDailyClose" ( + "secid" TEXT NOT NULL, + "tradeDate" TEXT NOT NULL, + "close" TEXT NOT NULL, + "createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT "MarketIndexDailyClose_pkey" PRIMARY KEY ("secid","tradeDate") +); + +CREATE TABLE "MarketIndexSyncState" ( + "secid" TEXT NOT NULL PRIMARY KEY, + "lastAttemptAt" DATETIME, + "lastSuccessAt" DATETIME, + "coveredFrom" TEXT, + "coveredTo" TEXT +); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4bcbf80..7fc478e 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -8,6 +8,24 @@ datasource db { provider = "sqlite" } +// Public MOEX index closes contain no user holdings or credentials. +model MarketIndexDailyClose { + secid String + tradeDate String + close String + createdAt DateTime @default(now()) + + @@id([secid, tradeDate]) +} + +model MarketIndexSyncState { + secid String @id + lastAttemptAt DateTime? + lastSuccessAt DateTime? + coveredFrom String? + coveredTo String? +} + model User { id String @id email String @unique diff --git a/src/integrations/moex/moex.client.spec.ts b/src/integrations/moex/moex.client.spec.ts index 6da226b..e9a6dd1 100644 --- a/src/integrations/moex/moex.client.spec.ts +++ b/src/integrations/moex/moex.client.spec.ts @@ -8,6 +8,73 @@ const emptyEvents = { describe('MoexClient', () => { afterEach(() => jest.restoreAllMocks()); + it('loads all pages of index closes without dropping a page', async () => { + const fetchMock = jest.spyOn(global, 'fetch').mockImplementation((input) => { + const start = Number(new URL(input as string).searchParams.get('start')); + return Promise.resolve({ + ok: true, + status: 200, + json: () => Promise.resolve({ + 'history': { + columns: ['TRADEDATE', 'CLOSE'], + data: start === 0 ? [['2026-10-01', 5752.22]] : [['2026-10-02', 5745.61]], + }, + 'history.cursor': { + columns: ['INDEX', 'TOTAL', 'PAGESIZE'], + data: [[start, 2, 1]], + }, + }), + } as Response); + }); + expect(await new MoexClient().indexHistory('MCFTRR', '2026-10-01', '2026-10-02')).toEqual([ + { tradeDate: '2026-10-01', close: '5752.22' }, + { tradeDate: '2026-10-02', close: '5745.61' }, + ]); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it('rejects a later malformed page instead of returning partial index history', async () => { + jest.spyOn(global, 'fetch').mockImplementation((input) => { + const start = Number(new URL(input as string).searchParams.get('start')); + return Promise.resolve({ + ok: true, + status: 200, + json: () => Promise.resolve({ + 'history': { + columns: ['TRADEDATE', 'CLOSE'], + data: [[start ? '2026-10-02' : '2026-10-01', start ? null : 100]], + }, + 'history.cursor': { columns: ['INDEX', 'TOTAL', 'PAGESIZE'], data: [[start, 2, 1]] }, + }), + } as Response); + }); + await expect(new MoexClient().indexHistory('IMOEX', '2026-10-01', '2026-10-02')) + .rejects.toThrow(); + }); + + it('rejects an unsuccessful index history response', async () => { + jest.spyOn(global, 'fetch').mockResolvedValue({ ok: false, status: 503 } as Response); + await expect(new MoexClient().indexHistory('MCFTRR', '2026-10-01', '2026-10-02')) + .rejects.toThrow(); + }); + + it('rejects a timed-out index history request', async () => { + jest.spyOn(global, 'fetch').mockRejectedValue(new DOMException('Timed out', 'TimeoutError')); + await expect(new MoexClient().indexHistory('IMOEX', '2026-10-01', '2026-10-02')) + .rejects.toThrow(); + }); + + it('rejects duplicate or out-of-order trading dates', async () => { + jest.spyOn(global, 'fetch').mockResolvedValue({ + ok: true, status: 200, + json: () => Promise.resolve({ + 'history': { columns: ['TRADEDATE', 'CLOSE'], data: [['2026-10-02', 100], ['2026-10-01', 101]] }, + 'history.cursor': { columns: ['INDEX', 'TOTAL', 'PAGESIZE'], data: [[0, 2, 100]] }, + }), + } as Response); + await expect(new MoexClient().indexHistory('IMOEX', '2026-10-01', '2026-10-02')) + .rejects.toThrow(); + }); it('uses effective yield only for the exact ISIN and board and preserves its own price and date', async () => { jest.spyOn(global, 'fetch').mockResolvedValue({ ok: true, diff --git a/src/integrations/moex/moex.client.ts b/src/integrations/moex/moex.client.ts index df037e0..742285d 100644 --- a/src/integrations/moex/moex.client.ts +++ b/src/integrations/moex/moex.client.ts @@ -10,6 +10,13 @@ const EMPTY_PAGE_SIZE = 0; const MIN_CURSOR_VALUE = 0; const FALLBACK_PAGE_SIZE = 100; +const INDEX_COLUMNS = ['TRADEDATE', 'CLOSE']; +const ISO_DATE_LENGTH = 10; +const ZERO_CLOSE = 0; +const ISO_DATE_START = 0; +const LAST_HISTORY_ROW = -1; + +const isoDayPattern = /^\d{4}-\d{2}-\d{2}$/; const baseUrl = 'https://iss.moex.com/iss'; const timeoutMs = 10_000; @@ -28,9 +35,64 @@ export type MoexBondYield = { tradeAt: string; value: string; }; +export type MoexIndexClose = { close: string; tradeDate: string }; @Injectable() export class MoexClient { + async indexHistory( + secid: 'IMOEX' | 'MCFTRR', + from: string, + till: string, + ): Promise { + const path = `history/engines/stock/markets/index/securities/${secid}.json`; + const rows: MoexIndexClose[] = []; + let start = FIRST_PAGE_OFFSET; + for (;;) { + const query = new URLSearchParams({ + from, + till, + 'start': String(start), + 'iss.meta': 'off', + 'history.columns': 'TRADEDATE,CLOSE', + }); + const response = await this.request(`${baseUrl}/${path}?${query}`); + if (!response) throw new ServiceUnavailableException({ code: MoexErrorCode.Unavailable }); + const block = response.history; + if ( + !block || !Array.isArray(block.columns) || + block.columns.length !== INDEX_COLUMNS.length || + // eslint-disable-next-line security/detect-object-injection -- Offset comes from a fixed-length validated columns array. + block.columns.some((column, offset) => column !== INDEX_COLUMNS[offset]) + ) throw new ServiceUnavailableException({ code: MoexErrorCode.Unavailable }); + const current = parseRows(block); + let previousDate = rows.at(LAST_HISTORY_ROW)?.tradeDate; + for (const row of current) { + const tradeDate = text(row.tradedate); + const close = numberText(row.close); + if ( + !tradeDate || !isoDayPattern.test(tradeDate) || + Number.isNaN(Date.parse(`${tradeDate}T00:00:00Z`)) || + new Date(`${tradeDate}T00:00:00Z`).toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH) !== tradeDate || + !close || Number(close) <= ZERO_CLOSE || + (previousDate !== undefined && tradeDate <= previousDate) + ) throw new ServiceUnavailableException({ code: MoexErrorCode.Unavailable }); + rows.push({ tradeDate, close }); + previousDate = tradeDate; + } + const cursor = parseRows(response['history.cursor']); + const index = integer(cursor[0]?.index); + const total = integer(cursor[0]?.total); + const pageSize = integer(cursor[0]?.pagesize); + if ( + cursor.length !== EXPECTED_CURSOR_ROWS || index !== start || + total === null || pageSize === null || pageSize === EMPTY_PAGE_SIZE || + current.length !== Math.min(pageSize, total - start) + ) throw new ServiceUnavailableException({ code: MoexErrorCode.Unavailable }); + if (index + pageSize >= total) return rows; + start = index + pageSize; + } + } + async bondYield( secid: string, isin: string, diff --git a/src/modules/investments/market-index/market-index.module.ts b/src/modules/investments/market-index/market-index.module.ts new file mode 100644 index 0000000..b927ec0 --- /dev/null +++ b/src/modules/investments/market-index/market-index.module.ts @@ -0,0 +1,13 @@ +import { Module } from '@nestjs/common'; + +import { MoexModule } from '#src/integrations/moex/moex.module'; + +import { MarketIndexRepository } from './market-index.repository'; +import { MarketIndexHistoryService } from './market-index.service'; + +@Module({ + imports: [MoexModule], + providers: [MarketIndexHistoryService, MarketIndexRepository], + exports: [MarketIndexHistoryService], +}) +export class MarketIndexModule {} diff --git a/src/modules/investments/market-index/market-index.repository.ts b/src/modules/investments/market-index/market-index.repository.ts new file mode 100644 index 0000000..62e3088 --- /dev/null +++ b/src/modules/investments/market-index/market-index.repository.ts @@ -0,0 +1,52 @@ +import { Injectable } from '@nestjs/common'; + +import type { MoexIndexClose } from '#src/integrations/moex/moex.client'; + +import { PrismaService } from '#src/prisma/prisma.service'; + +@Injectable() +export class MarketIndexRepository { + constructor(private readonly prisma: PrismaService) {} + + async read(secid: 'IMOEX' | 'MCFTRR', from: string, to: string) { + const [closes, state] = await Promise.all([ + this.prisma.marketIndexDailyClose.findMany({ + where: { secid, tradeDate: { gte: from, lte: to } }, + orderBy: { tradeDate: 'asc' }, + }), + this.prisma.marketIndexSyncState.findUnique({ where: { secid } }), + ]); + return { closes, state }; + } + + async attempt(secid: 'IMOEX' | 'MCFTRR', at: Date) { + await this.prisma.marketIndexSyncState.upsert({ + where: { secid }, create: { secid, lastAttemptAt: at }, update: { lastAttemptAt: at }, + }); + } + + // coveredTo is the `till` MOEX answered for: a close for today that is not + // published yet is re-checked by the service while the last close < to. + async save(secid: 'IMOEX' | 'MCFTRR', closes: MoexIndexClose[], from: string, to: string, at: Date) { + await this.prisma.$transaction(async (tx) => { + for (const close of closes) { + await tx.marketIndexDailyClose.upsert({ + where: { secid_tradeDate: { secid, tradeDate: close.tradeDate } }, + create: { secid, ...close }, + update: { close: close.close }, + }); + } + const state = await tx.marketIndexSyncState.findUniqueOrThrow({ where: { secid } }); + const joined = Boolean(state.coveredFrom && state.coveredTo && + from <= state.coveredTo && to >= state.coveredFrom); + await tx.marketIndexSyncState.update({ + where: { secid }, + data: { + lastSuccessAt: at, + coveredFrom: joined && state.coveredFrom && state.coveredFrom < from ? state.coveredFrom : from, + coveredTo: joined && state.coveredTo && state.coveredTo > to ? state.coveredTo : to, + }, + }); + }); + } +} diff --git a/src/modules/investments/market-index/market-index.service.spec.ts b/src/modules/investments/market-index/market-index.service.spec.ts new file mode 100644 index 0000000..6bed5c6 --- /dev/null +++ b/src/modules/investments/market-index/market-index.service.spec.ts @@ -0,0 +1,121 @@ +import { MarketIndexHistoryService } from './market-index.service'; + +jest.mock('./market-index.repository', () => ({ MarketIndexRepository: class {} })); + +type State = { + coveredFrom: null | string; + coveredTo: null | string; + lastAttemptAt: Date | null; + lastSuccessAt: Date | null; +}; +type Close = { close: string; tradeDate: string }; + +const HOUR_MS = 3_600_000; + +// In-memory stand-in mirroring the repository's merge rules. +function setup(history: Close[]) { + let state: null | State = null; + const stored = new Map(); + const repository = { + read: jest.fn((_: string, from: string, to: string) => Promise.resolve({ + closes: [...stored.entries()] + .filter(([date]) => date >= from && date <= to) + .sort(([a], [b]) => a.localeCompare(b)) + .map(([tradeDate, close]) => ({ tradeDate, close })), + state: state && { ...state }, + })), + attempt: jest.fn((_: string, at: Date) => { + state = { coveredFrom: null, coveredTo: null, lastSuccessAt: null, ...state, lastAttemptAt: at }; + return Promise.resolve(); + }), + save: jest.fn((_: string, closes: Close[], from: string, to: string, at: Date) => { + for (const close of closes) stored.set(close.tradeDate, close.close); + const current = state as State; + const joined = Boolean(current.coveredFrom && current.coveredTo && + from <= current.coveredTo && to >= current.coveredFrom); + state = { + ...current, + lastSuccessAt: at, + coveredFrom: joined && current.coveredFrom && current.coveredFrom < from ? current.coveredFrom : from, + coveredTo: joined && current.coveredTo && current.coveredTo > to ? current.coveredTo : to, + }; + return Promise.resolve(); + }), + }; + const moex = { + indexHistory: jest.fn((_: string, from: string, till: string) => Promise.resolve( + history.filter((row) => row.tradeDate >= from && row.tradeDate <= till), + )), + }; + const service = new MarketIndexHistoryService( + repository as never, moex as never, { event: jest.fn() } as never, + ); + return { service, moex, setHistory: (rows: Close[]) => history.splice(0, history.length, ...rows) }; +} + +const closes = (...dates: string[]) => dates.map((tradeDate) => ({ tradeDate, close: '100' })); + +describe('MarketIndexHistoryService', () => { + afterEach(() => jest.useRealTimers()); + + it('loads a ten-day lookback once and reuses the confirmed range', async () => { + const { service, moex } = setup(closes('2026-09-30', '2026-10-02', '2026-10-05')); + expect(await service.read('MCFTRR', '2026-10-02', '2026-10-05')).toHaveLength(3); + expect(await service.read('MCFTRR', '2026-10-02', '2026-10-05')).toHaveLength(3); + expect(moex.indexHistory).toHaveBeenCalledTimes(1); + expect(moex.indexHistory).toHaveBeenCalledWith('MCFTRR', '2026-09-22', '2026-10-05'); + }); + + it('extends history backward at once, even right after a successful load', async () => { + const { service, moex } = setup(closes('2023-01-03', '2026-09-30', '2026-10-05')); + await service.read('MCFTRR', '2026-10-01', '2026-10-05'); + expect(await service.read('MCFTRR', '2023-01-05', '2026-10-05')).toHaveLength(3); + expect(moex.indexHistory).toHaveBeenLastCalledWith('MCFTRR', '2022-12-26', '2026-10-05'); + }); + + it('re-checks a missing tail close after an hour, but not sooner', async () => { + jest.useFakeTimers({ now: new Date('2026-10-05T12:00:00Z') }); + const { service, moex, setHistory } = setup(closes('2026-09-30', '2026-10-02')); + await service.read('MCFTRR', '2026-10-02', '2026-10-05'); + setHistory(closes('2026-09-30', '2026-10-02', '2026-10-05')); + await service.read('MCFTRR', '2026-10-02', '2026-10-05'); + expect(moex.indexHistory).toHaveBeenCalledTimes(1); + jest.setSystemTime(Date.now() + HOUR_MS + 1); + const rows = await service.read('MCFTRR', '2026-10-02', '2026-10-05'); + expect(rows?.at(-1)?.tradeDate).toBe('2026-10-05'); + expect(moex.indexHistory).toHaveBeenCalledTimes(2); + }); + + it('keeps saved history usable when a later refresh fails and throttles retries', async () => { + jest.useFakeTimers({ now: new Date('2026-10-05T12:00:00Z') }); + const { service, moex } = setup(closes('2026-09-30', '2026-10-02')); + await service.read('MCFTRR', '2026-10-02', '2026-10-02'); + jest.setSystemTime(Date.now() + 1000); + moex.indexHistory.mockRejectedValue(new Error('MOEX down')); + expect(await service.read('MCFTRR', '2026-10-02', '2026-10-05')).toHaveLength(2); + expect(await service.read('MCFTRR', '2026-10-02', '2026-10-05')).toHaveLength(2); + expect(moex.indexHistory).toHaveBeenCalledTimes(2); + }); + + it('returns no history when nothing was ever loaded and MOEX fails', async () => { + const { service, moex } = setup([]); + moex.indexHistory.mockRejectedValue(new Error('MOEX down')); + expect(await service.read('MCFTRR', '2026-10-02', '2026-10-05')).toBeNull(); + }); + + it('refuses a range that does not reach the last cash flow', async () => { + const { service, moex } = setup(closes('2026-09-30', '2026-10-02')); + await service.read('MCFTRR', '2026-10-02', '2026-10-02'); + moex.indexHistory.mockRejectedValue(new Error('MOEX down')); + expect(await service.read('MCFTRR', '2026-10-02', '2026-10-09', '2026-10-08')).toBeNull(); + }); + + it('shares one request for parallel reads', async () => { + const { service, moex } = setup(closes('2026-09-30', '2026-10-02')); + await Promise.all([ + service.read('IMOEX', '2026-10-02', '2026-10-05'), + service.read('IMOEX', '2026-10-02', '2026-10-05'), + ]); + expect(moex.indexHistory).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/modules/investments/market-index/market-index.service.ts b/src/modules/investments/market-index/market-index.service.ts new file mode 100644 index 0000000..190dd0f --- /dev/null +++ b/src/modules/investments/market-index/market-index.service.ts @@ -0,0 +1,115 @@ +import { Injectable } from '@nestjs/common'; + +import { StructuredLogger } from '#src/common/logging/structured-logger.service'; +import { MoexClient, type MoexIndexClose } from '#src/integrations/moex/moex.client'; + +import { MarketIndexRepository } from './market-index.repository'; + +const HOUR_MS = 3_600_000; +const LOOKBACK_DAYS = 10; +const ISO_DATE_LENGTH = 10; +const ISO_DATE_START = 0; +const LAST_CLOSE = -1; +const MAX_REFRESH_ROUNDS = 2; +const FIRST_ROUND = 0; + +type IndexId = 'IMOEX' | 'MCFTRR'; +type Saved = Awaited>; + +function shiftDays(date: string, days: number): string { + const value = new Date(`${date}T00:00:00Z`); + value.setUTCDate(value.getUTCDate() + days); + return value.toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH); +} + +function within(at: Date | null | undefined, ms: number) { + return Boolean(at && Date.now() - at.valueOf() < ms); +} + +@Injectable() +export class MarketIndexHistoryService { + private readonly pending = new Map>(); + + constructor( + private readonly repository: MarketIndexRepository, + private readonly moex: MoexClient, + private readonly logger: StructuredLogger, + ) {} + + /** + * Closes covering [from − 10 days, to]. `requiredThrough` is the latest date + * that must have a close on or after it (the last cash flow); null when the + * confirmed history does not cover the period. + */ + async read( + secid: IndexId, + from: string, + to: string, + requiredThrough = from, + ): Promise { + const rangeFrom = shiftDays(from, -LOOKBACK_DAYS); + let saved = await this.repository.read(secid, rangeFrom, to); + for (let round = FIRST_ROUND; round < MAX_REFRESH_ROUNDS; round++) { + const range = this.refreshRange(saved, rangeFrom, to); + if (!range) break; + const job = this.pending.get(secid) ?? this.start(secid, range.from, to); + await job; + saved = await this.repository.read(secid, rangeFrom, to); + } + return this.covered(saved, from, rangeFrom, to, requiredThrough) + ? saved.closes.map(({ tradeDate, close }) => ({ tradeDate, close })) + : null; + } + + // coveredFrom..coveredTo is the range MOEX was successfully asked about. + // Extending it backward is fetched at once; an unpublished tail close is + // re-checked at most hourly; a failed attempt is retried at most hourly. + private refreshRange(saved: Saved, rangeFrom: string, to: string) { + const state = saved.state; + const lastAttempt = state?.lastAttemptAt ?? null; + const lastSuccess = state?.lastSuccessAt ?? null; + // Success is stamped with its own attempt time, so equal means success. + const failed = lastAttempt !== null && (lastSuccess === null || lastSuccess.valueOf() !== lastAttempt.valueOf()); + if (failed && within(lastAttempt, HOUR_MS)) return null; + if (!state?.coveredFrom || !state.coveredTo || state.coveredFrom > rangeFrom) return { from: rangeFrom }; + if (state.coveredTo < to) return { from: state.coveredTo }; + const last = saved.closes.at(LAST_CLOSE)?.tradeDate; + if ((!last || last < to) && !within(lastSuccess, HOUR_MS)) return { from: last ?? rangeFrom }; + return null; + } + + private covered(saved: Saved, from: string, rangeFrom: string, to: string, requiredThrough: string) { + const last = saved.closes.at(LAST_CLOSE)?.tradeDate; + return Boolean( + saved.state?.coveredFrom && saved.state.coveredFrom <= rangeFrom && + saved.closes.some((row) => row.tradeDate <= from) && + last && last >= shiftDays(to, -LOOKBACK_DAYS) && + // The last cash flow must lie inside the checked range, otherwise it + // would be priced by a stale close. + saved.state.coveredTo && saved.state.coveredTo >= requiredThrough, + ); + } + + private start(secid: IndexId, from: string, to: string): Promise { + const job = this.refresh(secid, from, to).finally(() => this.pending.delete(secid)); + this.pending.set(secid, job); + return job; + } + + private async refresh(secid: IndexId, from: string, to: string): Promise { + const at = new Date(); + const started = Date.now(); + await this.repository.attempt(secid, at); + try { + const closes = await this.moex.indexHistory(secid, from, to); + await this.repository.save(secid, closes, from, to, at); + this.logger.event('info', 'integration.moex.index_history.succeeded', { + operation: secid, count: closes.length, durationMs: Date.now() - started, + }); + } catch { + this.logger.event('warn', 'integration.moex.index_history.failed', { + operation: secid, durationMs: Date.now() - started, + }); + } + } +}