tradingview-websocket进阶

谷梁英毅
2023-12-01

1、websocket

用户量量大,数据量大,而且要求实时更新数据的时候,需要使用websocket。
tradingview正好就是这样的应用场景。

2、tradingview和websocket结合

  • getBars方法。tradingview图表插件响应用户操作,根据用户界面渲染需要的数据时间段,调用getBars方法,传递参数function(symbolInfo, resolution, rangeStartDate, rangeEndDate, onLoadedCallback),我们需要构造出预期的参数返回给onLoadedCallback方法,即onLoadedCallback(data)。
  • onLoadedCallback方法。该方法接收的data有两种,一种是list数组。另一种是newBars对象。list是历史数据,rangeStartDate - rangeEndDate时间段的数据,根据时间颗粒resolution来划分。newBars是增量数据,同样是rangeStartDate - rangeEndDate时间段的数据,但是这个数据必定是在当前的resolution内。当我们把数据返回给onLoadedCallback,图表会渲染这些数据。

3、数据获取和填充

  1. 页面初始化,实例化图表,连接websocket。
  2. 从websocket得到的数据存入缓存cacheData。
  3. 因为getBars是图表自己调用的,所以当我们存入数据到cacheData的时候,在getBars方法中调用onLoadedCallback就可以。

4、实例

图表调用的getBars方法:

TVjsApi.prototype.getBars = function(symbolInfo, resolution, rangeStartDate, rangeEndDate, onLoadedCallback) {
    //console.log(' >> :', this.formatt(rangeStartDate*1000), this.formatt(rangeEndDate*1000))
    var ticker = this.symbol + "-" + resolution;
    var tickerload = ticker + "load";
    var tickerstate = ticker + "state";

    if(!this.cacheData[ticker] && !this.cacheData[tickerstate]){
        //如果缓存没有数据,而且未发出请求,记录当前节点开始时间
        this.cacheData[tickerload] = rangeStartDate;
        //发起请求,从websocket获取当前时间段的数据
        this.initMessage(symbolInfo, resolution, rangeStartDate, rangeEndDate, onLoadedCallback);
        //设置状态为true
        this.cacheData[tickerstate] = !0;
        return false;
    }

    if(!this.cacheData[tickerload] || this.cacheData[tickerload] > rangeStartDate){
        //如果缓存有数据,但是没有当前时间段的数据,更新当前节点时间
        this.cacheData[tickerload] = rangeStartDate;
        //发起请求,从websocket获取当前时间段的数据
        this.initMessage(symbolInfo, resolution, rangeStartDate, rangeEndDate, onLoadedCallback);
        //设置状态为true
        this.cacheData[tickerstate] = !0;
        return false;
    }
    if(this.cacheData[tickerstate]){
        //正在从websocket获取数据,禁止一切操作
        return false;
    }
    //ticker = this.symbol + "-" + this.interval;
    //如果缓存有数据,且长度不为0,构造数据
    if (this.cacheData[ticker] && this.cacheData[ticker].length) {
        this.isLoading = false
        var newBars = []
        this.cacheData[ticker].forEach(item => {
            if (item.time >= rangeStartDate * 1000 && item.time <= rangeEndDate * 1000) {
                newBars.push(item)
            }
        })
        onLoadedCallback(newBars)
    } else {
    //如果没有数据,等待10ms,再执行一次
        var self = this
        this.getBarTimer = setTimeout(function() {
            self.getBars(symbolInfo, resolution, rangeStartDate, rangeEndDate, onLoadedCallback)
        }, 10)
    }
}

获取历史数据list的init方法:

TVjsApi.prototype.initMessage = function(symbolInfo, resolution, rangeStartDate, rangeEndDate, onLoadedCallback){
    //console.log('发起请求,从websocket获取当前时间段的数据');
    var that = this;
    //保留当前回调
    that.cacheData['onLoadedCallback'] = onLoadedCallback;
    //获取需要请求的数据数目
    var limit = that.initLimit(resolution, rangeStartDate, rangeEndDate);
    //商品名
    var symbol = that.symbol;
    //如果当前时间颗粒已经改变,停止上一个时间颗粒的订阅,修改时间节点值
    if(that.interval !== resolution){
        that.unSubscribe(that.interval)
        that.interval = resolution;
    }    
    //获取当前时间段的数据,在onMessage中执行回调onLoadedCallback
    if (that.interval < 60) {
        that.socket.send({
            cmd: 'req',
            args: ["candle.M"+resolution+"."+symbol, limit, rangeEndDate],
            id: 'trade.M'+ resolution +'.'+ symbol.toLowerCase()
        })
    } else if (that.interval >= 60) {
        that.socket.send({
            cmd: 'req',
            args: ["candle.H"+(resolution/60)+"."+symbol, limit, rangeEndDate],
            id: 'trade.H'+ (resolution / 60) +'.'+ symbol.toLowerCase()
        })
    } else {
        that.socket.send({
            cmd: 'req',
            args: ["candle.D1."+symbol, limit, rangeEndDate],
            id: 'trade.D1.'+ symbol.toLowerCase()
        })
    }
}

响应websocket的Message方法:

TVjsApi.prototype.onMessage = function(data) {
    var thats = this;
    //  console.log("这是后台返回的数据"+count+":"+JSON.stringify(data) )
    
    if (data.data && data.data.length) {
        //websocket返回的值,数组代表时间段历史数据,不是增量
        var list = []
        var ticker = thats.symbol + "-" + thats.interval;
        var tickerstate = ticker + "state";
        
        //var that = thats;
        //遍历数组,构造缓存数据
        data.data.forEach(function(element) {
            list.push({
                time: element.id*1000,
                open: element.open,
                high: element.high,
                low: element.low,
                close: element.close,
                volume: element.quote_vol
            })
        }, thats)
        //如果没有缓存数据,则直接填充,发起订阅
        if(!thats.cacheData[ticker]){
            /*thats.cacheData[ticker] = thats.cacheData[ticker].concat(list);
            thats.cacheData['onLoadedCallback'](list);
        }else{*/
            thats.cacheData[ticker] = list;
            thats.subscribe()
        }
        //新数据即当前时间段需要的数据,直接喂给图表插件
        if(thats.cacheData['onLoadedCallback']){
            thats.cacheData['onLoadedCallback'](list);
        }
        //请求完成,设置状态为false
        thats.cacheData[tickerstate] = !1;
        //记录当前缓存时间,即数组最后一位的时间
        thats.lastTime = thats.cacheData[ticker][thats.cacheData[ticker].length - 1].time            
    }
    if (data.type && data.type.indexOf(thats.symbol.toLowerCase()) !== -1) {
        // console.log(' >> sub:', data.type)
        // data带有type,即返回的是订阅数据,
        //缓存的key
        var ticker = thats.symbol + "-" + thats.interval;
        //构造增量更新数据
        var barsData = {
            time: data.id * 1000,
            open: data.open,
            high: data.high,
            low: data.low,
            close: data.close,
            volume: data.quote_vol
        }
        /*if (barsData.time >= thats.lastTime && thats.cacheData[ticker] && thats.cacheData[ticker].length) {
            thats.cacheData[ticker][thats.cacheData[ticker].length - 1] = barsData
        }*/
        //如果增量更新数据的时间大于缓存时间,而且缓存有数据,数据长度大于0
        if (barsData.time > thats.lastTime && thats.cacheData[ticker] && thats.cacheData[ticker].length) {
            //增量更新的数据直接加入缓存数组
            thats.cacheData[ticker].push(barsData)
            //修改缓存时间
            thats.lastTime = barsData.time
        }else if(barsData.time == thats.lastTime && thats.cacheData[ticker].length){
            //如果增量更新的时间等于缓存时间,即在当前时间颗粒内产生了新数据,更新当前数据
            thats.cacheData[ticker][thats.cacheData[ticker].length - 1] = barsData
        }
        // 通知图表插件,可以开始增量更新的渲染了
        thats.datafeeds.barsUpdater.updateData()
    }
}

代码逻辑如下:

  1. getBars方法由图表插件调用,带有当前需要渲染的时间参数、时间颗粒和callback。
  2. 如果缓存中有当前时间段的数据,构造newBars,调用onLoadedCallback(newBars)。
  3. 如果缓存中没有数据,判断当前需要的数据类型,如果是list历史记录,那么我们需要通过websocket来获取,执行initMessage。
  4. 判断当前缓存中的时间颗粒resolution是否已经改变,如果已经改变,需要清除缓存数据,更新缓存cacheData = list,返回图表onLoadedCallback(list),另外需要订阅当前时间颗粒的增量数据,取消上一次订阅。
  5. 如果时间颗粒resolution没有改变,websocket得到的历史数据不需要存入缓存,直接调用onLoadedCallback(list)就可以,因为图表插件的渲染是根本得到的数据时间段来执行的,即:图表插件会缓存已经渲染过的数据。
  6. 如果当前请求的数据为增量数据,但是缓存中没有,等待10ms,因为增量数据来源于websocket,我们已经订阅,需要等待websocket返回,相当于pending。

以上逻辑基本含括了用户的两个重要操作:切换时间颗粒、查看历史记录。
可运行代码在GitHub上已经更新,可预览

 类似资料: