import numpy as np from dataclasses import dataclass, field from enum import Enum from typing import List, Optional, Tuple # 全局常量配置写入元协议固化参数 WINDOW_LEN 300 STEP_SIZE 30 DRIFT_SIGMA_THRESH 2.0STABLE_WINDOW_REQUIRED 3 SDI_LEAD_TIME 4.1 SRI_LEAD_TIME 2.0 # 代谢区间划分 class MetabolicZone(Enum): HIGH high MID middle LOW low # 三阶耦合阈值矩阵 THRESHOLD_MATRIX { MetabolicZone.HIGH: { lambda2_sigma: 0.6, sri_scale: 0.5, sdi_scale: 0.5 }, MetabolicZone.MID: { lambda2_sigma: 0.4, sri_scale: 1.0, sdi_scale: 1.0 }, MetabolicZone.LOW: { lambda2_sigma: 0.2, sri_scale: 1.5, sdi_scale: 1.5 } } dataclass class WindowSnapshot: timestamp_start: int timestamp_end: int lambda2_mean: float lambda2_std: float global_drift: float is_drifted: bool False is_permanent_anchor: bool False dataclass class WarningTriple: lambda2_inflection: Optional[int] None sri_rise_start: Optional[int] None sdi_cross_time: Optional[int] None # 1. 滑窗基线管理器解决自指基线漂移 class BaselineTrajectoryManager: def __init__(self): self.window_queue: List[WindowSnapshot] [] self.permanent_anchors: List[Tuple[float, float]] [] # (mean, std) def push_new_window(self, t_now: int, lambda2_series: np.ndarray): win_mean np.mean(lambda2_series) win_std np.std(lambda2_series) drift self._calc_global_drift(win_mean) drifted drift DRIFT_SIGMA_THRESH snap WindowSnapshot( timestamp_startt_now - WINDOW_LEN, timestamp_endt_now, lambda2_meanwin_mean, lambda2_stdwin_std, global_driftdrift, is_drifteddrifted ) self.window_queue.append(snap) self._check_anchor_candidate() def _calc_global_drift(self, current_mean: float) - float: if not self.permanent_anchors: return 0.0 ref_mean, ref_std self.permanent_anchors[-1] return abs(current_mean - ref_mean) / ref_std def _check_anchor_candidate(self): # 连续N个无漂移窗口固化为永久基线锚点 stable_count 0 for snap in reversed(self.window_queue): if not snap.is_drifted: stable_count 1 else: break if stable_count STABLE_WINDOW_REQUIRED: latest self.window_queue[-1] latest.is_permanent_anchor True self.permanent_anchors.append((latest.lambda2_mean, latest.lambda2_std)) def get_active_baseline(self) - Tuple[float, float]: if self.permanent_anchors: return self.permanent_anchors[-1] return self.window_queue[-1].lambda2_mean, self.window_queue[-1].lambda2_std # 2. 时序语法解析器预警句子化拒绝孤立指标 class TemporalGrammarParser: def __init__(self): self.triple WarningTriple() self.false_positive_pool: List[WarningTriple] [] def mark_lambda2_inflection(self, t: int): self.triple.lambda2_inflection t def mark_sri_rising(self, t: int): self.triple.sri_rise_start t def mark_sdi_cross(self, t: int): self.triple.sdi_cross_time t def is_legal_warning(self) - Tuple[bool, str]: t_l2 self.triple.lambda2_inflection t_sri self.triple.sri_rise_start t_sdi self.triple.sdi_cross_time if None in (t_l2, t_sri, t_sdi): self.false_positive_pool.append(self.triple) return False, Missing grammar component # 校验严格时序 超前时差匹配 cond_order t_l2 t_sri t_sdi cond_offset (t_sdi - t_l2) SDI_LEAD_TIME0.5 cond_sri_offset (t_sdi - t_sri) SRI_LEAD_TIME0.3 if cond_order and cond_offset and cond_sri_offset: return True, Valid sequential alert else: self.false_positive_pool.append(self.triple) return False, Sequence disorder / offset mismatch def reset_triple(self): self.triple WarningTriple() # 3. 代谢ξ三阶阈值控制器 class XiThresholdController: staticmethod def get_zone(xi: float) - MetabolicZone: if xi 0.8: return MetabolicZone.HIGH elif 0.4 xi 0.8: return MetabolicZone.MID else: return MetabolicZone.LOW staticmethod def resolve_thresholds(xi: float, base_mean: float, base_std: float): zone XiThresholdController.get_zone(xi) cfg THRESHOLD_MATRIX[zone] l2_thresh base_mean - cfg[lambda2_sigma] * base_std s2_scale cfg[sri_scale] sd_scale cfg[sdi_scale] return l2_thresh, s2_scale, sd_scale # 4. 顶层自指调度总入口 class SelfReferGuardian: def __init__(self): self.baseline_mgr BaselineTrajectoryManager() self.grammar_parser TemporalGrammarParser() self.xi_controller XiThresholdController() def step_loop(self, t_now: int, lambda2: float, sri: float, sdi: float, xi: float, window_buffer: np.ndarray): # 1. 每滑动步长更新一次基线 if t_now % STEP_SIZE 0: self.baseline_mgr.push_new_window(t_now, window_buffer) base_mean, base_std self.baseline_mgr.get_active_baseline() # 2. 根据代谢状态动态解算阈值 l2_thresh, sri_scale, sdi_scale self.xi_controller.resolve_thresholds(xi, base_mean, base_std) # 3. 指标拐点标记 if lambda2 l2_thresh and self.grammar_parser.triple.lambda2_inflection is None: self.grammar_parser.mark_lambda2_inflection(t_now) if sri (1.0 * sri_scale) and self.grammar_parser.triple.sri_rise_start is None: self.grammar_parser.mark_sri_rising(t_now) if sdi (2.5 * sdi_scale) and self.grammar_parser.triple.sdi_cross_time is None: self.grammar_parser.mark_sdi_cross(t_now) # 4. 语法校验 输出指令 valid, reason self.grammar_parser.is_legal_warning() output { timestamp: t_now, valid_alert: valid, reason: reason, meta: { baseline_drift: self.baseline_mgr.window_queue[-1].global_drift, metabolic_zone: XiThresholdController.get_zone(xi).value, false_pos_count: len(self.grammar_parser.false_positive_pool) } } # 合法预警下发干预无论是否合法都保留假阳性作为认知边界样本 if valid: self.dispatch_intervention(output) self.grammar_parser.reset_triple() return output def dispatch_intervention(self, alert_msg): # 对接EXP-II三级预警总线 if alert_msg[meta][metabolic_zone] low: level 3 elif alert_msg[meta][metabolic_zone] middle: level 2 else: level 1 # 推送至归藏层干预执行器 print(f[GUICANG ALERT] Level {level} | {alert_msg[reason]})参考来源数学分析(九)-定积分4-定积分的性质2-1-积分中值定理2积分第一中值定理的几何意义【f(ξ)[1/(b-a)]·∫ₐᵇf(x)dx可理解为f(x)在区间[a,b]上所有函数值的平均值】别再死记公式了用PythonSPICE仿真直观理解运放频率响应中的Q与ξ区间套定理别再死记公式了用PythonSPICE仿真直观理解运放频率响应中的Q与ξ别再死记公式了用PythonNumPy手把手分析运放频率响应直观理解Q和ξ
