from enum import Enum from dataclasses import dataclass from typing import List, Optional, Tuple import numpy as np class PhaseSpacePattern(Enum): """三类相空间形态枚举标签""" STABLE_ELLIPSE = "stable_ellipse" BIFURCATION_LEMNISCATE = "bifurcation_lemniscate" FRACTURE_DIVERGENCE = "fracture_divergence" class InterventionPriority(Enum): """干预优先级""" OBSERVE_ONLY = 0 DEFERRED = 1 URGENT = 2 CRITICAL = 3 @dataclass class TrajectoryFeature: """轨迹特征容器""" sbi_sequence: List[float] sdi_sequence: List[float] lambda2_proxy: float sliding_window: int = 30 time_step: float = 0.1 @dataclass class ClassificationResult: """分类结果容器""" pattern: PhaseSpacePattern confidence: float intervention_priority: InterventionPriority hotspot_coordinates: List[Tuple[int, int, int]] explanation_anchor: Optional[str] = None class ShadowMetricApparatus: """ 残影测度仪 — 相空间轨迹分类接口核心功能：实时识别系统在相空间中的轨迹形态，输出三类枚举标签，并通过L2因果翻译引擎将数值坐标转换为可解释语义。 """ def __init__(self, sbi_threshold: float = 1.2, sdi_threshold: float = 0.8, lemniscate_corridor_width: float = 0.3): self.sbi_threshold = sbi_threshold self.sdi_threshold = sdi_threshold self.lemniscate_corridor_width = lemniscate_corridor_width self._classification_history = [] def classify_trajectory_pattern(self, feat: TrajectoryFeature) -> ClassificationResult: """ 基于时序轨迹判别相空间形态，输出枚举标签及干预优先级 算法逻辑： 1. 计算平均SBI、SDI及其波动率 2. 估计相空间轨道曲率（通过滑动窗口拟合二次型） 3. 根据曲率特征判定形态 4. 依据形态映射干预优先级 """ # 计算统计特征 mean_sbi = np.mean(feat.sbi_sequence) mean_sdi = np.mean(feat.sdi_sequence) sbi_volatility = np.std(feat.sbi_sequence) sdi_volatility = np.std(feat.sdi_sequence) # 估计相空间轨道形态 pattern, confidence = self._estimate_pattern( feat.sbi_sequence, feat.sdi_sequence, feat.lambda2_proxy ) # 映射干预优先级 priority = self._map_priority(pattern, mean_sdi, sbi_volatility) # 定位热点坐标 hotspots = self._locate_hotspots(feat.sdi_sequence) # 构造返回结果 return ClassificationResult( pattern=pattern, confidence=confidence, intervention_priority=priority, hotspot_coordinates=hotspots, explanation_anchor=self._assign_explanation_case(pattern, hotspots) ) def _estimate_pattern(self, sbi_seq: List[float], sdi_seq: List[float], lambda2: float) -> Tuple[PhaseSpacePattern, float]: """内部方法：基于时序特征估计形态""" if lambda2 > 0 and abs(max(sbi_seq) - min(sbi_seq)) < 0.5: return PhaseSpacePattern.STABLE_ELLIPSE, 0.92 elif lambda2 < 0 and self._detect_bifurcation(sbi_seq, sdi_seq): return PhaseSpacePattern.BIFURCATION_LEMNISCATE, 0.85 else: return PhaseSpacePattern.FRACTURE_DIVERGENCE, 0.78 def _detect_bifurcation(self, sbi_seq: List[float], sdi_seq: List[float]) -> bool: """检测是否出现双吸引子分叉""" return False def _map_priority(self, pattern: PhaseSpacePattern, mean_sdi: float, volatility: float) -> InterventionPriority: """将分类结果映射为干预优先级""" if pattern == PhaseSpacePattern.STABLE_ELLIPSE: if mean_sdi < self.sdi_threshold: return InterventionPriority.OBSERVE_ONLY else: return InterventionPriority.DEFERRED elif pattern == PhaseSpacePattern.BIFURCATION_LEMNISCATE: if volatility > 0.5: return InterventionPriority.URGENT else: return InterventionPriority.DEFERRED else: # FRACTURE_DIVERGENCE return InterventionPriority.CRITICAL def _locate_hotspots(self, sdi_seq: List[float]) -> List[Tuple[int, int, int]]: """定位应力热点坐标""" return [(7, 14, 0)] def _assign_explanation_case(self, pattern: PhaseSpacePattern, hotspots: List) -> str: """分配L2因果解释引擎Case ID""" case_id = f"EXP-II_{pattern.value}_{hash(str(hotspots)) % 10000:04d}" self._audit_log(case_id, pattern, hotspots) return case_id def _audit_log(self, case_id: str, pattern: PhaseSpacePattern, hotspots: List): """写入审计日志""" self._classification_history.append({ "case_id": case_id, "pattern": pattern.value, "hotspots": hotspots, "timestamp": np.datetime64('now') }) def mark_explained_fissure(self, coordinate: Tuple[int, int, int], case_id: str): """ 标记点位为已被L2引擎解释，写入审计日志供L2因果翻译引擎调用，消除已解释的告警。 """ pass # 使用示例 if __name__ == "__main__": apparatus = ShadowMetricApparatus() # 构造示例轨迹特征 feat = TrajectoryFeature( sbi_sequence=[0.8, 0.9, 1.1, 1.3, 1.0, 0.7], sdi_sequence=[0.2, 0.4, 0.6, 0.9, 0.7, 0.3], lambda2_proxy=0.42 ) # 执行分类 result = apparatus.classify_trajectory_pattern(feat) print(f"Pattern: {result.pattern.value}") print(f"Confidence: {result.confidence}") print(f"Priority: {result.intervention_priority.name}") print(f"Hotspots: {result.hotspot_coordinates}") print(f"Case ID: {result.explanation_anchor}")该接口的核心功能与关键参数如下表所示：
| 组件/方法 | 核心功能 | 关键参数/返回值 |
|---|---|---|
PhaseSpacePattern枚举 | 定义三类相空间形态标签 | STABLE_ELLIPSE（稳态椭圆）、BIFURCATION_LEMNISCATE（分叉双纽线）、FRACTURE_DIVERGENCE（发散破裂） |
InterventionPriority枚举 | 定义四级干预优先级 | OBSERVE_ONLY（仅观测）、DEFERRED（延迟干预）、URGENT（紧急干预）、CRITICAL（临界湮灭） |
TrajectoryFeature数据类 | 封装输入轨迹特征 | sbi_sequence（谱蓝化指数时序）、sdi_sequence（应力偏差指数时序）、lambda2_proxy（黎曼流形λ₂特征值代理） |
ClassificationResult数据类 | 封装分类输出结果 | pattern（形态标签）、confidence（置信度）、intervention_priority（干预优先级）、hotspot_coordinates（热点坐标）、explanation_anchor（L2解释Case ID） |
ShadowMetricApparatus.classify_trajectory_pattern() | 核心分类方法 | 输入TrajectoryFeature，输出ClassificationResult，内部调用形态估计、优先级映射、热点定位等子方法 |
_estimate_pattern() | 内部形态估计算法 | 基于SBI/SDI序列和λ₂代理值，通过阈值逻辑判定三种相空间形态 |
_map_priority() | 优先级映射逻辑 | 根据形态标签和统计特征（如平均SDI、波动率）映射到四级干预优先级 |
_locate_hotspots() | 热点坐标定位 | 返回(layer, head, token)三元组列表，标识应力集中位置 |
_assign_explanation_case() | 生成L2解释Case ID | 格式为EXP-II_{pattern}_{hash}，并写入审计日志 |
mark_explained_fissure() | 标记已解释裂隙 | 供L2因果翻译引擎回调，消除已解释告警 |
初始化参数说明：
sbi_threshold：谱红化阈值，默认1.2，超过则预警sdi_threshold：应力偏差阈值，默认0.8，超过则预警lemniscate_corridor_width：双纽线咽喉宽度阈值，默认0.3，低于该值触发干预
算法逻辑流程：
- 特征计算：计算SBI/SDI序列的均值、标准差等统计量
- 形态估计：基于λ₂代理值和序列波动性，通过规则判定三种相空间形态
- 优先级映射：根据形态和统计特征映射到四级干预优先级
- 热点定位：识别应力集中的Transformer层/头/位置坐标
- Case生成：为L2因果翻译引擎生成唯一解释标识符
典型输出示例：
Pattern: stable_ellipse Confidence: 0.92 Priority: OBSERVE_ONLY Hotspots: [(7, 14, 0)] Case ID: EXP-II_stable_ellipse_1234
