""" 自定义板块流入金额 """ import copy import itertools import os import constant import l2_data_util from db import mysql_data_delegate as mysql_data from huaxin_client import l1_subscript_codes_manager from third_data.kpl_data_constant import LimitUpCodesBlockRecordManager from third_data.third_blocks_manager import BlockMapManager from utils import tool, global_util @tool.singleton class CodeInMoneyManager: def __init__(self): # 总的净流入 self.__code_money_dict = {} # 总买单信息：{"code":[金额, 数量]} self.__code_buy_money_dict = {} # 总卖单信息：{"code":[金额, 数量]} self.__code_sell_money_dict = {} # 净流入大单买金额 self.__code_big_buy_money_list_dict = {} # 净流出大单卖金额 self.__code_big_sell_money_list_dict = {} self.__latest_price = {} self.__load_data() def __load_data(self): _path = f"{constant.get_path_prefix()}/logs/huaxin_local/l2/transaction_big_order.{tool.get_now_date_str()}.log" if os.path.exists(_path): with open(_path) as f: lines = f.readlines() for line in lines: line = line.split(" - ")[1] item = eval(line) self.add_data(item) def add_data(self, item): """ 添加数据 @param item: (代码,类型, 订单数据) 订单数据：(订单号, 量, 金额, 时间, 最新成交价格) @return: """ code = item[0] if code not in self.__code_money_dict: self.__code_money_dict[code] = 0 if code not in self.__code_buy_money_dict: self.__code_buy_money_dict[code] = [0, 0] if code not in self.__code_sell_money_dict: self.__code_sell_money_dict[code] = [0, 0] if not tool.is_ge_code(code): big_money = l2_data_util.get_big_money_val(item[2][4]) if item[2][2] < big_money: # 不是常规定义的大单就返回 return if tool.is_ge_code(code) and item[2][2] < 299e4 and item[2][1] < 290000: return if item[1] == 0: # item[2]的数据结构： (买单号, 量, 金额, 时间, 最新成交价格) self.__code_money_dict[code] += item[2][2] self.__code_buy_money_dict[code][0] += item[2][2] self.__code_buy_money_dict[code][1] += 1 if code not in self.__code_big_buy_money_list_dict: self.__code_big_buy_money_list_dict[code] = [] # 大买单信息：(金额, 最新价格, 订单号) if len(item[2]) >= 5: self.__code_big_buy_money_list_dict[code].append((item[2][2], item[2][4], item[2][0])) else: self.__code_money_dict[code] -= item[2][2] self.__code_sell_money_dict[code][0] += item[2][2] self.__code_sell_money_dict[code][1] += 1 # 大卖单信息 if code not in self.__code_big_sell_money_list_dict: self.__code_big_sell_money_list_dict[code] = [] if len(item[2]) >= 5: # 大卖单信息：(金额, 最新价格, 订单号) self.__code_big_sell_money_list_dict[code].append((item[2][2], item[2][4], item[2][0])) self.__latest_price[code] = item[2][4] def get_code_money_dict(self): return self.__code_money_dict def get_money(self, code): if code in self.__code_money_dict: return self.__code_money_dict.get(code) return 0 def get_money_info(self, code): """ 获取代码流入信息 @param code: 代码信息 @return: 净流入金额,[大单买金额, 大单买数量],[大单卖金额,大单卖数量] """ return self.__code_money_dict.get(code), self.__code_buy_money_dict.get(code), self.__code_sell_money_dict.get( code) def set_money(self, code, money): self.__code_money_dict[code] = money def get_big_buy_money_list(self, code): """ 获取代码的大买单列表 @param code: @return:[(金额, 价格, 订单号)] """ return self.__code_big_buy_money_list_dict.get(code) def get_big_sell_money_list(self, code): """ 获取代码的大买单列表 @param code: @return:[(金额, 价格, 订单号)] """ return self.__code_big_sell_money_list_dict.get(code) def get_latest_price(self, code): return self.__latest_price.get(code) @tool.singleton class BlockInMoneyRankManager: """ 板块流入流出管理 """ __mysql_db = mysql_data.Mysqldb() __code_blocks = {} __in_list = [] __out_list = [] # 最近这段时间的代码涨停次数 __history_code_limit_up_count = {} # 被排除的代码 __exclude_codes = set() def __load_codes(self): codes = [] codes_sh, codes_sz = l1_subscript_codes_manager.get_codes(False) for b in codes_sh: codes.append(b.decode()) for b in codes_sz: codes.append(b.decode()) return codes def __load_blocks(self): if self.codes: for code in self.codes: before_fblocks = LimitUpCodesBlockRecordManager().get_radical_buy_blocks(code) if not before_fblocks: before_fblocks = set() fblocks = BlockMapManager().filter_blocks(before_fblocks) if fblocks: fblocks -= constant.KPL_INVALID_BLOCKS self.__code_blocks[code] = fblocks def __load_exclude_codes(self): """ 获取之前4个交易日的数据 @return: """ max_day = tool.get_now_date_str() min_day = tool.date_sub(max_day, 30) sql = f"select * from (select distinct(r.`_day`) as 'day' from `kpl_limit_up_record` r where r.`_day`<'{max_day}' and r.`_day`>'{min_day}') a order by a.day desc limit 4" results = self.__mysql_db.select_all(sql) dates = [r[0] for r in results] # 获取之前4天涨停次数>=2次的代码 day_codes = {} for day in dates: sql = f"select distinct(r._code) from kpl_limit_up_record r where r.`_day`='{day}'" results = self.__mysql_db.select_all(sql) day_codes[day] = set([x[0] for x in results]) codes_list = [day_codes[k] for k in day_codes] # 统计代码的涨停天数 code_limit_up_count_dict = {} for codes in codes_list: for c in codes: if c not in code_limit_up_count_dict: code_limit_up_count_dict[c] = 0 code_limit_up_count_dict[c] += 1 self.__history_code_limit_up_count = code_limit_up_count_dict self.load_today_limit_up_codes() def load_today_limit_up_codes(self): # 加载今日涨停代码 day = tool.get_now_date_str() sql = f"select distinct(r._code) from kpl_limit_up_record r where r.`_day`='{day}'" results = self.__mysql_db.select_all(sql) codes = set([x[0] for x in results]) # 计算需要排除的代码 temp_limit_up_count = {} # 统计总共涨停天数 for c in self.__history_code_limit_up_count: if c in codes: temp_limit_up_count[c] = self.__history_code_limit_up_count[c] + 1 else: temp_limit_up_count[c] = self.__history_code_limit_up_count[c] exclude_codes = set() for c in temp_limit_up_count: if temp_limit_up_count[c] < 3: continue exclude_codes.add(c) self.__exclude_codes = exclude_codes def __init__(self): self.codes = self.__load_codes() self.__load_blocks() self.__load_exclude_codes() print("排除的代码", self.__exclude_codes) def get_codes(self): return self.codes def compute(self): codes = self.codes block_money = {} for code in codes: if code in self.__exclude_codes: continue money = CodeInMoneyManager().get_money(code) if money is None: continue # 大自由流通市值的流出不算 if money < 0: price = CodeInMoneyManager().get_latest_price(code) zylt_volume = global_util.zylt_volume_map.get(code) if price and zylt_volume and zylt_volume * price > 200e8: continue before_fblocks = LimitUpCodesBlockRecordManager().get_radical_buy_blocks(code) if not before_fblocks: before_fblocks = set() fblocks = BlockMapManager().filter_blocks(before_fblocks) if fblocks: fblocks -= constant.KPL_INVALID_BLOCKS for b in fblocks: if b not in block_money: block_money[b] = 0 block_money[b] += money temp_list = [(x, block_money[x]) for x in block_money] temp_list.sort(key=lambda x: x[1], reverse=True) self.__in_list = temp_list temp_list = copy.deepcopy(temp_list) temp_list.sort(key=lambda x: x[1]) self.__out_list = temp_list def get_block_codes_money(self, block): """ 获取板块中代码的流入流出 @param block: @return:(代码, 金额, 是否被排除) """ codes = self.codes fdatas = [] for code in codes: if code in self.__exclude_codes: continue money = CodeInMoneyManager().get_money(code) if money is None: continue # 大自由流通市值的流出不算 if money < 0: price = CodeInMoneyManager().get_latest_price(code) zylt_volume = global_util.zylt_volume_map.get(code) if price and zylt_volume and zylt_volume * price > 200e8: continue before_fblocks = LimitUpCodesBlockRecordManager().get_radical_buy_blocks(code) if not before_fblocks: before_fblocks = set() fblocks = BlockMapManager().filter_blocks(before_fblocks) if block not in fblocks: continue fdatas.append((code, money, code in self.__exclude_codes)) return fdatas def get_in_list(self): return self.__in_list def get_out_list(self): return self.__out_list if __name__ == '__main__': code = "600839" before_fblocks = LimitUpCodesBlockRecordManager().get_radical_buy_blocks(code) print(before_fblocks) # print(CodeInMoneyManager().get_money("300264")) # BlockInMoneyRankManager().compute() # print(BlockInMoneyRankManager().get_in_list()[:20]) # print(BlockInMoneyRankManager().get_out_list()[:20])