# 示例：基于 Prophet 的 VPS 流量预测
from prophet import Prophet
import pandas as pd

# 加载历史流量数据（每 5 分钟采集一次）
df = pd.read_csv("/var/log/vps/traffic_hourly.csv")
df.columns = ["ds", "y"]  # Prophet 要求列名为 ds（日期）和 y（数值）

# 训练模型
model = Prophet(
    yearly_seasonality=True,   # 年度周期性
    weekly_seasonality=True,   # 周度周期性
    daily_seasonality=True,    # 日度周期性
    changepoint_prior_scale=0.1  # 流量突变的敏感度
)
model.fit(df)

# 预测未来 24 小时
future = model.make_future_dataframe(periods=24, freq="H")
forecast = model.predict(future)

# 输出预测结果
print(f"预测峰值流量: {forecast['yhat'].max():.2f} Gbps")
print(f"预测谷值流量: {forecast['yhat'].min():.2f} Gbps")
print(f"预测波动率: {(forecast['yhat'].max() - forecast['yhat'].min()) / forecast['yhat'].mean():.2%}")

# nginx_upstream_dynamic.yaml — AI 动态生成的负载均衡配置
upstream backend_pool {
    # 基于流量预测动态调整权重
    server 10.0.1.10:8080 weight=15;   # 低负载实例，权重提高
    server 10.0.1.11:8080 weight=8;    # 高负载实例，权重降低
    server 10.0.1.12:8080 weight=12;   # 正常负载
    server 10.0.1.13:8080 weight=5;    # 边缘节点，仅承接突发流量
}

# AI 根据预测自动添加预热实例
# 预测未来 2 小时流量将增长 300%，自动扩容
server 10.0.1.14:8080 weight=10 backup;  # 预热的备用实例

#!/bin/bash
# ai_traffic_scheduler.sh — AI 驱动的动态流量调度脚本

PREDICTION_ENDPOINT="http://localhost:8000/api/traffic/predict"
NGINX_CONF="/etc/nginx/conf.d/upstream_dynamic.yaml"
LOG_FILE="/var/log/vps/ai_scheduler.log"

# 获取 AI 流量预测
FORECAST=$(curl -s "${PREDICTION_ENDPOINT}?hours=2")
PEAK_TRAFFIC=$(echo "$FORECAST" | jq '.peak_gbps')
CURRENT_TRAFFIC=$(echo "$FORECAST" | jq '.current_gbps')

# 计算流量增长率
GROWTH_RATE=$(echo "scale=2; ($PEAK_TRAFFIC - $CURRENT_TRAFFIC) / $CURRENT_TRAFFIC * 100" | bc)

echo "[$(date)] 当前流量: ${CURRENT_TRAFFIC} Gbps, 预测峰值: ${PEAK_TRAFFIC} Gbps, 增长率: ${GROWTH_RATE}%" | tee -a "$LOG_FILE"

# 根据增长率动态调整实例数量
if (( $(echo "$GROWTH_RATE > 50" | bc -l) )); then
    echo "流量激增 50%+，自动扩容备用实例" | tee -a "$LOG_FILE"
    # 激活预热实例
    sed -i 's/server 10.0.1.14:8080 weight=10 backup;/server 10.0.1.14:8080 weight=10;/' "$NGINX_CONF"
    nginx -s reload
elif (( $(echo "$GROWTH_RATE < -30" | bc -l) )); then
    echo "流量下降 30%+，收缩实例节省资源" | tee -a "$LOG_FILE"
    # 关闭备用实例
    sed -i 's/server 10.0.1.14:8080 weight=10;/server 10.0.1.14:8080 weight=10 backup;/g' "$NGINX_CONF"
    nginx -s reload
fi

# bandwidth_cost_optimizer.py — AI 带宽成本优化器
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class BandwidthCostOptimizer:
    def __init__(self, traffic_data_path):
        self.df = pd.read_csv(traffic_data_path, parse_dates=["timestamp"])
        self.df.set_index("timestamp", inplace=True)
        self.df.sort_index(inplace=True)
        
        # 计费模式价格（示例：DigitalOcean / Vultr / AWS）
        self.pricing = {
            "fixed_100mbps": {"base": 20, "overage": 0},       # $20/月固定 100Mbps
            "pay_per_gb": {"base": 0, "overage": 0.05},        # $0.05/GB
            "95th_percentile": {"base": 0, "overage": 0.04},   # $0.04/GB (95峰值)
            "hybrid": {"base": 10, "overage": 0.03},           # $10基础 + $0.03/GB超出
        }
    
    def calculate_cost(self, traffic_gb, mode):
        """计算指定计费模式下的月度成本"""
        config = self.pricing[mode]
        return config["base"] + traffic_gb * config["overage"]
    
    def analyze(self):
        """分析过去 30 天的流量数据，推荐最优计费模式"""
        # 按小时聚合
        hourly = self.df["bytes"].resample("1H").sum()
        daily = self.df["bytes"].resample("1D").sum()
        
        total_gb = hourly.sum() / (1024 ** 3)
        
        # 模拟各计费模式
        results = {}
        for mode in self.pricing:
            cost = self.calculate_cost(total_gb, mode)
            
            if mode == "95th_percentile":
                # 95峰值：排序后取第 95 百分位
                sorted_hours = hourly.dropna().sort_values(ascending=False)
                p95_index = int(len(sorted_hours) * 0.05)
                p95_gbps = sorted_hours.iloc[p95_index] / (1024 ** 3) * 8  # 转为 Gbps
                # 需要购买对应带宽的固定费用 + 超出部分
                base_bandwidth = max(1, int(np.ceil(p95_gbps)))
                cost = base_bandwidth * 10 + max(0, total_gb - base_bandwidth * 730 * 0.1) * 0.04
            
            results[mode] = {
                "monthly_cost": round(cost, 2),
                "total_traffic_gb": round(total_gb, 2),
            }
        
        # 排序推荐
        sorted_results = sorted(results.items(), key=lambda x: x[1]["monthly_cost"])
        
        return {
            "analysis_date": datetime.now().strftime("%Y-%m-%d"),
            "total_traffic_gb": round(total_gb, 2),
            "recommendation": sorted_results[0][0],
            "savings_vs_current": f"{((results['pay_per_gb']['monthly_cost'] - sorted_results[0][1]['monthly_cost']) / results['pay_per_gb']['monthly_cost'] * 100):.1f}%",
            "all_modes": results,
        }

# 使用示例
optimizer = BandwidthCostOptimizer("/var/log/vps/bandwidth_hourly.csv")
report = optimizer.analyze()
print(f"推荐计费模式: {report['recommendation']}")
print(f"预计节省: {report['savings_vs_current']}")
print(f"\n各模式对比:")
for mode, data in report["all_modes"].items():
    print(f"  {mode}: ${data['monthly_cost']}/月")

# AI 自动切换计费模式的决策逻辑
def auto_switch_billing(current_mode, analysis_result):
    recommended = analysis_result["recommendation"]
    
    if recommended != current_mode:
        # 检查切换条件（避免频繁切换）
        if can_switch_now(current_mode, recommended):
            switch_cost = get_switch_cost(current_mode, recommended)
            monthly_saving = analysis_result["all_modes"][recommended]["monthly_cost"] - \
                           analysis_result["all_modes"][current_mode]["monthly_cost"]
            
            if monthly_saving < 0 and switch_cost > abs(monthly_saving) * 3:
                # 切换成本高于 3 个月节省，不建议切换
                return False, "切换成本过高，建议保持当前模式"
            
            # 执行切换
            execute_billing_switch(current_mode, recommended)
            return True, f"已切换到 {recommended}，预计月节省 ${abs(monthly_saving)}"
    
    return False, "当前模式已是最优"

# cdn_cache_optimizer.py — AI CDN 缓存优化
import requests
from collections import defaultdict

class CDNCacheOptimizer:
    def __init__(self, origin_bandwidth_log, cdn_cache_log):
        self.origin_log = pd.read_csv(origin_bandwidth_log)
        self.cdn_log = pd.read_csv(cdn_cache_log)
    
    def analyze_cache_hit_rate(self):
        """分析 CDN 命中率，识别可优化的资源"""
        # 低命中率的路径（< 30%）意味着 CDN 没有发挥作用
        low_hit_paths = self.cdn_log.groupby("path").agg({
            "hits": "sum",
            "misses": "sum"
        }).assign(hit_rate=lambda x: x["hits"] / (x["hits"] + x["misses"]))
        
        candidates = low_hit_paths[low_hit_paths["hit_rate"] < 0.3]
        
        return {
            "low_hit_paths": candidates.index.tolist(),
            "potential_savings_gb": candidates.apply(
                lambda row: row["misses"] * 2.5  # 假设平均每请求 2.5MB
            ).sum()
        }
    
    def optimize_cache_rules(self, low_hit_paths):
        """AI 生成 CDN 缓存规则"""
        rules = []
        for path in low_hit_paths:
            if path.endswith((".jpg", ".png", ".css", ".js")):
                rules.append({
                    "path": path,
                    "action": "cache",
                    "ttl": "7d",
                    "reason": "静态资源，长期缓存可显著降低源站带宽"
                })
            elif "/api/" in path:
                rules.append({
                    "path": path,
                    "action": "no_cache",
                    "ttl": "0",
                    "reason": "API 响应动态变化，缓存意义不大"
                })
            else:
                rules.append({
                    "path": path,
                    "action": "cache",
                    "ttl": "1h",
                    "reason": "中等频率更新内容，短期缓存"
                })
        return rules

# network_diagnosis_engine.py — AI 网络故障诊断
import subprocess
import re
from datetime import datetime, timedelta

class NetworkDiagnosisEngine:
    def __init__(self):
        self.check_interval = 60  # 每 60 秒检查一次
        self.fault知识库 = self.load_fault_library()
    
    def load_fault_library(self):
        """加载已知故障模式和修复方案"""
        return {
            "dns_failure": {
                "symptoms": ["named: resolution time", "getaddrinfo: temporary failure", 
                             "nslookup timeout", "dig SERVFAIL"],
                "diagnosis": self._check_dns,
                "fix": self._fix_dns,
                "severity": "high",
                "auto_fix": True,
            },
            "route_blackhole": {
                "symptoms": ["Destination Host Unreachable", "No route to host",
                             "netstat: 0 active connections"],
                "diagnosis": self._check_routes,
                "fix": self._fix_routes,
                "severity": "critical",
                "auto_fix": True,
            },
            "port_blocked": {
                "symptoms": ["Connection timed out", "ECONNREFUSED", 
                             "nc: connect timed out"],
                "diagnosis": self._check_ports,
                "fix": self._fix_ports,
                "severity": "medium",
                "auto_fix": True,
            },
            "ssh_brute_force": {
                "symptoms": ["Failed password for root", "Invalid user",
                             "Too many authentication failures"],
                "diagnosis": self._check_ssh_logs,
                "fix": self._fix_ssh,
                "severity": "high",
                "auto_fix": True,
            },
            "ssl_expired": {
                "symptoms": ["certificate has expired", "SSL handshake failed",
                             "ERR_CERT_DATE_INVALID"],
                "diagnosis": self._check_ssl,
                "fix": self._fix_ssl,
                "severity": "critical",
                "auto_fix": True,
            },
        }
    
    def diagnose(self):
        """执行全量网络诊断"""
        results = []
        for fault_type, config in self.fault知识库.items():
            is_fault, detail = config["diagnosis"]()
            if is_fault:
                result = {
                    "type": fault_type,
                    "severity": config["severity"],
                    "detail": detail,
                    "auto_fix": config["auto_fix"],
                    "fix_command": config["fix"](),
                    "detected_at": datetime.now().isoformat(),
                }
                results.append(result)
        return results
    
    def _check_dns(self):
        """检查 DNS 解析是否正常"""
        try:
            result = subprocess.run(
                ["dig", "+short", "+time=3", "+tries=1", "google.com"],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode != 0 or not result.stdout.strip():
                # DNS 解析失败，尝试切换 DNS
                return True, f"DNS 解析失败: {result.stderr}"
            return False, ""
        except subprocess.TimeoutExpired:
            return True, "DNS 查询超时"
    
    def _fix_dns(self):
        """修复 DNS 问题"""
        fixes = []
        # 1. 切换备用 DNS
        fixes.append("echo 'nameserver 1.1.1.1' > /etc/resolv.conf")
        fixes.append("systemctl restart systemd-resolved")
        # 2. 检查 DNS 服务状态
        fixes.append("systemctl status systemd-resolved --no-pager")
        return "; ".join(fixes)
    
    def _check_routes(self):
        """检查路由表是否正常"""
        try:
            result = subprocess.run(
                ["ip", "route", "show"],
                capture_output=True, text=True, timeout=5
            )
            # 检查默认路由是否存在
            if "default via" not in result.stdout:
                return True, "默认路由缺失"
            return False, ""
        except Exception as e:
            return True, f"路由检查失败: {e}"
    
    def _fix_routes(self):
        """修复路由问题"""
        # 需要根据具体环境调整
        return "ip route add default via <gateway> dev <interface>"
    
    def _check_ports(self):
        """检查关键端口是否可访问"""
        critical_ports = [22, 80, 443]
        blocked = []
        for port in critical_ports:
            result = subprocess.run(
                ["nc", "-z", "-w", "3", "127.0.0.1", str(port)],
                capture_output=True
            )
            if result.returncode != 0:
                blocked.append(port)
        
        if blocked:
            return True, f"关键端口被封锁: {blocked}"
        return False, ""
    
    def _check_ssh_logs(self):
        """检查 SSH 暴力破解日志"""
        try:
            result = subprocess.run(
                ["journalctl", "-u", "ssh", "--since", "1 hour ago"],
                capture_output=True, text=True
            )
            failed_count = len(re.findall(r"Failed password", result.stdout))
            if failed_count > 50:
                return True, f"1 小时内 {failed_count} 次 SSH 失败登录，疑似暴力破解"
            return False, ""
        except Exception:
            return False, ""
    
    def _fix_ssh(self):
        """修复 SSH 安全问题"""
        return "crowdsec bump; fail2ban-client reload"
    
    def _check_ssl(self):
        """检查 SSL 证书状态"""
        try:
            result = subprocess.run(
                ["openssl", "x509", "-enddate", "-noout", "-in", "/etc/ssl/certs/server.crt"],
                capture_output=True, text=True
            )
            # 解析有效期
            end_date_str = result.stdout.split("=")[1]
            end_date = datetime.strptime(end_date_str.strip(), "%b %d %H:%M:%S %Y %Z")
            days_left = (end_date - datetime.now()).days
            
            if days_left < 7:
                return True, f"SSL 证书将在 {days_left} 天后过期"
            elif days_left < 0:
                return True, "SSL 证书已过期！"
            return False, f"SSL 证书有效，剩余 {days_left} 天"
        except FileNotFoundError:
            # 尝试自动续期
            return True, "SSL 证书不存在，尝试自动续期"
    
    def _fix_ssl(self):
        """修复 SSL 证书问题"""
        return "certbot renew --force-renewal && systemctl reload nginx"

# ai_network_agent.py — AI 网络自愈 Agent
import asyncio
import json
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ai_network_agent")

class NetworkSelfHealingAgent:
    def __init__(self):
        self.diagnosis_engine = NetworkDiagnosisEngine()
        self.notification = NotificationService()
        self.audit_log = AuditLogger()
    
    async def run_cycle(self):
        """执行一次完整的诊断-修复循环"""
        logger.info(f"[{datetime.now()}] 开始网络诊断循环...")
        
        # 1. 诊断
        faults = self.diagnosis_engine.diagnose()
        
        if not faults:
            logger.info("网络状态正常，无故障")
            return
        
        logger.warning(f"检测到 {len(faults)} 个故障")
        
        # 2. 按严重程度排序
        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        faults.sort(key=lambda x: severity_order.get(x["severity"], 99))
        
        # 3. 执行修复
        for fault in faults:
            if fault["auto_fix"]:
                logger.info(f"自动修复: {fault['type']}")
                fix_result = await self._execute_fix(fault)
                
                if fix_result["success"]:
                    logger.info(f"✅ {fault['type']} 修复成功")
                    self.audit_log.log(fault, fix_result, status="fixed")
                else:
                    logger.error(f"❌ {fault['type']} 修复失败: {fix_result['error']}")
                    self.audit_log.log(fault, fix_result, status="failed")
                    # 发送告警
                    await self.notification.send(
                        f"🚨 VPS 网络故障: {fault['type']} 自动修复失败",
                        details=fault["detail"]
                    )
            else:
                # 需要人工介入
                logger.warning(f"⚠️ {fault['type']} 需要人工介入")
                self.audit_log.log(fault, None, status="manual_required")
                await self.notification.send(
                    f"⚠️ VPS 网络故障: {fault['type']} 需要人工确认",
                    details=fault["detail"],
                    fix_command=fault.get("fix_command", "")
                )
        
        # 4. 效果验证
        await asyncio.sleep(30)  # 等待修复生效
        re_check = self.diagnosis_engine.diagnose()
        remaining = len(re_check)
        logger.info(f"修复后剩余故障: {remaining}")
    
    async def _execute_fix(self, fault):
        """执行故障修复"""
        try:
            # 执行修复命令
            fix_cmd = fault["fix_command"]
            result = await asyncio.to_thread(
                subprocess.run, fix_cmd, shell=True,
                capture_output=True, text=True, timeout=60
            )
            
            return {
                "success": result.returncode == 0,
                "stdout": result.stdout,
                "stderr": result.stderr,
            }
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    async def run_scheduled(self, interval=60):
        """定时运行诊断循环"""
        while True:
            try:
                await self.run_cycle()
            except Exception as e:
                logger.error(f"诊断循环异常: {e}")
            await asyncio.sleep(interval)

# 启动 Agent
if __name__ == "__main__":
    agent = NetworkSelfHealingAgent()
    asyncio.run(agent.run_scheduled(interval=60))

# ai_cdn_orchestrator.py — AI CDN 策略引擎
class AICDNOrchestrator:
    def __init__(self):
        self.traffic_analyzer = TrafficPatternAnalyzer()
        self.cost_model = CDNCostModel()
    
    def generate_cdn_rules(self, traffic_data):
        """基于流量分析生成 CDN 规则"""
        patterns = self.traffic_analyzer.analyze(traffic_data)
        rules = []
        
        for path_pattern, pattern_info in patterns.items():
            # AI 决策逻辑
            if pattern_info["content_type"] in ["image", "css", "js", "font"]:
                # 静态资源：长期缓存
                ttl = self._decide_static_ttl(pattern_info)
                rules.append({
                    "match": path_pattern,
                    "action": "cache",
                    "ttl": ttl,
                    "cache_key": "full_url",
                    "reason": f"静态资源，命中率 {pattern_info['hit_rate']:.1%}"
                })
            
            elif "/api/" in path_pattern or pattern_info["is_dynamic"]:
                # 动态内容：不缓存或短缓存
                rules.append({
                    "match": path_pattern,
                    "action": "no_cache" if pattern_info["hit_rate"] < 0.1 else "short_cache",
                    "ttl": "0" if pattern_info["hit_rate"] < 0.1 else "60",
                    "reason": f"动态内容，命中率仅 {pattern_info['hit_rate']:.1%}"
                })
            
            elif pattern_info["hit_rate"] > 0.7:
                # 高命中率路径：启用缓存
                rules.append({
                    "match": path_pattern,
                    "action": "cache",
                    "ttl": "1h",
                    "reason": f"高命中率 ({pattern_info['hit_rate']:.1%})，缓存收益大"
                })
            
            else:
                # 默认：短缓存
                rules.append({
                    "match": path_pattern,
                    "action": "short_cache",
                    "ttl": "300",
                    "reason": "默认短缓存策略"
                })
        
        return rules
    
    def _decide_static_ttl(self, pattern_info):
        """AI 决定静态资源的最佳缓存时间"""
        # 基于内容更新频率和命中率
        if pattern_info["update_frequency"] == "never":
            return "365d"   # 编译产物，长期缓存
        elif pattern_info["update_frequency"] == "rare":
            return "30d"   #  rarely updated
        elif pattern_info["hit_rate"] > 0.9:
            return "7d"    # 超高命中率，可以长期缓存
        else:
            return "1d"    # 一般命中率，短期缓存
    
    def calculate_savings(self, current_rules, new_rules, traffic_data):
        """计算 CDN 策略优化后的成本节省"""
        current_cost = self._calculate_cost(current_rules, traffic_data)
        new_cost = self._calculate_cost(new_rules, traffic_data)
        
        return {
            "current_monthly_cost": current_cost["total"],
            "optimized_monthly_cost": new_cost["total"],
            "savings": current_cost["total"] - new_cost["total"],
            "savings_percentage": f"{(current_cost['total'] - new_cost['total']) / current_cost['total'] * 100:.1f}%",
            "origin_bandwidth_saved_gb": current_cost["origin_gb"] - new_cost["origin_gb"],
        }

# 由 AI 自动生成的 Nginx + CDN 配置
# 基于历史流量分析，最优缓存策略

# 1. 静态资源 — 长期缓存（7-365天）
location ~* \.(jpg|jpeg|png|gif|ico|css|js|svg|woff2?|ttf|eot)$ {
    proxy_pass http://origin;
    proxy_cache valid_cache;
    proxy_cache_valid 200 365d;
    proxy_cache_valid 404 1d;
    add_header Cache-Control "public, max-age=31536000, immutable";
    add_header X-Cache-Strategy "AI-Optimized: static-7d-min";
}

# 2. HTML 页面 — 短期缓存（1小时）
location ~* \.html$ {
    proxy_pass http://origin;
    proxy_cache valid_cache;
    proxy_cache_valid 200 1h;
    proxy_cache_valid 404 10m;
    add_header Cache-Control "public, max-age=3600";
    add_header X-Cache-Strategy "AI-Optimized: html-1h";
}

# 3. API 请求 — 不缓存（动态内容）
location /api/ {
    proxy_pass http://origin;
    proxy_no_cache 1;
    proxy_cache_bypass 1;
    add_header Cache-Control "no-store, no-cache, must-revalidate";
    add_header X-Cache-Strategy "AI-Optimized: api-no-cache";
}

# 4. 高命中率动态路径 — 短缓存（60秒）
location ~* ^/api/v[12]/public/ {
    proxy_pass http://origin;
    proxy_cache valid_cache;
    proxy_cache_valid 200 60s;
    add_header Cache-Control "public, max-age=60";
    add_header X-Cache-Strategy "AI-Optimized: high-hit-short-cache";
}

# ai_ddos_detector.py — AI DDoS 智能检测
import numpy as np
from sklearn.ensemble import IsolationForest
from collections import defaultdict, deque
import time

class AIDDosDetector:
    def __init__(self, window_size=300):
        self.window_size = window_size  # 5 分钟滑动窗口
        self.connection_history = defaultdict(lambda: deque(maxlen=window_size))
        self.request_history = deque(maxlen=window_size * 10)  # 每秒采样
        self.detector = IsolationForest(contamination=0.05, random_state=42)
        self.threat_level = "normal"
        
        # 攻击特征定义
        self.attack_signatures = {
            "syn_flood": {"tcp_syn_per_sec": 1000, "connection_rate": 500},
            "http_flood": {"requests_per_sec": 500, "same_path_ratio": 0.8},
            "dns_amplification": {"dns_queries_per_sec": 200, "response_ratio": 50},
            "slowloris": {"slow_connections": 100, "connection_duration_avg": 300},
        }
    
    def analyze_traffic(self, current_metrics):
        """分析当前流量指标，检测 DDoS 攻击"""
        features = self._extract_features(current_metrics)
        
        # 使用隔离森林检测异常
        prediction = self.detector.predict([features])
        anomaly_score = self.detector.score_samples([features])[0]
        
        # 结合规则检测
        rule_violations = self._check_rules(current_metrics)
        
        # 综合评估威胁等级
        threat_score = self._calculate_threat_score(
            anomaly_score, rule_violations, current_metrics
        )
        
        return {
            "is_attack": threat_score > 0.7,
            "threat_level": self._map_threat_level(threat_score),
            "attack_type": self._identify_attack_type(current_metrics, rule_violations),
            "threat_score": threat_score,
            "anomaly_score": anomaly_score,
            "rule_violations": rule_violations,
            "recommended_action": self._get_recommended_action(threat_score),
        }
    
    def _extract_features(self, metrics):
        """提取流量特征向量"""
        return [
            metrics["connections_per_sec"],
            metrics["packets_per_sec"],
            metrics["bytes_per_sec"],
            metrics["unique_ips_per_sec"],
            metrics["syn_ratio"],
            metrics["avg_packet_size"],
            metrics["request_rate_per_ip"],
            metrics["error_rate"],
        ]
    
    def _check_rules(self, metrics):
        """检查已知攻击规则"""
        violations = []
        
        if metrics["syn_ratio"] > 0.8:
            violations.append({
                "rule": "syn_flood",
                "severity": "high",
                "detail": f"SYN 包比例 {metrics['syn_ratio']:.2%} 超过阈值 80%"
            })
        
        if metrics["unique_ips_per_sec"] < 10 and metrics["requests_per_sec"] > 500:
            violations.append({
                "rule": "single_source_flood",
                "severity": "medium",
                "detail": f"单源 IP 产生 {metrics['requests_per_sec']:.0f} QPS"
            })
        
        if metrics["error_rate"] > 0.5:
            violations.append({
                "rule": "high_error_rate",
                "severity": "medium",
                "detail": f"错误率 {metrics['error_rate']:.1%} 异常偏高"
            })
        
        return violations
    
    def _calculate_threat_score(self, anomaly_score, violations, metrics):
        """综合计算威胁分数"""
        # 基础分数来自 ML 模型
        base_score = max(0, -anomaly_score)  # IsolationForest 分数越负越异常
        
        # 规则违反加分
        rule_score = sum(
            0.3 if v["severity"] == "high" else 0.15
            for v in violations
        )
        
        # 流量异常加分
        if metrics["requests_per_sec"] > 1000:
            base_score += 0.2
        
        return min(1.0, base_score + rule_score)
    
    def _identify_attack_type(self, metrics, violations):
        """识别攻击类型"""
        if any(v["rule"] == "syn_flood" for v in violations):
            return "SYN Flood"
        elif metrics["requests_per_sec"] > 500 and metrics["unique_ips_per_sec"] < 10:
            return "HTTP Flood (Single Source)"
        elif metrics["error_rate"] > 0.5:
            return "Application Layer Attack"
        return "Unknown"
    
    def _get_recommended_action(self, threat_score):
        """生成应对建议"""
        if threat_score > 0.9:
            return {
                "action": "block_and_alert",
                "steps": [
                    "立即启用 DDoS 防护模式",
                    "封锁异常 IP 段",
                    "启用 CDN 清洗中心",
                    "通知安全团队",
                ],
                "automation_level": "full_auto",
            }
        elif threat_score > 0.7:
            return {
                "action": "throttle_and_monitor",
                "steps": [
                    "对异常 IP 限流",
                    "启用请求频率限制",
                    "加强日志记录",
                    "准备手动干预",
                ],
                "automation_level": "auto_with_confirmation",
            }
        elif threat_score > 0.5:
            return {
                "action": "monitor_and_warn",
                "steps": [
                    "增加监控频率",
                    "发送预警通知",
                    "准备防护预案",
                ],
                "automation_level": "warn_only",
            }
        return {
            "action": "normal",
            "steps": [],
            "automation_level": "none",
        }

# ai_ddos_response.py — AI DDoS 自动响应
class DDoSAutoResponder:
    def __init__(self):
        self.iptables = IPTablasManager()
        self.cloudflare = CloudflareAPI()
        self.notification = NotificationService()
    
    async def respond(self, detection_result):
        """根据检测结果自动执行响应"""
        action_config = detection_result["recommended_action"]
        
        if action_config["action"] == "normal":
            return
        
        # 全自动化响应
        if action_config["automation_level"] == "full_auto":
            await self._execute_full_auto_response(detection_result)
        
        # 需确认的自动化响应
        elif action_config["automation_level"] == "auto_with_confirmation":
            confirmation = await self._request_confirmation(action_config)
            if confirmation:
                await self._execute_full_auto_response(detection_result)
            else:
                await self.notification.send(
                    "⚠️ DDoS 攻击需人工确认",
                    details=detection_result
                )
        
        # 仅告警
        elif action_config["automation_level"] == "warn_only":
            await self.notification.send(
                "📡 检测到异常流量，建议关注",
                details=detection_result
            )
    
    async def _execute_full_auto_response(self, result):
        """执行完整的自动化响应"""
        attack_type = result["attack_type"]
        
        # 1. 封锁攻击源 IP 段
        if attack_type in ["SYN Flood", "HTTP Flood (Single Source)"]:
            suspicious_ips = self._identify_attack_sources(result)
            for ip in suspicious_ips[:100]:  # 最多封锁 100 个 IP
                self.iptables.block(ip)
        
        # 2. 启用 CDN 防护
        await self.cloudflare.enable_protection(mode="under_attack")
        
        # 3. 调整 Nginx 限流配置
        self._apply_rate_limiting(aggressive=True)
        
        # 4. 记录事件
        self._log_incident(result)
        
        # 5. 发送告警
        await self.notification.send(
            "🚨 DDoS 攻击已自动响应",
            details={
                "type": attack_type,
                "ips_blocked": len(self._identify_attack_sources(result)),
                "cdn_protection": "enabled",
            }
        )

# 1. 安装依赖
pip install prophet scikit-learn pandas numpy

# 2. 部署 AI Agent
git clone https://github.com/your-org/ai-network-agent.git
cd ai-network-agent
docker-compose up -d

# 3. 配置监控数据源
# 确保 Prometheus 采集了以下指标:
# - nginx_request_rate
# - nginx_bytes_transferred
# - nginx_active_connections
# - system_network_packets
# - system_network_errors

# 4. 配置通知渠道
# 编辑 config.yaml:
# notification:
#   slack:
#     webhook_url: "https://hooks.slack.com/..."
#   email:
#     smtp_host: "smtp.gmail.com"
#     recipient: "admin@yourdomain.com"

# 5. 启动 AI 网络 Agent
systemctl start ai-network-agent
systemctl enable ai-network-agent

# 6. 查看诊断报告
curl http://localhost:8080/api/diagnosis/latest

# 效果反馈循环：让 AI 模型越来越准
def update_model_with_feedback(old_prediction, actual_value, model):
    """用实际数据更新预测模型"""
    error = abs(actual_value - old_prediction) / old_prediction
    
    if error > 0.2:
        # 预测误差超过 20%，重新训练模型
        recent_data = load_recent_traffic_data(hours=24)
        model.retrain(recent_data)
        log_model_update("重训练完成，误差阈值 20%")
    elif error > 0.1:
        # 误差超过 10%，调整模型参数
        model.adjust_parameters(error=error)
        log_model_update("参数调整完成，误差阈值 10%")
    else:
        log_model_update("预测准确，无需调整")