Slurm集群作业调度系统实战指南

import jsonimport mathfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Tuple, Optional  @dataclassclass ProcurementConfig:    """绿电采购配置参数"""    annual_consumption_mwh: float  # 年耗电量(兆瓦时)    grid_emission_factor: float    # 电网碳排放因子(吨CO2/MWh)    grid_electricity_price: float  # 电网电价(元/度)    analysis_period_years: int = 20  # 分析周期(年)    discount_rate: float = 0.08     # 折现率  @dataclassclass PPAStrategy:    """PPA购电协议策略"""    ppa_price: float               # PPA电价(元/度)    annual_volume_mwh: float       # 年度采购量(MWh)    contract_years: int            # 合同年限    escalation_rate: float = 0.02  # 年电价上涨率  @dataclassclass GreenCertStrategy:    """绿电证书策略"""    cert_price_per_mwh: float      # 绿证价格(元/MWh)    annual_volume_mwh: float       # 年度采购量(MWh)  @dataclassclass DistributedSolar:    """分布式光伏配置"""    capacity_kw: float             # 装机容量(kW)    capacity_factor: float = 0.15  # 容量系数    investment_per_kw: float = 3500  # 单位投资(元/kW)    annual_om_cost_pct: float = 0.02  # 年运维成本占比    degradation_rate: float = 0.005   # 年衰减率    lifetime_years: int = 25          # 使用寿命(年)  class GreenPowerTCOCalculator:    """绿电采购TCO计算器"""     def __init__(self, config: ProcurementConfig):        self.config = config     def calc_ppa_cost(self, ppa: PPAStrategy) -> Dict:        """计算PPA模式的成本"""        annual_costs = []        total_cost = 0.0        total_volume = 0.0         for year in range(1, self.config.analysis_period_years + 1):            if year <= ppa.contract_years:                price = ppa.ppa_price * (1 + ppa.escalation_rate) ** (year - 1)                cost = price * ppa.annual_volume_mwh            else:                price = self.config.grid_electricity_price * (                    1 + 0.03                ) ** (year - 1)                cost = price * ppa.annual_volume_mwh             discount_factor = 1 / (1 + self.config.discount_rate) ** year            pv_cost = cost * discount_factor             annual_costs.append({                "year": year,                "price_per_kwh": round(price, 4),                "annual_cost": round(cost, 2),                "pv_cost": round(pv_cost, 2)            })            total_cost += pv_cost            total_volume += ppa.annual_volume_mwh         avg_price = total_cost / total_volume if total_volume > 0 else 0         return {            "strategy": "PPA",            "total_pv_cost": round(total_cost, 2),            "avg_price_per_kwh": round(avg_price, 4),            "annual_volume_mwh": ppa.annual_volume_mwh,            "annual_costs": annual_costs        }     def calc_green_cert_cost(self, cert: GreenCertStrategy) -> Dict:        """计算绿电证书模式的成本"""        annual_cert_cost = cert.cert_price_per_mwh * cert.annual_volume_mwh        total_pv_cost = 0.0         annual_costs = []        for year in range(1, self.config.analysis_period_years + 1):            price = cert.cert_price_per_mwh * (1 + 0.03) ** (year - 1)            cost = price * cert.annual_volume_mwh            discount_factor = 1 / (1 + self.config.discount_rate) ** year            pv_cost = cost * discount_factor             annual_costs.append({                "year": year,                "cert_price_per_mwh": round(price, 2),                "annual_cost": round(cost, 2),                "pv_cost": round(pv_cost, 2)            })            total_pv_cost += pv_cost         return {            "strategy": "GreenCertificate",            "total_pv_cost": round(total_pv_cost, 2),            "avg_price_per_mwh": round(                total_pv_cost / (                    cert.annual_volume_mwh * self.config.analysis_period_years                ), 4            ),            "annual_volume_mwh": cert.annual_volume_mwh,            "annual_costs": annual_costs        }     def calc_carbon_reduction(        self,        ppa_volume_mwh: float,        cert_volume_mwh: float,        solar_volume_mwh: float    ) -> Dict:        """计算碳减排量"""        total_green_mwh = ppa_volume_mwh + cert_volume_mwh + solar_volume_mwh        emission_reduced = total_green_mwh * self.config.grid_emission_factor        carbon_price = 100.0  # 元/吨CO2        carbon_value = emission_reduced * carbon_price         remaining_emission = (            self.config.annual_consumption_mwh - total_green_mwh        ) * self.config.grid_emission_factor        remaining_emission = max(0, remaining_emission)         return {            "total_green_mwh": round(total_green_mwh, 2),            "green_ratio": round(                total_green_mwh / self.config.annual_consumption_mwh, 4            ),            "emission_reduced_tons": round(emission_reduced, 2),            "carbon_value_yuan": round(carbon_value, 2),            "remaining_emission_tons": round(remaining_emission, 2),            "grid_emission_factor": self.config.grid_emission_factor        }     def calc_solar_npv(self, solar: DistributedSolar) -> Dict:        """计算分布式光伏的NPV分析"""        annual_generation = (            solar.capacity_kw * 24 * 365 * solar.capacity_factor        )        total_investment = solar.capacity_kw * solar.investment_per_kw        annual_om = total_investment * solar.annual_om_cost_pct         cash_flows = []        npv = -total_investment        cumulative = -total_investment        payback_year = None         for year in range(1, min(            self.config.analysis_period_years, solar.lifetime_years        ) + 1):            generation = annual_generation * (                1 - solar.degradation_rate            ) ** (year - 1)            savings = generation * self.config.grid_electricity_price * (                1 + 0.03            ) ** (year - 1)            net_cash = savings - annual_om * (1 + 0.02) ** (year - 1)            discount_factor = 1 / (1 + self.config.discount_rate) ** year            pv_cash = net_cash * discount_factor             npv += pv_cash            cumulative += net_cash             if payback_year is None and cumulative >= 0:                payback_year = year             cash_flows.append({                "year": year,                "generation_kwh": round(generation, 2),                "savings_yuan": round(savings, 2),                "net_cash_flow": round(net_cash, 2),                "pv_cash_flow": round(pv_cash, 2),                "cumulative": round(cumulative, 2)            })         irr = self._calc_irr(            [-total_investment] + [                cf["net_cash_flow"] for cf in cash_flows            ]        )         return {            "total_investment": round(total_investment, 2),            "annual_generation_kwh": round(annual_generation, 2),            "npv": round(npv, 2),            "irr": round(irr, 4),            "payback_years": payback_year,            "lifetime_savings": round(                sum(cf["savings_yuan"] for cf in cash_flows), 2            ),            "cash_flows": cash_flows        }     def compare_strategies(        self,        ppa: Optional[PPAStrategy],        cert: Optional[GreenCertStrategy],        solar: Optional[DistributedSolar],    ) -> Dict:        """比较不同采购策略"""        results = {"strategies": []}        total_cost = 0.0        total_green_mwh = 0.0         if ppa:            ppa_result = self.calc_ppa_cost(ppa)            results["strategies"].append(ppa_result)            total_cost += ppa_result["total_pv_cost"]            total_green_mwh += ppa.annual_volume_mwh * (                self.config.analysis_period_years            )         if cert:            cert_result = self.calc_green_cert_cost(cert)            results["strategies"].append(cert_result)            total_cost += cert_result["total_pv_cost"]            total_green_mwh += cert.annual_volume_mwh * (                self.config.analysis_period_years            )         if solar:            solar_result = self.calc_solar_npv(solar)            results["strategies"].append(solar_result)            total_cost -= solar_result["npv"]            solar_annual = solar_result["annual_generation_kwh"] / 1000            total_green_mwh += solar_annual * self.config.analysis_period_years         baseline_cost = 0.0        for year in range(1, self.config.analysis_period_years + 1):            price = self.config.grid_electricity_price * (                1 + 0.03            ) ** (year - 1)            cost = price * self.config.annual_consumption_mwh            discount_factor = 1 / (1 + self.config.discount_rate) ** year            baseline_cost += cost * discount_factor         carbon_result = self.calc_carbon_reduction(            ppa.annual_volume_mwh if ppa else 0,            cert.annual_volume_mwh if cert else 0,            (solar.capacity_kw * 24 * 365 * solar.capacity_factor / 1000             ) if solar else 0        )         results["summary"] = {            "baseline_pv_cost": round(baseline_cost, 2),            "green_strategy_pv_cost": round(total_cost, 2),            "total_savings": round(baseline_cost - total_cost, 2),            "savings_ratio": round(                (baseline_cost - total_cost) / baseline_cost, 4            ),            "carbon_reduction": carbon_result        }         return results     def _calc_irr(self, cash_flows: List[float]) -> float:        """计算内部收益率IRR"""        low, high = -0.5, 5.0        for _ in range(100):            mid = (low + high) / 2            npv = sum(                cf / (1 + mid) ** i for i, cf in enumerate(cash_flows)            )            if abs(npv) < 0.01:                return mid            if npv > 0:                low = mid            else:                high = mid        return (low + high) / 2  def main():    """主函数: 运行绿电采购TCO分析"""    config = ProcurementConfig(        annual_consumption_mwh=50000,        grid_emission_factor=0.88,        grid_electricity_price=0.72,        analysis_period_years=20,        discount_rate=0.08    )     ppa = PPAStrategy(        ppa_price=0.28,        annual_volume_mwh=35000,        contract_years=15,        escalation_rate=0.02    )     cert = GreenCertStrategy(        cert_price_per_mwh=40.0,        annual_volume_mwh=10000    )     solar = DistributedSolar(        capacity_kw=2000,        capacity_factor=0.15,        investment_per_kw=3500,        annual_om_cost_pct=0.02,        degradation_rate=0.005,        lifetime_years=25    )     calc = GreenPowerTCOCalculator(config)    results = calc.compare_strategies(ppa, cert, solar)     print("=" * 60)    print("绿电采购TCO分析报告")    print("=" * 60)    print(f"年耗电量: {config.annual_consumption_mwh} MWh")    print(f"电网电价: {config.grid_electricity_price} 元/度")    print(f"碳排放因子: {config.grid_emission_factor} 吨CO2/MWh")    print(f"分析周期: {config.analysis_period_years} 年")    print(f"折现率: {config.discount_rate * 100}%")    print()     summary = results["summary"]    print(f"基准方案(全电网)PV成本: {summary['baseline_pv_cost']:,.0f} 元")    print(f"绿电方案PV成本: {summary['green_strategy_pv_cost']:,.0f} 元")    print(f"总节省: {summary['total_savings']:,.0f} 元")    print(f"节省比例: {summary['savings_ratio'] * 100:.1f}%")    print()     cr = summary["carbon_reduction"]    print(f"绿电比例: {cr['green_ratio'] * 100:.1f}%")    print(f"年碳减排: {cr['emission_reduced_tons']:,.0f} 吨CO2")    print(f"碳减排价值: {cr['carbon_value_yuan']:,.0f} 元/年")    print(f"剩余排放: {cr['remaining_emission_tons']:,.0f} 吨CO2")    print()     for s in results["strategies"]:        if s["strategy"] == "PPA":            print(f"PPA策略 PV成本: {s['total_pv_cost']:,.0f} 元")        elif s["strategy"] == "GreenCertificate":            print(f"绿证策略 PV成本: {s['total_pv_cost']:,.0f} 元")        elif "npv" in s:            print(f"分布式光伏 NPV: {s['npv']:,.0f} 元")            print(f"  投资: {s['total_investment']:,.0f} 元")            print(f"  IRR: {s['irr'] * 100:.1f}%")            print(f"  回收期: {s['payback_years']} 年")     output = json.loads(json.dumps(results, default=str))    print()    print("详细结果已生成(JSON格式)")     return results  if __name__ == "__main__":    main()