import copy
|
import logging
|
import threading
|
import time
|
import time as t
|
|
from cancel_strategy.s_l_h_cancel_strategy import HourCancelBigNumComputer, LCancelRateManager
|
from cancel_strategy.s_l_h_cancel_strategy import LCancelBigNumComputer
|
from cancel_strategy.s_l_h_cancel_strategy import SCancelBigNumComputer
|
from code_attribute import big_money_num_manager, code_volumn_manager, code_data_util, limit_up_time_manager, \
|
gpcode_manager, code_nature_analyse
|
import constant
|
from db.redis_manager_delegate import RedisUtils
|
from l2.huaxin import l2_huaxin_util, huaxin_delegate_postion_manager
|
from l2.l2_limitup_sell_data_manager import L2LimitUpSellDataManager
|
from l2.l2_sell_manager import L2MarketSellManager, L2LimitUpSellManager
|
from l2.l2_transaction_data_manager import HuaXinSellOrderStatisticManager, BigOrderDealManager, HuaXinBuyOrderManager
|
from l2.place_order_single_data_manager import L2TradeSingleDataProcessor
|
from log_module import async_log_util, log_export
|
from third_data import kpl_data_manager, block_info
|
from third_data.kpl_data_constant import LimitUpDataConstant
|
from trade.buy_radical.block_special_codes_manager import BlockSpecialCodesManager
|
from trade.buy_radical.radical_buy_data_manager import EveryLimitupBigDealOrderManager
|
from utils import global_util, tool, buy_condition_util, buy_strategy_util, trade_util
|
import l2_data_util
|
from db import redis_manager_delegate as redis_manager
|
from third_data.code_plate_key_manager import CodePlateKeyBuyManager, KPLCodeJXBlockManager
|
from trade import trade_manager, trade_queue_manager, l2_trade_factor, l2_trade_util, \
|
trade_result_manager, current_price_process_manager, trade_data_manager, trade_huaxin, trade_record_log_util, \
|
trade_constant, buy_open_limit_up_strategy
|
from trade.buy_radical import radical_buy_data_manager, radical_buy_strategy
|
from l2 import l2_data_manager, l2_log, l2_data_source_util, code_price_manager, \
|
transaction_progress, cancel_buy_strategy, place_order_single_data_manager
|
from l2.cancel_buy_strategy import DCancelBigNumComputer, \
|
LatestCancelIndexManager, \
|
NewGCancelBigNumComputer, JCancelBigNumComputer, L2DataComputeUtil, RDCancelBigNumComputer
|
from l2.l2_data_manager import L2DataException, OrderBeginPosInfo
|
from l2.l2_data_util import local_today_datas, L2DataUtil, local_today_num_operate_map, local_today_buyno_map, \
|
local_latest_datas, local_today_canceled_buyno_map, local_today_sellno_map
|
import l2.l2_data_util
|
from log_module.log import logger_l2_trade_buy, logger_l2_process, logger_l2_error, logger_debug, \
|
logger_l2_not_buy_reasons, logger_real_place_order_position, logger_l2_trade_buy_queue
|
|
from trade.trade_data_manager import CodeActualPriceProcessor, PlaceOrderCountManager, AccountMoneyManager, \
|
RadicalBuyDealCodesManager
|
|
from trade.trade_manager import TradeTargetCodeModeManager, CodesTradeStateManager
|
|
from settings.trade_setting import MarketSituationManager, TradeBlockBuyModeManager
|
|
import concurrent.futures
|
|
|
class L2DataManager:
|
# 格式化数据
|
def format_data(self, datas):
|
format_datas = []
|
for data in datas:
|
format_datas.append({"val": data, "re": 1})
|
return format_datas
|
|
# 获取新增数据
|
def get_add_datas(self, format_datas):
|
pass
|
|
# 从数据库加载数据
|
def load_data(self, code=None, force=False):
|
pass
|
|
# 保存数据
|
def save_datas(self, add_datas, datas):
|
pass
|
|
|
# m值大单处理
|
|
|
class L2BigNumForMProcessor:
|
_db = 1
|
_redis_manager = redis_manager.RedisManager(1)
|
m_big_money_begin_cache = {}
|
m_big_money_process_index_cache = {}
|
__instance = None
|
|
def __new__(cls, *args, **kwargs):
|
if not cls.__instance:
|
cls.__instance = super(L2BigNumForMProcessor, cls).__new__(cls, *args, **kwargs)
|
cls.__load_datas()
|
return cls.__instance
|
|
@classmethod
|
def __get_redis(cls):
|
return cls._redis_manager.getRedis()
|
|
@classmethod
|
def __load_datas(cls):
|
_redis = cls._redis_manager.getRedis()
|
try:
|
keys = RedisUtils.keys(_redis, "m_big_money_begin-*")
|
for k in keys:
|
code = k.split("-")[-1]
|
val = RedisUtils.get(_redis, k)
|
tool.CodeDataCacheUtil.set_cache(cls.m_big_money_begin_cache, code, int(val))
|
|
keys = RedisUtils.keys(_redis, "m_big_money_process_index-*")
|
for k in keys:
|
code = k.split("-")[-1]
|
val = RedisUtils.get(_redis, k)
|
tool.CodeDataCacheUtil.set_cache(cls.m_big_money_process_index_cache, code, int(val))
|
|
|
|
finally:
|
RedisUtils.realse(_redis)
|
|
# 保存计算开始位置
|
def set_begin_pos(self, code, index):
|
if self.__get_begin_pos_cache(code) is None:
|
tool.CodeDataCacheUtil.set_cache(self.m_big_money_begin_cache, code, index)
|
# 保存位置
|
key = "m_big_money_begin-{}".format(code)
|
RedisUtils.setex_async(self._db, key, tool.get_expire(), index)
|
|
# 获取计算开始位置
|
def __get_begin_pos(self, code):
|
key = "m_big_money_begin-{}".format(code)
|
val = RedisUtils.get(self.__get_redis(), key)
|
if val is None:
|
return None
|
return int(val)
|
|
def __get_begin_pos_cache(self, code):
|
cache_result = tool.CodeDataCacheUtil.get_cache(self.m_big_money_begin_cache, code)
|
if cache_result[0]:
|
return cache_result[1]
|
return None
|
|
# 清除已经处理的数据
|
def clear_processed_end_index(self, code):
|
tool.CodeDataCacheUtil.clear_cache(self.m_big_money_process_index_cache, code)
|
key = "m_big_money_process_index-{}".format(code)
|
RedisUtils.delete_async(self._db, key)
|
|
# 添加已经处理过的单
|
def __set_processed_end_index(self, code, index):
|
tool.CodeDataCacheUtil.set_cache(self.m_big_money_process_index_cache, code, index)
|
key = "m_big_money_process_index-{}".format(code)
|
RedisUtils.setex_async(self._db, key, tool.get_expire(), index)
|
|
# 是否已经处理过
|
def __get_processed_end_index(self, code):
|
key = "m_big_money_process_index-{}".format(code)
|
val = RedisUtils.get(self.__get_redis(), key)
|
if val is None:
|
return None
|
return int(val)
|
|
def __get_processed_end_index_cache(self, code):
|
cache_result = tool.CodeDataCacheUtil.get_cache(self.m_big_money_process_index_cache, code)
|
if cache_result[0]:
|
return cache_result[1]
|
return None
|
|
# 处理大单
|
def process(self, code, start_index, end_index, limit_up_price):
|
|
begin_pos = self.__get_begin_pos_cache(code)
|
if begin_pos is None:
|
# 没有获取到开始买入信号
|
return
|
# 上次处理到的坐标
|
processed_index = self.__get_processed_end_index_cache(code)
|
if processed_index is None:
|
processed_index = 0
|
if processed_index >= end_index:
|
return
|
|
start_time = round(t.time() * 1000)
|
total_datas = local_today_datas[code]
|
|
num_splites = [round(5000 / limit_up_price), round(10000 / limit_up_price), round(20000 / limit_up_price),
|
round(30000 / limit_up_price)]
|
total_num = 0
|
buyno_map = local_today_buyno_map.get(code)
|
for i in range(max(start_index, processed_index), end_index + 1):
|
data = total_datas[i]
|
if not L2DataUtil.is_limit_up_price_buy_cancel(data["val"]) and not L2DataUtil.is_limit_up_price_buy(
|
data["val"]):
|
continue
|
# 如果是涨停买撤信号需要看数据位置是否比开始处理时间早
|
if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]):
|
# 获取买入信号
|
buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data_v2(total_datas[i], buyno_map)
|
if buy_index is not None and buy_index < begin_pos:
|
continue
|
|
# 计算成交金额
|
num = int(data["val"]["num"])
|
temp = 0
|
if num < num_splites[0]:
|
pass
|
elif num < num_splites[1]:
|
temp = 1
|
elif num < num_splites[2]:
|
temp = round(4 / 3, 3)
|
elif num < num_splites[3]:
|
temp = 2
|
else:
|
temp = 4
|
count = int(temp * data["re"] * 1000)
|
if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]):
|
count = 0 - count
|
total_num += count
|
self.__set_processed_end_index(code, end_index)
|
big_money_num_manager.add_num(code, total_num)
|
|
# print("m值大单计算范围：{}-{} 时间：{}".format(max(start_index, processed_index), end_index,
|
# round(t.time() * 1000) - start_time))
|
|
|
class L2TradeDataProcessor:
|
unreal_buy_dict = {}
|
volume_rate_info = {}
|
# 最近的闪电下单位置,格式为{code:(总卖时间,总卖额)}
|
__latest_fast_place_order_info_dict = {}
|
__codeActualPriceProcessor = CodeActualPriceProcessor()
|
__ths_l2_trade_queue_manager = trade_queue_manager.thsl2tradequeuemanager()
|
__thsBuy1VolumnManager = trade_queue_manager.THSBuy1VolumnManager()
|
__l2PlaceOrderParamsManagerDict = {}
|
__last_buy_single_dict = {}
|
__TradeBuyQueue = transaction_progress.TradeBuyQueue()
|
__latest_process_order_unique_keys = {}
|
__latest_process_not_order_unique_keys_count = {}
|
__trade_log_placr_order_info_dict = {} # 下单信息保存
|
# 初始化
|
__TradePointManager = l2_data_manager.TradePointManager()
|
__SCancelBigNumComputer = SCancelBigNumComputer()
|
__HourCancelBigNumComputer = HourCancelBigNumComputer()
|
__LCancelBigNumComputer = LCancelBigNumComputer()
|
__GCancelBigNumComputer = NewGCancelBigNumComputer()
|
__TradeStateManager = trade_manager.TradeStateManager()
|
__CodesTradeStateManager = trade_manager.CodesTradeStateManager()
|
__PauseBuyCodesManager = gpcode_manager.PauseBuyCodesManager()
|
__Buy1PriceManager = code_price_manager.Buy1PriceManager()
|
__AccountMoneyManager = AccountMoneyManager()
|
__TradeBuyDataManager = trade_data_manager.TradeBuyDataManager()
|
__LimitUpTimeManager = limit_up_time_manager.LimitUpTimeManager()
|
__BlackListCodeManager = gpcode_manager.BlackListCodeManager()
|
__WhiteListCodeManager = gpcode_manager.WhiteListCodeManager()
|
__WantBuyCodesManager = gpcode_manager.WantBuyCodesManager()
|
__TradeTargetCodeModeManager = TradeTargetCodeModeManager()
|
__TradeOrderIdManager = trade_huaxin.TradeOrderIdManager()
|
__LatestCancelIndexManager = LatestCancelIndexManager()
|
__L2MarketSellManager = L2MarketSellManager()
|
__L2LimitUpSellManager = L2LimitUpSellManager()
|
__PlaceOrderCountManager = PlaceOrderCountManager()
|
__CodeNatureRecordManager = code_nature_analyse.CodeNatureRecordManager()
|
__MarketSituationManager = MarketSituationManager()
|
__re_compute_threading_pool = concurrent.futures.ThreadPoolExecutor(max_workers=10)
|
|
# 买入锁
|
__buy_lock_dict = {}
|
|
# 当前批次正在处理的数据索引
|
__processing_data_indexes = {}
|
|
# 下次买的时间
|
__next_buy_time_dict = {}
|
|
# 中断本批次买入数据处理
|
__break_current_batch_data_for_buy_dict = {}
|
|
# 最近的执行位置
|
__latest_exec_indexes = {}
|
|
# 获取代码评分
|
@classmethod
|
def get_code_scores(cls):
|
score_dict = {}
|
for code in cls.__l2PlaceOrderParamsManagerDict:
|
score = cls.__l2PlaceOrderParamsManagerDict[code].score
|
score_dict[code] = score
|
return score_dict
|
|
@classmethod
|
# 数据处理入口
|
# datas: 本次截图数据
|
# capture_timestamp:截图时间戳
|
def process(cls, code, datas, capture_timestamp):
|
__start_time = round(t.time() * 1000)
|
try:
|
if len(datas) > 0:
|
# 判断价格区间是否正确
|
if not code_data_util.is_same_code_with_price(code, float(datas[0]["val"]["price"])):
|
raise L2DataException(L2DataException.CODE_PRICE_ERROR,
|
"股价不匹配 code-{} price-{}".format(code, datas[0]["val"]["price"]))
|
# 加载历史数据,返回数据是否正常
|
is_normal = l2.l2_data_util.load_l2_data(code)
|
if not is_normal:
|
# print("历史数据异常:", code)
|
# 数据不正常需要禁止交易
|
l2_trade_util.forbidden_trade(code, msg="L2历史数据异常", force=True)
|
# 纠正数据
|
if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_THS:
|
# 同花顺需要纠正数据，其他渠道不需要
|
datas = l2.l2_data_util.L2DataUtil.correct_data(code, local_latest_datas.get(code), datas)
|
_start_index = 0
|
if local_today_datas.get(code) is not None and len(
|
local_today_datas[code]) > 0:
|
_start_index = local_today_datas[code][-1]["index"] + 1
|
add_datas = l2.l2_data_util.L2DataUtil.get_add_data(code, local_latest_datas.get(code), datas,
|
_start_index)
|
# -------------数据增量处理------------
|
try:
|
cls.process_add_datas(code, add_datas, capture_timestamp, __start_time)
|
finally:
|
# 保存数据
|
__start_time = round(t.time() * 1000)
|
l2.l2_data_util.save_l2_data(code, datas, add_datas)
|
# __start_time = l2_data_log.l2_time(code,
|
# round(t.time() * 1000) - __start_time,
|
# "保存数据时间（{}）".format(len(add_datas)))
|
finally:
|
if code in cls.unreal_buy_dict:
|
cls.unreal_buy_dict.pop(code)
|
|
@classmethod
|
def set_real_place_order_index(cls, code, index, order_begin_pos: OrderBeginPosInfo):
|
trade_record_log_util.add_real_place_order_position_log(code, index, order_begin_pos.buy_single_index)
|
total_datas = local_today_datas.get(code)
|
use_time = tool.trade_time_sub_with_ms(L2DataUtil.get_time_with_ms(total_datas[index]["val"]) , L2DataUtil.get_time_with_ms(
|
total_datas[order_begin_pos.buy_exec_index]["val"]))
|
trade_record_log_util.add_place_order_use_time(code, f"执行位时间：{L2DataUtil.get_time_with_ms(total_datas[order_begin_pos.buy_exec_index]['val'])} 耗时：{use_time}")
|
l2_log.debug(code, "设置真实下单位：{}", index)
|
cancel_buy_strategy.set_real_place_position(code, index, order_begin_pos.buy_single_index, is_default=False)
|
# 获取真实下单位置之后需要判断F撤
|
try:
|
cancel_result = cancel_buy_strategy.FCancelBigNumComputer().need_cancel_for_deal_fast(code)
|
if cancel_result[0]:
|
L2TradeDataProcessor.cancel_buy(code, f"F撤:{cancel_result[1]}",
|
cancel_type=trade_constant.CANCEL_TYPE_F)
|
return
|
else:
|
l2_log.f_cancel_debug(code, f"获取真实成交位的F撤未生效：{cancel_result[1]}")
|
except Exception as e:
|
logger_debug.exception(e)
|
# 判断与执行位置的间隔时间
|
if order_begin_pos.mode != OrderBeginPosInfo.MODE_RADICAL:
|
# 非扫入下单要判断执行位置与真实下单位的间隔时间
|
total_datas = local_today_datas.get(code)
|
if tool.trade_time_sub_with_ms(L2DataUtil.get_time_with_ms(total_datas[index]["val"]),
|
L2DataUtil.get_time_with_ms(
|
total_datas[order_begin_pos.buy_exec_index]["val"])) > 2000:
|
L2TradeDataProcessor.cancel_buy(code, f"真实下单位({index})与执行位({order_begin_pos.buy_exec_index})相差2s以上",
|
cancel_type=trade_constant.CANCEL_TYPE_F)
|
return
|
|
# 处理华鑫L2数据
|
@classmethod
|
def process_huaxin(cls, code, origin_datas):
|
datas = None
|
try:
|
# 加载历史的L2数据
|
is_normal = l2.l2_data_util.l2_data_is_normal(code)
|
if not is_normal:
|
# 数据不正常需要禁止交易
|
l2_trade_util.forbidden_trade(code, msg="L2历史数据异常", force=True)
|
# 转换数据格式
|
_start_index = 0
|
total_datas = local_today_datas.get(code)
|
if code not in local_today_datas:
|
local_today_datas[code] = []
|
if total_datas:
|
_start_index = total_datas[-1]["index"] + 1
|
|
# 如果是板上，就需要剔除板上卖数据, 默认不剔除
|
filter_limit_up_sell = False
|
trade_price_info = HuaXinSellOrderStatisticManager.get_latest_trade_price_info(code)
|
limit_up_price = gpcode_manager.get_limit_up_price(code)
|
if limit_up_price:
|
limit_up_price = round(float(limit_up_price), 2)
|
# if trade_price_info and limit_up_price and trade_price_info[0] == limit_up_price:
|
# filter_limit_up_sell = True
|
|
datas = l2_huaxin_util.get_format_l2_datas(code, origin_datas, limit_up_price, _start_index,
|
filter_limit_up_sell)
|
L2LimitUpSellDataManager.add_l2_data(code, datas)
|
__start_time = round(t.time() * 1000)
|
if len(datas) > 0:
|
cls.process_add_datas(code, datas, 0, __start_time)
|
except Exception as e:
|
async_log_util.error(logger_l2_error, f"code:{code}")
|
# async_log_util.exception(logger_l2_error, e)
|
logger_l2_error.exception(e)
|
finally:
|
if datas:
|
l2.l2_data_util.save_l2_data(code, None, datas)
|
origin_datas.clear()
|
|
@classmethod
|
def __recompute_real_order_index(cls, code, pre_real_order_index, order_info, compute_type):
|
# 1s之后重新计算
|
time.sleep(1)
|
real_order_index = huaxin_delegate_postion_manager.RealDelegateOrderPositionManager().recompute_l2_place_order_position(
|
code, order_info, pre_real_order_index, compute_type)
|
if real_order_index and pre_real_order_index != real_order_index:
|
try:
|
exec_index = order_info[6]
|
order_begin_pos = cls.__get_order_begin_pos(
|
code)
|
async_log_util.info(logger_debug,
|
f"下单位矫正：真实下单位-{real_order_index} 订单信息-{order_info} 下单信息-{order_begin_pos.to_dict()}")
|
if order_begin_pos and order_begin_pos.buy_exec_index == exec_index:
|
cls.set_real_place_order_index(code, real_order_index, order_begin_pos)
|
async_log_util.info(logger_real_place_order_position,
|
f"真实下单位置矫正：{code}-({real_order_index},1)")
|
except Exception as e:
|
logger_debug.exception(e)
|
async_log_util.error(logger_debug, f"重新寻找真实下单位异常：{code}-{order_info}")
|
|
@classmethod
|
def process_add_datas(cls, code, add_datas, capture_timestamp, __start_time):
|
now_time_str = tool.get_now_time_str()
|
# 将本次中断设置为
|
cls.__break_current_batch_data_for_buy_dict[code] = False
|
if add_datas:
|
# 记录当前批数据的索引
|
cls.__processing_data_indexes[code] = (add_datas[0]["index"], add_datas[-1]["index"])
|
if code not in cls.__trade_log_placr_order_info_dict:
|
cls.__trade_log_placr_order_info_dict[code] = trade_record_log_util.PlaceOrderInfo()
|
# 拼接数据
|
local_today_datas[code].extend(add_datas)
|
l2.l2_data_util.load_num_operate_map(local_today_num_operate_map, code, add_datas)
|
l2.l2_data_util.load_buy_no_map(local_today_buyno_map, code, add_datas)
|
l2.l2_data_util.load_sell_no_map(local_today_sellno_map, code, add_datas)
|
l2.l2_data_util.load_canceled_buy_no_map(local_today_canceled_buyno_map, code, add_datas)
|
if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_HUAXIN:
|
try:
|
if constant.TEST:
|
pass
|
# order_begin_pos = cls.__get_order_begin_pos(code)
|
# if order_begin_pos.buy_exec_index and order_begin_pos.buy_exec_index>=0:
|
# place_order_index = add_datas[-1]["index"]
|
# cls.set_real_place_order_index(code, place_order_index, order_begin_pos.buy_single_index)
|
else:
|
# ------------09:30:00之前处理排1的问题--------------
|
if int(add_datas[-1]["val"]["time"].replace(":", "")) < 93000:
|
# 获取是否下单
|
buy_info = buy_open_limit_up_strategy.BuyOpenLimitupDataManager().get_place_order_info(code)
|
# 设置虚拟下单信息
|
if buy_info:
|
try:
|
buy_single_index, buy_exec_index, buy_exec_index = add_datas[0]["index"], \
|
add_datas[0]["index"], \
|
add_datas[0]["index"]
|
buy_volume_rate = 0
|
sell_info = ("09:15:00", 0)
|
threshold_money = 0
|
order_begin_pos_info = OrderBeginPosInfo(buy_single_index=buy_single_index,
|
buy_exec_index=buy_exec_index,
|
buy_compute_index=buy_exec_index,
|
num=1, count=1,
|
max_num_set=set(),
|
buy_volume_rate=buy_volume_rate,
|
mode=OrderBeginPosInfo.MODE_OPEN_LIMIT_UP,
|
mode_desc=f"排1下单",
|
sell_info=sell_info,
|
threshold_money=threshold_money)
|
cls.__save_order_begin_data(code, order_begin_pos_info)
|
# 设置下单信息
|
limit_up_price = gpcode_manager.get_limit_up_price_as_num(code)
|
shadow_price = tool.get_shadow_price(limit_up_price)
|
huaxin_delegate_postion_manager.place_order(code, limit_up_price, buy_info[0],
|
buy_exec_index,
|
add_datas[0], buy_info[2],
|
shadow_price=shadow_price,
|
shadow_volume=buy_info[1])
|
cls.__place_order_success(code, order_begin_pos_info, None)
|
except Exception as e:
|
logger_l2_error.exception(e)
|
finally:
|
buy_open_limit_up_strategy.BuyOpenLimitupDataManager().remove_place_order_info(code)
|
|
# 获取下单位置
|
|
if constant.IS_NEW_VERSION_PLACE_ORDER:
|
place_order_index, order_info, compute_type = huaxin_delegate_postion_manager.RealDelegateOrderPositionManager.compute_l2_place_order_position(
|
code, add_datas)
|
else:
|
place_order_index, order_info, compute_type = huaxin_delegate_postion_manager.get_l2_place_order_position(
|
code, float(
|
gpcode_manager.get_limit_up_price(code)), add_datas)
|
|
if place_order_index:
|
order_begin_pos = cls.__get_order_begin_pos(
|
code)
|
cls.set_real_place_order_index(code, place_order_index, order_begin_pos)
|
try:
|
cls.__re_compute_threading_pool.submit(
|
cls.__recompute_real_order_index, code, place_order_index, order_info,
|
compute_type)
|
except:
|
pass
|
async_log_util.info(logger_l2_process, f"code:{code} 获取到下单真实位置：{place_order_index}")
|
|
# 处理涨停卖与涨停卖撤
|
try:
|
for d in add_datas:
|
if L2DataUtil.is_limit_up_price_sell(d['val']):
|
L2TradeSingleDataProcessor.add_l2_delegate_limit_up_sell(code, d)
|
elif L2DataUtil.is_limit_up_price_sell_cancel(d['val']):
|
L2TradeSingleDataProcessor.add_l2_delegate_limit_up_sell_cancel(code,
|
d['val']['orderNo'])
|
except Exception as e:
|
logger_debug.exception(e)
|
except:
|
async_log_util.error(logger_l2_error, f"{code} 处理真实下单位置出错")
|
# 第1条数据是否为09:30:00
|
if add_datas[0]["val"]["time"] == "09:30:00":
|
if global_util.cuurent_prices.get(code):
|
price_data = global_util.cuurent_prices.get(code)
|
if price_data[1]:
|
# 当前涨停价，设置涨停时间
|
async_log_util.info(logger_l2_process, f"开盘涨停：{code}")
|
# 保存涨停时间
|
cls.__LimitUpTimeManager.save_limit_up_time(code, "09:30:00")
|
|
total_datas = local_today_datas[code]
|
# __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time,
|
# "l2数据预处理时间")
|
|
# 9:29:55开始处理数据
|
place_ordered_desc = ""
|
if len(add_datas) > 0 and int(tool.get_now_time_str().replace(":", "")) > int("092955"):
|
# 是否为首板代码
|
is_first_code = True # gpcode_manager.FirstCodeManager().is_in_first_record(code)
|
# 计算量
|
current_sell = cls.__L2MarketSellManager.get_current_total_sell_data(code)
|
total_sell_volume = 0
|
if current_sell and len(current_sell) > 2:
|
total_sell_volume = current_sell[2]
|
volume_rate = code_volumn_manager.CodeVolumeManager().get_volume_rate(code,
|
total_sell_volume=total_sell_volume)
|
volume_rate_index = code_volumn_manager.CodeVolumeManager().get_volume_rate_index(volume_rate)
|
# 计算分值
|
limit_up_time = cls.__LimitUpTimeManager.get_limit_up_time_cache(code)
|
if limit_up_time is None:
|
limit_up_time = tool.get_now_time_str()
|
|
score = None
|
cls.__l2PlaceOrderParamsManagerDict[code] = l2_trade_factor.L2PlaceOrderParamsManager(code, is_first_code,
|
volume_rate,
|
volume_rate_index,
|
score,
|
total_datas[-1][
|
'val']['time'])
|
cls.volume_rate_info[code] = (volume_rate, volume_rate_index)
|
|
latest_time = add_datas[-1]["val"]["time"]
|
|
# __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time,
|
# "l2数据准备时间")
|
# 时间差不能太大才能处理
|
if not l2_trade_util.is_in_forbidden_trade_codes(code):
|
# 计算板上卖,当数据少时才计算，否则不计算
|
try:
|
if len(add_datas) < 20:
|
has_limit_up_sell = False
|
for d in add_datas:
|
if L2DataUtil.is_limit_up_price_sell(d["val"]):
|
if d["val"]["num"] * float(d["val"]["price"]) < 5000:
|
continue
|
cls.__L2LimitUpSellManager.add_limit_up_sell(code, d["index"])
|
has_limit_up_sell = True
|
if has_limit_up_sell:
|
LCancelRateManager.compute_big_num_deal_info(code)
|
# elif L2DataUtil.is_limit_up_price_sell_cancel(d["val"]):
|
# cls.__L2LimitUpSellManager.add_limit_up_sell(code, d["index"])
|
except Exception as e:
|
async_log_util.error(logger_l2_error, f"计算板上卖出错：{str(e)}")
|
|
# 判断是否已经挂单
|
state = cls.__CodesTradeStateManager.get_trade_state_cache(code)
|
start_index = len(total_datas) - len(add_datas)
|
end_index = len(total_datas) - 1
|
|
if state == trade_constant.TRADE_STATE_BUY_DELEGATED or state == trade_constant.TRADE_STATE_BUY_PLACE_ORDER or state == trade_constant.TRADE_STATE_BUY_SUCCESS:
|
# 已挂单
|
place_ordered_desc = "已下单"
|
cls.__process_order(code, start_index, end_index, capture_timestamp, is_first_code)
|
else:
|
place_ordered_desc = "未下单"
|
# 未挂单,时间相差不大才能挂单
|
# tool.trade_time_sub(latest_time, "09:32:00") < 0
|
if l2.l2_data_util.L2DataUtil.is_same_time(
|
now_time_str, latest_time):
|
cls.__process_not_order(code, start_index, end_index, capture_timestamp, is_first_code)
|
else:
|
place_ordered_desc += ":数据延迟"
|
|
l2_log.info(code, logger_l2_process, "code:{} 处理数据范围: {}-{} 处理时间:{} 线程ID：{} 处理情况：{}", code,
|
add_datas[0]["index"],
|
add_datas[-1]["index"], round(t.time() * 1000) - __start_time,
|
l2_log.threadIds.get(code), place_ordered_desc)
|
|
# 处理未挂单
|
@classmethod
|
def __process_not_order(cls, code, start_index, end_index, capture_time, is_first_code):
|
__start_time = round(t.time() * 1000)
|
# 获取阈值
|
threshold_money, msg = cls.__get_threshmoney(code)
|
cls.__start_compute_buy(code, start_index, end_index, threshold_money, capture_time, is_first_code)
|
|
# 测试专用
|
@classmethod
|
def process_order(cls, code, start_index, end_index, capture_time, is_first_code, new_add=True):
|
cls.__process_order(code, start_index, end_index, capture_time, is_first_code, new_add)
|
|
# 处理已挂单
|
@classmethod
|
def __process_order(cls, code, start_index, end_index, capture_time, is_first_code, new_add=True):
|
# 增加推出机制
|
unique_key = f"{start_index}-{end_index}"
|
if cls.__latest_process_order_unique_keys.get(code) == unique_key:
|
async_log_util.error(logger_l2_error, f"重复处理数据：code-{code} start_index-{start_index} end_index-{end_index}")
|
return
|
cls.__latest_process_order_unique_keys[code] = unique_key
|
|
# H撤
|
def h_cancel(_buy_single_index, _buy_exec_index):
|
_start_time = round(t.time() * 1000)
|
try:
|
b_need_cancel, b_cancel_data = cls.__HourCancelBigNumComputer.need_cancel(code, _buy_single_index,
|
_buy_exec_index, start_index,
|
end_index, total_data,
|
code_volumn_manager.CodeVolumeManager().get_volume_rate_index(
|
order_begin_pos.buy_volume_rate),
|
cls.volume_rate_info[code][1],
|
is_first_code)
|
if b_need_cancel and b_cancel_data:
|
return b_cancel_data, "H撤"
|
except Exception as e:
|
if constant.TEST:
|
logging.exception(e)
|
# TODO 可能耗时
|
logger_l2_error.exception(e)
|
async_log_util.error(logger_l2_error,
|
f"H撤出错 参数：buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} {str(e)}")
|
async_log_util.exception(logger_l2_error, e)
|
# logger_l2_error.exception(e)
|
finally:
|
# l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-H撤大单计算")
|
pass
|
return None, ""
|
|
# L撤
|
def l_cancel(_buy_single_index, _buy_exec_index):
|
_start_time = round(t.time() * 1000)
|
try:
|
b_need_cancel, b_cancel_data, extra_msg, b_cancel_type = cls.__LCancelBigNumComputer.need_cancel(code,
|
_buy_exec_index,
|
start_index,
|
end_index,
|
total_data,
|
is_first_code)
|
if b_need_cancel and b_cancel_data:
|
return b_cancel_data, f"L撤({extra_msg})", b_cancel_type
|
except Exception as e:
|
async_log_util.error(logger_l2_error,
|
f"L撤出错 参数：buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} 错误原因：{str(e)}")
|
# logger_l2_error.exception(e)
|
async_log_util.exception(logger_l2_error, e)
|
finally:
|
# l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-L撤大单计算")
|
pass
|
return None, "", None
|
|
# G撤
|
def g_cancel(_buy_single_index, _buy_exec_index):
|
try:
|
b_need_cancel, b_cancel_data, extra_msg = cls.__GCancelBigNumComputer.need_cancel(code,
|
_buy_exec_index,
|
start_index,
|
end_index)
|
if b_need_cancel and b_cancel_data:
|
return b_cancel_data, f"G撤({extra_msg})"
|
except Exception as e:
|
async_log_util.error(logger_l2_error,
|
f"G撤出错 参数：buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} 错误原因：{str(e)}")
|
# logger_l2_error.exception(e)
|
async_log_util.exception(logger_l2_error, e)
|
finally:
|
# l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "已下单-L撤大单计算")
|
pass
|
return None, ""
|
|
# B撤
|
def b_cancel(_buy_single_index, _buy_exec_index):
|
try:
|
b_need_cancel, b_cancel_data, extra_msg = cls.__GCancelBigNumComputer.need_cancel_for_b(code,
|
start_index,
|
end_index)
|
if b_need_cancel and b_cancel_data:
|
return b_cancel_data, f"G撤({extra_msg})"
|
except Exception as e:
|
async_log_util.error(logger_l2_error,
|
f"B撤出错 参数：buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} 错误原因：{str(e)}")
|
return None, ""
|
|
# J撤
|
def j_cancel(_buy_single_index, _buy_exec_index):
|
try:
|
b_need_cancel, b_cancel_data, extra_msg = JCancelBigNumComputer().need_cancel(code, start_index,
|
end_index)
|
if b_need_cancel and b_cancel_data:
|
return b_cancel_data, f"J撤({extra_msg})"
|
except Exception as e:
|
async_log_util.error(logger_l2_error,
|
f"J撤出错 参数：buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} 错误原因：{str(e)}")
|
return None, ""
|
|
# RD撤
|
def rd_cancel(_buy_single_index, _buy_exec_index):
|
try:
|
b_need_cancel, b_cancel_data, extra_msg = RDCancelBigNumComputer().need_cancel(code, start_index,
|
end_index)
|
if b_need_cancel and b_cancel_data:
|
big_order_info = radical_buy_data_manager.get_total_deal_big_order_info(code,
|
gpcode_manager.get_limit_up_price_as_num(
|
code))
|
if big_order_info[0] > 0:
|
return b_cancel_data, f"RD撤({extra_msg})", trade_constant.CANCEL_TYPE_RD
|
except Exception as e:
|
async_log_util.error(logger_l2_error,
|
f"RD撤出错 参数：buy_single_index-{_buy_single_index} buy_exec_index-{_buy_exec_index} 错误原因：{str(e)}")
|
return None, "", trade_constant.CANCEL_TYPE_RD
|
|
if start_index < 0:
|
start_index = 0
|
|
if end_index < start_index:
|
return
|
total_data = local_today_datas.get(code)
|
_start_time = tool.get_now_timestamp()
|
# 获取买入信号起始点
|
order_begin_pos = cls.__get_order_begin_pos(
|
code)
|
# 默认量为0.2
|
if order_begin_pos.buy_volume_rate is None:
|
order_begin_pos.buy_volume_rate = 0.2
|
cancel_data, cancel_msg, cancel_type = None, "", None
|
if order_begin_pos.mode == OrderBeginPosInfo.MODE_RADICAL:
|
|
if not cancel_data:
|
cancel_data, cancel_msg, cancel_type = rd_cancel(order_begin_pos.buy_single_index,
|
order_begin_pos.buy_exec_index)
|
|
# 扫入下单只有L撤
|
if not cancel_data:
|
cancel_data, cancel_msg, cancel_type = l_cancel(order_begin_pos.buy_single_index,
|
order_begin_pos.buy_exec_index)
|
else:
|
if not cancel_data:
|
cancel_data, cancel_msg = g_cancel(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
|
cancel_type = trade_constant.CANCEL_TYPE_G
|
# 依次处理
|
if not cancel_data:
|
cancel_data, cancel_msg, cancel_type = l_cancel(order_begin_pos.buy_single_index,
|
order_begin_pos.buy_exec_index)
|
# B撤
|
if not cancel_data:
|
cancel_data, cancel_msg = b_cancel(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
|
cancel_type = trade_constant.CANCEL_TYPE_G
|
if not cancel_data:
|
cancel_data, cancel_msg = h_cancel(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
|
cancel_type = trade_constant.CANCEL_TYPE_H
|
# J撤
|
if not cancel_data:
|
cancel_data, cancel_msg = j_cancel(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
|
cancel_type = trade_constant.CANCEL_TYPE_J
|
|
if cancel_data and not DCancelBigNumComputer().has_auto_cancel_rules(code):
|
l2_log.debug(code, "触发撤单，撤单位置：{} ，撤单原因：{}", cancel_data["index"], cancel_msg)
|
# 撤单
|
cls.cancel_buy(code, cancel_msg, cancel_index=cancel_data["index"], cancel_type=cancel_type)
|
# 撤单成功，继续计算下单
|
cls.__process_not_order(code, cancel_data["index"] + 1, end_index, capture_time, is_first_code)
|
else:
|
pass
|
|
@classmethod
|
def __is_pre_can_buy(cls, code):
|
"""
|
预判断是否可买
|
@param code:
|
@return：
|
"""
|
if not cls.__TradeStateManager.is_can_buy_cache():
|
return False, True, f"今日已禁止交易", True
|
|
if l2_trade_util.is_in_forbidden_trade_codes(code):
|
return False, True, f"代码禁止交易", True
|
|
if cls.__PauseBuyCodesManager.is_in_cache(code):
|
return False, True, f"该代码被暂停交易", True
|
now_time_int = int(tool.get_now_time_str().replace(":", ""))
|
if now_time_int >= 145700:
|
return False, True, f"14:57后不能交易", True
|
# if now_time_int < 93100:
|
# return False, True, f"09:31之前不能交易", True
|
# 二板以上的票不买
|
yesterday_codes = kpl_data_manager.get_yesterday_limit_up_codes()
|
if yesterday_codes and code in yesterday_codes:
|
return False, True, f"不买高位板", True
|
|
if cls.__TradeTargetCodeModeManager.get_mode_cache() == TradeTargetCodeModeManager.MODE_ONLY_BUY_WANT_CODES:
|
if not cls.__WantBuyCodesManager.is_in_cache(
|
code) and not gpcode_manager.GreenListCodeManager().is_in_cache(code):
|
return False, True, f"只买想买：没在想买单和绿单", True
|
if cls.__TradeTargetCodeModeManager.get_mode_cache() == TradeTargetCodeModeManager.MODE_ONLY_BUY_SPECIAL_CODES:
|
special_blocks = BlockSpecialCodesManager().get_code_blocks(code)
|
if not special_blocks:
|
return False, True, f"只买辨识度", True
|
# if not cls.__WantBuyCodesManager.is_in_cache(code):
|
average_rate = cls.__Buy1PriceManager.get_average_rate(code)
|
if average_rate:
|
if tool.is_ge_code(code):
|
if average_rate < 0.8:
|
return False, True, f"均价涨幅({average_rate})小于8%", True
|
else:
|
if average_rate < 0.04:
|
return False, True, f"均价涨幅({average_rate})小于4%", True
|
return True, False, f"", False
|
|
@classmethod
|
def start_buy(cls, code, last_data, last_data_index, is_first_code, block_info):
|
"""
|
开始买入
|
@param code:
|
@param last_data:
|
@param last_data_index:
|
@param is_first_code:
|
@param block_info:板块信息:[(板块,流入信息)]
|
@return:
|
"""
|
return cls.__buy(code, 0, last_data, last_data_index, is_first_code, block_info=block_info)
|
|
@classmethod
|
def get_active_buy_blocks(cls, code):
|
"""
|
获取积极买入的板块
|
@param code:
|
@return:
|
"""
|
if tool.is_sh_code(code):
|
# 上证不积极下单
|
return None
|
|
current_total_sell_data = L2MarketSellManager().get_current_total_sell_data(code)
|
place_order_count = trade_data_manager.PlaceOrderCountManager().get_place_order_count(code)
|
# 只有初次下单 + 总抛压大于500w能积极下单
|
if place_order_count > 0 or current_total_sell_data is None or current_total_sell_data[1] < 500 * 10000:
|
return None
|
|
if global_util.zyltgb_map.get(code) > 50 * 100000000:
|
# 50亿以上的无法积极下单
|
return None
|
can_buy_result = CodePlateKeyBuyManager.can_buy(code)
|
if can_buy_result:
|
if can_buy_result[5]:
|
return can_buy_result[5]
|
elif not can_buy_result[0] and can_buy_result[1]:
|
# 10:00:00 前的独苗可积极买入
|
if tool.trade_time_sub(tool.get_now_time_str(), "10:00:00") <= 0:
|
return ["独苗"]
|
return None
|
|
@classmethod
|
def __place_order_success(cls, code, order_begin_pos, block_info):
|
################下单成功处理################
|
trade_result_manager.real_buy_success(code, cls.__TradePointManager)
|
l2_log.debug(code, "处理买入成功1")
|
cancel_buy_strategy.set_real_place_position(code, local_today_datas.get(code)[-1]["index"],
|
order_begin_pos.buy_single_index, is_default=True)
|
l2_log.debug(code, "处理买入成功2")
|
params_desc = cls.__l2PlaceOrderParamsManagerDict[code].get_buy_rank_desc()
|
l2_log.debug(code, params_desc)
|
#################清除本次下单的大单数据###############
|
EveryLimitupBigDealOrderManager.clear(code, "下单成功")
|
############记录下单时的数据############
|
try:
|
jx_blocks, jx_blocks_by = KPLCodeJXBlockManager().get_jx_blocks_cache(
|
code), KPLCodeJXBlockManager().get_jx_blocks_cache(code, by=True)
|
if jx_blocks:
|
jx_blocks = jx_blocks[0]
|
if jx_blocks_by:
|
jx_blocks_by = jx_blocks_by[0]
|
|
info = cls.__trade_log_placr_order_info_dict[code]
|
info.mode = order_begin_pos.mode
|
info.mode_desc = order_begin_pos.mode_desc
|
info.set_buy_index(order_begin_pos.buy_single_index, order_begin_pos.buy_exec_index)
|
info.set_sell_info(order_begin_pos.sell_info)
|
info.set_block_info(block_info)
|
if jx_blocks:
|
info.set_kpl_blocks(list(jx_blocks))
|
elif jx_blocks_by:
|
info.set_kpl_blocks(list(jx_blocks_by))
|
else:
|
info.set_kpl_blocks([])
|
can_buy_result = CodePlateKeyBuyManager.can_buy(code)
|
if can_buy_result:
|
if not can_buy_result[0] and can_buy_result[1]:
|
info.set_kpl_match_blocks(["独苗"])
|
elif not can_buy_result[0] and not can_buy_result[1]:
|
info.set_kpl_match_blocks(["非独苗不满足身位"])
|
else:
|
temps = []
|
temps.extend([f"{x[0]}" for x in can_buy_result[0]])
|
if can_buy_result[5]:
|
temps.append(f"积极买入：{can_buy_result[5]}")
|
info.set_kpl_match_blocks(temps)
|
trade_record_log_util.add_place_order_log(code, info)
|
except Exception as e:
|
async_log_util.error(logger_l2_error, f"加入买入记录日志出错：{str(e)}")
|
|
@classmethod
|
def __buy(cls, code, capture_timestamp, last_data, last_data_index, is_first_code, block_info=None):
|
pre_result = cls.__is_pre_can_buy(code)
|
if not pre_result[0]:
|
async_log_util.info(logger_l2_not_buy_reasons, f"{code}#{pre_result[2]}")
|
return False
|
# 添加买入锁
|
if code not in cls.__buy_lock_dict:
|
cls.__buy_lock_dict[code] = threading.Lock()
|
|
with cls.__buy_lock_dict[code]:
|
# 判断是否可以下单，不处于可下单状态需要返回
|
state = cls.__CodesTradeStateManager.get_trade_state_cache(code)
|
if not trade_util.is_can_order_by_state(state):
|
# 不处于可下单状态
|
return False
|
__start_time = tool.get_now_timestamp()
|
order_begin_pos = cls.__get_order_begin_pos(
|
code)
|
|
if order_begin_pos.MODE_RADICAL == order_begin_pos.mode:
|
# 积极下单不判断条件
|
can, need_clear_data, reason, is_valid_exec_index = True, False, "扫入下单", True
|
else:
|
return False
|
# can, need_clear_data, reason, is_valid_exec_index = cls.__can_buy_first(code)
|
|
# 删除虚拟下单
|
if code in cls.unreal_buy_dict:
|
cls.unreal_buy_dict.pop(code)
|
|
if not can:
|
if not is_valid_exec_index:
|
if cls.__latest_exec_indexes.get(code) and cls.__latest_exec_indexes[code][
|
-1] == order_begin_pos.buy_exec_index:
|
# 如果执行位置算做无效，就需要删除当前执行位置
|
cls.__latest_exec_indexes[code].pop()
|
l2_log.debug(code, "不可以下单，原因：{}", reason)
|
trade_record_log_util.add_cant_place_order_log(code, reason)
|
if need_clear_data:
|
# 中断买入
|
cls.__break_current_batch_data_for_buy_dict[code] = True
|
trade_result_manager.real_cancel_success(code, order_begin_pos.buy_single_index,
|
order_begin_pos.buy_exec_index,
|
local_today_datas.get(code))
|
return False
|
else:
|
l2_log.debug(code, "可以下单，原因：{}， 下单模式：{}", reason, order_begin_pos.mode)
|
try:
|
|
# 判断是否为首封下单
|
order_begin_pos.first_limit_up_buy = radical_buy_data_manager.is_first_limit_up_buy(code)
|
|
l2_log.debug(code, "开始执行买入")
|
trade_manager.start_buy(code, capture_timestamp, last_data,
|
last_data_index, order_begin_pos.mode, order_begin_pos.buy_exec_index)
|
l2_log.debug(code, "执行买入成功")
|
|
cls.__place_order_success(code, order_begin_pos, block_info)
|
|
except Exception as e:
|
async_log_util.exception(logger_l2_error, e)
|
l2_log.debug(code, "执行买入异常:{}", str(e))
|
pass
|
finally:
|
# l2_log.debug(code, "m值影响因子：{}", l2_trade_factor.L2TradeFactorUtil.factors_to_string(code))
|
pass
|
return True
|
|
# 是否可以取消
|
@classmethod
|
def __can_cancel(cls, code):
|
if constant.TEST:
|
return True, ""
|
if cls.__WhiteListCodeManager.is_in_cache(code):
|
return False, "代码在白名单中"
|
|
# 暂时注释掉
|
# 14点后如果是板块老大就不需要取消了
|
# now_time_str = tool.get_now_time_str()
|
# if int(now_time_str.replace(":", "")) >= 140000:
|
# industry, codes = ths_industry_util.get_same_industry_codes(code, gpcode_manager.get_gp_list())
|
# if industry is None:
|
# return True, "没有获取到行业"
|
# codes_index = industry_codes_sort.sort_codes(codes, code)
|
# if codes_index is not None and codes_index.get(code) is not None:
|
# # 同一板块中老二后面的不能买
|
# if codes_index.get(code) == 0:
|
# return False, "14:00后老大不能撤单"
|
# elif codes_index.get(code) == 1:
|
# # 判断老大是否都是09：30：00涨停的
|
# # 同1板块老大是09:30:00涨停，老二14：00砸开的不撤
|
# first_count = 0
|
# for key in codes_index:
|
# if codes_index[key] == 0:
|
# first_count += 1
|
# if limit_up_time_manager.get_limit_up_time(key) == "09:30:00":
|
# first_count -= 1
|
# if first_count == 0:
|
# return False, "14:00后老大都开盘涨停，老二不能撤单"
|
|
return True, ""
|
|
@classmethod
|
def __is_at_limit_up_buy(cls, code, buy_exec_index=None):
|
"""
|
是否是板上放量买
|
@param code:
|
@param buy_exec_index:
|
@return:
|
"""
|
total_data = local_today_datas.get(code)
|
if buy_exec_index is None:
|
# 执行位置为空，就取最近的一条数据
|
buy_exec_index = total_data[-1]["index"]
|
latest_exec_indexes = cls.__latest_exec_indexes.get(code)
|
if not latest_exec_indexes:
|
latest_exec_indexes = []
|
# 判断是否是炸开后买入
|
last_exec_index = 0
|
if len(latest_exec_indexes) > 1:
|
last_exec_index = latest_exec_indexes[-2]
|
# 获取最近的非涨停价成交时间
|
not_limit_up_trade_time_with_ms = current_price_process_manager.get_trade_not_limit_up_time_with_ms(
|
code)
|
is_limit_up_buy = True
|
if not_limit_up_trade_time_with_ms:
|
t1 = int(
|
L2DataUtil.get_time_with_ms(total_data[last_exec_index]["val"]).replace(":", "").replace(".",
|
""))
|
t2 = int(not_limit_up_trade_time_with_ms.replace(":", "").replace(".", ""))
|
t3 = int(L2DataUtil.get_time_with_ms(total_data[buy_exec_index]["val"]).replace(":", "").replace(
|
".", ""))
|
if t1 < t2 <= t3:
|
# 炸板时间在两次下单时间中间
|
is_limit_up_buy = False
|
return is_limit_up_buy
|
|
# 获取可以买的板块
|
@classmethod
|
def __get_can_buy_block(cls, code):
|
can_buy_result = CodePlateKeyBuyManager.can_buy(code)
|
if can_buy_result is None:
|
async_log_util.warning(logger_debug, "没有获取到板块缓存，将获取板块")
|
latest_current_limit_up_records = kpl_data_manager.get_latest_current_limit_up_records()
|
|
codes_delegate = set(CodesTradeStateManager().get_codes_by_trade_states_cache(
|
{trade_constant.TRADE_STATE_BUY_DELEGATED, trade_constant.TRADE_STATE_BUY_PLACE_ORDER}))
|
codes_success = set(CodesTradeStateManager().get_codes_by_trade_states_cache(
|
{trade_constant.TRADE_STATE_BUY_SUCCESS}))
|
|
CodePlateKeyBuyManager.update_can_buy_blocks(code,
|
kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas,
|
kpl_data_manager.KPLLimitUpDataRecordManager.total_datas,
|
latest_current_limit_up_records,
|
block_info.get_before_blocks_dict(),
|
kpl_data_manager.KPLLimitUpDataRecordManager.get_current_limit_up_reason_codes_dict(),
|
codes_delegate, codes_success)
|
can_buy_result = CodePlateKeyBuyManager.can_buy(code)
|
return can_buy_result
|
|
@classmethod
|
def can_buy_first(cls, code, limit_up_price):
|
now_timestamp = int(tool.get_now_time_str().replace(":", ""))
|
# 判断板块
|
# (可以买的板块列表, 是否是独苗, 消息简介,可买的强势主线, 板块关键词)
|
can_buy_result = cls.__get_can_buy_block(code)
|
# l2_log.debug(code, "获取到的板块信息：{}", can_buy_result)
|
if can_buy_result is None:
|
return False, True, "尚未获取到板块信息"
|
# -------量的约束--------
|
volume_rate_thresholds = buy_condition_util.get_volume_rate_by_level(
|
1), buy_condition_util.get_volume_rate_by_level(2)
|
|
k_format = code_nature_analyse.CodeNatureRecordManager().get_k_format_cache(code)
|
# 是否有辨识度
|
is_special = True if k_format and k_format[8][0] else False
|
|
# 是否是强势10分钟
|
is_in_strong_time = now_timestamp <= int("094000")
|
# 是否是强势30分钟
|
is_in_strong_time_30 = now_timestamp <= int("100000")
|
|
# 量参考是否在最近30天
|
# is_refer_volume_date_in_30_days = False
|
# try:
|
# volume_refer_date, volume_refer_date_distance = code_volumn_manager.get_volume_refer_date(code)
|
# if volume_refer_date_distance < 30:
|
# is_refer_volume_date_in_30_days = True
|
# except:
|
# pass
|
|
# 获取市场行情
|
situation = cls.__MarketSituationManager.get_situation_cache()
|
zylt_threshold_as_yi = buy_condition_util.get_zyltgb_threshold(situation)
|
zyltgb_as_yi = round(global_util.zyltgb_map.get(code) / 100000000, 2)
|
|
if zyltgb_as_yi >= zylt_threshold_as_yi[1]:
|
return False, True, f"{zylt_threshold_as_yi[1]}亿以上的都不买（{zyltgb_as_yi}）"
|
|
if zyltgb_as_yi < zylt_threshold_as_yi[0]:
|
return False, True, f"{zylt_threshold_as_yi[0]}亿以下的都不买（{zyltgb_as_yi}）"
|
|
if constant.ALL_ACTIVE_BUY:
|
return True, False, "买所有"
|
|
# 最优市值
|
is_best_zylt = True if zylt_threshold_as_yi[4] <= zyltgb_as_yi <= zylt_threshold_as_yi[5] else False
|
|
if is_special:
|
# 具有辨识度，算作最优市值
|
zyltgb_as_yi = zylt_threshold_as_yi[2] + 1
|
|
# 是否是最优自由流通市值
|
is_better_zylt = True if zylt_threshold_as_yi[2] <= zyltgb_as_yi <= zylt_threshold_as_yi[3] else False
|
|
if is_in_strong_time_30 and is_best_zylt:
|
# 强势30分钟，最优市值, 有可以下单的板块，不看量
|
return True, False, can_buy_result[2]
|
|
msg_list = []
|
if is_in_strong_time:
|
msg_list.append("强势10分钟")
|
|
# 独苗
|
if not can_buy_result[0] and can_buy_result[1]:
|
msg_list.append("独苗")
|
if not is_better_zylt:
|
# 如果没有辨识度才不买
|
if not is_special:
|
return False, True, f"强势10分钟，无辨识度， 独苗（{can_buy_result[4]}）不下单（{can_buy_result[4]}）自由流通市值（{zyltgb_as_yi}）不是特优市值"
|
elif not can_buy_result[0]:
|
return False, True, f"非独苗不满足身位：{can_buy_result[2]}"
|
else:
|
msg_list.append("非独苗")
|
if not is_better_zylt:
|
msg_list.append("不满足自由流通")
|
# 注释量的影响
|
# if k_format and (k_format[1][0] or k_format[3][0]):
|
# # 股价创新高或者逼近前高
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < volume_rate_thresholds[1]:
|
# return False, True, f"强势10分钟，后排，不满足自由市值，股价创新高或者逼近前高,当日量比({cls.volume_rate_info[code][0]})小于{volume_rate_thresholds[1]}"
|
# else:
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < volume_rate_thresholds[
|
# 0] and not is_special:
|
# return False, True, f"强势10分钟，后排,无辨识度，不满足自由市值，当日量比({cls.volume_rate_info[code][0]})小于{volume_rate_thresholds[0]}"
|
else:
|
msg_list.append("满足自由流通")
|
# 后排，满足自由流通市值需要下单
|
return True, False, can_buy_result[2]
|
return True, False, can_buy_result[2]
|
else:
|
|
# 注释量的影响
|
# # 上5个交易日有炸板之后
|
# if has_open_limit_up_in_5:
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < 0.29:
|
# return False, True, f"非强势10分钟,上5个交易日炸板,量未达到{0.29}({cls.volume_rate_info[code][0]})"
|
#
|
# # 上5个交易日有跌停
|
# if has_limit_down_in_5:
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < 0.29:
|
# return False, True, f"非强势10分钟,上5个交易日跌停,量未达到{0.29}({cls.volume_rate_info[code][0]})"
|
|
# 注释量的影响
|
# # 获取量的参考日期
|
# if code in global_util.max60_volumn:
|
# day = global_util.max60_volumn[code][1]
|
# if day in HistoryKDatasUtils.get_latest_trading_date_cache(5):
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < 0.29 and not is_special:
|
# return False, True, f"参考量在最近5天，无辨识度，量未达到0.29({cls.volume_rate_info[code][0]})"
|
|
# 非强势10分钟只买主线
|
if not can_buy_result[0] and can_buy_result[1]:
|
if not is_special and not is_best_zylt:
|
return False, True, f"非强势10分钟，无辨识度，非特优市值,独苗（{can_buy_result[4]}）不下单"
|
# 注释量的影响
|
# if can_buy_result[3]:
|
# # 强势主线
|
# if not is_better_zylt:
|
# if k_format and (k_format[1][0] or k_format[3][0]):
|
# # 股价创新高或者逼近前高
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < volume_rate_thresholds[1]:
|
# return False, True, f"非强势10分钟，强势主线后排，不满足自由市值，股价创新高或者逼近前高,当日量比({cls.volume_rate_info[code][0]})小于{volume_rate_thresholds[1]}"
|
# else:
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < volume_rate_thresholds[
|
# 0] and not is_special:
|
# return False, True, f"非强势10分钟，强势主线后排,不满足自由市值，无辨识度，当日量比({cls.volume_rate_info[code][0]})小于{volume_rate_thresholds[0]}"
|
# else:
|
# if k_format and (k_format[1][0] or k_format[3][0]):
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < volume_rate_thresholds[0]:
|
# return False, True, f"非强势10分钟，强势主线后排,满足自由市值，股价创新高或者逼近前高, 当日量比({cls.volume_rate_info[code][0]})小于{volume_rate_thresholds[0]}"
|
# else:
|
# # 非强势主线
|
# if not is_better_zylt:
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < volume_rate_thresholds[
|
# 1] and not is_special:
|
# return False, True, f"非强势10分钟，非强势主线后排【主线后排】，不满足自由市值,无辨识度， 当日量比({cls.volume_rate_info[code][0]})小于{volume_rate_thresholds[1]}"
|
# else:
|
# if k_format and (k_format[1][0] or k_format[3][0]):
|
# # 股价创新高或者逼近前高
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < volume_rate_thresholds[1]:
|
# return False, True, f"非强势10分钟，非强势主线后排【主线后排】，满足自由市值，股价创新高或者逼近前高, 当日量比({cls.volume_rate_info[code][0]})小于{volume_rate_thresholds[1]}"
|
# else:
|
# if code in cls.volume_rate_info and cls.volume_rate_info[code][0] < volume_rate_thresholds[
|
# 0] and not is_special:
|
# return False, True, f"非强势10分钟，非强势主线后排【主线后排】, 满足自由市值，无辨识度，当日量比({cls.volume_rate_info[code][0]})小于{volume_rate_thresholds[0]}"
|
return True, False, can_buy_result[2]
|
|
@classmethod
|
def can_buy_first_new(cls, code, limit_up_price):
|
now_timestamp = int(tool.get_now_time_str().replace(":", ""))
|
# 判断板块
|
# (可以买的板块列表, 是否是独苗, 消息简介,可买的强势主线, 板块关键词)
|
can_buy_result = cls.__get_can_buy_block(code)
|
if can_buy_result is None:
|
return False, True, "尚未获取到板块信息"
|
|
# 是否是强势30分钟
|
is_in_strong_time_30 = now_timestamp <= int("100000")
|
|
# 获取市场行情
|
situation = cls.__MarketSituationManager.get_situation_cache()
|
zylt_threshold_as_yi = buy_condition_util.get_zyltgb_threshold(situation)
|
zyltgb_as_yi = round(global_util.zyltgb_map.get(code) / 100000000, 2)
|
|
if zyltgb_as_yi < zylt_threshold_as_yi[0]:
|
return False, True, f"{zylt_threshold_as_yi[0]}亿以下的都不买（{zyltgb_as_yi}）"
|
# 测试时可取消注释
|
# if 1 > 0:
|
# return True, False, "买所有"
|
|
if zyltgb_as_yi >= zylt_threshold_as_yi[1]:
|
return False, True, f"{zylt_threshold_as_yi[1]}亿以上的都不买（{zyltgb_as_yi}）"
|
|
msg_list = []
|
if is_in_strong_time_30:
|
msg_list.append("强势30分钟")
|
# 独苗
|
if not can_buy_result[0] and can_buy_result[1]:
|
msg_list.append("独苗")
|
return True, False, "强势30分钟，独苗"
|
elif not can_buy_result[0]:
|
return False, True, f"强势30分钟，非独苗不满足身位：{can_buy_result[2]}"
|
else:
|
msg_list.append("非独苗")
|
# 后排，满足自由流通市值需要下单
|
return True, False, can_buy_result[2]
|
else:
|
place_order_count = trade_data_manager.PlaceOrderCountManager().get_place_order_count(code)
|
if place_order_count > 0:
|
return True, False, "之前下过单"
|
if not can_buy_result[0]:
|
# 没有板块
|
if can_buy_result[1]:
|
# 是独苗
|
if not TradeBlockBuyModeManager().can_buy_unique_block():
|
# 不能买独苗
|
return False, True, f"非强势30分钟，独苗，不满足身位：{can_buy_result[2]}"
|
else:
|
# 非独苗，没有可以买入的板块
|
return False, True, f"非强势30分钟，非独苗，不满足身位：{can_buy_result[2]}"
|
return True, False, can_buy_result[2]
|
|
@classmethod
|
def __cancel_buy(cls, code):
|
if code not in cls.__buy_lock_dict:
|
cls.__buy_lock_dict[code] = threading.Lock()
|
with cls.__buy_lock_dict[code]:
|
try:
|
l2_log.debug(code, "开始执行撤单")
|
trade_manager.start_cancel_buy(code)
|
l2_log.debug(code, "执行撤单成功")
|
return True
|
except Exception as e:
|
logging.exception(e)
|
l2_log.debug(code, "执行撤单异常：{}", str(e))
|
return False
|
finally:
|
pass
|
|
@classmethod
|
def cancel_buy(cls, code, msg=None, source="l2", cancel_index=None, cancel_type=None):
|
|
finally_can_cancel = True
|
total_datas = local_today_datas.get(code)
|
order_begin_pos = cls.__get_order_begin_pos(
|
code)
|
if cancel_type != trade_constant.CANCEL_TYPE_HUMAN:
|
# 是否是交易队列触发
|
# 扫入下单只有L撤能撤单
|
if order_begin_pos and order_begin_pos.mode == OrderBeginPosInfo.MODE_RADICAL and cancel_type not in {
|
trade_constant.CANCEL_TYPE_L_DOWN, trade_constant.CANCEL_TYPE_L, trade_constant.CANCEL_TYPE_RD,
|
trade_constant.CANCEL_TYPE_P}:
|
l2_log.cancel_debug(code, "撤单中断，原因：{}", "扫入下单不是L撤")
|
return False
|
# 加绿只有L撤/人撤生效
|
if gpcode_manager.GreenListCodeManager().is_in_cache(code):
|
if cancel_type not in {trade_constant.CANCEL_TYPE_L, trade_constant.CANCEL_TYPE_L_UP,
|
trade_constant.CANCEL_TYPE_L_DOWN, trade_constant.CANCEL_TYPE_RD,
|
trade_constant.CANCEL_TYPE_P}:
|
l2_log.cancel_debug(code, "撤单中断，原因：{}", "加绿不是L撤")
|
return False
|
|
if not total_datas:
|
return False
|
if source == "trade_queue":
|
# 交易队列触发的需要下单后5s
|
if order_begin_pos.buy_exec_index is not None and order_begin_pos.buy_exec_index > 0:
|
now_time_str = tool.get_now_time_str()
|
if tool.trade_time_sub(now_time_str,
|
total_datas[order_begin_pos.buy_exec_index]["val"]["time"]) < 5:
|
return False
|
|
if code in cls.unreal_buy_dict:
|
cls.unreal_buy_dict.pop(code)
|
# 取消买入标识
|
trade_result_manager.virtual_cancel_success(code, order_begin_pos.buy_single_index,
|
order_begin_pos.buy_exec_index, total_datas)
|
else:
|
can_cancel, reason = cls.__can_cancel(code)
|
if not can_cancel:
|
# 不能取消且不是人工撤单
|
l2_log.cancel_debug(code, "撤单中断，原因：{}", reason)
|
l2_log.debug(code, "撤单中断，原因：{}", reason)
|
return False
|
|
if finally_can_cancel:
|
if cancel_index is None:
|
cancel_index = total_datas[-1]["index"]
|
cls.__LatestCancelIndexManager.set_latest_cancel_index(code, cancel_index)
|
|
# 记录撤单日志
|
real_place_order_index = SCancelBigNumComputer().get_real_place_order_index_cache(code)
|
# 添加撤单日志记录
|
trade_record_log_util.add_cancel_msg_log(code, real_place_order_index, msg)
|
cancel_result = cls.__cancel_buy(code)
|
if cancel_result:
|
trade_result_manager.real_cancel_success(code, order_begin_pos.buy_single_index,
|
order_begin_pos.buy_exec_index, total_datas,
|
from_real_cancel=True)
|
l2_log.debug(code, "执行撤单结束，原因：{}", msg)
|
return True
|
|
# 虚拟下单
|
@classmethod
|
def __virtual_buy(cls, code, buy_single_index, buy_exec_index, capture_time):
|
cls.unreal_buy_dict[code] = (buy_exec_index, capture_time)
|
trade_result_manager.virtual_buy_success(code)
|
|
@classmethod
|
def __process_with_find_exec_index(cls, code, order_begin_pos: OrderBeginPosInfo, compute_end_index, block_info):
|
"""
|
处理找到执行位置
|
@return:
|
"""
|
total_datas = local_today_datas.get(code)
|
l2_log.debug(code, "获取到买入执行位置：{} m值：{} 纯买手数：{} 纯买单数：{} 是否板上买：{} 数据：{} ,量比:{} ,下单模式：{}",
|
order_begin_pos.buy_exec_index,
|
order_begin_pos.threshold_money,
|
order_begin_pos.num,
|
order_begin_pos.count, order_begin_pos.at_limit_up, total_datas[order_begin_pos.buy_exec_index],
|
cls.volume_rate_info[code], order_begin_pos.mode)
|
cls.__save_order_begin_data(code, order_begin_pos)
|
cls.__LimitUpTimeManager.save_limit_up_time(code, total_datas[order_begin_pos.buy_exec_index]["val"]["time"])
|
l2_log.debug(code, "delete_buy_cancel_point")
|
if code not in cls.__latest_exec_indexes:
|
cls.__latest_exec_indexes[code] = []
|
cls.__latest_exec_indexes[code].append(order_begin_pos.buy_exec_index)
|
# 保留最近3次的买入执行位置
|
if len(cls.__latest_exec_indexes[code]) > 3:
|
cls.__latest_exec_indexes[code] = cls.__latest_exec_indexes[code][-3:]
|
|
# 直接下单
|
ordered = cls.__buy(code, 0, total_datas[-1], total_datas[-1]["index"], True, block_info=block_info)
|
|
# 数据是否处理完毕
|
if order_begin_pos.buy_exec_index < compute_end_index:
|
if ordered:
|
cls.__process_order(code, order_begin_pos.buy_exec_index + 1, compute_end_index, 0, True,
|
False)
|
else:
|
cls.__start_compute_buy(code, order_begin_pos.buy_exec_index + 1, compute_end_index,
|
order_begin_pos.threshold_money,
|
0,
|
True, False)
|
return ordered
|
|
@classmethod
|
def __start_compute_buy(cls, code, compute_start_index, compute_end_index, threshold_money, capture_time,
|
is_first_code,
|
new_add=True):
|
|
# 中断当前批次买入数据处理
|
if cls.__break_current_batch_data_for_buy_dict.get(code):
|
return
|
|
# 判断下次买入时间是否正确
|
if code in cls.__next_buy_time_dict and t.time() < cls.__next_buy_time_dict[code]:
|
l2_log.debug(code, f"下次可下单时间（{compute_start_index}-{compute_end_index}）：{cls.__next_buy_time_dict[code]}")
|
return
|
if code in cls.__next_buy_time_dict:
|
cls.__next_buy_time_dict.pop(code)
|
|
if compute_end_index < compute_start_index:
|
return
|
|
unique_key = f"{code}-{compute_start_index}-{compute_end_index}"
|
if cls.__latest_process_not_order_unique_keys_count.get(
|
unique_key) and cls.__latest_process_not_order_unique_keys_count.get(unique_key) > 2:
|
async_log_util.error(logger_l2_error,
|
f"重复处理数据：code-{code} start_index-{compute_start_index} end_index-{compute_end_index}")
|
return
|
if unique_key not in cls.__latest_process_not_order_unique_keys_count:
|
cls.__latest_process_not_order_unique_keys_count[unique_key] = 0
|
cls.__latest_process_not_order_unique_keys_count[unique_key] += 1
|
|
_start_time = tool.get_now_timestamp()
|
total_datas = local_today_datas[code]
|
|
# ---------计算激进买入的信号---------
|
radical_result = cls.__compute_radical_order_begin_pos(code, compute_start_index, compute_end_index)
|
|
if radical_result[0]:
|
buy_single_index, buy_exec_index = radical_result[1], radical_result[1]
|
buy_volume_rate = cls.volume_rate_info[code][0]
|
refer_sell_data = cls.__L2MarketSellManager.get_refer_sell_data(code, total_datas[buy_single_index]["val"][
|
"time"])
|
if refer_sell_data:
|
sell_info = (refer_sell_data[0], refer_sell_data[1])
|
else:
|
sell_info = (total_datas[buy_single_index]["val"]["time"], 0)
|
threshold_money = 0
|
|
order_begin_pos_info = OrderBeginPosInfo(buy_single_index=buy_single_index,
|
buy_exec_index=buy_exec_index,
|
buy_compute_index=buy_exec_index,
|
num=total_datas[buy_single_index]["val"]["num"], count=1,
|
max_num_set=set(),
|
buy_volume_rate=buy_volume_rate,
|
mode=OrderBeginPosInfo.MODE_RADICAL,
|
mode_desc=f"大单不足扫入:{radical_result[2]}",
|
sell_info=sell_info,
|
threshold_money=threshold_money)
|
order_begin_pos_info.at_limit_up = cls.__is_at_limit_up_buy(code)
|
ordered = cls.__process_with_find_exec_index(code, order_begin_pos_info, compute_end_index,
|
block_info=radical_result[3])
|
if ordered:
|
radical_buy_data_manager.BlockPlaceOrderRecordManager().add_record(code, radical_result[2])
|
# 监听大单
|
RDCancelBigNumComputer().set_watch_indexes(code, radical_result[4])
|
return
|
else:
|
radical_result = cls.__compute_radical_order_begin_pos_for_many_sell(code, compute_start_index,
|
compute_end_index)
|
if radical_result[0]:
|
buy_single_index, buy_exec_index = radical_result[0][0], radical_result[0][1]
|
buy_volume_rate = cls.volume_rate_info[code][0]
|
refer_sell_data = cls.__L2MarketSellManager.get_refer_sell_data(code,
|
total_datas[buy_single_index]["val"][
|
"time"])
|
if refer_sell_data:
|
sell_info = (refer_sell_data[0], refer_sell_data[1])
|
else:
|
sell_info = (total_datas[buy_single_index]["val"]["time"], 0)
|
threshold_money = 0
|
|
order_begin_pos_info = OrderBeginPosInfo(buy_single_index=buy_single_index,
|
buy_exec_index=buy_exec_index,
|
buy_compute_index=buy_exec_index,
|
num=total_datas[buy_single_index]["val"]["num"], count=1,
|
max_num_set=set(),
|
buy_volume_rate=buy_volume_rate,
|
mode=OrderBeginPosInfo.MODE_RADICAL,
|
mode_desc=f"总抛压大扫入:{radical_result[2]}",
|
sell_info=sell_info,
|
threshold_money=threshold_money)
|
order_begin_pos_info.at_limit_up = cls.__is_at_limit_up_buy(code)
|
ordered = cls.__process_with_find_exec_index(code, order_begin_pos_info, compute_end_index,
|
block_info=radical_result[2])
|
if ordered:
|
radical_buy_data_manager.BlockPlaceOrderRecordManager().add_record(code, radical_result[2])
|
if RadicalBuyDealCodesManager().get_code_blocks(code):
|
# 已经扫入下过单
|
return
|
|
if not constant.CAN_COMMON_BUY:
|
# 常规买不买入了
|
return
|
|
# 获取买入信号计算起始位置
|
order_begin_pos = cls.__get_order_begin_pos(code)
|
|
# 是否为新获取到的位置
|
new_get_single = False
|
buy_single_index = order_begin_pos.buy_single_index
|
if buy_single_index is None:
|
# ------------------------------确定信号种类----------------------------------
|
# 第一步：获取积极下单信号
|
# if code.find('60') == 0:
|
# 积极买
|
continue_count = 1
|
has_single, _index, sell_info, single_msg, mode = cls.__compute_active_order_begin_pos(code, continue_count,
|
compute_start_index,
|
compute_end_index)
|
if not has_single:
|
# 没有信号，如果当前数据是涨停买就记录日志，防止记录过多的日志
|
if L2DataUtil.is_limit_up_price_buy(
|
total_datas[compute_start_index]["val"]) and L2DataUtil.is_limit_up_price_buy(
|
total_datas[compute_end_index]["val"]):
|
async_log_util.info(logger_l2_trade_buy_queue,
|
"尚未获取到买入信号： code-{} start_index-{} end_index-{} msg-{}", code,
|
compute_start_index, compute_end_index, sell_info)
|
fast_msg = None
|
if has_single:
|
order_begin_pos.mode = mode
|
order_begin_pos.mode_desc = f"委托触发： {single_msg}"
|
order_begin_pos.sell_info = sell_info
|
fast_msg = sell_info
|
# 用了信号就必须清除掉原有信号
|
place_order_single_data_manager.L2TradeSingleDataManager.clear_data(code)
|
|
if cls.__last_buy_single_dict.get(code) == _index:
|
has_single = None
|
_index = None
|
buy_single_index = _index
|
if has_single:
|
# 判断是否是板上买
|
order_begin_pos.at_limit_up = cls.__is_at_limit_up_buy(code)
|
# -----------------------------买入计算初始化,计算纯买额阈值-----------------------
|
cls.__last_buy_single_dict[code] = buy_single_index
|
new_get_single = True
|
order_begin_pos.num = 0
|
order_begin_pos.count = 0
|
order_begin_pos.buy_single_index = buy_single_index
|
order_begin_pos.threshold_money = int(sell_info[1] * 1.0)
|
order_begin_pos.max_num_set = set()
|
|
l2_log.debug(code, "获取到买入信号起始点：{} ,计算范围：{}-{} ，量比：{}，是否板上买：{}，数据：{} 模式：{}（{}） 信号类型：{}",
|
buy_single_index,
|
compute_start_index,
|
compute_end_index, cls.volume_rate_info[code], order_begin_pos.at_limit_up,
|
total_datas[buy_single_index],
|
order_begin_pos.mode, fast_msg, single_msg)
|
|
# _start_time = l2_data_log.l2_time(code, tool.get_now_timestamp() - _start_time, "下单信号计算时间")
|
|
if order_begin_pos.buy_single_index is None:
|
# 未获取到买入信号，终止程序
|
return None
|
|
# 开始计算的位置
|
start_process_index = max(order_begin_pos.buy_single_index, compute_start_index)
|
if new_get_single:
|
start_process_index = order_begin_pos.buy_single_index
|
|
# 设置为总卖额
|
new_buy_exec_index, buy_nums, buy_count, rebegin_buy_pos, threshold_money_new, max_num_set_new, not_buy_msg, clear_buy_single = cls.__sum_buy_num_for_order_active(
|
code,
|
start_process_index,
|
compute_end_index,
|
order_begin_pos.num,
|
order_begin_pos.count,
|
order_begin_pos.threshold_money,
|
order_begin_pos.buy_single_index, order_begin_pos.max_num_set, order_begin_pos.sell_info[1])
|
threshold_money = threshold_money_new
|
order_begin_pos.threshold_money = threshold_money
|
|
l2_log.debug(code, "范围：{}-{} m值-{} 量比:{} rebegin_buy_pos:{} clear_buy_single:{}", compute_start_index,
|
compute_end_index,
|
threshold_money, cls.volume_rate_info[code][0],
|
rebegin_buy_pos, clear_buy_single)
|
|
# 买入信号位与计算位置间隔2s及以上了
|
if rebegin_buy_pos is not None:
|
# 需要重新计算纯买额
|
cls.__start_compute_buy(code, rebegin_buy_pos, compute_end_index, threshold_money, capture_time,
|
is_first_code, False)
|
return
|
|
if new_buy_exec_index is not None:
|
cls.__process_with_find_exec_index(code, OrderBeginPosInfo(buy_single_index=buy_single_index,
|
buy_exec_index=new_buy_exec_index,
|
buy_compute_index=new_buy_exec_index,
|
num=buy_nums, count=buy_count,
|
max_num_set=max_num_set_new,
|
buy_volume_rate=cls.volume_rate_info[code][0],
|
mode=order_begin_pos.mode,
|
mode_desc=order_begin_pos.mode_desc,
|
sell_info=order_begin_pos.sell_info,
|
threshold_money=threshold_money),
|
compute_end_index, block_info=None)
|
else:
|
# 未达到下单条件，保存纯买额，设置纯买额
|
# 记录买入信号位置
|
if not clear_buy_single:
|
# 没有清除信号位置就保存下单位置信息
|
cls.__save_order_begin_data(code,
|
OrderBeginPosInfo(buy_single_index=buy_single_index, buy_exec_index=-1,
|
buy_compute_index=compute_end_index, num=buy_nums,
|
count=buy_count,
|
mode_desc=order_begin_pos.mode_desc,
|
max_num_set=max_num_set_new, mode=order_begin_pos.mode,
|
sell_info=order_begin_pos.sell_info,
|
threshold_money=threshold_money))
|
# 记录没下单原因
|
async_log_util.info(logger_l2_not_buy_reasons, f"{code}#{not_buy_msg}")
|
_start_time = t.time()
|
|
# 获取下单起始信号
|
@classmethod
|
def __get_order_begin_pos(cls, code) -> OrderBeginPosInfo:
|
order_begin_pos = cls.__TradePointManager.get_buy_compute_start_data_cache(
|
code)
|
return order_begin_pos
|
|
# 保存下单起始信号
|
@classmethod
|
def __save_order_begin_data(cls, code, info: OrderBeginPosInfo):
|
cls.__TradePointManager.set_buy_compute_start_data_v2(code, info)
|
|
@classmethod
|
def save_order_begin_data(cls, code, info: OrderBeginPosInfo):
|
cls.__save_order_begin_data(code, info)
|
|
# 计算下单起始信号
|
# compute_data_count 用于计算的l2数据数量
|
@classmethod
|
def __compute_order_begin_pos(cls, code, start_index, continue_count, end_index):
|
|
second_930 = 9 * 3600 + 30 * 60 + 0
|
# 倒数100条数据查询
|
datas = local_today_datas[code]
|
if end_index - start_index + 1 < continue_count:
|
return False, None
|
__time = None
|
|
last_index = None
|
count = 0
|
start = None
|
# now_time_s = tool.get_time_as_second(tool.get_now_time_str())
|
for i in range(start_index, end_index + 1):
|
_val = datas[i]["val"]
|
time_s = L2DataUtil.get_time_as_second(_val["time"])
|
# 时间要>=09:30:00
|
if time_s < second_930:
|
continue
|
|
# if not constant.TEST:
|
# if abs(now_time_s - time_s) > 2:
|
# # 正式环境下不处理2s外的数据
|
# continue
|
|
if L2DataUtil.is_limit_up_price_buy(_val):
|
# 金额要大于50万
|
if _val["num"] * float(_val["price"]) < 5000:
|
continue
|
# 寻找前面continue_count-1个涨停买
|
# for j in range(start_index - 1, -1, -1):
|
# if datas[j]["val"]
|
if last_index is None or (datas[last_index]["val"]["time"] == datas[i]["val"]["time"]):
|
if start is None:
|
start = i
|
last_index = i
|
count += datas[i]["re"]
|
if count >= continue_count:
|
return True, start
|
else:
|
# 本条数据作为起点
|
last_index = i
|
count = datas[i]["re"]
|
start = i
|
|
elif not L2DataUtil.is_sell(_val) and not L2DataUtil.is_sell_cancel(_val):
|
# 剔除卖与卖撤
|
last_index = None
|
count = 0
|
start = None
|
|
return False, None
|
|
# 快速买入法的信号位置查找
|
@classmethod
|
def __compute_fast_order_begin_pos(cls, code, start_index, end_index):
|
total_datas = local_today_datas[code]
|
start_time_str = total_datas[start_index]["val"]["time"]
|
refer_sell_data = cls.__L2MarketSellManager.get_refer_sell_data(code, start_time_str)
|
if refer_sell_data is None:
|
return False, -1, "总卖为空"
|
if cls.__L2MarketSellManager.is_refer_sell_time_used(code, refer_sell_data[0]):
|
return False, -1, "总卖统计时间已被使用"
|
# 是否大于2000万
|
if refer_sell_data[1] <= 2000 * 10000:
|
return False, -1, "总卖小于指定金额"
|
# 统计之前的卖
|
threshold_money = refer_sell_data[1]
|
for i in range(start_index - 1, -1, -1):
|
val = total_datas[i]["val"]
|
if tool.compare_time(val["time"], refer_sell_data[0]) <= 0:
|
break
|
if L2DataUtil.is_sell(val):
|
threshold_money += val["num"] * int(float(val["price"]) * 100)
|
elif L2DataUtil.is_sell_cancel(val):
|
threshold_money -= val["num"] * int(float(val["price"]) * 100)
|
# 是否为本秒的第一个涨停买
|
for i in range(start_index, end_index + 1):
|
data = total_datas[i]
|
val = data['val']
|
if not L2DataUtil.is_limit_up_price_buy(val):
|
# 要统计卖与卖撤
|
if L2DataUtil.is_sell(val):
|
threshold_money += val["num"] * int(float(val["price"]) * 100)
|
elif L2DataUtil.is_sell_cancel(val):
|
threshold_money -= val["num"] * int(float(val["price"]) * 100)
|
continue
|
# 50 万以下的不需要
|
if val["num"] * float(val["price"]) < 5000:
|
continue
|
# 是否为本s的第一次涨停
|
is_first_limit_up = True
|
for j in range(i - 1, -1, -1):
|
temp_val = total_datas[j]["val"]
|
if temp_val["time"] == val["time"]:
|
if L2DataUtil.is_limit_up_price_buy(temp_val) and temp_val["num"] * float(
|
temp_val["price"]) >= 5000:
|
is_first_limit_up = True
|
break
|
else:
|
break
|
if is_first_limit_up:
|
return True, i, [refer_sell_data[0], threshold_money]
|
return False, None, None
|
|
@classmethod
|
def __get_active_single_start_index(cls, code, start_index, end_index, continue_count, valid_single=False):
|
"""
|
获取主动买的信号起始索引
|
@param code:
|
@param start_index:
|
@param end_index:
|
@param continue_count:
|
@param valid_single: 是否验证成交买入信号
|
@return:
|
"""
|
total_datas = local_today_datas[code]
|
last_index = None
|
count = 0
|
start = None
|
for i in range(start_index, end_index + 1):
|
_val = total_datas[i]["val"]
|
if L2DataUtil.is_limit_up_price_buy(_val):
|
# 时间要>=09:30:00
|
if tool.trade_time_sub(_val["time"], "09:30:00") < 0:
|
continue
|
# 金额要大于50万
|
if _val["num"] * float(_val["price"]) < 5000:
|
continue
|
if last_index is None or (
|
total_datas[last_index]["val"]["time"] == total_datas[i]["val"]["time"]):
|
# 深证非板上放量需要判断是否有信号
|
if valid_single:
|
_single = place_order_single_data_manager.L2TradeSingleDataManager.get_valid_trade_single(
|
code,
|
tool.to_time_with_ms(
|
_val['time'],
|
_val['tms']))
|
if not _single:
|
continue
|
if start is None:
|
start = i
|
last_index = i
|
count += total_datas[i]["re"]
|
if count >= continue_count:
|
return start
|
else:
|
# 本条数据作为起点
|
last_index = i
|
count = total_datas[i]["re"]
|
start = i
|
elif not L2DataUtil.is_sell(_val) and not L2DataUtil.is_sell_cancel(_val):
|
# 剔除卖与卖撤
|
last_index = None
|
count = 0
|
start = None
|
return None
|
|
# 计算积极买的下单信号
|
@classmethod
|
def __compute_active_order_begin_pos(cls, code, continue_count, start_index, end_index):
|
"""
|
计算买入信号
|
@param code:
|
@param continue_count:
|
@param start_index:
|
@param end_index:
|
@return:
|
"""
|
total_datas = local_today_datas[code]
|
start_time_str = total_datas[start_index]["val"]["time"]
|
if end_index - start_index + 1 < continue_count:
|
return False, -1, "信号不连续", '', OrderBeginPosInfo.MODE_NORMAL
|
# 获取最近的总卖信息
|
# (time_str, round(money), volume, sell_1_info)
|
refer_sell_data = cls.__L2MarketSellManager.get_refer_sell_data(code, start_time_str)
|
active_buy_blocks = cls.get_active_buy_blocks(code)
|
if tool.is_sh_code(code):
|
if refer_sell_data is None:
|
# 设置默认卖信息
|
refer_sell_data = (start_time_str, 0, 0, (round(float(total_datas[start_index]["val"]["price"]), 2), 0))
|
# 上证:只要有一个涨停买的就是信号
|
for i in range(start_index, end_index + 1):
|
_val = total_datas[i]["val"]
|
if not L2DataUtil.is_limit_up_price_buy(_val):
|
continue
|
# 时间要>=09:30:00
|
if tool.trade_time_sub(_val["time"], "09:30:00") < 0:
|
continue
|
# 金额要大于50万
|
if _val["num"] * float(_val["price"]) < 5000:
|
continue
|
|
return True, i, [refer_sell_data[0],
|
0], '上证买入', OrderBeginPosInfo.MODE_ACTIVE if active_buy_blocks else OrderBeginPosInfo.MODE_FAST
|
return False, -1, "未获取到积极买的起始信号", '', OrderBeginPosInfo.MODE_NORMAL
|
else:
|
# 深证
|
if refer_sell_data is None:
|
return False, -1, "总卖为空", '', OrderBeginPosInfo.MODE_NORMAL
|
if refer_sell_data is None:
|
refer_sell_data = (start_time_str, 0, 0, (round(float(total_datas[start_index]["val"]["price"]), 2), 0))
|
l2_log.debug(code, f"丢失总卖额，设置默认为0，计算范围：{start_index}-{end_index}")
|
# 判断是否为真的板上放量
|
single = place_order_single_data_manager.L2TradeSingleDataManager.get_valid_trade_single(
|
code,
|
tool.to_time_with_ms(total_datas[end_index]['val']['time'], total_datas[end_index]['val']['tms']))
|
trade_price_info = HuaXinSellOrderStatisticManager.get_latest_trade_price_info(code)
|
limit_up_price = gpcode_manager.get_limit_up_price(code)
|
# p
|
is_limit_up = False
|
if limit_up_price and trade_price_info and abs(trade_price_info[0] - float(limit_up_price)) < 0.001:
|
is_limit_up = True
|
if refer_sell_data[1] > 0 or single or not is_limit_up:
|
# 不是板上放量
|
# 判断最近有没有涨停卖数据
|
limit_up_sell_count = L2TradeSingleDataProcessor.get_latest_limit_up_sell_order_count(code)
|
if (limit_up_sell_count == 0 or active_buy_blocks) and not single:
|
|
# 如果没有涨停卖数据/积极下单而且还没有成交买入信号，就按照原来的总卖额计算
|
threshold_money, sell_1_price = copy.deepcopy(refer_sell_data[1]), refer_sell_data[3][0]
|
for i in range(start_index - 1, -1, -1):
|
val = total_datas[i]["val"]
|
if tool.compare_time(val["time"], refer_sell_data[0]) < 0:
|
# 读取的L2的总卖额是不包含当前s的数据，所以需要将当前s的数据纳入计算
|
break
|
if L2DataUtil.is_sell(val):
|
threshold_money += val["num"] * int(float(val["price"]) * 100)
|
elif L2DataUtil.is_sell_cancel(val):
|
threshold_money -= val["num"] * int(float(val["price"]) * 100)
|
elif L2DataUtil.is_buy(val):
|
# 判断价格（大于卖1） 被买吃掉
|
if round(float(val["price"]), 2) - sell_1_price >= 0:
|
threshold_money -= val["num"] * int(float(val["price"]) * 100)
|
buy_single_index = cls.__get_active_single_start_index(code, start_index, end_index, continue_count,
|
valid_single=False)
|
if buy_single_index is not None:
|
return True, buy_single_index, [refer_sell_data[0],
|
threshold_money], f"上板无涨停卖：涨停卖数据-{limit_up_sell_count}", OrderBeginPosInfo.MODE_ACTIVE if active_buy_blocks else OrderBeginPosInfo.MODE_NORMAL
|
|
else:
|
# 按照成交买入信号计算
|
buy_single_index = cls.__get_active_single_start_index(code, start_index, end_index, continue_count,
|
valid_single=True)
|
if buy_single_index is not None:
|
return True, buy_single_index, [refer_sell_data[0],
|
0], "上板有成交买入信号", OrderBeginPosInfo.MODE_ACTIVE if active_buy_blocks else OrderBeginPosInfo.MODE_FAST
|
else:
|
# 板上放量：只需要一笔涨停买可作为信号
|
buy_single_index = cls.__get_active_single_start_index(code, start_index, end_index, continue_count,
|
valid_single=False)
|
if buy_single_index is not None:
|
return True, buy_single_index, [refer_sell_data[0],
|
0], "板上放量", OrderBeginPosInfo.MODE_ACTIVE if active_buy_blocks else OrderBeginPosInfo.MODE_NORMAL
|
|
return False, -1, "未获取到积极买的起始信号", '', OrderBeginPosInfo.MODE_NORMAL
|
|
# 计算激进买的下单信号
|
# 计算激进买的下单信号
|
@classmethod
|
def __compute_radical_order_begin_pos(cls, code, start_index, end_index):
|
"""
|
计算激进买入信号：
|
1.非板上放量
|
2.检测到299万大单买入
|
@param code:
|
@param start_index:
|
@param end_index:
|
@return: (是否获取到信号, 信号位置, 扫入板块/消息, 扫入板块大单流入信息, 需要监听的大单)
|
"""
|
|
# 激进买信号的时间
|
|
def __can_order():
|
# 判断是否是板上放量
|
if cls.__is_at_limit_up_buy(code, start_index):
|
return False, None, "板上放量", None
|
total_datas = local_today_datas[code]
|
limit_up_price = gpcode_manager.get_limit_up_price_as_num(code)
|
bigger_money = l2_data_util.get_big_money_val(limit_up_price, tool.is_ge_code(code))
|
min_num = int(bigger_money / limit_up_price / 100)
|
|
refer_sell_data = L2MarketSellManager().get_refer_sell_data(code, radical_data[3])
|
# 参考总卖额
|
refer_sell_money = 0
|
if refer_sell_data:
|
refer_sell_money = refer_sell_data[1]
|
big_order_deal_enough_result = radical_buy_data_manager.is_big_order_deal_enough(code,
|
code_volumn_manager.CodeVolumeManager().get_volume_rate_refer_in_5days(
|
code),
|
refer_sell_money,
|
for_buy=True,
|
is_almost_open_limit_up=
|
radical_data[5])
|
# 缺乏的大单金额
|
lack_money = big_order_deal_enough_result[3]
|
# 如果有大单成交就不需要看大单
|
if constant.CAN_RADICAL_BUY_NEED_BIG_ORDER_EVERYTIME:
|
# 每次下单都需要大单
|
current_big_order_deal_money_info = EveryLimitupBigDealOrderManager.get_big_buy_deal_order_money_info(
|
code)
|
if current_big_order_deal_money_info and tool.trade_time_sub(tool.get_now_time_str(),
|
current_big_order_deal_money_info[1]) > 60:
|
# 60s以上就不下单了
|
return False, None, "距离上次统计大单时间过去60s", set()
|
|
if lack_money == 0:
|
if not tool.is_sh_code(code):
|
# 非上证的票看50w
|
min_num = int(5000 / limit_up_price)
|
# 需要监听的大单
|
watch_indexes = set()
|
# 总委托大单金额
|
total_delegating_big_money = 0
|
single_index = None
|
# 从成交进度位开始看
|
trade_index, is_default = cls.__TradeBuyQueue.get_traded_index(code)
|
if trade_index is None:
|
trade_index = 0
|
canceled_buyno_map = local_today_canceled_buyno_map.get(code)
|
for i in range(trade_index, end_index + 1):
|
data = total_datas[i]
|
val = data["val"]
|
if not L2DataUtil.is_limit_up_price_buy(val):
|
continue
|
if val["num"] < min_num:
|
continue
|
# 撤单不算
|
left_count = l2_data_source_util.L2DataSourceUtils.get_limit_up_buy_no_canceled_count_v2(code, i,
|
total_datas,
|
canceled_buyno_map)
|
if left_count == 0:
|
continue
|
|
# 上证有可能是部分成交的大单
|
if i == start_index and tool.is_sh_code(code):
|
dealing_active_order_info = HuaXinBuyOrderManager().get_dealing_active_order_info(code)
|
if dealing_active_order_info and dealing_active_order_info[0] == int(val["orderNo"]):
|
# 判断是否为大单
|
order_money = dealing_active_order_info[2] + round(val["price"], 2) * val["num"] * 100
|
if order_money >= bigger_money:
|
lack_money -= order_money
|
watch_indexes.add(i)
|
if lack_money < 0:
|
single_index = i
|
break
|
|
if int(val["orderNo"]) <= radical_data[1]:
|
# 主动买单后的数据不算
|
continue
|
watch_indexes.add(i)
|
lack_money -= round(val["price"], 2) * val["num"] * 100
|
if lack_money < 0:
|
single_index = i
|
break
|
if single_index is not None:
|
return True, single_index, "有大单", watch_indexes
|
return False, None, f"大单不足：{trade_index}-{end_index} 缺少的大单-{lack_money}", watch_indexes
|
|
radical_data = RadicalBuyDealCodesManager.buy_by_l2_delegate_expire_time_dict.get(code)
|
record_codes = radical_buy_data_manager.BlockPlaceOrderRecordManager().get_codes()
|
# 是否扫入过
|
is_radical_buy = code in record_codes
|
|
if not radical_data:
|
return False, None, "不满足激进买的条件"
|
|
if constant.CAN_RADICAL_BUY_AT_LIMIT_UP:
|
# 板上放量可扫入
|
if t.time() > radical_data[0] and not is_radical_buy:
|
# 没扫入过才需要判断时间
|
return False, None, "超过生效时间"
|
else:
|
# 板上放量不可扫入
|
if t.time() > radical_data[0]:
|
return False, None, "超过生效时间"
|
|
result = __can_order()
|
l2_log.debug(code, f"L2扫入判断：{result}")
|
if result[0]:
|
# 已经扫入下过单且允许板上放量扫入的就需要判断板上放量的距离
|
if is_radical_buy and constant.CAN_RADICAL_BUY_AT_LIMIT_UP:
|
is_limit_up_buy = cls.__is_at_limit_up_buy(code)
|
if is_limit_up_buy:
|
# 判断成交进度到当前数据的笔数，如果少于10笔且还有未成交的大单（>=299）就可以下单
|
trade_index, is_default = cls.__TradeBuyQueue.get_traded_index(code)
|
if trade_index is None:
|
trade_index = 0
|
can_place_order, msg = buy_strategy_util.is_near_by_trade_index(code, trade_index)
|
if not can_place_order:
|
return False, result[1], "扫入过的代码板上放量距离远"
|
else:
|
# 判断该板块前排是否已经有成交
|
deal_codes = RadicalBuyDealCodesManager().get_deal_codes()
|
buy_blocks = radical_buy_data_manager.is_block_can_radical_buy(code, radical_data[2],
|
deal_codes)
|
if not buy_blocks:
|
return False, result[1], f"板块代码已有成交：{radical_data[2]}"
|
# 如果板上放量不可买入就需要删除信号
|
if not constant.CAN_RADICAL_BUY_AT_LIMIT_UP and code in RadicalBuyDealCodesManager.buy_by_l2_delegate_expire_time_dict:
|
RadicalBuyDealCodesManager.buy_by_l2_delegate_expire_time_dict.pop(code)
|
return True, result[1], radical_data[2], radical_data[4], result[3]
|
else:
|
async_log_util.info(logger_l2_not_buy_reasons, f"{code}#{result[2]}")
|
return result
|
|
# 总卖额参考时间使用记录
|
__refer_sell_used_times = {}
|
|
@classmethod
|
def __compute_radical_order_begin_pos_for_many_sell(cls, code, start_index, end_index):
|
"""
|
计算深证高抛压的卖的买入信号
|
@param code:
|
@param start_index:
|
@param end_index:
|
@return: 信号信息（信号位,执行位）, 消息, 可买入的板块
|
"""
|
if not tool.is_sz_code(code):
|
return None, "非深证的票", None
|
# 判断抛压是否大于5000w
|
total_datas = local_today_datas.get(code)
|
refer_sell_data = cls.__L2MarketSellManager.get_refer_sell_data(code, total_datas[-1]["val"]["time"])
|
if not refer_sell_data or refer_sell_data[1] < 5e7:
|
return None, "总卖额小于5000万", None
|
if code in cls.__refer_sell_used_times and refer_sell_data[0] in cls.__refer_sell_used_times:
|
return None, f"时间已经被使用：{refer_sell_data[0]}", None
|
|
deal_codes = RadicalBuyDealCodesManager().get_deal_codes()
|
if code in deal_codes:
|
return None, f"已经成交", None
|
|
f_buy_blocks, orgin_buy_blocks = radical_buy_strategy.compute_can_radical_buy_blocks(code, deal_codes)
|
if not f_buy_blocks:
|
return None, f"板块不可买入", None
|
|
# 判断是否有涨停买的数据
|
has_limit_up = False
|
for i in range(start_index, end_index + 1):
|
data = total_datas[i]
|
val = data["val"]
|
if L2DataUtil.is_limit_up_price_buy(val):
|
has_limit_up = True
|
break
|
if not has_limit_up:
|
return None, "无涨停买数据", None
|
# 查找计数起点
|
refer_sell_time_int = int(refer_sell_data[0].replace(":", ""))
|
begin_index = start_index
|
for i in range(start_index - 1, -1, -1):
|
data = total_datas[i]
|
val = data["val"]
|
if int(val["time"].replace(":", "")) < refer_sell_time_int:
|
begin_index = i + 1
|
break
|
threshold_num = int(round(refer_sell_data[1] / gpcode_manager.get_limit_up_price_as_num(code) / 100))
|
total_num = 0
|
for i in range(begin_index, end_index + 1):
|
data = total_datas[i]
|
val = data["val"]
|
if L2DataUtil.is_limit_up_price_buy(val):
|
total_num += val["num"]
|
elif L2DataUtil.is_limit_up_price_buy_cancel(val):
|
total_num -= val["num"]
|
if total_num > threshold_num:
|
if code not in cls.__refer_sell_used_times:
|
cls.__refer_sell_used_times[code] = set()
|
cls.__refer_sell_used_times[code].add(refer_sell_data[0])
|
return (begin_index, end_index), "可以下单", f_buy_blocks
|
return None, "总买额不满足", None
|
|
@classmethod
|
def test__compute_active_order_begin_pos(cls, code, continue_count, start_index, end_index):
|
return cls.__compute_active_order_begin_pos(code, continue_count, start_index, end_index)
|
|
@classmethod
|
def __get_threshmoney(cls, code):
|
m, msg = cls.__l2PlaceOrderParamsManagerDict[code].get_m_val()
|
if trade_manager.CodesTradeStateManager().get_trade_state_cache(code) == trade_constant.TRADE_STATE_NOT_TRADE:
|
# 首次下单m值扩大1.5倍
|
m = int(m * 1.5)
|
return m, msg
|
|
# 计算万手哥笔数
|
@classmethod
|
def __compute_big_money_count(cls, total_datas, start_index, end_index):
|
count = 0
|
for i in range(start_index, end_index + 1):
|
if L2DataUtil.is_limit_up_price_buy(total_datas[i]["val"]):
|
count += total_datas[i]["re"]
|
elif L2DataUtil.is_limit_up_price_buy_cancel(total_datas[i]["val"]):
|
count -= total_datas[i]["re"]
|
return count
|
|
# 将已经成交的过滤掉
|
|
@classmethod
|
def __filter_not_deal_indexes(cls, code, indexes):
|
trade_index, is_default = transaction_progress.TradeBuyQueue().get_traded_index(code)
|
if is_default:
|
return indexes
|
if trade_index is None:
|
return indexes
|
findexes = set()
|
for index in indexes:
|
if index < trade_index:
|
continue
|
findexes.add(index)
|
return findexes
|
|
# 返回(买入执行点, 总手, 总笔数, 从新计算起点, 纯买额阈值)
|
# 计算快速买入
|
@classmethod
|
def __sum_buy_num_for_order_fast(cls, code, compute_start_index, compute_end_index, origin_num, origin_count,
|
threshold_money_origin, buy_single_index, max_num_set):
|
_start_time = t.time()
|
total_datas = local_today_datas[code]
|
# is_first_code = gpcode_manager.FirstCodeManager().is_in_first_record_cache(code)
|
|
buy_nums = origin_num
|
buy_count = origin_count
|
limit_up_price = gpcode_manager.get_limit_up_price(code)
|
if limit_up_price is None:
|
raise Exception("涨停价无法获取")
|
limit_up_price = float(limit_up_price)
|
|
threshold_money = threshold_money_origin
|
# 目标手数
|
threshold_num = round(threshold_money / (limit_up_price * 100))
|
|
# buy_single_time_seconds = L2DataUtil.get_time_as_second(total_datas[buy_single_index]["val"]["time"])
|
|
# 可以触发买，当有涨停买信号时才会触发买
|
trigger_buy = True
|
# 间隔最大时间为3s
|
max_space_time_ms = 3 * 1000
|
# 不下单的信息
|
not_buy_msg = ""
|
max_buy_num_set = set(max_num_set)
|
place_order_count = trade_data_manager.PlaceOrderCountManager().get_place_order_count(code)
|
if place_order_count is None:
|
place_order_count = 0
|
is_ge_code = tool.is_ge_code(code)
|
buy_no_map = local_today_buyno_map.get(code)
|
for i in range(compute_start_index, compute_end_index + 1):
|
data = total_datas[i]
|
_val = total_datas[i]["val"]
|
trigger_buy = False
|
# 必须为连续2秒内的数据
|
if L2DataUtil.time_sub_as_ms(_val, total_datas[buy_single_index]["val"]) > max_space_time_ms:
|
cls.__TradePointManager.delete_buy_point(code)
|
if i == compute_end_index:
|
# 数据处理完毕
|
return None, buy_nums, buy_count, None, threshold_money, max_buy_num_set, f"【{i}】信号不连续，囊括时间-{max_space_time_ms}ms"
|
else:
|
# 计算买入信号，不能同一时间开始计算
|
for ii in range(buy_single_index + 1, compute_end_index + 1):
|
if total_datas[buy_single_index]["val"]["time"] != total_datas[ii]["val"]["time"]:
|
return None, buy_nums, buy_count, ii, threshold_money, max_buy_num_set, f"【{i}】信号不连续，囊括时间-{max_space_time_ms}ms"
|
if L2DataUtil.is_sell(_val):
|
threshold_money += _val["num"] * int(float(_val["price"]) * 100)
|
threshold_num = round(threshold_money / (limit_up_price * 100))
|
elif L2DataUtil.is_sell_cancel(_val):
|
threshold_money -= _val["num"] * int(float(_val["price"]) * 100)
|
threshold_num = round(threshold_money / (limit_up_price * 100))
|
# 涨停买
|
elif L2DataUtil.is_limit_up_price_buy(_val):
|
if l2_data_util.is_big_money(_val, is_ge_code):
|
max_buy_num_set.add(i)
|
trigger_buy = True
|
# 只统计59万以上的金额
|
buy_nums += int(_val["num"]) * int(total_datas[i]["re"])
|
buy_count += int(total_datas[i]["re"])
|
if buy_nums >= threshold_num:
|
l2_log.info(code, logger_l2_trade_buy,
|
f"{code}获取到买入执行点(快速买入)：{i} 统计纯买手数：{buy_nums} 目标纯买手数：{threshold_num} 统计纯买单数：{buy_count}")
|
elif L2DataUtil.is_limit_up_price_buy_cancel(_val):
|
# 判断买入位置是否在买入信号之前
|
buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data_v2(total_datas[i], buy_no_map)
|
if buy_index is not None:
|
# 找到买撤数据的买入点
|
if buy_index >= buy_single_index:
|
max_buy_num_set.discard(buy_index)
|
buy_nums -= int(_val["num"]) * int(data["re"])
|
buy_count -= int(data["re"])
|
l2_log.buy_debug(code, "{}数据在买入信号之后 撤买纯买手数：{} 目标手数：{}", i, buy_nums, threshold_num)
|
else:
|
l2_log.buy_debug(code, "{}数据在买入信号之前，买入位：{}", i, buy_index)
|
if total_datas[buy_single_index]["val"]["time"] == total_datas[buy_index]["val"]["time"]:
|
# 同一秒,当作买入信号之后处理
|
buy_nums -= int(_val["num"]) * int(data["re"])
|
buy_count -= int(data["re"])
|
max_buy_num_set.discard(buy_index)
|
# 大单撤销
|
l2_log.buy_debug(code, "{}数据买入位与预估买入位在同一秒", i)
|
else:
|
# 未找到买撤数据的买入点
|
l2_log.buy_debug(code, "未找到买撤数据的买入点: 位置-{} 数据-{}", i, data)
|
buy_nums -= int(_val["num"]) * int(total_datas[i]["re"])
|
buy_count -= int(total_datas[i]["re"])
|
l2_log.buy_debug(code, "位置-{}，总手数：{}，目标手数：{}", i,
|
buy_nums, threshold_num)
|
# 纯买额足够，且笔数大于5笔
|
if buy_nums < threshold_num:
|
not_buy_msg = f"【{i}】纯买额不够，{buy_nums}/{threshold_num}"
|
continue
|
if not trigger_buy:
|
not_buy_msg = f"【{i}】没有买单触发"
|
continue
|
if buy_count < 5:
|
not_buy_msg = f"【{i}】安全笔数不足，{buy_count}/{5}"
|
continue
|
|
max_buy_num_set = cls.__filter_not_deal_indexes(code, max_buy_num_set)
|
big_money_count_threshhold = cls.__l2PlaceOrderParamsManagerDict[code].get_big_num_count()
|
if len(max_buy_num_set) < big_money_count_threshhold:
|
not_buy_msg = f"【{i}】大单不满足要求，需要：{big_money_count_threshhold} 总：{len(max_buy_num_set)}"
|
continue
|
|
try:
|
info = cls.__trade_log_placr_order_info_dict[code]
|
info.set_trade_factor(threshold_money, 0, [])
|
except Exception as e:
|
async_log_util.error(logger_l2_error, f"记录交易因子出错：{str(e)}")
|
|
return i, buy_nums, buy_count, None, threshold_money, max_buy_num_set, "可以下单"
|
|
l2_log.buy_debug(code, "尚未获取到买入执行点(快速买入)，起始计算位置：{} 统计纯买手数：{} 目标纯买手数：{} 统计纯买单数：{}",
|
compute_start_index,
|
buy_nums,
|
threshold_num, buy_count)
|
|
return None, buy_nums, buy_count, None, threshold_money, max_buy_num_set, not_buy_msg
|
|
# 返回(买入执行点, 总手, 总笔数, 从新计算起点, 纯买额阈值)
|
# 计算激进买入
|
@classmethod
|
def __sum_buy_num_for_order_active(cls, code, compute_start_index, compute_end_index, origin_num, origin_count,
|
threshold_money_origin, buy_single_index, max_num_set, total_sell_money):
|
_start_time = t.time()
|
total_datas = local_today_datas[code]
|
# is_first_code = gpcode_manager.FirstCodeManager().is_in_first_record_cache(code)
|
|
buy_nums = origin_num
|
buy_count = origin_count
|
limit_up_price = gpcode_manager.get_limit_up_price(code)
|
if limit_up_price is None:
|
raise Exception("涨停价无法获取")
|
limit_up_price = float(limit_up_price)
|
|
is_at_limit_up = False
|
current_sell_data = cls.__L2MarketSellManager.get_current_total_sell_data(code)
|
if current_sell_data and current_sell_data[1] == 0:
|
# 板上放量买
|
is_at_limit_up = True
|
|
threshold_money = threshold_money_origin
|
# 目标手数
|
threshold_num = round(threshold_money / (limit_up_price * 100))
|
bigger_threshold_num = round(5000 / (limit_up_price))
|
|
# buy_single_time_seconds = L2DataUtil.get_time_as_second(total_datas[buy_single_index]["val"]["time"])
|
|
# 可以触发买，当有涨停买信号时才会触发买
|
trigger_buy = True
|
# 间隔最大时间为3s
|
max_space_time_ms = 3 * 1000
|
if tool.is_sz_code(code) and not is_at_limit_up:
|
# 深证非板上放量
|
max_space_time_ms = 1 * 1000
|
|
# 上证的间隔时间为1s
|
if tool.is_sh_code(code):
|
max_space_time_ms = 1 * 1000
|
|
# 不下单的信息
|
not_buy_msg = ""
|
max_buy_num_set = set(max_num_set)
|
active_buy_blocks = cls.get_active_buy_blocks(code)
|
is_ge_code = tool.is_ge_code(code)
|
buyno_map = local_today_buyno_map.get(code)
|
for i in range(compute_start_index, compute_end_index + 1):
|
data = total_datas[i]
|
_val = total_datas[i]["val"]
|
trigger_buy = False
|
# 必须为连续2秒内的数据
|
if L2DataUtil.time_sub_as_ms(_val, total_datas[buy_single_index]["val"]) > max_space_time_ms:
|
# 清除买点数据
|
cls.__TradePointManager.delete_buy_point(code)
|
if i == compute_end_index:
|
# 数据处理完毕
|
return None, buy_nums, buy_count, None, threshold_money, max_buy_num_set, f"【{i}】信号不连续，囊括时间-{max_space_time_ms}ms", True
|
else:
|
# 计算买入信号，不能同一时间开始计算
|
for ii in range(buy_single_index + 1, compute_end_index + 1):
|
if total_datas[buy_single_index]["val"]["time"] != total_datas[ii]["val"]["time"]:
|
return None, buy_nums, buy_count, ii, threshold_money, max_buy_num_set, f"【{i}】信号不连续，囊括时间-{max_space_time_ms}ms", True
|
# if L2DataUtil.is_sell(_val):
|
# threshold_money += _val["num"] * int(float(_val["price"]) * 100)
|
# threshold_num = round(threshold_money / (limit_up_price * 100))
|
# elif L2DataUtil.is_sell_cancel(_val):
|
# threshold_money -= _val["num"] * int(float(_val["price"]) * 100)
|
# threshold_num = round(threshold_money / (limit_up_price * 100))
|
# 涨停买
|
elif L2DataUtil.is_limit_up_price_buy(_val):
|
if _val['num'] < bigger_threshold_num:
|
continue
|
if l2_data_util.is_big_money(_val, is_ge_code):
|
max_buy_num_set.add(i)
|
trigger_buy = True
|
# 只统计59万以上的金额
|
buy_nums += int(_val["num"]) * int(total_datas[i]["re"])
|
buy_count += int(total_datas[i]["re"])
|
if buy_nums >= threshold_num:
|
l2_log.info(code, logger_l2_trade_buy,
|
f"{code}获取到买入执行点(积极下单)：{i} 统计纯买手数：{buy_nums} 目标纯买手数：{threshold_num} 统计纯买单数：{buy_count}")
|
elif L2DataUtil.is_limit_up_price_buy_cancel(_val):
|
# 判断买入位置是否在买入信号之前
|
buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data_v2(total_datas[i], buyno_map)
|
if buy_index is not None:
|
|
# 找到买撤数据的买入点
|
if buy_index >= buy_single_index:
|
max_buy_num_set.discard(buy_index)
|
buy_nums -= int(_val["num"]) * int(data["re"])
|
buy_count -= int(data["re"])
|
l2_log.buy_debug(code, "{}数据在买入信号之后 撤买纯买手数：{} 目标手数：{}", i, buy_nums, threshold_num)
|
else:
|
l2_log.buy_debug(code, "{}数据在买入信号之前，买入位：{}", i, buy_index)
|
if total_datas[buy_single_index]["val"]["time"] == total_datas[buy_index]["val"]["time"]:
|
# 同一秒,当作买入信号之后处理
|
buy_nums -= int(_val["num"]) * int(data["re"])
|
buy_count -= int(data["re"])
|
max_buy_num_set.discard(buy_index)
|
# 大单撤销
|
l2_log.buy_debug(code, "{}数据买入位与预估买入位在同一秒", i)
|
else:
|
# 未找到买撤数据的买入点
|
l2_log.buy_debug(code, "未找到买撤数据的买入点: 位置-{} 数据-{}", i, data)
|
buy_nums -= int(_val["num"]) * int(total_datas[i]["re"])
|
buy_count -= int(total_datas[i]["re"])
|
l2_log.buy_debug(code, "位置-{}，总手数：{}，目标手数：{}", i,
|
buy_nums, threshold_num)
|
|
# 计算信号位置之后的主动卖，加入到阈值之中
|
sell_orders = HuaXinSellOrderStatisticManager.get_latest_transaction_datas(code, min_sell_order_no=int(
|
total_datas[buy_single_index]['val']['orderNo']), min_deal_time=tool.trade_time_add_second(
|
total_datas[buy_single_index]['val']['time'], -1))
|
sell_order_num = sum([x[1] for x in sell_orders]) // 100
|
# 纯买额足够，且笔数大于2笔
|
if buy_nums < threshold_num + sell_order_num:
|
l2_log.buy_debug(code, f"激进买主动卖手数：{sell_order_num}")
|
not_buy_msg = f"【{i}】纯买额不够，{buy_nums}/{threshold_num}"
|
continue
|
if not trigger_buy:
|
not_buy_msg = f"【{i}】没有买单触发"
|
continue
|
|
safe_count = 1
|
if active_buy_blocks and current_sell_data[1] > 5000000:
|
# 有激进板块且大于500w才能扫入下单
|
safe_count = 1
|
else:
|
# 无激进下单板块
|
if total_sell_money <= 0:
|
# 所有的板上放量都需要3笔
|
# 板上下单需要安全笔数3笔
|
safe_count = 3
|
else:
|
if tool.is_sz_code(code):
|
# 深证上板
|
safe_count = 2
|
elif tool.is_sh_code(code):
|
# 上证安全笔数为3
|
safe_count = 3
|
if tool.is_sz_code(code):
|
money_y = code_volumn_manager.CodeVolumeManager().get_reference_volume_as_money_y(code)
|
# 大于8亿的安全笔数必须有8笔
|
if money_y >= 8:
|
safe_count = 8
|
# 9:31之前下单，安全笔数最小为5笔
|
if int(tool.get_now_time_str().replace(":", "")) < int("093100"):
|
safe_count = max(safe_count, 5)
|
# 深证的票，第一次下单必须要有3笔安全笔数
|
if tool.is_sz_code(code):
|
place_order_count = trade_data_manager.PlaceOrderCountManager().get_place_order_count(code)
|
if place_order_count == 0:
|
safe_count = max(safe_count, 3)
|
|
if buy_count < safe_count:
|
not_buy_msg = f"【{i}】安全笔数不足，{buy_count}/{safe_count}"
|
continue
|
|
# max_buy_num_set = cls.__filter_not_deal_indexes(code, max_buy_num_set)
|
# big_money_count_threshhold = cls.__l2PlaceOrderParamsManagerDict[code].get_big_num_count()
|
# if len(max_buy_num_set) < big_money_count_threshhold:
|
# not_buy_msg = f"【{i}】大单不满足要求，需要：{big_money_count_threshhold} 总：{len(max_buy_num_set)}"
|
# continue
|
|
try:
|
info = cls.__trade_log_placr_order_info_dict[code]
|
info.set_trade_factor(threshold_money, 0, [])
|
except Exception as e:
|
async_log_util.error(logger_l2_error, f"记录交易因子出错：{str(e)}")
|
l2_log.buy_debug(code, f"激进买主动卖手数：{sell_order_num}")
|
|
return i, buy_nums, buy_count, None, threshold_money, max_buy_num_set, "可以下单", False
|
|
l2_log.buy_debug(code, "尚未获取到买入执行点(激进买入)，起始计算位置：{} 统计纯买手数：{} 目标纯买手数：{} 统计纯买单数：{} ",
|
compute_start_index,
|
buy_nums,
|
threshold_num, buy_count)
|
|
return None, buy_nums, buy_count, None, threshold_money, max_buy_num_set, not_buy_msg, False
|
|
|
def test_trade_record():
|
code = "000333"
|
__trade_log_placr_order_info_dict = {code: trade_record_log_util.PlaceOrderInfo()}
|
try:
|
jx_blocks, jx_blocks_by = KPLCodeJXBlockManager().get_jx_blocks_cache(
|
code), KPLCodeJXBlockManager().get_jx_blocks_cache(code, by=True)
|
info = __trade_log_placr_order_info_dict[code]
|
info.set_buy_index(0, 1)
|
if jx_blocks:
|
info.set_kpl_blocks(list(jx_blocks))
|
elif jx_blocks_by:
|
info.set_kpl_blocks(list(jx_blocks_by))
|
else:
|
info.set_kpl_blocks([])
|
|
trade_record_log_util.add_place_order_log(code, info)
|
except:
|
pass
|
|
|
if __name__ == "__main__":
|
# test_trade_record()
|
# yesterday_limit_up_data_records = kpl_data_manager.get_current_limit_up_data_records(1)[0][1]
|
# yesterday_codes = set([x[0] for x in yesterday_limit_up_data_records])
|
# print(yesterday_codes)
|
code = "603003"
|
datas = log_export.load_l2_from_log()
|
datas = datas.get(code)
|
if datas is None:
|
datas = []
|
l2.l2_data_util.local_today_datas[code] = datas[:191]
|
l2.l2_data_util.load_buy_no_map(l2.l2_data_util.local_today_buyno_map, code,
|
l2.l2_data_util.local_today_datas[code])
|
l2.l2_data_util.load_canceled_buy_no_map(l2.l2_data_util.local_today_canceled_buyno_map, code,
|
l2.l2_data_util.local_today_datas[code])
|
start_index = 73
|
end_index = 190
|
LCancelBigNumComputer().compute_watch_index(code, start_index, end_index)
|
