AI时代的生物防御：从监测到响应的韧性体系

"""
minimal_bio_surveillance.py
最小化症候群异常监测管线——可独立运行，可扩展接入真实数据源

依赖: pip install numpy pandas scipy
"""

import numpy as np
import pandas as pd
from scipy import stats
from datetime import datetime, timedelta

# ── 1. 模拟数据：30天 × 5个区域的每日发热就诊数 ──
def generate_synthetic_data(days=30, regions=5, seed=42):
    np.random.seed(seed)
    base_rates = [120, 85, 200, 60, 150]  # 各区域基线就诊量
    dates = [datetime(2025, 5, 1) + timedelta(days=i) for i in range(days)]
    records = []
    for d in dates:
        for r in range(regions):
            # 正常波动 + 第20天在区域2注入一个异常突增
            noise = np.random.normal(0, base_rates[r] * 0.08)
            spike = 0
            if d.day == 20 and r == 2:
                spike = 90  # 模拟异常事件：额外90例
            count = int(base_rates[r] + noise + spike)
            records.append({"date": d, "region": f"R{r}", "fever_cases": max(count, 0)})
    return pd.DataFrame(records)

df = generate_synthetic_data()

# ── 2. 异常检测：滑动窗口Z-score法 ──
WINDOW = 7  # 用过去7天作为基线窗口
Z_THRESHOLD = 2.5  # Z-score超过2.5即标记异常

def detect_anomalies(group):
    """对单个区域的时序数据做滑动窗口Z-score检测"""
    group = group.sort_values("date")
    anomalies = []
    for i in range(WINDOW, len(group)):
        baseline = group["fever_cases"].iloc[i - WINDOW:i]
        current = group["fever_cases"].iloc[i]
        z = (current - baseline.mean()) / (baseline.std() + 1e-9)
        anomalies.append({
            "date": group["date"].iloc[i],
            "region": group["region"].iloc[i],
            "fever_cases": current,
            "baseline_mean": round(baseline.mean(), 1),
            "z_score": round(z, 2),
            "is_anomaly": abs(z) > Z_THRESHOLD,
        })
    return pd.DataFrame(anomalies)

anomaly_df = df.groupby("region").apply(detect_anomalies).reset_index(drop=True)

# ── 3. 风险评分：结合异常幅度与人口密度 ──
POPULATION_DENSITY = {"R0": 5000, "R1": 3000, "R2": 12000, "R3": 1500, "R4": 8000}
PATHOGEN_SEVERITY_WEIGHT = 1.5  # 呼吸道病原体严重度系数（可调）

def compute_risk(row):
    if not row["is_anomaly"]:
        return 0.0
    density = POPULATION_DENSITY.get(row["region"], 5000)
    # 风险 = 异常幅度 × 人口密度因子 × 严重度系数
    density_factor = np.log1p(density / 1000)
    risk = abs(row["z_score"]) * density_factor * PATHOGEN_SEVERITY_WEIGHT
    return round(min(risk, 10), 2)  # 风险分上限10

anomaly_df["risk_score"] = anomaly_df.apply(compute_risk, axis=1)

# ── 4. 输出：只显示异常事件 ──
alerts = anomaly_df[anomaly_df["is_anomaly"]].sort_values("risk_score", ascending=False)
print("=" * 70)
print("🚨 生物监测异常警报")
print("=" * 70)
for _, row in alerts.iterrows():
    level = "🔴高危" if row["risk_score"] >= 5 else "🟡关注"
    print(f"{row['date'].strftime('%Y-%m-%d')} | {row['region']} | "
          f"就诊{row['fever_cases']}例 (基线{row['baseline_mean']}) | "
          f"Z={row['z_score']} | 风险={row['risk_score']} | {level}")

if alerts.empty:
    print("✅ 当前窗口无异常信号")

异常信号 → AI初筛(分钟级) → 人工确认(小时级) → 自动化响应准备(并行)
                                              ↓
                                    ① 试剂/疫苗生产指令预激活
                                    ② 区域隔离方案生成
                                    ③ 医疗资源调度优化
                                    ④ 公众沟通模板渲染+审批

# bio_surveillance_stack.yaml
# 最小化生物监测AI栈——单机可运行

version: "3.9"
services:
  data-ingestion:
    image: python:3.11-slim
    working_dir: /app
    volumes:
      - ./ingestion:/app
    command: python scheduler.py  # 定时拉取ProMED/WHO数据
    environment:
      - DATA_SOURCES=promed_rss,who_api,hospital_feed
      - INGESTION_INTERVAL_MINUTES=15
    restart: always

  anomaly-detector:
    image: python:3.11-slim
    working_dir: /app
    volumes:
      - ./detector:/app
      - ./model_weights:/models
    command: python serve_detector.py  # Flask/FastAPI服务，接收数据流返回异常标记
    environment:
      - MODEL_PATH=/models/zscore_v1.pkl
      - Z_THRESHOLD=2.5
      - WINDOW_DAYS=7
    ports:
      - "8001:8001"
    depends_on:
      - data-ingestion
    restart: always

  risk-scorer:
    image: python:3.11-slim
    working_dir: /app
    volumes:
      - ./scorer:/app
    command: python serve_scorer.py
    environment:
      - POPULATION_DB=/app/pop_density.csv
      - SEVERITY_WEIGHT=1.5
    ports:
      - "8002:8002"
    depends_on:
      - anomaly-detector
    restart: always

  alert-gateway:
    image: python:3.11-slim
    working_dir: /app
    volumes:
      - ./alerts:/app
    command: python gateway.py  # 路由警报到邮件/Slack/内部系统
    environment:
      - ALERT_CHANNELS=email,slack,webhook
      - HIGH_RISK_THRESHOLD=5.0
    ports:
      - "8003:8003"
    depends_on:
      - risk-scorer
    restart: always