import datetime
|
import time
|
import strategy.data_cache
|
|
from trade import middle_api_protocol
|
|
|
class JueJinHttpApi:
|
@classmethod
|
def __request(cls, path_str, data_json):
|
def deformat_date(val):
|
if type(val) == str and val.find('T') > -1 and val.find(':') > -1 and val.find('+') > -1:
|
return datetime.datetime.fromisoformat(val)
|
return val
|
fdata = middle_api_protocol.load_juejin(path_str, data_json)
|
__start_time = time.time()
|
try:
|
fdata = middle_api_protocol.request(fdata)
|
finally:
|
__use_time = time.time() - __start_time
|
if fdata:
|
if type(fdata) == list:
|
for d in fdata:
|
if type(d) != dict:
|
continue
|
for k in d:
|
d[k] = deformat_date(d[k])
|
elif type(fdata) == dict:
|
for k in fdata:
|
fdata[k] = deformat_date(fdata[k])
|
return fdata
|
else:
|
return None
|
|
@classmethod
|
def get_exchanges_codes(cls, exchanges, sec_types, skip_suspended, skip_st, fields):
|
return cls.__request("get_instruments",
|
{"exchanges": exchanges, "sec_types": sec_types, "skip_suspended": skip_suspended,
|
"skip_st": skip_st, "fields": fields})
|
|
|
class JueJinApi:
|
# 获取交易所的代码
|
@classmethod
|
def get_exchanges_codes(cls, exchanges, skip_suspended=True, skip_st=True):
|
return JueJinHttpApi.get_exchanges_codes(exchanges=exchanges, sec_types=[1], skip_suspended=skip_suspended,
|
skip_st=skip_st,
|
fields="symbol,sec_type,sec_id,sec_name,listed_date,sec_level,"
|
"is_suspended,pre_close")
|
|
@classmethod
|
def get_target_codes(cls):
|
"""
|
获取目标代码
|
:return:
|
"""
|
datas = JueJinApi.get_exchanges_codes(["SHSE", "SZSE"])
|
fdatas = []
|
for d in datas:
|
if d["sec_level"] != 1:
|
continue
|
fdatas.append(d)
|
return fdatas
|
|
|
|
if __name__ == '__main__':
|
# 获取目标代码（获取目标票）
|
print(JueJinApi.get_exchanges_codes(["SHSE", "SZSE"]))
|
#
|
strategy.data_cache.all_stocks = JueJinApi.get_exchanges_codes(["SHSE", "SZSE"])
|
print(len(JueJinApi.get_exchanges_codes(["SHSE", "SZSE"])))
|
