Shulou(Shulou.com)06/02 Report--

This article introduces the knowledge of "how to add a wide data source for VNPY 2". In the operation of actual cases, many people will encounter such a dilemma, so let the editor lead you to learn how to deal with these situations. I hope you can read it carefully and be able to achieve something!

The code is as follows:

# encoding: UTF-8import jsonimport timefrom datetime import datetime, timedeltafrom typing import Listimport jqdatasdk as jqfrom vnpy.trader.constant import Exchange Intervalfrom vnpy.trader.database import database_managerfrom vnpy.trader.object import (BarData) class JQDataService: "Service for download market data from Joinquant" def _ init__ (self): # load configuration config = open ('config.json') self.setting = json.load (config) USERNAME = self.setting [' jqdata.Username'] PASSWORD = self.setting ['jqdata.Password'] try: jq.auth (USERNAME PASSWORD) except Exception as ex: print ("jq auth fail:" + repr (ex)) def to_jq_symbol (self, symbol: str, exchange: Exchange): "" CZCE product of RQData has symbol like "TA1905" while vt symbol is "TA905.CZCE" so need to add "1" in symbol. " If exchange in [Exchange.SSE Exchange.SZSE]: if exchange = = Exchange.SSE: jq_symbol = f "{symbol} .XSHG" # Shanghai Stock Exchange else: jq_symbol = f "{symbol} .XSHE" # Shenzhen Stock Exchange elif exchange = = Exchange.SHFE: jq_symbol = f "{symbol} .XSGE" # elif exchange = = Exchange.CFFEX: Jq_symbol = f "{symbol} .CCFX" # elif exchange = = Exchange.DCE: jq_symbol = f "{symbol} .XDCE" # elif exchange = = Exchange.INE: jq_symbol = f "{symbol} .XINE" # Shanghai International Energy Futures Exchange elif exchange = = Exchange.CZCE: # contract code of Zheng Shangsuo Only three digits of the year require special handling of for count Word in enumerate (symbol): if word.isdigit (): break # Check for index symbol time_str = symbol [count:] if time_str in ["88", "888", "99" "8888"]: return f "{symbol} .XZCE" # noinspection PyUnboundLocalVariable product = symbol [: count] year = symbol [count] month = symbol [count + 1:] if year = = "9": year = "1" + year else: year = "2" + year jq_symbol = f "{product} { Year} {month} .XZCE "return jq_symbol.upper () def query_history (self Symbol, exchange, start, end, interval='1m'): "Query history bar data from JQData and update Database. " Jq_symbol = self.to_jq_symbol (symbol Exchange) # if jq_symbol not in self.symbols: # return None # For querying night trading period data # end + = timedelta (1) now = datetime.now () if end > = now: end = now elif end.year = = now.year and end.month = = now.month and end.day = = now.day: end = now df = jq.get_price (jq_symbol, frequency=interval Fields= ["open", "high", "low", "close", "volume"], start_date=start, end_date=end, skip_paused=True) data: List [BarData] = [] if df is not None: for ix, row in df.iterrows (): bar = BarData (symbol=symbol, exchange=exchange) Interval=Interval.MINUTE, datetime=row.name.to_pydatetime ()-timedelta (minutes=1), open_price=row ["open"], high_price=row ["high"], low_price=row ["low"], close_price=row ["close"], volume=row ["volume"] Gateway_name= "JQ") data.append (bar) database_manager.save_bar_data (data) return data def downloadAllMinuteBar (self Days=1): "" download minute line data of contracts in all configurations "" if days! = 0: startDt = datetime.today ()-days * timedelta (1) enddt = datetime.today () else: startDt = datetime.today ()-10 * timedelta (1) enddt = datetime.today () print ('-'* 50) print (u 'start downloading contract minute line data') print ('-'* 50) if 'Bar.Min' in self.setting: l = self.setting ["Bar.Min"] for VNSymbol in l: dt0 = time.process_time () symbol = VNSymbol.split (".") [0] exchange = Exchange (VNSymbol.split (".") [1] ) self.query_history (symbol Exchange, startDt, enddt, interval='1m') cost = (time.process_time ()-dt0) print (u 'contract% s minute K line data download completed% s -% s % s seconds'% (symbol, startDt, enddt, cost) print (jq.get_query_count ()) print ('-'* 50) print u' contract minute line data download completed 'print (' -'* 50) return Noneif _ _ name__ = ='_ _ main__': JQdata = JQDataService () JQdata.downloadAllMinuteBar (days=30)

Config.json

{"jqdata.Username": "," jqdata.Password ":", "Bar.Min": ["MA8888.CZCE", "MA009.CZCE", "SR8888.CZCE", "RM8888.CZCE",]} "how to add a wide data source to VNPY 2". Thank you for reading. If you want to know more about the industry, you can follow the website, the editor will output more high-quality practical articles for you!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.