[程式碼分享]用Python在虛擬交易環境中策略實戰｜股票量化交易從零開始（六）

from tech_analysis_api_v2.api import *
from tech_analysis_api_v2.model import *
import threading
import pandas as pd
import sys
from MasterTradePy.api import MasterTradeAPI
from MasterTradePy.model import *
from MasterTradePy.constant import PriceType, OrderType, TradingSession, Side, TradingUnit, RCode


# 請先安裝 MasterTradePy 套件 以及 技術分析 套件

event = threading.Event()
golden_cross_event = threading.Event()

## 請填寫參賽帳號
username = 'your_username'
## 填寫密碼
password = 'your_password'
## 股票代碼
stock_id = '2330'

class ConcreteMarketTrader(MarketTrader):
    def OnNewOrderReply(self, data) -> None:
        print(data)

    def OnChangeReply(self, data) -> None:
        print(data)

    def OnCancelReply(self, data) -> None:
        print(data)

    def OnReport(self, data) -> None:
        if type(data) is ReportOrder:
            if data.order.tableName == "ORD:TwsOrd":
                print(f'回報資料: 委託書號={data.order.ordNo}, 股票代號={data.order.symbol}, 委託股數={data.orgOrder.qty}, 成交股數={data.order.cumQty}, 訊息={data.lastMessage}, 狀態={data.order.status}')
            elif data.order.tableName == "RPT:TwsDeal":
                print(f'回報資料: 委託書號={data.order.ordNo}, 股票代號={data.order.symbol}, 成交價格={data.order.dealPri}, 成交股數={data.order.cumQty}, 剩餘股數={data.order.leavesQty} 訊息={data.lastMessage}, 狀態={data.order.status}')
            elif data.order.tableName == "RPT:TwsNew":
                print(f'回報資料: 委託書號={data.order.ordNo}, 股票代號={data.order.symbol}, 委託價格={data.orgOrder.price}, 委託股數={data.orgOrder.qty}, 訊息={data.lastMessage}, 狀態={data.order.status}')
            else:
                print(f'回報資料: 委託書號={data.order.ordNo}, 股票代號={data.order.symbol}, 委託價格={data.orgOrder.price}, 委託股數={data.orgOrder.qty}, 成交股數={data.order.cumQty}, 訊息={data.lastMessage}, 狀態={data.order.status}')
    
    def OnReqResult(self, workID: str, data) -> None:
        pass

    def OnSystemEvent(self, data: SystemEvent) -> None:
        print(f'OnSystemEvent{data}')
    
    def OnAnnouncementEvent(self, data)->None:
        print(f'OnAnnouncementEvent:{data}')
                  
    def OnError(self, data):
        print(data)

def OnDigitalSSOEvent(aIsOK, aMsg):
    print(f'OnDigitalSSOEvent: {aIsOK} {aMsg}')

def OnTAConnStuEvent(aIsOK):
    print(f'OnTAConnStuEvent: {aIsOK}')
    if aIsOK:
        event.set()

class TradingLogic:
    # 交易邏輯寫這邊，可以使用元富提供的技術指標API 又或者你自己想辦法使用其他套件來運算你的技術指標
    def __init__(self):
        self.previous_k = None
        self.previous_d = None

    def OnUpdate(self, ta_Type: eTA_Type, aResultPre, aResultLast):
        if aResultLast is not None:
            if ta_Type == eTA_Type.KD:
                current_k = aResultLast.K
                current_d = aResultLast.D
                print(f'最新 Time:{aResultLast.KBar.TimeSn_Dply}, K:{current_k}, D:{current_d}')

                if self.previous_k is not None and self.previous_d is not None:
                    if self.previous_k < self.previous_d and current_k > current_d:
                        print("發現KD黃金交叉，下單")
                        golden_cross_event.set()

                self.previous_k = current_k
                self.previous_d = current_d

    def OnRcvDone(self, ta_Type: eTA_Type, aResult):
        if ta_Type == eTA_Type.KD:
            df = pd.DataFrame([{
                'Time': x.KBar.TimeSn_Dply,
                'K': x.K,
                'D': x.D
            } for x in aResult])
            display_dataframe_to_user("KD技術指標數據", df)

    def get_kd_config(self):
        ProdID = stock_id
        SNK = '5'
        STA_Type = 'KD'
        DateBegin = (pd.Timestamp.now() - pd.Timedelta(days=1)).strftime('%Y%m%d')

        NK = eNK_Kind.K_5m

        TA_Type = eTA_Type.KD

        return TechAnalysis.get_k_setting(ProdID, TA_Type, NK, DateBegin)

def execute_order(api, stock_id):
    symbol = stock_id
    # 下單帳號(請填寫參賽帳號)
    account = 'your_account'
    price = ''  # 空白表示市價下單
    qty = '1000'  # 1張請輸入1000
    orderType = OrderType.ROD

    if not price:
        priceType = PriceType.MKT
    else:
        priceType = PriceType.LMT

    order = Order(tradingSession=TradingSession.NORMAL,
                  side=Side.Buy,
                  symbol=symbol,
                  priceType=priceType,
                  price=price,
                  tradingUnit=TradingUnit.COMMON, 
                  qty=qty,
                  orderType=orderType,
                  tradingAccount=account,
                  userDef='')
    rcode = api.NewOrder(order)
    if rcode == RCode.OK:
        print('已送出委託')
    else:
        print('下單失敗! 請再次執行程式，依據回報資料修正輸入')

def main():
    trading_logic = TradingLogic()
    ta = TechAnalysis(OnDigitalSSOEvent, OnTAConnStuEvent, trading_logic.OnUpdate, trading_logic.OnRcvDone)
    ta.Login(username, password)
    event.wait()

    k_config = trading_logic.get_kd_config()
    ta.SubTA(k_config)

    trader = ConcreteMarketTrader()
    api = MasterTradeAPI(trader)
    api.SetConnectionHost('solace140.masterlink.com.tw:55555')
    # {是否連測試(True/False)} {是否單一帳號通過強制登入(True/False)} {是否連接競賽主機(True/False)} 這邊是
    rc = api.Login(username, password, True, False, True)
    if rc == RCode.OK:
        print('交易主機連線成功，進行雙因子認證')
        accounts = [x[4:] for x in api.accounts]
        rcc = api.CheckAccs(tradingAccounts=accounts)
        if rcc == RCode.OK:
            print('驗證已通過 可執行API交易功能')

    golden_cross_event.wait()
    execute_order(api, stock_id)

    input("Press Enter to finish...\n")
    ta.UnSubTA(k_config)

main()