guanlan-src/app_ontology/app_ontology_guanlan/sop/price.py

# -*- coding: utf-8 -*-
"""price.py — 柱2「智能排程」电价层: 参数化假定价形状 + 接真电价 (provenance→verdict 联动).

B1 排程机会成本 = Σ(功率×电价)。此前价形状写死在 runner(maint_schedule_gjd 280×五段式);
本模块把价形状抽成 config 驱动(value_assumptions.price) + 留真电价接口:
  - 真价源在 (spot.source 文件 / spot.hourly_real dict) → 用真价, provenance=real → 【结算】
  - 否则 benchmark × tou_factor × season 展开假定形状 → provenance=assumed → 【参考·参数假定·待业主结算价】
verdict 自动联动: ¥ 用假定价则交付降【参考·假定】; 真结算价到货 → 切【结算】 (不改一行排程逻辑)。

纯函数 (build_price_by_hour): 除 spot.source 显式文件路径外不碰 IO/时钟; 展开逻辑无副作用。
配套: templates/value_layer_assumptions.example.yaml (价形状模板) / schedule.py (消费 price_by_hour)。
"""
import json
from pathlib import Path

ASSUMED_TAG = '【参考·参数假定·待业主结算价】'
SETTLEMENT_TAG = '【结算】'


def _parse_hours(spec):
    """'00-08' | '08-17,22-24' | '23-2'(跨零点) → set(hours) ⊂ [0,24)。半开 [a,b)。"""
    out = set()
    for part in str(spec or '').split(','):
        part = part.strip()
        if not part:
            continue
        if '-' in part:
            a, b = (int(x) for x in part.split('-', 1))
            rng = range(a, b) if b > a else list(range(a, 24)) + list(range(0, b))  # 跨零点
            out.update(h % 24 for h in rng)
        else:
            out.add(int(part) % 24)
    return out


def _parse_months(spec):
    """'7-8' | '12-1'(跨年) | '3' → set(months) ⊂ [1,12]。闭区间。"""
    out = set()
    for part in str(spec or '').split(','):
        part = part.strip()
        if not part:
            continue
        if '-' in part:
            a, b = (int(x) for x in part.split('-', 1))
            out.update(range(a, b + 1) if b >= a else list(range(a, 13)) + list(range(1, b + 1)))
        else:
            out.add(int(part))
    return out


def _season_factor(season_cfg, month):
    """month → (factor, name)。命中具名季则用其 factor, 否则 normal(缺省1.0)。"""
    if not isinstance(season_cfg, dict) or month is None:
        return 1.0, 'normal'
    for name, band in season_cfg.items():
        if name == 'normal' or not isinstance(band, dict):
            continue
        if band.get('months') and month in _parse_months(band['months']):
            return float(band.get('factor', 1.0)), name
    return float((season_cfg.get('normal') or {}).get('factor', 1.0)), 'normal'


def _load_source(source):
    """真电价源 → {hour: 元/MWh}。支持: dict 直传 / .json({hour:price}) / .csv(hour,price 两列)。
    不存在/null/不支持 → None (回落假定形状)。"""
    if not source:
        return None
    if isinstance(source, dict):
        return {int(k): float(v) for k, v in source.items()}
    p = Path(str(source))
    if not p.is_file():
        return None
    try:
        if p.suffix == '.json':
            d = json.loads(p.read_text(encoding='utf-8'))
            return {int(k): float(v) for k, v in d.items()}
        if p.suffix == '.csv':
            out = {}
            for ln in p.read_text(encoding='utf-8').splitlines()[1:]:  # 跳表头
                cells = ln.split(',')
                if len(cells) >= 2:
                    out[int(float(cells[0]))] = float(cells[1])
            return out or None
    except (ValueError, OSError, json.JSONDecodeError):
        return None
    return None


def build_price_by_hour(price_cfg, month=None):
    """价 config (value_assumptions.price) → ({hour: 元/MWh}, provenance)。

    provenance: {mode, tariff_anchor, assumed(bool), verdict_tag, season, note}。
    ★接真电价: spot.hourly_real(dict) 或 spot.source(文件) 在 → 用真价, assumed=False, 【结算】。
    ★假定形状: benchmark × tou_factor(时段) × season_factor(月) 展开 24 时; 任一 assumed 标 → 整体假定。
    """
    if not isinstance(price_cfg, dict):
        return {}, {'error': 'price_cfg 非 dict', 'verdict_tag': 'INSUFFICIENT'}
    spot = price_cfg.get('spot') or {}
    real = spot.get('hourly_real') or _load_source(spot.get('source'))
    if real:
        price = {int(h) % 24: float(p) for h, p in real.items()}
        return price, {'mode': 'real', 'tariff_anchor': None, 'assumed': False,
                       'verdict_tag': SETTLEMENT_TAG, 'season': None,
                       'note': f'真电价 (源={spot.get("source") or "hourly_real"}, {len(price)}时段)'}

    bench = price_cfg.get('benchmark_cny_per_mwh')
    if bench is None:
        return {}, {'error': '缺 benchmark_cny_per_mwh 且无真价源', 'verdict_tag': 'INSUFFICIENT'}
    tou = spot.get('tou_factor') or {}
    hourly_factor = {h: 1.0 for h in range(24)}
    for band in tou.values():
        if isinstance(band, dict):
            for h in _parse_hours(band.get('hours', '')):
                hourly_factor[h] = float(band.get('factor', 1.0))
    sf, sname = _season_factor(spot.get('season_factor'), month)
    price = {h: round(float(bench) * hourly_factor[h] * sf, 2) for h in range(24)}
    assumed = bool(price_cfg.get('benchmark_assumed', True) or spot.get('tou_assumed', True)
                   or spot.get('season_assumed', True))
    return price, {
        'mode': 'assumed_shape', 'tariff_anchor': float(bench), 'assumed': assumed,
        'season': sname, 'season_factor': sf,
        'verdict_tag': ASSUMED_TAG if assumed else SETTLEMENT_TAG,
        'note': f'benchmark {bench}元/MWh × tou × season({sname}×{sf}) 展开; '
                + ('参数假定→待业主结算价' if assumed else '标定为真'),
    }


def stamp_schedule_verdict(sched_result, provenance):
    """schedule.optimize_schedule 结果 + 价 provenance → 打交付 verdict (¥ 用假定价则降级)。
    排程逻辑不变(数字照算), 只在交付层据价来源标 verdict — 真价到货切【结算】不改一行。"""
    if not isinstance(sched_result, dict):
        return sched_result
    out = dict(sched_result)
    out['price_provenance'] = provenance
    out['delivery_verdict'] = provenance.get('verdict_tag', ASSUMED_TAG)
    base = out.get('_note', '')
    out['_note'] = f'{base} | 价格: {provenance.get("note", "")} → {out["delivery_verdict"]}'
    return out