Patch for source

[milikest]

Hello,
I am a newbie in blokeh, so this can be an easy problem for you perhaps. I am trying to write an app to display stock prices on browser.
In the code I created an source:

source.stream(dict(close_time=close_time, open=open, close=close, high=high, low=low, inc=inc, dec=dec))

And with a while loop I am trying to get new price in every 5 seconds and trying to update the price bar.
To do so first I code checks if new price is not same with previous price and if it is different than creates a patch dictionary:

if close_new[-1] != close[-1]:
    patches_1_1 ={'close': [(len(close)-1, close_new[-1])]}

Than, updating the source:

source.patch(patches_1_1)

When I write len(close) it throws an index error, which is strange, and I thought it can be different indexing in patch so I chosed (len(close)-1 . There is a len(close)-1 index eventually. This is not a problem in anyway.

But this code creates a very long error:

  File "C:\ProgramData\Anaconda3\lib\threading.py", line 932, in _bootstrap_inner
    self.run()
  File "C:\ProgramData\Anaconda3\lib\threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\milik\PycharmProjects\Interview\price.py", line 114, in blocking_task
    source.patch(patches_1_1)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\models\sources.py", line 684, in patch
    self.data._patch(self.document, self, patches, setter)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\core\property\wrappers.py", line 470, in _patch
    self._notify_owners(old,
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\core\property\wrappers.py", line 150, in _notify_owners
    descriptor._notify_mutated(owner, old, hint=hint)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\core\property\descriptors.py", line 869, in _notify_mutated
    self._real_set(obj, old, value, hint=hint)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\core\property\descriptors.py", line 832, in _real_set
    self._trigger(obj, old, value, hint=hint, setter=setter)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\core\property\descriptors.py", line 909, in _trigger
    obj.trigger(self.name, old, value, hint, setter)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\model.py", line 664, in trigger
    super().trigger(attr, old, new, hint=hint, setter=setter)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\util\callback_manager.py", line 157, in trigger
    self._document._notify_change(self, attr, old, new, hint, setter, invoke)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 1061, in _notify_change
    self._trigger_on_change(event)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 1156, in _trigger_on_change
    self._with_self_as_curdoc(invoke_callbacks)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 1169, in _with_self_as_curdoc
    return f()
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 1155, in invoke_callbacks
    cb(event)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 723, in <lambda>
    self._callbacks[receiver] = lambda event: event.dispatch(receiver)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\events.py", line 269, in dispatch
    super().dispatch(receiver)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\events.py", line 124, in dispatch
    receiver._document_patched(self)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\server\session.py", line 218, in _document_patched
    raise RuntimeError("_pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes")

In the source there are several columns but I need to update only one value in one column in a time.

What am I missing?

[Bryan]

Hi @milikest it’s not possible to speculate without a Minimal Reproducible Example

[milikest]

Hello, to be more clear, here is the code:
As u see get_bars gets the ohlc knowledge from binance api.

def get_bars(symbol='BTCUSDT', interval='1m', limit='500'):
    root_url = 'https://api.binance.com/api/v1/klines'
    url = root_url + '?symbol=' + symbol + '&interval=' + interval + '&limit=' + limit
    data = json.loads(requests.get(url).text)
    df = pd.DataFrame(data)
    df.columns = ['Open_time',
                  'Open', 'High', 'Low', 'Close', 'Volume',
                  'Close_time', 'Quote_asset_volume', 'Number_of_trades',
                  'Taker_buy_base_asset_volume', 'Taker_buy_quote_asset_volume', 'Ignore']
    df['Open'] = df['Open'].astype('float')
    df['High'] = df['High'].astype('float')
    df['Low'] = df['Low'].astype('float')
    df['Close'] = df['Close'].astype('float')
    df['Volume'] = df['Volume'].astype('float')
    df['Quote_asset_volume'] = df['Quote_asset_volume'].astype('float')
    df['Number_of_trades'] = df['Number_of_trades'].astype('float')
    df['Taker_buy_base_asset_volume'] = df['Taker_buy_base_asset_volume'].astype('float')
    df['Taker_buy_quote_asset_volume'] = df['Taker_buy_quote_asset_volume'].astype('float')
    df['Ignore'] = df['Ignore'].astype('float')

    df.index = [dt.datetime.fromtimestamp(x / 1000.0) for x in df.Close_time]
    df['pre_index'] = df.index
    df.index = pd.RangeIndex(len(df.index))
    return df


df = get_bars()

close_time = np.array(df['Close_time'])
open = np.array(df['Open'])
close = np.array(df['Close'])
high = np.array(df['High'])
low = np.array(df['Low'])
inc = np.array(df.Close > df.Open)
dec = np.array(df.Open > df.Close)
close_time_inc = close_time[inc]
close_time_dec = close_time[dec]
open_dec = open[dec]
open_inc = open[inc]
close_dec = close[dec]
close_inc = close[inc]

This is from the bokeh sample : candlestick.py — Bokeh 2.4.2 Documentation

I created 3 sources because they are going to have different lengths but similar in each:

source = ColumnDataSource(data=dict(close_time=close_time, open=open, close=close, high=high, low=low, inc=inc, dec=dec))
source_2 = ColumnDataSource(data=dict(close_time_inc=close_time_inc, open_inc=open_inc, close_inc=close_inc))
source_3 = ColumnDataSource(data=dict(close_time_dec=close_time_dec, open_dec=open_dec, close_dec=close_dec))

doc = curdoc()

@gen.coroutine

def update(close_time, open, close, high, low, inc, dec, close_time_inc, close_time_dec, open_dec, open_inc, close_dec,

           close_inc):

       source.stream(dict(close_time=close_time, open=open, close=close, high=high, low=low, inc=inc, dec=dec))

       source_2.stream(dict(close_time_inc=close_time_inc, open_inc=open_inc, close_inc=close_inc))       

       source_3.stream(dict(close_time_dec=close_time_dec, open_dec=open_dec, close_dec=close_dec)

And here is to while loop:
get_bars limit parameter how many bars it will take from api. limit=1 means get me the last ohlc knowledge of that interval. Interval can be 1minute, 5minute, 1 hour,…etc. df_new will get last ohlc values so in the chart it will be updating only last bar.

def blocking_task():
    while True:
        global df
        global close_time
        global open
        global close
        global high
        global low
        global inc
        global dec
        global close_time_inc
        global close_time_dec
        global open_dec
        global open_inc
        global close_dec
        global close_inc
        time.sleep(5)
        df_new = get_bars(limit='1')
        close_time_new = np.array(df_new['Close_time'])
        open_new = np.array(df_new['Open'])
        close_new = np.array(df_new['Close'])
        high_new = np.array(df_new['High'])
        low_new = np.array(df_new['Low'])
        inc_new = np.array(df_new.Close > df_new.Open)
        dec_new = np.array(df_new.Open > df_new.Close)
        close_time_inc_new = close_time_new[inc_new]
        close_time_dec_new = close_time_new[dec_new]
        open_dec_new = open_new[dec_new]
        open_inc_new = open_new[inc_new]
        close_dec_new = close_new[dec_new]
        close_inc_new = close_new[inc_new]
        if close_time_new[-1] == close_time[-1]:
            patches_1 = {}
            patches_2 = {}
            patches_3 = {}
            if close_new[-1] != close[-1]:
                patches_1_1 ={'close': [(len(close)-1, close_new[-1])]}
                if len(patches_1_1)!=0:
                    source.patch(patches_1_1)
            if high_new[-1] != high[-1]:
                patches_1_2 ={'high': [(len(high)-1, high_new[-1])]}
                patches_1.update(**patches_1_2)
                if len(patches_1_2)!=0:
                    source.patch(patches_1_2)
           .
           .
           .
           .

This if statements goes to end of ‘dec’ variable updating for source and source_2 and source_3 is being updated with same logic.
And when the requested timestamps passes to next time interval:

        else:
            doc.add_next_tick_callback(
                partial(update,close_time=close_time, open=open, close=close, high=high, low=low, inc=inc, dec=dec ,
                close_time_inc=close_time_inc, close_time_dec=close_time_dec, open_dec=open_dec, open_inc=open_inc,
                 close_dec=close_dec, close_inc=close_inc))
            df = pd.concat([df, df_new])
            close_time = np.array(df['Close_time'])
            open = np.array(df['Open'])
            close = np.array(df['Close'])
            high = np.array(df['High'])
            low = np.array(df['Low'])
            inc = np.array(df.Close > df.Open)
            dec = np.array(df.Open > df.Close)
            close_time_inc = close_time[inc]
            close_time_dec = close_time[dec]
            open_dec = open[dec]
            open_inc = open[inc]
            close_dec = close[dec]
            close_inc = close[inc]

Last prices (ohlc) are being added to df and df_new will be new values again.
And the plotting last part:

TOOLS = "pan,wheel_zoom,box_zoom,reset,save,crosshair, hover, poly_draw,lasso_select"
p = figure(x_axis_type="datetime", sizing_mode='scale_height', tools=TOOLS, plot_width=1250, title="Candlestick")

w = 12 * 60 * 60 * 1
p.xaxis.major_label_orientation = pi / 4
p.grid.grid_line_alpha = 0.3
p.segment(x0='close_time', y0='high', x1='close_time', y1='low', source=source, color="black")
p.vbar('close_time_inc', w/2, 'open_inc', 'close_inc', source=source_2, fill_color="#077C47", line_color="black")
p.vbar('close_time_dec', w/2, 'open_dec', 'close_dec', source=source_3, fill_color="#ED051F", line_color="black")

doc.add_root(p)

thread = Thread(target=blocking_task)
thread.start()

And for sure to run the server on terminal I type : bokeh serve --show myapp.py
I guess I exceeded the maximum description for a problem…
Sorry for that.
And thanks anyway.