# coding=utf-8 from __future__ import print_function, absolute_import, unicode_literals import decimal from log_module.log import logger_common # from datetime import datetime from strategy import data_cache from utils import hx_qc_value_util # 获取logger实例 logger = logger_common # 将纯数字代码转化为=》掘金格式股票代码 def format_stock_symbol(code): if code.startswith('6'): return "SHSE." + code elif code.startswith('0'): return "SZSE." + code elif code.startswith('2'): return "SZSE." + code elif code.startswith('3'): return "SZSE." + code else: print(f"这个代码有异常，无法处理{code}") return code # 将掘金格式股票代码转化=》纯数字代码 def format_stock_code(symbol): stock_code = symbol.split('.')[1] # 将symbol转为纯数字编号股票代码 return stock_code # 查询前N个交易日的具体交易日期函数:在调用时date_of_the_day应该传入当天日期，num是前跨天数。循环取值后 返回的date_of_the_day值就是目标具体的交易日期， def pre_num_trading_day(date_of_the_day, num): for i in range(num): pre_date = hx_qc_value_util.get_previous_trading_date(date_of_the_day) # 获取前一个交易日API date_of_the_day = pre_date return date_of_the_day # 计算当日涨幅公式 def intraday_growth(price, pre_close): if price is not None and pre_close is not None: today_growth = (price - pre_close) / pre_close * 100 # 计算涨幅百分比 today_growth = round(today_growth, 2) return today_growth # # open_growth = intraday_growth(10.75, 11.08) # print(f"open_growth=={open_growth}") # # 计算瞬时涨幅公式 # # 初始化历史价格 # price_history = {} # # # def calculate_growth(symbol, price): # try: # if symbol not in price_history: # return 0 # 不足两个历史价格，无法计算涨幅 # last_price = price_history[symbol] # # print(f"price_history[symbol]~~~~~~~~~{price_history[symbol]}") # if last_price != 0: # growth = (price - last_price) / last_price * 100 # 计算涨幅百分比 # return growth # finally: # price_history[symbol] = price # # print(f"price======={price}") # # print(f"price_history=={price_history}") # # print(f"price_history[symbol]=={price_history[symbol]}") # # 计算瞬时间隔涨幅公式【为卖出策略单独创建函数】 # 初始化历史价格 history_price = {} def tick_growth(symbol, price): try: if symbol not in history_price: return 0 # 不足两个历史价格，无法计算涨幅 last_price = history_price[symbol] # print(f"last_price===={last_price}") if last_price != 0: growth = (price - last_price) / last_price * 100 # 计算涨幅百分比 return growth finally: history_price[symbol] = price # print(f"price======={price}") # print(f"price_history=={price_history}") # print(f"price_history[symbol]=={price_history[symbol]}") # 计算当日涨停价函数 形参pre_close 实参应传入 pre_close 【return的结果需要小数点后两位,在第三位进行四舍五入】 def limit_up_price(pre_close): limit_up_price_data = decimal.Decimal(str(pre_close)) * decimal.Decimal("1.1") limit_up_price_value = limit_up_price_data.quantize(decimal.Decimal("0.00"), decimal.ROUND_HALF_UP) return float(limit_up_price_value) # 测试计算用 # limit_up = limit_up_price(15) # print(f"limit_up=={limit_up}") # 计算当日跌停价函数 形参pre_close 实参应传入 pre_close 【return的结果需要小数点后两位round取整】 def limit_down_price(pre_close): limit_down_price_data = decimal.Decimal(str(pre_close)) * decimal.Decimal("0.9") limit_down_price_value = limit_down_price_data.quantize(decimal.Decimal("0.00"), decimal.ROUND_HALF_UP) return float(limit_down_price_value) # 测试计算用 # limit_down = limit_down_price(20) # print(f"limit_up=={limit_down}") # print(limit_up_price(24.95)) # print(calculate_growth("000333",54.3)) # 历史K线累计涨停天数函数 def count_limit_up_day(k_line_data): limit_up_day = 0 # 初始化涨停天数 # 计算90天内涨停天数 for i in k_line_data: if 'attribute' in i and (i['attribute'] in data_cache.limit_up_type): limit_up_day += 1 # print(f"涨停提日期：{i['bob']}") # print(f"涨停天数----limit_up_day======{limit_up_day}") return limit_up_day # 是否接近三种大抛压位计算函数 def position_of_throwing_pressure(k_line_data, current_open, current_volume, today_limit_up_price): """ :param k_line_data:个股对应的K线属性数据 :param current_open:个股当前最新价 :param current_volume:个股当日当时交易量 :param today_limit_up_price:个股当日涨停价 :return:是否接近涨停抛压位，是否接近炸板抛压位，是否接近跌停抛压位， """ limit_up_throwing_pressure = False frying_plate_throwing_pressure = False limit_down_throwing_pressure = False # 找到最近7日内的有涨停的序号 limit_up_day_min_index = next((i for i, d in enumerate(k_line_data[0:7]) if 'attribute' in d and d['attribute'] in data_cache.limit_up_type), None) # 如果没有找到，返回None # 找到最近7日内的有炸板的序号 frying_plate_day_min_index = next((i for i, d in enumerate(k_line_data[0:7]) if 'attribute' in d and d['attribute'] in data_cache.frying_plate_type), None) # 如果没有找到，返回None # 找到最近7日内的有跌停的序号 limit_down_day_min_index = next((i for i, d in enumerate(k_line_data[0:7]) if 'attribute' in d and d['attribute'] in data_cache.limit_down_type), None) # 如果没有找到，返回None # 最近的涨停序号存在 且 非昨日 if limit_up_day_min_index is not None and limit_up_day_min_index > 0: # 开盘价 < 涨停当日最高价 <= 今日涨停价 if current_open < k_line_data[limit_up_day_min_index]['high'] <= today_limit_up_price: # 涨停次日量 < 涨停当日量 * 1.2 if k_line_data[limit_up_day_min_index - 1]['volume'] < k_line_data[limit_up_day_min_index]['volume'] * 1.2: # 涨停次日最高价 < 涨停当日最高价的1.02倍。 if k_line_data[limit_up_day_min_index - 1]['high'] < k_line_data[limit_up_day_min_index]['high'] * 1.02: logger.info(f"【{k_line_data[0]['sec_name']}】,涨停大抛压日期：{k_line_data[limit_up_day_min_index]['bob']}") limit_up_throwing_pressure = True # 最近的炸板序号存在 且 非昨日 if frying_plate_day_min_index is not None and frying_plate_day_min_index > 0: # 开盘价<炸板当日最高价<=今日涨停价 或 开盘价<炸板当日收盘价（基本意味着炸板当日的均价）<=今日涨停价 或 开盘价<炸板次日最高价<=今日涨停价 if (current_open < k_line_data[frying_plate_day_min_index]['high'] <= today_limit_up_price) or (current_open < k_line_data[frying_plate_day_min_index]['close'] <= today_limit_up_price) or (current_open < k_line_data[frying_plate_day_min_index - 1]['high'] <= today_limit_up_price): # 炸板次日量 < 炸板当日量 * 1.2 if k_line_data[frying_plate_day_min_index - 1]['volume'] < k_line_data[frying_plate_day_min_index]['volume'] * 1.2: # 炸板次日最高价 < 炸板当日最高价的1.02倍 if k_line_data[frying_plate_day_min_index - 1]['high'] < k_line_data[frying_plate_day_min_index]['high'] * 1.02: logger.info(f"【{k_line_data[0]['sec_name']}】,炸板大抛压日期：{k_line_data[frying_plate_day_min_index]['bob']}，炸板次日量：{round(k_line_data[frying_plate_day_min_index - 1]['volume']/1000000, 2)} 万手") frying_plate_throwing_pressure = True # 最近的跌停序号存在 且 非昨日 if limit_down_day_min_index is not None and limit_down_day_min_index > 0: # 开盘价 < 跌停当日最低价 <= 今日涨停价 或 开盘价 < 跌停当日最高价 <= 今日涨停价 或 开盘价 < 跌停当日收盘价 <= 今日涨停价 if (current_open < k_line_data[limit_down_day_min_index]['low'] <= today_limit_up_price) or (current_open < k_line_data[limit_down_day_min_index]['high'] <= today_limit_up_price) or (current_open < k_line_data[limit_down_day_min_index - 1]['close'] <= today_limit_up_price): # 跌停次日量 < 跌停当日量 if k_line_data[limit_down_day_min_index - 1]['volume'] < k_line_data[limit_down_day_min_index]['volume']: logger.info(f"【{k_line_data[0]['sec_name']}】,跌停大抛压日期：{k_line_data[limit_down_day_min_index]['bob']}") limit_down_throwing_pressure = True return limit_up_throwing_pressure, frying_plate_throwing_pressure, limit_down_throwing_pressure # 安全交易量公式 用于计算不同时间段理论的安全交易量值 def secure_volume(now_date_time): # 定义时间段的开始和结束时间（使用字符串格式） time_slots = [ (("09:30:00", "09:30:30"), 0.04), (("09:30:30", "09:31:00"), 0.08), (("09:31:00", "09:31:30"), 0.1), (("09:31:30", "09:32:00"), 0.15), (("09:32:00", "09:32:30"), 0.18), (("09:32:30", "09:33:00"), 0.2), (("09:33:00", "09:34:00"), 0.25), (("09:35:00", "10:40:00"), 0.3), (("09:40:00", "10:00:00"), 0.4), (("10:00:00", "11:00:00"), 0.6), (("11:00:00", "14:00:00"), 0.8), (("14:00:00", "15:00:00"), 1.2), ] # 将now格式化为字符串以便比较 now_str = now_date_time.strftime("%H:%M:%S") # 遍历时间段，找到匹配的时间段并返回对应的交易量比 for (start_str, end_str), ratio in time_slots: if start_str <= now_str < end_str: return ratio # 如果没有匹配的时间段，返回0 return 0 # 充分交易量公式 用于计算日内涨幅段理论的安全交易量值 def sufficient_volume(current_volume, yesterday_volume, today_growth): if today_growth > 0: if current_volume > yesterday_volume * 0.8 * today_growth: return True else: return False # 计算 委买和委卖的比例函数(获取买盘强度数据)【掘金数据结构】 def buying_and_selling_ratio(current_quotes): buying_strong = False if current_quotes[0]['bid_v'] > current_quotes[0]['ask_v'] and ( current_quotes[0]['bid_v'] + current_quotes[1]['bid_v'] + current_quotes[2]['bid_v'] + current_quotes[3]['bid_v'] + current_quotes[4]['bid_v']) > ( current_quotes[0]['ask_v'] + current_quotes[1]['ask_v'] + current_quotes[2]['ask_v'] + current_quotes[3]['ask_v'] + current_quotes[4]['ask_v']): buying_strong = True return buying_strong # 计算 委买和委卖的比例函数(获取买盘强度数据)【华鑫数据结构】 # current_quotes_buy == [[23.25, 800], [23.24, 1100], [23.23, 1100], [23.22, 2200], [23.21, 2300]] # current_quotes_sell == [[23.27, 500], [23.29, 200], [23.3, 6900], [23.31, 500], [23.32, 200]] def get_buying_strength(current_quotes_buy, current_quotes_sell): buying_strong = False if current_quotes_buy[0][1] > current_quotes_sell[0][1] and ( current_quotes_buy[0][1] + current_quotes_buy[1][1] + current_quotes_buy[2][1] + current_quotes_buy[3][1] + current_quotes_buy[4][1]) > ( current_quotes_sell[0][1] + current_quotes_sell[1][1] + current_quotes_sell[2][1] + current_quotes_sell[3][1] + current_quotes_sell[4][1]): buying_strong = True return buying_strong # 最大化资金利用率股数下单比例 def maximum_buying_ratio(today_addition_num): # 查询今日新增持仓数量 仅仅在本地测试时使用 # today_addition_num = len(data_cache.addition_position_symbols_set) # 可用余额比例分配方法函数 if today_addition_num < 1: buying_ratio = 0.25 elif today_addition_num < 2: buying_ratio = 0.33 elif today_addition_num < 3: buying_ratio = 0.5 elif today_addition_num < 4: buying_ratio = 0.75 else: buying_ratio = 0.01 return buying_ratio