ai/datafeeds/udf/lib/data-pulse-provider.js

import { getErrorMessage, logMessage, } from './helpers';
export class DataPulseProvider {
    constructor(historyProvider, updateFrequency) {
        this._subscribers = {};
        this._requestsPending = 0;
        this._historyProvider = historyProvider;
        setInterval(this._updateData.bind(this), updateFrequency);
    }
    subscribeBars(symbolInfo, resolution, newDataCallback, listenerGuid) {
        if (this._subscribers.hasOwnProperty(listenerGuid)) {
            logMessage(`DataPulseProvider: already has subscriber with id=${listenerGuid}`);
            return;
        }
        this._subscribers[listenerGuid] = {
            lastBarTime: null,
            listener: newDataCallback,
            resolution: resolution,
            symbolInfo: symbolInfo,
        };
        logMessage(`DataPulseProvider: subscribed for #${listenerGuid} - {${symbolInfo.name}, ${resolution}}`);
    }
    unsubscribeBars(listenerGuid) {
        delete this._subscribers[listenerGuid];
        logMessage(`DataPulseProvider: unsubscribed for #${listenerGuid}`);
    }
    _updateData() {
        if (this._requestsPending > 0) {
            return;
        }
        this._requestsPending = 0;
        // eslint-disable-next-line guard-for-in
        for (const listenerGuid in this._subscribers) {
            this._requestsPending += 1;
            this._updateDataForSubscriber(listenerGuid)
                .then(() => {
                this._requestsPending -= 1;
                logMessage(`DataPulseProvider: data for #${listenerGuid} updated successfully, pending=${this._requestsPending}`);
            })
                .catch((reason) => {
                this._requestsPending -= 1;
                logMessage(`DataPulseProvider: data for #${listenerGuid} updated with error=${getErrorMessage(reason)}, pending=${this._requestsPending}`);
            });
        }
    }
    _updateDataForSubscriber(listenerGuid) {
        const subscriptionRecord = this._subscribers[listenerGuid];
        const rangeEndTime = parseInt((Date.now() / 1000).toString());
        // BEWARE: please note we really need 2 bars, not the only last one
        // see the explanation below. `10` is the `large enough` value to work around holidays
        const rangeStartTime = rangeEndTime - periodLengthSeconds(subscriptionRecord.resolution, 10);
        return this._historyProvider.getBars(subscriptionRecord.symbolInfo, subscriptionRecord.resolution, {
            from: rangeStartTime,
            to: rangeEndTime,
            countBack: 2,
            firstDataRequest: false,
        })
            .then((result) => {
            this._onSubscriberDataReceived(listenerGuid, result);
        });
    }
    _onSubscriberDataReceived(listenerGuid, result) {
        // means the subscription was cancelled while waiting for data
        if (!this._subscribers.hasOwnProperty(listenerGuid)) {
            logMessage(`DataPulseProvider: Data comes for already unsubscribed subscription #${listenerGuid}`);
            return;
        }
        const bars = result.bars;
        if (bars.length === 0) {
            return;
        }
        const lastBar = bars[bars.length - 1];
        const subscriptionRecord = this._subscribers[listenerGuid];
        if (subscriptionRecord.lastBarTime !== null && lastBar.time < subscriptionRecord.lastBarTime) {
            return;
        }
        const isNewBar = subscriptionRecord.lastBarTime !== null && lastBar.time > subscriptionRecord.lastBarTime;
        // Pulse updating may miss some trades data (ie, if pulse period = 10 secods and new bar is started 5 seconds later after the last update, the
        // old bar's last 5 seconds trades will be lost). Thus, at fist we should broadcast old bar updates when it's ready.
        if (isNewBar) {
            if (bars.length < 2) {
                throw new Error('Not enough bars in history for proper pulse update. Need at least 2.');
            }
            const previousBar = bars[bars.length - 2];
            subscriptionRecord.listener(previousBar);
        }
        subscriptionRecord.lastBarTime = lastBar.time;
        subscriptionRecord.listener(lastBar);
    }
}
function periodLengthSeconds(resolution, requiredPeriodsCount) {
    let daysCount = 0;
    if (resolution === 'D' || resolution === '1D') {
        daysCount = requiredPeriodsCount;
    }
    else if (resolution === 'M' || resolution === '1M') {
        daysCount = 31 * requiredPeriodsCount;
    }
    else if (resolution === 'W' || resolution === '1W') {
        daysCount = 7 * requiredPeriodsCount;
    }
    else {
        daysCount = requiredPeriodsCount * parseInt(resolution) / (24 * 60);
    }
    return daysCount * 24 * 60 * 60;
}