Загрузка данных
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 RealtimeState {
previousTime: number | null;
candles: Candle[];
sessionStart: number | 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 isValidCandle({ time, open, high, low, close, volume }: Candle): boolean {
const valuesAreValid = [time, open, high, low, close].every(Number.isFinite);
return (
valuesAreValid &&
(volume === undefined || Number.isFinite(volume)) &&
high >= Math.max(open, close) &&
low <= Math.min(open, close)
);
}
function aggregateCandles(candles: Candle[], time: number): Candle | undefined {
const firstCandle = candles[0];
const lastCandle = candles[candles.length - 1];
if (!firstCandle || !lastCandle) {
return undefined;
}
const high = Math.max(...candles.map((candle) => candle.high));
const low = Math.min(...candles.map((candle) => candle.low));
const volume = candles.reduce((total, candle) => total + (candle.volume ?? 0), 0);
return {
time,
open: firstCandle.open,
high,
low,
close: lastCandle.close,
volume,
};
}
function timeframeConvolution(
data: Candle[],
requestedTimeframe: Timeframes,
): {
candles: Candle[];
state: RealtimeState;
} {
const timeframeSeconds = getTimeframeSeconds(requestedTimeframe);
const sortedData = [...data].sort((first, second) => first.time - second.time);
const firstCandle = sortedData[0];
if (!firstCandle) {
return {
candles: [],
state: {
previousTime: null,
candles: [],
sessionStart: null,
},
};
}
const result: Candle[] = [];
let sessionStart = firstCandle.time;
let bucketStart = sessionStart;
let previousTime: number | null = null;
let candleGroup: Candle[] = [];
sortedData.forEach((candle) => {
const isNewSession = previousTime !== null && candle.time - previousTime > timeframeSeconds;
if (isNewSession) {
const aggregatedCandle = aggregateCandles(candleGroup, bucketStart);
if (aggregatedCandle) {
result.push(aggregatedCandle);
}
sessionStart = candle.time;
bucketStart = candle.time;
candleGroup = [candle];
} else {
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);
}
previousTime = candle.time;
});
const aggregatedCandle = aggregateCandles(candleGroup, bucketStart);
if (aggregatedCandle) {
result.push(aggregatedCandle);
}
return {
candles: result,
state: {
previousTime,
candles: candleGroup,
sessionStart,
},
};
}
function realtimeConvolution(
timeframe: Timeframes,
data: Candle,
currentState: RealtimeState,
): {
candle: Candle | undefined;
state: RealtimeState;
} {
const timeframeSeconds = getTimeframeSeconds(timeframe);
const previousTime = currentState.previousTime;
let sessionStart = currentState.sessionStart;
let candles = [...currentState.candles];
const isNewSession =
previousTime === null ||
sessionStart === null ||
data.time - previousTime > timeframeSeconds;
if (isNewSession) {
sessionStart = data.time;
candles = [data];
} else {
const previousBucketStart =
sessionStart + Math.floor((previousTime - sessionStart) / timeframeSeconds) * timeframeSeconds;
const currentBucketStart =
sessionStart + Math.floor((data.time - sessionStart) / timeframeSeconds) * timeframeSeconds;
if (currentBucketStart !== previousBucketStart) {
candles = [data];
} else {
const candleIndex = candles.findIndex((candle) => candle.time === data.time);
if (candleIndex === -1) {
candles.push(data);
} else {
candles[candleIndex] = data;
}
}
}
const bucketStart =
sessionStart + Math.floor((data.time - sessionStart) / timeframeSeconds) * timeframeSeconds;
return {
candle: aggregateCandles(candles, bucketStart),
state: {
previousTime: data.time,
candles,
sessionStart,
},
};
}
// По хорошему - класс должен быть синглтоном, чтобы кормить MoexChart одинаковой датой,
// и не плодить несколько подключений на одни символа
class DataSourceProvider {
private readonly realtimeStates = new Map<string, RealtimeState>();
private realtimeTimer: ReturnType<typeof setInterval> | null = null;
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;
}
cb?.(timeframe);
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,
});
const validData = data.filter(isValidCandle);
if (validData.length === 0) {
return null;
}
if (moexChartToIssTimeframe(timeframe) === timeframe) {
if (!until) {
this.realtimeStates.set(symbol, {
previousTime: validData[validData.length - 1]?.time ?? null,
candles: [],
sessionStart: null,
});
}
return validData;
}
const { candles, state } = timeframeConvolution(validData, timeframe);
if (!until) {
this.realtimeStates.set(symbol, state);
}
return candles;
};
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();
Promise.allSettled(
getSymbols().map((symbolId) =>
this.updateRealtimeSymbol(symbolId, timeframe, update, indicativeData),
),
);
}, periodMs);
return () => {
if (this.realtimeTimer) {
clearInterval(this.realtimeTimer);
}
this.realtimeTimer = null;
};
}
private async updateRealtimeSymbol(
symbolId: string,
timeframe: Timeframes,
update: (symbolId: string, candle: Candle) => void,
indicativeData?: ChartIndicativeData,
): Promise<void> {
const symbol = getRequestSymbol(symbolId);
if (!symbol) {
return;
}
const data = await requestRealtimeBars({
currencyPair: symbol.replaceAll(':', '.'),
interval: moexChartTimeConverter(timeframe),
ticker: symbol,
indicativeData,
});
if (!data || !isValidCandle(data)) {
return;
}
if (moexChartToIssTimeframe(timeframe) === timeframe) {
this.realtimeStates.set(symbol, {
previousTime: data.time,
candles: [],
sessionStart: null,
});
update(symbol, data);
return;
}
const currentState = this.realtimeStates.get(symbol) ?? {
previousTime: null,
candles: [],
sessionStart: null,
};
const { candle, state } = realtimeConvolution(timeframe, data, currentState);
this.realtimeStates.set(symbol, state);
if (candle) {
update(symbol, candle);
}
}
}
export { DataSourceProvider };