""" L2数据交易影响因子 """ # l2交易因子 import functools import big_money_num_manager import global_data_loader import global_util import limit_up_time_manager import trade_data_manager class L2TradeFactorUtil: # 获取基础m值，返回单位为元 @classmethod def get_base_safe_val(cls, zyltgb): yi = round(zyltgb / 100000000) if yi < 1: yi = 1 m = 5000000 + (yi - 1) * 500000 return round(m * 0.7) # 获取行业影响比例 # total_limit_percent为统计的比例之和乘以100 @classmethod def get_industry_rate(cls, total_limit_percent): if total_limit_percent is None: return 0 t = total_limit_percent / 10 if t < 0.9: return 0 elif t <= 1.1: return 0.7 elif t <= 1.6: return 0.4 else: rate = 0 for i in range(0, 30): if t <= 2.1 + 0.5 * i: rate = 0.18 + 0.08 * i break if rate > 0.9: rate = 0.9 return round(rate, 4) # 获取量影响比例 @classmethod def get_volumn_rate(cls, day60_max, yest, today): if day60_max is None: return 0 if yest is None: return 0 if today is None: return 0 old_volumn = int(yest) if int(day60_max) > int(yest): old_volumn = int(day60_max) r = round(int(today) / old_volumn, 2) if r < 0.01: r = 0.01 print("比例：", r) rate = 0 if r <= 0.5: rate = 0.6 - (r - 0.01) * 2 elif r <= 0.85: rate = -0.38 + (r - 0.5) * 2.8 elif r <= 1.15: rate = 0.6 - (r - 0.85) * 4 else: rate = -0.6 return round(rate, 4) @classmethod def get_volumn_rate_by_code(cls, code): volumn_day60_max, volumn_yest, volumn_today = cls.__get_volumns(code) rate = cls.get_volumn_rate(volumn_day60_max, volumn_yest, volumn_today) return rate # 当前股票首次涨停时间的影响比例 @classmethod def get_limit_up_time_rate(cls, time_str): times = time_str.split(":") start_m = 9 * 60 + 30 m = int(times[0]) * 60 + int(times[1]) dif = m - start_m base_rate = 0.3 rate = 0 if dif < 1: rate = base_rate elif dif <= 120: # 11:30之前 rate = base_rate - dif * 0.0035 else: rate = base_rate - (dif - 89) * 0.0035 if rate < -0.3020: rate = -0.3020 return round(rate, 4) # 纯万手哥影响值（手数》=9000 OR 金额》=500w） @classmethod def get_big_money_rate(cls, num): if num < 4: return 0 rate = (num - 4) * 0.035 / 4 + 0.06 if rate > 0.9: rate = 0.9 return round(rate, 4) @classmethod def compute_rate(cls, zyltgb, total_industry_limit_percent, volumn_day60_max, volumn_yest, volumn_today, limit_up_time, big_money_num): # 行业涨停影响比例 industry_rate = 0 if total_industry_limit_percent is not None: industry_rate = cls.get_industry_rate(total_industry_limit_percent) # 量影响比例 volumn_rate = 0 if volumn_day60_max is not None and volumn_yest is not None and volumn_today is not None: volumn_rate = cls.get_volumn_rate(int(volumn_day60_max), int(volumn_yest), int(volumn_today)) # 涨停时间影响比例 limit_up_time_rate = 0 if limit_up_time is not None: limit_up_time_rate = cls.get_limit_up_time_rate(limit_up_time) # 万手哥影响 big_money_rate = 0 if big_money_num is not None: big_money_rate = cls.get_big_money_rate(big_money_num) msg = "zyltgb：{} industry_rate：{} volumn_rate：{} limit_up_time_rate：{} big_money_rate：{}".format(zyltgb, industry_rate, volumn_rate, limit_up_time_rate, big_money_rate) final_rate = round(1 - (industry_rate + volumn_rate + limit_up_time_rate + big_money_rate), 4) if final_rate < 0.1: final_rate = 0.1 return final_rate, msg @classmethod def compute_rate_by_code(cls, code): factors = cls.__get_rate_factors(code) return cls.compute_rate(factors[0], factors[1], factors[2], factors[3], factors[4], factors[5], factors[6]) # 获取代码当前所在的行业热度 @classmethod def __get_industry_limit_percent(cls, code): # 获取行业热度 industry = global_util.code_industry_map.get(code) if industry is None: global_data_loader.load_industry() industry = global_util.code_industry_map.get(code) total_industry_limit_percent = global_util.industry_hot_num.get(industry) if industry is not None else None # 当前票是否涨停 # if total_industry_limit_percent is not None: # if code in global_util.limit_up_codes_percent: # # 减去当前票的涨幅 # total_industry_limit_percent -= global_util.limit_up_codes_percent[code] return total_industry_limit_percent # 获取量 @classmethod def __get_volumns(cls, code): volumn_day60_max, volumn_yest, volumn_today = global_util.max60_volumn.get( code), global_util.yesterday_volumn.get(code), global_util.today_volumn.get(code) if volumn_day60_max is None or volumn_yest is None: global_data_loader.load_volumn() volumn_day60_max, volumn_yest, volumn_today = global_util.max60_volumn.get( code), global_util.yesterday_volumn.get(code), global_util.today_volumn.get(code) return volumn_day60_max, volumn_yest, volumn_today @classmethod def __get_rate_factors(cls, code): zyltgb = global_util.zyltgb_map.get(code) total_industry_limit_percent = cls.__get_industry_limit_percent(code) # 获取量 volumn_day60_max, volumn_yest, volumn_today = cls.__get_volumns(code) # 首次涨停时间 limit_up_time = global_util.limit_up_time.get(code) if limit_up_time is None: limit_up_time = limit_up_time_manager.get_limit_up_time(code) big_money_num = global_util.big_money_num.get(code) if big_money_num is None: big_money_num = big_money_num_manager.get_num(code) return ( zyltgb, total_industry_limit_percent, volumn_day60_max, volumn_yest, volumn_today, limit_up_time, big_money_num) @classmethod def factors_to_string(cls, code): vals = cls.__get_rate_factors(code) return "zyltgb:%s, total_industry_limit_percent:%s, volumn_day60_max:%s, volumn_yest:%s, volumn_today:%s,limit_up_time:%s, big_money_num:%s" % vals @classmethod def get_zyltgb(cls, code): zyltgb = global_util.zyltgb_map.get(code) if zyltgb is None: global_data_loader.load_zyltgb() zyltgb = global_util.zyltgb_map.get(code) return zyltgb @classmethod def compute_m_value(cls, code): zyltgb = global_util.zyltgb_map.get(code) if zyltgb is None: global_data_loader.load_zyltgb() zyltgb = global_util.zyltgb_map.get(code) if zyltgb is None: print("没有获取到自由流通市值") return 10000000 zyltgb = cls.get_base_safe_val(zyltgb) rate, msg = cls.compute_rate_by_code(code) # print("m值获取：", code, round(zyltgb * rate)) return round(zyltgb * rate), msg # 获取安全笔数 @classmethod def get_safe_buy_count(cls, code): gb = cls.get_zyltgb(code) if not gb: # 默认8笔 return 8 count = gb // 100000000 if count <= 6: count = 8 elif count < 32: count = round(8 + 0.5 * (count - 6)) else: count = 21 volumn_day60_max, volumn_yest, volumn_today = cls.__get_volumns(code) rate = cls.get_volumn_rate(volumn_day60_max, volumn_yest, volumn_today) # 取大单影响值与行业影响值的较大值 count = round(count * (1 - rate)) if count < 8: count = 8 elif count > 21: count = 21 return count # l2因子归因数据 class L2TradeFactorSourceDataUtil: # 是否为大单 @classmethod def is_big_money(cls, data): if int(data["val"]["limitPrice"]) != 1: return False if int(data["val"]["num"]) >= 9000: return True money = round(float(data["val"]["price"]) * int(data["val"]["num"]) * 100) if money >= 5000000: return True return False if __name__ == "__main__": print(L2TradeFactorUtil.get_safe_buy_count("003005")) # print(L2TradeFactorUtil.get_rate_factors("003004")) # print(L2TradeFactorUtil.factors_to_string("003004")) print(L2TradeFactorUtil.get_safe_buy_count("002864")) # print(L2TradeFactorUtil.get_limit_up_time_rate("11:30:00")) # print(L2TradeFactorUtil.get_limit_up_time_rate("13:00:00")) # print(L2TradeFactorUtil.get_limit_up_time_rate("13:48:00")) # print(L2TradeFactorUtil.get_limit_up_time_rate("13:53:23")) # print(L2TradeFactorUtil.get_limit_up_time_rate("14:23:23")) # print(L2TradeFactorUtil.get_big_money_rate(2)) # print(L2TradeFactorUtil.get_big_money_rate(3))