guanlan-src/src/sop/schedule.py

# -*- coding: utf-8 -*-
"""schedule.py — 柱2「智能排程」核心: 检修停机窗口机会成本最小化 (farm-agnostic).

B1 检修排程 (docs/B1检修排程_模块设计_v0.1.md) 的可复用内核: 把 scripts/maint_schedule_gjd.py
的郭家店写死数学抽成纯函数 + 接柱1 cases 当任务池。
一句话: 同样的必要停机, 移到低(风×价)窗口 — 不修任何东西的纯排程收益。

★物理内核: 检修窗机会成本 = Σ_窗内(预期功率 × 预期电价)。现货省份两轴合一 = 低(风×价)积窗。
★纪律: 输出 best+mean+worst + saving (禁单点最优, 风/价预期都有误差); ¥ 用假定价 → verdict【参考·参数假定】。
★柱1×柱2 桥: tasks_from_cases 把柱1 CONFIRMED/预警 case 转成停机任务池 (发现→排程闭环)。

纯函数 (optimize_schedule): 不碰文件/时钟; per-farm 数据加载在 runner。
"""


def optimize_schedule(day_costs, task_turbine_days, per_day_capacity=2):
    """核心: 给每个可选作业日的停机成本 → 排 task_turbine_days 台日 → best/mean/worst 机会成本。

    day_costs: {day_key: 元/台日} — 每个候选作业日停一台的机会成本 (低风×低价日=便宜)。
    task_turbine_days: 总停机台日 (如 偏航治理 59 台日 / D29 大修 3 台日)。
    per_day_capacity: 每日最多同时停几台 (班组约束)。
    返回 {best_cny, mean_cny, worst_cny, saving_vs_mean, saving_pct, need_days, best_days, worst_days}。
    禁单点最优: 同时给最差/均匀基线 = 决策者看区间不看点。
    """
    if not day_costs or task_turbine_days <= 0 or per_day_capacity <= 0:
        return {'verdict': 'INSUFFICIENT', 'reason': '空任务池/无候选日/容量非正'}
    import math
    days_sorted = sorted(day_costs.items(), key=lambda kv: kv[1])   # 便宜→贵
    costs = [c for _, c in days_sorted]
    need_days = math.ceil(task_turbine_days / per_day_capacity)
    if need_days > len(costs):
        return {'verdict': 'INSUFFICIENT',
                'reason': f'需 {need_days} 作业日 > 可选 {len(costs)} 日 (候选窗不足)'}

    def _sum_schedule(day_cost_list):
        """按台日铺进日: 前 need_days-1 日各满容量, 末日铺剩余。"""
        rem, total = task_turbine_days, 0.0
        for c in day_cost_list[:need_days]:
            put = min(per_day_capacity, rem)
            total += c * put
            rem -= put
        return total

    best = _sum_schedule(costs)                     # 最便宜 need_days 日
    worst = _sum_schedule(costs[::-1])              # 最贵 need_days 日
    mean_day = sum(costs) / len(costs)
    mean = mean_day * task_turbine_days             # 均匀随机基线
    saving = mean - best
    return {
        'verdict': '参考',
        'best_cny': round(best), 'mean_cny': round(mean), 'worst_cny': round(worst),
        'saving_vs_mean': round(saving),
        'saving_pct': round(saving / mean * 100, 1) if mean else None,
        'need_days': need_days, 'n_candidate_days': len(costs),
        'best_days': [k for k, _ in days_sorted[:need_days]],
        'worst_days': [k for k, _ in days_sorted[::-1][:need_days]],
        '_note': 'best/mean/worst 三值 (禁单点最优); ¥ 若用假定价 → 交付标【参考·参数假定·待业主结算价】',
    }


def day_costs_from_power_price(power_by_day_hour, price_by_hour, work_hours=range(8, 17)):
    """构造 day_costs: 每完整覆盖作业窗的日 → Σ_作业时(场中位功率 kW/1000 × 逐时价 元/MWh)。

    power_by_day_hour: {(day, hour): 场中位功率kW}; price_by_hour: {hour: 元/MWh}。
    只取作业窗被完整覆盖的日 (缺时段的日不可比)。单台停机成本 = 该日作业窗少发的电 × 价。
    """
    wh = list(work_hours)
    by_day = {}
    for (day, hour), pkw in power_by_day_hour.items():
        # NaN-safe: pkw != pkw 逮 NaN (缺测小时不计入, 该日作业窗覆盖会因此不足而剔)
        if hour in wh and price_by_hour.get(hour) is not None and pkw is not None and pkw == pkw:
            by_day.setdefault(day, {})[hour] = max(pkw, 0.0) / 1000.0 * price_by_hour[hour]
    # 只保留作业窗每小时都在的日
    return {day: round(sum(hc.values()), 2) for day, hc in by_day.items() if len(hc) == len(wh)}


def tasks_from_cases(cases, states=('CONFIRMED', 'TICKETED', 'RELAPSED'),
                     verdicts=('准定论·预警', '定论'), td_per_turbine=1.0):
    """柱1×柱2 桥: 柱1 cases → 停机任务池。默认取 CONFIRMED/工单态 (真需停机的); 可放宽到预警候选。

    states/verdicts: 纳入排程的门槛 (默认只排已确诊/已开工单; 放宽 verdicts 含预警可做"预排程")。
    返回 {'turbines': [...], 'turbine_days': N, 'basis': ...}。
    """
    picked = []
    for c in (cases or []):
        if not isinstance(c, dict):
            continue
        st_ok = c.get('state') in states
        vd_ok = (c.get('verdict') or '').strip('【】') in verdicts
        if st_ok or vd_ok:
            picked.append({'turbine': c.get('turbine'), 'system': c.get('system'),
                           'case_id': c.get('case_id'), 'state': c.get('state'), 'verdict': c.get('verdict')})
    tds = round(len(picked) * td_per_turbine, 1)
    return {'turbines': picked, 'turbine_days': tds,
            'basis': f'{len(picked)} 台 (state∈{states} ∨ verdict∈{verdicts}) × {td_per_turbine}台日/台'}