Загрузка данных
import dayjs from 'dayjs';
import duration from 'dayjs/plugin/duration';
import { parseTimeframe } from 'moex-chart';
import { moexChartTimeConverter, moexChartToIssTimeframe } from '@utils/chartToReqTimeConverter';
import { DEFAULT_SYMBOL } from '@widgets/Chart/const';
import { requestBars, requestRealtimeBars } from '../../requestBars';
import { ChartIndicativeData } from '../../types';
import type { Candle, Timeframes } from 'moex-chart';
dayjs.extend(duration);
interface HistoryRequestState {
untilTime?: number;
request: Promise<Candle[] | null> | null;
}
function getRequestSymbol(symbolRaw?: string): string | undefined {
const symbol = String(symbolRaw ?? '').trim();
if (!symbol || symbol === DEFAULT_SYMBOL) {
return undefined;
}
return symbol;
}
function getTimeframeSeconds(timeframe: Timeframes): number {
const { candleWidth, dayjsUnit } = parseTimeframe(timeframe);
return dayjs.duration(candleWidth, dayjsUnit).asSeconds();
}
function aggregateCandles(candles: Candle[], time: number): Candle | undefined {
const firstCandle = candles[0];
const lastCandle = candles[candles.length - 1];
if (!firstCandle || !lastCandle) {
return undefined;
}
return {
time,
open: firstCandle.open,
high: Math.max(...candles.map(({ high }) => high)),
low: Math.min(...candles.map(({ low }) => low)),
close: lastCandle.close,
volume: candles.reduce((total, candle) => total + (candle.volume ?? 0), 0),
};
}
// По хорошему - класс должен быть синглтоном, чтобы кормить MoexChart одинаковой датой,
// и не плодить несколько подключений на одни символа
class DataSourceProvider {
private prevRealtimeDataArr: Candle[] = [];
private prevRealtimeData: Candle | undefined;
private realtimeShouldBeConvoluted = false;
private realtimeSessionStart: number | null = null;
private realtimeTimer: ReturnType<typeof setInterval> | null = null;
private historyRequests = new Map<string, HistoryRequestState>();
public getDataSource =
(indicativeData?: ChartIndicativeData, cb?: (timeframe: Timeframes) => void) =>
async (timeframe: Timeframes, symbolId: string, until?: Candle): Promise<Candle[] | null> => {
const symbol = getRequestSymbol(symbolId);
if (!symbol) {
return null;
}
const historyRequestKey = `${symbol}:${timeframe}`;
const historyRequestState = this.historyRequests.get(historyRequestKey);
if (historyRequestState && historyRequestState.untilTime === until?.time) {
if (historyRequestState.request) {
return historyRequestState.request;
}
if (until) {
return null;
}
}
cb?.(timeframe);
const historyRequest = this.requestHistoryData({
timeframe,
symbol,
until,
indicativeData,
});
this.historyRequests.set(historyRequestKey, {
untilTime: until?.time,
request: historyRequest,
});
try {
return await historyRequest;
} finally {
if (this.historyRequests.get(historyRequestKey)?.request === historyRequest) {
if (until) {
this.historyRequests.set(historyRequestKey, {
untilTime: until.time,
request: null,
});
} else {
this.historyRequests.delete(historyRequestKey);
}
}
}
};
public startRealtime({
getSymbols,
getTimeframe,
update,
periodMs = 5000,
indicativeData,
}: {
getSymbols: () => string[];
getTimeframe: () => Timeframes;
update: (symbolId: string, candle: Candle) => void;
periodMs?: number;
indicativeData?: ChartIndicativeData;
}): () => void {
if (this.realtimeTimer) {
clearInterval(this.realtimeTimer);
}
this.realtimeTimer = setInterval(() => {
const timeframe = getTimeframe();
const symbolIds = getSymbols();
Promise.all(
symbolIds.map(async (symbolId) => {
const symbol = getRequestSymbol(symbolId);
if (!symbol) {
return;
}
const data = await requestRealtimeBars({
currencyPair: symbol.replaceAll(':', '.'),
interval: moexChartTimeConverter(timeframe),
ticker: symbol,
indicativeData,
});
if (!data) {
return;
}
if (this.prevRealtimeData && JSON.stringify(data) === JSON.stringify(this.prevRealtimeData)) {
return;
}
if (!this.realtimeShouldBeConvoluted) {
this.prevRealtimeData = data;
update(symbol, data);
return;
}
this.realtimeConvolution(timeframe, data, (candle) => {
update(symbol, candle);
});
}),
);
}, periodMs);
return () => {
if (this.realtimeTimer) {
clearInterval(this.realtimeTimer);
}
this.realtimeTimer = null;
};
}
private async requestHistoryData({
timeframe,
symbol,
until,
indicativeData,
}: {
timeframe: Timeframes;
symbol: string;
until?: Candle;
indicativeData?: ChartIndicativeData;
}): Promise<Candle[] | null> {
const interval = moexChartTimeConverter(timeframe);
const date = until?.time ?? Math.round(Date.now() / 1000);
const data = await requestBars({
currencyPair: symbol.replaceAll(':', '.'),
interval,
periodParams: {
firstDataRequest: true,
to: date,
from: Math.round(Date.now() / 1000),
countBack: 2000,
},
ticker: symbol,
indicativeData,
});
if (data.length === 0) {
return null;
}
const issTimeframe = moexChartToIssTimeframe(timeframe);
if (issTimeframe === timeframe) {
this.realtimeShouldBeConvoluted = false;
if (!until) {
this.prevRealtimeData = data[data.length - 1];
this.prevRealtimeDataArr = [];
this.realtimeSessionStart = null;
}
return data;
}
this.realtimeShouldBeConvoluted = true;
return this.timeframeConvolution(data, timeframe, !until);
}
private timeframeConvolution(data: Candle[], requestedTimeframe: Timeframes, syncRealtime: boolean): Candle[] {
const timeframeSeconds = getTimeframeSeconds(requestedTimeframe);
const sortedData = [...data].sort((first, second) => first.time - second.time);
const firstCandle = sortedData[0];
if (!firstCandle) {
return [];
}
const result: Candle[] = [];
let sessionStart = firstCandle.time;
let bucketStart = sessionStart;
let candleGroup: Candle[] = [];
sortedData.forEach((candle, index) => {
const previousCandle = sortedData[index - 1];
const isNewSession = previousCandle && candle.time - previousCandle.time > timeframeSeconds;
if (isNewSession) {
const aggregatedCandle = aggregateCandles(candleGroup, bucketStart);
if (aggregatedCandle) {
result.push(aggregatedCandle);
}
sessionStart = candle.time;
bucketStart = candle.time;
candleGroup = [candle];
return;
}
const currentBucketStart =
sessionStart + Math.floor((candle.time - sessionStart) / timeframeSeconds) * timeframeSeconds;
if (currentBucketStart !== bucketStart) {
const aggregatedCandle = aggregateCandles(candleGroup, bucketStart);
if (aggregatedCandle) {
result.push(aggregatedCandle);
}
bucketStart = currentBucketStart;
candleGroup = [];
}
candleGroup.push(candle);
});
const aggregatedCandle = aggregateCandles(candleGroup, bucketStart);
if (aggregatedCandle) {
result.push(aggregatedCandle);
}
if (syncRealtime) {
this.realtimeSessionStart = sessionStart;
this.prevRealtimeDataArr = [...candleGroup];
this.prevRealtimeData = sortedData[sortedData.length - 1];
}
return result;
}
private realtimeConvolution(timeframe: Timeframes, data: Candle, update: (candle: Candle) => void): void {
const timeframeSeconds = getTimeframeSeconds(timeframe);
if (!this.prevRealtimeData || this.realtimeSessionStart === null) {
this.realtimeSessionStart = data.time;
this.prevRealtimeDataArr = [data];
} else {
const isNewSession = data.time - this.prevRealtimeData.time > timeframeSeconds;
if (isNewSession) {
this.realtimeSessionStart = data.time;
this.prevRealtimeDataArr = [data];
} else {
const previousBucketStart =
this.realtimeSessionStart +
Math.floor((this.prevRealtimeData.time - this.realtimeSessionStart) / timeframeSeconds) * timeframeSeconds;
const currentBucketStart =
this.realtimeSessionStart +
Math.floor((data.time - this.realtimeSessionStart) / timeframeSeconds) * timeframeSeconds;
if (currentBucketStart !== previousBucketStart) {
this.prevRealtimeDataArr = [data];
} else {
const candleIndex = this.prevRealtimeDataArr.findIndex((candle) => candle.time === data.time);
if (candleIndex === -1) {
this.prevRealtimeDataArr.push(data);
} else {
this.prevRealtimeDataArr[candleIndex] = data;
}
}
}
}
this.prevRealtimeData = data;
const sessionStart = this.realtimeSessionStart;
if (sessionStart === null) {
return;
}
const bucketStart = sessionStart + Math.floor((data.time - sessionStart) / timeframeSeconds) * timeframeSeconds;
const candle = aggregateCandles(this.prevRealtimeDataArr, bucketStart);
if (candle) {
update(candle);
}
}
}
export { DataSourceProvider };
import dayjs from 'dayjs';
import utc from 'dayjs/plugin/utc';
import { indicativeQuotesController } from '@api/controllers/indicativeQuotesController';
import api from '@api/index';
import { candleToBar } from '@utils/candleToBar';
import { ChartIndicativeData } from './types';
import { isIndicativeTicker } from './utils/isIndicativeTicker';
import { transformKeyToLowerCase } from './utils/transformKeyToLowerCase';
import type { Candle } from 'moex-chart';
dayjs.extend(utc);
export interface PeriodParams {
from: number;
to: number;
countBack: number;
firstDataRequest: boolean;
}
export interface HistoryMetadata {
noData: boolean;
}
export type HistoryCallback = (candles: Candle[], metadata: HistoryMetadata) => void;
export type SubscribeBarsCallback = (candle: Candle) => void;
interface RequestBarsArgs {
currencyPair: string;
interval: string;
periodParams: PeriodParams;
onHistoryCallback?: HistoryCallback;
ticker?: string;
indicativeData?: ChartIndicativeData;
}
interface RequestRealtimeBarsArgs {
currencyPair: string;
interval: string;
ticker?: string;
indicativeData?: ChartIndicativeData;
onRealtimeCallback?: SubscribeBarsCallback;
}
export type CustomBarsResolver = (args: {
ticker?: string;
currencyPair: string;
periodParams: PeriodParams;
interval: string;
}) => Promise<Candle[]>;
const barsResolvers = new Map<string, CustomBarsResolver>();
export const registerBarsResolver = (boardKey: string, resolver: CustomBarsResolver) => {
barsResolvers.set(boardKey, resolver);
};
const getResolverKey = (ticker?: string) => {
const [source, board] = ticker?.split(/[:.]/) ?? [];
if (!source || !board) {
return '';
}
return `${source}:${board}`;
};
const getBarsResolver = (ticker?: string) => barsResolvers.get(getResolverKey(ticker));
export async function requestBars({
currencyPair,
interval,
periodParams,
onHistoryCallback,
ticker,
indicativeData,
}: RequestBarsArgs): Promise<Candle[]> {
const customResolver = getBarsResolver(ticker);
if (customResolver) {
try {
const customBars = await customResolver({ ticker, currencyPair, periodParams, interval });
const toMs = periodParams.to * 1000;
const olderBars = customBars.filter((bar) => bar.time < toMs);
onHistoryCallback?.(olderBars, { noData: olderBars.length === 0 });
return olderBars;
} catch (e) {
onHistoryCallback?.([], { noData: true });
return [];
}
}
const date = new Date(periodParams.to * 1000);
const year = date.getUTCFullYear();
const month = `0${date.getUTCMonth() + 1}`.slice(-2);
const day = `0${date.getUTCDate()}`.slice(-2);
const hours = `0${date.getUTCHours()}`.slice(-2);
const minutes = `0${date.getUTCMinutes()}`.slice(-2);
const seconds = `0${date.getUTCSeconds()}`.slice(-2);
const hasIndicativeBoardInTicker = isIndicativeTicker(ticker);
// если при инициализации графика были данные indicativeData и текущий инструмент совпадает
// то отправляем запрос на индикатив
// иначе на инструменты
const isIndicativeInstrument = (indicativeData && indicativeData.key === ticker) || hasIndicativeBoardInTicker;
if (isIndicativeInstrument) {
const dateStr = `${year}-${month}-${day}T${hours}:${minutes}:${seconds}`;
const lowerCaseKey = transformKeyToLowerCase(currencyPair);
if (!lowerCaseKey) {
onHistoryCallback?.([], { noData: true });
return [];
}
try {
const { data } = await indicativeQuotesController.getCandles({
count: periodParams.countBack,
key: lowerCaseKey,
date: dateStr,
interval,
});
const candles = data.indicativeCandles.reverse().map(candleToBar);
onHistoryCallback?.(candles, {
noData: candles.length === 0,
});
return candles;
} catch {
onHistoryCallback?.([], {
noData: true,
});
return [];
}
}
const dateStr = `${year}-${month}-${day}%20${hours}:${minutes}:${seconds}`;
try {
const { data } = await api.getBars({
currencyPair,
date: dateStr,
interval,
count: periodParams.countBack,
ticker,
});
const bars = data.reverse().map(candleToBar);
onHistoryCallback?.(bars, {
noData: bars.length === 0,
});
return bars;
} catch {
onHistoryCallback?.([], {
noData: true,
});
return [];
}
}
export async function requestRealtimeBars({
currencyPair,
interval,
ticker,
onRealtimeCallback,
indicativeData,
}: RequestRealtimeBarsArgs): Promise<Candle | undefined> {
if (getBarsResolver(ticker)) {
return;
}
const hasIndicativeBoardInTicker = isIndicativeTicker(ticker);
// если при инициализации графика были данные indicativeData и текущий инструмент совпадает
// то отправляем запрос на индикатив
// иначе на инструменты
const isIndicativeInstrument = (indicativeData && indicativeData.key === ticker) || hasIndicativeBoardInTicker;
if (isIndicativeInstrument) {
const lowerCaseKey = transformKeyToLowerCase(currencyPair);
if (!lowerCaseKey) {
return undefined;
}
try {
const { data } = await indicativeQuotesController.getCandles({
count: 1,
key: lowerCaseKey,
date: dayjs().utc().add(1, 'minute').format('YYYY-MM-DDTHH:mm:ss'),
interval,
});
if (data.indicativeCandles.length === 0) {
return undefined;
}
const sortedData = [...data.indicativeCandles].sort(
(first, second) => new Date(first.end).valueOf() - new Date(second.end).valueOf(),
);
const firstCandle = sortedData[0];
if (!firstCandle) {
return undefined;
}
const bar = candleToBar(firstCandle);
onRealtimeCallback?.(bar);
return bar;
} catch (error) {
console.error('error from requestRealTimeBars indicativeQuotesController: ', error);
return undefined;
}
}
try {
const { data } = await api.getBars({
currencyPair,
date: dayjs().utc().add(1, 'minute').format('YYYY-MM-DD%20HH:mm:ss'),
interval,
count: 1,
ticker,
});
if (data.length === 0) {
return undefined;
}
const sortedData = [...data].sort(
(first, second) => new Date(first.end).valueOf() - new Date(second.end).valueOf(),
);
const firstCandle = sortedData[0];
if (!firstCandle) {
return undefined;
}
const bar = candleToBar(firstCandle);
onRealtimeCallback?.(bar);
return bar;
} catch (error) {
console.error('error from requestRealTimeBars: ', error);
return undefined;
}
}