jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
be442783a1353e006ddea9174be1f6ef5ed66aab
Chan/chanlun/pipeline/builders/kline.py
T
jackyu66gitandCursor bfdb2f2e2a 引擎四处「算了没人要的东西」，append_bar 13.9ms → 6.6ms 
服务端把增量上线后回传两个热点：add_indicators 为加一根重算全表（占 34%）、
cal_bi_list 整表重扫（51%）。顺着查下来四处都不是算法慢，是算了没人读的结果。

1. cal_trend 挂到 lean 下。它不是笔的依赖（bi.py:221 在它自己的循环里读自身
   序列状态），服务端 verify_incr_parity 三币 1800 根已对拍定论。web 走非
   lean，klc_trend 图层不受影响。

2. check_fx_pattern 删掉拼完就丢的字符串。它把 klu.to_string() 拼成 p 只为
   一行注释掉的 print——2000 根上近 3 万次 f-string 加 6 万次 enum 格式化，
   而且在 cal_bi_list 内层。klu.pattern 只被 cal_klu_pattern 自己的双K/三K
   判定读，不出模块不进 web，所以整个调用在 lean 下也跳过。

3. ChanBI.add_klc 去二次方。去重原本线性扫 klc_list，且每加一根就把整笔所有
   KLU 的 macdhist 重累一遍，往一笔加 k 根是 O(k²)。改成下标集合加
   macd_hist/macd_div 惰性求值。这两个值只有背驰判定（bsp.py）读，lean 下
   bsp 根本不算。

4. add_indicators 批量挂列。2001 行上 TA 计算合计只有 2.5ms，而 30 多次
   df['x']= 要 3.6ms——开销大头是 BlockManager 逐列插入不是计算，改为一次
   concat。cal_volume_ratio 里为算一列 rolling 而 copy() 整张 40 列表，一并去掉。

实测（本机，2001 根窗口。服务端基线 21.8ms 是另一台机器，别直接比绝对值）：
  append_bar        13.9 → 6.6ms
  └ rebuild_bi_zs    8.7 → 2.8ms
  └ add_indicators   4.4 → 3.5ms
  TF_DF lean        49.8 → 32.9ms
  TF_DF full        72.7 → 64.9ms

对拍用 git worktree 检出改动前的提交，同一份 BTC 1m 4000 根跑 38 项指纹：
full 模式 19 项全部一致（web 那条路没动）；lean 模式差 2 项，正是设计要它差
的 klc.trend 和 klu.pattern，而 lean 下 bi/zs/seg/bsp/dataframe 全部一致——
这就是「这两个字段没人读」的实测证据：打空它们，下游一位不变。

瓶颈已经换位置了。新增 probe_inner.py 拆 inner_ms 分档：本机 TF_DF 两条腿占
70%、build_htf_zones 13%、htf_fx_timeline 6%，而服务端报的是 chan 构建 22ms /
信号链 86ms，机器差解释不了这个四倍差距。曾怀疑是 payload 反序列化，实测
_rebuild 只有 1.0ms，假设不成立。两边跑同一探针对分档表才能定位。

HANDOFF 顺带修掉一处 5.6 重号（增量落地那节改为 5.7，本节挂 5.71）。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-28 16:01:26 +08:00

574 lines
22 KiB
Python
Raw Blame History

"""TF_DF builder mixin — 由 split_tfdf_builders 自动生成，逻辑与原 TF_DF 一致。"""
from __future__ import annotations
from datetime import timedelta
from decimal import Decimal
import numpy as np
import pandas as pd
from pandas import DataFrame
from chanlun.core.ChanBI import ChanBI
from chanlun.core.ChanBIZS import ChanBIZS
from chanlun.core.ChanBSP import ChanBSP
from chanlun.core.ChanEnum import (
Chan_BI_DIR,
Chan_BSP_DIR,
Chan_BSP_TYPE,
Chan_FX_TYPE,
Chan_K_DIR,
Chan_KLC_FX,
Chan_KLC_STATE,
Chan_KLINE_DIR,
Chan_KLU_PATTERN,
Chan_PRICE_TREND,
Chan_SEG_DIR,
Chan_ZS_DIR,
)
from chanlun.core.ChanKLC import ChanKLC
from chanlun.core.ChanKLU import ChanKLU
from chanlun.core.ChanSBI import ChanSBI
from chanlun.core.ChanSEG import ChanSEG
from chanlun.core.ChanZS import ChanZS, ChanZS_Big
from chanlun.indicators.ChanMACD import ChanMACD
class KlineBuilderMixin:
def get_klu_state(self, dataframe):
klc_list = self.get_klc_list(self.get_klu_list(dataframe))
bi_list = self.cal_bi_list(klc_list)
klu_state_list = []
klc_index = 0
for index in range(0, len(dataframe)):
if klc_index == len(klc_list):
klc_index = len(klc_list) - 1
klc = klc_list[klc_index]
if klc.end_klu and klc.end_klu.idx == index:
if klc.klc_state == Chan_KLC_STATE.S10:
klu_state_list.append("10")
#print(klc.end_time, klc.klc_fx_type)
elif klc.klc_state == Chan_KLC_STATE.S_10:
klu_state_list.append("-10")
#print(klc.end_time, klc.klc_fx_type)
elif klc.klc_state == Chan_KLC_STATE.S11:
klu_state_list.append("11")
#print(klc.end_time, klc.klc_fx_type)
elif klc.klc_state == Chan_KLC_STATE.S_11:
klu_state_list.append("-11")
#print(klc.end_time, klc.klc_fx_type)
else:
klu_state_list.append("00")
klc_index += 1
else:
klu_state_list.append("00")
print(klu_state_list[:20])
return klu_state_list
def check_fx1(self, klc):
if klc.pre and klc.next:
if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low:
if klc.pre.pre and klc.next.next:
if klc.high > klc.pre.pre.high and klc.high > klc.next.next.high:
#if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0:
klc.set_fx(Chan_FX_TYPE.TOP)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP")
return Chan_FX_TYPE.TOP
elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high:
#if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0:
if klc.pre.pre and klc.next.next:
if klc.low < klc.pre.pre.low and klc.low < klc.next.next.low:
klc.set_fx(Chan_FX_TYPE.BOTTOM)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM")
return Chan_FX_TYPE.BOTTOM
return Chan_FX_TYPE.UNKNOWN
def check_fx(self, klc):
# 右K未完成（仍在包含合并）时不分型：否则确认笔会随 next 扩区间被 check_*_fx 收回
if klc.pre and klc.next and klc.next.end_klu is not None:
if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low:
#if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0:
klc.set_fx(Chan_FX_TYPE.TOP)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP")
return Chan_FX_TYPE.TOP
elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high:
#if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0:
klc.set_fx(Chan_FX_TYPE.BOTTOM)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM")
return Chan_FX_TYPE.BOTTOM
return Chan_FX_TYPE.UNKNOWN
def check_fx3(self, klc):
# 右K未完成（仍在包含合并）时不分型：否则确认笔会随 next 扩区间被 check_*_fx 收回
if klc.pre and klc.next and klc.next.end_klu is not None:
next_klu = klc.next.end_klu.next
if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low and klc.high > next_klu.high:
#if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0:
klc.set_fx(Chan_FX_TYPE.TOP)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP")
return Chan_FX_TYPE.TOP
elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high and klc.low < next_klu.low:
#if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0:
klc.set_fx(Chan_FX_TYPE.BOTTOM)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM")
return Chan_FX_TYPE.BOTTOM
return Chan_FX_TYPE.UNKNOWN
def check_fx2(self, klc):
if klc.pre and klc.next:
if klc.high > klc.pre.close and klc.close > klc.next.close and klc.close > klc.pre.close and klc.close > klc.next.close:
#if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0:
klc.set_fx(Chan_FX_TYPE.TOP)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP")
return Chan_FX_TYPE.TOP
elif klc.low < klc.pre.close and klc.close < klc.next.close and klc.close < klc.pre.close and klc.close < klc.next.close:
#if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0:
klc.set_fx(Chan_FX_TYPE.BOTTOM)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM")
return Chan_FX_TYPE.BOTTOM
return Chan_FX_TYPE.UNKNOWN
def check_fx_pattern(self, klc):
"""给分型前后三根 KLC 的裸 K 打 pattern 标记。
原本还会把 `klu.to_string()` 拼成一个字符串——那是给下面那行注释掉的
print 用的，拼完就丢。它在 cal_bi_list 的内层，2000 根上要跑近三万次
f-string + 六万次 enum 格式化，是纯废动作，已删。
`klu.pattern` 只被 cal_klu_pattern 自己的双 K / 三 K 判定读，
不出这个模块，也不进 web 序列化。所以 lean 下整个调用可跳。
"""
if getattr(self, 'lean', False):
return
klu_list = klc.pre.klu_list + klc.klu_list + klc.next.klu_list
self.cal_klu_pattern(klu_list)
def cal_volume_ratio(self, dataframe, window=10):
"""当根量 / 前 window 根均量。前 window-1 根无基准，填 1.0。
原写法先 `dataframe.copy()` 再挂两列——为算一列 rolling 复制了整张
四十列的表。直接在 Series 上算，结果逐值相同。
"""
vol = dataframe['volume']
return (vol / vol.rolling(window=window).mean()).fillna(1.0).rename('volume_ratio')
def cal_kl_data(self, dataframe:DataFrame):
"""按行构造 KLU 链。
这里刻意不用 `dataframe.iloc[i]`：那会为每一根新建一个几十列的 Series，
随后 set_indicators 再在其上做几十次逐键查找。实测这两件事合计占 TF_DF
构建耗时的 96%。改为先把用到的列取成 ndarray，循环里只做整数下标访问。
"""
n = len(dataframe)
if n == 0:
return []
times = self._format_times(dataframe['date'])
o_a = dataframe['open'].to_numpy(dtype=float)
h_a = dataframe['high'].to_numpy(dtype=float)
l_a = dataframe['low'].to_numpy(dtype=float)
c_a = dataframe['close'].to_numpy(dtype=float)
v_a = dataframe['volume'].to_numpy(dtype=float)
has_ind = 'macd' in dataframe.columns
ind_cols = {}
if has_ind:
for _attr, col in ChanKLU.INDICATOR_FIELDS:
if col in dataframe.columns:
ind_cols[col] = dataframe[col].to_numpy(dtype=float)
klu_list = []
last_klu = None
for i in range(n):
klu = ChanKLU(times[i], o_a[i], h_a[i], l_a[i], c_a[i], v_a[i])
klu.set_idx(i)
klu_list.append(klu)
if last_klu:
last_klu.set_next(klu)
klu.set_pre(last_klu)
last_klu = klu
if has_ind:
klu.set_indicators_from(ind_cols, i)
return klu_list
@staticmethod
def _format_times(col):
"""向量化 strftime。非 datetime 列（少见）退回逐个格式化。"""
fmt = '%Y-%m-%d %H:%M:%S'
try:
return col.dt.strftime(fmt).to_numpy()
except AttributeError:
return np.array([d.strftime(fmt) for d in col], dtype=object)
def get_kl_data(self, dataframe:DataFrame):
return self.cal_kl_data(dataframe)
def _push_klu_into_klc_list(self, klc_list, klu, last_klu):
"""把一根 KLU 并入包含K线列表。与 get_klc_list 的几何规则相同。"""
if len(klc_list) > 0:
last_klc = klc_list[-1]
if klu.exception:
ddir = Chan_KLINE_DIR.DOWN
if last_klc.high < klu.high:
ddir = Chan_KLINE_DIR.UP
klc = ChanKLC(klu, index=len(klc_list), ddir=ddir)
klc.high = klu.close if klu.close > klu.open else klu.open
klc.low = klu.open if klu.close > klu.open else klu.close
klc_list.append(klc)
last_klc.set_next(klc)
klc.set_pre(last_klc)
last_klc.set_end_klu(last_klu)
klc.set_pre_fx()
else:
included = last_klc.check_klu_included(klu)
if not included:
ddir = Chan_KLINE_DIR.DOWN
if last_klc.high < klu.high:
ddir = Chan_KLINE_DIR.UP
klc = ChanKLC(klu, index=len(klc_list), ddir=ddir)
klc_list.append(klc)
last_klc.set_next(klc)
klc.set_pre(last_klc)
last_klc.set_end_klu(last_klu)
klc.set_pre_fx()
else:
last_klc.add_klu(klu)
else:
ddir = Chan_KLINE_DIR.UP
if klu.open > klu.close:
ddir = Chan_KLINE_DIR.DOWN
klc = ChanKLC(klu, 0, ddir)
klc_list.append(klc)
def get_klc_list(self, klu_list):
klc_list = []
# ChanMACD.__init__ 已调用 cal_macd_state，切勿再调一次（会重复堆积 seg/unittf）。
# lean 模式跳过整套 MACD 状态机：它只服务于 bsp/背驰/web 展示，笔与中枢不依赖它。
if getattr(self, 'lean', False):
self._last_chan_macd = None
else:
macd = ChanMACD(klu_list)
klu_list = macd.klu_list
self._last_chan_macd = macd
last_klu = None
for klu in klu_list:
self._push_klu_into_klc_list(klc_list, klu, last_klu)
last_klu = klu
# cal_trend 只产出 klc.trend，而全仓只有它自己（经 prev_klcs 读自身序列
# 状态）、一个 __str__ 和 web 的 klc_trend 图层消费它——笔与中枢不读。
# 增量路径的 klc 其 trend 恒为 UNKNOWN 却与批量构建逐字段相同，是实证。
# 所以 lean 下可跳；web 走非 lean，图层不受影响。
if not getattr(self, 'lean', False):
klc_list = self.cal_trend(klc_list)
return klc_list
def get_klu_list(self, dataframe):
klu_list = self.get_kl_data(dataframe)
#klu_list = self.cal_klu_pattern(klu_list)
return klu_list
def cal_klu_pattern(self, klu_list):
"""
计算裸K的pattern - 识别反转形态
"""
if not klu_list or len(klu_list) < 3:
return klu_list
for i, klu in enumerate(klu_list):
# 单根K线反转模式识别
self._detect_single_reversal_pattern(klu)
# 双根K线形态识别
if i >= 1:
self._detect_double_pattern(klu_list[i-1], klu)
# 三根K线形态识别
if i >= 2:
self._detect_triple_pattern(klu_list[i-2], klu_list[i-1], klu)
#if klu.pattern != Chan_KLU_PATTERN.UNKNOWN:
#print(klu.time, klu.pattern, klu.lower_shadow_ratio, klu.upper_shadow_ratio, klu.body_ratio, klu.lower_shadow_ratio/klu.body_ratio, klu.upper_shadow_ratio/klu.body_ratio)
return klu_list
def _detect_single_reversal_pattern(self, klu):
"""检测单根K线反转模式"""
body = abs(klu.close - klu.open)
upper_shadow = klu.high - max(klu.close, klu.open)
lower_shadow = min(klu.close, klu.open) - klu.low
total_range = klu.high - klu.low
# 避免除零
if total_range == 0:
return
body_ratio = body / total_range
upper_ratio = upper_shadow / total_range
lower_ratio = lower_shadow / total_range
#print(klu.time, upper_ratio, lower_ratio, body_ratio, upper_ratio/body_ratio, lower_ratio/body_ratio)
# 避免body_ratio为0时的除零错误
if body_ratio == 0:
return
# 锤子线/上吊线 - 反转信号
if lower_ratio / body_ratio >= 2:
# 锤子线：底部反转，需要前面一段
if klu.close > klu.open and klu.pre:
klu.set_pattern(Chan_KLU_PATTERN.HAMMER) # 底部反转
# 上吊线：顶部反转，需要前一根是上涨趋势
elif klu.close < klu.open and klu.pre:
klu.set_pattern(Chan_KLU_PATTERN.HANGING_MAN) # 顶部反转
# 倒锤子线/射击之星 - 反转信号
elif upper_ratio / body_ratio >= 2:
# 倒锤子线：底部反转，需要前一根是下跌趋势
if klu.close > klu.open and klu.pre:
klu.set_pattern(Chan_KLU_PATTERN.INVERTED_HAMMER) # 底部反转
# 射击之星：顶部反转，需要前一根是上涨趋势
elif klu.close < klu.open and klu.pre:
klu.set_pattern(Chan_KLU_PATTERN.SHOOTING_STAR) # 顶部反转
# 十字星 - 反转信号
elif body_ratio <= 0.1:
if upper_ratio > 0.4 and lower_ratio > 0.4:
klu.set_pattern(Chan_KLU_PATTERN.LONG_LEGGED_DOJI) # 强烈反转信号
elif upper_ratio > 0.4 and lower_ratio <= 0.1:
# 墓碑十字星：顶部反转，需要前一根是上涨趋势
if klu.pre and klu.pre.close > klu.pre.open:
klu.set_pattern(Chan_KLU_PATTERN.GRAVESTONE_DOJI) # 顶部反转
elif lower_ratio > 0.4 and upper_ratio <= 0.1:
# 蜻蜓十字星：底部反转，需要前一根是下跌趋势
if klu.pre and klu.pre.close < klu.pre.open:
klu.set_pattern(Chan_KLU_PATTERN.DRAGONFLY_DOJI) # 底部反转
else:
klu.set_pattern(Chan_KLU_PATTERN.DOJI) # 一般反转信号
def _detect_double_pattern(self, prev_klu, curr_klu):
"""检测两根K线形成的形态
包括：吞没形态(看涨/看跌)、乌云盖顶、曙光初现
"""
# 如果前一根K线已经有形态，不再识别双K线形态
if prev_klu.pattern != Chan_KLU_PATTERN.UNKNOWN:
return
# 计算K线实体
prev_body = abs(prev_klu.close - prev_klu.open)
curr_body = abs(curr_klu.close - curr_klu.open)
# 判断K线颜色（阴阳）
prev_bullish = prev_klu.close > prev_klu.open
curr_bullish = curr_klu.close > curr_klu.open
# 检查是否存在长期趋势（至少需要5根K线的趋势）
def check_long_trend(klu, bullish_trend=True, min_bars=5):
"""检查是否存在长期趋势
bullish_trend=True: 检查上涨趋势
bullish_trend=False: 检查下跌趋势
min_bars: 最少需要多少根K线形成趋势
"""
if not klu or not klu.pre:
return False
return True
# 使用EMA指标判断长期趋势
if klu.ema52 > 0:
if bullish_trend and klu.close < klu.ema52:
return False
if not bullish_trend and klu.close > klu.ema52:
return False
# 检查连续的K线方向
count = 0
current = klu.pre
while current and count < min_bars:
if not current.pre:
break
if bullish_trend:
# 上涨趋势：当前收盘价高于前一根收盘价
if current.close <= current.pre.close:
break
else:
# 下跌趋势：当前收盘价低于前一根收盘价
if current.close >= current.pre.close:
break
count += 1
current = current.pre
return count >= min_bars
# 1. 看涨吞没形态：前阴后阳，后者完全吞没前者
# 要求前面有明显的下跌趋势
if not prev_bullish and curr_bullish and \
abs(curr_klu.open - prev_klu.close) < 10 and \
curr_klu.close > prev_klu.open and \
check_long_trend(prev_klu, bullish_trend=False, min_bars=5):
curr_klu.set_pattern(Chan_KLU_PATTERN.BULLISH_ENGULFING)
return
# 2. 看跌吞没形态：前阳后阴，后者完全吞没前者
# 要求前面有明显的上涨趋势
if prev_bullish and not curr_bullish and \
abs(curr_klu.open - prev_klu.close) < 10 and \
curr_klu.close < prev_klu.open and \
check_long_trend(prev_klu, bullish_trend=True, min_bars=5):
curr_klu.set_pattern(Chan_KLU_PATTERN.BEARISH_ENGULFING)
return
# 3. 乌云盖顶：前阳后阴，后者开盘价高于前者最高价，收盘价在前者实体中部以下
# 要求前面有明显的上涨趋势
if prev_bullish and not curr_bullish and \
curr_klu.open > prev_klu.high and \
curr_klu.close < (prev_klu.open + prev_klu.close) / 2 and \
curr_klu.close > prev_klu.open and \
check_long_trend(prev_klu, bullish_trend=True, min_bars=5):
curr_klu.set_pattern(Chan_KLU_PATTERN.DARK_CLOUD_COVER)
return
# 4. 曙光初现：前阴后阳，后者开盘价低于前者最低价，收盘价在前者实体中部以上
# 要求前面有明显的下跌趋势
if not prev_bullish and curr_bullish and \
curr_klu.open < prev_klu.low and \
curr_klu.close > (prev_klu.open + prev_klu.close) / 2 and \
curr_klu.close < prev_klu.open and \
check_long_trend(prev_klu, bullish_trend=False, min_bars=5):
curr_klu.set_pattern(Chan_KLU_PATTERN.PIERCING_LINE)
return
# 平顶和平底移至三根K线形态中判断
def _detect_triple_pattern(self, first_klu, second_klu, third_klu):
"""检测三根K线形成的形态
包括：早晨之星、黄昏之星、平顶、平底
"""
# 如果前两根K线已经有形态，不再识别三K线形态
if first_klu.pattern != Chan_KLU_PATTERN.UNKNOWN or \
second_klu.pattern != Chan_KLU_PATTERN.UNKNOWN:
return
# 判断K线颜色（阴阳）
first_bullish = first_klu.close > first_klu.open
second_bullish = second_klu.close > second_klu.open
third_bullish = third_klu.close > third_klu.open
# 计算实体大小
first_body = abs(first_klu.close - first_klu.open)
second_body = abs(second_klu.close - second_klu.open)
third_body = abs(third_klu.close - third_klu.open)
# 检查是否存在长期趋势（至少需要5根K线的趋势）
def check_long_trend(klu, bullish_trend=True, min_bars=5):
"""检查是否存在长期趋势
bullish_trend=True: 检查上涨趋势
bullish_trend=False: 检查下跌趋势
min_bars: 最少需要多少根K线形成趋势
"""
if not klu or not klu.pre:
return False
# 使用EMA指标判断长期趋势
if klu.ema52 > 0:
if bullish_trend and klu.close < klu.ema52:
return False
if not bullish_trend and klu.close > klu.ema52:
return False
# 检查连续的K线方向
count = 0
current = klu.pre
while current and count < min_bars:
if not current.pre:
break
if bullish_trend:
# 上涨趋势：当前收盘价高于前一根收盘价
if current.close <= current.pre.close:
break
else:
# 下跌趋势：当前收盘价低于前一根收盘价
if current.close >= current.pre.close:
break
count += 1
current = current.pre
return count >= min_bars
# 1. 早晨之星：第一根阴线，第二根十字星或小实体，第三根阳线
# 要求前面有明显的下跌趋势
if not first_bullish and third_bullish and \
second_body < first_body * 0.3 and \
third_body > first_body * 0.5 and \
max(second_klu.open, second_klu.close) < first_klu.close and \
min(second_klu.open, second_klu.close) < third_klu.open and \
third_klu.close > (first_klu.open + first_klu.close) / 2 and \
check_long_trend(first_klu, bullish_trend=False, min_bars=7):
third_klu.set_pattern(Chan_KLU_PATTERN.MORNING_STAR)
return
# 2. 黄昏之星：第一根阳线，第二根十字星或小实体，第三根阴线
# 要求前面有明显的上涨趋势
if first_bullish and not third_bullish and \
second_body < first_body * 0.3 and \
third_body > first_body * 0.5 and \
min(second_klu.open, second_klu.close) > first_klu.close and \
max(second_klu.open, second_klu.close) > third_klu.open and \
third_klu.close < (first_klu.open + first_klu.close) / 2 and \
check_long_trend(first_klu, bullish_trend=True, min_bars=7):
third_klu.set_pattern(Chan_KLU_PATTERN.EVENING_STAR)
return
# 3. 平顶：三根K线的最高点几乎相同（上升趋势中更有意义）
# 要求前面有明显的上涨趋势
if (abs(first_klu.high - second_klu.high) / first_klu.high < 0.0002 and
abs(second_klu.high - third_klu.high) / second_klu.high < 0.0002 and
check_long_trend(first_klu, bullish_trend=True, min_bars=7)):
# 额外确认：价格接近阻力位或关键技术指标
is_near_resistance = False
# 检查是否接近EMA52阻力位
if first_klu.ema52 > 0:
resistance_level = first_klu.ema52
if abs(first_klu.high - resistance_level) / resistance_level < 0.01:
is_near_resistance = True
# 检查是否有成交量确认（成交量减少表示上涨动能减弱）
volume_confirmation = False
if (first_klu.volume > 0 and second_klu.volume > 0 and third_klu.volume > 0 and
third_klu.volume < second_klu.volume and second_klu.volume < first_klu.volume):
volume_confirmation = True
if is_near_resistance or volume_confirmation:
third_klu.set_pattern(Chan_KLU_PATTERN.TWEEZER_TOP)
return
# 4. 平底：三根K线的最低点几乎相同（下降趋势中更有意义）
# 要求前面有明显的下跌趋势
if (abs(first_klu.low - second_klu.low) / first_klu.low < 0.0002 and
abs(second_klu.low - third_klu.low) / second_klu.low < 0.0002 and
check_long_trend(first_klu, bullish_trend=False, min_bars=7)):
# 额外确认：价格接近支撑位或关键技术指标
is_near_support = False
# 检查是否接近EMA52支撑位
if first_klu.ema52 > 0:
support_level = first_klu.ema52
if abs(first_klu.low - support_level) / support_level < 0.01:
is_near_support = True
# 检查是否有成交量确认（成交量减少表示下跌动能减弱）
volume_confirmation = False
if (first_klu.volume > 0 and second_klu.volume > 0 and third_klu.volume > 0 and
third_klu.volume < second_klu.volume and second_klu.volume < first_klu.volume):
volume_confirmation = True
if is_near_support or volume_confirmation:
third_klu.set_pattern(Chan_KLU_PATTERN.TWEEZER_BOTTOM)
return
Reference in New Issue View Git Blame Copy Permalink