'初始化导入'

admin

2024-06-21 e15f8d3a986b9e8ccea31d28e0f7e26ac73d85b5

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
from local_api.util import tool
from local_api.util.tool import MARKET_TYPE_SZSE, MARKET_TYPE_SSE
from gm import api as gmapi
 
 
def get_juejin_code_list_with_prefix(codes):
    """
    根据代码获取symbol
    :param codes:
    :return:
    """
    list = []
    for d in codes:
        if tool.get_market_type(d) == MARKET_TYPE_SZSE:
            list.append("SZSE.{}".format(d))
        elif tool.get_market_type(d) == MARKET_TYPE_SSE:
            list.append("SHSE.{}".format(d))
        else:
            list.append(d)
 
    return list
 
 
class JueJinApi:
    strategy_id = ""
    token = ""
    account = ""
 
    __instance = None
 
    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(JueJinApi, cls).__new__(cls, *args, **kwargs)
        return cls.__instance
 
    @classmethod
    def init(cls, strategy_id, token, account=""):
        cls.strategy_id = strategy_id
        cls.token = token
        cls.account = account
 
    def getJueJinAccountInfo(self):
        return self.account, self.strategy_id, self.token
 
    def get_gp_latest_info(self, codes, fields=None):
        if not codes:
            return []
        symbols = get_juejin_code_list_with_prefix(codes)
        account_id, s_id, token = self.getJueJinAccountInfo()
        gmapi.set_token(token)
        data = gmapi.get_instruments(symbols=",".join(symbols), fields=fields)
        return data
 
    def get_history_tick_n(self, code, count, fields=None):
        symbols = get_juejin_code_list_with_prefix([code])
        account_id, s_id, token = self.getJueJinAccountInfo()
        gmapi.set_token(token)
        # 前除权
        results = gmapi.history_n(symbol=symbols[0], frequency="1d", count=count, adjust=1, fields=fields)
        return results
 
    def get_gp_current_info(self, codes):
        if not codes:
            return []
        symbols = get_juejin_code_list_with_prefix(codes)
 
        account_id, s_id, token = self.getJueJinAccountInfo()
        gmapi.set_token(token)
        data = gmapi.current(symbols=",".join(symbols))
        return data
 
    # 获取交易所的代码
    def get_exchanges_codes(self, exchanges, fields="symbol,sec_type,sec_id,sec_name,listed_date,sec_level,"
                                                    "is_suspended,pre_close"):
        account_id, s_id, token = self.getJueJinAccountInfo()
        gmapi.set_token(token)
        return gmapi.get_instruments(exchanges=exchanges, sec_types=[1], skip_suspended=True, skip_st=True,
                                     fields=fields)
 
    def get_previous_trading_date(self, date):
 
        account_id, s_id, token = self.getJueJinAccountInfo()
        gmapi.set_token(token)
        return gmapi.get_previous_trading_date("SHSE", date)
 
    # 返回指定日期的下个交易日
    def get_next_trading_date(self, date):
 
        account_id, s_id, token = self.getJueJinAccountInfo()
        gmapi.set_token(token)
        return gmapi.get_next_trading_date("SHSE", date)
 
    def get_trading_dates(self, start_date, end_date):
        account_id, s_id, token = self.getJueJinAccountInfo()
        gmapi.set_token(token)
        return gmapi.get_trading_dates("SHSE", start_date, end_date)