"""
|
本地http服务器
|
"""
|
import json
|
import threading
|
import uuid
|
from http.server import SimpleHTTPRequestHandler, HTTPServer
|
import urllib.parse as urlparse
|
from urllib.parse import parse_qs
|
|
import constant
|
from code_attribute import gpcode_manager
|
from db.mysql_data_delegate import Mysqldb
|
|
from strategy.data.data_manager import LowSuctionOriginDataExportManager
|
from strategy.strategy_variable_factory import DataLoader
|
from strategy.time_series_backtest import BackTest
|
from third_data import kpl_util
|
from utils import tool, output_util, huaxin_util
|
|
# 正在运行的回测任务字典
|
running_task_dict = {}
|
|
|
class CORSRequestHandler(SimpleHTTPRequestHandler):
|
# 数据加载字典
|
__data_loader_dict = {}
|
|
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")
|
date = ps_dict.get("date")
|
plates = set(json.loads(plates))
|
sql = f" select code, jx_blocks from kpl_code_blocks where day = '{date}' and " + " and ".join(
|
[f"jx_blocks like '%{p}%'" for p in plates])
|
datas = Mysqldb().select_all(sql)
|
fdatas = [(x[0], gpcode_manager.get_code_name(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})
|
|
elif url.path == "/get_codes_by_limit_up_count":
|
# 根据精选板块获取代码
|
plates = ps_dict.get("plates")
|
date = ps_dict.get("date")
|
plates = set(json.loads(plates))
|
if date not in self.__data_loader_dict:
|
self.__data_loader_dict[date] = DataLoader(date)
|
|
dates = set(self.__data_loader_dict.keys())
|
dates.discard(date)
|
for d in dates:
|
self.__data_loader_dict.pop(d)
|
results = self.__data_loader_dict[date].load_target_plate_and_codes()
|
codes = set()
|
for p in plates:
|
codes |= set(results.get(p))
|
fdatas = [(code, gpcode_manager.get_code_name(code)) for code in codes]
|
response_data = json.dumps({"code": 0, "data": fdatas})
|
|
elif url.path == "/start_run_backtest":
|
def run_backtest(task_id):
|
__back_test = BackTest(date, r"D:\workspace\trade_low_suction\strategy\strategy_script_v6.py",
|
target_codes=codes)
|
running_task_dict[task_id] = __back_test
|
__back_test.run()
|
|
date = ps_dict.get("date")
|
codes = ps_dict.get("codes")
|
codes = json.loads(codes)
|
# 开始运行回撤
|
task_id = str(uuid.uuid1())
|
threading.Thread(target=lambda: run_backtest(task_id), daemon=True).start()
|
response_data = json.dumps({"code": 0, "data": {"task_id": task_id}})
|
elif url.path == "/stop_run_backtest":
|
task_id = ps_dict.get("task_id")
|
if task_id not in running_task_dict:
|
response_data = json.dumps({"code": 1, "msg": "任务不存在"})
|
else:
|
backtest: BackTest = running_task_dict.pop(task_id)
|
del backtest
|
response_data = json.dumps({"code": 0, "data": {}})
|
|
elif url.path == "/get_backtest_result":
|
task_id = ps_dict.get("task_id")
|
if task_id not in running_task_dict:
|
response_data = json.dumps({"code": 1, "msg": "任务不存在"})
|
else:
|
backtest: BackTest = running_task_dict.get(task_id)
|
progress = tool.trade_time_sub(backtest.current_time, backtest.RANGE_TIMES[0]) / tool.trade_time_sub(
|
backtest.RANGE_TIMES[1], backtest.RANGE_TIMES[0])
|
backtest.backtest_results.sort(key=lambda x: (0 if x[3][0] else 1, x[0]))
|
response_data = json.dumps(
|
{"code": 0, "data": {"results": backtest.backtest_results, "finish": backtest.finish,
|
"progress": round(progress * 100, 2)}},
|
cls=tool.SetEncoder)
|
|
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()
|
