feat(market-index): store MOEX index closes with confirmed coverage (WealthLedger/main#54)

2026-11-08 14:20:05 +03:00
parent 84433ee207
commit 74dea05113
8 changed files with 463 additions and 0 deletions
@@ -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 {}
@@ -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,
+        },
+      });
+    });
+  }
+}
@@ -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<string, string>();
+  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);
+  });
+});
@@ -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<ReturnType<MarketIndexRepository['read']>>;
+
+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<IndexId, Promise<void>>();
+
+  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<MoexIndexClose[] | null> {
+    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<void> {
+    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<void> {
+    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,
+      });
+    }
+  }
+}