【Python交易程式開發現場38】產生交易訊號 (2) | 相關程式碼分析

 

【交易程式架構】

1. 視窗物件區，view鈕，槽函數 【V】

2. (tw函數1) 呼叫API函數區，tw呼叫類函數（連線，查報價，下單，查倉位）【III.1】

3. (tw函數2) API事件回傳用，tw回傳處理類函數（顯示回傳訊息，存Tick值，畫Tick圖，計算指標，產生訊號，自動下單）【III.2】

4. API事件回傳區

5. 外部py函數檔(將.py整個檔視為一個副程式, 用import .py載入, 如: 畫Tick圖)

call_twMain.py程式碼主要分五部份:

I. Qt 自訂Signal/Slot 類別

II - 群益API物件 & 公用變數 & 工具函數

III. 群益API 事件所對應之相關 tw函數 (於V.槽函數所會用到的子工具函數區)

IV - 群益API事件

V - View | 主視窗類別 | 按鈕送出訊號所對應之槽函數

call_twMain.py

##############################################################################

# V - View | 主視窗類別 | 按鈕送出訊號所對應之槽函數

##############################################################################

class MyMainWindow(QMainWindow, Ui_Form):

    def __init__(self, parent=None):    

        super(MyMainWindow, self).__init__(parent)

        self.setupUi(self)

        #self.q_push_button = numpy.array([])

        self.p_button = []

        for i in range(10):

            btn = QPushButton("下單")

            btn.clicked.connect(self.twTest12)   #槽函數要用帶有self參數的內部函數, 不能直接叫用twOptionOrder()

            self.p_button.append(btn)

        self.pushButton.clicked.connect(self.twTest1) #ctw signal

        self.pushButton_2.clicked.connect(self.twTest2) #ctw signal

        self.pushButton_3.clicked.connect(self.twTest3) #ctw signal

        self.pushButton_4.clicked.connect(self.twTest8) #ctw signal

        self.pushButton_5.clicked.connect(self.twTest4) #ctw signal

        self.pushButton_6.clicked.connect(self.twTest5) #ctw signal      

        self.pushButton_7.clicked.connect(self.twTest6) #ctw signal

        self.pushButton_8.clicked.connect(self.twTest7) #ctw signal

        self.pushButton_9.clicked.connect(self.twTest9) #ctw signal    

        self.pushButton_10.clicked.connect(self.twTest10) #ctw signal  

        self.pushButton_11.clicked.connect(self.twTest11) #ctw signal    

        self.pushButton_12.clicked.connect(self.twTest12) #ctw signal  

     

    def twTest6(self):  #ctw slot function

        print("6-畫1分K")    

        self.listWidget.addItem("6-畫1分K")

        #self.listWidget.addItem("m_nCode: " + str(twDisconnect()))  

        import twCandleStick_History #直接將.py檔當副程式用(直接執行.py檔所有指令)

    # --------------------------> key point 3: Plot Tick

    def twTest7(self):  #ctw slot function

        print("7-畫Tick圖")    

        self.listWidget.addItem("7-畫Tick圖")

        #self.listWidget.addItem("m_nCode: " + str(twDisconnect()))

        import twPlotTick_History #直接將.py檔當副程式用(直接執行.py檔所有指令)

twPlotTick_History.py

from twIndicator import twClosePriceModel

import matplotlib as mpl

import pickle

import matplotlib.pyplot as plt

import twGlobal

tw_close = twGlobal.twClose

tw_kbar1m = twGlobal.twKBar1M

with open('twclose.pickle', 'rb') as f:

    tw_close = pickle.load(f)

from time import perf_counter

import numpy as np

import pyqtgraph as pg

win = pg.GraphicsLayoutWidget(show=True)

win.setWindowTitle('pyqtgraph example: Scrolling Plots')

# 1) Simplest approach -- update data in the array such that plot appears to scroll

#    In these examples, the array size is fixed.

p1 = win.addPlot()

#p2 = win.addPlot()

#data1 = np.random.normal(size=300)

showLength = 100   # Tick圖總長度

leftLength = 500   # 動態看最後leftLength筆Ticks

tickLengh = len(tw_close.GetClose())  #取得之所有Tick數目

print("Tick總數: ", tickLengh)

#data1 = tw_close.GetClose()[0:100]

# data1為顯示在Tick圖上之所有data

data1 = tw_close.GetClose()[(tickLengh-leftLength-showLength):(tickLengh-leftLength)]

print("Show Tick總數: ", len(data1))

#data2_close = tw_kbar1m.GetClose()[0:100]

#data2_high = tw_kbar1m.GetHigh()[0:100]

#data2_low = tw_kbar1m.GetLow()[0:100]

curve1 = p1.plot(data1)

#curve2 = p2.plot(data1)

#startTickID = 100

#ptr1 = 101

ptr1 = tickLengh-leftLength+1 #為了取得顯示在Tick圖上的Tick之後的下一個Tick

def update1():

    global data1, ptr1

    data1[:-1] = data1[1:]  # shift data in the array one sample left

                            #[:-1]意思是->取全部,除了最後一個

                            #[1:]意思是->取全部,除了第一個

                            #[-1]意思是->取最後一個

                            # (see also: np.roll)

    #data1[-1] = np.random.normal()

    data1[-1] = tw_close.GetClose()[ptr1]

    curve1.setData(data1)

    ptr1 += 1

    print("ptr1=", ptr1)

    #curve2.setData(data1)

    #curve2.setPos(ptr1, 0)

   

# update all plots

def update():

    update1()

    #update2()

    #update3()

timer = pg.QtCore.QTimer()

timer.timeout.connect(update)

timer.start(50)

# print(tw_close)    

#plt.axis([30,8000,16800,17000])  #會造成走勢線出不來?

#plt.plot(tw_close.GetTime(),tw_close.GetClose(),label="close")

#plt.plot(tw_close.GetTime(),tw_close.GetSMA(), label='sma')

"""

plt.plot(tw_close.GetClose(),label="close")

plt.plot(tw_close.GetSMA(), label='sma')

plt.legend()

plt.show()

"""

if __name__ == '__main__':

    pg.exec()