forked from Vespa314/chan.py
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathBaoStockAPI.py
More file actions
114 lines (98 loc) · 3.6 KB
/
Copy pathBaoStockAPI.py
File metadata and controls
114 lines (98 loc) · 3.6 KB
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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
import baostock as bs
from Common.CEnum import AUTYPE, DATA_FIELD, KL_TYPE
from Common.CTime import CTime
from Common.func_util import kltype_lt_day, str2float
from KLine.KLine_Unit import CKLine_Unit
from .CommonStockAPI import CCommonStockApi
def create_item_dict(data, column_name):
for i in range(len(data)):
data[i] = parse_time_column(data[i]) if i == 0 else str2float(data[i])
return dict(zip(column_name, data))
def parse_time_column(inp):
# 20210902113000000
# 2021-09-13
if len(inp) == 10:
year = int(inp[:4])
month = int(inp[5:7])
day = int(inp[8:10])
hour = minute = 0
elif len(inp) == 17:
year = int(inp[:4])
month = int(inp[4:6])
day = int(inp[6:8])
hour = int(inp[8:10])
minute = int(inp[10:12])
elif len(inp) == 19:
year = int(inp[:4])
month = int(inp[5:7])
day = int(inp[8:10])
hour = int(inp[11:13])
minute = int(inp[14:16])
else:
raise Exception(f"unknown time column from baostock:{inp}")
return CTime(year, month, day, hour, minute)
def GetColumnNameFromFieldList(fileds: str):
_dict = {
"time": DATA_FIELD.FIELD_TIME,
"date": DATA_FIELD.FIELD_TIME,
"open": DATA_FIELD.FIELD_OPEN,
"high": DATA_FIELD.FIELD_HIGH,
"low": DATA_FIELD.FIELD_LOW,
"close": DATA_FIELD.FIELD_CLOSE,
"volume": DATA_FIELD.FIELD_VOLUME,
"amount": DATA_FIELD.FIELD_TURNOVER,
"turn": DATA_FIELD.FIELD_TURNRATE,
}
return [_dict[x] for x in fileds.split(",")]
class CBaoStock(CCommonStockApi):
is_connect = None
def __init__(self, code, k_type=KL_TYPE.K_DAY, begin_date=None, end_date=None, autype=AUTYPE.QFQ):
super(CBaoStock, self).__init__(code, k_type, begin_date, end_date, autype)
def get_kl_data(self):
# 天级别以上才有详细交易信息
if kltype_lt_day(self.k_type):
if not self.is_stock:
raise Exception("没有获取到数据，注意指数是没有分钟级别数据的！")
fields = "time,open,high,low,close"
else:
fields = "date,open,high,low,close,volume,amount,turn"
autype_dict = {AUTYPE.QFQ: "2", AUTYPE.HFQ: "1", AUTYPE.NONE: "3"}
rs = bs.query_history_k_data_plus(
code=self.code,
fields=fields,
start_date=self.begin_date,
end_date=self.end_date,
frequency=self.__convert_type(),
adjustflag=autype_dict[self.autype],
)
if rs.error_code != '0':
raise Exception(rs.error_msg)
while rs.error_code == '0' and rs.next():
yield CKLine_Unit(create_item_dict(rs.get_row_data(), GetColumnNameFromFieldList(fields)))
def SetBasciInfo(self):
rs = bs.query_stock_basic(code=self.code)
if rs.error_code != '0':
raise Exception(rs.error_msg)
code, code_name, ipoDate, outDate, stock_type, status = rs.get_row_data()
self.name = code_name
self.is_stock = (stock_type == '1')
@classmethod
def do_init(cls):
if not cls.is_connect:
cls.is_connect = bs.login()
@classmethod
def do_close(cls):
if cls.is_connect:
bs.logout()
cls.is_connect = None
def __convert_type(self):
_dict = {
KL_TYPE.K_DAY: 'd',
KL_TYPE.K_WEEK: 'w',
KL_TYPE.K_MON: 'm',
KL_TYPE.K_5M: '5',
KL_TYPE.K_15M: '15',
KL_TYPE.K_30M: '30',
KL_TYPE.K_60M: '60',
}
return _dict[self.k_type]