"""
|
安全笔数管理
|
"""
|
# 下单L2的安全笔数管理
|
import json
|
|
from db.redis_manager_delegate import RedisUtils
|
from l2 import l2_data_source_util
|
from trade import l2_trade_factor
|
from db import redis_manager_delegate as redis_manager
|
from utils import tool
|
from l2.l2_data_util import L2DataUtil, local_today_buyno_map
|
|
|
class BuyL2SafeCountManager(object):
|
__db = 0
|
__redis_manager = redis_manager.RedisManager(0)
|
__instance = None
|
latest_place_order_info_cache = {}
|
safe_count_l2_cache = {}
|
|
def __new__(cls, *args, **kwargs):
|
if not cls.__instance:
|
cls.__instance = super(BuyL2SafeCountManager, cls).__new__(cls, *args, **kwargs)
|
cls.__load_datas()
|
return cls.__instance
|
|
def __init__(self):
|
self.last_buy_queue_data = {}
|
|
@classmethod
|
def __get_redis(cls):
|
return cls.__redis_manager.getRedis()
|
|
@classmethod
|
def __load_datas(cls):
|
__redis = cls.__get_redis()
|
try:
|
keys = RedisUtils.keys(__redis, "safe_count_l2-*")
|
for k in keys:
|
ks = k.split("-")
|
code, last_buy_single_index = ks[1], int(ks[2])
|
val = RedisUtils.get(__redis, k)
|
val = json.loads(val)
|
tool.CodeDataCacheUtil.set_cache(cls.safe_count_l2_cache, f"{code}-{last_buy_single_index}", val)
|
keys = RedisUtils.keys(__redis, "latest_place_order_info-*")
|
for k in keys:
|
ks = k.split("-")
|
code = ks[-1]
|
val = RedisUtils.get(__redis, k)
|
val = json.loads(val)
|
tool.CodeDataCacheUtil.set_cache(cls.latest_place_order_info_cache, code, val)
|
finally:
|
RedisUtils.realse(__redis)
|
|
# 记录每一次的处理进度
|
def __save_compute_progress(self, code, last_buy_single_index, process_index, buy_num, cancel_num):
|
key = "safe_count_l2-{}-{}".format(code, last_buy_single_index)
|
tool.CodeDataCacheUtil.set_cache(self.safe_count_l2_cache, f"{code}-{last_buy_single_index}",
|
(last_buy_single_index, process_index, buy_num, cancel_num))
|
RedisUtils.setex_async(self.__db, key, tool.get_expire(),
|
json.dumps((last_buy_single_index, process_index, buy_num, cancel_num)))
|
|
# 返回数据与更新时间
|
def __get_compute_progress(self, code, last_buy_single_index):
|
key = "safe_count_l2-{}-{}".format(code, last_buy_single_index)
|
val = RedisUtils.get(self.__get_redis(), key)
|
if val is None:
|
return None, -1, 0, 0
|
val = json.loads(val)
|
return val[0], val[1], val[2], val[3]
|
|
def __get_compute_progress_cache(self, code, last_buy_single_index):
|
cache_result = tool.CodeDataCacheUtil.get_cache(self.safe_count_l2_cache, f"{code}-{last_buy_single_index}")
|
if cache_result[0]:
|
return cache_result[1]
|
return None, -1, 0, 0
|
|
# 保存最近的下单信息
|
def __save_latest_place_order_info(self, code, buy_single_index, buy_exec_index, cancel_index):
|
tool.CodeDataCacheUtil.set_cache(self.latest_place_order_info_cache, code,
|
(buy_single_index, buy_exec_index, cancel_index))
|
key = "latest_place_order_info-{}".format(code)
|
RedisUtils.setex_async(self.__db, key, tool.get_expire(),
|
json.dumps((buy_single_index, buy_exec_index, cancel_index)))
|
|
def __get_latest_place_order_info(self, code):
|
key = "latest_place_order_info-{}".format(code)
|
val = RedisUtils.get(self.__get_redis(), key)
|
if val is None:
|
return None, None, None
|
val = json.loads(val)
|
return val[0], val[1], val[2]
|
|
def __get_latest_place_order_info_cache(self, code):
|
cache_result = tool.CodeDataCacheUtil.get_cache(self.latest_place_order_info_cache, code)
|
if cache_result[0]:
|
return cache_result[1]
|
return None, None, None
|
|
def __get_all_compute_progress(self, code):
|
key_regex = f"safe_count_l2-{code}-*"
|
keys = RedisUtils.keys(self.__get_redis(), key_regex)
|
vals = []
|
for k in keys:
|
val = RedisUtils.get(self.__get_redis(), k)
|
val = json.loads(val)
|
vals.append(val)
|
return vals
|
|
def clear_data(self, code):
|
key_regex = f"safe_count_l2-{code}-*"
|
keys = RedisUtils.keys(self.__get_redis(), key_regex)
|
for k in keys:
|
RedisUtils.delete(self.__get_redis(), k)
|
|
tool.CodeDataCacheUtil.clear_cache(self.latest_place_order_info_cache, code)
|
key = f"latest_place_order_info-{code}"
|
RedisUtils.delete_async(self.__db, key)
|
|
# 获取基础的安全笔数
|
def __get_base_save_count(self, code, is_first):
|
return l2_trade_factor.L2TradeFactorUtil.get_safe_buy_count(code, is_first)
|
|
# 获取最后的安全笔数
|
def get_safe_count(self, code, is_first_code, rate):
|
# rate = self.__get_rate(code)
|
count, min_count, max_count = self.__get_base_save_count(code, is_first_code)
|
# 第4次下单按第一次算
|
# if place_order_count and place_order_count >= 3:
|
# rate = 1
|
# print("--------------------------------")
|
# print("安全笔数比例：", rate)
|
# print("--------------------------------")
|
# count, min_count, max_count = self.__get_base_save_count(code, is_first_code)
|
# count = round(count * rate)
|
# if count < min_count:
|
# count = min_count
|
# if count > max_count:
|
# count = max_count
|
return int(round(count * (1 + rate), 0))
|
|
# 计算留下来的比例
|
# last_buy_single_index 上一次下单信号起始位置
|
# cancel_index 上一次取消下单的位置
|
# start_index 数据开始位置
|
# end_index 数据结束位置
|
def compute_left_rate(self, code, start_index, end_index, total_datas,
|
local_today_num_operate_map):
|
last_buy_single_index, buy_exec_index, cancel_index = self.__get_latest_place_order_info_cache(code)
|
if last_buy_single_index is None:
|
return
|
cancel_time = None
|
if cancel_index is not None:
|
cancel_time = total_datas[cancel_index]["val"]["time"]
|
# 获取处理的进度
|
last_buy_single_index_, process_index, buy_num, cancel_num = self.__get_compute_progress_cache(code,
|
last_buy_single_index)
|
|
break_index = -1
|
for i in range(start_index, end_index):
|
data = total_datas[i]
|
val = data["val"]
|
# 如果没有取消位置就一直计算下去， 计算截至时间不能大于取消时间
|
if cancel_time and int(cancel_time.replace(":", "")) < int(val["time"].replace(":", "")):
|
break_index = i
|
break
|
if break_index >= 0:
|
end_index = break_index - 1
|
# 获取开始计算的位置
|
start_compute_index = min(start_index, last_buy_single_index)
|
if start_compute_index <= process_index:
|
start_compute_index = process_index + 1
|
|
for i in range(start_compute_index, end_index):
|
data = total_datas[i]
|
val = data["val"]
|
if process_index >= i:
|
continue
|
if L2DataUtil.is_limit_up_price_buy(val):
|
# 涨停买
|
buy_num += int(val["num"]) * data["re"]
|
elif L2DataUtil.is_limit_up_price_buy_cancel(val):
|
# 获取买入信息
|
buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data_v2(data, local_today_buyno_map.get(code))
|
if buy_index is not None:
|
if last_buy_single_index <= buy_index <= end_index:
|
cancel_num += int(val["num"]) * data["re"]
|
|
process_index = end_index
|
# 保存处理进度与数量
|
self.__save_compute_progress(code, last_buy_single_index, process_index, buy_num, cancel_num)
|
|
# 获取比例
|
def __get_rate(self, code):
|
vals = self.__get_all_compute_progress(code)
|
rate = (1 - 0)
|
for val in vals:
|
temp_rate = (1 - round((val[2] - val[3]) / val[2], 4))
|
if temp_rate > 1:
|
temp_rate = 1
|
rate *= temp_rate
|
return rate
|
|
# 下单成功
|
def save_place_order_info(self, code, buy_single_index, buy_exec_index, cancel_index):
|
self.__save_latest_place_order_info(code, buy_single_index, buy_exec_index, cancel_index)
|
