企业级电商数据采集架构：基于Pangolin Scrape API的云原生解决方案

┌─────────────────────────────────────────────────────────────────┐
│                        企业级数据采集平台                        │
├─────────────────────────────────────────────────────────────────┤
│  前端层    │  API网关  │  业务服务层  │  数据层  │  基础设施层  │
├─────────────────────────────────────────────────────────────────┤
│            │           │              │          │              │
│ 管理控制台 │  腾讯云   │   采集服务   │ 腾讯云   │   腾讯云     │
│ 数据大屏   │  API网关  │   解析服务   │ COS      │   CVM        │
│ 监控告警   │           │   调度服务   │ CDB      │   TKE        │
│            │           │   通知服务   │ Redis    │   VPC        │
│            │           │              │ Kafka    │   CLB        │
└─────────────────────────────────────────────────────────────────┘
                                │
                                ▼
                    ┌─────────────────────┐
                    │   Pangolin API      │
                    │                     │
                    │ - 数据采集          │
                    │ - 智能解析          │
                    │ - 反检测技术        │
                    └─────────────────────┘

# services/scheduler/scheduler.py
import asyncio
import json
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
from tencentcloud.common import credential
from tencentcloud.cvm.v20170312 import cvm_client, models as cvm_models
from tencentcloud.tke.v20180525 import tke_client, models as tke_models

@dataclass
class ScrapingTask:
    task_id: str
    task_type: str  # 'product', 'search', 'category'
    priority: int   # 1-10, 10为最高优先级
    urls: List[str]
    schedule_time: datetime
    retry_count: int = 0
    max_retries: int = 3
    status: str = 'pending'  # pending, running, completed, failed
    created_at: datetime = None
    updated_at: datetime = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()
        self.updated_at = datetime.now()

class EnterpriseScheduler:
    def __init__(self, config: Dict):
        self.config = config
        self.cred = credential.Credential(
            config['secret_id'], 
            config['secret_key']
        )
        self.task_queue = asyncio.PriorityQueue()
        self.running_tasks = {}
        self.completed_tasks = {}
        
        # 初始化腾讯云客户端
        self.cvm_client = cvm_client.CvmClient(self.cred, config['region'])
        self.tke_client = tke_client.TkeClient(self.cred, config['region'])
    
    async def submit_task(self, task: ScrapingTask) -> str:
        """提交采集任务"""
        # 任务验证
        if not self._validate_task(task):
            raise ValueError("Invalid task configuration")
        
        # 优先级队列：负数表示高优先级
        priority = -task.priority
        await self.task_queue.put((priority, task.task_id, task))
        
        # 记录任务到数据库
        await self._persist_task(task)
        
        return task.task_id
    
    async def process_tasks(self):
        """处理任务队列"""
        while True:
            try:
                # 获取任务
                priority, task_id, task = await self.task_queue.get()
                
                # 检查资源可用性
                if await self._check_resource_availability():
                    # 执行任务
                    await self._execute_task(task)
                else:
                    # 资源不足，重新排队
                    await asyncio.sleep(30)
                    await self.task_queue.put((priority, task_id, task))
                
            except Exception as e:
                logger.error(f"Task processing error: {e}")
                await asyncio.sleep(10)
    
    async def _execute_task(self, task: ScrapingTask):
        """执行采集任务"""
        task.status = 'running'
        task.updated_at = datetime.now()
        self.running_tasks[task.task_id] = task
        
        try:
            # 根据任务类型选择执行策略
            if task.task_type == 'product':
                result = await self._execute_product_task(task)
            elif task.task_type == 'search':
                result = await self._execute_search_task(task)
            elif task.task_type == 'category':
                result = await self._execute_category_task(task)
            else:
                raise ValueError(f"Unknown task type: {task.task_type}")
            
            # 任务完成
            task.status = 'completed'
            task.updated_at = datetime.now()
            self.completed_tasks[task.task_id] = task
            
            # 存储结果
            await self._store_results(task.task_id, result)
            
        except Exception as e:
            # 任务失败处理
            task.retry_count += 1
            if task.retry_count <= task.max_retries:
                task.status = 'pending'
                # 指数退避重试
                delay = 2 ** task.retry_count * 60
                task.schedule_time = datetime.now() + timedelta(seconds=delay)
                await self.task_queue.put((-task.priority, task.task_id, task))
            else:
                task.status = 'failed'
                await self._handle_task_failure(task, str(e))
        
        finally:
            # 清理运行中任务
            self.running_tasks.pop(task.task_id, None)
    
    async def _check_resource_availability(self) -> bool:
        """检查资源可用性"""
        try:
            # 检查CVM实例状态
            req = cvm_models.DescribeInstancesRequest()
            resp = self.cvm_client.DescribeInstances(req)
            
            running_instances = sum(1 for instance in resp.InstanceSet 
                                  if instance.InstanceState == 'RUNNING')
            
            # 检查当前负载
            current_load = len(self.running_tasks)
            max_concurrent = self.config.get('max_concurrent_tasks', 100)
            
            return (running_instances > 0 and 
                   current_load < max_concurrent)
            
        except Exception as e:
            logger.error(f"Resource check failed: {e}")
            return False
    
    async def auto_scale_resources(self):
        """自动扩缩容"""
        while True:
            try:
                queue_size = self.task_queue.qsize()
                running_count = len(self.running_tasks)
                
                # 扩容条件：队列积压 > 50 且运行任务 < 最大值
                if queue_size > 50 and running_count < self.config['max_concurrent_tasks']:
                    await self._scale_up()
                
                # 缩容条件：队列为空且运行任务 < 10
                elif queue_size == 0 and running_count < 10:
                    await self._scale_down()
                
                await asyncio.sleep(300)  # 5分钟检查一次
                
            except Exception as e:
                logger.error(f"Auto scaling error: {e}")
                await asyncio.sleep(60)
    
    async def _scale_up(self):
        """扩容处理"""
        # 通过TKE扩展Pod数量
        try:
            # 这里可以调用TKE API进行扩容
            logger.info("Scaling up resources...")
            # 实际实现需要根据具体的K8s配置
        except Exception as e:
            logger.error(f"Scale up failed: {e}")
    
    async def _scale_down(self):
        """缩容处理"""
        try:
            logger.info("Scaling down resources...")
            # 实际实现需要根据具体的K8s配置
        except Exception as e:
            logger.error(f"Scale down failed: {e}")

# services/storage/cos_storage.py
import json
import gzip
from typing import Dict, List, Optional
from datetime import datetime
from tencentcloud.cos.v20180717 import cos_client, models as cos_models
from tencentcloud.cdb.v20170320 import cdb_client, models as cdb_models

class EnterpriseDataStorage:
    def __init__(self, config: Dict):
        self.config = config
        self.cred = credential.Credential(
            config['secret_id'], 
            config['secret_key']
        )
        
        # 初始化COS客户端（对象存储）
        self.cos_client = cos_client.CosClient(self.cred, config['region'])
        
        # 初始化CDB客户端（云数据库）
        self.cdb_client = cdb_client.CdbClient(self.cred, config['region'])
        
        self.bucket_name = config['cos_bucket']
        self.db_instance_id = config['cdb_instance_id']
    
    async def store_raw_data(self, task_id: str, data: Dict) -> str:
        """存储原始数据到COS"""
        try:
            # 数据压缩
            json_data = json.dumps(data, ensure_ascii=False)
            compressed_data = gzip.compress(json_data.encode('utf-8'))
            
            # 生成存储路径
            date_path = datetime.now().strftime('%Y/%m/%d')
            object_key = f"raw_data/{date_path}/{task_id}.json.gz"
            
            # 上传到COS
            req = cos_models.PutObjectRequest()
            req.Bucket = self.bucket_name
            req.Key = object_key
            req.Body = compressed_data
            req.ContentType = 'application/gzip'
            
            resp = self.cos_client.PutObject(req)
            
            # 记录元数据到数据库
            await self._record_storage_metadata(task_id, object_key, len(compressed_data))
            
            return object_key
            
        except Exception as e:
            logger.error(f"Failed to store raw data: {e}")
            raise
    
    async def store_processed_data(self, task_id: str, products: List[Dict]) -> bool:
        """存储处理后的结构化数据"""
        try:
            # 批量插入到数据库
            await self._batch_insert_products(products)
            
            # 同时备份到COS
            date_path = datetime.now().strftime('%Y/%m/%d')
            object_key = f"processed_data/{date_path}/{task_id}_products.json"
            
            json_data = json.dumps(products, ensure_ascii=False, indent=2)
            
            req = cos_models.PutObjectRequest()
            req.Bucket = self.bucket_name
            req.Key = object_key
            req.Body = json_data.encode('utf-8')
            req.ContentType = 'application/json'
            
            self.cos_client.PutObject(req)
            
            return True
            
        except Exception as e:
            logger.error(f"Failed to store processed data: {e}")
            return False
    
    async def _batch_insert_products(self, products: List[Dict]):
        """批量插入产品数据"""
        # 这里使用云数据库CDB进行批量插入
        # 实际实现需要根据具体的数据库schema
        
        sql_template = """
        INSERT INTO products (
            asin, title, price, currency, brand, category,
            rating, review_count, availability, features,
            images, description, scraped_at
        ) VALUES %s
        ON DUPLICATE KEY UPDATE
            price = VALUES(price),
            rating = VALUES(rating),
            review_count = VALUES(review_count),
            availability = VALUES(availability),
            scraped_at = VALUES(scraped_at)
        """
        
        # 构建批量插入数据
        values = []
        for product in products:
            values.append((
                product.get('asin', ''),
                product.get('title', ''),
                product.get('price', 0),
                product.get('currency', 'USD'),
                product.get('brand', ''),
                product.get('category', ''),
                product.get('rating', 0),
                product.get('review_count', 0),
                product.get('availability', ''),
                json.dumps(product.get('features', [])),
                json.dumps(product.get('images', [])),
                product.get('description', ''),
                datetime.now()
            ))
        
        # 执行批量插入（这里简化了实际的数据库操作）
        logger.info(f"Batch inserting {len(values)} products")
    
    async def get_data_statistics(self) -> Dict:
        """获取数据统计信息"""
        try:
            # 从数据库获取统计信息
            stats = {
                'total_products': 0,
                'today_scraped': 0,
                'storage_usage_gb': 0,
                'avg_daily_growth': 0
            }
            
            # 查询总产品数
            # stats['total_products'] = await self._query_total_products()
            
            # 查询今日采集数量
            # stats['today_scraped'] = await self._query_today_scraped()
            
            # 查询存储使用量
            stats['storage_usage_gb'] = await self._get_cos_usage()
            
            return stats
            
        except Exception as e:
            logger.error(f"Failed to get statistics: {e}")
            return {}
    
    async def _get_cos_usage(self) -> float:
        """获取COS存储使用量"""
        try:
            req = cos_models.GetBucketRequest()
            req.Bucket = self.bucket_name
            
            resp = self.cos_client.GetBucket(req)
            
            total_size = sum(obj.Size for obj in resp.Contents)
            return total_size / (1024 ** 3)  # 转换为GB
            
        except Exception as e:
            logger.error(f"Failed to get COS usage: {e}")
            return 0.0

# services/monitoring/monitor.py
import asyncio
from typing import Dict, List
from dataclasses import dataclass
from datetime import datetime, timedelta
from tencentcloud.monitor.v20180724 import monitor_client, models as monitor_models
from tencentcloud.sms.v20210111 import sms_client, models as sms_models

@dataclass
class AlertRule:
    name: str
    metric: str
    threshold: float
    comparison: str  # '>', '<', '>=', '<=', '=='
    duration: int    # 持续时间（秒）
    severity: str    # 'critical', 'warning', 'info'
    enabled: bool = True

class EnterpriseMonitor:
    def __init__(self, config: Dict):
        self.config = config
        self.cred = credential.Credential(
            config['secret_id'], 
            config['secret_key']
        )
        
        # 初始化监控客户端
        self.monitor_client = monitor_client.MonitorClient(self.cred, config['region'])
        self.sms_client = sms_client.SmsClient(self.cred, config['region'])
        
        # 告警规则
        self.alert_rules = [
            AlertRule('API成功率', 'api_success_rate', 95.0, '<', 300, 'critical'),
            AlertRule('任务队列积压', 'queue_size', 1000, '>', 600, 'warning'),
            AlertRule('数据库连接数', 'db_connections', 80, '>', 300, 'warning'),
            AlertRule('存储使用率', 'storage_usage', 85.0, '>', 900, 'warning'),
            AlertRule('内存使用率', 'memory_usage', 90.0, '>', 300, 'critical'),
        ]
        
        self.metrics_cache = {}
        self.alert_history = {}
    
    async def collect_metrics(self):
        """收集系统指标"""
        while True:
            try:
                current_time = datetime.now()
                
                # 收集各类指标
                metrics = {
                    'api_success_rate': await self._get_api_success_rate(),
                    'queue_size': await self._get_queue_size(),
                    'db_connections': await self._get_db_connections(),
                    'storage_usage': await self._get_storage_usage(),
                    'memory_usage': await self._get_memory_usage(),
                    'cpu_usage': await self._get_cpu_usage(),
                    'network_io': await self._get_network_io(),
                    'disk_io': await self._get_disk_io(),
                }
                
                # 更新指标缓存
                self.metrics_cache[current_time] = metrics
                
                # 清理过期数据（保留24小时）
                cutoff_time = current_time - timedelta(hours=24)
                self.metrics_cache = {
                    k: v for k, v in self.metrics_cache.items() 
                    if k > cutoff_time
                }
                
                # 上报到腾讯云监控
                await self._report_custom_metrics(metrics)
                
                # 检查告警规则
                await self._check_alert_rules(metrics)
                
                await asyncio.sleep(60)  # 每分钟收集一次
                
            except Exception as e:
                logger.error(f"Metrics collection error: {e}")
                await asyncio.sleep(30)
    
    async def _get_api_success_rate(self) -> float:
        """获取API成功率"""
        # 从应用日志或数据库查询最近5分钟的API调用成功率
        # 这里简化实现
        return 98.5
    
    async def _get_queue_size(self) -> int:
        """获取任务队列大小"""
        # 从Redis或消息队列获取当前队列大小
        return 150
    
    async def _get_db_connections(self) -> int:
        """获取数据库连接数"""
        # 从数据库监控获取当前连接数
        return 45
    
    async def _get_storage_usage(self) -> float:
        """获取存储使用率"""
        # 从COS监控获取存储使用率
        return 72.3
    
    async def _get_memory_usage(self) -> float:
        """获取内存使用率"""
        # 从CVM监控获取内存使用率
        return 68.5
    
    async def _get_cpu_usage(self) -> float:
        """获取CPU使用率"""
        return 45.2
    
    async def _get_network_io(self) -> Dict:
        """获取网络IO"""
        return {'in': 1024000, 'out': 2048000}  # bytes/s
    
    async def _get_disk_io(self) -> Dict:
        """获取磁盘IO"""
        return {'read': 50, 'write': 30}  # MB/s
    
    async def _report_custom_metrics(self, metrics: Dict):
        """上报自定义指标到腾讯云监控"""
        try:
            req = monitor_models.PutMonitorDataRequest()
            req.Metrics = []
            
            for metric_name, value in metrics.items():
                if isinstance(value, (int, float)):
                    metric = monitor_models.MetricDatum()
                    metric.MetricName = metric_name
                    metric.Value = float(value)
                    metric.Timestamp = int(datetime.now().timestamp())
                    req.Metrics.append(metric)
            
            resp = self.monitor_client.PutMonitorData(req)
            logger.debug(f"Reported {len(req.Metrics)} metrics to cloud monitor")
            
        except Exception as e:
            logger.error(f"Failed to report metrics: {e}")
    
    async def _check_alert_rules(self, current_metrics: Dict):
        """检查告警规则"""
        for rule in self.alert_rules:
            if not rule.enabled:
                continue
            
            metric_value = current_metrics.get(rule.metric)
            if metric_value is None:
                continue
            
            # 检查阈值
            triggered = self._evaluate_condition(metric_value, rule.threshold, rule.comparison)
            
            if triggered:
                # 检查持续时间
                if await self._check_duration(rule, current_metrics):
                    await self._send_alert(rule, metric_value)
    
    def _evaluate_condition(self, value: float, threshold: float, comparison: str) -> bool:
        """评估告警条件"""
        if comparison == '>':
            return value > threshold
        elif comparison == '<':
            return value < threshold
        elif comparison == '>=':
            return value >= threshold
        elif comparison == '<=':
            return value <= threshold
        elif comparison == '==':
            return abs(value - threshold) < 0.001
        return False
    
    async def _check_duration(self, rule: AlertRule, current_metrics: Dict) -> bool:
        """检查告警持续时间"""
        # 检查过去rule.duration秒内是否持续触发
        cutoff_time = datetime.now() - timedelta(seconds=rule.duration)
        
        triggered_count = 0
        total_count = 0
        
        for timestamp, metrics in self.metrics_cache.items():
            if timestamp < cutoff_time:
                continue
            
            total_count += 1
            metric_value = metrics.get(rule.metric)
            if metric_value and self._evaluate_condition(metric_value, rule.threshold, rule.comparison):
                triggered_count += 1
        
        # 如果80%以上的时间都在触发，则认为满足持续时间条件
        return total_count > 0 and (triggered_count / total_count) >= 0.8
    
    async def _send_alert(self, rule: AlertRule, current_value: float):
        """发送告警"""
        alert_key = f"{rule.name}_{rule.severity}"
        
        # 防止重复告警（1小时内同一告警只发送一次）
        last_alert_time = self.alert_history.get(alert_key)
        if last_alert_time and (datetime.now() - last_alert_time).seconds < 3600:
            return
        
        # 构建告警消息
        message = f"""
        🚨 【{rule.severity.upper()}】系统告警
        
        告警规则: {rule.name}
        当前值: {current_value}
        阈值: {rule.comparison} {rule.threshold}
        时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
        
        请及时处理！
        """
        
        # 发送短信告警
        await self._send_sms_alert(message)
        
        # 发送邮件告警
        await self._send_email_alert(rule, message)
        
        # 记录告警历史
        self.alert_history[alert_key] = datetime.now()
        
        logger.warning(f"Alert sent: {rule.name} = {current_value}")
    
    async def _send_sms_alert(self, message: str):
        """发送短信告警"""
        try:
            req = sms_models.SendSmsRequest()
            req.PhoneNumberSet = self.config.get('alert_phones', [])
            req.SmsSdkAppId = self.config['sms_app_id']
            req.SignName = self.config['sms_sign']
            req.TemplateId = self.config['sms_template_id']
            req.TemplateParamSet = [message[:100]]  # 短信长度限制
            
            resp = self.sms_client.SendSms(req)
            logger.info(f"SMS alert sent to {len(req.PhoneNumberSet)} recipients")
            
        except Exception as e:
            logger.error(f"Failed to send SMS alert: {e}")
    
    async def _send_email_alert(self, rule: AlertRule, message: str):
        """发送邮件告警"""
        # 这里可以集成邮件服务
        logger.info(f"Email alert would be sent: {rule.name}")
    
    async def get_dashboard_data(self) -> Dict:
        """获取监控面板数据"""
        if not self.metrics_cache:
            return {}
        
        # 获取最近的指标数据
        latest_metrics = list(self.metrics_cache.values())[-1]
        
        # 计算趋势数据
        recent_data = list(self.metrics_cache.items())[-60:]  # 最近60个数据点
        
        dashboard = {
            'current_metrics': latest_metrics,
            'trends': {},
            'alerts': {
                'active_count': len([r for r in self.alert_rules if r.enabled]),
                'recent_alerts': list(self.alert_history.keys())[-10:]
            },
            'system_health': self._calculate_system_health(latest_metrics)
        }
        
        # 计算各指标的趋势
        for metric_name in latest_metrics.keys():
            values = [data[metric_name] for _, data in recent_data if metric_name in data]
            if len(values) >= 2:
                dashboard['trends'][metric_name] = {
                    'current': values[-1],
                    'previous': values[-2],
                    'change_percent': ((values[-1] - values[-2]) / values[-2] * 100) if values[-2] != 0 else 0,
                    'history': values[-20:]  # 最近20个点
                }
        
        return dashboard
    
    def _calculate_system_health(self, metrics: Dict) -> str:
        """计算系统健康状态"""
        critical_metrics = {
            'api_success_rate': (95.0, '>'),
            'memory_usage': (90.0, '<'),
            'cpu_usage': (85.0, '<'),
            'storage_usage': (90.0, '<')
        }
        
        health_score = 100
        
        for metric, (threshold, comparison) in critical_metrics.items():
            value = metrics.get(metric, 0)
            
            if comparison == '>' and value < threshold:
                health_score -= 20
            elif comparison == '<' and value > threshold:
                health_score -= 20
        
        if health_score >= 90:
            return 'excellent'
        elif health_score >= 70:
            return 'good'
        elif health_score >= 50:
            return 'warning'
        else:
            return 'critical'

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: pangolin-scraper
  namespace: data-collection
spec:
  replicas: 3
  selector:
    matchLabels:
      app: pangolin-scraper
  template:
    metadata:
      labels:
        app: pangolin-scraper
    spec:
      containers:
      - name: scraper
        image: pangolin-scraper:latest
        ports:
        - containerPort: 8000
        env:
        - name: PANGOLIN_API_KEY
          valueFrom:
            secretKeyRef:
              name: pangolin-secret
              key: api-key
        - name: REDIS_URL
          value: "redis://redis-service:6379"
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: connection-string
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: pangolin-scraper-service
  namespace: data-collection
spec:
  selector:
    app: pangolin-scraper
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

# k8s/hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: pangolin-scraper-hpa
  namespace: data-collection
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: pangolin-scraper
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  - type: Pods
    pods:
      metric:
        name: queue_size
      target:
        type: AverageValue
        averageValue: "100"
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 100
        periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60

# k8s/configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: pangolin-config
  namespace: data-collection
data:
  config.yaml: |
    # 应用配置
    app:
      name: "pangolin-scraper"
      version: "1.0.0"
      environment: "production"
    
    # API配置
    pangolin:
      base_url: "https://api.pangolinfo.com"
      timeout: 30
      max_retries: 3
      rate_limit: 1000  # requests per minute
    
    # 数据库配置
    database:
      pool_size: 20
      max_overflow: 30
      pool_timeout: 30
      pool_recycle: 3600
    
    # 缓存配置
    redis:
      pool_size: 20
      socket_timeout: 5
      socket_connect_timeout: 5
      retry_on_timeout: true
    
    # 监控配置
    monitoring:
      metrics_interval: 60
      health_check_interval: 30
      alert_cooldown: 3600
    
    # 日志配置
    logging:
      level: "INFO"
      format: "json"
      max_file_size: "100MB"
      backup_count: 10

# utils/cost_optimizer.py
import asyncio
from typing import Dict, List
from datetime import datetime, timedelta
from dataclasses import dataclass

@dataclass
class ResourceUsage:
    timestamp: datetime
    cpu_usage: float
    memory_usage: float
    network_io: float
    storage_io: float
    cost_per_hour: float

class CostOptimizer:
    def __init__(self, config: Dict):
        self.config = config
        self.usage_history = []
        self.optimization_rules = [
            self._optimize_instance_type,
            self._optimize_storage_tier,
            self._optimize_network_bandwidth,
            self._optimize_scheduling
        ]
    
    async def analyze_usage_patterns(self) -> Dict:
        """分析资源使用模式"""
        if len(self.usage_history) < 24:  # 至少需要24小时数据
            return {}
        
        # 按小时分组分析
        hourly_usage = {}
        for usage in self.usage_history[-168:]:  # 最近7天
            hour = usage.timestamp.hour
            if hour not in hourly_usage:
                hourly_usage[hour] = []
            hourly_usage[hour].append(usage)
        
        # 计算每小时平均使用率
        patterns = {}
        for hour, usages in hourly_usage.items():
            patterns[hour] = {
                'avg_cpu': sum(u.cpu_usage for u in usages) / len(usages),
                'avg_memory': sum(u.memory_usage for u in usages) / len(usages),
                'avg_cost': sum(u.cost_per_hour for u in usages) / len(usages),
                'peak_cpu': max(u.cpu_usage for u in usages),
                'peak_memory': max(u.memory_usage for u in usages)
            }
        
        return patterns
    
    async def generate_optimization_recommendations(self) -> List[Dict]:
        """生成优化建议"""
        recommendations = []
        
        for rule in self.optimization_rules:
            try:
                suggestion = await rule()
                if suggestion:
                    recommendations.append(suggestion)
            except Exception as e:
                logger.error(f"Optimization rule failed: {e}")
        
        return recommendations
    
    async def _optimize_instance_type(self) -> Dict:
        """实例类型优化"""
        patterns = await self.analyze_usage_patterns()
        if not patterns:
            return {}
        
        # 计算平均资源使用率
        avg_cpu = sum(p['avg_cpu'] for p in patterns.values()) / len(patterns)
        avg_memory = sum(p['avg_memory'] for p in patterns.values()) / len(patterns)
        
        current_cost = sum(p['avg_cost'] for p in patterns.values())
        
        # 推荐实例类型
        if avg_cpu < 30 and avg_memory < 50:
            recommended_type = "S5.MEDIUM4"
            estimated_savings = current_cost * 0.3
        elif avg_cpu < 60 and avg_memory < 70:
            recommended_type = "S5.LARGE8"
            estimated_savings = current_cost * 0.15
        else:
            return {}
        
        return {
            'type': 'instance_optimization',
            'title': '实例规格优化',
            'description': f'当前CPU使用率{avg_cpu:.1f}%，内存使用率{avg_memory:.1f}%，建议调整实例规格',
            'recommendation': f'建议使用{recommended_type}实例',
            'estimated_savings': estimated_savings,
            'implementation_effort': 'medium'
        }
    
    async def _optimize_storage_tier(self) -> Dict:
        """存储层级优化"""
        # 分析数据访问模式
        hot_data_ratio = 0.2  # 热数据比例
        warm_data_ratio = 0.3  # 温数据比例
        cold_data_ratio = 0.5  # 冷数据比例
        
        # 计算存储成本优化
        current_storage_cost = 1000  # 假设当前月存储成本
        
        # 使用智能分层存储
        optimized_cost = (
            current_storage_cost * hot_data_ratio * 1.0 +  # 标准存储
            current_storage_cost * warm_data_ratio * 0.7 +  # 低频存储
            current_storage_cost * cold_data_ratio * 0.3    # 归档存储
        )
        
        savings = current_storage_cost - optimized_cost
        
        if savings > 100:  # 节省超过100元才推荐
            return {
                'type': 'storage_optimization',
                'title': '存储分层优化',
                'description': '通过智能分层存储降低存储成本',
                'recommendation': '启用COS智能分层，自动将冷数据迁移到低成本存储',
                'estimated_savings': savings,
                'implementation_effort': 'low'
            }
        
        return {}
    
    async def _optimize_network_bandwidth(self) -> Dict:
        """网络带宽优化"""
        # 分析网络使用模式
        peak_bandwidth = 100  # MB/s
        avg_bandwidth = 30   # MB/s
        
        if peak_bandwidth / avg_bandwidth > 3:
            return {
                'type': 'network_optimization',
                'title': '网络带宽优化',
                'description': '网络使用存在明显峰谷差异，建议使用弹性带宽',
                'recommendation': '配置自动调整带宽，按实际使用量计费',
                'estimated_savings': 200,  # 月节省金额
                'implementation_effort': 'low'
            }
        
        return {}
    
    async def _optimize_scheduling(self) -> Dict:
        """任务调度优化"""
        # 分析任务执行模式
        patterns = await self.analyze_usage_patterns()
        if not patterns:
            return {}
        
        # 找出低负载时段
        low_load_hours = [
            hour for hour, pattern in patterns.items()
            if pattern['avg_cpu'] < 40 and pattern['avg_memory'] < 50
        ]
        
        if len(low_load_hours) >= 6:  # 有6小时以上的低负载时段
            return {
                'type': 'scheduling_optimization',
                'title': '任务调度优化',
                'description': f'发现{len(low_load_hours)}小时低负载时段，可优化任务调度',
                'recommendation': '将非紧急任务调度到低负载时段执行，提高资源利用率',
                'estimated_savings': 150,
                'implementation_effort': 'medium'
            }
        
        return {}
    
    async def calculate_roi(self, time_period_days: int = 30) -> Dict:
        """计算投资回报率"""
        # 计算使用Pangolin API vs 自建团队的成本对比
        
        # 自建团队成本（月）
        self_built_costs = {
            'developer_salary': 50000,      # 2名高级开发工程师
            'devops_salary': 25000,         # 1名运维工程师
            'infrastructure': 15000,        # 服务器、带宽等
            'proxy_services': 8000,         # 代理服务
            'maintenance': 5000,            # 维护成本
            'total': 103000
        }
        
        # Pangolin API成本（月）
        pangolin_costs = {
            'api_calls': 20000,             # API调用费用
            'infrastructure': 5000,         # 基础设施（简化）
            'development': 10000,           # 开发集成成本（摊销）
            'total': 35000
        }
        
        # 计算ROI
        monthly_savings = self_built_costs['total'] - pangolin_costs['total']
        annual_savings = monthly_savings * 12
        
        # 计算投资回报率
        initial_investment = 50000  # 初始集成成本
        roi_percentage = (annual_savings - initial_investment) / initial_investment * 100
        
        return {
            'monthly_savings': monthly_savings,
            'annual_savings': annual_savings,
            'roi_percentage': roi_percentage,
            'payback_period_months': initial_investment / monthly_savings if monthly_savings > 0 else float('inf'),
            'cost_breakdown': {
                'self_built': self_built_costs,
                'pangolin': pangolin_costs
            }
        }