446 lines
16 KiB
Python
446 lines
16 KiB
Python
from ChanEnum import Chan_FX_TYPE
|
|
class ChanKLU:
|
|
def __init__(self, time, open, high, low, close, volume):
|
|
# _time, _close, _open, _high, _low, _extra_info={}
|
|
self.kl_type = None
|
|
self.time = time
|
|
self.close = close
|
|
self.open = open
|
|
self.high = high
|
|
self.low = low
|
|
self.volume = volume
|
|
self.idx = 0
|
|
self.index = 0
|
|
self.macd = 0
|
|
self.signal = 0
|
|
self.macdhist = 0
|
|
self.ma5 = 0
|
|
self.ma10 = 0
|
|
self.ma30 = 0
|
|
self.ma50 = 0
|
|
self.ma200 = 0
|
|
self.ma250 = 0
|
|
self.rsi = 0
|
|
self.volume_ratio = 0
|
|
|
|
# === 新增：实时分型相关属性 ===
|
|
self.pre = None # 前一根K线
|
|
self.next = None # 后一根K线
|
|
self.fx_type = Chan_FX_TYPE.UNKNOWN # 分型类型：0=无分型，1=顶分型，-1=底分型
|
|
self.fx_strength = 0 # 分型强度：0-100
|
|
self.fx_confirmed = False # 分型是否确认
|
|
|
|
def set_next(self, next):
|
|
self.next = next
|
|
self.update_realtime_analysis()
|
|
#if self.fx_type != Chan_FX_TYPE.UNKNOWN and self.fx_strength > 1:
|
|
#print(self.index, self.time, self.fx_type, self.fx_confirmed, self.fx_strength)
|
|
def set_pre(self, pre):
|
|
self.pre = pre
|
|
def detect_realtime_fx(self):
|
|
"""
|
|
实时检测K线分型（不等待KLC确认）
|
|
基于原始K线的即时分型识别
|
|
"""
|
|
if not self.pre or not self.next:
|
|
self.fx_type = Chan_FX_TYPE.UNKNOWN
|
|
return False
|
|
|
|
# 顶分型检测
|
|
if (self.high > self.pre.high and
|
|
self.high > self.next.high):
|
|
self.fx_type = Chan_FX_TYPE.TOP
|
|
self.fx_confirmed = True
|
|
return True
|
|
|
|
# 底分型检测
|
|
elif (self.low < self.pre.low and
|
|
self.low < self.next.low):
|
|
self.fx_type = Chan_FX_TYPE.BOTTOM
|
|
self.fx_confirmed = True
|
|
return True
|
|
|
|
self.fx_type = Chan_FX_TYPE.UNKNOWN
|
|
self.fx_confirmed = False
|
|
return False
|
|
|
|
def calculate_realtime_fx_strength(self):
|
|
"""
|
|
用self.pre和self.next实现分型强弱判断（与KLC中cal_fx_strength一致）
|
|
|
|
核心缠论原理：
|
|
- 强分型：出现在笔的末端，能够终结当前笔，标志着趋势转折
|
|
- 弱分型：出现在笔的中间，是中继性质，笔还会继续延伸
|
|
|
|
返回值：
|
|
3: 极强分型（笔终结+强确认）
|
|
2: 强分型（笔终结）
|
|
1: 偏强分型（可能终结笔）
|
|
0: 中性分型
|
|
-1: 偏弱分型（中继特征明显）
|
|
-2: 弱分型（明显中继）
|
|
-3: 极弱分型（无效分型）
|
|
"""
|
|
# 检查是否为分型，且有前后K线数据
|
|
if self.fx_type == Chan_FX_TYPE.UNKNOWN:
|
|
return 0
|
|
if not self.pre or not self.next:
|
|
return 100
|
|
# === 核心判断：分型在笔中的位置 ===
|
|
|
|
# 1. 检查这个分型是否能够终结当前笔
|
|
is_bi_end = self._check_if_bi_ending_fx()
|
|
|
|
# 2. 检查分型的后续走势确认
|
|
post_fx_confirmation = self._check_post_fx_confirmation()
|
|
|
|
# 3. 检查分型的标准性和强度
|
|
fx_quality = self._check_fx_quality()
|
|
|
|
# === 综合评分 ===
|
|
base_score = 0
|
|
|
|
# 笔位置是最重要的判断标准
|
|
if is_bi_end == 2: # 强烈确认笔终结
|
|
base_score = 2
|
|
elif is_bi_end == 1: # 可能笔终结
|
|
base_score = 1
|
|
elif is_bi_end == -1: # 明显中继
|
|
base_score = -2
|
|
elif is_bi_end == -2: # 强烈中继特征
|
|
base_score = -3
|
|
else: # 不确定
|
|
base_score = 0
|
|
|
|
# 后续确认调整
|
|
base_score += post_fx_confirmation
|
|
|
|
# 分型质量调整
|
|
base_score += fx_quality
|
|
|
|
# 限制在-3到3范围内
|
|
final_score = max(-3, min(3, base_score))
|
|
self.fx_strength = final_score
|
|
# 转换为0-100分制以保持接口一致性
|
|
#self.fx_strength = int((final_score + 3) * 100 / 6) # -3到3映射到0-100
|
|
if final_score > 1.8:
|
|
print(self.time, final_score, is_bi_end, post_fx_confirmation, fx_quality)
|
|
#print(self.time, final_score, is_bi_end, post_fx_confirmation, fx_quality)
|
|
return self.fx_strength
|
|
|
|
def _check_if_bi_ending_fx(self):
|
|
"""
|
|
检查分型是否为笔终结分型
|
|
返回值：
|
|
2: 强烈确认笔终结
|
|
1: 可能笔终结
|
|
0: 不确定
|
|
-1: 明显中继
|
|
-2: 强烈中继特征
|
|
"""
|
|
# 检查是否有足够的后续数据来判断
|
|
if not self.next or not hasattr(self.next, 'next'):
|
|
return 0
|
|
|
|
# 获取分型后的几根K线数据
|
|
subsequent_klus = []
|
|
temp = self.next
|
|
for i in range(2): # 检查后续2根K线
|
|
if temp:
|
|
subsequent_klus.append(temp)
|
|
temp = temp.next if hasattr(temp, 'next') else None
|
|
else:
|
|
break
|
|
|
|
if len(subsequent_klus) < 2:
|
|
return 0
|
|
|
|
if self.fx_type == Chan_FX_TYPE.TOP:
|
|
return self._check_top_bi_ending(subsequent_klus)
|
|
else: # BOTTOM
|
|
return self._check_bottom_bi_ending(subsequent_klus)
|
|
|
|
def _check_top_bi_ending(self, subsequent_klus):
|
|
"""检查顶分型是否为笔终结"""
|
|
# 强烈笔终结特征：
|
|
# 1. 后续K线持续下跌，且跌破关键位置
|
|
# 2. 没有新的更高的高点出现
|
|
|
|
broken_key_levels = 0
|
|
new_highs = 0
|
|
downward_trend = 0
|
|
|
|
# 检查关键价位突破
|
|
first_low = self.pre.low
|
|
middle_low = self.low
|
|
key_support = min(first_low, middle_low)
|
|
|
|
for i, klu in enumerate(subsequent_klus):
|
|
# 检查是否跌破关键支撑
|
|
if klu.low < key_support:
|
|
broken_key_levels += 1
|
|
|
|
# 检查是否出现新高
|
|
if klu.high > self.high:
|
|
new_highs += 1
|
|
|
|
# 检查下跌趋势
|
|
if i > 0 and klu.close < subsequent_klus[i-1].close:
|
|
downward_trend += 1
|
|
|
|
# 强烈笔终结：跌破关键位且无新高
|
|
if broken_key_levels >= 1 and new_highs == 0 and downward_trend >= 2:
|
|
return 2
|
|
|
|
# 可能笔终结：部分条件满足
|
|
if (broken_key_levels >= 1 and new_highs <= 1) or (new_highs == 0 and downward_trend >= 3):
|
|
return 1
|
|
|
|
# 明显中继：出现新高且未跌破关键位
|
|
if new_highs >= 2 and broken_key_levels == 0:
|
|
return -2
|
|
|
|
# 中继倾向：出现新高
|
|
if new_highs >= 1:
|
|
return -1
|
|
|
|
return 0
|
|
|
|
def _check_bottom_bi_ending(self, subsequent_klus):
|
|
"""检查底分型是否为笔终结"""
|
|
# 强烈笔终结特征：
|
|
# 1. 后续K线持续上涨，且突破关键位置
|
|
# 2. 没有新的更低的低点出现
|
|
|
|
broken_key_levels = 0
|
|
new_lows = 0
|
|
upward_trend = 0
|
|
|
|
# 检查关键价位突破
|
|
first_high = self.pre.high
|
|
middle_high = self.high
|
|
key_resistance = max(first_high, middle_high)
|
|
|
|
for i, klu in enumerate(subsequent_klus):
|
|
# 检查是否突破关键阻力
|
|
if klu.high > key_resistance:
|
|
broken_key_levels += 1
|
|
|
|
# 检查是否出现新低
|
|
if klu.low < self.low:
|
|
new_lows += 1
|
|
|
|
# 检查上涨趋势
|
|
if i > 0 and klu.close > subsequent_klus[i-1].close:
|
|
upward_trend += 1
|
|
|
|
# 强烈笔终结：突破关键位且无新低
|
|
if broken_key_levels >= 1 and new_lows == 0 and upward_trend >= 2:
|
|
return 2
|
|
|
|
# 可能笔终结：部分条件满足
|
|
if (broken_key_levels >= 1 and new_lows <= 1) or (new_lows == 0 and upward_trend >= 3):
|
|
return 1
|
|
|
|
# 明显中继：出现新低且未突破关键位
|
|
if new_lows >= 2 and broken_key_levels == 0:
|
|
return -2
|
|
|
|
# 中继倾向：出现新低
|
|
if new_lows >= 1:
|
|
return -1
|
|
|
|
return 0
|
|
|
|
def _check_post_fx_confirmation(self):
|
|
"""
|
|
检查分型后的走势确认
|
|
返回值：-1到1的调整分数
|
|
"""
|
|
if not self.next:
|
|
return 0
|
|
|
|
score = 0
|
|
|
|
# 检查第三根K线的确认
|
|
third_klu = self.next
|
|
|
|
if self.fx_type == Chan_FX_TYPE.TOP:
|
|
# 顶分型：第三根K线应该走弱
|
|
middle_price = (self.high + self.low) / 2
|
|
|
|
if third_klu.close < middle_price:
|
|
score += 0.5
|
|
if third_klu.low < self.pre.low: # 跌破第一根K线低点
|
|
score += 0.5
|
|
if third_klu.close < third_klu.open and abs(third_klu.close - third_klu.open) > abs(self.close - self.open) * 0.5:
|
|
score += 0.3 # 明显阴线
|
|
|
|
else: # BOTTOM
|
|
# 底分型：第三根K线应该走强
|
|
middle_price = (self.high + self.low) / 2
|
|
|
|
if third_klu.close > middle_price:
|
|
score += 0.5
|
|
if third_klu.high > self.pre.high: # 突破第一根K线高点
|
|
score += 0.5
|
|
if third_klu.close > third_klu.open and abs(third_klu.close - third_klu.open) > abs(self.close - self.open) * 0.5:
|
|
score += 0.3 # 明显阳线
|
|
|
|
return min(1, max(-1, score))
|
|
|
|
def _check_fx_quality(self):
|
|
"""
|
|
检查分型本身的质量
|
|
返回值：-1到1的调整分数
|
|
"""
|
|
score = 0
|
|
|
|
# 检查分型的标准性
|
|
if self.fx_type == Chan_FX_TYPE.TOP:
|
|
# 高点突出程度
|
|
high_diff1 = (self.high - self.pre.high) / self.high if self.high > 0 else 0
|
|
high_diff2 = (self.high - self.next.high) / self.high if self.high > 0 else 0
|
|
min_diff = min(high_diff1, high_diff2)
|
|
|
|
if min_diff > 0.03: # 非常突出
|
|
score += 0.5
|
|
elif min_diff > 0.01: # 比较突出
|
|
score += 0.2
|
|
elif min_diff < 0.003: # 不够突出
|
|
score -= 0.5
|
|
|
|
else: # BOTTOM
|
|
# 低点突出程度
|
|
low_diff1 = (self.pre.low - self.low) / self.pre.low if self.pre.low > 0 else 0
|
|
low_diff2 = (self.next.low - self.low) / self.next.low if self.next.low > 0 else 0
|
|
min_diff = min(low_diff1, low_diff2)
|
|
|
|
if min_diff > 0.03: # 非常突出
|
|
score += 0.5
|
|
elif min_diff > 0.01: # 比较突出
|
|
score += 0.2
|
|
elif min_diff < 0.003: # 不够突出
|
|
score -= 0.5
|
|
|
|
# 检查量价配合
|
|
avg_volume = self._get_avg_volume(lookback=5)
|
|
if avg_volume > 0:
|
|
volume_ratio = self.volume / avg_volume
|
|
if volume_ratio > 1.5:
|
|
score += 0.3
|
|
elif volume_ratio < 0.7:
|
|
score -= 0.2
|
|
|
|
return min(1, max(-1, score))
|
|
|
|
def _get_avg_volume(self, lookback=5):
|
|
"""获取前N根K线平均成交量"""
|
|
volumes = []
|
|
temp = self.pre
|
|
|
|
for i in range(lookback):
|
|
if temp:
|
|
volumes.append(temp.volume)
|
|
temp = temp.pre if hasattr(temp, 'pre') else None
|
|
else:
|
|
break
|
|
|
|
return sum(volumes) / len(volumes) if volumes else self.volume
|
|
|
|
def get_fx_signal(self):
|
|
"""
|
|
获取分型交易信号
|
|
返回: (信号类型, 强度, 建议)
|
|
"""
|
|
if not self.fx_confirmed:
|
|
return ("无信号", 0, "等待分型确认")
|
|
|
|
strength_level = "弱"
|
|
if self.fx_strength >= 80:
|
|
strength_level = "极强"
|
|
elif self.fx_strength >= 65:
|
|
strength_level = "强"
|
|
elif self.fx_strength >= 50:
|
|
strength_level = "中等"
|
|
|
|
if self.fx_type == Chan_FX_TYPE.TOP:
|
|
signal_type = f"{strength_level}顶分型"
|
|
if self.fx_strength >= 65:
|
|
suggestion = "考虑减仓或止盈"
|
|
else:
|
|
suggestion = "谨慎观望"
|
|
else:
|
|
signal_type = f"{strength_level}底分型"
|
|
if self.fx_strength >= 65:
|
|
suggestion = "考虑建仓或加仓"
|
|
else:
|
|
suggestion = "谨慎观望"
|
|
|
|
return (signal_type, self.fx_strength, suggestion)
|
|
|
|
def update_realtime_analysis(self):
|
|
"""
|
|
更新实时分析（在每根K线完成时调用）
|
|
"""
|
|
self.detect_realtime_fx()
|
|
if self.fx_confirmed:
|
|
self.calculate_realtime_fx_strength()
|
|
|
|
def set_idx(self, idx):
|
|
self.idx = idx
|
|
self.index = idx
|
|
|
|
def set_indicators(self, item):
|
|
self.macd = float(item['macd']) if 'macd' in item and item['macd'] else 0
|
|
self.signal = float(item['macdsignal']) if 'macdsignal' in item and item['macdsignal'] else 0
|
|
self.macdhist = float(item['macdhist']) if 'macdhist' in item and item['macdhist'] else 0
|
|
self.ma5 = float(item['ma5']) if 'ma5' in item and item['ma5'] else 0
|
|
self.ma10 = float(item['ma10']) if 'ma10' in item and item['ma10'] else 0
|
|
self.ma30 = float(item['ma30']) if 'ma30' in item and item['ma30'] else 0
|
|
|
|
# 安全检查 ma250、ma50 和 ma200
|
|
self.ma250 = float(item['ma250']) if 'ma250' in item and item['ma250'] else 0
|
|
self.ma50 = float(item['ma50']) if 'ma50' in item and item['ma50'] else 0
|
|
self.ma200 = float(item['ma200']) if 'ma200' in item and item['ma200'] else 0
|
|
|
|
self.rsi = float(item['rsi']) if 'rsi' in item and item['rsi'] else 0
|
|
self.volume_ratio = float(item['volume_ratio']) if 'volume_ratio' in item and item['volume_ratio'] else 0
|
|
|
|
# 设置指标后更新实时分析
|
|
self.update_realtime_analysis()
|
|
|
|
def get_feature_data(self):
|
|
features = dict()
|
|
features['klu_close'] = self.close
|
|
features['klu_open'] = self.open
|
|
features['klu_high'] = self.high
|
|
features['klu_low'] = self.low
|
|
features['klu_volume'] = self.volume
|
|
features['klu_index'] = self.index
|
|
features['klu_macd'] = self.macd
|
|
features['klu_signal'] = self.signal
|
|
features['klu_macdhist'] = self.macdhist
|
|
features['klu_ma5'] = self.ma5
|
|
features['klu_ma10'] = self.ma10
|
|
features['klu_ma30'] = self.ma30
|
|
features['klu_ma50'] = self.ma50
|
|
features['klu_ma200'] = self.ma200
|
|
features['klu_ma250'] = self.ma250
|
|
features['klu_rsi'] = self.rsi
|
|
features['klu_volume_ratio'] = self.volume_ratio
|
|
|
|
# === 新增：实时分型特征 ===
|
|
# 将枚举转换为数值：UNKNOWN=0, TOP=1, BOTTOM=-1
|
|
if self.fx_type == Chan_FX_TYPE.TOP:
|
|
fx_type_value = 1
|
|
elif self.fx_type == Chan_FX_TYPE.BOTTOM:
|
|
fx_type_value = -1
|
|
else:
|
|
fx_type_value = 0
|
|
|
|
features['klu_fx_type'] = fx_type_value
|
|
features['klu_fx_strength'] = self.fx_strength
|
|
features['klu_fx_confirmed'] = 1 if self.fx_confirmed else 0
|
|
|
|
return features |