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

本篇文章會介紹如何利用技術指標進行自動化交易，並且以程式範例詳細說明。

參賽者將學習如何使用元富的技術分析API與交易API進行自動化交易，希望能幫助大家在競賽中獲得好成績。

文章尾巴會有完整程式碼，懶人的話可以直接跳到後半部。

台灣虛擬貨幣交易所排名，手續費折扣整理
 全球虛擬貨幣交易所排名，手續費折扣整理

目錄 Content 隱藏

執行訂單

結語

環境設置與套件安裝

首先，你需要安裝 MasterTradePy 和 tech_analysis_api_v2 及官網上的證券交易跟技術分析兩個套件。

這些套件提供了交易與技術分析的功能，能夠幫助你實現自動化交易策略。

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

初始化設定

設定參賽帳號、密碼以及目標股票代碼。

username = 'your_username'
password = 'your_password'
stock_id = '2330'

定義交易邏輯

這段程式碼的核心在於交易邏輯。這裡我們主要使用 KD 技術指標（KDJ）來進行交易決策。當 K 值上穿 D 值時，意味著黃金交叉，這是一個買入信號。

class TradingLogic:
    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

設置技術分析

使用技術分析 API 來設定並訂閱 KD 技術指標。

這裡我們以 5 分鐘的 K 線數據來計算 KD 值。也可以改接當下的報價（需要行情套件），但需要另外使用套件（如：TAlib）運算技術指標。

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)

執行訂單

當發現KD黃金交叉時，我們將進行市價買入操作。

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('下單失敗! 請再次執行程式，依據回報資料修正輸入')

主程式流程

整個程式的主要流程如下：

登入技術分析API並訂閱KD技術指標。
登入交易API，並進行雙因子認證。
當發現KD黃金交叉時，自動下單買入股票。

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')
    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()

結語

這篇文章展示了如何利用元富的API進行自動化交易的範例。希望透過這個範例，能夠更好地理解技術指標應用及自動化交易的流程。

【系列閱讀】元富證券系列文章

完整程式碼

以下是整個策略到下單的程式碼

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()

加入LINE社群「量化交易討論群」無壓力討論與分享！

加入Discord 「量化交易討論群」即時獲取實用的資源！

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

環境設置與套件安裝

初始化設定

定義交易邏輯

設置技術分析

執行訂單

主程式流程

結語

完整程式碼

量化通

發佈留言取消回覆

環境設置與套件安裝

初始化設定

定義交易邏輯

設置技術分析

執行訂單

主程式流程

結語

完整程式碼

量化通

相關文章

股票手續費計算，台股交易成本分析｜整筆與零股交易手續費、單筆最低手續費、證交稅

量化交易教學懶人包，自動交易從零開始，軟體平台選擇分享｜量化交易新手入門

Python程式交易教學懶人包，學Python量化交易必須知道的5件事｜新手入門

發佈留言取消回覆