iT邦幫忙

2023 iThome 鐵人賽
DAY 13
1
AI & Data

從零到英雄：用GCP建立AI交易體系系列 第 13 篇

Day13 API上雲(上)

15th鐵人賽
544 瀏覽

  • 分享至 

  • xImage
    •  

串接模型與下單

我們要上雲之前，還是需要串接模型而後進行下單。因此有請我們上次的程式碼：

from shioaji import TickSTKv1, Exchange, Shioaji, constant, BidAskSTKv1
from threading import Event # event模組用於維持程式運作

api = Shioaji()
accounts =  api.login(api_key="", 
                      secret_key=""
                      )

#訂閱個股盤中tick資訊
api.quote.subscribe(
    api.Contracts.Stocks["2330"], 
    quote_type = constant.QuoteType.Tick,
    version = constant.QuoteVersion.v1
)

# 維持程式運作等待交易資料送入
Event().wait()

# 定義quote_callback，即回傳報價資訊時所要執行的動作
@api.on_tick_stk_v1()
def quote_callback(exchange: Exchange, tick:TickSTKv1):
    print(f"Exchange: {exchange}, Tick: {tick}") #將報價資訊內容輸出

api.logout()

這一個程式碼會訂閱股票代號 2330 的股票，並且秀出報價資料。
但是我們目前的策略如下：

  • 如果模型給出1(買進)訊號時買入。
  • 如果已經買入，買入的標的達到標準差兩倍的價格時就賣出(不論是上漲或下跌)

因此我們的設計只需要取得昨天算出來可以交易的股票，並且在程式運作時維持監控停損即可。以下是需要的幾個API：

下單憑證

下單前要有憑證，在永豐官網那邊可以取得，今天先不要使用，避免胡亂下單。

result = api.activate_ca(
    ca_path=os.getenv('YOUR_CA_PATH'), # 下單電子憑證路徑及檔案名稱
    ca_passwd=os.getenv('YOUR_CA_PASS'), # 下單電子憑證密碼
    person_id=os.getenv('YOUR_PERSON_ID'), # 身份證字號
)

委託

下委託單前要先建立order物件，詳細的參數請前看兩天前的那一篇：

order = api.Order(price=,
				  quantity=,
				  action=,
				  price_type=,
				  order_type=,
				  order_cond=,
				  order_lot=,
				  first_sell=,
				  account=
				 )

委託建立之後要發送出去：

api.place_order(
        contract: Contract,
        order: Order,
        timeout: int = 5000
    )

定時查詢及時庫存，以進行監控停損停利。

api.list_positions(
    account: shioaji.account.Account = None, #交易帳戶，預設為None
    unit: shioaji.constant.Unit = <Unit.Common: 'Common'>, #單位，預設為整股
    timeout: int = 5000 #timeout預設為5000ms，即延遲時間?
)

我們整理一下需要的業務邏輯，我們會需要幾個物件去管理業務邏輯：

  1. 監控停損
  2. 開盤買進

其實還會需要：

  1. 資金控制
  2. 速度優化
    但是今天先不做這些。

開盤買進

一開始我們先假設已經取得要下單的股票代號列表，先讓程式可以開盤買進：

from shioaji import TickSTKv1, Exchange, Shioaji, constant, BidAskSTKv1
from threading import Event # event模組用於維持程式運作
from datetime import datetime

stock_list = ['2330']

api = Shioaji(simulation=True)
accounts =  api.login(api_key="", 
                      secret_key="",
                      )

def open_market_buy(stock_list):
    '''
    開盤買進，理論上程式會在開盤前就執行，因此我們在開盤前預掛委託單，
    '''
    if time_chack(): # 確定時間是否在開盤之前
        for stock_id in stock_list:
            # 製作委託單
            order = api.Order(price=api.Contracts.Stocks[stock_id].limit_up,
				  quantity=1,
				  action='Buy',
				  price_type='LMT', # 限價單
				  order_type='ROD',
				  order_cond='Cash',
				  order_lot='Common', # 整股
				  first_sell='false',
				  account=accounts[0]
				 )
            
            # 發送委託單
            api.place_order(
                contract=api.Contracts.Stocks[stock_id],
                order=order,
                timeout=5000
            )
            print(stock_id)
    else:
        print('開盤後')
        pass

def time_chack():
    '''
    確定時間是否在開盤之前
    '''
    # 獲取當前時間
    now = datetime.now()

    # 獲取當前的時和分
    current_hour = now.hour
    current_minute = now.minute

    # 如果啟動時已經開盤了就不做交易，如果還沒有開盤就執行交易。
    if current_hour < 9 or (current_hour == 9 and current_minute == 0):
        return True
    else:
        return False

open_market_buy(stock_list=stock_list)

# 維持程式運作等待交易資料送入
Event().wait()

# 定義quote_callback，即回傳報價資訊時所要執行的動作
@api.on_tick_stk_v1()
def quote_callback(exchange: Exchange, tick:TickSTKv1):
    print(f"Exchange: {exchange}, Tick: {tick}") #將報價資訊內容輸出

api.logout()

上面的程式碼我們先用這部分去完成時間的檢驗，如果已經開盤了我們就不交易了，如果沒有開盤我們才預掛委託單。

def time_chack():
    '''
    確定時間是否在開盤之前
    '''
    # 獲取當前時間
    now = datetime.now()

    # 獲取當前的時和分
    current_hour = now.hour
    current_minute = now.minute

    # 如果啟動時已經開盤了就不做交易，如果還沒有開盤就執行交易。
    if current_hour < 9 or (current_hour == 9 and current_minute == 0):
        return True
    else:
        return False

接著我們再用以下方法處理交易的部分，這裡要注意開盤前是不能掛市價單的，所以我們就掛漲停限價單。

def open_market_buy(stock_list):
    '''
    開盤買進，理論上程式會在開盤前就執行，因此我們在開盤前預掛委託單，
    '''
    if time_chack(): # 確定時間是否在開盤之前
        for stock_id in stock_list:
            # 製作委託單
            order = api.Order(price=api.Contracts.Stocks[stock_id].limit_up,
				  quantity=1,
				  action='Buy',
				  price_type='LMT', # 限價單
				  order_type='ROD',
				  order_cond='Cash',
				  order_lot='Common', # 整股
				  first_sell='false',
				  account=accounts[0]
				 )
            
            # 發送委託單
            api.place_order(
                contract=api.Contracts.Stocks[stock_id],
                order=order,
                timeout=5000
            )
            print(stock_id)
    else:
        print('開盤後')
        pass

接下來實作監控停損

監控停損

我們把判斷式寫再cellback函式中，並且訂閱相關需要的股票代號。

from shioaji import TickSTKv1, Exchange, Shioaji, constant, BidAskSTKv1
from threading import Event # event模組用於維持程式運作
from datetime import datetime

stock_list = ['2330']
sell_price = {'2330': {
    'high':400,
    'low':100
    }
}

api = Shioaji(simulation=True)
accounts =  api.login(api_key="", 
                      secret_key="",
                      )

def open_market_buy(stock_list):
    '''
    開盤買進，理論上程式會在開盤前就執行，因此我們在開盤前預掛委託單，
    '''
    if time_chack(): # 確定時間是否在開盤之前
        for stock_id in stock_list:
            # 製作委託單
            order = api.Order(price=api.Contracts.Stocks[stock_id].limit_up,
				  quantity=1,
				  action='Buy',
				  price_type='LMT', # 限價單
				  order_type='ROD',
				  order_cond='Cash',
				  order_lot='Common', # 整股
				  first_sell='false',
				  account=accounts[0]
				 )
            
            # 發送委託單
            api.place_order(
                contract=api.Contracts.Stocks[stock_id],
                order=order,
                timeout=5000
            )

            #為了監控買進的標的是否需要停損，所以訂閱tick資訊
            api.quote.subscribe(
                api.Contracts.Stocks[stock_id], 
                quote_type = constant.QuoteType.Tick,
                version = constant.QuoteVersion.v1
            )
            print(stock_id)
    else:
        print('開盤後')
        pass

def time_chack():
    '''
    確定時間是否在開盤之前
    '''
    # 獲取當前時間
    now = datetime.now()

    # 獲取當前的時和分
    current_hour = now.hour
    current_minute = now.minute

    # 如果啟動時已經開盤了就不做交易，如果還沒有開盤就執行交易。
    if current_hour < 9 or (current_hour == 9 and current_minute == 0):
        return True
    else:
        return False

open_market_buy(stock_list=stock_list)

# 為了要能夠監控盤中的個股狀況，我們需要訂閱需要監控的個股
for stock_id in stock_list:
    #訂閱個股盤中tick資訊
    api.quote.subscribe(
        api.Contracts.Stocks[stock_id], 
        quote_type = constant.QuoteType.Tick,
        version = constant.QuoteVersion.v1
    )

# 維持程式運作等待交易資料送入
Event().wait()

# 定義quote_callback，即回傳報價資訊時所要執行的動作
@api.on_tick_stk_v1()
def quote_callback(exchange: Exchange, tick:TickSTKv1):
    print(f"Exchange: {exchange}, Tick: {tick}") #將報價資訊內容輸出
    # 停利
    if tick.close >= sell_price[tick.code]['high']:
        # 製作委託單
        order = api.Order(price=api.Contracts.Stocks[tick.code].limit_down,
                quantity=1,
                action='Sell',
                price_type='LMT', # 限價單
                order_type='ROD',
                order_cond='Cash',
                order_lot='Common', # 整股
                first_sell='false',
                account=accounts[0]
                )
        
        # 發送委託單
        api.place_order(
            contract=api.Contracts.Stocks[tick.code],
            order=order,
            timeout=5000
        )
    # 停損
    elif tick.close <= sell_price[tick.code]['low']:
        # 製作委託單
        order = api.Order(price=api.Contracts.Stocks[stock_id].limit_down,
                quantity=1,
                action='Sell',
                price_type='LMT', # 限價單
                order_type='ROD',
                order_cond='Cash',
                order_lot='Common', # 整股
                first_sell='false',
                account=accounts[0]
                )
        
        # 發送委託單
        api.place_order(
            contract=api.Contracts.Stocks[stock_id],
            order=order,
            timeout=5000
        )
    else:
        pass

api.logout()

整理一下程式碼

將程式碼封裝到函數中，提高易讀性以及可維護性。

from shioaji import TickSTKv1, Exchange, Shioaji, constant, BidAskSTKv1
from threading import Event
from datetime import datetime

def init_api():
    """初始化API"""
    return Shioaji(simulation=True)

def login(api, api_key, secret_key):
    """登入API"""
    return api.login(api_key=api_key, secret_key=secret_key)

def is_market_open():
    """檢查市場是否開放"""
    now = datetime.now()
    return 9 <= now.hour < 15

def place_order(api, stock_id, action, price, accounts):
    """下單"""
    order = api.Order(
        price=price,
        quantity=1,
        action=action,
        price_type='LMT',  # 限價單
        order_type='ROD',  # 僅今日有效
        order_cond='Cash', # 現股交易
        order_lot='Common', # 整股
        first_sell='false', 
        account=accounts[0]
    )
    api.place_order(
        contract=api.Contracts.Stocks[stock_id],
        order=order,
        timeout=5000
    )

def subscribe_stock(api, stock_id):
    """訂閱股票報價"""
    api.quote.subscribe(
        api.Contracts.Stocks[stock_id], 
        quote_type=constant.QuoteType.Tick,
        version=constant.QuoteVersion.v1
    )

def quote_callback(exchange: Exchange, tick: TickSTKv1):
    """報價回調函數"""
    print(f"Exchange: {exchange}, Tick: {tick}")
    if tick.close >= sell_price[tick.code]['high']:
        place_order(api, tick.code, 'Sell', api.Contracts.Stocks[tick.code].limit_down, accounts)
    elif tick.close <= sell_price[tick.code]['low']:
        place_order(api, tick.code, 'Sell', api.Contracts.Stocks[tick.code].limit_down, accounts)

if __name__ == "__main__":
    stock_list = ['2330']
    sell_price = {'2330': {'high': 400, 'low': 100}}
    
    api = init_api()
    accounts = login(api, "9EwVF2HtBmfeexfLBYUExWvFbaL6f1jenPfxESjSo4x5", "5BW5ZxGc3Vz186y5EzhCeg7b25qC3bD2kk823t9L7kyx")

    if not is_market_open():
        for stock_id in stock_list:
            place_order(api, stock_id, 'Buy', api.Contracts.Stocks[stock_id].limit_up, accounts)  # 預掛開盤買進的委託單
            subscribe_stock(api, stock_id)  # 訂閱開盤買進的個股
    else:
        print("市場已經開盤。")
    
    for stock_id in stock_list:
        subscribe_stock(api, stock_id)  # 訂閱盤中需要監控的個股
    
    api.on_tick_stk_v1(quote_callback)

    api.on_tick_stk_v1(quote_callback)  # 設定報價回調函數
    Event().wait()  # 維持程式運作，等待交易資料送入
    api.logout()  # 登出API

上一篇
Day12 修正數據(重跑要花時間尚未更新完)
下一篇
Day14 趕火車之上雲(中)
系列文
從零到英雄：用GCP建立AI交易體系 共 34 篇
  1. 30
    Day 30 回測統計(三) 連續獲利或虧損
  2. 31
    Day 30 回測統計(四) 實際執行會碰上的問題
  3. 32
    Day 31 回測統計(五) 資金風險效率
  4. 33
    Day 32 回測統計(分類評分)
  5. 34
    Day 33 回測統計(屬性分析)
完整目錄
圖片
  直播研討會
圖片
{{ item.channelVendor }} {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言
iThome鐵人賽
參賽組數
1064 組
團體組數
40 組
累計文章數
22203 篇
完賽人數
602 人
iT+看影片追技術 看影片追技術 看更多
熱門tag 看更多
15th鐵人賽 16th鐵人賽 13th鐵人賽 14th鐵人賽 12th鐵人賽 11th鐵人賽 鐵人賽 2019鐵人賽 javascript 2018鐵人賽 python 2017鐵人賽 windows php c# windows server linux css react vue.js
熱門問題
熱門回答
熱門文章
IT邦幫忙