蜗牛之路

蜗牛博客VNPY源码学习系列文章：

VNPY源码（一）CTP封装及K线合成
VNPY源码（二）API获取行情和script_trader
VNPY源码（三）主引擎MainEngine
VNPY源码（四）DataRecorder
VNPY源码（五）CtaEngine实盘引擎
VNPY源码（六）BacktesterEngine回测引擎
VNPY源码（七）限价单与停止单
VNPY源码（八）VNPY的数据流

一、init_cli_trading
init_cli_trading位于C:\vnstudio\Lib\site-packages\vnpy\app\script_trader下面的cli.py文件中，这个文件夹下面还有其他的文件，但是通过 from vnpy.app.script_trader import init_cli_trading就可以导入进来了。

在C:\vnstudio\Lib\site-packages\vnpy\gateway\ctptest新建test.py文件。

from vnpy.gateway.ctp.ctp_gateway import CtpGateway
from vnpy.app.script_trader import init_cli_trading

setting = {
    "用户名": "1xx11",
    "密码": "axx4",
    "经纪商代码": "9999",
    "交易服务器": "180.168.146.187:10130",
    "行情服务器": "180.168.146.187:10131;",
    "产品名称": "simnow_client_test",
    "授权编码": "0000000000000000",
    "产品信息": "11111"
}
engine = init_cli_trading([CtpGateway])
engine.connect_gateway(setting,"CTP")

成果展示：

然后在jupyter中运行，

# 查询所有合约
engine.get_all_contracts(use_df=True)

其他命令：

# 查询资金
engine.get_all_accounts(use_df=True)

# 查询持仓
engine.get_all_positions(use_df=True)

# 查询活动委托
engine.get_all_active_orders(use_df=True)



# 订阅行情
engine.subscribe(["zn1910.SHFE"])

# 查询行情
engine.get_tick("zn1910.SHFE", use_df=True)

# 委托下单
vt_orderid = engine.buy("zn1910.SHFE", 32, 100)
print(vt_orderid)

# 查询特定委托
engine.get_order(vt_orderid)

# 委托撤单
engine.cancel_order(vt_orderid)

二、利用vnpy获取行情

from pathlib import Path
from threading import Thread
from time import sleep

from vnpy.api.ctp.vnctpmd import MdApi


EVENT_LOG = "log"
EVENT_CONTRACT = "contract"
EVENT_TICK = "tick"
EVENT_BAR = "bar"
EVENT_ERROR = "error"
EVENT_POSITION = "position"
EVENT_TRADE = "trade"
EVENT_ORDER = "order"
EVENT_ACCOUNT = "account"
EVENT_SHARED = "shared"
EVENT_LAST = "last"
EVENT_INIT_FINISHED = "init"


def _get_trader_dir(temp_name: str):
    """
    Get path where trader is running in.
    """
    cwd = Path.cwd()
    temp_path = cwd.joinpath(temp_name)

    # If .vntrader folder exists in current working directory,
    # then use it as trader running path.
    if temp_path.exists():
        return cwd, temp_path

    # Otherwise use home path of system.
    home_path = Path.home()
    temp_path = home_path.joinpath(temp_name)

    if not temp_path.exists():
        temp_path.mkdir()

    return home_path, temp_path


def get_folder_path(folder_name: str):
    """
    Get path for temp folder with folder name.
    """
    TRADER_DIR, TEMP_DIR = _get_trader_dir(".ctpbee")
    folder_path = TEMP_DIR.joinpath(folder_name)
    if not folder_path.exists():
        folder_path.mkdir()
    return folder_path


class BeeMdApi(MdApi):
    """"""

    def __init__(self):
        """Constructor"""
        super(BeeMdApi, self).__init__()

        self.gateway_name = "ctp"

        self.reqid = 0
        self.connect_status = False
        self.login_status = False
        self.subscribed = set()

        self.userid = ""
        self.password = ""
        self.brokerid = 0

    @property
    def md_status(self):
        return self.login_status

    def on_event(self, type, data):
        print(f"事件类型{type}, data={data}")

    def onFrontConnected(self):
        """
        Callback when front server is connected.
        """
        self.connect_status = True
        self.on_event(type=EVENT_LOG, data="行情服务器连接成功")
        self.login()

    def onFrontDisconnected(self, reason: int):
        """
        Callback when front server is disconnected.
        """
        self.connect_status = False
        self.login_status = False
        self.on_event(type=EVENT_LOG, data=f"行情连接断开，原因{reason}")

    def onRspUserLogin(self, data: dict, error: dict, reqid: int, last: bool):
        """
        Callback when user is logged in.
        """
        if not error["ErrorID"]:
            self.login_status = True
            self.on_event(type=EVENT_LOG, data="行情服务器登录成功")

            for symbol in self.subscribed:
                self.subscribeMarketData(symbol)
        else:
            error["detail"] = "行情登录失败"
            self.on_event(type=EVENT_ERROR, data=error)

    def onRspError(self, error: dict, reqid: int, last: bool):
        """
        Callback when error occured.
        """
        error['detail'] = "行情接口报错"
        self.on_event(type=EVENT_ERROR, data=error)

    def onRspSubMarketData(self, data: dict, error: dict, reqid: int, last: bool):
        """"""
        if not error or not error["ErrorID"]:
            return
        error['detail'] = "行情订阅失败"
        self.on_event(type=EVENT_ERROR, data=error)

    def onRtnDepthMarketData(self, data: dict):
        """
        Callback of tick data update.
        """
        print(data)

    def connect(self, info: dict):
        """
        Start connection to server.
        """
        self.userid = info['userid']
        self.password = info['password']
        self.brokerid = info['brokerid']

        # If not connected, then start connection first.
        if not self.connect_status:
            path = get_folder_path(self.gateway_name.lower() + f"/{self.userid}")
            self.createFtdcMdApi(str(path) + "\\Md")
            self.registerFront(info['md_address'])
            self.init()
        # If already connected, then login immediately.
        elif not self.login_status:
            self.login()

    def login(self):
        """
        Login onto server.
        """
        req = {
            "UserID": self.userid,
            "Password": self.password,
            "BrokerID": self.brokerid
        }

        self.reqid += 1
        self.reqUserLogin(req, self.reqid)

    def subscribe(self, symbol):
        """
        Subscribe to tick data update.
        """
        result = None
        if self.login_status:
            result = self.subscribeMarketData(symbol)
        self.subscribed.add(symbol)
        return result

    def close(self):
        """
        Close the connection.
        """
        if self.connect_status:
            self.exit()


def run_login():
    gateway = BeeMdApi()
    info = {
        "CONNECT_INFO": {
            "userid": "",
            "password": "",
            "brokerid": "9999",
            "md_address": "tcp://180.168.146.187:10131",
            "td_address": "tcp://180.168.146.187:10130",
            # "md_address": "tcp://218.202.237.33:10112",
            # "td_address": "tcp://218.202.237.33:10102",
            "product_info": "",
            "appid": "simnow_client_test",
            "auth_code": "0000000000000000",
        },
        "INTERFACE": "ctp",
        "TD_FUNC": True,
        "MD_FUNC": True,
    }
    gateway.connect(
        info["CONNECT_INFO"]
    )

    gateway.subscribe("ag1912")
    while True:
        sleep(1)
        print("-"*30)

run_login()

成果展示：

可参考：
https://github.com/vnpy/vnpy/blob/master/examples/notebook_trading/demo_notebook.ipynb