""" 结构价值区 (Structure Zone) 系统 将多时间周期的 Chan 中枢边界 (ZD/ZG/GG/DD) 和 EMA52 统一表示为带强度评分的价值区对象。 """ from dataclasses import dataclass, field from typing import List, Dict, Optional, Any from datetime import datetime # ============================================================ # Dataclasses # ============================================================ @dataclass class RawZonePoint: """内部中间结构:从 Chan 中枢提取的单个价格点""" price: float timeframe: str # '5m', '1h', '4h' 等 structure_type: str # 'bi_zhongshu' | 'xd_zhongshu' | 'ema52' boundary_type: str # 'ZD' | 'ZG' | 'GG' | 'DD' | 'EMA52' source_zs_id: int # 来源 ZS 在列表中的 index(调试用) is_sure: bool # 来源 ZS 是否已完成 candle_time: Optional[str] = None # 来源 ZS 的 end_time(用于 recency 计算) @dataclass class StructureZone: """统一的价值区对象""" id: int lower: float upper: float center: float # (lower + upper) / 2 width_pct: float # (upper - lower) / center * 100 zone_type: str # 'support' | 'resistance' | 'neutral' timeframes: List[str] # 参与形成此区间的时间周期 structure_types: List[str] # 参与形成的结构类型 boundary_types: List[str] # 参与形成的边界类型 overlap_count: int # 聚类中的原始点数 touch_count: int # MVP: 等于 overlap_count recency_score: float # 0.0 - 1.0, 1.0 = 最近 ema52_distance_pct: float # 到最近 EMA52 的距离百分比 ema52_aligned: bool # 是否有 EMA52 落在区间内 strength_score: float # 0-100 综合评分 confidence: float # 0.0 - 1.0 first_seen: Optional[str] # 最早的 candle_time last_seen: Optional[str] # 最晚的 candle_time metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class StructureZoneConfig: """StructureZone 提取与评分配置""" cluster_radius_pct: float = 0.5 # 价格聚类半径(百分比) min_overlap_for_zone: int = 2 # 最少重叠点数才能形成区间 max_zones: int = 20 # 返回的最大区间数 recency_halflife_bars: int = 50 # recency 衰减半衰期(K线数) zone_timeframes: List[str] = field(default_factory=lambda: ['4h', '1h', '30m', '15m', '5m']) kl_lines_per_tf: int = 500 # 每个时间周期使用最近多少根K线 structure_weights: Dict[str, float] = field(default_factory=lambda: { 'bi_zhongshu': 1.0, # 笔中枢 — 最直接的价格行为 'xd_zhongshu': 0.8, # 线段中枢 — 较高级别但粒度较粗 'ema52': 0.4, # EMA — 趋势参考,弱于结构 }) # ============================================================ # Extraction # ============================================================ def extract_raw_points_from_tf_df( tf_df_dict: Dict[str, Any], ema_symbols: List[str], config: StructureZoneConfig, ) -> List[RawZonePoint]: """ 从 ChanLun.tf_df_dict 中提取所有原始价格点。 仅处理 config.zone_timeframes 中存在的时间周期。 """ points: List[RawZonePoint] = [] for tf_name in config.zone_timeframes: if tf_name not in tf_df_dict: continue tf_df = tf_df_dict[tf_name] # 1. 笔中枢 (ChanBIZS) try: if hasattr(tf_df, 'seg_list') and tf_df.seg_list: bi_zs_result = tf_df.cal_bi_zs(tf_df.seg_list) if bi_zs_result: _extract_from_zs_objects( points, tf_name, 'bi_zhongshu', bi_zs_result, config.kl_lines_per_tf ) except Exception: pass # 2. 线段中枢 (ChanZS) try: zs_list = getattr(tf_df, 'zs_list', None) if zs_list: _extract_from_zs_objects( points, tf_name, 'xd_zhongshu', zs_list, config.kl_lines_per_tf ) except Exception: pass # 3. EMA52 值 for tf_name in config.zone_timeframes: if tf_name in tf_df_dict: try: ema_val = tf_df_dict[tf_name].get_ema52() if ema_val is not None and ema_val > 0: points.append(RawZonePoint( price=float(ema_val), timeframe=tf_name, structure_type='ema52', boundary_type='EMA52', source_zs_id=-1, is_sure=True, candle_time=None, )) except Exception: pass return points def _extract_from_zs_objects( points: List[RawZonePoint], tf_name: str, structure_type: str, zs_list, kl_limit: int, ): """从 ZS 链表中提取 ZD/ZG/GG/DD 点""" count = 0 node = zs_list while hasattr(node, 'next'): node = node.next # 从链表头开始遍历 head = zs_list # 收集所有节点 all_nodes = [] cur = head while cur is not None and hasattr(cur, 'next'): all_nodes.append(cur) cur = cur.next # 只取最近 kl_limit 根K线内的 ZS all_nodes = all_nodes[-kl_limit:] if len(all_nodes) > kl_limit else all_nodes for idx, zs in enumerate(all_nodes): if not getattr(zs, 'is_sure', False): continue try: zg = float(zs.zg) zd = float(zs.zd) gg = float(zs.gg) if getattr(zs, 'gg', 0) else zg dd = float(zs.dd) if getattr(zs, 'dd', 0) else zd end_time = str(zs.end_time) if hasattr(zs, 'end_time') and zs.end_time else None except (ValueError, TypeError, AttributeError): continue if zg <= 0 or zd <= 0: continue zs_id = getattr(zs, 'index', idx) points.append(RawZonePoint(price=zg, timeframe=tf_name, structure_type=structure_type, boundary_type='ZG', source_zs_id=zs_id, is_sure=True, candle_time=end_time)) points.append(RawZonePoint(price=zd, timeframe=tf_name, structure_type=structure_type, boundary_type='ZD', source_zs_id=zs_id, is_sure=True, candle_time=end_time)) points.append(RawZonePoint(price=gg, timeframe=tf_name, structure_type=structure_type, boundary_type='GG', source_zs_id=zs_id, is_sure=True, candle_time=end_time)) points.append(RawZonePoint(price=dd, timeframe=tf_name, structure_type=structure_type, boundary_type='DD', source_zs_id=zs_id, is_sure=True, candle_time=end_time)) def extract_raw_points_from_serialized( analyses: Dict[str, Dict], ema52_dict: Dict[str, Optional[float]], config: StructureZoneConfig, ) -> List[RawZonePoint]: """ 从已序列化的分析结果中提取价格点(用于 web API,避免重复计算)。 analyses: {'5m': {'zs_list': [...], 'bi_zs_list': [...]}, '15m': {...}, ...} ema52_dict: {'5m': 123.45, '15m': None, ...} """ points: List[RawZonePoint] = [] for tf_name in config.zone_timeframes: if tf_name not in analyses: continue analysis = analyses[tf_name] # 笔中枢 bi_zs_items = analysis.get('bi_zs_list', []) for idx, zs in enumerate(bi_zs_items): if not zs.get('is_sure', False): continue try: zg = float(zs['zg']); zd = float(zs['zd']) gg = float(zs.get('gg', zg)); dd = float(zs.get('dd', zd)) end_time = zs.get('end_time') except (ValueError, KeyError): continue if zg <= 0 or zd <= 0: continue points.append(RawZonePoint(price=zg, timeframe=tf_name, structure_type='bi_zhongshu', boundary_type='ZG', source_zs_id=idx, is_sure=True, candle_time=str(end_time) if end_time else None)) points.append(RawZonePoint(price=zd, timeframe=tf_name, structure_type='bi_zhongshu', boundary_type='ZD', source_zs_id=idx, is_sure=True, candle_time=str(end_time) if end_time else None)) points.append(RawZonePoint(price=gg, timeframe=tf_name, structure_type='bi_zhongshu', boundary_type='GG', source_zs_id=idx, is_sure=True, candle_time=str(end_time) if end_time else None)) points.append(RawZonePoint(price=dd, timeframe=tf_name, structure_type='bi_zhongshu', boundary_type='DD', source_zs_id=idx, is_sure=True, candle_time=str(end_time) if end_time else None)) # 线段中枢 zs_items = analysis.get('zs_list', []) for idx, zs in enumerate(zs_items): if not zs.get('is_sure', False): continue try: zg = float(zs['zg']); zd = float(zs['zd']) gg = float(zs.get('gg', zg)); dd = float(zs.get('dd', zd)) end_time = zs.get('end_time') except (ValueError, KeyError): continue if zg <= 0 or zd <= 0: continue points.append(RawZonePoint(price=zg, timeframe=tf_name, structure_type='xd_zhongshu', boundary_type='ZG', source_zs_id=idx, is_sure=True, candle_time=str(end_time) if end_time else None)) points.append(RawZonePoint(price=zd, timeframe=tf_name, structure_type='xd_zhongshu', boundary_type='ZD', source_zs_id=idx, is_sure=True, candle_time=str(end_time) if end_time else None)) points.append(RawZonePoint(price=gg, timeframe=tf_name, structure_type='xd_zhongshu', boundary_type='GG', source_zs_id=idx, is_sure=True, candle_time=str(end_time) if end_time else None)) points.append(RawZonePoint(price=dd, timeframe=tf_name, structure_type='xd_zhongshu', boundary_type='DD', source_zs_id=idx, is_sure=True, candle_time=str(end_time) if end_time else None)) # EMA52 for tf_name in config.zone_timeframes: ema_val = ema52_dict.get(tf_name) if ema_val is not None and ema_val > 0: points.append(RawZonePoint( price=float(ema_val), timeframe=tf_name, structure_type='ema52', boundary_type='EMA52', source_zs_id=-1, is_sure=True, candle_time=None, )) return points # ============================================================ # Clustering # ============================================================ def cluster_raw_points( points: List[RawZonePoint], config: StructureZoneConfig, ) -> List[List[RawZonePoint]]: """ 贪心单通聚类:将价格相近的 RawZonePoint 归为一组。 仅在 1D 价格轴上操作,O(n log n)。 """ if not points: return [] sorted_points = sorted(points, key=lambda p: p.price) clusters: List[List[RawZonePoint]] = [] for p in sorted_points: placed = False for cluster in reversed(clusters): # 检查是否可以放入当前聚类(与聚类均价比较) avg_price = sum(pt.price for pt in cluster) / len(cluster) if abs(p.price - avg_price) / avg_price * 100 <= config.cluster_radius_pct: cluster.append(p) placed = True break if not placed: clusters.append([p]) # 过滤点数不足的聚类 return [c for c in clusters if len(c) >= config.min_overlap_for_zone] # ============================================================ # Scoring & Building # ============================================================ def build_structure_zones( clusters: List[List[RawZonePoint]], current_price: float, ema52_values: Dict[str, Optional[float]], latest_candle_time: Optional[str], config: StructureZoneConfig, ) -> List[StructureZone]: """ 从聚类构建 StructureZone 列表,计算所有字段和评分。 """ zones: List[StructureZone] = [] # 收集所有 EMA52 值 ema_prices = [v for v in ema52_values.values() if v is not None and v > 0] for zone_id, cluster in enumerate(clusters): prices = [p.price for p in cluster] lower = min(prices) upper = max(prices) center = (lower + upper) / 2 width_pct = (upper - lower) / center * 100 if center > 0 else 0.0 # 区间类型 if upper < current_price: zone_type = 'support' # 区间在当前价格下方 → 支撑 elif lower > current_price: zone_type = 'resistance' # 区间在当前价格上方 → 阻力 else: zone_type = 'neutral' # 区间跨越当前价格 timeframes = sorted(set(p.timeframe for p in cluster)) structure_types = sorted(set(p.structure_type for p in cluster)) boundary_types = sorted(set(p.boundary_type for p in cluster)) overlap_count = len(cluster) # Recency times = [p.candle_time for p in cluster if p.candle_time] first_seen = min(times) if times else None last_seen = max(times) if times else None recency_score = _calc_recency(last_seen, latest_candle_time, config.recency_halflife_bars) # EMA52 alignment ema52_distance_pct = 999.0 ema52_aligned = False if ema_prices: distances = [abs(center - ep) / ep * 100 for ep in ema_prices] ema52_distance_pct = round(min(distances), 2) ema52_aligned = any(lower <= ep <= upper for ep in ema_prices) # Strength score strength_score = _calc_strength(cluster, config, recency_score, ema52_aligned, ema52_distance_pct, width_pct) # Confidence confidence = _calc_confidence(overlap_count, len(timeframes), cluster) zones.append(StructureZone( id=zone_id + 1, lower=round(lower, 2), upper=round(upper, 2), center=round(center, 2), width_pct=round(width_pct, 2), zone_type=zone_type, timeframes=timeframes, structure_types=structure_types, boundary_types=boundary_types, overlap_count=overlap_count, touch_count=overlap_count, # MVP: 等于 overlap_count recency_score=round(recency_score, 3), ema52_distance_pct=ema52_distance_pct, ema52_aligned=ema52_aligned, strength_score=round(strength_score, 1), confidence=round(confidence, 2), first_seen=first_seen, last_seen=last_seen, )) # 按强度降序排列 zones.sort(key=lambda z: z.strength_score, reverse=True) # 截断 if config.max_zones > 0 and len(zones) > config.max_zones: zones = zones[:config.max_zones] return zones def _calc_recency( last_seen: Optional[str], latest_time: Optional[str], halflife_bars: int, ) -> float: """计算 recency 分数:越近越高""" if not last_seen or not latest_time: return 0.5 try: # 尝试解析 ISO 格式时间 from dateutil import parser t_last = parser.parse(last_seen) t_latest = parser.parse(latest_time) offset_seconds = (t_latest - t_last).total_seconds() if offset_seconds < 0: return 1.0 # 假设每根K线平均 5 分钟 bar_seconds = 300 offset_bars = offset_seconds / bar_seconds # 指数衰减: 2 ^ (-offset / halflife) score = 2.0 ** (-offset_bars / halflife_bars) return float(score) except Exception: return 0.5 def _calc_strength( cluster: List[RawZonePoint], config: StructureZoneConfig, recency_score: float, ema52_aligned: bool, ema52_distance_pct: float, width_pct: float, ) -> float: """计算综合强度评分 (0-100)""" # 组件 1: 结构类型多样性 (0-40) structure_type_counts: Dict[str, int] = {} for p in cluster: structure_type_counts[p.structure_type] = structure_type_counts.get(p.structure_type, 0) + 1 total = sum(structure_type_counts.values()) structure_score = 0.0 for st, count in structure_type_counts.items(): weight = config.structure_weights.get(st, 0.5) structure_score += weight * count structure_score = min(structure_score / max(1, total), 1.0) c1 = structure_score * 40 # 组件 2: 多周期确认 (0-25) tf_set = set(p.timeframe for p in cluster) tf_diversity = len(tf_set) c2 = min(tf_diversity / 5, 1.0) * 25 # 组件 3: 区间紧密度 (0-15) — 越窄越强 tightness = max(0.0, 1.0 - (width_pct / 3.0)) c3 = tightness * 15 # 组件 4: Recency (0-10) c4 = recency_score * 10 # 组件 5: EMA52 共振 (0-10) if ema52_aligned: ema_proximity = max(0.0, 1.0 - (ema52_distance_pct / 2.0)) c5 = ema_proximity * 10 else: c5 = 0.0 return c1 + c2 + c3 + c4 + c5 def _calc_confidence( overlap_count: int, tf_count: int, cluster: List[RawZonePoint], ) -> float: """计算置信度 (0-1)""" base = min(overlap_count / 6.0, 0.85) # 多周期加分 tf_bonus = min(tf_count / 5.0, 0.1) # 是否所有点都来自 sure 的 ZS all_sure = all(p.is_sure for p in cluster) sure_bonus = 0.05 if all_sure else 0.0 return min(base + tf_bonus + sure_bonus, 1.0) # ============================================================ # Top-level pipeline # ============================================================ def analyze_structure_zones( tf_df_dict: Dict[str, Any], ema_symbols: List[str], current_price: Optional[float] = None, config: Optional[StructureZoneConfig] = None, ) -> List[StructureZone]: """ 一站式分析:提取 → 聚类 → 评分 → 返回排序后的 StructureZone 列表。 """ if config is None: config = StructureZoneConfig() # 提取 raw_points = extract_raw_points_from_tf_df(tf_df_dict, ema_symbols, config) if not raw_points: return [] # 获取当前价格 if current_price is None: for tf_name in config.zone_timeframes: if tf_name in tf_df_dict: try: ema_val = tf_df_dict[tf_name].get_ema52() if ema_val and ema_val > 0: current_price = float(ema_val) break except Exception: pass if current_price is None: current_price = 0.0 # EMA52 值 ema52_values = {} for tf_name in config.zone_timeframes: if tf_name in tf_df_dict: try: ema52_values[tf_name] = tf_df_dict[tf_name].get_ema52() except Exception: ema52_values[tf_name] = None # 最晚时间 latest_time = None times = [p.candle_time for p in raw_points if p.candle_time] if times: latest_time = max(times) # 聚类 clusters = cluster_raw_points(raw_points, config) # 构建 & 评分 return build_structure_zones(clusters, current_price, ema52_values, latest_time, config) def analyze_structure_zones_from_serialized( analyses: Dict[str, Dict], ema52_dict: Dict[str, Optional[float]], current_price: float, config: Optional[StructureZoneConfig] = None, ) -> List[StructureZone]: """ 从已序列化的分析结果构建 StructureZone(用于 web API)。 """ if config is None: config = StructureZoneConfig() raw_points = extract_raw_points_from_serialized(analyses, ema52_dict, config) if not raw_points: return [] # 最晚时间 latest_time = None times = [p.candle_time for p in raw_points if p.candle_time] if times: latest_time = max(times) # EMA52 值(用于 alignment 检测) ema_values = {tf: v for tf, v in ema52_dict.items() if v is not None and v > 0} clusters = cluster_raw_points(raw_points, config) return build_structure_zones(clusters, current_price, ema_values, latest_time, config)