From b15ba06a009f587d2e9d3e00e78ae93ebcba4f78 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Mon, 18 Aug 2025 01:14:36 +0800 Subject: [PATCH] =?UTF-8?q?add=20macd=E5=88=A4=E6=96=AD=E8=83=8C=E7=A6=BB?= =?UTF-8?q?=EF=BC=8C=E4=BD=BF=E7=94=A8=E5=88=86=E5=9E=8B=E5=92=8C=E9=9A=90?= =?UTF-8?q?=E5=BD=A2=E5=BD=A2=E6=80=81=E5=AE=9E=E7=8E=B0=E7=AC=AC=E4=BA=8C?= =?UTF-8?q?=E7=B1=BB=E4=B9=B0=E5=8D=96=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ChanEnum.py | 35 +- ChanKLC.py | 15 +- ChanKLU.py | 1416 +++++++++++++++++++----------------- ChanMACD.py | 129 ++-- ChanMACDHistSet.py | 11 +- ChanMACDSeg.py | 20 +- ChanMACDUnitTF.py | 26 +- K线动能理论.txt | 4 +- config/ChanLun_MACD.json | 83 +++ strategies/ChanLun_MACD.py | 346 +++++++++ web/app.py | 270 ++++++- web/templates/index.html | 883 ++++++++++++++++++++-- 12 files changed, 2422 insertions(+), 816 deletions(-) create mode 100644 config/ChanLun_MACD.json create mode 100644 strategies/ChanLun_MACD.py diff --git a/ChanEnum.py b/ChanEnum.py index 76a7b6a..cc6de9c 100644 --- a/ChanEnum.py +++ b/ChanEnum.py @@ -67,26 +67,45 @@ class Chan_KLC_FX(Enum): TOP2 = auto() TOP3 = auto() TOP4 = auto() + TOP5 = auto() BOTTOM1 = auto() BOTTOM2 = auto() BOTTOM3 = auto() BOTTOM4 = auto() + BOTTOM5 = auto() UNKNOWN = auto() +# 统一的MACD状态枚举，包含所有可能的状态 class Chan_MACD_STATE(Enum): - GW = auto() - GWK = auto() - UP = auto() - DOWN = auto() - CROSS0 = auto() - UNKNOWN = auto() - START = auto() - NEAR0 = auto() + """MACD状态枚举 - 包含所有可能的状态""" + # 穿越状态 + CROSS0_UP = auto() # 向上穿越零轴 + CROSS0_DOWN = auto() # 向下穿越零轴 + + # 趋势状态 + NEAR0 = auto() # 接近零轴 + + # 位置状态 + HIGH = auto() # 高位：MACD黄白线离开零轴到高点，能量柱最大开始减弱 + HIGH_EMPTY = auto() # 高位空：MACD黄白线处于高位，能量柱衰减，与黄白线形成空间夹角 + RETURN_ZERO = auto() # 归零轴：能量柱呈现一根比一根短的排列方式 + UP = auto() # 归零轴后的上涨 + DOWN = auto() # 归零轴后的下跌 + # 基础状态 + UNKNOWN = auto() # 未知 + START = auto() # 开始 class Chan_MACDSEG_DIR(Enum): ABOVE = auto() UNDER = auto() +class Chan_MACDUNITTF_TYPE(Enum): + START = auto() + CROSS0 = auto() + NEAR0 = auto() class Chan_MACDHISTSET_DIR(Enum): ABOVE = auto() UNDER = auto() +class Chan_MACDUNITTF_DIR(Enum): + ABOVE = auto() + UNDER = auto() class Chan_MACDHIST_STATE(Enum): UP = auto() DOWN = auto() diff --git a/ChanKLC.py b/ChanKLC.py index 68f004f..530c4bd 100644 --- a/ChanKLC.py +++ b/ChanKLC.py @@ -30,7 +30,7 @@ class ChanKLC(): self.klc_fx_type = Chan_KLC_FX.UNKNOWN self.rsi = klu.rsi self.volume_ratio = klu.volume_ratio - self.macdhist = 0 + self.macdhist = klu.macdhist self.body = klu.body self.upper_shadow = klu.upper_shadow self.lower_shadow = klu.lower_shadow @@ -65,25 +65,30 @@ class ChanKLC(): self.cal_bb_out() if self.pre: self.pre.cal_bb_out() - #self.bb_out = True def cal_bb_out(self): for klu in self.klus: if self.high >= klu.bbup302 and klu.bbup302 > 0 and (self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2): self.bb_out = True #print(self.end_time, self.high, klu.bbup302, self.klc_fx_type) if self.high >= klu.bbup30 and klu.bbup30 > 0 and self.next and (self.next.macd - self.macd) < 0: - self.klc_fx_type = Chan_KLC_FX.TOP4 + self.klc_fx_type = Chan_KLC_FX.TOP4 if self.low <= klu.bblow302 and klu.bblow302 > 0 and (self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2): self.bb_out = True if self.low <= klu.bblow30 and klu.bblow30 > 0 and self.next and (self.macd - self.next.macd) < 0: - self.klc_fx_type = Chan_KLC_FX.BOTTOM4 - if self.fx ==Chan_FX_TYPE.TOP: + self.klc_fx_type = Chan_KLC_FX.BOTTOM4 + if self.fx == Chan_FX_TYPE.TOP: + #print(self.end_time, self.fx, self.macd, self.macdhist, len(self.klus)) if self.macd > 0: self.bb_out = True + if self.macdhist < 0: + self.klc_fx_type = Chan_KLC_FX.TOP5 else: if self.fx == Chan_FX_TYPE.BOTTOM: if self.macd < 0: self.bb_out = True + if self.macdhist > 0: + self.klc_fx_type = Chan_KLC_FX.BOTTOM5 + #self.bb_out = True def cal_macd_state(self, dir): macd_state = 0 return macd_state diff --git a/ChanKLU.py b/ChanKLU.py index c92cc38..352744d 100644 --- a/ChanKLU.py +++ b/ChanKLU.py @@ -1,675 +1,745 @@ from ChanEnum import Chan_FX_TYPE, Chan_KLU_TYPE, Chan_K_DIR, Chan_MACD_STATE, Chan_MACDHIST_STATE 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.bbp120 = 0 - self.bbp365 = 0 - self.bb120 = 0 - self.bb365 = 0 - self.bbp302 = 0 - self.bbup302 = 0 - self.bblow302 = 0 - self.bbup30 = 0 - self.bblow30 = 0 - # === 新增：K线类型 === - self.kline_type = None # K线类型：大阳线、大阴线、小阳线、小阴线 - - # === 新增：实时分型相关属性 === - 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 # 分型是否确认 - self.klu_type = None - self.cal_klu_min_max() - self.range = self.high - self.low - self.body = abs(self.close - self.open) - self.upper_shadow = self.high - max(self.close, self.open) - self.lower_shadow = min(self.close, self.open) - self.low - self.body_ratio = self.body / self.range - self.upper_shadow_ratio = self.upper_shadow / self.body - self.lower_shadow_ratio = self.lower_shadow / self.body - self.candle_dir = Chan_K_DIR.CROSS if self.close == self.open else Chan_K_DIR.BULL if self.close > self.open else Chan_K_DIR.BEAR - self.strength = 0 if self.candle_dir == Chan_K_DIR.CROSS else self.cal_klu_strength() - - self.ema52 = 0 - self.ema24 = 0 - self.macd_slop = 0 - self.signal_slop = 0 - self.hist_slop = 0 - self.hist_state = Chan_MACDHIST_STATE.UNKNOWN - self.macd_state = Chan_MACD_STATE.UNKNOWN - self.macd_hist_gap = 0 - #print(self.open, self.close, self.high, self.low, self.candle_dir, self.strength) - def cal_klu_strength(self): - strength = 0 - if range != 0: - if self.candle_dir == Chan_K_DIR.BULL and (self.upper_shadow + self.lower_shadow) != 0: - strength += self.body / self.range - strength += self.body / (self.upper_shadow + self.lower_shadow) - elif self.candle_dir == Chan_K_DIR.BEAR and (self.upper_shadow + self.lower_shadow) != 0: - strength -=self.body / self.range - strength -= self.body / (self.upper_shadow + self.lower_shadow) - #print(self.time, self.body, self.range, self.upper_shadow, self.lower_shadow, self.candle_dir, strength) - return strength - return strength - def cal_klu_min_max(self): - """ - 计算K线类型：大阳线、大阴线、小阳线、小阴线 - """ - if self.open <= 0: # 避免除零错误 - self.kline_type = None - return - - # 计算涨跌幅 - price_change_ratio = (self.close - self.open) / self.open - strength = 0 - # 判断K线类型 - if price_change_ratio > 0.005: # 涨幅超过2% - self.kline_type = Chan_KLU_TYPE.BigBull - strength += abs(price_change_ratio) - elif price_change_ratio > 0: # 涨幅0-2% - self.kline_type = Chan_KLU_TYPE.SmallBull - strength += abs(price_change_ratio) - elif price_change_ratio < -0.005: # 跌幅超过2% - self.kline_type = Chan_KLU_TYPE.BigBear - strength += abs(price_change_ratio) - elif price_change_ratio < 0: # 跌幅0-2% - self.kline_type = Chan_KLU_TYPE.SmallBear - strength += abs(price_change_ratio) - else: # 开盘价等于收盘价 - self.kline_type = Chan_KLU_TYPE.Cross - strength += abs(price_change_ratio) - return strength - 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 cal_fx(self): - """ - 根据缠论经典规则计算分型强弱 - 返回分型强度：3=极强，2=强，1=中等，0=弱，-1=极弱 - """ - if self.fx_type == Chan_FX_TYPE.UNKNOWN or not self.pre or not self.next: - return 0 - - if self.fx_type == Chan_FX_TYPE.TOP: - return self._cal_top_fx_strength() - else: # BOTTOM - return self._cal_bottom_fx_strength() - - def _check_contain_relation(self, k1, k2): - """检查两根K线是否存在包含关系""" - return (k1.high >= k2.high and k1.low <= k2.low) or (k2.high >= k1.high and k2.low <= k1.low) - - def _is_big_yang_line(self, klu): - """判断是否为大阳线""" - return klu.close > klu.open and (klu.close - klu.open) / klu.open > 0.02 - - def _is_big_yin_line(self, klu): - """判断是否为大阴线""" - return klu.close < klu.open and (klu.open - klu.close) / klu.open > 0.02 - - def _is_small_line(self, klu): - """判断是否为小K线""" - return abs(klu.close - klu.open) / klu.open < 0.01 - - def _has_long_upper_shadow(self, klu): - """判断是否有长上影线""" - body_size = abs(klu.close - klu.open) - upper_shadow = klu.high - max(klu.close, klu.open) - return upper_shadow > body_size * 1.5 - - def _cal_top_fx_strength(self): - """计算顶分型强度""" - strength = 0 - k1, k2, k3 = self.pre, self, self.next - - # (1) 检查包含关系 - 没有包含关系加分 - has_contain_12 = self._check_contain_relation(k1, k2) - has_contain_23 = self._check_contain_relation(k2, k3) - - if not has_contain_12 and not has_contain_23: - strength += 1 # 完全没有包含关系，加1分 - - # (2) 检查第1条K线是大阳线，第2、3条是小K线的情况 - if self._is_big_yang_line(k1) and self._is_small_line(k2) and self._is_small_line(k3): - strength -= 2 # 中继顶分型特征，减2分 - - # (3) 检查第2条K线有长上影线或大阴线，且第3条K线条件 - k2_mid = (k2.high + k2.low) / 2 - k3_is_yang = k3.close > k3.open - k3_close_above_mid = k3.close > k2_mid - - if (self._has_long_upper_shadow(k2) or self._is_big_yin_line(k2)) and not (k3_is_yang and k3_close_above_mid): - strength += 2 # 力度大的顶分型，加2分 - - # (4) 检查第2、3条K线包含关系，第3条为大阴线"吃掉"第2条 - if has_contain_23 and self._is_big_yin_line(k3) and k3.low < k2.low and k3.high < k2.high: - strength += 1 # 最坏包含关系，但对顶分型有利，加1分 - - # (5) 第3条K线跌破第1条K线底部且不能高于第1条K线区间一半之上 - k1_mid = (k1.high + k1.low) / 2 - if k3.low < k1.low and k3.high < k1_mid: - strength -= 1 # 较弱的顶分型，减1分 - - # 额外检查：第3条K线收盘价相对第1条K线的位置 - if k3.close < k1.low: - strength += 1 # 强烈下跌确认，加1分 - - return max(-1, min(3, strength)) # 限制在-1到3范围内 - - def _cal_bottom_fx_strength(self): - """计算底分型强度""" - strength = 0 - k1, k2, k3 = self.pre, self, self.next - - # 底分型上边沿 - fx_top = max(k1.high, k2.high) - - # (1) 第3条K线高点远高于第1条K线高点 - if k3.high > k1.high * 1.02: # 高出2%以上认为是"远高于" - strength += 2 # 较强走势，加2分 - - # (2) 第3条K线高点正好是第1根K线高点，或略微高于底分型上边沿 - elif k1.high * 0.99 <= k3.high <= fx_top * 1.01: # 在合理范围内 - strength += 0 # 一般走势，不加分也不减分 - - # (3) 第3条K线高点低于第1条K线高点 - elif k3.high < k1.high: - strength -= 1 # 较弱走势，减1分 - - # 检查包含关系 - has_contain_12 = self._check_contain_relation(k1, k2) - has_contain_23 = self._check_contain_relation(k2, k3) - - if not has_contain_12 and not has_contain_23: - strength += 1 # 完全没有包含关系，加1分 - - # 检查第3条K线是否为强阳线 - if self._is_big_yang_line(k3): - strength += 1 # 强阳线确认，加1分 - - # (4) 检查后续第1条K线（如果存在） - if hasattr(k3, 'next') and k3.next: - next_k = k3.next - if next_k.low > fx_top: - strength += 2 # 后续K线低点高于底分型上边沿，强烈确认，加2分 - elif next_k.low <= k2.low: - strength -= 1 # 后续K线跌破分型低点，减1分 - - return max(-1, min(3, strength)) # 限制在-1到3范围内 - - 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) - #self.fx_strength = self.cal_fx() - 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.bbp120 = float(item['bbp120']) if 'bbp120' in item and item['bbp120'] else 0 - self.bbp365 = float(item['bbp365']) if 'bbp365' in item and item['bbp365'] else 0 - self.bb120 = float(item['bb120']) if 'bb120' in item and item['bb120'] else 0 - self.bb365 = float(item['bb365']) if 'bb365' in item and item['bb365'] else 0 - self.bbp30 = float(item['bbp30']) if 'bbp30' in item and item['bbp30'] else 0 - self.bbup30 = float(item['bbup30']) if 'bbup30' in item and item['bbup30'] else 0 - self.bblow30 = float(item['bblow30']) if 'bblow30' in item and item['bblow30'] else 0 - self.bbp302 = float(item['bbp302']) if 'bbp302' in item and item['bbp302'] else 0 - self.bbup302 = float(item['bbup302']) if 'bbup302' in item and item['bbup302'] else 0 - self.bblow302 = float(item['bblow302']) if 'bblow302' in item and item['bblow302'] else 0 - self.bbup120 = float(item['bbup120']) if 'bbup120' in item and item['bbup120'] else 0 - self.bblow120 = float(item['bblow120']) if 'bblow120' in item and item['bblow120'] else 0 - self.bbup365 = float(item['bbup365']) if 'bbup365' in item and item['bbup365'] else 0 - self.bblow365 = float(item['bblow365']) if 'bblow365' in item and item['bblow365'] else 0 - # 设置指标后更新实时分析 - self.update_realtime_analysis() - self.cal_macd_state() - def cal_macd_state(self): - if self.macd == 0 and self.signal == 0 and self.macdhist == 0: - return Chan_MACD_STATE.UNKNOWN - if self.pre and self.pre.macd_state != Chan_MACD_STATE.UNKNOWN: - self.macd_slop = self.macd - self.pre.macd - self.signal_slop = self.signal - self.pre.signal - self.hist_slop = self.macdhist - self.pre.macdhist - self.macd_hist_gap = abs(self.macdhist - self.macd) - if self.pre.signal >= 0 and self.signal < 0 and self.ema52 > self.close and self.next and self.next.signal < 0: - self.macd_state = Chan_MACD_STATE.CROSS0 - elif self.pre.signal <= 0 and self.signal > 0 and self.ema52 < self.close and self.next and self.next.signal > 0: - self.macd_state = Chan_MACD_STATE.CROSS0 - elif self.macd > 0 and self.signal > 0 and self.macdhist > 0: - if self.signal > self.macdhist: - self.macd_state = Chan_MACD_STATE.GW - if self.macdhist < self.pre.macdhist and self.macd_slop > 0 and self.signal_slop > 0 and self.macd_slop < self.pre.macd_slop and self.signal_slop < self.pre.signal_slop and self.macd_hist_gap > self.pre.macd_hist_gap: - self.macd_state = Chan_MACD_STATE.GWK - elif self.macd_slop > 0 and self.signal_slop > 0: - self.macd_state = Chan_MACD_STATE.UP - elif self.macd < 0 and self.signal < 0 and self.macdhist < 0: - if self.signal < self.macdhist: - self.macd_state = Chan_MACD_STATE.GW - if self.macdhist > self.pre.macdhist and self.macd_slop < 0 and self.signal_slop < 0 and self.macd_slop > self.pre.macd_slop and self.signal_slop > self.pre.signal_slop and self.macd_hist_gap > self.pre.macd_hist_gap: - self.macd_state = Chan_MACD_STATE.GWK - elif self.macd_slop < 0 and self.signal_slop < 0: - self.macd_state = Chan_MACD_STATE.DOWN - else: - self.macd_state = Chan_MACD_STATE.START - return self.macd_state - - 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 \ No newline at end of file + 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.bbp120 = 0 + self.bbp365 = 0 + self.bb120 = 0 + self.bb365 = 0 + self.bbp302 = 0 + self.bbup302 = 0 + self.bblow302 = 0 + self.bbup30 = 0 + self.bblow30 = 0 + # === 新增：K线类型 === + self.kline_type = None # K线类型：大阳线、大阴线、小阳线、小阴线 + + # === 新增：实时分型相关属性 === + 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 # 分型是否确认 + self.klu_type = None + self.cal_klu_min_max() + self.range = self.high - self.low + self.body = abs(self.close - self.open) + self.upper_shadow = self.high - max(self.close, self.open) + self.lower_shadow = min(self.close, self.open) - self.low + self.body_ratio = self.body / self.range + self.upper_shadow_ratio = self.upper_shadow / self.body + self.lower_shadow_ratio = self.lower_shadow / self.body + self.candle_dir = Chan_K_DIR.CROSS if self.close == self.open else Chan_K_DIR.BULL if self.close > self.open else Chan_K_DIR.BEAR + self.strength = 0 if self.candle_dir == Chan_K_DIR.CROSS else self.cal_klu_strength() + + self.ema52 = 0 + self.ema24 = 0 + self.macd_slop = 0 + self.signal_slop = 0 + self.hist_slop = 0 + self.hist_state = Chan_MACDHIST_STATE.UNKNOWN + self.macd_state = Chan_MACD_STATE.UNKNOWN + self.macd_hist_gap = 0 + # === 归零轴细化与模式/背离 === + self.zero_axis = False # 是否归零轴（穿越或接近） + self.zero_axis_state = "none" # {none,crossing,near} + self.zero_axis_side = 0 # 1:above, -1:under, 0:none + self.zero_axis_score = 0 # 0-100 综合评分 + self.mode1_touch_ema52 = False # 单边后触碰EMA52 + self.mode2_fast_to_zero = False # 快线向零收敛 + self.mode3_double_tf = False # 双周期归零（近似占位，由上层填充高周期确认） + self.mode3_dir = "none" # {long_strong_rebound, short_strong_rebound, none} + self.mode4_touch52_no_zero = False # 先触碰EMA52但黄白线未归零 + self.div_type = "none" # {bearish, bullish, hidden_bearish, hidden_bullish, none} + self.div_score = 0.0 # 背离强度(0-100) + #print(self.open, self.close, self.high, self.low, self.candle_dir, self.strength) + def cal_klu_strength(self): + strength = 0 + if range != 0: + if self.candle_dir == Chan_K_DIR.BULL and (self.upper_shadow + self.lower_shadow) != 0: + strength += self.body / self.range + strength += self.body / (self.upper_shadow + self.lower_shadow) + elif self.candle_dir == Chan_K_DIR.BEAR and (self.upper_shadow + self.lower_shadow) != 0: + strength -=self.body / self.range + strength -= self.body / (self.upper_shadow + self.lower_shadow) + #print(self.time, self.body, self.range, self.upper_shadow, self.lower_shadow, self.candle_dir, strength) + return strength + return strength + def cal_klu_min_max(self): + """ + 计算K线类型：大阳线、大阴线、小阳线、小阴线 + """ + if self.open <= 0: # 避免除零错误 + self.kline_type = None + return + + # 计算涨跌幅 + price_change_ratio = (self.close - self.open) / self.open + strength = 0 + # 判断K线类型 + if price_change_ratio > 0.005: # 涨幅超过2% + self.kline_type = Chan_KLU_TYPE.BigBull + strength += abs(price_change_ratio) + elif price_change_ratio > 0: # 涨幅0-2% + self.kline_type = Chan_KLU_TYPE.SmallBull + strength += abs(price_change_ratio) + elif price_change_ratio < -0.005: # 跌幅超过2% + self.kline_type = Chan_KLU_TYPE.BigBear + strength += abs(price_change_ratio) + elif price_change_ratio < 0: # 跌幅0-2% + self.kline_type = Chan_KLU_TYPE.SmallBear + strength += abs(price_change_ratio) + else: # 开盘价等于收盘价 + self.kline_type = Chan_KLU_TYPE.Cross + strength += abs(price_change_ratio) + return strength + 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 set_histset(self, histset): + """设置HistSet关联""" + self.histset = histset + + def set_seg(self, seg): + """设置Seg关联""" + self.seg = seg + + def set_unittf(self, unittf): + """设置UnitTF关联""" + self.unittf = unittf + 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 cal_fx(self): + """ + 根据缠论经典规则计算分型强弱 + 返回分型强度：3=极强，2=强，1=中等，0=弱，-1=极弱 + """ + if self.fx_type == Chan_FX_TYPE.UNKNOWN or not self.pre or not self.next: + return 0 + + if self.fx_type == Chan_FX_TYPE.TOP: + return self._cal_top_fx_strength() + else: # BOTTOM + return self._cal_bottom_fx_strength() + + def _check_contain_relation(self, k1, k2): + """检查两根K线是否存在包含关系""" + return (k1.high >= k2.high and k1.low <= k2.low) or (k2.high >= k1.high and k2.low <= k1.low) + + def _is_big_yang_line(self, klu): + """判断是否为大阳线""" + return klu.close > klu.open and (klu.close - klu.open) / klu.open > 0.02 + + def _is_big_yin_line(self, klu): + """判断是否为大阴线""" + return klu.close < klu.open and (klu.open - klu.close) / klu.open > 0.02 + + def _is_small_line(self, klu): + """判断是否为小K线""" + return abs(klu.close - klu.open) / klu.open < 0.01 + + def _has_long_upper_shadow(self, klu): + """判断是否有长上影线""" + body_size = abs(klu.close - klu.open) + upper_shadow = klu.high - max(klu.close, klu.open) + return upper_shadow > body_size * 1.5 + + def _cal_top_fx_strength(self): + """计算顶分型强度""" + strength = 0 + k1, k2, k3 = self.pre, self, self.next + + # (1) 检查包含关系 - 没有包含关系加分 + has_contain_12 = self._check_contain_relation(k1, k2) + has_contain_23 = self._check_contain_relation(k2, k3) + + if not has_contain_12 and not has_contain_23: + strength += 1 # 完全没有包含关系，加1分 + + # (2) 检查第1条K线是大阳线，第2、3条是小K线的情况 + if self._is_big_yang_line(k1) and self._is_small_line(k2) and self._is_small_line(k3): + strength -= 2 # 中继顶分型特征，减2分 + + # (3) 检查第2条K线有长上影线或大阴线，且第3条K线条件 + k2_mid = (k2.high + k2.low) / 2 + k3_is_yang = k3.close > k3.open + k3_close_above_mid = k3.close > k2_mid + + if (self._has_long_upper_shadow(k2) or self._is_big_yin_line(k2)) and not (k3_is_yang and k3_close_above_mid): + strength += 2 # 力度大的顶分型，加2分 + + # (4) 检查第2、3条K线包含关系，第3条为大阴线"吃掉"第2条 + if has_contain_23 and self._is_big_yin_line(k3) and k3.low < k2.low and k3.high < k2.high: + strength += 1 # 最坏包含关系，但对顶分型有利，加1分 + + # (5) 第3条K线跌破第1条K线底部且不能高于第1条K线区间一半之上 + k1_mid = (k1.high + k1.low) / 2 + if k3.low < k1.low and k3.high < k1_mid: + strength -= 1 # 较弱的顶分型，减1分 + + # 额外检查：第3条K线收盘价相对第1条K线的位置 + if k3.close < k1.low: + strength += 1 # 强烈下跌确认，加1分 + + return max(-1, min(3, strength)) # 限制在-1到3范围内 + + def _cal_bottom_fx_strength(self): + """计算底分型强度""" + strength = 0 + k1, k2, k3 = self.pre, self, self.next + + # 底分型上边沿 + fx_top = max(k1.high, k2.high) + + # (1) 第3条K线高点远高于第1条K线高点 + if k3.high > k1.high * 1.02: # 高出2%以上认为是"远高于" + strength += 2 # 较强走势，加2分 + + # (2) 第3条K线高点正好是第1根K线高点，或略微高于底分型上边沿 + elif k1.high * 0.99 <= k3.high <= fx_top * 1.01: # 在合理范围内 + strength += 0 # 一般走势，不加分也不减分 + + # (3) 第3条K线高点低于第1条K线高点 + elif k3.high < k1.high: + strength -= 1 # 较弱走势，减1分 + + # 检查包含关系 + has_contain_12 = self._check_contain_relation(k1, k2) + has_contain_23 = self._check_contain_relation(k2, k3) + + if not has_contain_12 and not has_contain_23: + strength += 1 # 完全没有包含关系，加1分 + + # 检查第3条K线是否为强阳线 + if self._is_big_yang_line(k3): + strength += 1 # 强阳线确认，加1分 + + # (4) 检查后续第1条K线（如果存在） + if hasattr(k3, 'next') and k3.next: + next_k = k3.next + if next_k.low > fx_top: + strength += 2 # 后续K线低点高于底分型上边沿，强烈确认，加2分 + elif next_k.low <= k2.low: + strength -= 1 # 后续K线跌破分型低点，减1分 + + return max(-1, min(3, strength)) # 限制在-1到3范围内 + + 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) + #self.fx_strength = self.cal_fx() + 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.bbp120 = float(item['bbp120']) if 'bbp120' in item and item['bbp120'] else 0 + self.bbp365 = float(item['bbp365']) if 'bbp365' in item and item['bbp365'] else 0 + self.bb120 = float(item['bb120']) if 'bb120' in item and item['bb120'] else 0 + self.bb365 = float(item['bb365']) if 'bb365' in item and item['bb365'] else 0 + self.bbp30 = float(item['bbp30']) if 'bbp30' in item and item['bbp30'] else 0 + self.bbup30 = float(item['bbup30']) if 'bbup30' in item and item['bbup30'] else 0 + self.bblow30 = float(item['bblow30']) if 'bblow30' in item and item['bblow30'] else 0 + self.bbp302 = float(item['bbp302']) if 'bbp302' in item and item['bbp302'] else 0 + self.bbup302 = float(item['bbup302']) if 'bbup302' in item and item['bbup302'] else 0 + self.bblow302 = float(item['bblow302']) if 'bblow302' in item and item['bblow302'] else 0 + self.bbup120 = float(item['bbup120']) if 'bbup120' in item and item['bbup120'] else 0 + self.bblow120 = float(item['bblow120']) if 'bblow120' in item and item['bblow120'] else 0 + self.bbup365 = float(item['bbup365']) if 'bbup365' in item and item['bbup365'] else 0 + self.bblow365 = float(item['bblow365']) if 'bblow365' in item and item['bblow365'] else 0 + # 设置指标后更新实时分析 + self.update_realtime_analysis() + #self.cal_macd_state() + def cal_macd_state(self): + # 按定义精简实现：优先级 CROSS0 > 位置(HIGH/HE/RETURN_ZERO) > NEAR0 > UNKNOWN + # 1) 首条或缺前一根 + if not hasattr(self, 'pre') or self.pre is None: + self.macd_state = Chan_MACD_STATE.START + return self.macd_state + + # 2) 基本校验 + if self.macd == 0 and self.signal == 0 and self.macdhist == 0: + self.macd_state = Chan_MACD_STATE.UNKNOWN + return self.macd_state + + # 调试：当前值 + #print(f"DEBUG: MACD值检查 - 时间: {self.time}, MACD: {self.macd:.4f}, Signal: {self.signal:.4f}, 前一根Signal: {self.pre.signal:.4f}") + + # 3) CROSS0 仅以 Signal 穿越零轴判定 + if self.pre.signal >= 0 and self.signal < 0: + self.macd_state = Chan_MACD_STATE.CROSS0_DOWN + #print(f"DEBUG: CROSS0触发(向下) - 时间: {self.time}, Signal: {self.pre.signal:.4f}->{self.signal:.4f}") + return self.macd_state + if self.pre.signal <= 0 and self.signal > 0: + self.macd_state = Chan_MACD_STATE.CROSS0_UP + #print(f"DEBUG: CROSS0触发(向上) - 时间: {self.time}, Signal: {self.pre.signal:.4f}->{self.signal:.4f}") + return self.macd_state + + # 4) 高位状态的位置状态， 高位，高位空，归零轴 + position_state = Chan_MACD_STATE.UNKNOWN + if abs(self.macd) > 50: + position_state = self._cal_position_state() + if position_state != Chan_MACD_STATE.UNKNOWN: + self.macd_state = position_state + #print(f"DEBUG: 位置状态设置 - 时间: {self.time}, 状态: {self.macd_state}, MACD: {self.macd:.4f}, Signal: {self.signal:.4f}") + return self.macd_state + + # 5) 趋近零轴：任一线接近零轴即可，无限接近零轴，EMA52没有加 + NEAR0_EPS = 15 + if abs(self.macd) <= NEAR0_EPS or abs(self.signal) <= NEAR0_EPS: + self.macd_state = Chan_MACD_STATE.NEAR0 + #print(f"DEBUG: 趋近零轴触发 - 时间: {self.time}, MACD: {self.macd:.4f}, Signal: {self.signal:.4f}") + return self.macd_state + # 6） 离开0轴开始上涨或者下跌阶段，高位之前的 + if self.macd_state == Chan_MACD_STATE.UNKNOWN: + if self.macd > 0 and self.pre: + if (self.pre.macd_state == Chan_MACD_STATE.NEAR0 or self.pre.macd_state == Chan_MACD_STATE.UP) and self.signal > self.pre.signal: + self.macd_state = Chan_MACD_STATE.UP + return self.macd_state + elif self.macd < 0 and self.pre: + if (self.pre.macd_state == Chan_MACD_STATE.NEAR0 or self.pre.macd_state == Chan_MACD_STATE.DOWN) and self.signal < self.pre.signal: + self.macd_state = Chan_MACD_STATE.DOWN + return self.macd_state + # 7) 其余情况 + self.macd_state = Chan_MACD_STATE.UNKNOWN + return self.macd_state + + def _cal_position_state(self): + """计算MACD位置状态：高位、高位空、归零轴""" + if not hasattr(self, 'pre') or self.pre is None: + return Chan_MACD_STATE.UNKNOWN + pre_hist = abs(self.pre.macdhist) + cur_hist = abs(self.macdhist) + if self.next: + next_hist = abs(self.next.macdhist) + if cur_hist > pre_hist and cur_hist > next_hist: + return Chan_MACD_STATE.HIGH + if self.pre.macd_state == Chan_MACD_STATE.HIGH or self.pre.macd_state == Chan_MACD_STATE.HIGH_EMPTY: + if self.macd > 0: + if self.macd > self.signal: + return Chan_MACD_STATE.HIGH_EMPTY + else: + return Chan_MACD_STATE.RETURN_ZERO + elif self.macd < 0: + if self.macd < self.signal: + return Chan_MACD_STATE.HIGH_EMPTY + else: + return Chan_MACD_STATE.RETURN_ZERO + return Chan_MACD_STATE.UNKNOWN + 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 \ No newline at end of file diff --git a/ChanMACD.py b/ChanMACD.py index 76b1fbc..da1cdee 100644 --- a/ChanMACD.py +++ b/ChanMACD.py @@ -1,5 +1,5 @@ from ChanKLU import ChanKLU -from ChanEnum import Chan_MACD_STATE, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR +from ChanEnum import Chan_MACD_STATE, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR, Chan_MACDUNITTF_DIR, Chan_MACDUNITTF_TYPE from ChanMACDSeg import ChanMACDSeg from ChanMACDUnitTF import ChanMACDUnitTF from ChanMACDHistSet import ChanMACDHistSet @@ -10,6 +10,12 @@ class ChanMACD(): self.seg_list = [] self.unittf_list = [] self.histset_list = [] + # 状态标记列表 + self.high_position_list = [] # 高位列表 + self.high_empty_list = [] # 高位空列表 + self.return_zero_list = [] # 归零轴列表 + self.cross0_up_list = [] # 向上穿越零轴列表 + self.cross0_down_list = [] # 向下穿越零轴列表 self.cal_macd() def cal_macd(self): last_seg = None @@ -17,68 +23,98 @@ class ChanMACD(): last_histset = None last_klu = None - for klu in self.klu_list: + for klu in self.klu_list: klu.cal_macd_state() - # 1) 只有当 MACD 已可用(非 UNKNOWN)时，才开始初始化段/单元 if last_seg is None: if klu.macd_state != Chan_MACD_STATE.UNKNOWN: - # 初始化首段与首单元 - seg_dir = Chan_MACDSEG_DIR.ABOVE if klu.macd >= 0 else Chan_MACDSEG_DIR.UNDER + # 初始化首个直方图集合（根据当前柱体正负） + if klu.macdhist >= 0: + histset = ChanMACDHistSet(klu.time, klu, None, Chan_MACDHISTSET_DIR.ABOVE) + else: + histset = ChanMACDHistSet(klu.time, klu, None, Chan_MACDHISTSET_DIR.UNDER) + self.histset_list.append(histset) + last_histset = histset + + # 初始化首段 + seg_dir = Chan_MACDSEG_DIR.ABOVE if klu.signal >= 0 else Chan_MACDSEG_DIR.UNDER seg = ChanMACDSeg(klu.time, klu, None, seg_dir) self.seg_list.append(seg) last_seg = seg - - unittf = ChanMACDUnitTF(klu.time, klu, None, None) - self.unittf_list.append(unittf) - last_unittf = unittf - - # 初始化首个直方图集合（根据当前柱体正负） - if klu.macdhist >= 0: - histset = ChanMACDHistSet(klu.time, klu, last_unittf, Chan_MACDHISTSET_DIR.ABOVE) - else: - histset = ChanMACDHistSet(klu.time, klu, last_unittf, Chan_MACDHISTSET_DIR.UNDER) - self.histset_list.append(histset) - last_histset = histset last_seg.add_histset(histset) + + # 初始化首个UnitTF + unittf_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER + unittf = ChanMACDUnitTF(klu.time, klu, None, unittf_dir, Chan_MACDUNITTF_TYPE.START) + self.unittf_list.append(unittf) + unittf.add_histset(histset) + last_unittf = unittf last_unittf.add_histset(histset) last_seg.add_unittf(unittf) # 未就绪则继续等下一根；已就绪亦已完成首个结构初始化，继续下一根 last_klu = klu continue - # 2) 过零切段/切单元 - if klu.macd_state == Chan_MACD_STATE.CROSS0: + # 2) 过零切段（使用KLU中的穿越状态） + if (klu.macd_state == Chan_MACD_STATE.CROSS0_UP or + klu.macd_state == Chan_MACD_STATE.CROSS0_DOWN): + # 结束旧 unittf + last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0) + # 新的单位时间周期 + new_dir = Chan_MACDUNITTF_DIR.UNDER if last_unittf.unittf_dir == Chan_MACDUNITTF_DIR.ABOVE else Chan_MACDUNITTF_DIR.ABOVE + unittf = ChanMACDUnitTF(klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.CROSS0) + self.unittf_list.append(unittf) + unittf.add_histset(last_histset) + last_unittf.set_next(unittf) + last_unittf = unittf # 收尾旧段 - last_seg.end_klu = last_klu - last_seg.end_time = last_klu.time + last_seg.set_end_klu(last_klu) # 新段方向取反 new_dir = Chan_MACDSEG_DIR.UNDER if last_seg.seg_dir == Chan_MACDSEG_DIR.ABOVE else Chan_MACDSEG_DIR.ABOVE seg = ChanMACDSeg(klu.time, klu, last_seg, new_dir) self.seg_list.append(seg) last_seg.set_next(seg) last_seg = seg - - # 切换 unit tf（以当前histset收尾为边界） - unittf = ChanMACDUnitTF(klu.time, klu, last_histset, None) - self.unittf_list.append(unittf) - if last_unittf: - last_unittf.end_klu = last_klu - last_unittf.end_time = last_klu.time - last_unittf.set_next(unittf) - last_unittf = unittf - + last_seg.add_histset(last_histset) + last_seg.add_unittf(unittf) + else: + # 4) UnitTF 状态机：用黄线Signal的归零轴 + if last_klu.macd_state == Chan_MACD_STATE.NEAR0 and last_unittf.near0_count > 1: + if klu.macd_state == Chan_MACD_STATE.UP: + last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0) + new_dir = Chan_MACDUNITTF_DIR.ABOVE if last_unittf.unittf_dir == Chan_MACDUNITTF_DIR.UNDER else Chan_MACDUNITTF_DIR.UNDER + unittf = ChanMACDUnitTF(klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0) + self.unittf_list.append(unittf) + unittf.add_histset(last_histset) + last_unittf.set_next(unittf) + last_unittf = unittf + last_seg.add_unittf(unittf) + elif klu.macd_state == Chan_MACD_STATE.DOWN: + last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0) + new_dir = Chan_MACDUNITTF_DIR.UNDER if last_unittf.unittf_dir == Chan_MACDUNITTF_DIR.ABOVE else Chan_MACDUNITTF_DIR.ABOVE + unittf = ChanMACDUnitTF(klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0) + self.unittf_list.append(unittf) + unittf.add_histset(last_histset) + last_unittf.set_next(unittf) + last_unittf = unittf + last_seg.add_unittf(unittf) + else: + last_unittf.add_klu(klu) + last_seg.add_klu(klu) + last_histset.add_klu(klu) + else: + last_unittf.add_klu(klu) + last_seg.add_klu(klu) + last_histset.add_klu(klu) # 3) 直方图集合（基于当前 unittf） if klu.macdhist >= 0: if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE: last_histset.add_klu(klu) else: # 结束旧 histset（以前一根结束更合理） - if last_histset: - end_klu = getattr(klu, 'pre', None) or klu - last_histset.end_klu = end_klu - last_histset.end_time = end_klu.time - histset = ChanMACDHistSet(klu.time, klu, last_unittf, Chan_MACDHISTSET_DIR.ABOVE) + if last_histset and last_klu: + last_histset.set_end_klu(last_klu) + histset = ChanMACDHistSet(klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE) self.histset_list.append(histset) if last_histset: last_histset.set_next(histset) @@ -91,11 +127,10 @@ class ChanMACD(): if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.UNDER: last_histset.add_klu(klu) else: - if last_histset: - end_klu = getattr(klu, 'pre', None) or klu - last_histset.end_klu = end_klu - last_histset.end_time = end_klu.time - histset = ChanMACDHistSet(klu.time, klu, last_unittf, Chan_MACDHISTSET_DIR.UNDER) + # 结束旧 histset（以前一根结束更合理） + if last_histset and last_klu: + last_histset.set_end_klu(last_klu) + histset = ChanMACDHistSet(klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.UNDER) self.histset_list.append(histset) if last_histset: last_histset.set_next(histset) @@ -104,14 +139,4 @@ class ChanMACD(): last_seg.add_histset(histset) if last_unittf: last_unittf.add_histset(histset) - last_klu = klu - # 循环结束后，收尾当前打开的结构 - if last_seg and getattr(last_seg, 'end_klu', None) is None: - last_seg.end_klu = last_klu - last_seg.end_time = last_klu.time - if last_unittf and getattr(last_unittf, 'end_klu', None) is None: - last_unittf.end_klu = last_klu - last_unittf.end_time = last_klu.time - if last_histset and getattr(last_histset, 'end_klu', None) is None: - last_histset.end_klu = last_klu - last_histset.end_time = last_klu.time \ No newline at end of file + last_klu = klu \ No newline at end of file diff --git a/ChanMACDHistSet.py b/ChanMACDHistSet.py index 48802c4..67d6a98 100644 --- a/ChanMACDHistSet.py +++ b/ChanMACDHistSet.py @@ -1,15 +1,20 @@ class ChanMACDHistSet(): - def __init__(self, start_time, start_klu, hist_set_dir, pre_histset): + def __init__(self, start_time, start_klu, pre_histset, dir): + self.start_time = start_time + self.end_time = None self.klu_list = [] self.klu_list.append(start_klu) self.ref_klu = None - self.hist_set_dir = hist_set_dir + self.histset_dir = dir self.next = None self.pre = pre_histset def set_next(self, next_histset): self.next = next_histset def add_klu(self, klu): self.klu_list.append(klu) - klu.set_histset(self) \ No newline at end of file + klu.set_histset(self) + def set_end_klu(self, end_klu): + self.end_klu = end_klu + self.end_time = end_klu.time \ No newline at end of file diff --git a/ChanMACDSeg.py b/ChanMACDSeg.py index 44673ef..8ad5072 100644 --- a/ChanMACDSeg.py +++ b/ChanMACDSeg.py @@ -1,4 +1,4 @@ - +from ChanEnum import Chan_MACDSEG_DIR class ChanMACDSeg(): @@ -14,14 +14,26 @@ class ChanMACDSeg(): self.seg_dir = seg_dir self.pre = pre_seg self.next = None + self.peak_klu = start_klu def set_next(self, next_seg): self.next = next_seg def add_klu(self, klu): - self.klu_list.append(klu) - klu.set_seg(self) + if klu: + self.klu_list.append(klu) + klu.set_seg(self) + if self.seg_dir == Chan_MACDSEG_DIR.ABOVE: + if klu.high > self.peak_klu.high: + self.peak_klu = klu + else: + if klu.low < self.peak_klu.low: + self.peak_klu = klu def add_unittf(self, unittf): self.unittf_list.append(unittf) unittf.set_next(self) def add_histset(self, histset): self.hist_set.append(histset) - histset.set_next(self) \ No newline at end of file + histset.set_next(self) + def set_end_klu(self, end_klu): + self.add_klu(end_klu) + self.end_klu = end_klu + self.end_time = end_klu.time \ No newline at end of file diff --git a/ChanMACDUnitTF.py b/ChanMACDUnitTF.py index 90a48a0..0e18236 100644 --- a/ChanMACDUnitTF.py +++ b/ChanMACDUnitTF.py @@ -1,8 +1,8 @@ - +from ChanEnum import Chan_MACD_STATE class ChanMACDUnitTF(): - def __init__(self, start_time, start_klu, start_histset, pre_unittf, dir): + def __init__(self, start_time, start_klu, pre_unittf, dir, start_type): self.start_time = start_time self.end_time = None self.start_klu = start_klu @@ -10,15 +10,29 @@ class ChanMACDUnitTF(): self.klu_list = [] self.klu_list.append(start_klu) self.histset_list = [] - self.histset_list.append(start_histset) self.next = None self.pre = pre_unittf - self.uinttf_dir = dir + self.unittf_dir = dir + self.start_type = start_type + self.end_type = None + self.peak_hist = start_klu.macdhist + self.near0_count = 1 def set_next(self, next_unittf): self.next = next_unittf def add_histset(self, histset): self.histset_list.append(histset) histset.set_next(self) def add_klu(self, klu): - self.klu_list.append(klu) - klu.set_unittf(self) \ No newline at end of file + if klu: + self.klu_list.append(klu) + klu.set_unittf(self) + if klu.macdhist > self.peak_hist: + self.peak_hist = klu.macdhist + if klu.macd_state == Chan_MACD_STATE.NEAR0 and len(self.histset_list) > 1: + #print(self.start_time, klu.time, self.near0_count) + self.near0_count += 1 + def set_end_klu(self, end_klu, end_type): + self.add_klu(end_klu) + self.end_type = end_type + self.end_klu = end_klu + self.end_time = end_klu.time \ No newline at end of file diff --git a/K线动能理论.txt b/K线动能理论.txt index 3ecfb76..b69a256 100644 --- a/K线动能理论.txt +++ b/K线动能理论.txt @@ -15,9 +15,9 @@ MACD黄白线和零轴的几种形态： 离开零轴 当MACD黄白线穿过零轴那么进入第一阶段离开零轴，此时能量柱变化越来越大，不断增长，k线加速上涨 高位 -当MACD黄白线离开零轴，到一高点时，能量柱此时处于最大，开始减弱时MACD处于高位 +当MACD黄白线离开零轴，到一高点时，能量柱此时处于最大，开始减弱时MACD处于高位，高位时MACD黄白线和能量柱是同向的，高位过后是高位空 高位空 -当MACD黄白线处于高位，随着K线出现缓慢上涨或者横盘整理，MACD黄白线保持高位出现平滑横盘走势，此时，MACD的能量柱出现衰减变化，同时能量柱喝黄白线之间形成一定的空间夹脚，随着能量柱的不断衰减就导致黄白线喝能量柱之间的空间夹脚越来越大，因此就形成高位空 +当MACD黄白线处于高位，随着K线出现缓慢上涨或者横盘整理，MACD黄白线保持高位出现平滑横盘走势，此时，MACD的能量柱出现衰减变化，同时能量柱和黄白线之间形成一定的空间夹脚，随着能量柱的不断衰减就导致黄白线和能量柱之间的空间夹脚越来越大，因此就形成高位空 归零轴 当MACD黄白线在高位，驱动K线上涨的能量所产生的加速度小于或者等于零，K线减速上涨或者下跌，能量变化越来越小，能量柱呈现出一根比一根短的排列方式 穿零轴 diff --git a/config/ChanLun_MACD.json b/config/ChanLun_MACD.json new file mode 100644 index 0000000..69c5739 --- /dev/null +++ b/config/ChanLun_MACD.json @@ -0,0 +1,83 @@ +{ + "$schema": "https://schema.freqtrade.io/schema.json", + "max_open_trades": 1, + "stake_currency": "USDT", + "stake_amount": "unlimited", + "tradable_balance_ratio": 0.99, + "fiat_display_currency": "USD", + "dry_run": true, + "db_url": "sqlite:///tradesv3.chanlun_btc_15.sqlite", + "dry_run_wallet": 1000, + "cancel_open_orders_on_exit": true, + "trading_mode": "futures", + "margin_mode": "isolated", + "can_short" : true, + "timeframe" : "5m", + "process_only_new_candles" : false, + "unfilledtimeout": { + "entry": 1, + "exit": 1, + "exit_timeout_count": 5, + "unit": "minutes" + }, + "entry_pricing": { + "price_side": "other", + "use_order_book": false, + "order_book_top": 1, + "price_last_balance": 0.0, + "check_depth_of_market": { + "enabled": false, + "bids_to_ask_delta": 1 + } + }, + "exit_pricing":{ + "price_side": "other", + "use_order_book": false, + "order_book_top": 1 + }, + "exchange": { + "name": "binance", + "key": "hvoXanRExQvcN4tyGFvEnsSF4gqxXp6ZJnBu5lnhvlVuHaDbj2PhLBQGCLkkyeI8", + "secret": "3UKA2oyDj7OoXrausmnaLwLlNfXmlNf2imBdmQqqKHArcJfk6X9xjaUF19wzu82l", + "ccxt_config": {}, + "ccxt_async_config": {}, + "pair_whitelist": [ + "BTC/USDT:USDT" + ], + "pair_blacklist": [ + "BNB/.*" + ] + }, + "pairlists": [ + { + "method": "StaticPairList", + "number_assets": 1, + "sort_key": "quoteVolume", + "min_value": 0, + "refresh_period": 1800 + } + ], + "telegram": { + "enabled": true, + "token": "7677670958:AAFL_jgZvNUTPR3R3vWieREX_tDVi9w2C1Y", + "chat_id": "580807463" + }, + "api_server": { + "enabled": true, + "listen_ip_address": "127.0.0.1", + "listen_port": 8813, + "verbosity": "error", + "enable_openapi": false, + "jwt_secret_key": "14d3510740e2c39a973a8895f1aa2704d98d08b86170260085709fa5ea48251d", + "ws_token": "dtKKDnafBrX4icq_ZCw7acJTahTK4h_yvg", + "CORS_origins": [], + "username": "freqtrader", + "password": "FreqTrade007" + }, + "bot_name": "freqtrade", + "initial_state": "running", + "force_entry_enable": false, + "internals": { + "process_throttle_secs": 2 + } +} \ No newline at end of file diff --git a/strategies/ChanLun_MACD.py b/strategies/ChanLun_MACD.py new file mode 100644 index 0000000..70fe50b --- /dev/null +++ b/strategies/ChanLun_MACD.py @@ -0,0 +1,346 @@ +# --- Do not remove these libs --- +from statistics import median +from freqtrade.strategy import IStrategy, stoploss_from_absolute +import sys +import os +# 添加父目录到系统路径 +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from ChanLun import ChanLun +from ChanLun_Classifier import ChanLunClassifier +from ChanEnum import Chan_FX_TYPE, Chan_KLC_FX, Chan_BI_DIR, Chan_KLC_FX +from ChanPY import ChanPY +# -------------------------------- +from technical.util import resample_to_interval, resampled_merge +import talib.abstract as ta +import numpy as np +import pandas as pd +from pandas import DataFrame +from datetime import datetime, timedelta +from freqtrade.persistence import Trade, Order +from typing import Optional +import logging +logger = logging.getLogger(__name__) +### Now you can use logger.info('asfd') to log +# freqtrade plot-dataframe --strategy ChanLun_MACD --datadir user_data/data/binance -c ./user_data/ChanLun_MACD.json --timerange=20250309- + +# freqtrade trade -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies +# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies --timerange=20250812- +# freqtrade download-data -c ./user_data/Chan/config/ChanLun_MACD.json -t 5m --pairs BTC/USDT:USDT --timerange=20240101- +# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_MACD.json -e 200 --timerange=20250201-20250401 + +# sudo docker compose run --rm chanlun_btc backtesting -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies --timerange=20250721- +# sudo docker compose run --rm chanlun_btc download-data -c ./user_data/Chan/config/ChanLun_MACD.json --pairs BTC/USDT:USDT -t 1m --timerange 20240101- +# sudo docker compose run --rm chanlun_btc trade -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies + +class ChanLun_MACD(IStrategy): + # 标准 Freqtrade 策略：使用 归零轴 + 背离/隐性形态 进行交易 + INTERFACE_VERSION: int = 3 + + # 基本参数 + timeframe = '5m' + startup_candle_count = 300 + can_short = True + + # ROI/止损（止损由自定义 ATR 控制，此处设大） + minimal_roi = {"0": 0.1} + stoploss = -0.3 + use_custom_stoploss = True + + # 使用市价单，避免回测限价成交不充分导致信号丢单 + order_types = { + "entry": "market", + "exit": "market", + "stoploss": "market", + "stoploss_on_exchange": False, + "stoploss_on_exchange_interval": 60, + } + + # 过滤：ATR 太小不进场 + # 为确保先跑出单，暂不限制 ATR（回测确认后再收紧） + min_atr_value = 0.0 + # 可调参数 + eps_zero_param = 0.06 + div_shift = 2 + zero_recent_lookback = 3 + + # ============ 指标计算 ============ + def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + # MACD + macd = ta.MACD(dataframe) + dataframe['macd'] = macd['macd'] + dataframe['macdsignal'] = macd['macdsignal'] + dataframe['macdhist'] = macd['macdhist'] + + # EMA24/EMA52（若无ema24，使用ema26近似） + dataframe['ema24'] = ta.EMA(dataframe, timeperiod=24) + dataframe['ema52'] = ta.EMA(dataframe, timeperiod=52) + + # ATR + dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) + + # 归零轴判定 + eps_zero = self.eps_zero_param # 可调 + dataframe['zero_cross'] = (dataframe['macd'].shift(1) * dataframe['macd'] <= 0) + dataframe['zero_near'] = (dataframe['macd'].abs() <= eps_zero) | ((dataframe['macd'].abs() <= eps_zero) & (dataframe['macdsignal'].abs() <= eps_zero)) + dataframe['zero_axis'] = dataframe['zero_cross'] | dataframe['zero_near'] + + # 价格触碰均线 + band24 = 0.006 # 放宽贴近阈值 + band52 = 0.008 + dataframe['near_ema24'] = (dataframe['ema24'] > 0) & ((dataframe['close'] - dataframe['ema24']).abs() / dataframe['ema24'] <= band24) + dataframe['near_ema52'] = (dataframe['ema52'] > 0) & ((dataframe['close'] - dataframe['ema52']).abs() / dataframe['ema52'] <= band52) + + # 背离/隐性背离（可调间隔），先简化到 MACD 快线 + sh = self.div_shift + dataframe['bull_div'] = (dataframe['close'] < dataframe['close'].shift(sh)) & (dataframe['macd'] > dataframe['macd'].shift(sh)) & (dataframe['macd'] < 0) + dataframe['bear_div'] = (dataframe['close'] > dataframe['close'].shift(sh)) & (dataframe['macd'] < dataframe['macd'].shift(sh)) & (dataframe['macd'] > 0) + dataframe['hidden_bull'] = (dataframe['close'] > dataframe['close'].shift(sh)) & (dataframe['macd'] < dataframe['macd'].shift(sh)) & (dataframe['macd'] < 0) + dataframe['hidden_bear'] = (dataframe['close'] < dataframe['close'].shift(sh)) & (dataframe['macd'] > dataframe['macd'].shift(sh)) & (dataframe['macd'] > 0) + + # 近N根出现过归零轴（解决“同一根同时满足”过严问题） + zr = dataframe['zero_axis'] + for i in range(1, self.zero_recent_lookback): + zr = zr | dataframe['zero_axis'].shift(i) + dataframe['zero_recent'] = zr.fillna(False) + + # 近零处快线穿越慢线（补充触发源） + near_zero_now = dataframe['macd'].abs() <= (eps_zero * 2) + cross_up = (dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macd'].shift(1) <= dataframe['macdsignal'].shift(1)) + cross_dn = (dataframe['macd'] < dataframe['macdsignal']) & (dataframe['macd'].shift(1) >= dataframe['macdsignal'].shift(1)) + dataframe['zero_cross_up_near'] = near_zero_now & cross_up + dataframe['zero_cross_dn_near'] = near_zero_now & cross_dn + + # ====== 结构：基于 ChanMACD 的 UnitTF 结束点（与零轴定义一致） ====== + try: + from ChanKLU import ChanKLU + from ChanMACD import ChanMACD + klu_list = [] + prev = None + for idx, row in dataframe.iterrows(): + klu = ChanKLU( + time=idx, + open=float(row.get('open', 0) or 0), + high=float(row.get('high', 0) or 0), + low=float(row.get('low', 0) or 0), + close=float(row.get('close', 0) or 0), + volume=float(row.get('volume', 0) or 0), + ) + klu.set_pre(prev) + if prev: + prev.set_next(klu) + klu.ema24 = float(row.get('ema24', 0) or 0) + klu.ema52 = float(row.get('ema52', 0) or 0) + klu.set_indicators({'macd': row.get('macd'), 'macdsignal': row.get('macdsignal'), 'macdhist': row.get('macdhist')}) + klu.set_idx(len(klu_list)) + klu_list.append(klu) + prev = klu + cm = ChanMACD(klu_list) + end_up = {} + end_dn = {} + for u in cm.unittf_list: + if not getattr(u, 'end_klu', None): + continue + dirv = getattr(u, 'dir', 0) + timev = u.end_klu.time + if dirv >= 0: + end_up[timev] = True + else: + end_dn[timev] = True + dataframe['unit_end_up'] = dataframe.index.to_series().apply(lambda t: bool(end_up.get(t, False))).astype(bool) + dataframe['unit_end_dn'] = dataframe.index.to_series().apply(lambda t: bool(end_dn.get(t, False))).astype(bool) + except Exception: + dataframe['unit_end_up'] = False + dataframe['unit_end_dn'] = False + + # ====== 高位空 / 低位多 形态（高位横盘柱衰/低位横盘柱回升） ====== + eps_high = max(eps_zero * 2, 0.08) + H = 5 + N = 3 + macd_high = (dataframe['macd'] > eps_high) + macd_low = (dataframe['macd'] < -eps_high) + # 黄白线高/低位区（结合快慢线） + dataframe['macd_high_zone'] = (dataframe['macd'] > eps_high) & (dataframe['macdsignal'] > eps_high) + dataframe['macd_low_zone'] = (dataframe['macd'] < -eps_high) & (dataframe['macdsignal'] < -eps_high) + hist_down = (dataframe['macdhist'].diff() < 0) + hist_up = (dataframe['macdhist'].diff() > 0) + dataframe['hs_window'] = macd_high.rolling(H).sum() == H + dataframe['ls_window'] = macd_low.rolling(H).sum() == H + dataframe['hist_down_streak'] = hist_down.rolling(N).sum() == N + dataframe['hist_up_streak'] = hist_up.rolling(N).sum() == N + dataframe['high_short_setup'] = (dataframe['hs_window'] & dataframe['hist_down_streak']).fillna(False) + dataframe['low_long_setup'] = (dataframe['ls_window'] & dataframe['hist_up_streak']).fillna(False) + + # ====== 基于枢轴点(局部高低点)的直方图背离/隐性背离检测 ====== + # 枢轴点定义：高点 high[i] > high[i-1] 且 >= high[i+1]；低点相反 + pivot_high = (dataframe['high'] > dataframe['high'].shift(1)) & (dataframe['high'] >= dataframe['high'].shift(-1)) + pivot_low = (dataframe['low'] < dataframe['low'].shift(1)) & (dataframe['low'] <= dataframe['low'].shift(-1)) + # 直方图峰/谷 + hist_peak = (dataframe['macdhist'] > dataframe['macdhist'].shift(1)) & (dataframe['macdhist'] >= dataframe['macdhist'].shift(-1)) + hist_trough = (dataframe['macdhist'] < dataframe['macdhist'].shift(1)) & (dataframe['macdhist'] <= dataframe['macdhist'].shift(-1)) + # 仅在对应象限判定 + hist_peak_pos = hist_peak & (dataframe['macd'] > 0) + hist_trough_neg = hist_trough & (dataframe['macd'] < 0) + # 抽取序列上的上一枢轴值 + ph_price = dataframe['high'].where(pivot_high) + pl_price = dataframe['low'].where(pivot_low) + ph_hist = dataframe['macdhist'].where(hist_peak_pos) + pl_hist = dataframe['macdhist'].where(hist_trough_neg) + prev_ph_price = ph_price.shift(1).ffill() + prev_pl_price = pl_price.shift(1).ffill() + prev_ph_hist = ph_hist.shift(1).ffill() + prev_pl_hist = pl_hist.shift(1).ffill() + # 经典背离 + bear_div_pivot = pivot_high & hist_peak_pos & (dataframe['high'] > prev_ph_price) & (dataframe['macdhist'] < prev_ph_hist) + bull_div_pivot = pivot_low & hist_trough_neg & (dataframe['low'] < prev_pl_price) & (dataframe['macdhist'] > prev_pl_hist) + # 隐性背离（顺势） + hidden_bear_pivot = pivot_high & hist_peak_pos & (dataframe['high'] < prev_ph_price) & (dataframe['macdhist'] > prev_ph_hist) + hidden_bull_pivot = pivot_low & hist_trough_neg & (dataframe['low'] > prev_pl_price) & (dataframe['macdhist'] < prev_pl_hist) + dataframe['bear_div_pivot'] = bear_div_pivot.fillna(False) + dataframe['bull_div_pivot'] = bull_div_pivot.fillna(False) + dataframe['hidden_bear_pivot'] = hidden_bear_pivot.fillna(False) + dataframe['hidden_bull_pivot'] = hidden_bull_pivot.fillna(False) + + # ====== 统计日志（便于回测定位信号规模） ====== + try: + pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A' + cnt_zero_recent = int(dataframe['zero_recent'].fillna(False).sum()) + cnt_zcup = int(dataframe['zero_cross_up_near'].fillna(False).sum()) + cnt_zcdn = int(dataframe['zero_cross_dn_near'].fillna(False).sum()) + cnt_bull_div = int(dataframe['bull_div'].fillna(False).sum()) + cnt_bear_div = int(dataframe['bear_div'].fillna(False).sum()) + cnt_hbull = int(dataframe['hidden_bull'].fillna(False).sum()) + cnt_hbear = int(dataframe['hidden_bear'].fillna(False).sum()) + cnt_u_end_up = int(dataframe.get('unit_end_up', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + cnt_u_end_dn = int(dataframe.get('unit_end_dn', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + cnt_hs = int(dataframe.get('high_short_setup', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + cnt_ll = int(dataframe.get('low_long_setup', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + cnt_bear_div_p = int(dataframe.get('bear_div_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + cnt_bull_div_p = int(dataframe.get('bull_div_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + cnt_hbear_p = int(dataframe.get('hidden_bear_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + cnt_hbull_p = int(dataframe.get('hidden_bull_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + zones_high = int(dataframe.get('macd_high_zone', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + zones_low = int(dataframe.get('macd_low_zone', pd.Series(False, index=dataframe.index)).fillna(False).sum()) + logger.info(f"[{pair}] IND zr={cnt_zero_recent} zcup={cnt_zcup} zcdn={cnt_zcdn} div(bull={cnt_bull_div},bear={cnt_bear_div},hb={cnt_hbull},hs={cnt_hbear}) piv(bull={cnt_bull_div_p},bear={cnt_bear_div_p},hb={cnt_hbull_p},hs={cnt_hbear_p}) zones(high={zones_high},low={zones_low}) unit_end(up={cnt_u_end_up},dn={cnt_u_end_dn}) setup(hs={cnt_hs},ll={cnt_ll})") + except Exception: + pass + + # 进场模板（在 populate_entry_trend 中使用） + return dataframe + + # ============ 入场/出场信号 ============ + def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + dataframe['enter_long'] = 0 + dataframe['enter_short'] = 0 + + # 做多：柱形枢轴底背离/隐性多 + 低位区 或 近零轴（允许只要枢轴+近零即可） + s_false = pd.Series(False, index=dataframe.index) + bull_hist = ( + dataframe.get('bull_div_pivot', s_false).fillna(False) + | + dataframe.get('hidden_bull_pivot', s_false).fillna(False) + ) + low_zone = dataframe.get('macd_low_zone', s_false).fillna(False) + # 放宽近零阈值以确保产生成交 + near_zero = (dataframe['macd'].abs() <= (self.eps_zero_param * 2.0)).fillna(False) + long_cond = (bull_hist & (low_zone | near_zero | dataframe['zero_recent'])) + dataframe.loc[long_cond, 'enter_long'] = 1 + + # 做空：柱形枢轴顶背离/隐性空 + 高位区 或 近零轴（允许只要枢轴+近零即可） + bear_hist = ( + dataframe.get('bear_div_pivot', s_false).fillna(False) + | + dataframe.get('hidden_bear_pivot', s_false).fillna(False) + ) + high_zone = dataframe.get('macd_high_zone', s_false).fillna(False) + short_cond = (bear_hist & (high_zone | near_zero | dataframe['zero_recent'])) + dataframe.loc[short_cond, 'enter_short'] = 1 + + # ====== 入场标签与统计（聚焦柱形+区位） ====== + try: + long_highzone = (bull_hist & low_zone).fillna(False) + long_nearzero = (bull_hist & near_zero).fillna(False) + short_highzone = (bear_hist & high_zone).fillna(False) + short_nearzero = (bear_hist & near_zero).fillna(False) + + dataframe['enter_tag'] = '' + dataframe['long_tag_tmp'] = np.select( + [long_highzone, long_nearzero], + ['HIST_BULL_DIV_HIGHZONE', 'HIST_BULL_DIV_NEARZERO'], + default='' + ) + dataframe['short_tag_tmp'] = np.select( + [short_highzone, short_nearzero], + ['HIST_BEAR_DIV_HIGHZONE', 'HIST_BEAR_DIV_NEARZERO'], + default='' + ) + dataframe.loc[dataframe['enter_long'] == 1, 'enter_tag'] = dataframe.loc[dataframe['enter_long'] == 1, 'long_tag_tmp'].replace('', 'OTHER') + dataframe.loc[dataframe['enter_short'] == 1, 'enter_tag'] = dataframe.loc[dataframe['enter_short'] == 1, 'short_tag_tmp'].replace('', 'OTHER') + + pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A' + cnt_long = int((dataframe['enter_long'] == 1).sum()) + cnt_short = int((dataframe['enter_short'] == 1).sum()) + cnt_l_hz = int(long_highzone.sum()); cnt_l_nz = int(long_nearzero.sum()) + cnt_s_hz = int(short_highzone.sum()); cnt_s_nz = int(short_nearzero.sum()) + # 最终可下单信号数量（enter_* 列） + el = int((dataframe.get('enter_long', 0) == 1).sum()) + es = int((dataframe.get('enter_short', 0) == 1).sum()) + logger.info(f"[{pair}] SIG long={cnt_long} short={cnt_short} long_parts(hz={cnt_l_hz},nz={cnt_l_nz}) short_parts(hz={cnt_s_hz},nz={cnt_s_nz}) ENTER(el={el},es={es})") + except Exception: + pass + + # 不追加 UnitTF 入场，聚焦柱形+区位组合 + try: + idx_long = list(dataframe.index[dataframe['enter_long'] == 1]) + idx_short = list(dataframe.index[dataframe['enter_short'] == 1]) + def _fmt(ts_list): + return [str(ts_list[i]) for i in range(min(5, len(ts_list)))] + (["..."] if len(ts_list) > 10 else []) + [str(ts_list[i]) for i in range(max(0, len(ts_list)-5), len(ts_list))] if ts_list else [] + pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A' + logger.info(f"[{pair}] ENTER_LONG idx samples: {_fmt(idx_long)}") + logger.info(f"[{pair}] ENTER_SHORT idx samples: {_fmt(idx_short)}") + except Exception: + pass + return dataframe + + def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + dataframe['exit_long'] = 0 + dataframe['exit_short'] = 0 + + # 退出：MACD 反向穿越零轴 或 触碰 EMA52 失败 + long_exit = (dataframe['macd'] < 0) | (dataframe['near_ema52'] & (dataframe['macd'] < dataframe['macdsignal'])) + short_exit = (dataframe['macd'] > 0) | (dataframe['near_ema52'] & (dataframe['macd'] > dataframe['macdsignal'])) + dataframe.loc[long_exit, 'exit_long'] = 1 + dataframe.loc[short_exit, 'exit_short'] = 1 + return dataframe + + # ============ 过滤与止损 ============ + def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, + time_in_force: str, current_time: datetime, entry_tag: str | None, + side: str, **kwargs) -> bool: + # 暂时放开所有过滤，确保先产生成交，再逐步收紧 + return True + + def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None: + # 首次进场保存 ATR 作为 1x 止损距离 + dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) + if dataframe is None or len(dataframe) == 0: + return None + last = dataframe.iloc[-1].squeeze() + if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side): + entry_atr = float(last.get('atr', 0) or 0) + trade.set_custom_data(key="entry_atr", value=entry_atr) + logger.info(f"保存开仓ATR: {entry_atr}") + return None + + def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, + current_rate: float, current_profit: float, after_fill: bool, + **kwargs) -> float | None: + # 1x ATR 止损 + entry_atr = trade.get_custom_data(key="entry_atr") + if entry_atr is None: + # 兜底：5% + return -0.05 + if trade.is_short: + stop_price = trade.open_rate + float(entry_atr) + else: + stop_price = trade.open_rate - float(entry_atr) + return stoploss_from_absolute(stop_price, current_rate, is_short=trade.is_short) \ No newline at end of file diff --git a/web/app.py b/web/app.py index 34233f3..bf29c40 100644 --- a/web/app.py +++ b/web/app.py @@ -18,8 +18,9 @@ import numpy as np sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ChanLun import ChanLun -from ChanEnum import Chan_BI_DIR, Chan_SEG_DIR, Chan_KLC_FX, Chan_FX_TYPE +from ChanEnum import Chan_BI_DIR, Chan_SEG_DIR, Chan_KLC_FX, Chan_FX_TYPE, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR from cn_stock_data import ChinaStockData +from ChanMACD import ChanMACD # 添加买卖点枚举类型 class TRADE_POINT_TYPE: @@ -363,6 +364,58 @@ def analyze_chan(df): bi.cal_macd_div() #print(bi.start_time, bi.macd_hist, bi.macd_div) + # 添加ChanMACD分析 + chan_macd = None + chan_macd_data = {} + try: + # 直接使用ChanLun的get_klu_list方法获取KLU列表 + klu_list = chan.get_klu_list(df) + + if klu_list and len(klu_list) > 0: + print(f"获取到KLU列表，长度: {len(klu_list)}") + chan_macd = ChanMACD(klu_list) + chan_macd_data = { + 'seg_list': chan_macd.seg_list, + 'unittf_list': chan_macd.unittf_list, + 'histset_list': chan_macd.histset_list, + 'high_position_list': chan_macd.high_position_list, + 'high_empty_list': chan_macd.high_empty_list, + 'low_position_list': getattr(chan_macd, 'low_position_list', []), + 'low_empty_list': getattr(chan_macd, 'low_empty_list', []), + 'return_zero_list': chan_macd.return_zero_list, + 'cross0_up_list': chan_macd.cross0_up_list, + 'cross0_down_list': chan_macd.cross0_down_list + } + print(f"ChanMACD分析完成: seg={len(chan_macd.seg_list)}, unittf={len(chan_macd.unittf_list)}, histset={len(chan_macd.histset_list)}") + else: + print("未能获取KLU列表或列表为空") + chan_macd_data = { + 'seg_list': [], + 'unittf_list': [], + 'histset_list': [], + 'high_position_list': [], + 'high_empty_list': [], + 'return_zero_list': [], + 'cross0_up_list': [], + 'cross0_down_list': [] + } + except Exception as e: + print(f"ChanMACD分析出错: {e}") + import traceback + traceback.print_exc() + chan_macd_data = { + 'seg_list': [], + 'unittf_list': [], + 'histset_list': [], + 'high_position_list': [], + 'high_empty_list': [], + 'low_position_list': [], + 'low_empty_list': [], + 'return_zero_list': [], + 'cross0_up_list': [], + 'cross0_down_list': [] + } + # 获取原始K线数据用于KLU分型分析 klu_list = [] try: @@ -490,7 +543,8 @@ def analyze_chan(df): 'zs_list': zs_list, 'trade_points': buy_sell_points, 'klc_fx_info': klc_fx_info, # KLC分型信息 - 'klu_fx_info': klu_fx_info # 添加KLU分型信息 + 'klu_fx_info': klu_fx_info, # 添加KLU分型信息 + 'chan_macd': chan_macd_data # 添加ChanMACD分析数据 } def generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, start_time=None, end_time=None): @@ -981,6 +1035,211 @@ def format_time_safely(time_obj, client_tz): # 已经是datetime对象 return time_obj.astimezone(client_tz).isoformat() +def serialize_chan_macd_data(chan_macd_data, client_tz): + """序列化ChanMACD数据为JSON可序列化格式""" + serialized_data = { + 'seg_list': [], + 'unittf_list': [], + 'histset_list': [], + # 状态标记数据 + 'high_position_list': [], + 'high_empty_list': [], + 'low_position_list': [], + 'low_empty_list': [], + 'return_zero_list': [], + 'cross0_up_list': [], + 'cross0_down_list': [] + } + + # 序列化seg_list + for seg in chan_macd_data.get('seg_list', []): + try: + seg_data = { + 'start_time': format_time_safely(seg.start_time, client_tz), + 'end_time': format_time_safely(seg.end_time, client_tz) if seg.end_time else None, + 'seg_dir': 'ABOVE' if seg.seg_dir == Chan_MACDSEG_DIR.ABOVE else 'UNDER', + 'klu_count': len(seg.klu_list) if hasattr(seg, 'klu_list') else 0, + 'unittf_count': len(seg.unittf_list) if hasattr(seg, 'unittf_list') else 0, + 'histset_count': len(seg.hist_set) if hasattr(seg, 'hist_set') else 0 + } + serialized_data['seg_list'].append(seg_data) + except Exception as e: + print(f"序列化seg出错: {e}") + continue + + # 序列化unittf_list（兼容新结构与枚举类型） + for unittf in chan_macd_data.get('unittf_list', []): + try: + dir_value = getattr(unittf, 'uinttf_dir', None) + dir_name = getattr(dir_value, 'name', dir_value if isinstance(dir_value, str) else None) + start_t = getattr(unittf, 'start_type', None) + start_type = getattr(start_t, 'name', start_t) + end_t = getattr(unittf, 'end_type', None) + end_type = getattr(end_t, 'name', end_t) + peak_abs = getattr(unittf, 'peak_abs', None) + if peak_abs is None: + peak_abs = getattr(unittf, 'peak_hist', None) + length = getattr(unittf, 'length', None) + if length is None: + length = len(unittf.klu_list) if hasattr(unittf, 'klu_list') else None + + unittf_data = { + 'start_time': format_time_safely(getattr(unittf, 'start_time', None), client_tz), + 'end_time': format_time_safely(getattr(unittf, 'end_time', None), client_tz) if getattr(unittf, 'end_time', None) else None, + 'dir': dir_name, # 'ABOVE' | 'UNDER' | None + 'start_type': start_type, # e.g. 'START' | 'CROSS0' | 'NEAR0_UP' | 'NEAR0_DOWN' + 'end_type': end_type, + 'invalid': getattr(unittf, 'invalid', False), + 'peak_abs': peak_abs, + 'length': length, + 'klu_count': len(unittf.klu_list) if hasattr(unittf, 'klu_list') else 0, + 'histset_count': len(unittf.histset_list) if hasattr(unittf, 'histset_list') else 0 + } + serialized_data['unittf_list'].append(unittf_data) + except Exception as e: + print(f"序列化unittf出错: {e}") + continue + + # 序列化histset_list + for histset in chan_macd_data.get('histset_list', []): + try: + histset_data = { + 'start_time': format_time_safely(getattr(histset, 'start_time', None), client_tz), + 'end_time': format_time_safely(getattr(histset, 'end_time', None), client_tz), + 'histset_dir': 'ABOVE' if histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE else 'UNDER', + 'klu_count': len(histset.klu_list) if hasattr(histset, 'klu_list') else 0 + } + serialized_data['histset_list'].append(histset_data) + except Exception as e: + print(f"序列化histset出错: {e}") + continue + + # 序列化状态标记数据 + # 序列化高位列表 + for high_pos in chan_macd_data.get('high_position_list', []): + try: + high_pos_data = { + 'time': format_time_safely(high_pos['time'], client_tz), + 'end_time': format_time_safely(high_pos.get('end_time'), client_tz) if high_pos.get('end_time') else None, + 'type': high_pos.get('type', 'start'), + 'macd': high_pos.get('macd'), + 'signal': high_pos.get('signal'), + 'macdhist': high_pos.get('macdhist'), + 'end_macd': high_pos.get('end_macd'), + 'end_signal': high_pos.get('end_signal'), + 'end_macdhist': high_pos.get('end_macdhist') + } + serialized_data['high_position_list'].append(high_pos_data) + except Exception as e: + print(f"序列化high_position出错: {e}") + continue + + # 序列化高位空列表 + for high_empty in chan_macd_data.get('high_empty_list', []): + try: + high_empty_data = { + 'time': format_time_safely(high_empty['time'], client_tz), + 'end_time': format_time_safely(high_empty.get('end_time'), client_tz) if high_empty.get('end_time') else None, + 'type': high_empty.get('type', 'start'), + 'macd': high_empty.get('macd'), + 'signal': high_empty.get('signal'), + 'macdhist': high_empty.get('macdhist'), + 'end_macd': high_empty.get('end_macd'), + 'end_signal': high_empty.get('end_signal'), + 'end_macdhist': high_empty.get('end_macdhist') + } + serialized_data['high_empty_list'].append(high_empty_data) + except Exception as e: + print(f"序列化high_empty出错: {e}") + continue + + # 序列化低位与低位空 + for low_pos in chan_macd_data.get('low_position_list', []): + try: + low_pos_data = { + 'time': format_time_safely(low_pos['time'], client_tz), + 'end_time': format_time_safely(low_pos.get('end_time'), client_tz) if low_pos.get('end_time') else None, + 'type': low_pos.get('type', 'start'), + 'macd': low_pos.get('macd'), + 'signal': low_pos.get('signal'), + 'macdhist': low_pos.get('macdhist'), + 'end_macd': low_pos.get('end_macd'), + 'end_signal': low_pos.get('end_signal'), + 'end_macdhist': low_pos.get('end_macdhist') + } + serialized_data['low_position_list'].append(low_pos_data) + except Exception as e: + print(f"序列化low_position出错: {e}") + continue + + for low_empty in chan_macd_data.get('low_empty_list', []): + try: + low_empty_data = { + 'time': format_time_safely(low_empty['time'], client_tz), + 'end_time': format_time_safely(low_empty.get('end_time'), client_tz) if low_empty.get('end_time') else None, + 'type': low_empty.get('type', 'start'), + 'macd': low_empty.get('macd'), + 'signal': low_empty.get('signal'), + 'macdhist': low_empty.get('macdhist'), + 'end_macd': low_empty.get('end_macd'), + 'end_signal': low_empty.get('end_signal'), + 'end_macdhist': low_empty.get('end_macdhist') + } + serialized_data['low_empty_list'].append(low_empty_data) + except Exception as e: + print(f"序列化low_empty出错: {e}") + continue + + # 序列化归零轴列表 + for return_zero in chan_macd_data.get('return_zero_list', []): + try: + return_zero_data = { + 'time': format_time_safely(return_zero['time'], client_tz), + 'end_time': format_time_safely(return_zero.get('end_time'), client_tz) if return_zero.get('end_time') else None, + 'type': return_zero.get('type', 'start'), + 'macd': return_zero.get('macd'), + 'signal': return_zero.get('signal'), + 'macdhist': return_zero.get('macdhist'), + 'end_macd': return_zero.get('end_macd'), + 'end_signal': return_zero.get('end_signal'), + 'end_macdhist': return_zero.get('end_macdhist') + } + serialized_data['return_zero_list'].append(return_zero_data) + except Exception as e: + print(f"序列化return_zero出错: {e}") + continue + + # 序列化穿越零轴列表 + for cross0_up in chan_macd_data.get('cross0_up_list', []): + try: + cross0_up_data = { + 'time': format_time_safely(cross0_up['time'], client_tz), + 'type': cross0_up.get('type', 'start'), + 'macd': cross0_up.get('macd'), + 'signal': cross0_up.get('signal'), + 'macdhist': cross0_up.get('macdhist') + } + serialized_data['cross0_up_list'].append(cross0_up_data) + except Exception as e: + print(f"序列化cross0_up出错: {e}") + continue + + for cross0_down in chan_macd_data.get('cross0_down_list', []): + try: + cross0_down_data = { + 'time': format_time_safely(cross0_down['time'], client_tz), + 'type': cross0_down.get('type', 'start'), + 'macd': cross0_down.get('macd'), + 'signal': cross0_down.get('signal'), + 'macdhist': cross0_down.get('macdhist') + } + serialized_data['cross0_down_list'].append(cross0_down_data) + except Exception as e: + print(f"序列化cross0_down出错: {e}") + continue + + return serialized_data + def is_smaller_timeframe(tf1, tf2): """判断时间周期tf1是否小于tf2""" # 定义时间周期的分钟数映射 @@ -1439,7 +1698,9 @@ def analyze(): 'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级 'is_strong_fx': bool(point['is_strong_fx']), # 是否为强分型 'fx_confirmed': bool(point['fx_confirmed']) # 分型是否确认 - } for point in analysis_result['klu_fx_info']] + } for point in analysis_result['klu_fx_info']], + # 添加ChanMACD分析数据 + 'chan_macd': serialize_chan_macd_data(analysis_result.get('chan_macd', {}), client_tz) }) # 如果生成了回放数据，添加到返回结果中 @@ -1563,6 +1824,9 @@ def analyze(): 'fx_confirmed': bool(point['fx_confirmed']) # 分型是否确认 } for point in element_analysis['klu_fx_info']] + # 添加次周期ChanMACD分析数据 + result['element_chan_macd'] = serialize_chan_macd_data(element_analysis.get('chan_macd', {}), client_tz) + pass return jsonify(result) diff --git a/web/templates/index.html b/web/templates/index.html index 47ae231..cc67112 100644 --- a/web/templates/index.html +++ b/web/templates/index.html @@ -909,6 +909,10 @@ +
+ + +
@@ -1037,6 +1041,8 @@
+ +