from __future__ import annotations import copy import hashlib import json import math from collections.abc import Callable, Mapping from datetime import date, timedelta from typing import Any from server.aps_domain.constraints import engine_constraint_flags from server.aps_domain.params import get_schedule_params from server.engines.base import EngineParams from server.engines.cp_engine import CP_DIAGNOSTIC_RHS_PARAMETERS, CpSatEngine from server.engines.queries import get_available_minutes from server.engines.solver_process import SolverProcessError, run_cp_rhs_diagnostic World = dict[str, Any] _METHOD = "cp-rhs-one-parameter-at-a-time-resolve.v1" _OBJECTIVE_UNIT = "weighted-tardiness-minute" _DEFAULT_INCREMENTS = { "C8_due_date_allowance": 60, "C12_team_capacity": 1, "C12_tooling_capacity": 1, } _AUTO_PARAMETERS = frozenset(_DEFAULT_INCREMENTS) _LABELS = { "C7_line_day_capacity_minutes": "产线日容量分钟", "C8_due_date_allowance": "交期容差", "C12_team_capacity": "班组并发容量", "C12_tooling_capacity": "工装并发容量", } _UNITS = { "C7_line_day_capacity_minutes": "minute", "C8_due_date_allowance": "minute", "C12_team_capacity": "capacity-unit", "C12_tooling_capacity": "capacity-unit", } _RATE_UNITS = { "C7_line_day_capacity_minutes": "currency-per-line-day-rhs-minute", "C8_due_date_allowance": "currency-per-due-allowance-minute", "C12_team_capacity": "currency-per-capacity-unit-horizon", "C12_tooling_capacity": "currency-per-capacity-unit-horizon", } def _digest(payload: Any) -> str: text = json.dumps( payload, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":"), default=str, ) return hashlib.sha256(text.encode("utf-8")).hexdigest() def _resolve_start_date(world: World, start_date: str | None) -> tuple[str, str]: source = "request" if start_date is not None else "world.businessDate" raw = start_date if start_date is not None else world.get("businessDate") if not isinstance(raw, str) or not raw.strip(): raise ValueError("startDate 必填；仅可省略于 world.businessDate 已设置时") value = raw.strip() try: parsed = date.fromisoformat(value) except ValueError as exc: raise ValueError("startDate 须为 YYYY-MM-DD") from exc if parsed.isoformat() != value: raise ValueError("startDate 须为 YYYY-MM-DD") return value, source def _solver_snapshot(meta: Mapping[str, Any]) -> dict[str, Any]: process = meta.get("solverProcess") or {} return { "status": meta.get("status"), "objective": meta.get("objective"), "bestBound": meta.get("bestBound"), "gap": meta.get("gap"), "wallTimeSec": meta.get("wallTimeSec"), "timeLimitSec": meta.get("timeLimitSec"), "assumptionConstraints": meta.get("assumptionConstraints") or [], "activeAssumptionConstraints": meta.get("activeAssumptionConstraints") or [], "constraintInstanceCounts": meta.get("constraintInstanceCounts") or {}, "rhsDiagnosticMode": meta.get("rhsDiagnosticMode"), "rhsPerturbation": meta.get("rhsPerturbation"), "rhsParameterState": meta.get("rhsParameterState"), "numSearchWorkers": meta.get("numSearchWorkers"), "randomSeed": meta.get("randomSeed"), "requestId": process.get("requestId"), "invocationId": process.get("invocationId"), "operation": process.get("operation"), "runtimeSafe": (process.get("runtimeIdentity") or {}).get("safe"), } def _rhs_topology(state: Mapping[str, Any] | None) -> dict[str, Any]: state = state or {} return { "dueDateEntryCount": state.get("dueDateEntryCount"), "lineDayResources": sorted( ( item.get("lineId"), item.get("bucketDate"), item.get("bucketStartMin"), item.get("bucketEndMin"), item.get("baseCapacityMinutes"), item.get("candidateLoadTermCount"), item.get("changeoverTermCount"), item.get("fixedFrozenLoadMinutes"), tuple(item.get("shiftIds") or []), tuple(item.get("shiftCalendarRowIds") or []), ) for item in (state.get("lineDailyCapacities") or []) if isinstance(item, Mapping) ), "teamResources": sorted( (item.get("resourceId"), item.get("intervalCount")) for item in (state.get("teamCapacities") or []) if isinstance(item, Mapping) ), "toolingResources": sorted( (item.get("resourceId"), item.get("intervalCount")) for item in (state.get("toolingCapacities") or []) if isinstance(item, Mapping) ), } def _model_identity(snapshot: Mapping[str, Any]) -> dict[str, Any]: return { "assumptionConstraints": snapshot.get("assumptionConstraints") or [], "activeAssumptionConstraints": snapshot.get("activeAssumptionConstraints") or [], "constraintInstanceCounts": snapshot.get("constraintInstanceCounts") or {}, "rhsTopology": _rhs_topology(snapshot.get("rhsParameterState")), "objectiveUnit": _OBJECTIVE_UNIT, } def _capacity_map(state: Mapping[str, Any], field: str) -> dict[int, int]: return { int(item["resourceId"]): int(item["capacity"]) for item in state.get(field) or [] if isinstance(item, Mapping) } def _line_capacity_map(state: Mapping[str, Any]) -> dict[tuple[int, str], int]: return { (int(item["lineId"]), str(item["bucketDate"])): int(item["capacityMinutes"]) for item in state.get("lineDailyCapacities") or [] if isinstance(item, Mapping) } def _parse_line_day_instance(instance_id: str) -> tuple[int, str]: parts = str(instance_id).split(":") if len(parts) != 3 or parts[0] != "line-day": raise ValueError(f"C7 rhsInstanceId 非法：{instance_id}") try: line_id = int(parts[1]) bucket_date = date.fromisoformat(parts[2]).isoformat() except (TypeError, ValueError) as exc: raise ValueError(f"C7 rhsInstanceId 非法：{instance_id}") from exc if line_id <= 0 or bucket_date != parts[2]: raise ValueError(f"C7 rhsInstanceId 非法：{instance_id}") return line_id, bucket_date def _validate_state_delta( baseline: Mapping[str, Any], perturbed: Mapping[str, Any], perturbation: Mapping[str, Any], ) -> tuple[int, int] | None: baseline_state = baseline.get("rhsParameterState") perturbed_state = perturbed.get("rhsParameterState") if not isinstance(baseline_state, Mapping) or not isinstance(perturbed_state, Mapping): return None parameter_id = str(perturbation["parameterId"]) increment = int(perturbation["increment"]) if parameter_id == "C7_line_day_capacity_minutes": baseline_caps = _line_capacity_map(baseline_state) perturbed_caps = _line_capacity_map(perturbed_state) key = (int(perturbation["lineId"]), str(perturbation["bucketDate"])) if key not in baseline_caps or set(baseline_caps) != set(perturbed_caps): return None expected = dict(baseline_caps) expected[key] += increment if perturbed_caps != expected: return None if perturbed_state.get("dueDateAllowanceMinutes") != baseline_state.get( "dueDateAllowanceMinutes" ): return None for field in ("teamCapacities", "toolingCapacities"): if _capacity_map(baseline_state, field) != _capacity_map(perturbed_state, field): return None return baseline_caps[key], perturbed_caps[key] if parameter_id == "C8_due_date_allowance": baseline_rhs = int(baseline_state.get("dueDateAllowanceMinutes") or 0) perturbed_rhs = int(perturbed_state.get("dueDateAllowanceMinutes") or 0) if perturbed_rhs != baseline_rhs + increment: return None if _capacity_map(baseline_state, "teamCapacities") != _capacity_map( perturbed_state, "teamCapacities" ) or _capacity_map(baseline_state, "toolingCapacities") != _capacity_map( perturbed_state, "toolingCapacities" ): return None if _line_capacity_map(baseline_state) != _line_capacity_map(perturbed_state): return None return baseline_rhs, perturbed_rhs field = "teamCapacities" if parameter_id == "C12_team_capacity" else "toolingCapacities" other = "toolingCapacities" if field == "teamCapacities" else "teamCapacities" baseline_caps = _capacity_map(baseline_state, field) perturbed_caps = _capacity_map(perturbed_state, field) resource_id = int(perturbation["resourceId"]) if resource_id not in baseline_caps or set(baseline_caps) != set(perturbed_caps): return None expected = dict(baseline_caps) expected[resource_id] += increment if perturbed_caps != expected: return None if _capacity_map(baseline_state, other) != _capacity_map(perturbed_state, other): return None if _line_capacity_map(baseline_state) != _line_capacity_map(perturbed_state): return None if perturbed_state.get("dueDateAllowanceMinutes") != baseline_state.get( "dueDateAllowanceMinutes" ): return None return baseline_caps[resource_id], perturbed_caps[resource_id] def _cost_fields( parameter_id: str, increment: int, cost_rates: Mapping[str, float], currency: str, *, instance_id: str | None = None, instance_cost_rates: Mapping[str, float] | None = None, ) -> dict[str, Any]: instance_rates = instance_cost_rates or {} if instance_id is not None and instance_id in instance_rates: rate = float(instance_rates[instance_id]) return { "costStatus": "configured", "costRate": rate, "costRateUnit": _RATE_UNITS[parameter_id], "costCurrency": currency, "estimatedCost": round(rate * increment, 6), "costScope": "rhs-instance", "costSource": "request.instanceCostRates", } if parameter_id not in cost_rates: return { "costStatus": "not_configured", "costRate": None, "costRateUnit": None, "costCurrency": None, "estimatedCost": None, "costScope": None, "costSource": None, } rate = float(cost_rates[parameter_id]) return { "costStatus": "configured", "costRate": rate, "costRateUnit": _RATE_UNITS[parameter_id], "costCurrency": currency, "estimatedCost": round(rate * increment, 6), "costScope": "site-default-for-parameter", "costSource": "request", } def run_cp_rhs_resolve( world: World, *, start_date: str | None = None, strategy: str = "COMPREHENSIVE", planning_horizon_days: int = 14, time_limit_seconds: float = 4.0, parameter_ids: list[str] | None = None, increments: Mapping[str, int] | None = None, cost_rates: Mapping[str, float] | None = None, instance_increments: Mapping[str, int] | None = None, instance_cost_rates: Mapping[str, float] | None = None, currency: str = "CNY", cancel_check: Callable[[], None] | None = None, ) -> dict[str, Any]: """Resolve one positive C7/C8/C12 RHS increment at a time without materialization.""" requested = sorted( _AUTO_PARAMETERS if parameter_ids is None else list(parameter_ids) ) if not requested: raise ValueError("parameterIds 不能为空") if len(requested) != len(set(requested)) or not all(isinstance(value, str) for value in requested): raise ValueError("parameterIds 须为不重复字符串数组") unknown = set(requested) - CP_DIAGNOSTIC_RHS_PARAMETERS if unknown: raise ValueError(f"不支持 CP RHS 参数：{sorted(unknown)}") if not 1 <= int(planning_horizon_days) <= 90: raise ValueError("planningHorizonDays 须在 1~90") if not 0.5 <= float(time_limit_seconds) <= 10.0: raise ValueError("timeLimitSeconds 须在 0.5~10") normalized_increments = dict(_DEFAULT_INCREMENTS) if increments is not None: if not isinstance(increments, Mapping) or set(increments) - set(requested): raise ValueError("increments 只能配置本次请求参数") normalized_increments.update(increments) for parameter_id in requested: if parameter_id == "C7_line_day_capacity_minutes": continue value = normalized_increments.get(parameter_id) limit = 10_080 if parameter_id == "C8_due_date_allowance" else 100 if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= limit: raise ValueError(f"{parameter_id} increment 须为 1~{limit} 整数") normalized_instance_increments = dict(instance_increments or {}) if not isinstance(instance_increments or {}, Mapping): raise TypeError("instanceIncrements 须为对象") if len(normalized_instance_increments) > 128: raise ValueError("instanceIncrements 单次最多 128 个实例") if normalized_instance_increments and "C7_line_day_capacity_minutes" not in requested: raise ValueError("instanceIncrements 仅可用于本次请求的 C7 参数") if "C7_line_day_capacity_minutes" in requested and not normalized_instance_increments: raise ValueError("C7 参数须显式提供 instanceIncrements") parsed_instances: dict[str, tuple[int, str]] = {} for instance_id, value in normalized_instance_increments.items(): if not isinstance(instance_id, str): raise TypeError("instanceIncrements 键须为 line-day 实例 ID") parsed_instances[instance_id] = _parse_line_day_instance(instance_id) if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 1_440: raise ValueError(f"{instance_id} increment 须为 1~1440 整数") normalized_rates = dict(cost_rates or {}) if not isinstance(cost_rates or {}, Mapping) or set(normalized_rates) - set(requested): raise ValueError("costRates 只能配置本次请求参数") for parameter_id, value in normalized_rates.items(): if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite( float(value) ) or float(value) < 0: raise ValueError(f"{parameter_id} cost rate 须为非负有限数") normalized_instance_rates = dict(instance_cost_rates or {}) if not isinstance(instance_cost_rates or {}, Mapping): raise TypeError("instanceCostRates 须为对象") if set(normalized_instance_rates) - set(normalized_instance_increments): raise ValueError("instanceCostRates 只能配置本次 instanceIncrements 实例") for instance_id, value in normalized_instance_rates.items(): if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite( float(value) ) or float(value) < 0: raise ValueError(f"{instance_id} instance cost rate 须为非负有限数") if not isinstance(currency, str) or len(currency) != 3 or not currency.isascii() or not currency.isalpha(): raise ValueError("currency 须为 3 位 ASCII 字母") currency = currency.upper() resolved_start, start_source = _resolve_start_date(world, start_date) start_day = date.fromisoformat(resolved_start) max_model_day = start_day + timedelta(days=max(int(planning_horizon_days) * 2, 7)) for instance_id, (line_id, bucket_date) in parsed_instances.items(): line = next(( row for row in world.get("lines") or [] if int(row.get("id", -1)) == line_id and row.get("status", "ACTIVE") == "ACTIVE" ), None) bucket_day = date.fromisoformat(bucket_date) if line is None: raise ValueError(f"C7 产线未启用：{line_id}") if not start_day <= bucket_day <= max_model_day: raise ValueError(f"C7 实例日期超出 CP 时间域：{instance_id}") if get_available_minutes(world, line_id, bucket_date) <= 0: raise ValueError(f"C7 实例不是有效工作日：{instance_id}") sandbox = copy.deepcopy(world) schedule_params = get_schedule_params(sandbox) params = EngineParams( orderIds=[], engineType="CP", strategyTemplate=strategy, planningHorizonDays=int(planning_horizon_days), startDate=resolved_start, constraints=engine_constraint_flags(sandbox), deliveryBufferRatio=float(schedule_params.get("deliveryBufferRatio") or 0.95), freezeWindowHours=float(schedule_params.get("freezeWindowHours") or 0.0), timeLimitSeconds=float(time_limit_seconds), name="cp-rhs-diagnostic", ) entries, _, source_count = CpSatEngine().collect_and_order(sandbox, params) objective_spec = { "kind": "min-weighted-tardiness", "unit": _OBJECTIVE_UNIT, "late": "max(0, jobEndMin-dueMin)", "weight": "customerLevelWeight*100 + rush/forecast adjustments", } world_digest = _digest(sandbox) entries_digest = _digest(entries) params_digest = _digest(params.model_dump(mode="json")) objective_spec_digest = _digest(objective_spec) input_digest = _digest({ "worldDigest": world_digest, "entriesDigest": entries_digest, "paramsDigest": params_digest, "objectiveSpecDigest": objective_spec_digest, "parameterIds": requested, "increments": { key: normalized_increments[key] for key in requested if key in normalized_increments }, "costRates": normalized_rates, "instanceIncrements": normalized_instance_increments, "instanceCostRates": normalized_instance_rates, "currency": currency, }) common = { "method": _METHOD, "modelScope": "operation-level-cp-sat-re-solve", "counterfactualMode": "one-rhs-parameter-increment-at-a-time", "objectiveUnit": _OBJECTIVE_UNIT, "isDualValue": False, "isUnitMarginalValue": False, "isFiniteDifference": True, "nonAdditiveAcrossParameters": True, "costCalibration": { "status": ( "configured" if normalized_rates or normalized_instance_rates else "not_configured" ), "configuredParameterIds": sorted(normalized_rates), "configuredInstanceIds": sorted(normalized_instance_rates), "currency": currency if normalized_rates or normalized_instance_rates else None, "objectiveAndCostAreNotNettable": True, }, "cancellationGranularity": "between-solves", "cancellationDoesNotInterruptActiveSolve": True, "activeSolveSupervisionTimeoutSeconds": max(15.0, float(time_limit_seconds) + 12.0), "cancellationCleanupTimeoutSeconds": 11.0, "cancellationLatencyUpperBoundSeconds": max( 15.0, float(time_limit_seconds) + 12.0 ) + 11.0, "track": "fixed", "strategy": strategy, "startDate": resolved_start, "startDateSource": start_source, "planningHorizonDays": int(planning_horizon_days), "timeLimitSeconds": float(time_limit_seconds), "inputDigest": input_digest, "worldDigest": world_digest, "entriesDigest": entries_digest, "paramsDigest": params_digest, "objectiveSpec": objective_spec, "objectiveSpecDigest": objective_spec_digest, "entryCount": len(entries), "sourceOrderCount": source_count, } if not entries: return { **common, "status": "no_demand", "evaluations": 0, "baseline": None, "rows": [], "summary": "当前 fixed 轨没有可进入 CP-SAT 的待排订单，未执行 RHS 重解。", } if cancel_check is not None: cancel_check() try: _, baseline_meta = run_cp_rhs_diagnostic( sandbox, entries, params, pipeline_label="CP-RHS-BASELINE", ) except SolverProcessError as exc: return { **common, "status": "baseline_solver_error", "evaluations": 1, "baseline": None, "rows": [], "summary": f"CP RHS 基线求解失败：{exc.code}", "error": exc.as_dict(), } if cancel_check is not None: cancel_check() baseline = _solver_snapshot(baseline_meta) baseline_status = str(baseline.get("status") or "") baseline_objective = baseline.get("objective") if baseline_status not in {"OPTIMAL", "FEASIBLE", "INFEASIBLE"} or ( baseline_status in {"OPTIMAL", "FEASIBLE"} and not isinstance(baseline_objective, (int, float)) ): return { **common, "status": "baseline_unavailable", "evaluations": 1, "baseline": baseline, "rows": [], "summary": f"CP RHS 基线状态 {baseline_status or 'UNKNOWN'}，无法比较业务目标。", } active_constraints = set(baseline.get("activeAssumptionConstraints") or []) cumulative_resources = baseline_meta.get("cumulative", {}).get("resources", []) line_day_resources = baseline_meta.get("lineDailyCapacity", {}).get("resources", []) variants: list[dict[str, Any]] = [] inactive_rows: list[dict[str, Any]] = [] for parameter_id in requested: if parameter_id == "C7_line_day_capacity_minutes": for instance_id in sorted(normalized_instance_increments): line_id, bucket_date = parsed_instances[instance_id] increment = int(normalized_instance_increments[instance_id]) resource = next(( item for item in line_day_resources if isinstance(item, Mapping) and int(item.get("lineId", -1)) == line_id and item.get("bucketDate") == bucket_date and int(item.get("candidateLoadTermCount") or 0) > 0 and int(item.get("baseCapacityMinutes") or 0) > 0 ), None) base_row = { "parameterId": parameter_id, "parameterName": _LABELS[parameter_id], "rhsInstanceId": instance_id, "lineId": line_id, "lineCode": resource.get("lineCode") if resource else None, "bucketDate": bucket_date, "bucketStartMin": resource.get("bucketStartMin") if resource else None, "bucketEndMin": resource.get("bucketEndMin") if resource else None, "bucketKind": "natural-calendar-day", "accountingMethod": "start-day-full-duration.v1", "capacitySource": "shift-calendar-effective-minutes", "isMaterializedShiftModel": False, "increment": increment, "parameterUnit": _UNITS[parameter_id], "isDualValue": False, "isUnitMarginalValue": False, "isFiniteDifference": True, "nonAdditiveAcrossParameters": True, **_cost_fields( parameter_id, increment, normalized_rates, currency, instance_id=instance_id, instance_cost_rates=normalized_instance_rates, ), } if resource is None or "C7_capacity" not in active_constraints: inactive_cost = {} if base_row["costStatus"] == "configured": inactive_cost = { "costStatus": "configured_inactive", "estimatedCost": None, } inactive_rows.append({ **base_row, "status": "inactive", "reason": "该 line/day 在本次 CP 模型中没有可增量的工作日产能实例。", "objectiveImprovement": None, **inactive_cost, }) continue variants.append({ **base_row, "perturbation": { "parameterId": parameter_id, "lineId": line_id, "bucketDate": bucket_date, "increment": increment, }, }) continue increment = int(normalized_increments[parameter_id]) base_row = { "parameterId": parameter_id, "parameterName": _LABELS[parameter_id], "increment": increment, "parameterUnit": _UNITS[parameter_id], "isDualValue": False, "isUnitMarginalValue": False, "isFiniteDifference": True, "nonAdditiveAcrossParameters": True, **_cost_fields(parameter_id, increment, normalized_rates, currency), } if parameter_id == "C8_due_date_allowance": variants.append({ **base_row, "rhsInstanceId": "fixed-entries:due-date-allowance", "perturbation": {"parameterId": parameter_id, "increment": increment}, }) continue constraint_id = "C12_team" if parameter_id == "C12_team_capacity" else "C12_tooling" kind = "team" if parameter_id == "C12_team_capacity" else "tooling" resources = sorted( ( resource for resource in cumulative_resources if isinstance(resource, Mapping) and resource.get("kind") == kind and int(resource.get("intervalCount") or 0) >= 2 ), key=lambda resource: int(resource["id"]), ) if constraint_id in active_constraints else [] if not resources: inactive_cost = {} if base_row["costStatus"] == "configured": inactive_cost = { "costStatus": "configured_inactive", "estimatedCost": None, } inactive_rows.append({ **base_row, "status": "inactive", "resourceKind": kind, "resourceId": None, "resourceCode": None, "reason": "该容量参数在本次 CP 模型中没有可增量的资源实例。", "objectiveImprovement": None, **inactive_cost, }) continue for resource in resources: resource_id = int(resource["id"]) variants.append({ **base_row, "resourceKind": kind, "resourceId": resource_id, "resourceCode": resource.get("code"), "rhsInstanceId": f"{kind}:{resource_id}", "perturbation": { "parameterId": parameter_id, "resourceId": resource_id, "increment": increment, }, }) rows = list(inactive_rows) evaluations = 1 monotonicity_violation = False for variant in variants: perturbation = variant.pop("perturbation") if cancel_check is not None: cancel_check() evaluations += 1 try: _, perturbed_meta = run_cp_rhs_diagnostic( sandbox, entries, params, pipeline_label=( f"CP-RHS-{perturbation['parameterId']}-" f"{perturbation.get('resourceId', perturbation.get('lineId', 'ALL'))}" ), rhs_perturbation=perturbation, ) except SolverProcessError as exc: rows.append({ **variant, "status": "solver_error", "reason": exc.code, "error": exc.as_dict(), "objectiveImprovement": None, }) continue if cancel_check is not None: cancel_check() perturbed = _solver_snapshot(perturbed_meta) rhs_pair = _validate_state_delta(baseline, perturbed, perturbation) if _model_identity(perturbed) != _model_identity(baseline) or rhs_pair is None: rows.append({ **variant, "status": "solver_error", "reason": "基线与增量重解的模型拓扑或 RHS 差值不一致", "perturbed": perturbed, "objectiveImprovement": None, }) continue baseline_rhs, perturbed_rhs = rhs_pair variant = {**variant, "baselineRhs": baseline_rhs, "perturbedRhs": perturbed_rhs} perturbed_status = str(perturbed.get("status") or "") perturbed_objective = perturbed.get("objective") if baseline_status == "INFEASIBLE": if perturbed_status in {"OPTIMAL", "FEASIBLE"}: if variant["parameterId"] == "C8_due_date_allowance": monotonicity_violation = True rows.append({ **variant, "status": "monotonicity_violation", "reason": "C8 仅改变延期目标 RHS，却改变了可行性，报告失败关闭。", "perturbed": perturbed, "objectiveImprovement": None, }) continue rows.append({ **variant, "status": "restores_feasibility", "comparisonQuality": ( "exact-optimal" if perturbed_status == "OPTIMAL" else "feasible-only" ), "perturbed": perturbed, "objectiveImprovement": None, "interpretation": "基线不可行，单个 RHS 增量后恢复可行；不计算目标差。", }) elif perturbed_status == "INFEASIBLE": rows.append({ **variant, "status": "does_not_restore_feasibility", "reason": "基线和 RHS 增量后均不可行。", "perturbed": perturbed, "objectiveImprovement": None, }) else: rows.append({ **variant, "status": "solver_error" if perturbed_status == "MODEL_INVALID" else "unavailable", "reason": f"RHS 增量后状态 {perturbed_status or 'UNKNOWN'}", "perturbed": perturbed, "objectiveImprovement": None, }) continue if perturbed_status == "INFEASIBLE": monotonicity_violation = True rows.append({ **variant, "status": "monotonicity_violation", "reason": "放宽 RHS 后反而不可行，违反单调性，整份报告失败关闭。", "perturbed": perturbed, "objectiveImprovement": None, }) continue if perturbed_status == "MODEL_INVALID": rows.append({ **variant, "status": "solver_error", "reason": "RHS 增量后 CP 模型无效。", "perturbed": perturbed, "objectiveImprovement": None, }) continue if perturbed_status not in {"OPTIMAL", "FEASIBLE"} or not isinstance( perturbed_objective, (int, float) ): rows.append({ **variant, "status": "unavailable", "reason": f"RHS 增量后求解状态 {perturbed_status or 'UNKNOWN'}", "perturbed": perturbed, "objectiveImprovement": None, }) continue exact = baseline_status == "OPTIMAL" and perturbed_status == "OPTIMAL" difference = float(baseline_objective) - float(perturbed_objective) increment = int(variant["increment"]) if exact: if difference < -1e-6: monotonicity_violation = True rows.append({ **variant, "status": "monotonicity_violation", "reason": "OPTIMAL 结果显示放宽 RHS 后目标恶化，违反单调性。", "perturbed": perturbed, "objectiveImprovement": None, }) continue rows.append({ **variant, "status": "available", "comparisonQuality": "exact-optimal", "isExactCounterfactual": True, "baselineObjective": float(baseline_objective), "perturbedObjective": float(perturbed_objective), "objectiveImprovement": round(difference, 6), "objectiveImprovementPerIncrementUnit": round(difference / increment, 6), "improvementLowerBound": round(difference, 6), "improvementUpperBound": round(difference, 6), "perturbed": perturbed, "interpretation": "单个正向 RHS 增量的 one-at-a-time 有限差分；不是对偶或货币净收益。", }) continue baseline_bound = float(baseline["bestBound"]) perturbed_bound = float(perturbed["bestBound"]) lower = max(0.0, baseline_bound - float(perturbed_objective)) upper = float(baseline_objective) - perturbed_bound if upper < -1e-6: monotonicity_violation = True rows.append({ **variant, "status": "monotonicity_violation", "reason": "FEASIBLE 界限显示放宽 RHS 可能恶化最优目标，报告失败关闭。", "perturbed": perturbed, "objectiveImprovement": None, }) continue if lower > upper + 1e-6: rows.append({ **variant, "status": "solver_error", "reason": "FEASIBLE 改善区间上下界矛盾，未输出数值。", "perturbed": perturbed, "objectiveImprovement": None, }) continue rows.append({ **variant, "status": "bounded", "comparisonQuality": "bound-interval", "isExactCounterfactual": False, "objectiveImprovement": None, "improvementLowerBound": round(lower, 6), "improvementUpperBound": round(max(0.0, upper), 6), "improvementLowerBoundPerIncrementUnit": round(lower / increment, 6), "improvementUpperBoundPerIncrementUnit": round(max(0.0, upper) / increment, 6), "perturbed": perturbed, "interpretation": "至少一侧仅 FEASIBLE，只报告由 incumbent/bound 推导的改善区间。", }) if monotonicity_violation: for row in rows: if row["status"] != "monotonicity_violation": row["suppressedStatus"] = row["status"] row["status"] = "suppressed" row["reason"] = "报告存在单调性违反，本行改善值已抑制。" for field in ( "objectiveImprovement", "objectiveImprovementPerIncrementUnit", "improvementLowerBound", "improvementUpperBound", "improvementLowerBoundPerIncrementUnit", "improvementUpperBoundPerIncrementUnit", ): row[field] = None rows.sort(key=lambda row: ( 0 if row["status"] == "available" else 1, -(row.get("objectiveImprovement") or 0.0), row["parameterId"], row.get("resourceId") or 0, row.get("lineId") or 0, row.get("bucketDate") or "", )) if cancel_check is not None: cancel_check() partial = any(row["status"] in {"solver_error", "unavailable"} for row in rows) report_status = "monotonicity_violation" if monotonicity_violation else ( "partial" if partial else "completed" ) return { **common, "status": report_status, "evaluations": evaluations, "baseline": baseline, "rows": rows, "summary": ( "RHS 报告因单调性异常失败关闭，所有改善值已抑制。" if monotonicity_violation else f"完成 {len(variants)} 个正向 RHS one-at-a-time 重解；成本仅在显式费率下估算。" ), } __all__ = ["run_cp_rhs_resolve"]