Chan/web/services/runtime/analyze.py

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)
	#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)
	
	# 添加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,  # 添加买卖点列表
		'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