使用 matplotlib 为多个轴设置动画，仅在新数据到达时更新图形

Question

我将数据以 JSON 格式存储在 stock_data 中（可以是任意数据）。我想绘制 4 个轴，当有新数据时，更新图表（通过我假设的动画）。

我只希望在使用 INTRADAY 数据时发生这种情况（如您所见，我在底部有一个 if 盘中检查）。我正在从 API 中提取这些盘中数据。此 API 每分钟左右更新一次，并且仅在特定时间更新。我不介意它是否不会立即更新，但最好在 1 分钟内更新新数据。

我尝试提取新数据并将其与旧 DF 进行比较（如您在代码末尾看到的那样）并将其放入 while True: 循环中，但是图形无法呈现。我试过简单地将整个函数放在一个循环中并每次都渲染图形 - 这不仅需要很长时间才能渲染，而且如果我放大图形，它会完全重置它。我认为这是重绘的问题？

最后，我也不确定要在 animation.FuncAnimation() 中添加什么。我排除了 ax3 和 ax4，因为出于演示目的，它们的行为与 ax2 相同。非常感谢您的帮助。

import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker
import matplotlib.dates as mdates
import matplotlib.animation as animation
from mpl_finance import candlestick_ohlc
import numpy as np

## CANDLESTICK GRAPH ## 
def candlestick(symbol, MA1 = 20, MA2 = 200):
    try:
        ## arbitrary colors ##
        candle_upcol = '#cccccc'
        candle_downcol = '#cccccc'
        fill_col = '#cccccc'
        bg_col = '#cccccc'
        spine_col = '#cccccc'
        ## load stocks ##
        stock_data = pd.DataFrame.from_dict(json.load(open('db/AAPL.txt')), orient = 'index', dtype = np.float64)
        stock_data = stock_data.values
        ## BEGIN PLOTTING ##
        start_point = len(stock_data[max(MA1, MA2)-1:])
        fig = plt.figure(facecolor=bg_col)
        #set grids
        ax1 = plt.subplot2grid((8,4), (1,0), rowspan = 5, colspan = 4, facecolor = bg_col)
        ax2 = plt.subplot2grid((8,4), (7,0), rowspan = 1, colspan = 4, sharex = ax1, facecolor= bg_col)
        ax3 = plt.subplot2grid((8,4), (0,0), rowspan = 1, colspan = 4, sharex = ax1, facecolor = bg_col)
        ax4 = plt.subplot2grid((8,4), (6,0), rowspan = 1, colspan = 4, sharex = ax1, facecolor = bg_col)

        #PRICE plot (AX1)
        candlestick_ohlc(ax1, stock_data[-start_point:,0:5], width = 0.6, colorup = candle_upcol, colordown = candle_downcol)
        ax1.xaxis.set_major_locator(mticker.MaxNLocator(10))
        ax1.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d'))
        ax1.grid(True)
        plt.setp(ax1.get_xticklabels(), visible=False) #remove x ticks

        #MOVING AVERAGES plot (AX1)
        if MA1 != 0:
            av1 = moving_average(stock_data[:,4], MA1) #using close prices
            label_ma1 = '{MA} SMA'.format(MA = str(MA1))
            ax1.plot(stock_data[-start_point:,0], av1[-start_point:], label = label_ma1, color = '#aec6cf', linewidth = .8)
        if MA2 != 0:
            av2 = moving_average(stock_data[:,4], MA2) #using close prices
            label_ma2 = '{MA} SMA'.format(MA = str(MA2))
            ax1.plot(stock_data[-start_point:,0], av2[-start_point:], label = label_ma2, color = '#ffb347', linewidth = .8)
        if MA1 != 0 or MA2 != 0:
            ax1.text(0, 1, 'MA ({MA1}, {MA2})'.format(MA1 = str(MA1), MA2 = str(MA2)), va = 'top', ha = 'left', color = 'w', transform = ax1.transAxes, alpha = 0.5, fontweight = 'bold')

        #VOLUME plot (AX2)
        volume_min = 0 #stock_data[:,5].min()
        ax2.plot(stock_data[-start_point:,0], stock_data[-start_point:,5], '#00ffe8', linewidth = .8)
        ax2.fill_between(stock_data[-start_point:,0], volume_min, stock_data[-start_point:,5], facecolor = fill_col, alpha = 0.5)
        ax2.axes.yaxis.set_ticklabels([]) #remove y ticks
        ax2.text(0, 1, 'VOLUME', va = 'top', ha = 'left', color = 'w', transform = ax2.transAxes, alpha = 0.5, fontweight = 'bold')

        #RSI plot (AX3)
        #similar to VOL

        #MACD plot (AX4)
        #similar to VOL

        #SHARED plot (ALL AX)
        for all_ax in (ax1, ax2''', ax3, ax4'''):
            plt.setp(all_ax.spines.values(), color=spine_col)
            all_ax.tick_params(axis='both', colors = 'w')
            all_ax.yaxis.label.set_color("w")
            all_ax.yaxis.tick_right()
            all_ax.xaxis.set_tick_params(labelsize=9)
            all_ax.yaxis.set_tick_params(labelsize=9)

        #ENTIRE plot
        plt.subplots_adjust(hspace = 0)
        fig.autofmt_xdate()
        fig.suptitle('{STOCK}'.format(STOCK = symbol), color = 'w', fontweight='bold', alpha = 0.75)
        print('Drawing graph.')
        if data_type != 'Intraday':
            print('Graphing complete.')
        else:
            #this will be replaced by an API fetch function at some point, this is just for testing if animation works.. needs a sleep function? and while True loop..?
            new_stock_data = pd.DataFrame.from_dict(json.load(open('db/AAPL_new.txt')), orient = 'index', dtype = np.float64)
            new_stock_data = new_stock_data.values

            if (new_stock_data[-1] == stock_data[-1]).all() == False:
                stock_data = np.vstack([stock_data, new_stock_data[-1]])

            #ani = animation.FuncAnimation(fig, '''???''', interval = 10000) #blit=True?
        plt.show()
    except:
        print('Failed main loop.')

Answer 1

FuncAnimation 将以interval 给出的速率重复绘制（或 blit）。如果不需要，可以使用计时器来代替。计时器调用一个函数，根据某些条件，该函数要么什么都不做，要么用新数据更新绘图。这样，您可以确保仅在有新数据可用时才绘制画布（即条件为真）。

import datetime
import numpy as np
import matplotlib.pyplot as plt

fig, ax = plt.subplots()
t = []
x = []
line, = ax.plot_date(t,x, ls="-")

def update():
    now = datetime.datetime.now()
    if np.random.rand() > 0.9:
        t.append(now)
        x.append(np.random.randn())
        line.set_data(t,x)
        ax.relim()
        ax.autoscale_view()
        fig.canvas.draw_idle()
        message = "new data drawn"
    else:
        message = "no new data"
    print(now.time(), message)


timer = fig.canvas.new_timer(interval=200)
timer.add_callback(update)
timer.start()

plt.show()