CTP程序化交易入門系列

本文將從多個方面詳細闡述CTP程序化交易入門系列，包括行情獲取、交易指令下達等。

一、行情獲取

在進行程序化交易前，需要獲取實時的行情信息。CTP提供了多種獲取行情的渠道，包括：

1、使用CTP API獲取行情：通過CTP API獲取行情數據，具有實時性和數據完整性。

// 示例代碼：
#include 
#include "ThostFtdcMdApi.h"

class CTraderApi : public CThostFtdcMdSpi
{
public:
    //......
    virtual void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField *pDepthMarketData);
};

void CTraderApi::OnRtnDepthMarketData(CThostFtdcDepthMarketDataField *pDepthMarketData)
{
    // 獲取行情數據
}

int main()
{
    CThostFtdcMdApi* pMdUserApi = CThostFtdcMdApi::CreateFtdcMdApi();
    CTraderApi* pTraderSpi = new CTraderApi();
    pMdUserApi->RegisterSpi(pTraderSpi);
    pMdUserApi->Init();

    return 0;
}

2、使用K線等歷史數據：通過請求歷史數據獲取行情，缺點是數據不夠實時。

// 示例代碼：
from ctpapi import ApiStruct
import ctpapi
class MyMdApi(ctpapi.CThostFtdcMdApi):
    def __init__(self, instruments):
        ctpapi.CThostFtdcMdApi.__init__(self)

        self.instruments = instruments

    def onRtnDepthMarketData(self, pDepthMarketData):
        # 獲取行情數據


if __name__ == '__main__':
    instruments = ['rb2105', 'i2105']
    user = 'xxx'
    password = 'xxx'
    broker_id = 'xxx'
    address = 'xxx'
    api = MyMdApi(instruments)
    api.Create("")

    api.RegisterFront(address)
    api.Init()

    loginReq = ApiStruct.ReqUserLogin(BrokerID=broker_id, UserID=user, Password=password)
    api.ReqUserLogin(loginReq, 1)

二、交易指令下達

獲取行情後，需要進行交易指令下達。交易指令下達有多種方式，包括：

1、使用CTP API下單：通過CTP API下單，具有實時性和交互性。

// 示例代碼：
#include 
#include "ThostFtdcTraderApi.h"

class CTraderApi : public CThostFtdcTraderSpi
{
public:
    //......
    virtual void OnRtnOrder(CThostFtdcOrderField *pOrder);
};

void CTraderApi::OnRtnOrder(CThostFtdcOrderField *pOrder)
{
    // 獲取下單結果
}

int main()
{
    CThostFtdcTraderApi* pUserApi = CThostFtdcTraderApi::CreateFtdcTraderApi();
    CTraderApi* pTraderSpi = new CTraderApi();
    pUserApi->RegisterSpi(pTraderSpi);
    pUserApi->SubscribePublicTopic(THOST_TERT_QUICK);
    pUserApi->SubscribePrivateTopic(THOST_TERT_QUICK);
    pUserApi->RegisterFront("tcp://xxx");
    pUserApi->Init();

    CThostFtdcInputOrderField order = {0};
    // 下單代碼
    pUserApi->ReqOrderInsert(ℴ, nRequestID++);
    return 0;
}

2、通過HTTP請求下單介面下單：使用HTTP請求下單介面發送下單請求，不需要本地安裝CTP API，但速度和穩定性相對較低。

// 示例代碼：
import requests
import json
def http_order(instrument, price, volume, direction):
    url = 'http://xxx/api/order'
    headers = {'content-type': 'application/json'}
    data = {'instrument': instrument, 'price': price, 'volume': volume, 'direction': direction}
    r = requests.post(url, data=json.dumps(data), headers=headers)
    result = json.loads(r.content)["result"]
    if result:
        # 下單成功
    else:
        # 下單失敗

三、風險管理

在進行程序化交易時，必須考慮風險控制。對於期貨交易，風險控制的主要手段包括：

1、資金管理：通過實時計算持倉市值、權益等信息，判斷是否需要進行追加保證金或平倉操作。

// 示例代碼：
#include 
#include "ThostFtdcTraderApi.h"

class CTraderApi : public CThostFtdcTraderSpi
{
public:
    //......
    virtual void OnRtnTradingNotice(CThostFtdcTradingNoticeInfoField *pTradingNoticeInfo);
};

void CTraderApi::OnRtnTradingNotice(CThostFtdcTradingNoticeInfoField *pTradingNoticeInfo)
{
    // 獲取到交易通知，判斷是否需要進行追加保證金或平倉操作
}

int main()
{
    CThostFtdcTraderApi* pUserApi = CThostFtdcTraderApi::CreateFtdcTraderApi();
    CTraderApi* pTraderSpi = new CTraderApi();
    pUserApi->RegisterSpi(pTraderSpi);
    pUserApi->SubscribePublicTopic(THOST_TERT_QUICK);
    pUserApi->SubscribePrivateTopic(THOST_TERT_QUICK);
    pUserApi->RegisterFront("tcp://xxx");
    pUserApi->Init();

    return 0;
}

2、止損設置：通過設置止損參數，對持倉進行風險控制。

// 示例代碼：
from ctpapi import ApiStruct
import ctpapi

class MyMdApi(ctpapi.CThostFtdcMdApi):
    def __init__(self, instruments, api):
        ctpapi.CThostFtdcMdApi.__init__(self)

        self.instruments = instruments
        self.api = api

    def onRtnDepthMarketData(self, pDepthMarketData):
        # 設置止損參數
        if (pDepthMarketData.LastPrice >= 3500):
            order = ApiStruct.InputOrder(
                InstrumentID='cu2105',
                LimitPrice=3300,
                VolumeTotalOriginal=1,
                OrderPriceType=ApiStruct.OPT_LIMIT_PRICE,
                Direction=ApiStruct.DIRECTION_SELL,
                CombOffsetFlag=ApiStruct.OFFSET_OPEN
            )
            self.api.ReqOrderInsert(order, api.nRequestID)
        else:
            # ...
    def set_api(self, api):
        self.api = api


class MyTraderApi(ctpapi.CThostFtdcTraderSpi):
    def __init__(self, user_id, password, broker_id, md_api):
        ctpapi.CThostFtdcTraderSpi.__init__(self)
        self.__req_id = 0
        self.__user_id = user_id
        self.__password = password
        self.__broker_id = broker_id
        self.__md_api = md_api

    def onFrontConnected(self):
        login_req = ApiStruct.ReqUserLogin(
            BrokerID=self.__broker_id, UserID=self.__user_id, Password=self.__password)
        self.__md_api.ReqUserLogin(login_req, self.__req_id)

    def onRtnOrder(self, order):
        # 處理訂單回報

    def onRtnTrade(self, trade):
        # 處理成交回報

    def onRtnInstrumentStatus(self, instrument_status):
        # 處理合約狀態

if __name__ == '__main__':
    instruments = ['rb2105', 'i2105']
    user = 'xxx'
    password = 'xxx'
    broker_id = 'xxx'
    address = 'xxx'
    api = ctpapi.CThostFtdcTraderApi_CreateFtdcTraderApi()
    api.RegisterFront(address)
    api.SubscribePrivateTopic(ctpapi.TERT_QUICK)
    api.SubscribePublicTopic(ctpapi.TERT_QUICK)

    md_api = MyMdApi(instruments, api)
    md_api.Create("")
    md_api.set_api(api)
    md_api.RegisterFront(address)
    md_api.Init()

    trader = MyTraderApi(user, password, broker_id, md_api)
    api.RegisterSpi(trader)
    api.Init()
    api.Join()

四、策略開發

在實際應用中，程序化交易主要需要用到策略開發，即開發交易策略並實現自動下單。主要步驟如下：

1、根據市場情況和投資組合確定交易策略。

2、編寫交易規則：定義入場/出場、止損/止盈條件、持倉跟蹤等參數。

// 示例代碼：
class MyMdApi(ctpapi.CThostFtdcMdApi):
    def __init__(self, instruments, api):
        ctpapi.CThostFtdcMdApi.__init__(self)

        self.instruments = instruments
        self.api = api

    def onRtnDepthMarketData(self, pDepthMarketData):
        # 確定交易策略
        if pDepthMarketData.InstrumentID == 'rb2105':
            if (pDepthMarketData.LastPrice >= 3500):
                # 行情符合交易策略
                order = ApiStruct.InputOrder(
                    InstrumentID='cu2105',
                    LimitPrice=3300,
                    VolumeTotalOriginal=1,
                    OrderPriceType=ApiStruct.OPT_LIMIT_PRICE,
                    Direction=ApiStruct.DIRECTION_SELL,
                    CombOffsetFlag=ApiStruct.OFFSET_OPEN
                )
                self.api.ReqOrderInsert(order, api.nRequestID)
        else:
            # ...

    def set_api(self, api):
        self.api = api


class MyTraderApi(ctpapi.CThostFtdcTraderSpi):
    def __init__(self, user_id, password, broker_id, md_api):
        ctpapi.CThostFtdcTraderSpi.__init__(self)
        self.__req_id = 0
        self.__user_id = user_id
        self.__password = password
        self.__broker_id = broker_id
        self.__md_api = md_api

    def onFrontConnected(self):
        login_req = ApiStruct.ReqUserLogin(
            BrokerID=self.__broker_id, UserID=self.__user_id, Password=self.__password)
        self.__md_api.ReqUserLogin(login_req, self.__req_id)

    def onRtnOrder(self, order):
        # 處理訂單回報

    def onRtnTrade(self, trade):
        # 處理成交回報

    def onRtnInstrumentStatus(self, instrument_status):
        # 處理合約狀態

if __name__ == '__main__':
    instruments = ['rb2105', 'i2105']
    user = 'xxx'
    password = 'xxx'
    broker_id = 'xxx'
    address = 'xxx'
    api = ctpapi.CThostFtdcTraderApi_CreateFtdcTraderApi()
    api.RegisterFront(address)
    api.SubscribePrivateTopic(ctpapi.TERT_QUICK)
    api.SubscribePublicTopic(ctpapi.TERT_QUICK)

    md_api = MyMdApi(instruments, api)
    md_api.Create("")
    md_api.set_api(api)
    md_api.RegisterFront(address)
    md_api.Init()

    trader = MyTraderApi(user, password, broker_id, md_api)
    api.RegisterSpi(trader)
    api.Init()
    api.Join()

3、進行回測和優化：使用歷史數據和交易規則進行模擬回測，評估策略的效果並進行優化和調整。

// 示例代碼：
def backtest(testdata: pd.DataFrame, strategy: object) -> pd.DataFrame:
    result = pd.DataFrame(columns=['datetime', 'price', 'action'])
    for i in range(0, len(testdata)):
        action = strategy.run(testdata.iloc[i])
        if action:
            result = result.append(
                {'datetime': testdata.iloc[i]['datetime'], 'price': testdata.iloc[i]['price'],
                 'action': action}, ignore_index=True)
    return result

class MyStrategy():
    def __init__(self):
        pass

    def run(self, data):
        # 執行交易規則
        if data['price'] >= 3500:
            return 'sell'
        return False

if __name__ == '__main__':
    testdata = pd.read_csv('testdata.csv')
    strategy = MyStrategy()
    backtest_result = backtest(testdata, strategy)
    # 輸出回測結果

原創文章，作者：RAVHS，如若轉載，請註明出處：https://www.506064.com/zh-tw/n/374675.html

CTP程序化交易入門系列

一、行情獲取

二、交易指令下達

三、風險管理

四、策略開發

相關推薦

發表回復