import numpy as np from dataclasses import dataclass from enum import Enum from typing import List, Optional, Tuple import time # 全局常量配置写入元协议固化参数 WINDOW_LEN 300 STEP_SIZE 30 DRIFT_SIGMA_THRESH 2.0STD_TREND_THRESHOLD 1.15 STABLE_WINDOW_REQUIRED 3 # 假阳性池全局配置 FP_MAX_SIZE 2000 FP_EXPIRE_DAYS 7 FP_SIMILAR_THRESH 0.15 # 代谢区间划分 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, sdi_lead_time: 3.0, sri_lead_time: 1.2 }, MetabolicZone.MID: { lambda2_sigma: 0.4, sri_scale: 1.0, sdi_scale: 1.0, sdi_lead_time: 4.1, sri_lead_time: 2.0 }, MetabolicZone.LOW: { lambda2_sigma: 0.2, sri_scale: 1.5, sdi_scale: 1.5, sdi_lead_time: 5.5, sri_lead_time: 2.8 } } # 假阳性池生命周期管理模块 dataclass(slotsTrue) class FalsePositiveEntry: timestamp: float lambda2: float sri: float sdi: float xi: float fp_reason: str feature_hash: int weight: float 1.0 class FalsePositivePool: __slots__ (max_size, age_threshold_days, similarity_threshold, pool) def __init__(self, max_size: int FP_MAX_SIZE, age_threshold_days: int FP_EXPIRE_DAYS, similarity_threshold: float FP_SIMILAR_THRESH): self.max_size max_size self.age_threshold_days age_threshold_days self.similarity_threshold similarity_threshold self.pool: List[FalsePositiveEntry] [] def add_candidate(self, lambda2: float, sri: float, sdi: float, xi: float, fp_reason: str) - bool: fhash self._compute_hash(lambda2, sri, sdi) # 相似样本只更新权重和时间戳不新增 for entry in self.pool: if self._is_similar(entry, lambda2, sri, sdi): entry.weight min(entry.weight 0.1, 3.0) entry.timestamp time.time() return False # 容量溢出触发淘汰 if len(self.pool) self.max_size: self._evict() if len(self.pool) self.max_size: self.pool.append(FalsePositiveEntry( timestamptime.time(), lambda2lambda2, srisri, sdisdi, xixi, fp_reasonfp_reason, feature_hashfhash )) return True return False def _evict(self): now time.time() # 第一步清理过期样本 expire_sec self.age_threshold_days * 86400 self.pool [e for e in self.pool if (now - e.timestamp) expire_sec] # 第二步按权重时间升序截断 if len(self.pool) self.max_size: self.pool.sort(keylambda e: e.weight * e.timestamp, reverseFalse) self.pool self.pool[:self.max_size] # 第三步冗余去重 self._dedup() def _dedup(self): if len(self.pool) 2: return remove_idx set() for i in range(len(self.pool)): for j in range(i 1, len(self.pool)): if self._is_similar(self.pool[i], self.pool[j].lambda2, self.pool[j].sri, self.pool[j].sdi): # 保留更新时间更近的一条 if self.pool[i].timestamp self.pool[j].timestamp: remove_idx.add(j) else: remove_idx.add(i) self.pool [e for idx, e in enumerate(self.pool) if idx not in remove_idx] def get_stats(self) - dict: if not self.pool: return { size: 0, avg_weight: 0.0, age_days_max: 0.0, fp_by_reason: {} } now time.time() avg_w np.mean([e.weight for e in self.pool]) max_age_d max([(now - e.timestamp) / 86400 for e in self.pool]) reason_cnt {} for entry in self.pool: reason_cnt[entry.fp_reason] reason_cnt.get(entry.fp_reason, 0) 1 return { size: len(self.pool), avg_weight: round(avg_w, 3), age_days_max: round(max_age_d, 2), fp_by_reason: reason_cnt } def _compute_hash(self, lambda2: float, sri: float, sdi: float) - int: return hash((round(lambda2, 3), round(sri, 3), round(sdi, 3))) def _is_similar(self, entry: FalsePositiveEntry, l2: float, sr: float, sd: float) - bool: v1 np.array([entry.lambda2, entry.sri, entry.sdi]) v2 np.array([l2, sr, sd]) return np.linalg.norm(v1 - v2) self.similarity_threshold # 数据结构体 dataclass class WindowSnapshot: timestamp_start: int timestamp_end: int lambda2_mean: float lambda2_std: float global_mean_drift: float global_std_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]] [] def push_new_window(self, t_now: int, lambda2_series: np.ndarray): win_mean np.mean(lambda2_series) win_std np.std(lambda2_series) mean_drift self._calc_mean_drift(win_mean) std_drift self._calc_std_trend_drift(win_std) drifted (mean_drift DRIFT_SIGMA_THRESH) or (std_drift STD_TREND_THRESHOLD) snap WindowSnapshot( timestamp_startt_now - WINDOW_LEN, timestamp_endt_now, lambda2_meanwin_mean, lambda2_stdwin_std, global_mean_driftmean_drift, global_std_driftstd_drift, is_drifteddrifted ) self.window_queue.append(snap) self._check_anchor_candidate() def _calc_mean_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 _calc_std_trend_drift(self, current_std: float) - float: if not self.permanent_anchors: return 1.0 _, ref_std self.permanent_anchors[-1] if ref_std 1e-9: return 1.0 return current_std / ref_std def _check_anchor_candidate(self): 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() 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, sdi_lead: float, sri_lead: float) - 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): return False, Missing grammar component cond_order t_l2 t_sri t_sdi cond_offset (t_sdi - t_l2) sdi_lead0.5 cond_sri_offset (t_sdi - t_sri) sri_lead0.3 if cond_order and cond_offset and cond_sri_offset: return True, Valid sequential alert else: 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 sri_scale cfg[sri_scale] sdi_scale cfg[sdi_scale] sdi_lead cfg[sdi_lead_time] sri_lead cfg[sri_lead_time] return l2_thresh, sri_scale, sdi_scale, sdi_lead, sri_lead # 4. 顶层自指调度总入口全模块集成完毕 class SelfReferGuardian: def __init__(self, modeshadow): self.mode mode self.baseline_mgr BaselineTrajectoryManager() self.grammar_parser TemporalGrammarParser() self.xi_controller XiThresholdController() # 挂载带生命周期的假阳性池 self.fp_pool FalsePositivePool() def step_loop(self, t_now: int, lambda2: float, sri: float, sdi: float, xi: float, window_buffer: np.ndarray): # 滑动步长触发基线更新 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() # 动态阈值动态超前时差解算 l2_thresh, sri_scale, sdi_scale, sdi_lead, sri_lead self.xi_controller.resolve_thresholds(xi, base_mean, base_std) # 拐点打点 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) # 语法校验 valid, reason self.grammar_parser.is_legal_warning(sdi_lead, sri_lead) latest_win self.baseline_mgr.window_queue[-1] if self.baseline_mgr.window_queue else None # 判定为假阳性则入池留存 if not valid: self.fp_pool.add_candidate(lambda2, sri, sdi, xi, reason) fp_stats self.fp_pool.get_stats() output { timestamp: t_now, valid_alert: valid, reason: reason, meta: { baseline_mean_drift: latest_win.global_mean_drift if latest_win else 0, baseline_std_drift: latest_win.global_std_drift if latest_win else 0, baseline_is_drifted: latest_win.is_drifted if latest_win else False, metabolic_zone: XiThresholdController.get_zone(xi).value, false_positive: fp_stats, dynamic_lead_times: { sdi_lead: sdi_lead, sri_lead: sri_lead } } } if valid: self.dispatch_intervention(output) self.grammar_parser.reset_triple() return output def dispatch_intervention(self, alert_msg): # 对接EXP-II三级预警总线 zone alert_msg[meta][metabolic_zone] if zone low: level 3 elif zone middle: level 2 else: level 1 print(f[GUICANG ALERT] Level {level} | {alert_msg[reason]}) # 调用示例 if __name__ __main__: guardian SelfReferGuardian(modeshadow) time_step 500 buffer np.random.randn(WINDOW_LEN) result guardian.step_loop( t_nowtime_step, lambda20.21, sri0.72, sdi2.6, xi0.35, window_bufferbuffer ) print(result)参考来源【信息科学与工程学】【安全领域】第一百四十二篇 零信任网络解决方案中的算法04