jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加klc识别ema线接触

Browse Source
This commit is contained in:
jackyu66git
2026-02-08 22:01:45 +08:00
parent b9c11b6163
commit 10b8ba398b
6 changed files with 305 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanKLC.py
+254 -9
View File
@@ -1,7 +1,7 @@
import copy
from typing import Dict, Optional
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_KLC_FX, Chan_K_DIR, Chan_MACD_STATE, Chan_PRICE_TREND
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_KLC_FX, Chan_K_DIR, Chan_MACD_STATE, Chan_PRICE_TREND, Chan_EMA_POS, Chan_EMA_SEMANTIC
import ChanKLU
import ChanCTime
@@ -45,11 +45,247 @@ class ChanKLC():
self.state = Chan_MACD_STATE.UNKNOWN
self.continue_div = False
self.separate_div = False
self.ema52 = klu.ema52
self.ema24 = klu.ema24
self.ema52 = klu.ema52
self.ema104 = klu.ema104
self.ema156 = klu.ema156
self.ema208 = klu.ema208
self.trend = Chan_PRICE_TREND.UNKNOWN
self.exception = klu.exception
self.klc_dir = Chan_KLINE_DIR.UP if klu.close > klu.open else Chan_KLINE_DIR.DOWN
self.ema_dir = klu.ema_dir
self.bsp = False
# EMA状态字典：key为EMA名称，value为 {'pos': Chan_EMA_POS, 'semantic': Chan_EMA_SEMANTIC}
self.ema_status = {}
# 向后兼容：保留 ema52_status 和 ema52_pos
self.ema52_status = 0
self.ema52_pos = Chan_EMA_POS.UNKNOWN
self.cal_all_ema_status()
# ==================== EMA 通用计算方法 ====================
@staticmethod
def cal_ema_pos(high, low, close, ema_value, threshold=0):
"""
计算K线与任意EMA的客观位置关系（与趋势方向无关，支持threshold容差）
参数:
high, low, close: K线的高低收盘价
ema_value: EMA的值
threshold: 容差值（绝对值），在此范围内视为"接近/触碰"
例如 BTC 价格 $100,000 时 threshold=100 表示差100点视为触碰
返回:
Chan_EMA_POS 枚举值
判断逻辑（以threshold=100, ema=97000为例）：
ema_zone = [96900, 97100] （EMA上下各扩展threshold）
ABOVE: low > 97100 K线完全在zone上方（远离EMA）
NEAR_ABOVE: 97000 < low <= 97100 K线在上方但下影线进入zone（接近EMA）
CROSS_CLOSE_ABOVE: close > 97000, low <= 97000 K线穿越EMA，收盘在上方
ON_EMA: abs(close - 97000) <= 100 收盘价在zone内
CROSS_CLOSE_BELOW: close < 97000, high >= 97000 K线穿越EMA，收盘在下方
NEAR_BELOW: 96900 <= high < 97000 K线在下方但上影线进入zone（接近EMA）
BELOW: high < 96900 K线完全在zone下方（远离EMA）
"""
if ema_value is None or ema_value == 0:
return Chan_EMA_POS.UNKNOWN
ema_upper = ema_value + threshold # EMA zone 上界
ema_lower = ema_value - threshold # EMA zone 下界
# 1. 收盘价在EMA附近（zone内）
if threshold > 0 and abs(close - ema_value) <= threshold:
# 收盘价在zone内，但还需要看是否有实际穿越
if low <= ema_value and close >= ema_value:
return Chan_EMA_POS.CROSS_CLOSE_ABOVE # 实际穿越了精确EMA线
elif high >= ema_value and close <= ema_value:
return Chan_EMA_POS.CROSS_CLOSE_BELOW
return Chan_EMA_POS.ON_EMA
# 2. K线实际穿越了精确的EMA线
if close > ema_value and low <= ema_value:
return Chan_EMA_POS.CROSS_CLOSE_ABOVE
if close < ema_value and high >= ema_value:
return Chan_EMA_POS.CROSS_CLOSE_BELOW
if close == ema_value:
return Chan_EMA_POS.ON_EMA
# 3. 没有实际穿越，检查是否"接近"（在threshold zone内）
if close > ema_value:
# K线在EMA上方
if threshold > 0 and low <= ema_upper:
return Chan_EMA_POS.NEAR_ABOVE # 下影线进入zone，接近但未触碰
return Chan_EMA_POS.ABOVE # 远离EMA
else:
# K线在EMA下方
if threshold > 0 and high >= ema_lower:
return Chan_EMA_POS.NEAR_BELOW # 上影线进入zone，接近但未触碰
return Chan_EMA_POS.BELOW # 远离EMA
@staticmethod
def cal_ema_semantic(ema_pos, kline_dir, ema_dir):
"""
根据客观位置 + K线方向 + 趋势方向，计算语义状态
参数:
ema_pos: Chan_EMA_POS 客观位置
kline_dir: Chan_KLINE_DIR K线方向 (UP/DOWN/COMBINE/INCLUDED)
ema_dir: int 趋势方向 (1=多头, -1=空头, 0=盘整)
返回:
Chan_EMA_SEMANTIC 枚举值
语义含义（以多头为例，空头完全对称）：
TOUCH_HOLD: 触碰EMA，收盘守住趋势侧（支撑/压力有效）
BREAK: 穿越EMA，收盘在逆势侧（支撑/压力失败）
DEEP_COUNTER: 完全在EMA逆势侧（深度回调/反抽）
TOUCH_FAIL: 逆势触碰EMA但未穿越（反弹/反抽力度不足）
RECOVER: 逆势后穿越EMA回到趋势侧（收复EMA）
TREND_SIDE: 完全在EMA趋势侧（正常运行）
STRONG_TREND: 顺势K线完全在EMA趋势侧（强势，远未及EMA）
WEAK_COUNTER: 逆势K线完全在EMA逆势侧（弱势，远未到EMA）
"""
if ema_pos == Chan_EMA_POS.UNKNOWN:
return Chan_EMA_SEMANTIC.NEUTRAL
# 统一处理：将多头/盘整和空头映射到同一套逻辑
# is_bull=True 时，"趋势侧"=上方，"逆势侧"=下方
# is_bull=False时，"趋势侧"=下方，"逆势侧"=上方
is_bull = ema_dir >= 0 # 多头和盘整都按多头逻辑处理
# K线是否是顺势方向（多头下UP为顺势，空头下DOWN为顺势）
is_trend_kline = (kline_dir == Chan_KLINE_DIR.UP) if is_bull else (kline_dir == Chan_KLINE_DIR.DOWN)
is_counter_kline = (kline_dir == Chan_KLINE_DIR.DOWN) if is_bull else (kline_dir == Chan_KLINE_DIR.UP)
# 位置映射：多头下 ABOVE=趋势侧, BELOW=逆势侧; 空头反过来
trend_side = Chan_EMA_POS.ABOVE if is_bull else Chan_EMA_POS.BELOW
counter_side = Chan_EMA_POS.BELOW if is_bull else Chan_EMA_POS.ABOVE
near_trend = Chan_EMA_POS.NEAR_ABOVE if is_bull else Chan_EMA_POS.NEAR_BELOW
near_counter = Chan_EMA_POS.NEAR_BELOW if is_bull else Chan_EMA_POS.NEAR_ABOVE
cross_to_trend = Chan_EMA_POS.CROSS_CLOSE_ABOVE if is_bull else Chan_EMA_POS.CROSS_CLOSE_BELOW
cross_to_counter = Chan_EMA_POS.CROSS_CLOSE_BELOW if is_bull else Chan_EMA_POS.CROSS_CLOSE_ABOVE
# COMBINE / INCLUDED 方向：只看位置，不区分强弱
if not is_trend_kline and not is_counter_kline:
if ema_pos == trend_side:
return Chan_EMA_SEMANTIC.TREND_SIDE
elif ema_pos in (near_trend, cross_to_trend, Chan_EMA_POS.ON_EMA):
return Chan_EMA_SEMANTIC.APPROACHING
elif ema_pos in (near_counter, cross_to_counter):
return Chan_EMA_SEMANTIC.APPROACHING
elif ema_pos == counter_side:
return Chan_EMA_SEMANTIC.DEEP_COUNTER
return Chan_EMA_SEMANTIC.NEUTRAL
# 逆势K线（多头下的下跌K线 / 空头下的上涨K线）
if is_counter_kline:
if ema_pos == trend_side:
return Chan_EMA_SEMANTIC.STRONG_TREND # 逆势K线仍在趋势侧（回调很浅）
elif ema_pos == near_trend:
return Chan_EMA_SEMANTIC.APPROACHING # 接近EMA，即将测试支撑/压力
elif ema_pos == cross_to_trend:
return Chan_EMA_SEMANTIC.TOUCH_HOLD # 触碰EMA后守住趋势侧
elif ema_pos == Chan_EMA_POS.ON_EMA:
return Chan_EMA_SEMANTIC.TOUCH_HOLD # 收盘在EMA附近，视为守住
elif ema_pos == cross_to_counter:
return Chan_EMA_SEMANTIC.BREAK # 穿越EMA到逆势侧
elif ema_pos == near_counter:
return Chan_EMA_SEMANTIC.BREAK # 接近EMA但收盘在逆势侧，也视为击穿
elif ema_pos == counter_side:
return Chan_EMA_SEMANTIC.DEEP_COUNTER # 完全在逆势侧
# 顺势K线（多头下的上涨K线 / 空头下的下跌K线）
if is_trend_kline:
if ema_pos == counter_side:
return Chan_EMA_SEMANTIC.WEAK_COUNTER # 顺势K线却在逆势侧（弱势）
elif ema_pos == near_counter:
return Chan_EMA_SEMANTIC.APPROACHING # 从逆势侧接近EMA
elif ema_pos == cross_to_counter:
return Chan_EMA_SEMANTIC.TOUCH_FAIL # 触碰EMA但未穿越回趋势侧
elif ema_pos == Chan_EMA_POS.ON_EMA:
return Chan_EMA_SEMANTIC.TOUCH_FAIL # 收盘在EMA附近，未确认突破
elif ema_pos == cross_to_trend:
return Chan_EMA_SEMANTIC.RECOVER # 从逆势侧穿越回趋势侧
elif ema_pos == near_trend:
return Chan_EMA_SEMANTIC.RECOVER # 接近趋势侧（刚收复EMA附近）
elif ema_pos == trend_side:
return Chan_EMA_SEMANTIC.TREND_SIDE # 完全在趋势侧（正常）
return Chan_EMA_SEMANTIC.NEUTRAL
@staticmethod
def semantic_to_int(semantic):
"""将 Chan_EMA_SEMANTIC 枚举转换为整数，兼容旧的 ema52_status 数值"""
mapping = {
Chan_EMA_SEMANTIC.TOUCH_HOLD: 1,
Chan_EMA_SEMANTIC.BREAK: 2,
Chan_EMA_SEMANTIC.DEEP_COUNTER: 3,
Chan_EMA_SEMANTIC.TOUCH_FAIL: 4,
Chan_EMA_SEMANTIC.RECOVER: 5,
Chan_EMA_SEMANTIC.TREND_SIDE: 6,
Chan_EMA_SEMANTIC.STRONG_TREND: 7,
Chan_EMA_SEMANTIC.WEAK_COUNTER: 8,
Chan_EMA_SEMANTIC.APPROACHING: 9,
Chan_EMA_SEMANTIC.NEUTRAL: 0,
}
return mapping.get(semantic, 0)
# threshold_pct: 阈值百分比，用于自动计算绝对阈值
# 例如 0.001 表示 EMA 值的 0.1%，BTC $100,000 时 threshold = $100
threshold_pct = 0.001
def cal_all_ema_status(self):
"""
统一计算所有EMA与K线的位置关系和语义状态
threshold 自动按 EMA 值的百分比计算（cls.threshold_pct，默认0.1%）
- BTC $100,000 时：threshold ≈ $100
- ETH $3,000 时：threshold ≈ $3
- SOL $200 时：threshold ≈ $0.2
结果存储在 self.ema_status 字典中，格式：
{
'ema24': {'pos': Chan_EMA_POS, 'semantic': Chan_EMA_SEMANTIC, 'value': float, 'threshold': float},
'ema52': {...},
...
}
同时保持向后兼容：self.ema52_pos 和 self.ema52_status
"""
ema_configs = {
'ema24': self.ema24,
'ema52': self.ema52,
'ema104': self.ema104,
'ema156': self.ema156,
'ema208': self.ema208,
}
self.ema_status = {}
for name, value in ema_configs.items():
# 按 EMA 值的百分比自动计算阈值
threshold = abs(value) * self.threshold_pct if value and self.threshold_pct > 0 else 0
pos = ChanKLC.cal_ema_pos(self.high, self.low, self.close, value, threshold)
semantic = ChanKLC.cal_ema_semantic(pos, self.dir, self.ema_dir)
self.ema_status[name] = {
'pos': pos,
'semantic': semantic,
'value': value,
'threshold': threshold,
}
# 向后兼容
self.ema52_pos = self.ema_status['ema52']['pos']
self.ema52_status = ChanKLC.semantic_to_int(self.ema_status['ema52']['semantic'])
def get_ema_pos(self, ema_name):
"""获取指定EMA的客观位置，如 klc.get_ema_pos('ema24')"""
if ema_name in self.ema_status:
return self.ema_status[ema_name]['pos']
return Chan_EMA_POS.UNKNOWN
def get_ema_semantic(self, ema_name):
"""获取指定EMA的语义状态，如 klc.get_ema_semantic('ema52')"""
if ema_name in self.ema_status:
return self.ema_status[ema_name]['semantic']
return Chan_EMA_SEMANTIC.NEUTRAL
def set_trend(self, trend):
self.trend = trend
def to_string(self):
@@ -123,14 +359,23 @@ class ChanKLC():
self.rsi += self.klu_list[index].rsi
self.volume_ratio += self.klu_list[index].volume_ratio
self.macdhist += self.klu_list[index].macdhist
self.ema52 += self.klu_list[index].ema52
self.ema24 += self.klu_list[index].ema24
self.rsi = self.rsi / len(self.klu_list)
self.volume_ratio = self.volume_ratio / len(self.klu_list)
self.volume = self.volume / len(self.klu_list)
self.macdhist = self.macdhist / len(self.klu_list)
self.ema52 = self.ema52 / len(self.klu_list)
self.ema24 = self.ema24 / len(self.klu_list)
self.ema52 += self.klu_list[index].ema52
self.ema104 += self.klu_list[index].ema104
self.ema156 += self.klu_list[index].ema156
self.ema208 += self.klu_list[index].ema208
if self.ema_dir != self.klu_list[index].ema_dir:
self.ema_dir = 0
n = len(self.klu_list)
self.rsi = self.rsi / n
self.volume_ratio = self.volume_ratio / n
self.volume = self.volume / n
self.macdhist = self.macdhist / n
self.ema24 = self.ema24 / n
self.ema52 = self.ema52 / n
self.ema104 = self.ema104 / n
self.ema156 = self.ema156 / n
self.ema208 = self.ema208 / n
if len(self.klu_list) > 0:
self.macd = self.klu_list[-1].macd
self.signal = self.klu_list[-1].signal