三阶阈值动态调控机制-三亩地-编程学习日记

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_start=t_now - WINDOW_LEN, timestamp_end=t_now, lambda2_mean=win_mean, lambda2_std=win_std, global_drift=drift, is_drifted=drifted ) 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]上所有函数值的平均值】
别再死记公式了！用Python+SPICE仿真，直观理解运放频率响应中的Q与ξ
区间套定理
别再死记公式了！用Python+SPICE仿真，直观理解运放频率响应中的Q与ξ
别再死记公式了！用Python+NumPy手把手分析运放频率响应，直观理解Q和ξ
日记详情

三阶阈值动态调控机制

参考来源
