"""
|
数据工具类
|
提供各种K线/涨停形态、技术指标和价格模式的计算方法
|
"""
|
import constant
|
from utils import tool
|
|
|
class KTickLineAnalyzer:
|
"""
|
TICK级K线分析工具类
|
"""
|
|
@classmethod
|
def calculate_upper_limit_price(cls, stock_code, pre_close):
|
"""
|
计算价格上限
|
@param stock_code: 证券代码
|
@param pre_close: 前一日收盘价
|
@return: 价格上限
|
"""
|
return round(tool.get_limit_up_rate(stock_code) * pre_close, 2)
|
|
@classmethod
|
def calculate_lower_limit_price(cls, stock_code, pre_close):
|
"""
|
计算价格下限
|
@param stock_code: 证券代码
|
@param pre_close: 前一日收盘价
|
@return: 价格下限
|
"""
|
# A股ST股票涨跌幅为5%，其他为10%
|
return round(tool.get_limit_down_rate(stock_code) * pre_close, 2)
|
|
@classmethod
|
def get_yesterday_close(cls, k_data):
|
"""
|
获取昨日收盘价
|
@param k_data: K线数据列表
|
@return: 昨日收盘价
|
"""
|
return k_data[0]['close']
|
|
@classmethod
|
def get_yesterday_high(cls, k_data):
|
"""
|
获取昨日最高价
|
@param k_data: K线数据列表
|
@return: 昨日最高价
|
"""
|
return k_data[0]['high']
|
|
@classmethod
|
def get_yesterday_amount(cls, k_data):
|
"""
|
获取昨日成交额
|
@param k_data: K线数据列表
|
@return: 昨日最高价
|
"""
|
return k_data[0]['amount']
|
|
@classmethod
|
def get_recent_days_high(cls, k_data, days):
|
"""
|
获取近几个交易日的最高价
|
@param k_data: K线数据列表
|
@param days: 交易日数量
|
@return: 最高价
|
"""
|
return max([d['high'] for d in k_data[:days]])
|
|
@classmethod
|
def is_yesterday_limit_up(cls, k_data):
|
"""
|
判断昨日是否涨停
|
@param k_data: K线数据列表
|
@return: True/False
|
"""
|
return k_data[0]['close'] >= cls.calculate_upper_limit_price(k_data[0]["sec_id"], k_data[0]["pre_close"])
|
|
@classmethod
|
def is_yesterday_limit_down(cls, k_data):
|
"""
|
判断昨日是否跌停
|
@param k_data: K线数据列表
|
@return: True/False
|
"""
|
return k_data[0]['close'] <= cls.calculate_lower_limit_price(k_data[0]["sec_id"], k_data[0]["pre_close"])
|
|
@classmethod
|
def is_yesterday_exploded(cls, k_data):
|
"""
|
判断昨日是否炸板
|
@param k_data: K线数据列表
|
@return: True/False
|
"""
|
return k_data[0]['high'] >= cls.calculate_upper_limit_price(k_data[0]["sec_id"], k_data[0]["pre_close"]) > \
|
k_data[0]['close']
|
|
@classmethod
|
def get_yesterday_volume(cls, k_data):
|
"""
|
获取昨日成交量
|
@param k_data: K线数据列表
|
@return: 成交量
|
"""
|
return k_data[0]['volume']
|
|
@classmethod
|
def get_recent_days_max_volume(cls, k_data, days):
|
"""
|
获取近几个交易日的最高量及对应K线数据
|
@param k_data: K线数据列表
|
@param days: 交易日数量
|
@return: (最高量, 对应K线数据)
|
"""
|
max_volume_day = max(k_data[:days], key=lambda d: d['volume'])
|
return max_volume_day['volume'], max_volume_day
|
|
@classmethod
|
def get_recent_limit_up_count(cls, k_data, days):
|
"""
|
获取近几个交易日的涨停次数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 涨停次数
|
"""
|
return sum(
|
1 for d in k_data[:days] if d['close'] >= cls.calculate_upper_limit_price(d["sec_id"], d["pre_close"]))
|
|
@classmethod
|
def get_recent_exploded_count(cls, k_data, days):
|
"""
|
获取近几个交易日的炸板次数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 炸板次数
|
"""
|
return sum(1 for d in k_data[:days] if
|
d['high'] >= cls.calculate_upper_limit_price(d["sec_id"], d["pre_close"]) > d['close'])
|
|
@classmethod
|
def get_recent_limit_down_count(cls, k_data, days):
|
"""
|
获取近几个交易日的跌停次数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 跌停次数
|
"""
|
return sum(
|
1 for d in k_data[:days] if d['close'] <= cls.calculate_lower_limit_price(d["sec_id"], d["pre_close"]))
|
|
@classmethod
|
def get_first_limit_up_avg_premium(cls, k_data, days):
|
"""
|
获取近几个交易日的首板涨停平均溢价率
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 平均溢价率
|
"""
|
premiums = []
|
for i in range(days):
|
if i + 1 >= len(k_data):
|
continue
|
if i < 1:
|
continue
|
# 判断当日涨停且前日非涨停
|
if (k_data[i]['close'] >= cls.calculate_upper_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] < cls.calculate_upper_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])):
|
# 计算溢价率 = (当日最高价 - 前日收盘价)/前日收盘价
|
premium = (k_data[i - 1]['high'] - k_data[i]['close']) / k_data[i]['close']
|
premiums.append(premium)
|
return sum(premiums) / len(premiums) if premiums else 0
|
|
@classmethod
|
def get_first_exploded_avg_premium(cls, k_data, days):
|
"""
|
获取近几个交易日的首板炸板平均溢价率
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 平均溢价率
|
"""
|
premiums = []
|
for i in range(days):
|
if i + 1 >= len(k_data):
|
continue
|
if i < 1:
|
continue
|
# 判断当日炸板且前日非涨停
|
if (k_data[i]['high'] >= cls.calculate_upper_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"]) >
|
k_data[i]['close']) and \
|
(k_data[i + 1]['close'] < cls.calculate_upper_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])):
|
# 计算溢价率 = (当日最高价 - 前日收盘价)/前日收盘价
|
premium = (k_data[i + 1]['high'] - k_data[i]['close']) / k_data[i]['close']
|
premiums.append(premium)
|
return sum(premiums) / len(premiums) if premiums else 0
|
|
@classmethod
|
def get_first_limit_up_days(cls, k_data, days):
|
"""
|
获取近几个交易日的首板天数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 首板天数
|
"""
|
count = 0
|
for i in range(days):
|
if i + 1 >= len(k_data):
|
continue
|
# 判断当日涨停且前日非涨停
|
if (k_data[i]['close'] >= cls.calculate_upper_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] < cls.calculate_upper_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])):
|
count += 1
|
return count
|
|
@classmethod
|
def get_second_limit_up_days(cls, k_data, days):
|
"""
|
获取近几个交易日的二板天数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 二板天数
|
"""
|
count = 0
|
for i in range(days):
|
if i + 2 >= len(k_data):
|
continue
|
# 判断连续两日涨停且第三日非涨停
|
if (k_data[i]['close'] >= cls.calculate_upper_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] >= cls.calculate_upper_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])) and \
|
(k_data[i + 2]['close'] < cls.calculate_upper_limit_price(k_data[i + 2]["sec_id"],
|
k_data[i + 2]["pre_close"])):
|
count += 1
|
return count
|
|
@classmethod
|
def get_third_limit_up_days(cls, k_data, days):
|
"""
|
获取近几个交易日的三板天数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 三板天数
|
"""
|
count = 0
|
for i in range(days):
|
if i + 3 >= len(k_data):
|
continue
|
# 判断连续三日涨停且第四日非涨停
|
if (k_data[i]['close'] >= cls.calculate_upper_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] >= cls.calculate_upper_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])) and \
|
(k_data[i + 2]['close'] >= cls.calculate_upper_limit_price(k_data[i + 2]["sec_id"],
|
k_data[i + 2]["pre_close"])) and \
|
(k_data[i + 3]['close'] < cls.calculate_upper_limit_price(k_data[i + 3]["sec_id"],
|
k_data[i + 3]["pre_close"])):
|
count += 1
|
return count
|
|
@classmethod
|
def get_fourth_or_more_limit_up_days(cls, k_data, days):
|
"""
|
获取近几个交易日的四板及以上天数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 四板及以上天数
|
"""
|
count = 0
|
for i in range(days):
|
if i + 3 >= len(k_data):
|
continue
|
# 判断连续三日涨停且第四日涨停
|
if (k_data[i]['close'] >= cls.calculate_upper_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] >= cls.calculate_upper_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])) and \
|
(k_data[i + 2]['close'] >= cls.calculate_upper_limit_price(k_data[i + 2]["sec_id"],
|
k_data[i + 2]["pre_close"])) and \
|
(k_data[i + 3]['close'] >= cls.calculate_upper_limit_price(k_data[i + 3]["sec_id"],
|
k_data[i + 3]["pre_close"])):
|
count += 1
|
return count
|
|
@classmethod
|
def get_first_limit_down_days(cls, k_data, days):
|
"""
|
获取近几个交易日的首跌停天数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 首跌停天数
|
"""
|
count = 0
|
for i in range(days):
|
if i + 1 >= len(k_data):
|
continue
|
# 判断当日跌停且前日非跌停
|
if (k_data[i]['close'] <= cls.calculate_lower_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] > cls.calculate_lower_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])):
|
count += 1
|
return count
|
|
@classmethod
|
def get_second_limit_down_days(cls, k_data, days):
|
"""
|
获取近几个交易日的二跌停天数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 二跌停天数
|
"""
|
count = 0
|
for i in range(days):
|
if i + 2 >= len(k_data):
|
continue
|
# 判断连续两日跌停且第三日非跌停
|
if (k_data[i]['close'] <= cls.calculate_lower_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] <= cls.calculate_lower_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])) and \
|
(k_data[i + 2]['close'] > cls.calculate_lower_limit_price(k_data[i + 2]["sec_id"],
|
k_data[i + 2]["pre_close"])):
|
count += 1
|
return count
|
|
@classmethod
|
def get_third_limit_down_days(cls, k_data, days):
|
"""
|
获取近几个交易日的三跌停天数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 三跌停天数
|
"""
|
count = 0
|
for i in range(days):
|
if i + 3 >= len(k_data):
|
continue
|
# 判断连续三日跌停且第四日非跌停
|
if (k_data[i]['close'] <= cls.calculate_lower_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] <= cls.calculate_lower_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])) and \
|
(k_data[i + 2]['close'] <= cls.calculate_lower_limit_price(k_data[i + 2]["sec_id"],
|
k_data[i + 2]["pre_close"])) and \
|
(k_data[i + 3]['close'] > cls.calculate_lower_limit_price(k_data[i + 3]["sec_id"],
|
k_data[i + 3]["pre_close"])):
|
count += 1
|
return count
|
|
@classmethod
|
def get_fourth_or_more_limit_down_days(cls, k_data, days):
|
"""
|
获取近几个交易日的四跌停及以上天数
|
@param k_data: K线数据列表(近150个交易日，不包含当前交易日，时间倒序)
|
@param days: 交易日数量
|
@return: 四跌停及以上天数
|
"""
|
count = 0
|
for i in range(days):
|
if i + 3 >= len(k_data):
|
continue
|
# 判断连续三日跌停且第四日跌停
|
if (k_data[i]['close'] <= cls.calculate_lower_limit_price(k_data[i]["sec_id"], k_data[i]["pre_close"])) and \
|
(k_data[i + 1]['close'] <= cls.calculate_lower_limit_price(k_data[i + 1]["sec_id"],
|
k_data[i + 1]["pre_close"])) and \
|
(k_data[i + 2]['close'] <= cls.calculate_lower_limit_price(k_data[i + 2]["sec_id"],
|
k_data[i + 2]["pre_close"])) and \
|
(k_data[i + 3]['close'] <= cls.calculate_lower_limit_price(k_data[i + 3]["sec_id"],
|
k_data[i + 3]["pre_close"])):
|
count += 1
|
return count
|
|
|
class K60SLineAnalyzer:
|
"""
|
60sk线分析
|
"""
|
|
@classmethod
|
def get_close_price_of_max_volume(cls, k_data):
|
"""
|
获取60sk线中最高量对应的收盘价
|
@param k_data: 60秒K线数据列表
|
@return: 最高成交量对应的收盘价，60sK线数据
|
"""
|
max_volume_item = max(k_data, key=lambda x: x['volume'])
|
return max_volume_item['close'], max_volume_item
|
|
|
class KPLLimitUpDataAnalyzer:
|
"""
|
开盘啦涨停数据分析
|
"""
|
|
@classmethod
|
def get_most_common_reasons(cls, limit_up_data, min_day, max_day):
|
"""
|
获取某段日期开盘啦出现最多原因出现的天数，排除特殊涨停原因，多个原因有一样的涨停天数，返回所有
|
@param limit_up_data: 近150个交易日某个代码的涨停数据，格式：[(代码,日期,涨停原因)]
|
@param min_day: 最小日期
|
@param max_day: 最大日期
|
@return: [(原因, 出现的天数)]
|
"""
|
reason_counts = {}
|
special_reasons = constant.KPL_INVALID_BLOCKS
|
if limit_up_data:
|
for _, date, reason in limit_up_data:
|
if min_day <= date <= max_day and reason not in special_reasons:
|
reason_counts[reason] = reason_counts.get(reason, 0) + 1
|
if not reason_counts:
|
return []
|
max_count = max(reason_counts.values())
|
return [(reason, count) for reason, count in reason_counts.items() if count == max_count]
|
