import json
|
import logging
|
import os
|
import threading
|
|
import constant
|
|
from api.outside_api_command_manager import ActionCallback
|
from code_attribute import gpcode_manager
|
from huaxin_client import l1_subscript_codes_manager
|
from huaxin_client.client_network import SendResponseSkManager
|
from log_module import async_log_util
|
from strategy import strategy_params_settings, env_info, strategy_manager
|
from strategy.env_info import RealTimeEnvInfo
|
from strategy.place_order_queue_manager import PlaceOrderRecordManager
|
from strategy.strategy_manager import PlateWhiteListManager
|
from strategy.strategy_params_settings import StrategyParamsSettingsManager, StrategyParamsSettings
|
from strategy.strategy_variable import StockVariables
|
from strategy.strategy_variable_factory import DataLoader
|
from third_data.history_k_data_manager import TradeDateManager
|
from third_data.kpl_block_manager import KPLCodeJXBlocksManager
|
from trade import trade_record_log_util
|
from trade.trade_manager import TradeStateManager, DealCodesManager, logger_trade, logger_debug
|
from utils import socket_util, middle_api_protocol, tool, huaxin_util
|
|
OPERRATE_SET = 1 # 设置
|
OPERRATE_DELETE = 2 # 删除
|
OPERRATE_GET = 3 # 获取
|
OPERRATE_ADD = 4 # 新增
|
|
|
class MyAPICallback(ActionCallback):
|
|
@classmethod
|
def __send_response(cls, data_bytes):
|
sk = SendResponseSkManager.create_send_response_sk(addr=middle_api_protocol.SERVER_HOST,
|
port=middle_api_protocol.SERVER_PORT)
|
try:
|
data_bytes = socket_util.load_header(data_bytes)
|
sk.sendall(data_bytes)
|
result, header_str = socket_util.recv_data(sk)
|
result = json.loads(result)
|
if result["code"] != 0:
|
raise Exception(result['msg'])
|
finally:
|
sk.close()
|
|
def send_response(self, data, _client_id, _request_id):
|
data_bytes = json.dumps({"type": "response", "data": data, "client_id": _client_id,
|
"request_id": _request_id}).encode('utf-8')
|
for i in range(3):
|
try:
|
self.__send_response(data_bytes)
|
# print("发送数据成功")
|
break
|
except Exception as e1:
|
logging.exception(e1)
|
|
def __on_get_settings(self):
|
"""
|
获取交易参数
|
@return:
|
"""
|
result = strategy_params_settings.StrategyParamsSettingsManager().get_settings().to_json_str()
|
result = json.loads(result)
|
result["trade_state"] = 1 if TradeStateManager().is_can_buy_cache() else 0
|
|
return {"code": 0, "data": result}
|
|
def __on_set_settings(self, data):
|
"""
|
设置交易参数
|
@return:
|
"""
|
settings = strategy_params_settings.StrategyParamsSettingsManager().get_settings()
|
settings_vars = vars(settings)
|
for k in data:
|
if k == "trade_state":
|
if data[k]:
|
TradeStateManager().open_buy()
|
else:
|
TradeStateManager().close_buy()
|
continue
|
if k not in settings_vars:
|
# 没有在属性里面
|
continue
|
settings.__setattr__(k, data[k])
|
strategy_params_settings.StrategyParamsSettingsManager().set_settings(settings)
|
# 同步参数设置
|
return {"code": 0, "data": {}}
|
|
def __on_get_env(self, need_hsitory_data):
|
"""
|
获取环境信息
|
@param need_hsitory_data: 是否需要历史数据
|
@return:
|
"""
|
fdata = {}
|
# 实时数据
|
fdata["real_time_data"] = RealTimeEnvInfo().to_dict()
|
# 历史数据
|
fdata["history_data"] = {}
|
print("获取环境", os.getpid())
|
if need_hsitory_data:
|
if tool.get_now_time_str() < '16:00:00':
|
# 如果在16:00之前采用当前日期
|
day = tool.get_now_date_str()
|
else:
|
# 如果在16:00之后采用下一个交易日
|
day = TradeDateManager().get_next_trade_day(tool.get_now_date_str())
|
fdata["history_data"][
|
"leading_limit_up_block_codes_count"] = env_info.get_leading_limit_up_block_codes_count(
|
day)
|
|
if tool.get_now_time_str() < '16:00:00':
|
# 如果在16:00之前采用当前日期
|
day = TradeDateManager().get_previous_trade_day(tool.get_now_date_str())
|
else:
|
# 如果在16:00之后采用下一个交易日
|
day = tool.get_now_date_str()
|
fdata["history_data"]["k_bars_count"] = env_info.get_history_k_bars(day)
|
|
day = tool.get_now_date_str()
|
fdata["history_data"]["kpl_code_jx_blocks_count"] = env_info.get_kpl_code_jx_blocks(day)
|
|
return {"code": 0, "data": fdata, "msg": "测试结果"}
|
|
def __on_update_leading_limit_up_datas(self):
|
"""
|
更新领涨代码信息
|
@return:
|
"""
|
|
def update():
|
plates = __DataLoader.get_limit_up_reasons_with_plate_code()
|
for p in plates:
|
print(p)
|
__DataLoader.load_plate_codes(p[0], p[1])
|
|
if tool.get_now_time_str() < '16:00:00':
|
# 如果在16:00之前采用当前日期
|
day = tool.get_now_date_str()
|
else:
|
# 如果在16:00之后采用下一个交易日
|
day = TradeDateManager().get_next_trade_day(tool.get_now_date_str())
|
__DataLoader = DataLoader(day)
|
threading.Thread(target=lambda: update(), daemon=True).start()
|
return {"code": 0}
|
|
def __on_update_kpl_code_jx_blocks_datas(self):
|
"""
|
更新开盘啦精选板块数据
|
@return:
|
"""
|
|
def update():
|
codes = set()
|
codes_sh, codes_sz = l1_subscript_codes_manager.get_codes()
|
codes |= set([x.decode() for x in codes_sh])
|
codes |= set([x.decode() for x in codes_sz])
|
KPLCodeJXBlocksManager(day, codes).start_download_blocks()
|
# 如果在16:00之前采用当前日期
|
|
day = tool.get_now_date_str()
|
threading.Thread(target=lambda: update(), daemon=True).start()
|
return {"code": 0}
|
|
def __on_init_data(self):
|
try:
|
strategy_manager.low_suction_strtegy.load_data()
|
return {"code": 0}
|
except Exception as e:
|
logging.exception(e)
|
return {"code": 1, "msg": str(e)}
|
|
def __on_plate_white_list(self, data):
|
operate = data["operate"]
|
if operate == OPERRATE_GET:
|
plates = PlateWhiteListManager().get_plates()
|
return {"code": 0, "data": list(plates)}
|
elif operate == OPERRATE_ADD:
|
plate = data["plate"]
|
PlateWhiteListManager().add_plate(plate)
|
return {"code": 0}
|
elif operate == OPERRATE_DELETE:
|
plate = data["plate"]
|
PlateWhiteListManager().remove_plate(plate)
|
return {"code": 0}
|
|
def OnCommonRequest(self, client_id, request_id, data):
|
ctype = data["ctype"]
|
result_json = {}
|
if ctype == "get_settings":
|
result_json = self.__on_get_settings()
|
elif ctype == 'set_settings':
|
del data["ctype"]
|
result_json = self.__on_set_settings(data)
|
elif ctype == 'get_env':
|
# 获取环境数据
|
result_json = self.__on_get_env(data.get("history"))
|
elif ctype == 'update_leading_limit_up_datas':
|
# 更新领涨数据
|
result_json = self.__on_update_leading_limit_up_datas()
|
elif ctype == 'update_kpl_code_jx_blocks_datas':
|
# 更新开盘啦精选数据
|
result_json = self.__on_update_kpl_code_jx_blocks_datas()
|
elif ctype == 'init_datas':
|
# 初始化数据
|
result_json = self.__on_init_data()
|
elif ctype == 'get_place_order_records':
|
# 获取下单记录
|
datas = trade_record_log_util.get_trade_records(trade_record_log_util.TYPE_PLACE_ORDER)
|
for data in datas:
|
data[3]["code_name"] = gpcode_manager.CodesNameManager().get_code_name(data[3]["code"])
|
result_json = {"code": 0, "data": datas}
|
elif ctype == 'get_can_buy_plates':
|
# 获取可以买的板块
|
settings: StrategyParamsSettings = StrategyParamsSettingsManager().get_settings()
|
try:
|
current_limit_up_plate_codes = strategy_manager.low_suction_strtegy.current_limit_up_plate_codes
|
plates = [plate for plate, codes in current_limit_up_plate_codes.items() if
|
len(codes) >= settings.limit_up_count_of_new_plate]
|
except:
|
plates = []
|
result_json = {"code": 0, "data": plates}
|
elif ctype == 'plate_white_list':
|
result_json = self.__on_plate_white_list(data)
|
elif ctype == 'get_not_process_place_order_record':
|
# 获取未处理的下单记录
|
record = PlaceOrderRecordManager(tool.get_now_date_str()).get_not_process_record()
|
if not record:
|
result_json = {"code": 1, "msg": '暂无记录'}
|
else:
|
# (ID,代码,板块信息, 大单信息, 时间, 价格, 涨幅)
|
fdata = {
|
"id": record[0],
|
"code": record[1],
|
"name": gpcode_manager.CodesNameManager().get_code_name(record[1]),
|
"platesInfo": {k: [(xx, gpcode_manager.CodesNameManager().get_code_name(xx)) for xx in v] for k, v in
|
record[2].items()},
|
"time_str": record[4],
|
"bigOrderInfo": record[3]
|
}
|
result_json = {"code": 0, "data": fdata}
|
elif ctype == 'set_place_order_buy':
|
id_ = data.get("id")
|
place_order_record_manager = PlaceOrderRecordManager(tool.get_now_date_str())
|
# (ID,代码,板块信息, 大单信息, 时间, 价格, 涨幅)
|
record = place_order_record_manager.get_not_process_record_by_id(id_)
|
if not record:
|
result_json = {"code": 1, "msg": '记录不存在'}
|
else:
|
code = record[1]
|
sv: StockVariables = strategy_manager.low_suction_strtegy.stock_variables_dict.get(code)
|
if sv is None:
|
result_json = {"code": 1, "msg": '代码变量对象不存在'}
|
else:
|
try:
|
place_order_record_manager.set_buy(record[0])
|
# 可以下单
|
# 判断是否可以买
|
order_ref = huaxin_util.create_order_ref()
|
price = tool.get_buy_max_price(sv.当前价)
|
volume = 100
|
DealCodesManager().place_order(set(record[2].keys()), record[1], order_ref, price, volume)
|
trade_record_log_util.add_place_order_log(code, trade_record_log_util.PlaceOrderInfo(code=code,
|
time_str=
|
tool.get_now_time_str(),
|
price=sv.当前价,
|
rate=round(( sv.当前价 - sv.昨日收盘价) * 100 / sv.昨日收盘价, 2),
|
plates=set(record[2].keys()),
|
plates_info= record[2],
|
info=record[3]
|
))
|
async_log_util.info(logger_trade, f"{code}下单，板块：{record[2].keys()}")
|
result_json = {"code": 0, "msg": '处理成功'}
|
except Exception as e:
|
logger_trade.exception(e)
|
result_json = {"code": 0, "msg": str(e)}
|
elif ctype == 'set_place_order_not_buy':
|
id_ = data.get("id")
|
place_order_record_manager = PlaceOrderRecordManager(tool.get_now_date_str())
|
# (ID,代码,板块信息, 大单信息, 时间, 价格, 涨幅)
|
record = place_order_record_manager.get_not_process_record_by_id(id_)
|
if not record:
|
result_json = {"code": 1, "msg": '记录不存在'}
|
else:
|
place_order_record_manager.set_not_buy(record[0], "手动驳回")
|
result_json = {"code": 0, "msg": '处理成功'}
|
elif ctype == 'add_black_list':
|
code = data.get("code")
|
gpcode_manager.BlackListCodeManager().add_code(code)
|
place_order_record_manager = PlaceOrderRecordManager(tool.get_now_date_str())
|
# (ID,代码,板块信息, 大单信息, 时间, 价格, 涨幅)
|
records = place_order_record_manager.get_not_process_records_by_code(code)
|
logger_debug.info(f"未处理记录：{code}-{len(records)}")
|
if records:
|
for r in records:
|
place_order_record_manager.set_not_buy(r[0], "代码拉黑")
|
result_json = {"code": 0, "msg": '添加成功'}
|
|
self.send_response(result_json, client_id, request_id)
|
