jackyu66gitandCursor
df27b4dde8
将 web/services/runtime.py 拆为 runtime/ 子模块并保持门面兼容；补齐 ESS 文档、门面/契约/TF_DF 测试与 CODE_REVIEW Approve。 Co-authored-by: Cursor <cursoragent@cursor.com>
275 lines
9.1 KiB
Python
275 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import talib.abstract as 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
|
|
|