训练流量预测模型实现自动扩容

┌─────────────────────────────────────────────────────────────────────┐
│                        离线训练流程                                   │
│                                                                     │
│  Prometheus ──▶ 历史指标采集 ──▶ 特征工程 ──▶ 模型训练 ──▶ 模型注册  │
│  (30天 QPS)     (Python 脚本)    (Pandas)    (Prophet/XGB)  (MLflow) │
└─────────────────────────────────────────────────────────────────────┘
                                                                     │
                                                          模型文件下载
                                                                     │
                                                                     ▼
┌─────────────────────────────────────────────────────────────────────┐
│                        在线推理 + 扩容流程                            │
│                                                                     │
│  当前指标 ──▶ 推理服务 ──▶ 预测未来 QPS ──▶ 扩容决策器 ──▶ K8s API   │
│  (Prometheus)  (FastAPI)    (未来30分钟)       (Go 程序)    (client-go)│
│                                                                     │
│                        ┌──────────────┐                              │
│                        │   HPA 兜底    │  ← 基于实时 CPU 做精细调整   │
│                        └──────────────┘                              │
└─────────────────────────────────────────────────────────────────────┘

# scripts/fetch_metrics.py
# 从 Prometheus 拉取历史 QPS 数据，保存为 CSV
# 运行：python fetch_metrics.py --prom-url http://prometheus:9090 --days 30

import requests
import pandas as pd
import argparse
from datetime import datetime, timedelta
import time

def fetch_qps_from_prometheus(prom_url, query, start_time, end_time, step='60s'):
    """从 Prometheus 拉取 range query 数据
    
    Args:
        prom_url: Prometheus 地址
        query: PromQL 查询语句
        start_time: 开始时间 (datetime)
        end_time: 结束时间 (datetime)
        step: 采样间隔，默认 60 秒
    
    Returns:
        DataFrame，包含 timestamp 和 value 两列
    """
    # Prometheus range query API
    # 一次最多拉 11000 个数据点，超过需要分段
    url = f"{prom_url}/api/v1/query_range"
    
    params = {
        'query': query,
        'start': start_time.timestamp(),
        'end': end_time.timestamp(),
        'step': step,
    }
    
    resp = requests.get(url, params=params, timeout=60)
    resp.raise_for_status()
    data = resp.json()
    
    if data['status'] != 'success':
        raise ValueError(f"Prometheus query failed: {data}")
    
    result = data['data']['result']
    if len(result) == 0:
        raise ValueError("No data returned from Prometheus")
    
    # 取第一个时间序列（如果有多条，后面再聚合）
    values = result[0]['values']
    
    df = pd.DataFrame(values, columns=['timestamp', 'value'])
    # Prometheus 返回的是 Unix 时间戳（字符串），需要转换
    df['timestamp'] = pd.to_datetime(df['timestamp'].astype(int), unit='s')
    # value 是字符串，转成 float
    df['value'] = df['value'].astype(float)
    df = df.set_index('timestamp')
    
    return df

def fetch_history(prom_url, days=30):
    """拉取 N 天的 QPS 数据，分段拉避免单次请求过大"""
    end_time = datetime.now()
    start_time = end_time - timedelta(days=days)
    
    # 用 sum(rate()) 算总 QPS
    # 注意：rate() 的窗口 [1m] 要跟采样间隔匹配
    # 如果服务有多个实例，用 sum 聚合
    query = 'sum(rate(http_requests_total{job="api-gateway"}[1m]))'
    
    all_dfs = []
    # 每次拉 1 天，避免单次请求超时
    current = start_time
    while current < end_time:
        chunk_end = min(current + timedelta(days=1), end_time)
        print(f"fetching {current} to {chunk_end}...")
        
        try:
            df = fetch_qps_from_prometheus(prom_url, query, current, chunk_end)
            all_dfs.append(df)
        except Exception as e:
            print(f"  warning: failed to fetch this chunk: {e}")
            # 跳过失败的段，不要让一个 chunk 失败导致整个拉取失败
        
        current = chunk_end
        time.sleep(1)  # 别把 Prometheus 打爆
    
    if not all_dfs:
        raise RuntimeError("No data fetched")
    
    result = pd.concat(all_dfs)
    # 去重（分段拉取可能有重叠）
    result = result[~result.index.duplicated(keep='first')]
    result = result.sort_index()
    
    return result

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--prom-url', default='http://prometheus:9090')
    parser.add_argument('--days', type=int, default=30)
    parser.add_argument('--output', default='qps_history.csv')
    args = parser.parse_args()
    
    df = fetch_history(args.prom_url, args.days)
    df.to_csv(args.output)
    print(f"saved {len(df)} rows to {args.output}")
    print(f"date range: {df.index[0]} to {df.index[-1]}")
    print(f"qps range: {df['value'].min():.1f} - {df['value'].max():.1f}")

# scripts/feature_engineering.py
# 特征工程：从原始 QPS 数据生成训练特征

import pandas as pd
import numpy as np
from datetime import datetime

def build_features(df):
    """从原始 QPS DataFrame 生成特征矩阵
    
    Args:
        df: DataFrame，index 是 datetime，有一列 'value' 是 QPS
    
    Returns:
        DataFrame，包含原始 QPS + 所有特征
    """
    df = df.copy()
    df = df.rename(columns={'value': 'qps'})
    
    # === 时间特征 ===
    # hour: 0-23，捕捉日内周期
    df['hour'] = df.index.hour
    # dayofweek: 0=周一, 6=周日
    df['dayofweek'] = df.index.dayofweek
    # is_weekend: 周末流量模式跟工作日不一样
    df['is_weekend'] = (df.index.dayofweek >= 5).astype(int)
    
    # 节假日特征：需要外部数据源
    # 我用的是中国法定节假日，自己维护一个日期列表
    # 如果有营销活动也在这里加
    holidays = [
        '2025-01-01', '2025-02-10', '2025-02-11', '2025-02-12',  # 春节
        '2025-04-04', '2025-04-05', '2025-04-06',                # 清明
        '2025-05-01', '2025-05-02', '2025-05-03',                # 劳动节
        '2025-06-14', '2025-06-15', '2025-06-16',                # 端午
        '2025-09-27', '2025-09-28', '2025-09-29',                # 中秋
        '2025-10-01', '2025-10-02', '2025-10-03',                # 国庆
        # 电商大促
        '2025-06-18', '2025-11-11', '2025-12-12',
    ]
    holiday_dates = pd.to_datetime(holidays).date
    df['is_holiday'] = df.index.date.isin(holiday_dates).astype(int)
    
    # === 滞后特征 ===
    # 1 小时前的 QPS：短期趋势信号
    df['qps_lag_1h'] = df['qps'].shift(1)
    # 24 小时前的 QPS：昨天同一时刻的 QPS，最强的周期信号
    df['qps_lag_24h'] = df['qps'].shift(24)
    # 7 天前同一时刻的 QPS：上周同一天的 QPS
    df['qps_lag_7d'] = df['qps'].shift(7 * 24)
    
    # === 滑动统计特征 ===
    # 过去 7 天同一时刻的均值和标准差
    # 用 shift(24) 先拿到"昨天同一时刻"，再 rolling 7 天
    # 这样每个窗口包含的是过去 7 天的同一时刻，而不是过去 7 天的所有时刻
    df['qps_same_hour_mean_7d'] = df['qps'].shift(24).rolling(window=7).mean()
    df['qps_same_hour_std_7d'] = df['qps'].shift(24).rolling(window=7).std()
    
    # 过去 1 小时的均值和变化率
    df['qps_roll_mean_1h'] = df['qps'].rolling(window=60).mean()  # 60 个 1 分钟点
    df['qps_roll_std_1h'] = df['qps'].rolling(window=60).std()
    
    # QPS 变化率（一阶差分）：捕捉趋势
    df['qps_diff_1'] = df['qps'].diff(1)
    df['qps_diff_5'] = df['qps'].diff(5)
    
    # === 异常值处理 ===
    # QPS 不应该有负值或极端大值
    # 用 3σ 法则裁剪，但别直接删，用 clip 限幅
    qps_mean = df['qps'].mean()
    qps_std = df['qps'].std()
    upper_bound = qps_mean + 5 * qps_std  # 5σ 而不是 3σ，避免裁掉正常峰值
    df['qps'] = df['qps'].clip(lower=0, upper=upper_bound)
    
    # 删除因为有 NaN 的行（前 7*24=168 行因为 lag 特效必然是 NaN）
    df = df.dropna()
    
    return df

if __name__ == '__main__':
    df = pd.read_csv('qps_history.csv', parse_dates=['timestamp'], index_col='timestamp')
    features = build_features(df)
    features.to_csv('qps_features.csv')
    print(f"generated {len(features.columns)} features, {len(features)} rows")
    print(f"features: {list(features.columns)}")

# scripts/train_prophet.py
# 训练 Prophet 模型，做基线预测

import pandas as pd
from prophet import Prophet
import joblib
import logging

# Prophet 的日志太吵了，设为 WARNING
logging.getLogger('prophet').setLevel(logging.WARNING)
logging.getLogger('cmdstanpy').setLevel(logging.WARNING)

def train_prophet(df):
    """训练 Prophet 模型
    
    Prophet 要求输入 DataFrame 有两列：
    - ds: 日期时间
    - y: 目标值
    
    Args:
        df: 特征工程后的 DataFrame，包含 qps 列
    
    Returns:
        训练好的 Prophet 模型
    """
    # Prophet 只需要 ds 和 y 两列，其他特征它自己会算季节性
    train_df = pd.DataFrame({
        'ds': df.index,
        'y': df['qps'].values,
    })
    
    # 初始化 Prophet
    # yearly_seasonality=False: 一年的数据不够学年度季节性
    # weekly_seasonality=True: 周周期很重要，默认开
    # daily_seasonality=True: 日周期是核心，默认开
    # changepoint_prior_scale: 控制趋势变化的灵活度，默认 0.05
    #   调大到 0.1 让模型更敏感地捕捉趋势变化（但别太大，会过拟合）
    model = Prophet(
        yearly_seasonality=False,
        weekly_seasonality=True,
        daily_seasonality=True,
        changepoint_prior_scale=0.1,
        changepoint_range=0.9,  # 用前 90% 的数据来检测趋势变化点
    )
    
    # 添加节假日效应
    # Prophet 内置了部分国家的节假日，但中国节假日要自己加
    holidays = pd.DataFrame({
        'holiday': 'cn_holiday',
        'ds': pd.to_datetime([
            '2025-01-01', '2025-02-10', '2025-02-11', '2025-02-12',
            '2025-04-04', '2025-05-01', '2025-06-18',
            '2025-09-27', '2025-10-01', '2025-10-02', '2025-10-03',
            '2025-11-11', '2025-12-12',
        ]),
        'lower_window': -1,  # 节假日前 1 天也开始有影响
        'upper_window': 1,   # 节假日后 1 天还有余温
    })
    # Prophet 内置了部分国家的节假日，但中国节假日不一定有
    # 尝试加载内置 CN 节假日，失败就用上面手动定义的 holidays
    try:
        model = model.add_country_holidays(country_name='CN')
    except Exception as e:
        print(f"no built-in CN holidays, using manual list: {e}")
    # 直接传 holidays 作为补充（跟内置节假日不冲突，Prophet 会合并）
    model.holidays = holidays
    
    print("training Prophet model...")
    model.fit(train_df)
    
    return model

if __name__ == '__main__':
    df = pd.read_csv('qps_features.csv', parse_dates=['timestamp'], index_col='timestamp')
    
    model = train_prophet(df)
    
    # 保存模型
    joblib.dump(model, 'prophet_model.pkl')
    print("model saved to prophet_model.pkl")
    
    # 快速验证：预测未来 30 分钟
    future = model.make_future_dataframe(periods=30, freq='min')
    forecast = model.predict(future)
    print("\n=== forecast for next 30 min ===")
    print(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(30).to_string(index=False))

# scripts/train_xgb.py
# 训练 XGBoost 模型，修正 Prophet 的预测残差

import pandas as pd
import numpy as np
import joblib
from prophet import Prophet
from xgboost import XGBRegressor
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import mean_absolute_error, mean_absolute_percentage_error
import logging

logging.getLogger('prophet').setLevel(logging.WARNING)

def train_residual_model(df, prophet_model):
    """用 XGBoost 学习 Prophet 预测的残差
    
    策略：先用 Prophet 预测训练集，算出残差（实际值 - 预测值），
    然后用 XGBoost 学习残差与特征之间的关系。
    
    最终预测 = Prophet 预测 + XGBoost 预测的残差
    """
    # 1. 用 Prophet 预测训练集（in-sample 预测）
    prophet_input = pd.DataFrame({
        'ds': df.index,
        'y': df['qps'].values,
    })
    prophet_pred = prophet_model.predict(prophet_input)
    
    # 2. 计算残差
    residual = df['qps'].values - prophet_pred['yhat'].values
    
    # 3. 准备 XGBoost 的特征和标签
    # 用工程化特征（hour, dayofweek, lag 特征等）来预测残差
    feature_cols = [
        'hour', 'dayofweek', 'is_weekend', 'is_holiday',
        'qps_lag_1h', 'qps_lag_24h', 'qps_lag_7d',
        'qps_same_hour_mean_7d', 'qps_same_hour_std_7d',
        'qps_roll_mean_1h', 'qps_roll_std_1h',
        'qps_diff_1', 'qps_diff_5',
    ]
    # 加上 Prophet 的预测值作为特征（让 XGB 知道基线是多少）
    df_features = df[feature_cols].copy()
    df_features['prophet_yhat'] = prophet_pred['yhat'].values
    
    X = df_features.values
    y = residual  # 标签是残差，不是原始 QPS
    
    # 4. 时间序列交叉验证
    # 不能用随机 K-Fold，必须按时间切分
    tscv = TimeSeriesSplit(n_splits=5)
    
    best_mae = float('inf')
    
    for fold, (train_idx, val_idx) in enumerate(tscv.split(X)):
        X_train, X_val = X[train_idx], X[val_idx]
        y_train, y_val = y[train_idx], y[val_idx]
        
        model = XGBRegressor(
            n_estimators=300,        # 树的数量，300 足够
            max_depth=6,              # 深度 6 防过拟合
            learning_rate=0.05,       # 小学习率 + 多树 = 更稳定
            subsample=0.8,           # 行采样
            colsample_bytree=0.8,    # 列采样
            reg_alpha=0.1,           # L1 正则化
            reg_lambda=1.0,          # L2 正则化
            random_state=42,
            n_jobs=-1,
        )
        
        model.fit(
            X_train, y_train,
            eval_set=[(X_val, y_val)],
            verbose=False,
        )
        
        # 验证集预测
        val_pred = model.predict(X_val)
        mae = mean_absolute_error(y_val, val_pred)
        
        # 最终预测 = Prophet 基线 + XGB 残差
        final_pred = prophet_pred['yhat'].values[val_idx] + val_pred
        final_mae = mean_absolute_error(df['qps'].values[val_idx], final_pred)
        final_mape = mean_absolute_percentage_error(
            df['qps'].values[val_idx] + 1,  # +1 避免除零
            final_pred + 1,
        )
        
        print(f"fold {fold}: residual MAE={mae:.1f}, final MAE={final_mae:.1f}, MAPE={final_mape:.1%}")
        
        if final_mae < best_mae:
            best_mae = final_mae
    
    # 用全部数据重新训练
    final_model = XGBRegressor(
        n_estimators=300, max_depth=6, learning_rate=0.05,
        subsample=0.8, colsample_bytree=0.8,
        reg_alpha=0.1, reg_lambda=1.0, random_state=42, n_jobs=-1,
    )
    final_model.fit(X, y, verbose=False)
    
    return final_model, feature_cols, best_mae

if __name__ == '__main__':
    df = pd.read_csv('qps_features.csv', parse_dates=['timestamp'], index_col='timestamp')
    
    # 先加载 Prophet 模型
    prophet_model = joblib.load('prophet_model.pkl')
    
    # 训练 XGBoost 残差模型
    xgb_model, feature_cols, best_mae = train_residual_model(df, prophet_model)
    
    # 保存
    joblib.dump({
        'model': xgb_model,
        'feature_cols': feature_cols,
    }, 'xgb_residual_model.pkl')
    
    print(f"\nmodel saved. best final MAE: {best_mae:.1f}")

# app/main.py
# FastAPI 推理服务：输入预测步数，返回未来 N 分钟的 QPS 预测

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging

# 日志配置
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="QPS Predictor", version="1.0.0")

# 全局变量，启动时加载模型
prophet_model = None
xgb_model = None
xgb_feature_cols = None

@app.on_event("startup")
def load_models():
    """启动时加载模型，避免每次请求都加载"""
    global prophet_model, xgb_model, xgb_feature_cols
    
    prophet_model = joblib.load('models/prophet_model.pkl')
    
    xgb_data = joblib.load('models/xgb_residual_model.pkl')
    xgb_model = xgb_data['model']
    xgb_feature_cols = xgb_data['feature_cols']
    
    logger.info("models loaded successfully")

class PredictRequest(BaseModel):
    steps: int = 30  # 预测未来多少分钟

class PredictPoint(BaseModel):
    timestamp: str
    qps: float
    qps_lower: float
    qps_upper: float

class PredictResponse(BaseModel):
    predictions: list[PredictPoint]
    model_version: str = "1.0"

@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest):
    """预测未来 N 分钟的 QPS
    
    流程：
    1. Prophet 做基线预测
    2. 构造特征，XGBoost 预测残差
    3. 最终预测 = Prophet 基线 + XGB 残差
    4. 加上置信区间
    """
    if req.steps <= 0 or req.steps > 120:
        raise HTTPException(status_code=400, detail="steps must be between 1 and 120")
    
    # 1. Prophet 基线预测
    future = prophet_model.make_future_dataframe(periods=req.steps, freq='min')
    prophet_forecast = prophet_model.predict(future)
    
    # 取最后 N 分钟的预测
    prophet_pred = prophet_forecast.tail(req.steps).copy()
    
    # 2. 构造 XGBoost 特征
    # 注意：预测时没有实际的 QPS 值，lag 特征要用 Prophet 的预测值来填充
    # 这是一个简化处理——更严谨的做法是用递归预测（每次预测一步，用预测值作为下一步的 lag）
    now = datetime.now()
    features = []
    
    for i, row in prophet_pred.iterrows():
        ts = row['ds']
        feat = {
            'hour': ts.hour,
            'dayofweek': ts.dayofweek,
            'is_weekend': int(ts.dayofweek >= 5),
            'is_holiday': 0,  # 简化：预测时不知道未来是否节假日，实际要查日历
            # lag 特征用 Prophet 预测值近似
            # 严格来说应该用递归预测，但 Prophet 预测已经足够准了
            'qps_lag_1h': row['yhat'],
            'qps_lag_24h': row['yhat'],
            'qps_lag_7d': row['yhat'],
            'qps_same_hour_mean_7d': row['yhat'],
            'qps_same_hour_std_7d': 0,
            'qps_roll_mean_1h': row['yhat'],
            'qps_roll_std_1h': 0,
            'qps_diff_1': 0,
            'qps_diff_5': 0,
            'prophet_yhat': row['yhat'],
        }
        features.append(feat)
    
    # 3. XGBoost 残差预测
    feature_df = pd.DataFrame(features)
    X = feature_df[xgb_feature_cols + ['prophet_yhat']].values
    residual_pred = xgb_model.predict(X)
    
    # 4. 最终预测 = 基线 + 残差
    final_pred = prophet_pred['yhat'].values + residual_pred
    # QPS 不能为负
    final_pred = np.maximum(final_pred, 0)
    
    # 置信区间：用 Prophet 的区间 + 残差的标准差
    # 这是一个近似，更准确的做法是用分位数回归
    residual_std = np.std(residual_pred)
    lower = final_pred - 1.96 * residual_std
    upper = final_pred + 1.96 * residual_std
    
    # 构造响应
    predictions = []
    for i, (_, row) in enumerate(prophet_pred.iterrows()):
        predictions.append(PredictPoint(
            timestamp=row['ds'].isoformat(),
            qps=round(final_pred[i], 1),
            qps_lower=round(max(0, lower[i]), 1),
            qps_upper=round(upper[i], 1),
        ))
    
    logger.info(f"predicted {req.steps} steps, max_qps={max(p.qps for p in predictions):.1f}")
    
    return PredictResponse(predictions=predictions)

@app.get("/health")
def health():
    """健康检查端点"""
    if prophet_model is None or xgb_model is None:
        raise HTTPException(status_code=503, detail="models not loaded")
    return {"status": "ok"}

# 启动命令：uvicorn app.main:app --host 0.0.0.0 --port 8080

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码和模型
COPY app/ ./app/
COPY models/ ./models/

EXPOSE 8080

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: qps-predictor
  namespace: aiops
spec:
  replicas: 2  # 两个副本做 HA
  selector:
    matchLabels:
      app: qps-predictor
  template:
    metadata:
      labels:
        app: qps-predictor
    spec:
      containers:
        - name: predictor
          image: registry.example.com/qps-predictor:v1.0
          ports:
            - containerPort: 8080
          resources:
            requests:
              cpu: 200m
              memory: 512Mi
            limits:
              cpu: 500m
              memory: 1Gi
          # 健康检查
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 10
            periodSeconds: 30
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: qps-predictor
  namespace: aiops
spec:
  selector:
    app: qps-predictor
  ports:
    - port: 8080
      targetPort: 8080

// pkg/scaler/decision.go
package scaler

import (
	"fmt"
	"math"
	"time"
)

// ScaleConfig 扩容配置
type ScaleConfig struct {
	MinReplicas       int32   // 最小副本数
	MaxReplicas       int32   // 最大副本数
	QPSPerPod         float64 // 每个 Pod 能承载的 QPS（压测得出）
	ScaleUpThreshold  float64 // 扩容阈值：预测 QPS 达到容量的多少百分比时扩容
	ScaleDownThreshold float64 // 缩容阈值
	CooldownSeconds    int      // 冷却时间：两次扩缩容之间最少间隔
	PredictSteps       int      // 预测步数（分钟）
	PodStartupSeconds  int      // Pod 启动时间（预热），用于提前扩容
}

// DefaultConfig 默认配置
func DefaultConfig() ScaleConfig {
	return ScaleConfig{
		MinReplicas:        3,
		MaxReplicas:        50,
		QPSPerPod:          200,  // 压测得出：单 Pod 200 QPS 时 P99 < 200ms
		ScaleUpThreshold:   0.7, // 达到 70% 容量就开始扩容，留余量
		ScaleDownThreshold: 0.3, // 降到 30% 容量才缩容，避免抖动
		CooldownSeconds:    120, // 2 分钟内不重复扩缩容
		PredictSteps:       15,  // 预测未来 15 分钟
		PodStartupSeconds:  60,  // Pod 启动 + 预热需要 60 秒
	}
}

// ScaleDecision 扩缩容决策结果
type ScaleDecision struct {
	CurrentReplicas int32   `json:"currentReplicas"`
	DesiredReplicas int32   `json:"desiredReplicas"`
	Reason          string  `json:"reason"`
	PredictedMaxQPS float64 `json:"predictedMaxQPS"`
	Action          string  `json:"action"` // scale_up / scale_down / no_action
}

// PredictedPoint 预测数据点
type PredictedPoint struct {
	Timestamp string  `json:"timestamp"`
	QPS       float64 `json:"qps"`
}

// Decide 根据预测结果和当前状态决定目标副本数
// 这是整个扩容系统的"大脑"
func Decide(
	predictions []PredictedPoint,
	currentReplicas int32,
	lastScaleTime time.Time,
	config ScaleConfig,
) ScaleDecision {
	decision := ScaleDecision{
		CurrentReplicas: currentReplicas,
	}

	if len(predictions) == 0 {
		decision.Action = "no_action"
		decision.Reason = "no prediction data"
		decision.DesiredReplicas = currentReplicas
		return decision
	}

	// 1. 找到预测窗口内的最大 QPS
	// 不看所有预测点，只看"Pod 启动时间"之后的点
	// 因为 Pod 启动需要时间，现在扩容要考虑预热
	startupDelay := time.Duration(config.PodStartupSeconds) * time.Second
	now := time.Now()

	var maxPredictedQPS float64
	for _, p := range predictions {
		ts, err := time.Parse(time.RFC3339, p.Timestamp)
		if err != nil {
			continue
		}
		// 跳过 Pod 启动时间内的预测——这段时间来不及扩容
		if ts.Sub(now) < startupDelay {
			continue
		}
		if p.QPS > maxPredictedQPS {
			maxPredictedQPS = p.QPS
		}
	}

	if maxPredictedQPS == 0 {
		maxPredictedQPS = predictions[0].QPS // fallback
	}

	decision.PredictedMaxQPS = maxPredictedQPS

	// 2. 计算需要的副本数
	// 公式：目标副本数 = ceil(预测 QPS / 单 Pod QPS)
	// 但要考虑阈值：达到 70% 容量就扩容，而不是等到 100%
	currentCapacity := float64(currentReplicas) * config.QPSPerPod
	thresholdForScaleUp := currentCapacity * config.ScaleUpThreshold

	// 3. 冷却期检查
	timeSinceLastScale := time.Since(lastScaleTime)
	if timeSinceLastScale < time.Duration(config.CooldownSeconds)*time.Second {
		decision.Action = "no_action"
		decision.Reason = fmt.Sprintf("cooldown: last scale %v ago, need %v",
			timeSinceLastScale.Round(time.Second),
			time.Duration(config.CooldownSeconds)*time.Second)
		decision.DesiredReplicas = currentReplicas
		return decision
	}

	// 4. 扩容判断
	if maxPredictedQPS > thresholdForScaleUp {
		// 需要扩容
		desired := int32(math.Ceil(maxPredictedQPS / config.QPSPerPod))
		// 留 20% 余量，避免刚扩完又得扩
		desired = int32(math.Ceil(float64(desired) * 1.2))
		desired = clampReplicas(desired, config.MinReplicas, config.MaxReplicas)

		if desired > currentReplicas {
			decision.Action = "scale_up"
			decision.Reason = fmt.Sprintf("predicted max QPS %.0f > threshold %.0f, scaling from %d to %d",
				maxPredictedQPS, thresholdForScaleUp, currentReplicas, desired)
			decision.DesiredReplicas = desired
			return decision
		}
	}

	// 5. 缩容判断
	// 缩容更保守：只有当预测 QPS 持续低于 30% 容量才缩
	thresholdForScaleDown := currentCapacity * config.ScaleDownThreshold
	if maxPredictedQPS < thresholdForScaleDown && currentReplicas > config.MinReplicas {
		desired := int32(math.Ceil(maxPredictedQPS / config.QPSPerPod))
		// 缩容别一次缩太多，每次最多缩 30%
		maxReduction := int32(math.Ceil(float64(currentReplicas) * 0.3))
		minDesired := currentReplicas - maxReduction
		// 取 QPS 需求和最大减幅限制的较大值，避免缩过头
		if desired < minDesired {
			desired = minDesired
		}
		if desired < config.MinReplicas {
			desired = config.MinReplicas
		}

		decision.Action = "scale_down"
		decision.Reason = fmt.Sprintf("predicted max QPS %.0f < threshold %.0f, scaling down from %d to %d",
			maxPredictedQPS, thresholdForScaleDown, currentReplicas, desired)
		decision.DesiredReplicas = desired
		return decision
	}

	// 6. 不需要调整
	decision.Action = "no_action"
	decision.Reason = fmt.Sprintf("predicted max QPS %.0f within [%.0f, %.0f]",
		maxPredictedQPS, thresholdForScaleDown, thresholdForScaleUp)
	decision.DesiredReplicas = currentReplicas
	return decision
}

func clampReplicas(desired, min, max int32) int32 {
	if desired < min {
		return min
	}
	if desired > max {
		return max
	}
	return desired
}

// cmd/scaler/main.go
package main

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"

	"yourapp/pkg/scaler"
)

func main() {
	kubeconfig := os.Getenv("KUBECONFIG")
	predictorURL := os.Getenv("PREDICTOR_URL") // 如 http://qps-predictor:8080
	targetDeployment := os.Getenv("TARGET_DEPLOYMENT")
	targetNamespace := os.Getenv("TARGET_NAMESPACE")

	if predictorURL == "" || targetDeployment == "" {
		log.Fatal("PREDICTOR_URL and TARGET_DEPLOYMENT are required")
	}

	// K8s client
	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		log.Fatalf("build config failed: %v", err)
	}
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		log.Fatalf("create clientset failed: %v", err)
	}

	scaleConfig := scaler.DefaultConfig()
	// 记录上次扩缩容时间
	lastScaleTime := time.Time{} // zero value 表示从未扩缩容过

	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer cancel()

	ticker := time.NewTicker(60 * time.Second) // 每分钟决策一次
	defer ticker.Stop()

	log.Printf("scaler started, target=%s/%s, predictor=%s",
		targetNamespace, targetDeployment, predictorURL)

	// 首次立即执行
	runScaleCycle(ctx, clientset, predictorURL, targetNamespace, targetDeployment,
		&lastScaleTime, scaleConfig)

	for {
		select {
		case <-ctx.Done():
			log.Println("shutting down...")
			return
		case <-ticker.C:
			runScaleCycle(ctx, clientset, predictorURL, targetNamespace, targetDeployment,
				&lastScaleTime, scaleConfig)
		}
	}
}

func runScaleCycle(
	ctx context.Context,
	clientset *kubernetes.Clientset,
	predictorURL, namespace, deployment string,
	lastScaleTime *time.Time,
	config scaler.ScaleConfig,
) {
	// 1. 调推理服务获取预测
	predictions, err := fetchPrediction(ctx, predictorURL, config.PredictSteps)
	if err != nil {
		log.Printf("fetch prediction failed: %v", err)
		return
	}

	// 2. 获取当前副本数
	dep, err := clientset.AppsV1().Deployments(namespace).Get(ctx, deployment, metav1.GetOptions{})
	if err != nil {
		log.Printf("get deployment failed: %v", err)
		return
	}
	currentReplicas := *dep.Spec.Replicas

	// 3. 决策
	decision := scaler.Decide(predictions, currentReplicas, *lastScaleTime, config)

	log.Printf("decision: action=%s, current=%d, desired=%d, predictedMaxQPS=%.0f, reason=%s",
		decision.Action, decision.CurrentReplicas, decision.DesiredReplicas,
		decision.PredictedMaxQPS, decision.Reason)

	// 4. 执行
	if decision.Action == "no_action" {
		return
	}

	if decision.DesiredReplicas == currentReplicas {
		return
	}

	// 用 Scale subresource 改副本数
	scale, err := clientset.AppsV1().Deployments(namespace).GetScale(ctx, deployment, metav1.GetOptions{})
	if err != nil {
		log.Printf("get scale failed: %v", err)
		return
	}

	scale.Spec.Replicas = decision.DesiredReplicas
	_, err = clientset.AppsV1().Deployments(namespace).UpdateScale(ctx, deployment, scale, metav1.UpdateOptions{})
	if err != nil {
		log.Printf("update scale failed: %v", err)
		return
	}

	*lastScaleTime = time.Now()
	log.Printf("scaled %s/%s from %d to %d",
		namespace, deployment, currentReplicas, decision.DesiredReplicas)
}

// httpClient 带超时的 HTTP 客户端，避免推理服务无响应时永久阻塞
var httpClient = &http.Client{Timeout: 30 * time.Second}

// fetchPrediction 调推理服务获取预测
// 推理服务的 FastAPI 端点从 JSON body 接收 {"steps": N}
func fetchPrediction(ctx context.Context, predictorURL string, steps int) ([]scaler.PredictedPoint, error) {
	url := fmt.Sprintf("%s/predict", predictorURL)

	// 构造 JSON body：FastAPI 端用 Pydantic 模型从 body 接收参数
	reqBody, err := json.Marshal(map[string]int{"steps": steps})
	if err != nil {
		return nil, fmt.Errorf("marshal request body failed: %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(reqBody))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/json")

	resp, err := httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("call predictor failed: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return nil, fmt.Errorf("predictor returned %d", resp.StatusCode)
	}

	var result struct {
		Predictions []scaler.PredictedPoint `json:"predictions"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, fmt.Errorf("decode response failed: %w", err)
	}

	return result.Predictions, nil
}

# api-gateway-hpa.yaml
# HPA 配置：预测器负责提前扩容，HPA 兜底处理突发流量
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: api-gateway-hpa
  namespace: prod
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: api-gateway
  minReplicas: 3       # 跟预测器的 MinReplicas 保持一致
  maxReplicas: 50       # 跟预测器的 MaxReplicas 保持一致
  metrics:
    # 基于 CPU 利用率做精细调整
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 60  # CPU > 60% 就扩容
    # 基于内存做保护
    - type: Resource
      resource:
        name: memory
        target:
          type: Utilization
          averageUtilization: 80
  behavior:
    # 扩容行为：快扩
    scaleUp:
      stabilizationWindowSeconds: 0  # 扩容不等待，立刻执行
      policies:
        # 每次最多扩容 100%（翻倍）
        - type: Percent
          value: 100
          periodSeconds: 15
        # 或者每次最多加 4 个
        - type: Pods
          value: 4
          periodSeconds: 15
      selectPolicy: Max  # 取两个策略中扩容更快的那个
    # 缩容行为：慢缩，避免抖动
    scaleDown:
      stabilizationWindowSeconds: 300  # 缩容前等 5 分钟确认流量真的降了
      policies:
        # 每次最多缩 10%
        - type: Percent
          value: 10
          periodSeconds: 60
      selectPolicy: Min  # 取更保守的策略

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  数据版本化  │───▶│  实验追踪    │───▶│  模型注册    │───▶│  CI/CD 部署  │
│  DVC        │    │  MLflow     │    │  MLflow     │    │  ArgoCD     │
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘
                                                                 │
                                                                 ▼
┌─────────────────────────────────────────────────────────────────────┐
│                        监控与漂移检测                                  │
│                                                                     │
│  预测误差监控 ──▶ MAPE > 20%? ──▶ 触发重训 ──▶ 回到数据版本化         │
│  (Prometheus)    (Alertmanager)  (Airflow)                          │
└─────────────────────────────────────────────────────────────────────┘

# 预测误差率（MAPE）
# prediction_error = abs(predicted_qps - actual_qps) / actual_qps
avg_over_time(
  prediction_abs_error[1h]
) / avg_over_time(
  actual_qps[1h]
) * 100 > 20