VeighNa量化社区
你的开源社区量化交易平台
brave_wlc's Avatar
Member
离线
8 帖子
声望: 0

ScriptTrader - 脚本策略模块,订阅行情后,怎么可以获取到实时行情推送的tick?
example给的是每次调用get_tick去获取最新tick。
tick = engine.get_tick(vt_symbol="rb1910.SHFE",use_df=False)

如果不调用engine.get_tick的话,应该怎么调用拿到实时推送tick数据?
新手,期待大牛解答,万分感谢~

附上代码:

from time import sleep
from vnpy.event import EventEngine
from vnpy.trader.engine import MainEngine
from vnpy_ctp import CtpGateway
from vnpy_scripttrader import ScriptEngine
from vnpy_scripttrader import ScriptTraderApp
from vnpy.trader.object import (
    SubscribeRequest,
)


def tick():
    ctp_setting = {
        "用户名": "XXX",
        "密码": "XXX",
        "经纪商代码": "9999",
        "交易服务器": "180.168.146.187:10130",
        "行情服务器": "180.168.146.187:10131",
        "产品名称": "simnow_client_test",
        "授权编码": "0000000000000000",
        "产品信息": ""
    }
    event_engine = EventEngine()
    main_engine = MainEngine(event_engine)
    main_engine.add_gateway(CtpGateway)

    main_engine.add_app(ScriptTraderApp)
    main_engine.write_log("主引擎创建成功")
    main_engine.connect(ctp_setting, "CTP")
    main_engine.write_log("连接CTP接口")

    sleep(10)
    script_engine = ScriptEngine(main_engine, event_engine)
    vt_symbol = 'IF2201.CFFEX'
    # add_tick_recording('IF2201.CFFEX')
    contract = script_engine.get_contract(vt_symbol="IF2201.CFFEX", use_df=False)
    # contract = script_engine.get_contract(vt_symbol)
    if not contract:
        main_engine.write_log(f"找不到合约:{vt_symbol}")
        return
    req: SubscribeRequest = SubscribeRequest(
        symbol=contract.symbol,
        exchange=contract.exchange
    )
    main_engine.subscribe(req, contract.gateway_name)

if __name__ == '__main__':
    tick()

另外在ctp_gateway.py中的onRtnDepthMarketData方法中写了 print(tick)

    def onRtnDepthMarketData(self, data: dict) -> None:
        """行情数据推送"""
        # 过滤没有时间戳的异常行情数据
        if not data["UpdateTime"]:
            return

        # 过滤还没有收到合约数据前的行情推送
        symbol: str = data["InstrumentID"]
        contract: ContractData = symbol_contract_map.get(symbol, None)
        if not contract:
            return

        # 对大商所的交易日字段取本地日期
        if contract.exchange == Exchange.DCE:
            date_str: str = self.current_date
        else:
            date_str: str = data["ActionDay"]

        timestamp: str = f"{date_str} {data['UpdateTime']}.{int(data['UpdateMillisec']/100)}"
        dt: datetime = datetime.strptime(timestamp, "%Y%m%d %H:%M:%S.%f")
        dt: datetime = CHINA_TZ.localize(dt)

        tick: TickData = TickData(
            symbol=symbol,
            exchange=contract.exchange,
            datetime=dt,
            name=contract.name,
            volume=data["Volume"],
            turnover=data["Turnover"],
            open_interest=data["OpenInterest"],
            last_price=adjust_price(data["LastPrice"]),
            limit_up=data["UpperLimitPrice"],
            limit_down=data["LowerLimitPrice"],
            open_price=adjust_price(data["OpenPrice"]),
            high_price=adjust_price(data["HighestPrice"]),
            low_price=adjust_price(data["LowestPrice"]),
            pre_close=adjust_price(data["PreClosePrice"]),
            bid_price_1=adjust_price(data["BidPrice1"]),
            ask_price_1=adjust_price(data["AskPrice1"]),
            bid_volume_1=data["BidVolume1"],
            ask_volume_1=data["AskVolume1"],
            gateway_name=self.gateway_name
        )

        if data["BidVolume2"] or data["AskVolume2"]:
            tick.bid_price_2 = adjust_price(data["BidPrice2"])
            tick.bid_price_3 = adjust_price(data["BidPrice3"])
            tick.bid_price_4 = adjust_price(data["BidPrice4"])
            tick.bid_price_5 = adjust_price(data["BidPrice5"])

            tick.ask_price_2 = adjust_price(data["AskPrice2"])
            tick.ask_price_3 = adjust_price(data["AskPrice3"])
            tick.ask_price_4 = adjust_price(data["AskPrice4"])
            tick.ask_price_5 = adjust_price(data["AskPrice5"])

            tick.bid_volume_2 = data["BidVolume2"]
            tick.bid_volume_3 = data["BidVolume3"]
            tick.bid_volume_4 = data["BidVolume4"]
            tick.bid_volume_5 = data["BidVolume5"]

            tick.ask_volume_2 = data["AskVolume2"]
            tick.ask_volume_3 = data["AskVolume3"]
            tick.ask_volume_4 = data["AskVolume4"]
            tick.ask_volume_5 = data["AskVolume5"]
        print(tick)
        self.gateway.on_tick(tick)

`from vnpy_scripttrader import init_cli_trading
from vnpy_ctp import CtpGateway
from time import sleep
from vnpy_scripttrader import ScriptEngine

engine = init_cli_trading([CtpGateway])
setting = {
"用户名": "xxxxxx",
"密码": "xxxxxxxxxx",
"经纪商代码": "9999",
"交易服务器": "180.168.146.187:10202",
"行情服务器": "180.168.146.187:10212",
"产品名称": "simnow_client_test",
"授权编码": "0000000000000000"
}
engine.connect_gateway(setting, "CTP")`
想要实现的功能就是,打开(.py文件打包封装成的)exe文件,实现行情登录,以及循环获取订阅行情数据?
目前通过vnpy软件连接CTP,打开ScriptTrader模块,打开.py文件,点击启动可以完成。
想咨询一下大佬们,怎么把登录和获取行情放在一个.py文件实现,使用pyinstaller应该怎么打包文件实现这个功能?

© 2015-2022 微信 18391752892
备案服务号:沪ICP备18006526号

沪公网安备 31011502017034号

【用户协议】
【隐私政策】
【免责条款】