★ FEATURED ARTICLE 特防科技反谍技术研究院2026-09-27阅读约 35 分钟1. 背景从技术链路到实战数据前三篇文章分别讲了五类设备的信号特征、三代技术路线对比、悟空系统五步技术链路。本篇进入实战数据复盘——327次检测的工程实现和数据分析代码。本文代码基于Python 3.10涉及numpy、scipy、scikit-learn、paho-mqtt。数据来自特防科技反谍技术研究院悟空系统的327次真实检测记录已脱敏处理。本文代码结构 1. 检测数据采集与清洗 2. 设备类型分布统计 3. 藏匿位置聚类分析 4. 检出率与误报率计算 5. 金箍系统白名单基线库 6. 金箍系统EKF轨迹追踪 7. 金箍系统SPC异常行为检测2. 检测数据采集与清洗2.1 数据结构定义每次检测产生一份检测记录包含场景信息、布点信息、设备清单、检出率等字段。以下为数据结构定义。# 检测记录数据结构 from dataclasses import dataclass, field from typing import List, Optional from datetime import datetime dataclass class DeviceRecord: 单个检测到的设备记录 device_id: str device_type: str # BLE / WiFi / GSM / 4G_CatM / Other protocol: str # BLE 5.0 / 802.11n / GSM900 ... freq_mhz: float rssi_dbm: float pos_x: float pos_y: float pos_z: float confidence: float location_type: str # 藏匿位置类型 dataclass class DetectionRecord: 一次完整检测的记录 record_id: str scene_type: str # office / meeting_room / hotel / residence scan_time: datetime scan_duration_min: int node_count: int devices: List[DeviceRecord] field(default_factorylist) detection_rate: Optional[float] None false_positive_rate: Optional[float] None2.2 数据加载与清洗# 加载327次检测记录 import json import pandas as pd from pathlib import Path def load_detection_records(data_dir: str) - pd.DataFrame: 加载检测记录返回扁平化的DataFrame 参数: data_dir: 检测记录JSON文件所在目录 返回: DataFrame每行一个设备记录 records [] for file in Path(data_dir).glob(*.json): with open(file, r, encodingutf-8) as f: data json.load(f) for device in data[devices]: records.append({ record_id: data[record_id], scene_type: data[scene_type], scan_time: data[scan_time], scan_duration_min: data[scan_duration_min], node_count: data[node_count], device_type: device[device_type], protocol: device[protocol], freq_mhz: device[freq_mhz], rssi_dbm: device[rssi_dbm], pos_x: device[pos_x], pos_y: device[pos_y], pos_z: device[pos_z], confidence: device[confidence], location_type: device[location_type] }) df pd.DataFrame(records) # 清洗剔除置信度低于0.5的误报 df df[df[confidence] 0.5] # 清洗剔除RSSI低于-95dBm的弱信号可能是噪声 df df[df[rssi_dbm] -95] return df # 使用示例 df load_detection_records(./detection_records/) print(f总记录数: {len(df)}) print(f场景分布:\n{df[scene_type].value_counts()})3. 设备类型分布统计3.1 整体分布# 设备类型整体分布 def analyze_device_distribution(df: pd.DataFrame) - pd.DataFrame: 统计设备类型分布 total len(df) dist df.groupby(device_type).agg( count(device_id, count), avg_rssi(rssi_dbm, mean), avg_confidence(confidence, mean) ).reset_index() dist[percentage] (dist[count] / total * 100).round(1) dist dist.sort_values(percentage, ascendingFalse) return dist # 运行结果 # device_type count avg_rssi avg_confidence percentage # BLE ... -72.3 0.89 38.0 # WiFi ... -58.6 0.91 27.0 # GSM ... -45.2 0.94 18.0 # 4G_CatM ... -78.4 0.85 12.0 # Other ... -70.1 0.82 5.03.2 各场景下的类型分布# 各场景下的设备类型分布 def analyze_scene_distribution(df: pd.DataFrame) - pd.DataFrame: 按场景统计设备类型分布 pivot df.pivot_table( indexdevice_type, columnsscene_type, valuesdevice_id, aggfunccount, fill_value0 ) # 转换为百分比每列除以该列总和 pivot_pct pivot.div(pivot.sum(axis0), axis1) * 100 return pivot_pct.round(1) # 运行结果示例 # scene_type meeting_room office hotel residence # BLE 48.0 42.0 22.0 18.0 # WiFi 20.0 24.0 38.0 36.0 # GSM 12.0 14.0 22.0 26.0 # 4G_CatM 14.0 15.0 12.0 14.0关键发现办公室和会议室场景中BLE占比接近一半42%和48%酒店和住宅场景中WiFi占比最高38%和36%。这说明不同场景的威胁类型存在显著差异检测方案的布点策略应因场景而异。4. 藏匿位置聚类分析藏匿位置的分布不是随机的存在明显的聚集特征。以下代码对藏匿位置进行聚类分析。# 藏匿位置聚类分析 from sklearn.cluster import KMeans import numpy as np def analyze_location_distribution(df: pd.DataFrame) - pd.DataFrame: 统计藏匿位置分布 total len(df) loc_dist df.groupby(location_type).agg( count(device_id, count), avg_confidence(confidence, mean) ).reset_index() loc_dist[percentage] (loc_dist[count] / total * 100).round(1) loc_dist loc_dist.sort_values(percentage, ascendingFalse) return loc_dist # 按设备类型统计藏匿位置偏好 def location_preference_by_type(df: pd.DataFrame) - pd.DataFrame: 分析不同设备类型的藏匿位置偏好 pivot df.pivot_table( indexlocation_type, columnsdevice_type, valuesdevice_id, aggfunccount, fill_value0 ) # 计算每个位置中不同设备类型的占比 pivot_pct pivot.div(pivot.sum(axis1), axis0) * 100 return pivot_pct.round(1) # 运行结果 # location_type BLE WiFi GSM 4G_CatM # 电源插座/充电器内部 45.2 30.1 10.3 12.1 # 天花板检修口 38.5 22.4 18.7 16.2 # 通风管道 35.1 25.8 15.4 19.5 # 会议桌下方/桌腿内部 72.3 8.5 5.2 10.4 # 绿植盆栽内部 62.8 15.3 8.7 10.2关键发现会议桌下方/桌腿内部的BLE占比高达72.3%绿植盆栽内部的BLE占比62.8%。这说明BLE设备更倾向于藏在“近距离、小体积、不易被注意”的位置。而电源插座/充电器内部则是各类设备共同偏好的位置BLE占比45.2%WiFi占比30.1%。5. 检出率与误报率计算5.1 检出率计算# 检出率计算 def calculate_detection_rate(df: pd.DataFrame, ground_truth: pd.DataFrame) - pd.DataFrame: 计算检出率 参数: df: 检测结果DataFrame ground_truth: 真实设备列表通过物理排查确认 返回: 各场景下的检出率统计 results [] for scene in df[scene_type].unique(): scene_df df[df[scene_type] scene] scene_gt ground_truth[ground_truth[scene_type] scene] # 真实存在的设备 true_devices set(scene_gt[device_id]) # 系统检测到的设备 detected set(scene_df[scene_df[confidence] 0.5][device_id]) # 检出率 检测到的真实设备数 / 真实设备总数 true_positive detected true_devices detection_rate len(true_positive) / len(true_devices) * 100 if true_devices else 0 # 误报率 检测到但不在真实列表中的设备数 / 检测到的总设备数 false_positive detected - true_devices false_positive_rate len(false_positive) / len(detected) * 100 if detected else 0 results.append({ scene_type: scene, true_devices: len(true_devices), detected_devices: len(detected), true_positive: len(true_positive), false_positive: len(false_positive), detection_rate: round(detection_rate, 1), false_positive_rate: round(false_positive_rate, 1) }) return pd.DataFrame(results)5.2 按设备类型的检出率# 按设备类型计算检出率 def detection_rate_by_type(df: pd.DataFrame, ground_truth: pd.DataFrame) - pd.DataFrame: 按设备类型计算检出率 results [] for device_type in ground_truth[device_type].unique(): type_gt ground_truth[ground_truth[device_type] device_type] true_ids set(type_gt[device_id]) type_df df[df[device_type] device_type] detected_ids set(type_df[type_df[confidence] 0.5][device_id]) true_positive detected_ids true_ids rate len(true_positive) / len(true_ids) * 100 if true_ids else 0 results.append({ device_type: device_type, total: len(true_ids), detected: len(true_positive), detection_rate: round(rate, 1) }) return pd.DataFrame(results).sort_values(detection_rate, ascendingFalse) # 运行结果示例 # device_type total detected detection_rate # GSM ... ... 95% # WiFi ... ... 95% # BLE ... ... 90% # 4G_CatM ... ... 85%6. 金箍系统白名单基线库金箍系统在悟空系统基础上增加了三个核心模块。第一个是设备指纹基线库用于建立合法设备白名单。6.1 白名单基线库实现# 设备指纹基线库 from collections import defaultdict from datetime import datetime, timedelta import numpy as np class BaselineLibrary: 设备指纹基线库7天学习期建立白名单 def __init__(self, learning_days: int 7): self.learning_days learning_days self.learning_start None self.device_profiles defaultdict(list) # MAC - 信号特征列表 self.whitelist {} # MAC - 设备指纹 def start_learning(self): 开始学习期 self.learning_start datetime.now() print(f学习期开始: {self.learning_start}) def collect(self, mac: str, features: dict): 采集设备信号特征 if self.learning_start is None: raise RuntimeError(请先调用 start_learning()) self.device_profiles[mac].append({ timestamp: datetime.now(), freq_mhz: features[freq_mhz], rssi_dbm: features[rssi_dbm], duty_cycle: features[duty_cycle], pos_x: features[pos_x], pos_y: features[pos_y], pos_z: features[pos_z] }) def finalize_learning(self): 学习期结束生成白名单 if self.learning_start is None: raise RuntimeError(学习期未开始) elapsed datetime.now() - self.learning_start if elapsed.days self.learning_days: print(f学习期未满: {elapsed.days}/{self.learning_days}天) # 对每个设备进行特征提取和行为建模 for mac, profiles in self.device_profiles.items(): # 过滤出现次数少于10次的设备视为临时设备 if len(profiles) 10: continue freqs [p[freq_mhz] for p in profiles] rssis [p[rssi_dbm] for p in profiles] duties [p[duty_cycle] for p in profiles] positions np.array([[p[pos_x], p[pos_y], p[pos_z]] for p in profiles]) # 时间规律统计设备出现的时段 hours [p[timestamp].hour for p in profiles] self.whitelist[mac] { freq_mean: np.mean(freqs), freq_std: np.std(freqs), rssi_mean: np.mean(rssis), rssi_std: np.std(rssis), duty_mean: np.mean(duties), pos_mean: positions.mean(axis0), pos_std: positions.std(axis0), active_hours: set(hours), sample_count: len(profiles) } print(f白名单生成完成共 {len(self.whitelist)} 个合法设备) def is_authorized(self, mac: str) - bool: 判断设备是否在白名单中 return mac in self.whitelist def check_anomaly(self, mac: str, features: dict) - dict: 检查设备是否异常针对白名单内设备 if mac not in self.whitelist: return {anomaly: True, reason: 不在白名单中} profile self.whitelist[mac] anomalies [] # 检查信号强度是否异常 if abs(features[rssi_dbm] - profile[rssi_mean]) 3 * profile[rssi_std]: anomalies.append(信号强度异常) # 检查占空比是否异常 if features[duty_cycle] 2 * profile[duty_mean]: anomalies.append(占空比异常) # 检查时间是否异常 current_hour datetime.now().hour if current_hour not in profile[active_hours]: anomalies.append(非常规时段出现) return { anomaly: len(anomalies) 0, reasons: anomalies }7. 金箍系统EKF轨迹追踪7.1 扩展卡尔曼滤波实现# EKF轨迹追踪 import numpy as np from typing import Tuple class EKFTracker: 扩展卡尔曼滤波连续估计移动设备轨迹 def __init__(self, process_noise0.1, measurement_noise1.0): 参数: process_noise: 过程噪声Q measurement_noise: 观测噪声R self.Q process_noise self.R measurement_noise self.state None # [x, y, z, vx, vy, vz] self.covariance None def initialize(self, position: np.ndarray): 初始化状态 self.state np.array([ position[0], position[1], position[2], 0, 0, 0 ]) self.covariance np.eye(6) * 1.0 def predict(self, dt: float 1.0): 预测步骤 # 状态转移矩阵匀速模型 F np.eye(6) F[0, 3] dt F[1, 4] dt F[2, 5] dt # 预测状态 self.state F self.state # 预测协方差 self.covariance F self.covariance F.T np.eye(6) * self.Q def update(self, measurement: np.ndarray): 更新步骤 # 测量矩阵只观测位置 H np.zeros((3, 6)) H[0, 0] 1 H[1, 1] 1 H[2, 2] 1 # 卡尔曼增益 S H self.covariance H.T np.eye(3) * self.R K self.covariance H.T np.linalg.inv(S) # 更新状态 y measurement - H self.state self.state self.state K y # 更新协方差 self.covariance (np.eye(6) - K H) self.covariance def get_position(self) - np.ndarray: 获取当前位置估计 return self.state[:3] def get_velocity(self) - np.ndarray: 获取当前速度估计 return self.state[3:] # 使用示例 tracker EKFTracker(process_noise0.1, measurement_noise1.0) tracker.initialize(np.array([5.0, 3.0, 2.5])) for measurement in measurements: tracker.predict(dt1.0) tracker.update(measurement) pos tracker.get_position() vel tracker.get_velocity() print(f位置: ({pos[0]:.1f}, {pos[1]:.1f}, {pos[2]:.1f}), f速度: ({vel[0]:.2f}, {vel[1]:.2f}, {vel[2]:.2f}))7.2 轨迹分类# 轨迹分类静止 / 周期性 / 异常移动 def classify_trajectory(positions: np.ndarray, timestamps: np.ndarray) - str: 对轨迹进行分类 参数: positions: N×3数组轨迹点 timestamps: N维数组时间戳 返回: static / periodic / anomalous if len(positions) 10: return insufficient_data # 计算位移总量 displacements np.linalg.norm(np.diff(positions, axis0), axis1) total_displacement displacements.sum() # 计算速度 time_diffs np.diff(timestamps) velocities displacements / time_diffs avg_velocity velocities.mean() # 静止设备位移小速度低 if total_displacement 1.0 and avg_velocity 0.01: return static # 周期性设备位移有规律位置在有限范围内变化 pos_range positions.max(axis0) - positions.min(axis0) if np.all(pos_range 3.0): return periodic # 异常移动设备位移大速度变化剧烈 return anomalous8. 金箍系统SPC异常行为检测8.1 统计过程控制实现# SPC异常行为检测 from collections import deque import numpy as np class SPCDetector: 基于统计过程控制的异常检测 def __init__(self, window_size3600, sigma_threshold3, consecutive_points3): 参数: window_size: 滑动窗口大小秒默认1小时 sigma_threshold: 控制限倍数默认3倍标准差 consecutive_points: 连续超限点数触发告警默认3个 self.window_size window_size self.sigma_threshold sigma_threshold self.consecutive_points consecutive_points self.buffers defaultdict(lambda: deque(maxlenwindow_size)) self.consecutive_counts defaultdict(int) def update(self, device_id: str, value: float, timestamp: float) - dict: 更新SPC检测 参数: device_id: 设备标识 value: 当前观测值如RSSI timestamp: 时间戳秒 返回: {anomaly: bool, reason: str} buffer self.buffers[device_id] buffer.append(value) # 至少需要30个采样点才能建立控制限 if len(buffer) 30: return {anomaly: False, reason: insufficient_data} values np.array(buffer) mean values.mean() std values.std() # 计算控制限 ucl mean self.sigma_threshold * std lcl mean - self.sigma_threshold * std # 判断是否超出控制限 if value ucl or value lcl: self.consecutive_counts[device_id] 1 else: self.consecutive_counts[device_id] 0 # 连续超限触发告警 if self.consecutive_counts[device_id] self.consecutive_points: return { anomaly: True, reason: f连续{self.consecutive_points}个采样点超出控制限, value: value, ucl: ucl, lcl: lcl } return {anomaly: False, reason: normal}8.2 异常事件检测器# 异常事件检测器综合白名单SPC轨迹 class AnomalyDetector: 金箍系统异常行为检测器 def __init__(self, baseline: BaselineLibrary): self.baseline baseline self.spc_detectors defaultdict(lambda: SPCDetector()) self.trackers {} def process(self, device_id: str, mac: str, features: dict, timestamp: float) - dict: 处理一条设备信号返回异常检测结果 anomalies [] # 1. 白名单检查 if not self.baseline.is_authorized(mac): anomalies.append({ type: new_device, level: 即时告警, message: 不在白名单中的新设备 }) else: # 白名单内设备检查异常行为 result self.baseline.check_anomaly(mac, features) if result[anomaly]: for reason in result[reasons]: anomalies.append({ type: behavior_anomaly, level: 可能被人激活, message: reason }) # 2. SPC检测 spc_result self.spc_detectors[device_id].update( device_id, features[rssi_dbm], timestamp ) if spc_result[anomaly]: anomalies.append({ type: signal_anomaly, level: 信号异常, message: spc_result[reason] }) # 3. 时间异常检测 current_hour datetime.fromtimestamp(timestamp).hour if current_hour 22 or current_hour 6: anomalies.append({ type: time_anomaly, level: 高风险告警, message: 非常规时段出现 }) return { device_id: device_id, mac: mac, timestamp: timestamp, anomalies: anomalies, has_anomaly: len(anomalies) 0 }9. 完整处理链路从检测记录到异常告警# 完整处理链路从检测记录到异常告警 def full_pipeline(data_dir: str, output_dir: str): 完整处理链路 # Step 1: 加载数据 df load_detection_records(data_dir) print(f加载 {len(df)} 条设备记录) # Step 2: 设备类型分布 dist analyze_device_distribution(df) dist.to_csv(f{output_dir}/device_distribution.csv, indexFalse) print(f设备类型分布:\n{dist}) # Step 3: 场景分布 scene_dist analyze_scene_distribution(df) scene_dist.to_csv(f{output_dir}/scene_distribution.csv) print(f场景分布:\n{scene_dist}) # Step 4: 藏匿位置分布 loc_dist analyze_location_distribution(df) loc_dist.to_csv(f{output_dir}/location_distribution.csv, indexFalse) print(f藏匿位置分布:\n{loc_dist}) # Step 5: 检出率计算 # 需要 ground truth 数据 # detection_rate calculate_detection_rate(df, ground_truth) # Step 6: 金箍系统初始化 baseline BaselineLibrary(learning_days7) detector AnomalyDetector(baseline) print(处理完成) # 运行 full_pipeline(./detection_records/, ./output/)10. 小结本文给出了327次反窃听检测数据的工程实现和数据分析代码。核心结论BLE设备占比38%是检出数量最多的类型办公室和会议室场景中BLE占比接近一半42%和48%藏匿位置第一名是电源插座/充电器内部占比31%会议桌下方/桌腿内部的BLE占比最高达72.3%悟空系统检出率95%BLE检出率90%误报率5%金箍系统通过白名单基线、EKF轨迹追踪、SPC异常检测实现7×24持续监测代码中涉及的关键技术点数据加载与清洗置信度0.5RSSI−95dBm设备分布统计按场景和设备类型交叉分析藏匿位置聚类分析不同设备类型的藏匿位置偏好白名单基线库7天学习期分三阶段建立EKF轨迹追踪六维状态向量1Hz更新频率SPC异常检测1小时滑动窗口3σ控制限连续3点触发本文数据来自特防科技反谍技术研究院《反窃听检测实战数据统计报告》2026基于悟空系统327次真实检测记录所有数据均已脱敏处理。金箍系统技术方案来自《多节点空间感知系统技术白皮书》2026。悟空系统为特防科技反谍技术研究院多节点空间感知方案的短期检测形态金箍系统为7×24持续监测形态。
阅读完成 · 觉得有帮助？
