jackyu66gitandCursor
2cd50f1e01
三周期分开关控制，只画数字不画图标；线段面积比沿用同向笔面积口径。 Co-authored-by: Cursor <cursoragent@cursor.com>
289 lines
9.7 KiB
Python
289 lines
9.7 KiB
Python
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
from chanlun.indicators import ta
|
||
|
||
from chanlun import TF_DF
|
||
from chanlun.core.ChanEnum import Chan_KLC_FX, Chan_FX_TYPE
|
||
from chanlun.indicators.ChanMACD import ChanMACD
|
||
|
||
from .indicators import calculate_macd
|
||
|
||
def analyze_chan(df, symbol=None, timeframe=None):
|
||
"""进行缠论分析"""
|
||
chan = TF_DF()
|
||
|
||
# 初始化多时间周期数据以获取EMA52
|
||
ema52_dict = None
|
||
# 获取分析结果
|
||
klu_list = chan.get_kl_data(df)
|
||
klc_list = chan.get_klc_list(klu_list)
|
||
bi_list = chan.cal_bi_list(klc_list)
|
||
#for index in range(0, 10):
|
||
#print(bi_list[index].start_time, bi_list[index].start_klc.end_time, bi_list[index].dir)
|
||
seg_list = chan.get_seg_list(bi_list)
|
||
zs_list = chan.calculate_seg_zs(seg_list)
|
||
# 计算笔中枢（BI中枢）并拍平成列表
|
||
|
||
bi_zs_list = chan.cal_bi_zs_list_pure(bi_list)
|
||
#bi_zs_list = chan.cal_bi_zs(seg_list)
|
||
bsp_list = []
|
||
if len(bi_zs_list) > 0:
|
||
bsp_list = chan.find_all_bsp(bi_list, bi_zs_list)
|
||
# 第四类买卖点（B4/S4）：复用上面刚算好的 bi_zs_list，不重复建中枢。
|
||
# 大级别由同一份 df 重采样得到，失败时退化为空列表，不拖垮主分析。
|
||
fast_bsp_list = []
|
||
try:
|
||
if len(bi_zs_list) > 0:
|
||
fast_bsp_list = chan.cal_fast_bsp(df=df, bi_zs_list=bi_zs_list, timeframe=timeframe)
|
||
except Exception as e:
|
||
print(f"第四类买卖点计算出错: {e}")
|
||
fast_bsp_list = []
|
||
#bsp_state_list = chan.get_bsp_state(df)
|
||
#for bsp in bsp_list:
|
||
#print(bsp.end_time, bsp.type, bsp.dir)
|
||
# 添加买卖点识别
|
||
for bi in bi_list:
|
||
bi.cal_macdhist()
|
||
for bi in bi_list:
|
||
bi.cal_macd_div()
|
||
#print(bi.start_time, bi.macd_hist, bi.macd_div)
|
||
for seg in seg_list:
|
||
seg.cal_macdhist()
|
||
for seg in seg_list:
|
||
seg.cal_macd_div()
|
||
|
||
# 添加ChanMACD分析（复用 get_klc_list 内已算好的结果，避免同周期二次全量分析）
|
||
chan_macd = None
|
||
chan_macd_data = {}
|
||
try:
|
||
|
||
if klu_list and len(klu_list) > 0:
|
||
print(f"获取到KLU列表，长度: {len(klu_list)}")
|
||
chan_macd = getattr(chan, '_last_chan_macd', None)
|
||
if chan_macd is None:
|
||
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,
|
||
'klu_list': chan_macd.klu_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线分型信息
|
||
klc_fx_info = []
|
||
for klc in klc_list:
|
||
if hasattr(klc, 'klc_fx_type') and klc.klc_fx_type != Chan_KLC_FX.UNKNOWN:
|
||
try:
|
||
# 计算分型强度
|
||
fx_strength = 0
|
||
fx_strength_level = ""
|
||
is_strong_fx = False
|
||
|
||
# 统一使用cal_fx_strength函数
|
||
if hasattr(klc, 'cal_fx_strength'):
|
||
fx_strength = klc.cal_fx_strength(5)
|
||
|
||
# 尝试获取分型强度等级
|
||
if hasattr(klc, 'get_fx_strength_level'):
|
||
fx_strength_level = klc.get_fx_strength_level()
|
||
|
||
# 尝试判断是否为强分型
|
||
if hasattr(klc, 'is_strong_fx'):
|
||
is_strong_fx = klc.is_strong_fx()
|
||
|
||
# 如果分型强度小于1，设为0
|
||
if fx_strength < 1:
|
||
fx_strength = 0
|
||
|
||
# KLC 分型框（起止时间+高低价）：
|
||
# 仅使用 cal_fx_box 通过 display 条件后生成的 klc.fx_box。
|
||
# 若无 fx_box，则前端不应绘制分型框。
|
||
fx_box = getattr(klc, 'fx_box', None)
|
||
box_start_time = getattr(fx_box, 'start_time', None) if fx_box else None
|
||
box_end_time = getattr(fx_box, 'end_time', None) if fx_box else None
|
||
box_high = getattr(fx_box, 'high', None) if fx_box else None
|
||
box_low = getattr(fx_box, 'low', None) if fx_box else None
|
||
|
||
if klc.bb_out:
|
||
klc_fx_info.append({
|
||
'time': klc.end_time,
|
||
'price': klc.low if klc.fx == Chan_FX_TYPE.BOTTOM else klc.high,
|
||
'fx_type': str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""),
|
||
'is_bottom': klc.fx == Chan_FX_TYPE.BOTTOM,
|
||
'fx_strength': fx_strength, # 分型强度分数 (0-100)
|
||
'fx_strength_level': fx_strength_level, # 分型强度等级 (极强/强/中等/弱/极弱)
|
||
'is_strong_fx': is_strong_fx, # 是否为强分型
|
||
|
||
# 虚线分型框信息（给前端画框用）
|
||
'start_time': box_start_time,
|
||
'end_time': box_end_time,
|
||
'high': float(box_high) if box_high is not None else None,
|
||
'low': float(box_low) if box_low is not None else None,
|
||
})
|
||
except Exception as e:
|
||
# 如果出错，仍然添加基本信息，但分型强度为0
|
||
fx_box = getattr(klc, 'fx_box', None)
|
||
box_start_time = getattr(fx_box, 'start_time', None) if fx_box else None
|
||
box_end_time = getattr(fx_box, 'end_time', None) if fx_box else None
|
||
box_high = getattr(fx_box, 'high', None) if fx_box else None
|
||
box_low = getattr(fx_box, 'low', None) if fx_box else None
|
||
|
||
klc_fx_info.append({
|
||
'time': klc.end_time,
|
||
'price': klc.low if klc.fx == Chan_FX_TYPE.BOTTOM else klc.high,
|
||
'fx_type': str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""),
|
||
'is_bottom': klc.fx == Chan_FX_TYPE.BOTTOM,
|
||
'fx_strength': 0,
|
||
'fx_strength_level': "",
|
||
'is_strong_fx': False,
|
||
|
||
# 虚线分型框信息（给前端画框用）
|
||
'start_time': box_start_time,
|
||
'end_time': box_end_time,
|
||
'high': float(box_high) if box_high is not None else None,
|
||
'low': float(box_low) if box_low is not None else None,
|
||
})
|
||
|
||
|
||
return {
|
||
'klc_list': klc_list,
|
||
'klu_list': klu_list, # 添加KLU列表
|
||
'bi_list': bi_list,
|
||
'seg_list': seg_list,
|
||
'zs_list': zs_list,
|
||
'bi_zs_list': bi_zs_list, # 添加BI中枢列表
|
||
'bsp_list': bsp_list, # 添加买卖点列表
|
||
'fast_bsp_list': fast_bsp_list, # 第四类买卖点（B4/S4）
|
||
'klc_fx_info': klc_fx_info, # KLC分型信息
|
||
'chan_macd': chan_macd_data, # 添加ChanMACD分析数据
|
||
'ema52_dict': ema52_dict # 添加多时间周期EMA52数据
|
||
}
|
||
|
||
def classify_trend_stage(df):
|
||
"""根据 EMA 斜率与多空排列判断趋势方向与阶段
|
||
返回: direction in {"bull","bear","sideways"}, stage in {"early","mid","late"}, strength_score (0-100)
|
||
"""
|
||
if df is None or len(df) < 60:
|
||
return "sideways", "early", 0
|
||
|
||
# 使用 EMA5/10/24/52
|
||
closes = df['close'].values
|
||
ema5 = df['ema5'].values if 'ema5' in df else ta.EMA(df, timeperiod=5)
|
||
ema10 = df['ema10'].values if 'ema10' in df else ta.EMA(df, timeperiod=10)
|
||
ema24 = df['ema24'].values if 'ema24' in df else ta.EMA(df, timeperiod=24)
|
||
ema52 = df['ema52'].values if 'ema52' in df else ta.EMA(df, timeperiod=52)
|
||
|
||
# 最近N根用于斜率与排列判定
|
||
lookback = min(30, len(df) - 1)
|
||
if lookback <= 5:
|
||
return "sideways", "early", 0
|
||
|
||
# 简单斜率: 最近k根的线性变化率近似
|
||
def slope(arr, k=10):
|
||
k = min(k, len(arr) - 1)
|
||
if k < 2:
|
||
return 0.0
|
||
y = arr[-k:]
|
||
x = np.arange(k)
|
||
# 最小二乘拟合斜率
|
||
denom = np.dot(x - x.mean(), x - x.mean())
|
||
if denom == 0:
|
||
return 0.0
|
||
m = np.dot(y - y.mean(), x - x.mean()) / denom
|
||
return float(m)
|
||
|
||
k_slope = 12 # 斜率窗口
|
||
s5 = slope(ema5, k_slope)
|
||
s10 = slope(ema10, k_slope)
|
||
s24 = slope(ema24, k_slope)
|
||
s52 = slope(ema52, k_slope)
|
||
|
||
# 多空排列
|
||
last5, last10, last24, last52 = ema5[-1], ema10[-1], ema24[-1], ema52[-1]
|
||
bull_stack = last5 > last10 > last24 > last52
|
||
bear_stack = last5 < last10 < last24 < last52
|
||
|
||
# 波动性与动量增强: MACD 柱体最近均值
|
||
macdhist = df['macdhist'].values if 'macdhist' in df else calculate_macd(df)['histogram']
|
||
hist_recent = macdhist[-lookback:]
|
||
hist_power = float(np.mean(np.abs(hist_recent))) if len(hist_recent) else 0.0
|
||
|
||
# 方向
|
||
if bull_stack and s24 > 0 and s52 > 0:
|
||
direction = "bull"
|
||
elif bear_stack and s24 < 0 and s52 < 0:
|
||
direction = "bear"
|
||
else:
|
||
# 用价格相对 EMA52 辅助
|
||
if closes[-1] > last52 and (s24 + s52) > 0:
|
||
direction = "bull"
|
||
elif closes[-1] < last52 and (s24 + s52) < 0:
|
||
direction = "bear"
|
||
else:
|
||
direction = "sideways"
|
||
|
||
# 阶段: 依据（斜率大小、与EMA52距离、MACD柱体扩张/收敛）
|
||
dist52 = float((closes[-1] - last52) / last52) if last52 else 0.0
|
||
slope_score = max(0.0, (abs(s24) + abs(s52)) * 1000.0) # 归一化
|
||
dist_score = min(50.0, abs(dist52) * 200.0)
|
||
hist_score = min(30.0, hist_power * 10.0)
|
||
strength = float(min(100.0, slope_score + dist_score + hist_score))
|
||
|
||
# 简单阶段判定
|
||
if direction == "sideways":
|
||
stage = "early"
|
||
strength = min(strength, 30.0)
|
||
else:
|
||
# 查看最近 hist 是否在扩大或收敛
|
||
if len(hist_recent) >= 6:
|
||
recent_growth = np.mean(np.abs(hist_recent[-3:])) - np.mean(np.abs(hist_recent[-6:-3]))
|
||
else:
|
||
recent_growth = 0.0
|
||
|
||
if recent_growth > 0 and abs(dist52) < 0.05:
|
||
stage = "early"
|
||
elif recent_growth > 0 and abs(dist52) >= 0.05:
|
||
stage = "mid"
|
||
else:
|
||
stage = "late"
|
||
|
||
return direction, stage, strength
|
||
|