"""Baseline calculation module""" import statistics from collections import defaultdict from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Optional, Tuple from ..data.csv_handler import CSVHandler from ..common.utils import safe_float class BaselineCalculator: """基线计算器""" def __init__(self, baseline_days: int = 30, min_sample_size: int = 7): """初始化基线计算器 Args: baseline_days: 基线计算使用的天数 min_sample_size: 最小样本数量 """ self.baseline_days = baseline_days self.min_sample_size = min_sample_size self.csv_handler = CSVHandler() def load_historical_data(self, data_dirs: List[Path], account_id: str, target_date: datetime, environment: str = None) -> List[Dict]: """加载历史数据 Args: data_dirs: 数据目录列表 account_id: 账户ID target_date: 目标日期 environment: 环境名称 Returns: 历史记录列表 """ all_records = [] start_date = target_date - timedelta(days=self.baseline_days) for data_dir in data_dirs: if not data_dir.exists(): continue # 查找匹配的CSV文件 pattern = f"*{account_id}_*.csv" files = list(data_dir.glob(pattern)) for file_path in files: try: records = self.csv_handler.read_file(str(file_path)) # 过滤日期范围和环境 for record in records: try: record_date = datetime.strptime(record['date'], '%Y-%m-%d') # 检查日期范围 if start_date <= record_date < target_date: # 检查环境匹配 if environment is None or record.get('environment') == environment: all_records.append(record) except (ValueError, KeyError): continue except Exception as e: print(f"读取文件失败 {file_path}: {e}") continue return all_records def calculate_daily_baseline(self, records: List[Dict], dimension: str = 'total') -> Dict: """计算日均基线 Args: records: 历史记录 dimension: 计算维度 ('total', 'account', 'service', 'tag') Returns: 基线统计信息 """ if dimension == 'total': return self._calculate_total_baseline(records) elif dimension == 'account': return self._calculate_account_baseline(records) elif dimension == 'service': return self._calculate_service_baseline(records) elif dimension == 'tag': return self._calculate_tag_baseline(records) else: raise ValueError(f"不支持的维度: {dimension}") def _calculate_total_baseline(self, records: List[Dict]) -> Dict: """计算总成本基线""" daily_costs = defaultdict(float) for record in records: date = record['date'] cost = safe_float(record.get('cost', 0)) daily_costs[date] += cost costs = [cost for cost in daily_costs.values() if cost > 0] return self._compute_statistics(costs) def _calculate_account_baseline(self, records: List[Dict]) -> Dict[str, Dict]: """计算按账户的基线""" account_daily_costs = defaultdict(lambda: defaultdict(float)) for record in records: account_id = record.get('account_id', 'unknown') date = record['date'] cost = safe_float(record.get('cost', 0)) account_daily_costs[account_id][date] += cost baselines = {} for account_id, daily_costs in account_daily_costs.items(): costs = [cost for cost in daily_costs.values() if cost > 0] if len(costs) >= self.min_sample_size: baselines[account_id] = self._compute_statistics(costs) return baselines def _calculate_service_baseline(self, records: List[Dict]) -> Dict[str, Dict]: """计算按服务的基线""" service_daily_costs = defaultdict(lambda: defaultdict(float)) for record in records: service = record.get('product', 'Unknown') date = record['date'] cost = safe_float(record.get('cost', 0)) service_daily_costs[service][date] += cost baselines = {} for service, daily_costs in service_daily_costs.items(): costs = [cost for cost in daily_costs.values() if cost > 0] if len(costs) >= self.min_sample_size: baselines[service] = self._compute_statistics(costs) return baselines def _calculate_tag_baseline(self, records: List[Dict], tag_field: str = 'tag_usedby') -> Dict[str, Dict]: """计算按标签的基线""" tag_daily_costs = defaultdict(lambda: defaultdict(float)) for record in records: tag_value = record.get(tag_field, '').strip() if not tag_value: continue date = record['date'] cost = safe_float(record.get('cost', 0)) tag_daily_costs[tag_value][date] += cost baselines = {} for tag, daily_costs in tag_daily_costs.items(): costs = [cost for cost in daily_costs.values() if cost > 0] if len(costs) >= self.min_sample_size: baselines[tag] = self._compute_statistics(costs) return baselines def _compute_statistics(self, costs: List[float]) -> Dict: """计算统计信息 Args: costs: 成本列表 Returns: 统计信息字典 """ if not costs: return { 'avg_cost': 0, 'median_cost': 0, 'std_cost': 0, 'max_cost': 0, 'min_cost': 0, 'sample_size': 0, 'percentile_75': 0, 'percentile_90': 0, 'percentile_95': 0 } sorted_costs = sorted(costs) n = len(sorted_costs) return { 'avg_cost': statistics.mean(costs), 'median_cost': statistics.median(costs), 'std_cost': statistics.stdev(costs) if len(costs) > 1 else 0, 'max_cost': max(costs), 'min_cost': min(costs), 'sample_size': len(costs), 'percentile_75': sorted_costs[int(n * 0.75)] if n > 0 else 0, 'percentile_90': sorted_costs[int(n * 0.90)] if n > 0 else 0, 'percentile_95': sorted_costs[int(n * 0.95)] if n > 0 else 0 } def calculate_trend(self, records: List[Dict], window_days: int = 7) -> Dict: """计算成本趋势 Args: records: 历史记录 window_days: 滑动窗口天数 Returns: 趋势信息 """ daily_costs = defaultdict(float) for record in records: date = record['date'] cost = safe_float(record.get('cost', 0)) daily_costs[date] += cost # 按日期排序 sorted_dates = sorted(daily_costs.keys()) if len(sorted_dates) < window_days * 2: return {'trend': 'insufficient_data'} # 计算早期和晚期的平均成本 early_costs = [daily_costs[date] for date in sorted_dates[:window_days]] late_costs = [daily_costs[date] for date in sorted_dates[-window_days:]] early_avg = statistics.mean(early_costs) if early_costs else 0 late_avg = statistics.mean(late_costs) if late_costs else 0 if early_avg > 0: trend_ratio = late_avg / early_avg change_percentage = (trend_ratio - 1) * 100 if trend_ratio > 1.1: trend_direction = 'increasing' elif trend_ratio < 0.9: trend_direction = 'decreasing' else: trend_direction = 'stable' else: trend_direction = 'unknown' change_percentage = 0 return { 'trend': trend_direction, 'early_avg': early_avg, 'late_avg': late_avg, 'change_percentage': change_percentage, 'sample_days': len(sorted_dates) } def detect_seasonality(self, records: List[Dict]) -> Dict: """检测季节性模式 Args: records: 历史记录 Returns: 季节性信息 """ weekday_costs = defaultdict(list) for record in records: try: date = datetime.strptime(record['date'], '%Y-%m-%d') weekday = date.weekday() # 0=Monday, 6=Sunday cost = safe_float(record.get('cost', 0)) weekday_costs[weekday].append(cost) except (ValueError, KeyError): continue # 计算每个工作日的平均成本 weekday_avg = {} weekday_names = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] for weekday in range(7): costs = weekday_costs[weekday] if costs: weekday_avg[weekday_names[weekday]] = statistics.mean(costs) else: weekday_avg[weekday_names[weekday]] = 0 # 找出成本最高和最低的工作日 if weekday_avg: max_day = max(weekday_avg, key=weekday_avg.get) min_day = min(weekday_avg, key=weekday_avg.get) max_cost = weekday_avg[max_day] min_cost = weekday_avg[min_day] variation = (max_cost - min_cost) / max_cost if max_cost > 0 else 0 else: max_day = min_day = 'unknown' variation = 0 return { 'weekday_averages': weekday_avg, 'highest_day': max_day, 'lowest_day': min_day, 'variation_percentage': variation * 100 }