"""
|
历史K线服务
|
"""
|
import decimal
|
|
import tool
|
from db import redis_manager
|
import gm.api as gmapi
|
|
|
class JueJinApi:
|
__redisManager = redis_manager.RedisManager(0)
|
|
# 获取掘金参数
|
@classmethod
|
def getJueJinAccountInfo(cls):
|
redis = cls.__redisManager.getRedis()
|
account_id = redis.get("juejin-account-id")
|
strategy_id = redis.get("juejin-strategy-id")
|
token = redis.get("juejin-token")
|
return account_id, strategy_id, token
|
|
@classmethod
|
def get_juejin_code_list_with_prefix(cls, codes):
|
list = []
|
for d in codes:
|
if d[0:2] == '00':
|
list.append("SZSE.{}".format(d))
|
elif d[0:2] == '60':
|
list.append("SHSE.{}".format(d))
|
return list
|
|
@classmethod
|
def get_gp_latest_info(cls, codes, fields=None):
|
if not codes:
|
return []
|
account_id, s_id, token = cls.getJueJinAccountInfo()
|
symbols = cls.get_juejin_code_list_with_prefix(codes)
|
gmapi.set_token(token)
|
data = gmapi.get_instruments(symbols=",".join(symbols), fields=fields)
|
print(data)
|
return data
|
|
@classmethod
|
def get_history_tick_n(cls, code, count, fields=None):
|
account_id, s_id, token = cls.getJueJinAccountInfo()
|
symbols = cls.get_juejin_code_list_with_prefix([code])
|
gmapi.set_token(token)
|
# 前除权
|
results = gmapi.history_n(symbol=symbols[0], frequency="1d", count=count, adjust=1, fields=fields)
|
return results
|
|
@classmethod
|
def get_gp_current_info(cls, codes):
|
if not codes:
|
return []
|
account_id, s_id, token = cls.getJueJinAccountInfo()
|
symbols = cls.get_juejin_code_list_with_prefix(codes)
|
gmapi.set_token(token)
|
data = gmapi.current(symbols=",".join(symbols))
|
print(data)
|
return data
|
|
# 返回指定日期的上个交易日
|
|
@classmethod
|
def get_previous_trading_date(cls, date):
|
account_id, s_id, token = cls.getJueJinAccountInfo()
|
gmapi.set_token(token)
|
return gmapi.get_previous_trading_date("SHSE", date)
|
|
# 返回指定日期的下个交易日
|
@classmethod
|
def get_next_trading_date(cls, date):
|
account_id, s_id, token = cls.getJueJinAccountInfo()
|
gmapi.set_token(token)
|
return gmapi.get_previous_trading_date("SHSE", date)
|
|
@classmethod
|
def get_trading_dates(cls, start_date, end_date):
|
account_id, s_id, token = cls.getJueJinAccountInfo()
|
gmapi.set_token(token)
|
return gmapi.get_trading_dates("SHSE", start_date, end_date)
|
|
|
class HistoryKDatasUtils(object):
|
|
@classmethod
|
def get_gp_latest_info(cls, codes, fields=None):
|
return JueJinApi.get_gp_latest_info(codes, fields)
|
|
@classmethod
|
def get_history_tick_n(cls, code, count, fields=None):
|
return JueJinApi.get_history_tick_n(code, count, fields)
|
|
@classmethod
|
def get_gp_current_info(cls, codes):
|
return JueJinApi.get_gp_current_info(codes)
|
|
# 返回指定日期的上个交易日
|
@classmethod
|
def get_previous_trading_date(cls, date):
|
return JueJinApi.get_previous_trading_date(date)
|
|
# 返回指定日期的下个交易日
|
@classmethod
|
def get_next_trading_date(cls, date):
|
return JueJinApi.get_next_trading_date(date)
|
|
@classmethod
|
def get_trading_dates(cls, start_date, end_date):
|
return JueJinApi.get_trading_dates(start_date, end_date)
|
|
@classmethod
|
def get_now_price(cls, codes):
|
data = cls.get_gp_current_info(codes)
|
prices = []
|
for item in data:
|
code = item["symbol"].split(".")[1]
|
price = item["price"]
|
prices.append((code, price))
|
return prices
|
|
# 获取代码的涨幅
|
@classmethod
|
def get_codes_limit_rate(cls, codes):
|
datas = cls.get_gp_latest_info(codes)
|
pre_price_dict = {}
|
for data in datas:
|
code = data["sec_id"]
|
pre_close = tool.to_price(decimal.Decimal(str(data['pre_close'])))
|
pre_price_dict[code] = pre_close
|
|
now_prices = cls.get_now_price(codes)
|
f_results = []
|
for data in now_prices:
|
code = data[0]
|
price = data[1]
|
pre_price = float(pre_price_dict.get(code))
|
rate = round((price - pre_price) * 100 / pre_price, 2)
|
f_results.append((code, rate))
|
f_results.sort(key=lambda tup: tup[1])
|
f_results.reverse()
|
return f_results
|
|
@classmethod
|
def get_lowest_price_rate(cls, code, count):
|
datas = cls.get_history_tick_n(code, count)
|
low_price = datas[0]["close"]
|
for data in datas:
|
if low_price > data["close"]:
|
low_price = data["close"]
|
return (datas[-1]["close"] - low_price) / low_price
|
|
@classmethod
|
def get_gp_codes_names(cls, codes):
|
datas = cls.get_gp_latest_info(codes)
|
results = {}
|
for data in datas:
|
code = data["symbol"].split(".")[1]
|
code_name = data['sec_name']
|
results[code] = code_name
|
return results
|
