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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
| import enum
| import json
|
|
| def parse_kpl_datas(results):
| start_y = -1
| end_x = -1
| index = 0
| datas = []
| for result in results:
| text = result[1]
| if text.find("股票名称") > -1:
| start_y = result[0][0][1]
|
| if text.find("竞价涨幅") > -1:
| end_x = result[0][0][0]
| if start_y > 0 and end_x > 0:
| if result[0][0][0] < end_x and result[0][0][1] > start_y and (result[0][1][0] - result[0][0][0]) > 30:
| datas.append(text)
| index += 1
| datas = datas[:3 * 5]
| fdatas = []
| temp = []
| for data in datas:
| temp.append(data)
| if len(temp) == 3:
| fdatas.append((temp[2][:6], temp[1]))
| temp = []
| return fdatas
|
|
| # 涨停代码：
| # (代码,名称,首次涨停时间,最近涨停时间,几板,涨停原因,板块,实际流通,主力净额)
| # (0,1,6,25,9,16,11,15,12)
| # 竞价代码：
| # (代码,名称,涨停委买额,板块,竞价成交额,实际流通)
| # (0,1,18,11,22,15)
| # 炸板：
| # (代码,名称,涨幅,板块,实际流通)
| # (0,1,4,11,15)
| # 跌停：
| # (代码,名称,板块,实际流通)
| # (0,1,11,15)
| # 曾跌停：
| # (代码,名称,涨幅,板块,实际流通)
| # (0,1,4,11,15)
|
| DABAN_TYPE_BIDDING = 8
| DABAN_TYPE_LIMIT_UP = 1
| DABAN_TYPE_OPEN_LIMIT_UP = 2
| DABAN_TYPE_LIMIT_DOWN = 3
| DABAN_TYPE_EVER_LIMIT_DOWN = 5
|
|
| class KPLDataType(enum.Enum):
| BIDDING = "biddings"
| LIMIT_UP = "limit_up"
| OPEN_LIMIT_UP = "open_limit_up"
| LIMIT_DOWN = "limit_down"
| EVER_LIMIT_DOWN = "ever_limit_down"
| FENG_KOU = "feng_kou"
| BEST_FENG_KOU = "best_feng_kou"
| FENG_XIANG = "feng_xiang"
| INDUSTRY_RANK = "industry_rank"
|
|
| def __parseDaBanItemData(data, type):
| if type == DABAN_TYPE_BIDDING:
| return data[0], data[1], data[18], data[11], data[22], data[15]
| elif type == DABAN_TYPE_LIMIT_UP:
| return data[0], data[1], data[6], data[25], data[9], data[16], data[11], data[15], data[12]
| elif type == DABAN_TYPE_OPEN_LIMIT_UP:
| return data[0], data[1], data[4], data[11], data[15]
| elif type == DABAN_TYPE_LIMIT_DOWN:
| return data[0], data[1], data[11], data[15]
| elif type == DABAN_TYPE_EVER_LIMIT_DOWN:
| return data[0], data[1], data[4], data[11], data[15]
|
| return None
|
|
| # 最强风口
| # (代码,名称,强度,涨幅,热门板块,所有板块)
| def __parseBestFengKouItemData(data):
| return data[0], data[1], data[2], data[4], data[12], data[10]
|
|
| # 市场风口
| # (代码,名称,涨幅,主力净额,风口概念)
| def __parseFengKouItemData(data):
| return data[0], data[1], data[3], data[7], data[11]
|
|
| # 风向标
| # (代码, 名称, 现价, 涨幅, 板块, 300万大单净额, 主力净额, 自由市值)
| def __parseFengXiangBiaoItemData(data):
| return data[0], data[1], data[5], data[6], data[4], data[-3], data[13], data[10]
|
|
| # 行业涨幅
| # (代码,名称,主力净额,涨跌幅)
| def __parseIndustry_rank(data):
| return data[0], data[1], data[6], data[3]
|
|
| def parseDaBanData(data, type_):
| if type(data) == str:
| data = json.loads(data)
| if int(data["errcode"]) != 0:
| raise Exception(f"解析数据出错，errcode：{data['errcode']}")
| list_ = data["list"]
| fresult_ = []
| for d in list_:
| pdata = __parseDaBanItemData(d, type_)
| if pdata:
| fresult_.append(pdata)
| return fresult_
|
|
| def parseFengKou(data):
| if type(data) == str:
| data = json.loads(data)
| if int(data["errcode"]) != 0:
| raise Exception(f"解析数据出错，errcode：{data['errcode']}")
| list_ = data["List"]
| fresult_ = []
| for d in list_:
| pdata = __parseFengKouItemData(d)
| if pdata:
| fresult_.append(pdata)
| return fresult_
|
|
| def parseBestFengKou(data):
| if type(data) == str:
| data = json.loads(data)
| if int(data["errcode"]) != 0:
| raise Exception(f"解析数据出错，errcode：{data['errcode']}")
| list_ = data["List"]
| fresult_ = []
| for d in list_:
| pdata = __parseBestFengKouItemData(d)
| if pdata:
| fresult_.append(pdata)
| return fresult_
|
|
| def parseFengXiang(data):
| if type(data) == str:
| data = json.loads(data)
| if int(data["errcode"]) != 0:
| raise Exception(f"解析数据出错，errcode：{data['errcode']}")
| list_ = data["list"]
| fresult_ = []
| for d in list_:
| pdata = __parseFengXiangBiaoItemData(d)
| if pdata:
| fresult_.append(pdata)
| return fresult_
|
|
| def parseIndustryRank(data):
| if type(data) == str:
| data = json.loads(data)
| if int(data["errcode"]) != 0:
| raise Exception(f"解析数据出错，errcode：{data['errcode']}")
| list_ = data["list"]
| fresult_ = []
| for d in list_:
| pdata = __parseIndustry_rank(d)
| if pdata:
| fresult_.append(pdata)
| return fresult_
|
|
