我做了一个程序,实时接收原油期货的每笔交易信息。基本上,OnReceiveRealData
当事务执行并调用时执行real_get
方法。在该方法中,收集当前时间、价格和数量数据并用它们制作字典。有更多方法可以从实时流数据中制作 OHLC 格式数据,但我的问题是如何update自定义图形项目(烛台图)real_get
方法被调用?
class COM_Receiver(QAxWidget):
def __init__(self):
super().__init__()
self._create_com_instance()
self.list_items = ['CLF18']
self.OnReceiveRealData.connect(self.real_get)
self.OnEventConnect.connect(self._event_connect)
def _create_com_instance(self):
self.setControl("KFOPENAPI.KFOpenAPICtrl.1")
def _event_connect(self, err_code):
if err_code == 0:
print("connected")
self.real_set()
else:
print("disconnected")
self.login_event_loop.exit()
def connect(self):
self.dynamicCall("CommConnect(1)")
self.login_event_loop = QEventLoop()
self.login_event_loop.exec_()
def real_set(self):
self.dynamicCall("SetInputValue(str, str)", "itemCode", ';'.join(self.list_items))
ret = self.dynamicCall("CommRqData(str, str, str, str)", "itemCurrent", "opt10005", "", "1001")
def real_get(self, code, realtype, realdata):
if realtype == 'itemCurrent':
eventTime = ( datetime.utcnow() + timedelta(hours=2) )
currentPrice = self.dynamicCall("GetCommRealData(str, int)", "itemCurrent", 140)
currentVolume = self.dynamicCall("GetCommRealData(str, int)", "itemCurrent", 15)
dic_current = {'eventTime':eventTime, 'currentPrice':currentPrice, 'currentVolume':currentVolume}
self.make_ohlc(self.plt, dic_current)
if __name__ == "__main__":
app = QApplication(sys.argv)
c = COM_Receiver()
c.connect()
sys.exit(app.exec_())
我参考了这篇文章(使用 pyqtgraph 添加新数据栏的最快方法)并意识到我可以更新烛台而无需删除和创建CandlestickItem()
每当收到新数据时都会实例化(这就是我现在的做法,它消耗了大量资源)。
我尝试制作 CandlestickItem() 实例并用set_data
然而,它不能很好地工作,并且除非我单击图表(plt = pg.plot()),否则不会显示更新的烛台。也就是说,如果我离开程序大约 10 分钟,图表不会显示任何差异,但是一旦我单击图表,它就会立即显示最后 10 分钟的所有新烛台。我希望它实时显示新的烛台,即使我不连续单击图表。
我怎样才能随时更新自定义图形项目real_get
方法被调用OnReceiveRealData
事件在COM_Receiver
班级?我认为item.set_data(data=dic_current)
应该运行在real_get
方法,但到目前为止我所尝试的方法并没有达到我的预期。
下面的源是本文中烛台示例的完整源。
import pyqtgraph as pg
from pyqtgraph import QtCore, QtGui
import random
## Create a subclass of GraphicsObject.
## The only required methods are paint() and boundingRect()
## (see QGraphicsItem documentation)
class CandlestickItem(pg.GraphicsObject):
def __init__(self):
pg.GraphicsObject.__init__(self)
self.flagHasData = False
def set_data(self, data):
self.data = data ## data must have fields: time, open, close, min, max
self.flagHasData = True
self.generatePicture()
self.informViewBoundsChanged()
def generatePicture(self):
## pre-computing a QPicture object allows paint() to run much more quickly,
## rather than re-drawing the shapes every time.
self.picture = QtGui.QPicture()
p = QtGui.QPainter(self.picture)
p.setPen(pg.mkPen('w'))
w = (self.data[1][0] - self.data[0][0]) / 3.
for (t, open, close, min, max) in self.data:
p.drawLine(QtCore.QPointF(t, min), QtCore.QPointF(t, max))
if open > close:
p.setBrush(pg.mkBrush('r'))
else:
p.setBrush(pg.mkBrush('g'))
p.drawRect(QtCore.QRectF(t-w, open, w*2, close-open))
p.end()
def paint(self, p, *args):
if self.flagHasData:
p.drawPicture(0, 0, self.picture)
def boundingRect(self):
## boundingRect _must_ indicate the entire area that will be drawn on
## or else we will get artifacts and possibly crashing.
## (in this case, QPicture does all the work of computing the bouning rect for us)
return QtCore.QRectF(self.picture.boundingRect())
app = QtGui.QApplication([])
data = [ ## fields are (time, open, close, min, max).
[1., 10, 13, 5, 15],
[2., 13, 17, 9, 20],
[3., 17, 14, 11, 23],
[4., 14, 15, 5, 19],
[5., 15, 9, 8, 22],
[6., 9, 15, 8, 16],
]
item = CandlestickItem()
item.set_data(data)
plt = pg.plot()
plt.addItem(item)
plt.setWindowTitle('pyqtgraph example: customGraphicsItem')
def update():
global item, data
data_len = len(data)
rand = random.randint(0, len(data)-1)
new_bar = data[rand][:]
new_bar[0] = data_len
data.append(new_bar)
item.set_data(data)
app.processEvents() ## force complete redraw for every plot
timer = QtCore.QTimer()
timer.timeout.connect(update)
timer.start(100)
## Start Qt event loop unless running in interactive mode or using pyside.
if __name__ == '__main__':
import sys
if (sys.flags.interactive != 1) or not hasattr(QtCore, 'PYQT_VERSION'):
QtGui.QApplication.instance().exec_()