74 lines
3.0 KiB
Python
74 lines
3.0 KiB
Python
|
"""Adjustable-potential assessment skill v1 (docs/05 §2.2 可调潜力辨识).
|
|||
|
|
|
|||
|
|
For each resource: adjustable capacity = certified capacity × rolling
|
|||
|
|
fulfillment rate (delivered / planned over the evidence window), with
|
|||
|
|
confidence from the profile's own confidence, the reliability score and the
|
|||
|
|
amount of evidence. A resource with no execution history keeps its certified
|
|||
|
|
figure at the profile's confidence — the "画像时效性" risk (docs/13 §0) is
|
|||
|
|
made visible through `evidence_days` rather than hidden.
|
|||
|
|
|
|||
|
|
The de-rating exponent and evidence saturation are placeholders: OPEN-QUESTION
|
|||
|
|
B10 (可靠性/潜力辨识口径). The target accuracy (≥ 90%) is judged by the eval
|
|||
|
|
harness on real execution feedback (docs/12 L2).
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from decimal import Decimal
|
|||
|
|
|
|||
|
|
import numpy as np
|
|||
|
|
|
|||
|
|
from vpp_contracts.potential_assessment_request import PotentialAssessmentRequest
|
|||
|
|
from vpp_contracts.potential_assessment_result import PotentialAssessmentResult
|
|||
|
|
|
|||
|
|
from . import SKILL_VERSIONS
|
|||
|
|
from .numeric import INTERVALS, SCALE_MW, curve, quantize
|
|||
|
|
|
|||
|
|
SKILL = "potential-assessment"
|
|||
|
|
|
|||
|
|
EVIDENCE_WINDOW_DAYS = 14 # OPEN-QUESTION B10
|
|||
|
|
EVIDENCE_SATURATION_DAYS = 7 # OPEN-QUESTION B10: confidence saturates after this many days
|
|||
|
|
|
|||
|
|
|
|||
|
|
def assess_potential(req: PotentialAssessmentRequest) -> PotentialAssessmentResult:
|
|||
|
|
assessments = []
|
|||
|
|
total = np.zeros(INTERVALS)
|
|||
|
|
for r in req.resources:
|
|||
|
|
profile = r.profile
|
|||
|
|
certified = float(Decimal(profile.certified_adjustable_mw))
|
|||
|
|
recent = sorted(r.fulfillment, key=lambda f: f.market_date)[-EVIDENCE_WINDOW_DAYS:]
|
|||
|
|
planned = sum(Decimal(f.planned_mwh) for f in recent)
|
|||
|
|
delivered = sum(Decimal(f.delivered_mwh) for f in recent)
|
|||
|
|
if planned > 0:
|
|||
|
|
rate = min(Decimal(1), delivered / planned)
|
|||
|
|
evidence_days = len(recent)
|
|||
|
|
else:
|
|||
|
|
rate = None
|
|||
|
|
evidence_days = 0
|
|||
|
|
|
|||
|
|
derate = float(rate) if rate is not None else 1.0
|
|||
|
|
adjustable = np.full(INTERVALS, certified * derate)
|
|||
|
|
evidence_factor = min(1.0, evidence_days / EVIDENCE_SATURATION_DAYS)
|
|||
|
|
base_conf = float(Decimal(profile.confidence))
|
|||
|
|
reliability = float(Decimal(profile.reliability_score))
|
|||
|
|
confidence = base_conf * (0.5 + 0.5 * evidence_factor) * (0.5 + 0.5 * reliability)
|
|||
|
|
|
|||
|
|
total += adjustable
|
|||
|
|
assessments.append(
|
|||
|
|
{
|
|||
|
|
"resource_id": profile.resource_id,
|
|||
|
|
"adjustable_mw": curve(adjustable, req.market_date, SCALE_MW),
|
|||
|
|
"confidence": quantize(confidence, 3),
|
|||
|
|
"fulfillment_rate": quantize(rate, 4) if rate is not None else None,
|
|||
|
|
"evidence_days": evidence_days,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
return PotentialAssessmentResult.model_validate(
|
|||
|
|
{
|
|||
|
|
"market_date": req.market_date,
|
|||
|
|
"assessments": assessments,
|
|||
|
|
"total_adjustable_mw": curve(total, req.market_date, SCALE_MW),
|
|||
|
|
"skill_version": SKILL_VERSIONS[SKILL],
|
|||
|
|
}
|
|||
|
|
)
|