import concurrent.futures
|
import copy
|
import hashlib
|
import http
|
import json
|
import logging
|
import socketserver
|
import time
|
from http.server import BaseHTTPRequestHandler
|
import urllib.parse as urlparse
|
|
import psutil
|
|
import constant
|
from db import redis_manager_delegate as redis_manager, mysql_data_delegate as mysql_data
|
from db.redis_manager_delegate import RedisUtils
|
from log_module import log_export, async_log_util
|
from log_module.log import hx_logger_l2_transaction, logger_debug, logger_request_api
|
from strategy import data_cache
|
from strategy.kpl_data_manager import KPLStockOfMarketsPlateLogManager, KPLMarketStockHeatLogManager
|
from strategy.trade_setting import TradeSetting
|
from trade import huaxin_trade_api, huaxin_trade_data_update
|
from trade.huaxin_trade_record_manager import DelegateRecordManager, DealRecordManager, MoneyManager, PositionManager
|
from utils import tool, huaxin_util, socket_util
|
|
|
class DataServer(BaseHTTPRequestHandler):
|
__data_process_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=10)
|
|
# 禁用日志输出
|
def log_message(self, format, *args):
|
pass
|
|
def do_GET(self):
|
|
def get_params(url):
|
return dict([(k, v[0]) for k, v in urlparse.parse_qs(url.query).items()])
|
|
path = self.path
|
url = urlparse.urlparse(path)
|
params_dict = get_params(url)
|
response_data = ""
|
if url.path == "/get_position_list":
|
# 获取持仓列表
|
results = PositionManager.get_position_cache()
|
results = copy.deepcopy(results)
|
for r in results:
|
r["auto_sell"] = 1 if r["securityID"] in data_cache.LIMIT_UP_SELL_CODES else 0
|
response_data = json.dumps({"code": 0, "data": results})
|
elif url.path == "/get_money":
|
# 获取资金信息
|
result = MoneyManager.get_cache()
|
response_data = json.dumps({"code": 0, "data": result})
|
elif url.path == "/get_deal_list":
|
# 获取成交列表
|
results = DealRecordManager.list_by_day(tool.get_now_date_str("%Y%m%d"))
|
if results:
|
for d in results:
|
d["securityName"] = data_cache.DataCache().code_name_dict.get(tool.get_symbol(d["securityID"]))
|
|
response_data = json.dumps({"code": 0, "data": results})
|
elif url.path == "/get_delegate_list":
|
# 获取委托列表
|
# 是否可撤单，如果不传默认拉取所有
|
print("参数", params_dict)
|
can_cancel = params_dict.get("can_cancel")
|
order_status = []
|
if can_cancel is not None:
|
if int(can_cancel):
|
print("获取未结委托")
|
order_status = [huaxin_util.TORA_TSTP_OST_Cached, huaxin_util.TORA_TSTP_OST_Unknown,
|
huaxin_util.TORA_TSTP_OST_Accepted, huaxin_util.TORA_TSTP_OST_PartTraded]
|
else:
|
print("获取已结委托")
|
order_status = [huaxin_util.TORA_TSTP_OST_AllTraded, huaxin_util.TORA_TSTP_OST_PartTradeCanceled,
|
huaxin_util.TORA_TSTP_OST_AllCanceled, huaxin_util.TORA_TSTP_OST_Rejected]
|
results = DelegateRecordManager.list_by_day(tool.get_now_date_str("%Y%m%d"), None, orderStatus=order_status)
|
response_data = json.dumps({"code": 0, "data": results})
|
|
elif url.path == "/refresh_trade_data":
|
# 刷新交易数据
|
_type = params_dict.get("type")
|
if _type == "money":
|
huaxin_trade_data_update.add_money_list()
|
elif _type == "delegate":
|
huaxin_trade_data_update.add_delegate_list("手动刷新")
|
elif _type == "deal":
|
huaxin_trade_data_update.add_deal_list()
|
elif _type == "position":
|
huaxin_trade_data_update.add_position_list()
|
response_data = json.dumps({"code": 0, "data": {}})
|
|
elif url.path == "/get_market_info":
|
# 获取市场行情信息
|
codes_str = params_dict.get("codes")
|
codes = json.loads(codes_str)
|
fdatas = []
|
for code in codes:
|
data = data_cache.latest_code_market_info_dict.get(code)
|
# logger_debug.info(f"获取L1行情接口：{code}-{data}")
|
if data:
|
fdatas.append(data)
|
response_data = json.dumps({"code": 0, "data": fdatas})
|
elif url.path == "/get_buy_money":
|
# 获取每次买入的金额
|
money = data_cache.BUY_MONEY_PER_CODE
|
response_data = json.dumps({"code": 0, "data": {"money": money}})
|
elif url.path == "/get_trade_settings":
|
fdata = {"running": TradeSetting().get_running(), "auto_sell": TradeSetting().get_auto_sell(),
|
"auto_buy": TradeSetting().get_auto_buy()}
|
response_data = json.dumps({"code": 0, "data": fdata})
|
|
elif url.path == "/set_trade_settings":
|
running = params_dict.get("running")
|
auto_sell = params_dict.get("auto_sell")
|
auto_buy = params_dict.get("auto_buy")
|
if running is not None:
|
TradeSetting().set_running(int(running))
|
if auto_sell is not None:
|
TradeSetting().set_auto_sell(int(auto_sell))
|
if auto_buy is not None:
|
TradeSetting().set_auto_buy(int(auto_buy))
|
response_data = json.dumps({"code": 0, "data": {}})
|
|
elif url.path == "/get_env":
|
request_id = params_dict.get("request_id")
|
use_time_list = []
|
try:
|
__start_time = time.time()
|
fdata = {}
|
# try:
|
# date = HistoryKDatasUtils.get_trading_dates(tool.date_sub(tool.get_now_date_str(), 10),
|
# tool.get_now_date_str())
|
# if date:
|
# fdata["juejin"] = 1
|
# except Exception as e:
|
# fdata["juejin"] = 0
|
# fdata["kpl"] = {}
|
# # 获取开盘啦数据
|
# kpl_types = [KPLDataType.LIMIT_UP.value, KPLDataType.JINGXUAN_RANK.value,
|
# KPLDataType.INDUSTRY_RANK.value]
|
# for kpl_type in kpl_types:
|
# if kpl_type in KPLDataManager.kpl_data_update_info:
|
# fdata["kpl"][kpl_type] = KPLDataManager.kpl_data_update_info.get(kpl_type)
|
|
try:
|
# 验证redis
|
RedisUtils.get(redis_manager.RedisManager(0).getRedis(), "test")
|
fdata["redis"] = 1
|
except:
|
fdata["redis"] = 0
|
use_time_list.append(("验证redis", time.time() - __start_time))
|
|
try:
|
# 验证mysql
|
mysql_data.Mysqldb().select_one("select 1")
|
fdata["mysql"] = 1
|
except:
|
fdata["mysql"] = 0
|
use_time_list.append(("验证mysql", time.time() - __start_time))
|
|
try:
|
# redis异步任务数量
|
fdata["redis_async_task_count"] = redis_manager.RedisUtils.get_async_task_count()
|
except:
|
pass
|
use_time_list.append(("验证异步任务数量", time.time() - __start_time))
|
|
# 获取交易通道
|
try:
|
can_access = huaxin_trade_api.test_trade_channel()
|
fdata["trade_channel_access"] = 1 if can_access else 0
|
except Exception as e:
|
logger_debug.exception(e)
|
fdata["trade_channel_access"] = 0
|
use_time_list.append(("验证交易通道", time.time() - __start_time))
|
|
# 获取CPU与内存适用情况
|
memory_info = psutil.virtual_memory()
|
cpu_percent = psutil.cpu_percent(interval=1)
|
fdata["device"] = {"cpu": cpu_percent, "memery": memory_info.percent}
|
|
use_time_list.append(("获取设备资源占用", time.time() - __start_time))
|
# 获取交易通道
|
result = {"code": 0, "data": fdata, "msg": ""}
|
# print("OnGetEnvInfo 成功")
|
response_data = json.dumps(result)
|
except Exception as e:
|
response_data = json.dumps({"code": 1, "msg": str(e)})
|
logger_debug.error(f"环境获取异常：{request_id}")
|
logger_debug.exception(e)
|
finally:
|
if use_time_list and use_time_list[-1][1] > 10:
|
logger_debug.warning(f"环境获取时间大于10s({request_id}):{use_time_list}")
|
# 获取板块强度数据
|
elif url.path == "/load_get_kpl_market_sift_plate":
|
# 加载数据
|
KPLStockOfMarketsPlateLogManager().load_data()
|
response_data = json.dumps({"code": 0, "msg": "暂无内容"})
|
elif url.path == "/get_kpl_market_sift_plate":
|
# 获取开盘啦流入板块详细信息
|
print("==========get_kpl_market_sift_plate==========")
|
try:
|
time_str = params_dict.get("time")
|
if not time_str:
|
time_str = tool.get_now_time_str()
|
fdatas = KPLStockOfMarketsPlateLogManager().get_filter_log_datas()
|
response_data = json.dumps({"code": 1, "msg": "暂无内容"})
|
for i in range(len(fdatas) - 1, -1, -1):
|
if fdatas[i][0] <= time_str:
|
response_data = json.dumps({"code": 0, "data": fdatas[i]})
|
break
|
except Exception as e:
|
logging.exception(e)
|
response_data = json.dumps({"code": 1, "msg": str(e)})
|
|
# 获取个股强度数据
|
elif url.path == "/load_get_kpl_market_stock_heat":
|
# 加载数据
|
KPLMarketStockHeatLogManager().load_data()
|
response_data = json.dumps({"code": 0, "msg": "暂无内容"})
|
elif url.path == "/get_get_kpl_market_stock_heat":
|
# 获取开盘啦流入板块详细信息
|
print("==========get_kpl_stock_of_markets_plate==========")
|
try:
|
time_str = params_dict.get("time")
|
if not time_str:
|
time_str = tool.get_now_time_str()
|
fdatas = KPLMarketStockHeatLogManager().get_filter_log_datas()
|
response_data = json.dumps({"code": 1, "msg": "暂无内容"})
|
for i in range(len(fdatas) - 1, -1, -1):
|
if fdatas[i][0] <= time_str:
|
response_data = json.dumps({"code": 0, "data": fdatas[i]})
|
break
|
except Exception as e:
|
logging.exception(e)
|
response_data = json.dumps({"code": 1, "msg": str(e)})
|
elif url.path == "/get_kpl_market_strong_records":
|
# 获取开盘啦市场强度记录
|
time_str = params_dict.get("time")
|
if not time_str:
|
time_str = tool.get_now_time_str()
|
datas = log_export.load_kpl_market_strong()
|
fdatas = []
|
for data in datas:
|
# (距离09:15:00的秒数, 时间, 强度)
|
fdatas.append((tool.trade_time_sub(data[0], "09:15:00"), data[0], data[1]))
|
response_data = json.dumps({"code": 0, "data": fdatas})
|
|
self.send_response(200)
|
# 发给请求客户端的响应数据
|
self.send_header('Content-type', 'application/json')
|
self.end_headers()
|
self.wfile.write(response_data.encode())
|
|
@classmethod
|
def __is_sign_right(cls, params):
|
ps = []
|
for k, v in params.items():
|
if k == 'sign':
|
continue
|
ps.append(f"{k}={v}")
|
ps.sort()
|
source_str = "&".join(ps) + "!@#lowSU*^cTion8888"
|
md5_hash = hashlib.md5()
|
# 将字符串编码为字节并更新哈希对象
|
md5_hash.update(source_str.encode('utf-8'))
|
# 获取十六进制表示的哈希值
|
md5_hexdigest = md5_hash.hexdigest()
|
if md5_hexdigest == params.get("sign"):
|
return True
|
return False
|
|
def do_POST(self):
|
result_str = ""
|
try:
|
path = self.path
|
print("接收到POST请求：", str(path))
|
url = urlparse.urlparse(path)
|
if url.path == "/trade_callback":
|
if constant.IS_SIMULATED_TRADE:
|
# 接受开盘啦数据
|
body = self.__parse_request()
|
if type(body) != str:
|
huaxin_trade_api.add_trade_callback_data(json.dumps(body))
|
else:
|
huaxin_trade_api.add_trade_callback_data(body)
|
result_str = json.dumps({"code": 0})
|
elif url.path == "/buy":
|
# 签名验证
|
params = self.__parse_request()
|
if not self.__is_sign_right(params):
|
result_str = json.dumps({"code": 1001, "msg": "签名错误"})
|
return
|
print("买入", params)
|
logger_request_api.info(f"买入：{params}")
|
# 买入
|
code = params.get("code") # 代码
|
volume = params.get("volume") # 量
|
price = params.get("price")
|
if not price:
|
# 没有上传价格，就需要获取最近的价格进行买入
|
data = data_cache.latest_code_market_info_dict.get(code)
|
if not data:
|
raise Exception("没有获取到L1数据")
|
pre_price = data[1]
|
current_price = data[2] if data[2] else data[5][0][0]
|
price = tool.get_buy_max_price(current_price)
|
price = min(price, tool.get_limit_up_price(code, pre_price))
|
else:
|
price = round(float(params.get("price")), 2) # 价格
|
result = huaxin_trade_api.order(1, code, volume, price, blocking=True)
|
result_str = json.dumps(result)
|
elif url.path == "/sell":
|
params = self.__parse_request()
|
# 签名验证
|
if not self.__is_sign_right(params):
|
result_str = json.dumps({"code": 1001, "msg": "签名错误"})
|
return
|
# 卖出
|
try:
|
print("卖出", params)
|
code = params.get("code") # 代码
|
volume = params.get("volume") # 量
|
price = params.get("price")
|
if not price:
|
# 没有上传价格，就需要获取最近的价格进行买入
|
data = data_cache.latest_code_market_info_dict.get(code)
|
if not data:
|
raise Exception("没有获取到L1数据")
|
pre_price = data[1]
|
current_price = data[2] if data[2] else data[5][0][0]
|
# 获取最新成交价格
|
latest_deal_price = data_cache.latest_deal_price_dict.get(code)
|
if latest_deal_price:
|
current_price = round(float(latest_deal_price), 2)
|
async_log_util.info(logger_debug, f"根据成交价卖出：{code}-{latest_deal_price}")
|
|
price = tool.get_buy_min_price(current_price)
|
price = max(price, tool.get_limit_down_price(code, pre_price))
|
else:
|
price = round(params.get("price"), 2) # 价格
|
result = huaxin_trade_api.order(2, code, volume, price, blocking=True)
|
result_str = json.dumps(result)
|
finally:
|
logger_request_api.info(f"卖出：{params}")
|
|
elif url.path == "/set_buy_money":
|
# 设置每次买入的金额
|
params = self.__parse_request()
|
# 签名验证
|
if not self.__is_sign_right(params):
|
result_str = json.dumps({"code": 1001, "msg": "签名错误"})
|
return
|
# 卖出
|
print("每次买入的金额", params)
|
money = params.get("money") # 金额
|
if money is None:
|
result_str = json.dumps({"code": 1, "msg": "未上传金额"})
|
return
|
money = int(money)
|
|
logger_debug.info(f"设置开仓金额：{money}")
|
data_cache.BUY_MONEY_PER_CODE = money
|
result_str = json.dumps({"code": 0})
|
|
elif url.path == "/set_limit_up_sell":
|
# 设置每次买入的金额
|
params = self.__parse_request()
|
# 签名验证
|
if not self.__is_sign_right(params):
|
result_str = json.dumps({"code": 1001, "msg": "签名错误"})
|
return
|
# 卖出
|
print("每次买入的金额", params)
|
code = params.get("code") #代码
|
enable = params.get("enable") # 是否开启
|
if code is None or enable is None:
|
result_str = json.dumps({"code": 1, "msg": "上传数据缺失"})
|
return
|
enable = int(enable)
|
if enable:
|
data_cache.LIMIT_UP_SELL_CODES.add(code)
|
else:
|
data_cache.LIMIT_UP_SELL_CODES.discard(code)
|
result_str = json.dumps({"code": 0})
|
|
elif url.path == "/cancel_order":
|
params = self.__parse_request()
|
# 签名验证
|
if not self.__is_sign_right(params):
|
result_str = json.dumps({"code": 1001, "msg": "签名错误"})
|
return
|
# 卖出
|
print("撤单", params)
|
direction = params.get("direction")
|
code = params.get("code") # 代码
|
orderSysID = params.get("orderSysID") # 系统订单编号
|
result = huaxin_trade_api.cancel_order(direction, code, orderSysID, blocking=True)
|
result_str = json.dumps(result)
|
elif url.path == "/upload_deal_big_orders":
|
# 成交大单传递
|
datas = self.rfile.read(int(self.headers['content-length']))
|
_str = str(datas, encoding="gbk")
|
datas = json.loads(_str)
|
for d in datas:
|
if d[1] != 0:
|
continue
|
code, data = d[0], d[2]
|
if code not in data_cache.big_order_deal_dict:
|
data_cache.big_order_deal_dict[code] = []
|
data_cache.big_order_deal_dict[code].append(d)
|
# 获取买大单数量
|
len(data_cache.big_order_deal_dict.get(code, []))
|
hx_logger_l2_transaction.info(_str)
|
# 记录日志
|
result_str = json.dumps({"code": 0})
|
except Exception as e:
|
result_str = json.dumps({"code": 1, "msg": str(e)})
|
finally:
|
self.__send_response(result_str)
|
|
def __send_response(self, data):
|
# 发给请求客户端的响应数据
|
self.send_response(200)
|
self.send_header('Content-type', 'application/json')
|
self.end_headers()
|
self.wfile.write(data.encode())
|
|
def __parse_request(self):
|
params = {}
|
datas = self.rfile.read(int(self.headers['content-length']))
|
_str = str(datas, encoding="gbk")
|
# print(_str)
|
try:
|
params = json.loads(_str)
|
return params
|
except:
|
return _str
|
|
|
class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
|
pass
|
|
|
def run(addr="0.0.0.0", port=12881):
|
handler = DataServer
|
try:
|
httpd = ThreadedHTTPServer((addr, port), handler)
|
print("HTTP server is at: http://%s:%d/" % (addr, port))
|
httpd.serve_forever()
|
except Exception as e:
|
pass
|
|
|
if __name__ == "__main__":
|
run()
|
