"""
|
本地http服务器
|
"""
|
import json
|
from http.server import SimpleHTTPRequestHandler, HTTPServer
|
import urllib.parse as urlparse
|
from urllib.parse import parse_qs
|
|
import constant
|
from db.mysql_data_delegate import Mysqldb
|
|
from strategy.low_suction_strategy import LowSuctionOriginDataExportManager
|
from strategy.strategy_variable_factory import DataLoader
|
from third_data import kpl_util
|
from utils import tool, output_util, huaxin_util
|
|
|
class CORSRequestHandler(SimpleHTTPRequestHandler):
|
def end_headers(self):
|
# 添加 CORS 头
|
self.send_header('Access-Control-Allow-Origin', '*')
|
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
|
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
|
super().end_headers()
|
|
# 处理 OPTIONS 预检请求
|
def do_OPTIONS(self):
|
self.send_response(200)
|
self.end_headers()
|
|
def do_GET(self) -> None:
|
path = self.path
|
url = urlparse.urlparse(path)
|
ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
|
response_data = ""
|
if url.path == "/get_limit_up_plate_list":
|
date = ps_dict.get("date")
|
time_str = ps_dict.get("time")
|
__LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(date)
|
# [(代码, 代码名称, 涨停原因, 涨停时间, 高度信息, 自由流通市值，是否炸板)]
|
results = __LowSuctionOriginDataExportManager.export_current_limit_up_records()
|
results = [x for x in results if tool.to_time_str(int(x[3])) <= time_str]
|
code_plates = __LowSuctionOriginDataExportManager.export_code_plates()
|
# 最终涨停的代码
|
limit_up_codes = [x[0] for x in results if not x[6]]
|
if date == tool.get_now_date_str():
|
limit_up_list = __LowSuctionOriginDataExportManager.export_limit_up_list()
|
if limit_up_list:
|
limit_up_list = [x for x in limit_up_list if x[0][:8] <= time_str]
|
limit_up_list = limit_up_list[-1][1]
|
limit_up_codes = [x[0] for x in limit_up_list]
|
|
code_infos = {x[0]: x for x in results}
|
plate_codes = {}
|
for code in code_infos.keys():
|
plates = code_plates.get(code)
|
if not plates:
|
plates = {kpl_util.filter_block(code_infos[code][2])}
|
plates -= constant.KPL_INVALID_BLOCKS
|
for p in plates:
|
if p not in plate_codes:
|
plate_codes[p] = set()
|
plate_codes[p].add(code)
|
# (板块名称, 涨停数，炸板数)
|
data = [(p, len(plate_codes[p]), len([code for code in plate_codes[p] if code not in limit_up_codes])) for p
|
in
|
plate_codes]
|
data.sort(key=lambda x: x[1] - x[2], reverse=True)
|
response_data = json.dumps({"code": 0, "data": data})
|
|
elif url.path == "/get_plate_codes":
|
date = ps_dict.get("date")
|
plate = ps_dict.get("plate")
|
time_str = ps_dict.get("time")
|
__LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(date)
|
results = __LowSuctionOriginDataExportManager.export_current_limit_up_records()
|
for r in results:
|
r[3] = tool.to_time_str(int(r[3]))
|
r[5] = output_util.money_desc(r[5])
|
results = [x for x in results if x[3] <= time_str]
|
# 最终涨停的代码
|
limit_up_codes = [x[0] for x in results if not x[6]]
|
if date == tool.get_now_date_str():
|
limit_up_list = __LowSuctionOriginDataExportManager.export_limit_up_list()
|
if limit_up_list:
|
limit_up_list = [x for x in limit_up_list if x[0][:8] <= time_str]
|
limit_up_list = limit_up_list[-1][1]
|
limit_up_codes = [x[0] for x in limit_up_list]
|
for x in results:
|
x[6] = x[0] not in limit_up_codes
|
|
# [(代码, 代码名称, 涨停原因, 涨停时间, 高度信息, 自由流通市值，是否炸板)]
|
code_plates = __LowSuctionOriginDataExportManager.export_code_plates()
|
code_infos = {x[0]: x for x in results}
|
plate_codes = {}
|
for code in code_infos.keys():
|
plates = code_plates.get(code)
|
if not plates:
|
plates = {kpl_util.filter_block(code_infos[code][2])}
|
plates -= constant.KPL_INVALID_BLOCKS
|
for p in plates:
|
if p not in plate_codes:
|
plate_codes[p] = set()
|
plate_codes[p].add(code)
|
codes = plate_codes.get(plate)
|
datas = [code_infos[code] for code in codes]
|
datas.sort(key=lambda x: x[3])
|
response_data = json.dumps({"code": 0, "data": datas})
|
elif url.path == "/get_big_order_list":
|
date = ps_dict.get("date")
|
code = ps_dict.get("code")
|
__LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(date)
|
big_orders_dict = __LowSuctionOriginDataExportManager.export_all_big_order_deal(200e4)
|
datas = big_orders_dict.get(code, [])
|
for x in datas:
|
x[2][3] = huaxin_util.convert_time(x[2][3], with_ms=False)
|
if len(x[2]) > 5:
|
x[2][5] = huaxin_util.convert_time(x[2][5], with_ms=False)
|
order_ids = set()
|
datas.reverse()
|
fdatas = []
|
for d in datas:
|
if d[2][0] in order_ids:
|
continue
|
fdatas.append(d)
|
order_ids.add(d[2][0])
|
|
response_data = json.dumps({"code": 0, "data": fdatas})
|
elif url.path == "/get_block_in_datas":
|
date = ps_dict.get("date")
|
time_str = ps_dict.get("time")
|
__LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(date)
|
block_in_datas = __LowSuctionOriginDataExportManager.export_block_in_datas()
|
fdatas = []
|
for d in reversed(block_in_datas):
|
if d[0] <= time_str:
|
fdatas = d[1]
|
break
|
fdatas = [(x[0], output_util.money_desc(x[1])) for x in fdatas]
|
response_data = json.dumps({"code": 0, "data": fdatas})
|
|
elif url.path == "/get_codes_by_jx_plates":
|
# 根据精选板块获取代码
|
plates = ps_dict.get("plates")
|
plates = set(json.loads(plates))
|
sql = " select code, jx_blocks from kpl_code_blocks where " + " and ".join(
|
[f"jx_blocks like '%{p}%'" for p in plates])
|
datas = Mysqldb().select_all(sql)
|
fdatas = [(x[0], "、".join([ f"<red>{dd}</red>" if dd in plates else dd for dd in json.loads(x[1])])) for x in datas if len(set(json.loads(x[1])) & plates) == len(plates)]
|
response_data = json.dumps({"code": 0, "data": fdatas})
|
|
print("GET请求")
|
self.send_response(200)
|
# 发给请求客户端的响应数据
|
self.send_header('Content-type', 'application/json')
|
self.end_headers()
|
self.wfile.write(response_data.encode())
|
|
def do_POST(self) -> None:
|
print("POST请求")
|
|
|
if __name__ == '__main__':
|
server_address = ('', 8000)
|
httpd = HTTPServer(server_address, CORSRequestHandler)
|
print("Server running on http://localhost:8000")
|
httpd.serve_forever()
|
