# -*- coding: utf-8 -*- # Qt相关和十字光标 from qtpy.QtGui import * from qtpy.QtCore import * from qtpy import QtWidgets, QtGui, QtCore from qtpy.QtWidgets import QApplication, QWidget from uiCrosshair import Crosshair import pyqtgraph as pg # 其他 import numpy as np import pandas as pd from functools import partial from datetime import datetime, timedelta from collections import deque # 字符串转换 # --------------------------------------------------------------------------------------- try: _fromUtf8 = QtCore.QString.fromUtf8 except AttributeError: def _fromUtf8(s): return s ######################################################################## # 键盘鼠标功能 ######################################################################## class KeyWraper(QtWidgets.QWidget): """键盘鼠标功能支持的元类""" # 初始化 # ---------------------------------------------------------------------- def __init__(self, parent=None): QtWidgets.QWidget.__init__(self, parent) # 重载方法keyPressEvent(self,event),即按键按下事件方法 # ---------------------------------------------------------------------- def keyPressEvent(self, event): if event.key() == QtCore.Qt.Key_Up: self.onUp() elif event.key() == QtCore.Qt.Key_Down: self.onDown() elif event.key() == QtCore.Qt.Key_Left: self.onLeft() elif event.key() == QtCore.Qt.Key_Right: self.onRight() elif event.key() == QtCore.Qt.Key_PageUp: self.onPre() elif event.key() == QtCore.Qt.Key_PageDown: self.onNxt() # 重载方法mousePressEvent(self,event),即鼠标点击事件方法 # ---------------------------------------------------------------------- def mousePressEvent(self, event): if event.button() == QtCore.Qt.RightButton: self.onRClick(event.pos()) elif event.button() == QtCore.Qt.LeftButton: self.onLClick(event.pos()) # 重载方法mouseReleaseEvent(self,event),即鼠标点击事件方法 # ---------------------------------------------------------------------- def mouseReleaseEvent(self, event): if event.button() == QtCore.Qt.RightButton: self.onRRelease(event.pos()) elif event.button() == QtCore.Qt.LeftButton: self.onLRelease(event.pos()) self.releaseMouse() # 重载方法wheelEvent(self,event),即滚轮事件方法 # ---------------------------------------------------------------------- def wheelEvent(self, event): if event.delta() > 0: self.onUp() else: self.onDown() # 重载方法dragMoveEvent(self,event),即拖动事件方法 # ---------------------------------------------------------------------- def paintEvent(self, event): self.onPaint() # PgDown键 # ---------------------------------------------------------------------- def onNxt(self): pass # PgUp键 # ---------------------------------------------------------------------- def onPre(self): pass # 向上键和滚轮向上 # ---------------------------------------------------------------------- def onUp(self): pass # 向下键和滚轮向下 # ---------------------------------------------------------------------- def onDown(self): pass # 向左键 # ---------------------------------------------------------------------- def onLeft(self): pass # 向右键 # ---------------------------------------------------------------------- def onRight(self): pass # 鼠标左单击 # ---------------------------------------------------------------------- def onLClick(self, pos): pass # 鼠标右单击 # ---------------------------------------------------------------------- def onRClick(self, pos): pass # 鼠标左释放 # ---------------------------------------------------------------------- def onLRelease(self, pos): pass # 鼠标右释放 # ---------------------------------------------------------------------- def onRRelease(self, pos): pass # 画图 # ---------------------------------------------------------------------- def onPaint(self): pass ######################################################################## # 选择缩放功能支持 ######################################################################## class CustomViewBox(pg.ViewBox): # ---------------------------------------------------------------------- def __init__(self, parent, *args, **kwds): pg.ViewBox.__init__(self, *args, **kwds) self.parent = parent # 拖动放大模式 # self.setMouseMode(self.RectMode) ## 右键自适应 # ---------------------------------------------------------------------- def mouseClickEvent(self, ev): if ev.button() == QtCore.Qt.RightButton: self.autoRange() # 重载方法mousePressEvent(self,event),即鼠标点击事件方法 # ---------------------------------------------------------------------- def mousePressEvent(self, event): pg.ViewBox.mousePressEvent(self, event) # 重载方法mouseDragEvent(self,event),即拖动事件方法 def mouseDragEvent(self, ev, axis=None): # if ev.start==True and ev.finish==False: ##判断拖拽事件是否结束 pos = ev.pos() lastPos = ev.lastPos() dif = pos - lastPos rect = self.sceneBoundingRect() pianyi = dif.x() * self.parent.countK * 2 / rect.width() self.parent.index -= int(pianyi) self.parent.index = max(self.parent.index, 60) xMax = self.parent.index + self.parent.countK ## xMin = self.parent.index - self.parent.countK if xMin < 0: xMin = 0 # self.parent.plotAll(False, xMin, xMax) #注释原因：拖动事件不需要先绘制图形界面 pg.ViewBox.mouseDragEvent(self, ev, axis) # ## 重载方法resizeEvent(self, ev) def resizeEvent(self, ev): self.linkedXChanged() self.linkedYChanged() self.updateAutoRange() self.updateViewRange() self._matrixNeedsUpdate = True self.sigStateChanged.emit(self) self.background.setRect(self.rect()) self.sigResized.emit(self) self.parent.refreshHeight() ######################################################################## # 时间序列，横坐标支持 ######################################################################## class MyStringAxis(pg.AxisItem): """时间序列横坐标支持""" # 初始化 # ---------------------------------------------------------------------- def __init__(self, xdict, *args, **kwargs): pg.AxisItem.__init__(self, *args, **kwargs) self.minVal = 0 self.maxVal = 0 self.xdict = xdict self.x_values = np.asarray(xdict.keys()) self.x_strings = xdict.values() self.setPen(color=(255, 255, 255, 255), width=0.8) self.setStyle(tickFont=QFont("Roman times", 10, QFont.Bold), autoExpandTextSpace=True) # 更新坐标映射表 # ---------------------------------------------------------------------- def update_xdict(self, xdict): self.xdict.update(xdict) self.x_values = np.asarray(self.xdict.keys()) self.x_strings = self.xdict.values() # 将原始横坐标转换为时间字符串,第一个坐标包含日期 # ---------------------------------------------------------------------- def tickStrings(self, values, scale, spacing): strings = [] for v in values: vs = v * scale if vs in self.x_values: vstr = self.x_strings[np.abs(self.x_values - vs).argmin()] if (isinstance(vstr, (str))): vstr = vstr else: vstr = vstr.strftime('%Y-%m-%d %H:%M:%S') else: vstr = "" strings.append(vstr) return strings ######################################################################## # K线图形对象 ######################################################################## class CandlestickItem(pg.GraphicsObject): """K线图形对象""" # 初始化 # ---------------------------------------------------------------------- def __init__(self, data): """初始化""" pg.GraphicsObject.__init__(self) # 数据格式: [ (time, open, close, low, high),...] self.data = data # 只重画部分图形，大大提高界面更新速度 self.setFlag(self.ItemUsesExtendedStyleOption) # 画笔和画刷 w = 0.4 self.offset = 0 self.low = 0 self.high = 1 self.picture = QtGui.QPicture() self.pictures = [] self.bPen = pg.mkPen(color=(0, 240, 240, 255), width=w * 2) self.bBrush = pg.mkBrush((0, 240, 240, 255)) self.rPen = pg.mkPen(color=(255, 60, 60, 255), width=w * 2) self.rBrush = pg.mkBrush((255, 60, 60, 255)) self.rBrush.setStyle(Qt.NoBrush) # 刷新K线 self.generatePicture(self.data) # 画K线 # ---------------------------------------------------------------------- def generatePicture(self, data=None, redraw=False): """重新生成图形对象""" # 重画或者只更新最后一个K线 if redraw: self.pictures = [] elif self.pictures: self.pictures.pop() w = 0.4 bPen = self.bPen bBrush = self.bBrush rPen = self.rPen rBrush = self.rBrush low, high = (data[0]['low'], data[0]['high']) if len(data) > 0 else (0, 1) for (t, open0, close0, low0, high0) in data: if t >= len(self.pictures): tShift = t low, high = (min(low, low0), max(high, high0)) picture = QtGui.QPicture() p = QtGui.QPainter(picture) # # 下跌蓝色（实心）, 上涨红色（空心） pen, brush, pmin, pmax = (bPen, bBrush, close0, open0) if open0 > close0 else (rPen, rBrush, open0, close0) p.setPen(pen) p.setBrush(brush) # 画K线方块和上下影线 if pmin > low0: p.drawLine(QtCore.QPointF(tShift, low0), QtCore.QPointF(tShift, pmin)) if high0 > pmax: p.drawLine(QtCore.QPointF(tShift, pmax), QtCore.QPointF(tShift, high0)) p.drawRect(QtCore.QRectF(tShift - w, open0, w * 2, close0 - open0)) # if open0 == close0: # p.drawRect(QtCore.QPointF(tShift - w, open0), QtCore.QPointF(tShift + w, close0)) # else: # p.drawRect(QtCore.QRectF(tShift - w, open0, w * 2, close0 - open0)) # if pmin > low0: # p.drawLine(QtCore.QPointF(tShift, low0), QtCore.QPointF(tShift, pmin)) # if high0 > pmax: # p.drawLine(QtCore.QPointF(tShift, pmax), QtCore.QPointF(tShift, high0)) p.end() self.pictures.append(picture) self.low, self.high = low, high # 手动重画 # ---------------------------------------------------------------------- def update(self): if not self.scene() is None: self.scene().update() # 自动重画 # ---------------------------------------------------------------------- def paint(self, p, o, w): rect = o.exposedRect xmin, xmax = (max(0, int(rect.left())), min(len(self.pictures), int(rect.right()))) [p.drawPicture(0, 0, pic) for pic in self.pictures[xmin:xmax]] # 定义边界 # ---------------------------------------------------------------------- def boundingRect(self): return QtCore.QRectF(0, self.low, len(self.pictures), (self.high - self.low)) ######################################################################## class KLineWidget(KeyWraper): """用于显示价格走势图""" # 保存K线数据的列表和Numpy Array对象 listBar = [] listVol = [] listHigh = [] listLow = [] listSig = [] listOpenInterest = [] arrows = [] # 是否完成了历史数据的读取 initCompleted = False # ---------------------------------------------------------------------- def __init__(self, parent=None, name=None): """Constructor""" self.parent = parent self.name = name super(KLineWidget, self).__init__(parent) # 当前序号 self.index = None # 下标 self.countK = 60 # 显示的Ｋ线范围 self.oldsize=0#rectd的hieght # 缓存数据 self.datas = [] self.listBar = [] self.listVol = [] self.listHigh = [] self.listLow = [] self.listSig = [] self.listOpenInterest = [] self.arrows = [] self.sars = [] # 所有K线上信号图 self.allColor = deque(['blue', 'green', 'yellow', 'white']) self.sigData = {} self.sigColor = {} self.sigPlots = {} # 初始化完成 self.initCompleted = False # 调用函数 self.initUi() # ---------------------------------------------------------------------- # 初始化相关 # ---------------------------------------------------------------------- def initUi(self): """初始化界面""" self.setWindowTitle(u'K线工具') # 主图 self.pw = pg.PlotWidget() # 界面布局 self.lay_KL = pg.GraphicsLayout(border=(100, 100, 100)) self.lay_KL.setContentsMargins(10, 10, 10, 10) self.lay_KL.setSpacing(0) self.lay_KL.setBorder(color=(255, 255, 255, 255), width=0.8) self.lay_KL.setZValue(0) self.lay_KL.setMinimumHeight(140) self.KLtitle = self.lay_KL.addLabel(u'') self.pw.setCentralItem(self.lay_KL) # 设置横坐标 xdict = {} self.axisTime = MyStringAxis(xdict, orientation='bottom') # 初始化子图 self.initplotKline() self.initplotVol() self.initplotOI() # 注册十字光标 self.crosshair = Crosshair(self.pw, self) # 设置界面 self.vb = QtWidgets.QVBoxLayout() self.vb.addWidget(self.pw) self.setLayout(self.vb) # 初始化完成 self.initCompleted = True self.oldsize=self.rect().height() self.customBox = {} # ---------------------------------------------------------------------- def makePI(self, name): """生成PlotItem对象""" vb = CustomViewBox(self) plotItem = pg.PlotItem(viewBox=vb, name=name, axisItems={'bottom': self.axisTime}) plotItem.setMenuEnabled(False) plotItem.setClipToView(True) plotItem.hideAxis('left') plotItem.showAxis('right') plotItem.setDownsampling(mode='peak') plotItem.setRange(xRange=(0, 1), yRange=(0, 1)) plotItem.getAxis('right').setWidth(60) plotItem.getAxis('right').setStyle(tickFont=QFont("Roman times", 10, QFont.Bold)) plotItem.getAxis('right').setPen(color=(255, 255, 255, 255), width=0.8) plotItem.showGrid(True, True) plotItem.hideButtons() return plotItem # ---------------------------------------------------------------------- def initplotVol(self): """初始化成交量子图""" self.pwVol = self.makePI('PlotVol' + self.name) self.volume = CandlestickItem(self.listVol) self.pwVol.addItem(self.volume) self.pwVol.setMaximumHeight((self.rect().height()-30)/4) self.pwVol.setMinimumHeight(12) self.pwVol.setXLink('PlotOI' + self.name) self.pwVol.hideAxis('bottom') self.lay_KL.nextRow() self.lay_KL.addItem(self.pwVol) self.lay_KL.adjustSize() # ---------------------------------------------------------------------- def initplotKline(self): """初始化K线子图""" self.pwKL = self.makePI('PlotKL' + self.name) self.candle = CandlestickItem(self.listBar) self.pwKL.addItem(self.candle) self.pwKL.setXLink('PlotOI' + self.name) self.pwKL.hideAxis('bottom') self.pwKL.setMinimumHeight((self.rect().height()-30)/3) self.lay_KL.nextRow() self.lay_KL.addItem(self.pwKL) # ---------------------------------------------------------------------- def initplotOI(self): """初始化持仓量子图""" self.pwOI = self.makePI('PlotOI' + self.name) self.curveOI = self.pwOI.plot() self.pwOI.setMaximumHeight((self.rect().height() - 30) / 4) self.pwOI.setMinimumHeight(20) self.lay_KL.nextRow() self.lay_KL.addItem(self.pwOI) # ---------------------------------------------------------------------- # 画图相关 # ---------------------------------------------------------------------- def plotVol(self, redraw=False, xmin=0, xmax=-1): """重画成交量子图""" if self.initCompleted: self.volume.generatePicture(self.listVol[xmin:xmax], redraw) # 画成交量子图 # ---------------------------------------------------------------------- def plotKline(self, redraw=False, xmin=0, xmax=-1): """重画K线子图""" if self.initCompleted: self.candle.generatePicture(self.listBar[xmin:xmax], redraw) # 画K线 self.plotMark() # 显示开平仓信号位置 # ---------------------------------------------------------------------- def plotOI(self, xmin=0, xmax=-1): """重画持仓量子图""" if self.initCompleted: # self.curveOI.setData(self.listOpenInterest[xmin:xmax]+[0], symbol='o', name="OpenInterest") self.curveOI.setData(self.listOpenInterest[xmin:xmax] + [0], name="OpenInterest" + self.name) # 去除symbol='o' def updateIndicatorSar(self, sarDatas, startIndex): """更新sar指标 startIndex 从0到 len -1""" if len(sarDatas) == 0: return while (len(self.sars) > (startIndex - 1)): sar = self.sars[-1] self.pwKL.removeItem(sar) self.sars.remove(sar) for i in range(startIndex, len(sarDatas)): arrow = pg.ArrowItem(pos=(i, sarDatas[i]), angle=90, tipAngle=5, headLen=2, tailWidth=4, pen={'width': 1}, brush=(0, 0, 255)) defaultOpts = { 'pxMode': True, 'angle': -150, ## If the angle is 0, the arrow points left 'pos': (0, 0), 'headLen': 10, 'tipAngle': 10, 'baseAngle': 0, 'tailLen': None, 'tailWidth': 3, 'pen': (200, 200, 200), 'brush': (255, 0, 0), } # arrow.setStyle(defaultOpts) self.pwKL.addItem(arrow) self.sars.append(arrow) def addIndicatorSar(self, sarDatas): """增加sar指标""" if len(sarDatas) == 0: return for sar in self.sars: self.pwKL.removeItem(sar) # 画信号 for i in range(len(sarDatas)): # arrow = pg.ArrowItem(pos=(i, sarDatas[i]),headLen=10, angle=90, brush=(0, 0, 255)) arrow = pg.ArrowItem(pos=(i, sarDatas[i]), angle=90, tipAngle=5, headLen=2, tailWidth=4, pen={'width': 1}, brush=(0, 0, 255)) defaultOpts = { 'pxMode': True, 'angle': -150, ## If the angle is 0, the arrow points left 'pos': (0, 0), 'headLen': 10, 'tipAngle': 10, 'baseAngle': 0, 'tailLen': None, 'tailWidth': 3, 'pen': (200, 200, 200), 'brush': (255, 0, 0), } # arrow.setStyle(defaultOpts) self.pwKL.addItem(arrow) self.sars.append(arrow) # ---------------------------------------------------------------------- def addSig(self, sig): """新增信号图""" if sig in self.sigPlots: self.pwKL.removeItem(self.sigPlots[sig]) self.sigPlots[sig] = self.pwKL.plot() self.sigColor[sig] = self.allColor[0] self.allColor.append(self.allColor.popleft()) # ---------------------------------------------------------------------- def showSig(self, datas): """刷新信号图""" for sig in self.sigPlots: self.sigData[sig] = datas[sig] [self.sigPlots[sig].setData(datas[sig], pen=self.sigColor[sig][0], name=sig) for sig in self.sigPlots] # if sig in datas] # ---------------------------------------------------------------------- def plotMark(self): """显示开平仓信号""" # 检查是否有数据 if len(self.datas) == 0: return for arrow in self.arrows: self.pwKL.removeItem(arrow) # 画买卖信号 for i in range(len(self.listSig)): # 无信号 if self.listSig[i] == None: continue # 买信号 elif self.listSig[i] != None: direction = self.listSig[i]["direction"] offset = self.listSig[i]["offset"] price = self.listSig[i]["price"] if direction == "空" and offset == "开仓": # arrow = pg.ArrowItem(pos=(i, price), angle=-90, brush=(255, 0, 0)) arrow = pg.ArrowItem(pos=(i, price), angle=180, tipAngle=60, headLen=8, tailLen=3, tailWidth=5, pen={'color': 'w', 'width': 1}, brush='r') elif direction == "多" and offset == "开仓": # arrow = pg.ArrowItem(pos=(i, price), angle=90, brush=(255, 0, 0)) arrow = pg.ArrowItem(pos=(i, price), angle=180, tipAngle=60, headLen=8, tailLen=3, tailWidth=5, pen={'color': 'w', 'width': 1}, brush='b') elif direction == "空" and offset == "平仓": # arrow = pg.ArrowItem(pos=(i, price), angle=-90, brush=(0, 0, 255)) arrow = pg.ArrowItem(pos=(i, price), angle=0, tipAngle=40, headLen=8, tailLen=None, tailWidth=8, pen={'color': 'w', 'width': 1}, brush='y') elif direction == "多" and offset == "平仓": # arrow = pg.ArrowItem(pos=(i, price), angle=90, brush=(0, 0, 255)) arrow = pg.ArrowItem(pos=(i, price), angle=0, tipAngle=40, headLen=8, tailLen=None, tailWidth=8, pen={'color': 'w', 'width': 1}, brush='y') self.pwKL.addItem(arrow) self.arrows.append(arrow) # ---------------------------------------------------------------------- def updateAll(self): """ 手动更新所有K线图形，K线播放模式下需要 """ datas = self.datas self.volume.update() self.candle.update() def update(view, low, high): vRange = view.viewRange() xmin = max(0, int(vRange[0][0])) xmax = max(0, int(vRange[0][1])) xmax = min(xmax, len(datas)) if len(datas) > 0 and xmax > xmin: ymin = min(datas[xmin:xmax][low]) ymax = max(datas[xmin:xmax][high]) if ymin and ymax: view.setRange(yRange=(ymin, ymax)) else: view.setRange(yRange=(0, 1)) update(self.pwKL.getViewBox(), 'low', 'high') update(self.pwVol.getViewBox(), 'volume', 'volume') #update(self.pwOI.getViewBox(), 'openInterest', 'openInterest') update(self.curveOI.getViewBox(), 'openInterest', 'openInterest') # ---------------------------------------------------------------------- def plotAll(self, redraw=True, xMin=0, xMax=-1): """ 重画所有界面 redraw ：False=重画最后一根K线; True=重画所有 xMin,xMax : 数据范围 """ # xMax = len(self.datas) if xMax < 0 else xMax # self.countK = xMax-xMin # self.index = int((xMax+xMin)/2) if redraw: xmax = len(self.datas) if xMax < 0 else xMax xmin=max(0,xmax-self.countK) self.index = int((xmax + xmin) / 2) self.pwOI.setLimits(xMin=xMin, xMax=xMax) self.pwKL.setLimits(xMin=xMin, xMax=xMax) self.pwVol.setLimits(xMin=xMin, xMax=xMax) self.plotKline(redraw, xMin, xMax) # K线图 self.plotVol(redraw, xMin, xMax) # K线副图，成交量 self.plotOI(0, len(self.datas)) # K线副图，持仓量 self.refresh() def refreshHeight(self): # super.__init__(QResizeEvent) # 如果窗口最大化，不修改比例，防御性设计 # if self.isMaximized(): # return if len(self.datas)!=0: height =self.rect().height() if height!=self.oldsize: self.oldsize=height height=(height-30)/4 self.pwKL.setMinimumHeight(height * 2-24) self.pwVol.setMaximumHeight(height) self.pwVol.setMinimumHeight(12) self.pwOI.setMaximumHeight(height) self.pwOI.setMinimumHeight(12) #print height # ---------------------------------------------------------------------- def refresh(self): """ 刷新三个子图的现实范围 """ datas = self.datas minutes = int(self.countK / 2) xmin = max(0, self.index - minutes) xmax = xmin + 2 * minutes self.pwOI.setRange(xRange=(xmin, xmax)) self.pwKL.setRange(xRange=(xmin, xmax)) self.pwVol.setRange(xRange=(xmin, xmax)) # ---------------------------------------------------------------------- # 快捷键相关 # ---------------------------------------------------------------------- def onNxt(self): """跳转到下一个开平仓点""" if len(self.listSig) > 0 and not self.index is None: datalen = len(self.listSig) self.index += 1 while self.index < datalen and self.listSig[self.index] == 0: self.index += 1 self.refresh() x = self.index y = self.datas[x]['close'] self.crosshair.signal.emit((x, y)) # ---------------------------------------------------------------------- def onPre(self): """跳转到上一个开平仓点""" if len(self.listSig) > 0 and not self.index is None: self.index -= 1 while self.index > 0 and self.listSig[self.index] == 0: self.index -= 1 self.refresh() x = self.index y = self.datas[x]['close'] self.crosshair.signal.emit((x, y)) # ---------------------------------------------------------------------- def onDown(self): """放大显示区间""" self.countK = min(len(self.datas), int(self.countK * 1.2) + 1) self.refresh() if len(self.datas) > 0: x = self.index - self.countK / 2 + 2 if int( self.crosshair.xAxis) < self.index - self.countK / 2 + 2 else int(self.crosshair.xAxis) x = self.index + self.countK / 2 - 2 if x > self.index + self.countK / 2 - 2 else x x=min(x,len(self.datas)-1) y = self.datas[x][2] # y = self.datas[x]['close'] self.crosshair.signal.emit((x, y)) # ---------------------------------------------------------------------- def onUp(self): """缩小显示区间""" # self.countK = max(3,int(self.countK/1.2)-1) self.countK = max(20, int(self.countK / 1.2) - 1) # 最小显示k线范围20 self.refresh() if len(self.datas) > 0: x = self.index - self.countK / 2 + 2 if int( self.crosshair.xAxis) < self.index - self.countK / 2 + 2 else int(self.crosshair.xAxis) x = self.index + self.countK / 2 - 2 if x > self.index + self.countK / 2 - 2 else x x = min(x, len(self.datas)-1) y = self.datas[x]['close'] self.crosshair.signal.emit((x, y)) # ---------------------------------------------------------------------- def onLeft(self): """向左移动""" if len(self.datas) > 0 and int(self.crosshair.xAxis) > 2: x = int(self.crosshair.xAxis) - 1 y = self.datas[x]['close'] if x <= self.index - self.countK / 2 + 2 and self.index > 1: self.index -= 1 self.refresh() self.crosshair.signal.emit((x, y)) # ---------------------------------------------------------------------- def onRight(self): """向右移动""" if len(self.datas) > 0 and int(self.crosshair.xAxis) < len(self.datas) - 1: x = int(self.crosshair.xAxis) + 1 y = self.datas[x]['close'] if x >= self.index + int(self.countK / 2) - 2: self.index += 1 self.refresh() self.crosshair.signal.emit((x, y)) # ---------------------------------------------------------------------- # 界面回调相关 # ---------------------------------------------------------------------- def onPaint(self): """界面刷新回调""" view = self.pwKL.getViewBox() vRange = view.viewRange() # xmin = max(0,int(vRange[0][0])) # xmax = max(0,int(vRange[0][1])) # self.index = int((xmin+xmax)/2)+1 # ---------------------------------------------------------------------- def resignData(self, datas): """更新数据，用于Y坐标自适应""" self.crosshair.datas = datas def viewXRangeChanged(low, high, self): vRange = self.viewRange() xmin = max(0, int(vRange[0][0])) xmax = max(0, int(vRange[0][1])) xmax = min(xmax, len(datas)) if len(datas) > 0 and xmax > xmin: ymin = min(datas[xmin:xmax][low]) ymax = max(datas[xmin:xmax][high]) if ymin and ymax: self.setRange(yRange=(ymin, ymax)) else: self.setRange(yRange=(0, 1)) view = self.pwKL.getViewBox() view.sigXRangeChanged.connect(partial(viewXRangeChanged, 'low', 'high')) view = self.pwVol.getViewBox() view.sigXRangeChanged.connect(partial(viewXRangeChanged, 'volume', 'volume')) view = self.pwOI.getViewBox() view.sigXRangeChanged.connect(partial(viewXRangeChanged, 'openInterest', 'openInterest')) # ---------------------------------------------------------------------- # 数据相关 # ---------------------------------------------------------------------- def clearData(self): """清空数据""" # 清空数据，重新画图 self.time_index = [] self.listBar = [] self.listVol = [] self.listLow = [] self.listHigh = [] self.listOpenInterest = [] self.listSig = [] self.sigData = {} self.arrows = [] self.datas = None # ---------------------------------------------------------------------- def updateSig(self, sig): """刷新买卖信号""" self.listSig = sig self.plotMark() # ---------------------------------------------------------------------- def onBar(self, bar, nWindow=20): """ 新增K线数据,K线播放模式 nWindow : 最大数据窗口 """ # 是否需要更新K线 newBar = False if len(self.datas) > 0 and bar.datetime == self.datas[-1].datetime else True nrecords = len(self.datas) if newBar else len(self.datas) - 1 bar.openInterest = np.random.randint(0, 3) if bar.openInterest == np.inf or bar.openInterest == -np.inf else bar.openInterest recordVol = (nrecords, bar.volume, 0, 0, bar.volume) if bar.close < bar.open else ( nrecords, 0, bar.volume, 0, bar.volume) if newBar and any(self.datas): self.datas.resize(nrecords + 1, refcheck=0) self.listBar.resize(nrecords + 1, refcheck=0) self.listVol.resize(nrecords + 1, refcheck=0) elif any(self.datas): self.listLow.pop() self.listHigh.pop() self.listOpenInterest.pop() if any(self.datas): self.datas[-1] = (bar.datetime, bar.open, bar.close, bar.low, bar.high, bar.volume, bar.openInterest) self.listBar[-1] = (nrecords, bar.open, bar.close, bar.low, bar.high) self.listVol[-1] = recordVol else: self.datas = np.rec.array( [(datetime, bar.open, bar.close, bar.low, bar.high, bar.volume, bar.openInterest)], names=('datetime', 'open', 'close', 'low', 'high', 'volume', 'openInterest')) self.listBar = np.rec.array([(nrecords, bar.open, bar.close, bar.low, bar.high)], names=('datetime', 'open', 'close', 'low', 'high')) self.listVol = np.rec.array([recordVol], names=('datetime', 'open', 'close', 'low', 'high')) self.resignData(self.datas) self.axisTime.update_xdict({nrecords: bar.datetime}) self.listLow.append(bar.low) self.listHigh.append(bar.high) self.listOpenInterest.append(bar.openInterest) # if newBar: # xMax = nrecords + 1 # xMin = max(0, xMax - self.countK) # 最小显示区间 # self.index=int(xMax+xMin)/2 # if self.pwKL.getViewBox().mousePress == True: # print("sssssssssss") # if self.pwKL.getViewBox().mousePress == False: ##self.mousePress==false,不在执行刷新操作 xMax = self.index + self.countK xMin = self.index - self.countK #xMin=(0,xMin) if xMin<0: xMin=0 if not newBar: self.updateAll() self.plotAll(False, xMin, xMax) self.crosshair.signal.emit((None, None)) # ---------------------------------------------------------------------- def loadData(self, datas): """ 载入pandas.DataFrame数据 datas : 数据格式，cols : datetime, open, close, low, high """ # 设置中心点时间 self.index = 0 # 绑定数据，更新横坐标映射，更新Y轴自适应函数，更新十字光标映射 datas['time_int'] = np.array(range(len(datas.index))) self.datas = datas[['open', 'close', 'low', 'high', 'volume', 'openInterest']].to_records() self.axisTime.xdict = {} xdict = dict(enumerate(datas.index.tolist())) self.axisTime.update_xdict(xdict) self.resignData(self.datas) # 更新画图用到的数据 self.listBar = datas[['time_int', 'open', 'close', 'low', 'high']].to_records(False) self.listHigh = list(datas['high']) self.listLow = list(datas['low']) self.listOpenInterest = list(datas['openInterest']) # 成交量颜色和涨跌同步，K线方向由涨跌决定 datas0 = pd.DataFrame() datas0['open'] = datas.apply(lambda x: 0 if x['close'] >= x['open'] else x['volume'], axis=1) datas0['close'] = datas.apply(lambda x: 0 if x['close'] < x['open'] else x['volume'], axis=1) datas0['low'] = datas0['open'] datas0['high'] = datas0['close'] datas0['time_int'] = np.array(range(len(datas.index))) self.listVol = datas0[['time_int', 'open', 'close', 'low', 'high']].to_records(False) # 调用画图函数 self.plotAll(True, 0, len(self.datas)) def loadDataBar(self, datas): """ 载入pandas.DataFrame数据 datas : 数据格式，'symbol','vtSymbol','exchange','open','high','low','close','date','time1','datetime','volume','openInterest' """ print(datas) self.index = 0 # 绑定数据，更新横坐标映射，更新Y轴自适应函数，更新十字光标映射 datas['time_int'] = np.array(range(len(datas.index))) self.datas = datas[['datetime', 'open', 'close', 'low', 'high', 'volume', 'openInterest']].to_records(False) # self.datas = datas[['symbol','vtSymbol','exchange','open','high','low','close','date','time1','datetime','volume','openInterest']].to_records() self.axisTime.xdict = {} # xdict = dict(enumerate(datas.index.tolist())) xdict = dict(enumerate(datas['datetime'].tolist())) self.axisTime.update_xdict(xdict) self.resignData(self.datas) # 更新画图用到的数据 self.listBar = datas[['time_int', 'open', 'close', 'low', 'high']].to_records(False) self.listHigh = list(datas['high']) self.listLow = list(datas['low']) self.listOpenInterest = list(datas['openInterest']) # 成交量颜色和涨跌同步，K线方向由涨跌决定 datas0 = pd.DataFrame() datas0['open'] = datas.apply(lambda x: 0 if x['close'] >= x['open'] else x['volume'], axis=1) datas0['close'] = datas.apply(lambda x: 0 if x['close'] < x['open'] else x['volume'], axis=1) datas0['low'] = datas0['open'] datas0['high'] = datas0['close'] datas0['time_int'] = np.array(range(len(datas.index))) self.listVol = datas0[['time_int', 'open', 'close', 'low', 'high']].to_records(False) # 调用画图函数 self.plotAll(True, 0, len(self.datas)) ##VtBarData 结构 def loadDataBarArray(self, record): txtData = pd.DataFrame.from_records(record, index="datetime") # txtData = txtData.rename(columns = {0:'symbol', 1:"vtSymbol",2:"exchange",3:"open",4:"high",5:"low",6:"close",7:"date",8:"time",9:"datetime",10:"volume",11:"openInterest"}) self.loadData(txtData) ######################################################################## # 功能测试 ######################################################################## import sys if __name__ == '__main__': app = QApplication(sys.argv) # 界面设置 cfgfile = QtCore.QFile('css.qss') cfgfile.open(QtCore.QFile.ReadOnly) styleSheet = cfgfile.readAll() styleSheet = unicode(styleSheet, encoding='utf8') app.setStyleSheet(styleSheet) # K线界面 # ui = KLineWidget() # ui.show() # ui.KLtitle.setText('rb1701',size='20pt') # ui.loadData(pd.DataFrame.from_csv('data.csv')) ui = KLineWidget(name="opt") ui.show() ui.KLtitle.setText('rb1701', size='20pt') # txtData = pd.DataFrame.from_csv('D:/data/day/rb1101.txt',header=None,index_col=7) txtData = pd.DataFrame.from_csv('E:/Data/day/rb1101.txt', header=None, index_col=7) txtData = txtData.rename( columns={0: 'symbol', 1: "vtSymbol", 2: "exchange", 3: "open", 4: "high", 5: "low", 6: "close", 7: "date", 8: "time", 9: "datetime", 10: "volume", 11: "openInterest"}) ui.loadDataBar(txtData) app.exec_()