# 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]}")
|
|
|
# 计算tick涨幅公式【为卖出策略单独创建函数】
|
# 初始化历史价格
|
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.05),
|
(("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
|
|
|
# 示例使用
|
# now = datetime.now()
|
# print(f"secure_volume(now)=={secure_volume(now)}")
|
|
# 计算 委买和委卖的比例函数(获取买盘强度数据)【掘金数据结构】
|
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
|
