Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import secrets | |
| from copy import deepcopy | |
| from math import ceil | |
| from statistics import mean | |
| from typing import Any | |
| from .config import cfg, load_reward_config, price_band | |
| from .generator import generate_recipe, sample_seed, to_internal_task_id, to_public_task_id | |
| from .models import ( | |
| CenterObservation, | |
| HiddenRecipe, | |
| LocalOrderSnapshot, | |
| OrderTemplate, | |
| PendingTransferProposal, | |
| V2JointAction, | |
| V2Observation, | |
| V2Reward, | |
| V2StepResult, | |
| WarehouseObservation, | |
| ) | |
| from .rules import compact_public_rules | |
| class V2SupplyMindEnv: | |
| def __init__(self, default_task_id: str = "scarcity_market") -> None: | |
| self.default_task_id = default_task_id | |
| self.recipe: HiddenRecipe | None = None | |
| self.task_id = default_task_id | |
| self.internal_task_id = to_internal_task_id(default_task_id) | |
| self.public_seed = 0 | |
| self.internal_seed = 0 | |
| self.seed_source = "provided" | |
| self.round_index = 0 | |
| self.inventory: dict[str, dict[str, int]] = {} | |
| self.inventory_age: dict[str, dict[str, float]] = {} | |
| self.center_sourced_inventory: dict[str, dict[str, int]] = {} | |
| self.center_margin_per_unit: dict[str, dict[str, float]] = {} | |
| self.central_depot_inventory: dict[str, int] = {} | |
| self.central_depot_inventory_age: dict[str, float] = {} | |
| self.depot_procurements: list[tuple[int, str, int]] = [] | |
| self.depot_trucks_available = 0 | |
| self.depot_truck_returns: list[int] = [] | |
| self.drivers_available: dict[str, int] = {} | |
| self.driver_returns: dict[str, list[int]] = {} | |
| self.order_status: dict[str, str] = {} | |
| self.pending_transfers: dict[str, PendingTransferProposal] = {} | |
| self.next_proposal_id = 1 | |
| self.trust: dict[str, float] = {} | |
| self.agent_rewards: dict[str, float] = {} | |
| self.cumulative_reward = 0.0 | |
| self.audit_metrics: dict[str, Any] = {} | |
| self.last_step_reward = 0.0 | |
| self.last_components: dict[str, float] = {} | |
| self.last_events: list[str] = [] | |
| self.last_invalid_action_details: list[dict[str, Any]] = [] | |
| self.done = False | |
| self.last_episode_summary: dict[str, Any] | None = None | |
| def reset(self, task_id: str | None = None, seed: int | None = None) -> V2Observation: | |
| self.internal_task_id = to_internal_task_id(task_id or self.default_task_id) | |
| self.seed_source = "provided" if seed is not None else "curated_pool" | |
| self.public_seed = seed if seed is not None else sample_seed(self.internal_task_id, secrets.randbelow(1_000_000)) | |
| self.task_id = to_public_task_id(self.internal_task_id) | |
| self.internal_seed = self.public_seed | |
| self._load_recipe(generate_recipe(self.internal_task_id, self.internal_seed)) | |
| return self.state() | |
| def reset_internal(self, task_id: str, seed: int) -> V2Observation: | |
| self.public_seed = seed | |
| self.internal_seed = seed | |
| self.seed_source = "provided" | |
| self.internal_task_id = to_internal_task_id(task_id) | |
| self.task_id = to_public_task_id(self.internal_task_id) | |
| self._load_recipe(generate_recipe(self.internal_task_id, self.internal_seed)) | |
| return self.state() | |
| def _load_recipe(self, recipe: HiddenRecipe) -> None: | |
| self.recipe = recipe | |
| self.round_index = 0 | |
| self.inventory = deepcopy(recipe.initial_inventory) | |
| self.inventory_age = { | |
| warehouse_id: {sku: 0.0 for sku in inventory} | |
| for warehouse_id, inventory in self.inventory.items() | |
| } | |
| self.center_sourced_inventory = { | |
| warehouse_id: {sku: 0 for sku in inventory} | |
| for warehouse_id, inventory in self.inventory.items() | |
| } | |
| self.center_margin_per_unit = { | |
| warehouse_id: {sku: 0.0 for sku in inventory} | |
| for warehouse_id, inventory in self.inventory.items() | |
| } | |
| self.central_depot_inventory = dict(recipe.central_depot_inventory) | |
| self.central_depot_inventory_age = {sku: 0.0 for sku in self.central_depot_inventory} | |
| self.depot_procurements = [] | |
| self.depot_trucks_available = recipe.profile.depot_trucks | |
| self.depot_truck_returns = [] | |
| self.drivers_available = dict(recipe.initial_drivers) | |
| self.driver_returns = {spec.warehouse_id: [] for spec in recipe.warehouse_specs} | |
| self.order_status = {order.order_id: "pending" for order in recipe.orders} | |
| self.pending_transfers = {} | |
| self.next_proposal_id = 1 | |
| self.trust = {spec.warehouse_id: 0.75 for spec in recipe.warehouse_specs} | |
| self.agent_rewards = {"center": 0.0, **{spec.warehouse_id: 0.0 for spec in recipe.warehouse_specs}} | |
| self.cumulative_reward = 0.0 | |
| self.audit_metrics = self._empty_audit_metrics() | |
| self.last_step_reward = 0.0 | |
| self.last_components = {} | |
| self.last_events = ["v2 environment reset"] | |
| self.last_invalid_action_details = [] | |
| self.done = False | |
| self.last_episode_summary = None | |
| def state(self) -> V2Observation: | |
| recipe = self._require_recipe() | |
| specs_by_id = {spec.warehouse_id: spec for spec in recipe.warehouse_specs} | |
| visible = self._visible_orders() | |
| warehouses = { | |
| spec.warehouse_id: WarehouseObservation( | |
| warehouse_id=spec.warehouse_id, | |
| label=spec.label, | |
| region=spec.region, | |
| inventory=dict(self.inventory[spec.warehouse_id]), | |
| inventory_age={key: round(value, 2) for key, value in self.inventory_age[spec.warehouse_id].items()}, | |
| drivers_available=self.drivers_available[spec.warehouse_id], | |
| local_orders=[ | |
| LocalOrderSnapshot( | |
| order_id=order.order_id, | |
| sku=order.sku, | |
| units=order.units, | |
| customer_value_per_unit=order.customer_value_per_unit, | |
| created_round=order.created_round, | |
| deadline_round=order.deadline_round, | |
| status=self.order_status[order.order_id], # type: ignore[arg-type] | |
| ) | |
| for order in visible | |
| if order.warehouse_id == spec.warehouse_id and self.order_status[order.order_id] in {"pending", "accepted"} | |
| ], | |
| order_action_context=[ | |
| self._order_action_context(order, spec.warehouse_id) | |
| for order in visible | |
| if order.warehouse_id == spec.warehouse_id and self.order_status[order.order_id] in {"pending", "accepted"} | |
| ], | |
| pending_transfer_proposals=[ | |
| proposal for proposal in self.pending_transfers.values() | |
| if proposal.from_warehouse == spec.warehouse_id | |
| ], | |
| safety_stock=dict(spec.safety_stock), | |
| route_costs=dict(spec.route_costs), | |
| last_reward=round(self.agent_rewards[spec.warehouse_id], 3), | |
| ) | |
| for spec in recipe.warehouse_specs | |
| } | |
| return V2Observation( | |
| round_index=self.round_index, | |
| remaining_rounds=max(0, recipe.profile.total_rounds - self.round_index), | |
| task_id=self.task_id, | |
| scenario_info={ | |
| "task_id": self.task_id, | |
| "seed": self.public_seed, | |
| "seed_source": self.seed_source, | |
| "warehouse_count": recipe.profile.warehouse_count, | |
| "total_rounds": recipe.profile.total_rounds, | |
| "depot_replenishment_cap": recipe.profile.depot_replenishment_cap, | |
| "depot_procurement_cap": recipe.profile.depot_procurement_cap, | |
| "depot_procurement_lead_time": recipe.profile.depot_procurement_lead_time, | |
| "transfer_cap": recipe.profile.transfer_cap, | |
| "public_forecasts": [ | |
| forecast for forecast in recipe.public_forecasts | |
| if forecast["available_round"] <= self.round_index | |
| ], | |
| "public_rules": compact_public_rules(), | |
| }, | |
| center=CenterObservation( | |
| round_index=self.round_index, | |
| remaining_rounds=max(0, recipe.profile.total_rounds - self.round_index), | |
| depot_inventory=dict(self.central_depot_inventory), | |
| depot_trucks_available=self.depot_trucks_available, | |
| inbound_procurements=[ | |
| {"arrival_round": round_index, "sku": sku, "units": units} | |
| for round_index, sku, units in sorted(self.depot_procurements) | |
| ], | |
| warehouse_summaries=[ | |
| { | |
| "warehouse_id": spec.warehouse_id, | |
| "region": spec.region, | |
| "inventory": dict(self.inventory[spec.warehouse_id]), | |
| "drivers_available": self.drivers_available[spec.warehouse_id], | |
| "trust": round(self.trust[spec.warehouse_id], 3), | |
| "pending_orders": sum(1 for order in visible if order.warehouse_id == spec.warehouse_id and self.order_status[order.order_id] == "pending"), | |
| "accepted_orders": sum(1 for order in visible if order.warehouse_id == spec.warehouse_id and self.order_status[order.order_id] == "accepted"), | |
| } | |
| for spec in recipe.warehouse_specs | |
| ], | |
| market_signals=[], | |
| pending_transfer_proposals=list(self.pending_transfers.values()), | |
| price_bands=load_reward_config()["price_bands"], | |
| depot_inventory_age={key: round(value, 2) for key, value in self.central_depot_inventory_age.items()}, | |
| ), | |
| warehouses=warehouses, | |
| feedback={ | |
| "last_step_reward": round(self.last_step_reward, 3), | |
| "cumulative_reward": round(self.cumulative_reward, 3), | |
| "invalid_action_details": list(self.last_invalid_action_details), | |
| "events": list(self.last_events), | |
| "episode_summary": None if self.last_episode_summary is None else dict(self.last_episode_summary), | |
| }, | |
| ) | |
| def step(self, action: V2JointAction, grade_terminal: bool = True) -> V2StepResult: | |
| if self.done: | |
| return V2StepResult( | |
| observation=self.state(), | |
| reward=V2Reward(step_reward=0.0, cumulative_reward=self.cumulative_reward, components={}), | |
| done=True, | |
| info={"episode_summary": self.last_episode_summary or {}, "agent_rewards": dict(self.agent_rewards)}, | |
| ) | |
| recipe = self._require_recipe() | |
| self._receive_returns() | |
| components: dict[str, float] = {} | |
| agent_delta = {key: 0.0 for key in self.agent_rewards} | |
| events: list[str] = [] | |
| self._current_invalid_action_details: list[dict[str, Any]] = [] | |
| self._resolve_transfer_responses(action, components, agent_delta, events) | |
| market_signals = self._collect_warehouse_market_actions(action, components, agent_delta, events) | |
| self._resolve_order_decisions(action, components, agent_delta, events) | |
| self._resolve_center_procurements(action, components, agent_delta, events) | |
| self._resolve_center_liquidations(action, components, agent_delta, events) | |
| self._resolve_center_replenishments(action, components, agent_delta, events) | |
| self._resolve_offer_matches(action, market_signals, components, agent_delta, events) | |
| self._queue_transfer_proposals(action, components, agent_delta, events) | |
| self._fulfill_orders(action, components, agent_delta, events) | |
| self._expire_orders(components, agent_delta, events) | |
| self._apply_holding_and_spoilage(components, agent_delta, events) | |
| will_done = self.round_index + 1 >= recipe.profile.total_rounds or all( | |
| status in {"fulfilled", "rejected", "expired"} for status in self.order_status.values() | |
| ) | |
| if will_done: | |
| self._apply_terminal_order_cleanup(components, agent_delta, events) | |
| self._apply_terminal_inventory_penalty(components, agent_delta, events) | |
| self._apply_terminal_fairness_penalty(components, events) | |
| step_reward = sum(value for key, value in components.items() if key.startswith("global_")) | |
| self._update_audit_metrics(components) | |
| self.cumulative_reward += step_reward | |
| self.last_step_reward = step_reward | |
| self.last_components = {key: round(value, 3) for key, value in components.items() if abs(value) > 1e-9} | |
| self.last_events = events or ["no material event this round"] | |
| self.last_invalid_action_details = list(getattr(self, "_current_invalid_action_details", [])) | |
| for key, value in agent_delta.items(): | |
| self.agent_rewards[key] += value | |
| self.round_index += 1 | |
| self.done = will_done | |
| info: dict[str, Any] = { | |
| "agent_rewards": {key: round(value, 3) for key, value in self.agent_rewards.items()}, | |
| "market_signals": market_signals, | |
| } | |
| if self.done and grade_terminal: | |
| from .grading import grade_episode | |
| avg_warehouse_reward = mean(value for key, value in self.agent_rewards.items() if key != "center") | |
| task_result = grade_episode( | |
| self.internal_task_id, | |
| self.internal_seed, | |
| self.cumulative_reward, | |
| self.agent_rewards["center"], | |
| avg_warehouse_reward, | |
| ) | |
| self.last_episode_summary = { | |
| "raw_reward": round(task_result.raw_reward, 3), | |
| "baseline_reward": round(task_result.baseline_reward, 3), | |
| "target_reward": round(task_result.target_reward, 3), | |
| "graded_score": round(task_result.score, 4), | |
| "center_reward": round(self.agent_rewards["center"], 3), | |
| "average_warehouse_reward": round(avg_warehouse_reward, 3), | |
| "center_role_score": round(task_result.center_role_score, 4), | |
| "warehouse_role_score": round(task_result.warehouse_role_score, 4), | |
| "baseline_center_reward": round(task_result.baseline_center_reward, 3), | |
| "target_center_reward": round(task_result.target_center_reward, 3), | |
| "baseline_warehouse_reward": round(task_result.baseline_warehouse_reward, 3), | |
| "target_warehouse_reward": round(task_result.target_warehouse_reward, 3), | |
| "audit_metrics": dict(self.audit_metrics), | |
| } | |
| info["episode_summary"] = dict(self.last_episode_summary) | |
| return V2StepResult( | |
| observation=self.state(), | |
| reward=V2Reward(step_reward=round(step_reward, 3), cumulative_reward=round(self.cumulative_reward, 3), components={}), | |
| done=self.done, | |
| info=info, | |
| ) | |
| def _resolve_transfer_responses(self, action: V2JointAction, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| recipe = self._require_recipe() | |
| specs_by_id = {spec.warehouse_id: spec for spec in recipe.warehouse_specs} | |
| for warehouse_id, warehouse_action in action.warehouse_actions.items(): | |
| for response in warehouse_action.transfer_responses: | |
| proposal = self.pending_transfers.pop(response.proposal_id, None) | |
| if proposal is None or proposal.from_warehouse != warehouse_id: | |
| self._invalid(components, "transfer_response", "proposal_missing_or_wrong_source", -3.0, {"warehouse_id": warehouse_id, "proposal_id": response.proposal_id}) | |
| continue | |
| if response.decision == "reject": | |
| _add(components, "global_rejected_transfer_penalty", -0.5) | |
| _add(components, "audit_rejected_transfer_count", 1.0) | |
| agent_delta["center"] -= cfg("center_rewards", "rejected_transfer_penalty") | |
| self.trust[warehouse_id] = max(0.0, self.trust[warehouse_id] - 0.04) | |
| events.append(f"{warehouse_id} rejected transfer {proposal.proposal_id}") | |
| continue | |
| if self.inventory[proposal.from_warehouse].get(proposal.sku, 0) < proposal.units: | |
| self._invalid(components, "transfer_response", "source_insufficient_stock", -4.0, {"warehouse_id": warehouse_id, "proposal_id": response.proposal_id, "sku": proposal.sku}) | |
| continue | |
| source = specs_by_id[proposal.from_warehouse] | |
| target = specs_by_id[proposal.to_warehouse] | |
| transfer_cost = source.route_costs[target.region] * proposal.units * 0.8 | |
| moved_age = self.inventory_age[proposal.from_warehouse].get(proposal.sku, 0.0) | |
| self._remove_warehouse_stock(proposal.from_warehouse, proposal.sku, proposal.units) | |
| self._add_warehouse_stock(proposal.to_warehouse, proposal.sku, proposal.units, moved_age) | |
| broker_fee = _broker_fee(proposal.units, proposal.compensation) | |
| _add(components, "global_transfer_cost", -transfer_cost) | |
| _add(components, "center_transfer_broker_fee", broker_fee) | |
| _add(components, "audit_transfer_count", 1.0) | |
| _add(components, "audit_successful_transfer_units", float(proposal.units)) | |
| agent_delta["center"] += broker_fee | |
| agent_delta[proposal.from_warehouse] += proposal.compensation - proposal.units * cfg("warehouse_rewards", "transfer_sacrifice_cost_per_unit") | |
| agent_delta[proposal.to_warehouse] -= proposal.compensation | |
| events.append(f"{proposal.from_warehouse} accepted transfer {proposal.units} {proposal.sku} to {proposal.to_warehouse}") | |
| def _collect_warehouse_market_actions(self, action: V2JointAction, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> list[dict[str, Any]]: | |
| signals: list[dict[str, Any]] = [] | |
| for warehouse_id, warehouse_action in action.warehouse_actions.items(): | |
| if warehouse_id not in self.inventory: | |
| self._invalid(components, "warehouse_action", "unknown_warehouse", -3.0, {"warehouse_id": warehouse_id}) | |
| continue | |
| for offer in warehouse_action.inventory_offers: | |
| units = min(max(0, offer.units), self.inventory[warehouse_id].get(offer.sku, 0)) | |
| if units > 0: | |
| signals.append({"signal_id": f"{warehouse_id}:offer:{offer.sku}", "warehouse_id": warehouse_id, "type": "offer", "sku": offer.sku, "units": units, "price": offer.ask_price}) | |
| for request in warehouse_action.inventory_requests: | |
| if request.units > 0: | |
| signals.append({"signal_id": f"{warehouse_id}:request:{request.sku}", "warehouse_id": warehouse_id, "type": "request", "sku": request.sku, "units": request.units, "price": request.max_price}) | |
| return signals | |
| def _resolve_order_decisions(self, action: V2JointAction, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| orders = {order.order_id: order for order in self._visible_orders()} | |
| for warehouse_id, warehouse_action in action.warehouse_actions.items(): | |
| for decision in warehouse_action.order_decisions: | |
| order = orders.get(decision.order_id) | |
| if order is None or order.warehouse_id != warehouse_id or self.order_status.get(order.order_id) != "pending": | |
| self._invalid(components, "order_decision", "order_missing_wrong_warehouse_or_not_pending", -2.0, {"warehouse_id": warehouse_id, "order_id": decision.order_id}) | |
| continue | |
| if decision.decision == "reject": | |
| self.order_status[order.order_id] = "rejected" | |
| penalty = order.units * order.customer_value_per_unit * cfg("order_rewards", "reject_penalty_multiplier") | |
| _add(components, "global_reject_penalty", -penalty) | |
| agent_delta[warehouse_id] -= penalty | |
| events.append(f"{warehouse_id} rejected {order.order_id}") | |
| else: | |
| self.order_status[order.order_id] = "accepted" | |
| events.append(f"{warehouse_id} accepted {order.order_id}") | |
| def _resolve_center_procurements(self, action: V2JointAction, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| recipe = self._require_recipe() | |
| used_units = 0 | |
| for procurement in action.central_action.central_procurements: | |
| band = price_band(procurement.sku) | |
| if procurement.units <= 0 or procurement.max_unit_cost < band["procurement_cost"] or used_units + procurement.units > recipe.profile.depot_procurement_cap: | |
| reason = "invalid_units" if procurement.units <= 0 else "max_unit_cost_below_procurement" if procurement.max_unit_cost < band["procurement_cost"] else "procurement_cap_exceeded" | |
| self._invalid(components, "central_procurement", reason, -4.0, {"sku": procurement.sku, "units": procurement.units}) | |
| agent_delta["center"] -= 2.0 | |
| continue | |
| cost = procurement.units * band["procurement_cost"] | |
| arrival = self.round_index + recipe.profile.depot_procurement_lead_time | |
| self.depot_procurements.append((arrival, procurement.sku, procurement.units)) | |
| used_units += procurement.units | |
| _add(components, "global_procurement_cost", -cost) | |
| agent_delta["center"] -= cost | |
| events.append(f"center procured {procurement.units} {procurement.sku}, arrival round {arrival}") | |
| def _resolve_center_liquidations(self, action: V2JointAction, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| used_units = 0 | |
| cap = int(cfg("center_rewards", "liquidation_cap_per_round")) | |
| recovery_multiplier = cfg("center_rewards", "liquidation_recovery_multiplier") | |
| for liquidation in action.central_action.central_liquidations: | |
| if liquidation.units <= 0 or used_units + liquidation.units > cap: | |
| reason = "invalid_units" if liquidation.units <= 0 else "liquidation_cap_exceeded" | |
| self._invalid(components, "central_liquidation", reason, -3.0, {"sku": liquidation.sku, "units": liquidation.units}) | |
| agent_delta["center"] -= 1.0 | |
| continue | |
| available = self.central_depot_inventory.get(liquidation.sku, 0) | |
| if liquidation.units > available: | |
| self._invalid(components, "central_liquidation", "insufficient_depot_stock", -3.0, {"sku": liquidation.sku, "units": liquidation.units, "available": available}) | |
| agent_delta["center"] -= 1.0 | |
| continue | |
| recovery = liquidation.units * price_band(liquidation.sku)["procurement_cost"] * recovery_multiplier | |
| self._remove_depot_stock(liquidation.sku, liquidation.units) | |
| used_units += liquidation.units | |
| _add(components, "global_liquidation_recovery", recovery) | |
| _add(components, "center_liquidation_recovery", recovery) | |
| agent_delta["center"] += recovery | |
| events.append(f"center liquidated {liquidation.units} {liquidation.sku} at markdown recovery") | |
| def _resolve_center_replenishments(self, action: V2JointAction, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| recipe = self._require_recipe() | |
| specs_by_id = {spec.warehouse_id: spec for spec in recipe.warehouse_specs} | |
| used_units = 0 | |
| for shipment in action.central_action.central_replenishments: | |
| band = price_band(shipment.sku) | |
| if ( | |
| shipment.to_warehouse not in self.inventory | |
| or shipment.units <= 0 | |
| or shipment.unit_price > band["max_wholesale_price"] | |
| or shipment.units > self.central_depot_inventory.get(shipment.sku, 0) | |
| or used_units + shipment.units > recipe.profile.depot_replenishment_cap | |
| or self.depot_trucks_available <= 0 | |
| ): | |
| reason = self._shipment_invalid_reason(shipment, used_units, recipe.profile.depot_replenishment_cap, band) | |
| self._invalid(components, "central_replenishment", reason, -4.0, {"to_warehouse": shipment.to_warehouse, "sku": shipment.sku, "units": shipment.units}) | |
| agent_delta["center"] -= 2.0 | |
| continue | |
| spec = specs_by_id[shipment.to_warehouse] | |
| shipment_cost = (1.2 + spec.route_costs[spec.region]) * shipment.units | |
| moved_age = self.central_depot_inventory_age.get(shipment.sku, 0.0) | |
| self._remove_depot_stock(shipment.sku, shipment.units) | |
| self._add_warehouse_stock(shipment.to_warehouse, shipment.sku, shipment.units, moved_age) | |
| self.center_sourced_inventory[shipment.to_warehouse][shipment.sku] += shipment.units | |
| self.center_margin_per_unit[shipment.to_warehouse][shipment.sku] = shipment.unit_price - band["procurement_cost"] | |
| self.depot_trucks_available -= 1 | |
| self.depot_truck_returns.append(self.round_index + spec.route_times[spec.region]) | |
| used_units += shipment.units | |
| wholesale_cost = shipment.unit_price * shipment.units | |
| _add(components, "global_center_shipment_cost", -shipment_cost) | |
| agent_delta["center"] -= shipment_cost | |
| agent_delta[shipment.to_warehouse] -= wholesale_cost | |
| events.append(f"center sold {shipment.units} {shipment.sku} to {shipment.to_warehouse} at {shipment.unit_price:.1f}") | |
| def _resolve_offer_matches(self, action: V2JointAction, market_signals: list[dict[str, Any]], components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| signals = {signal["signal_id"]: signal for signal in market_signals} | |
| moved = 0 | |
| for match in action.central_action.offer_matches: | |
| offer = signals.get(match.offer_signal_id) | |
| request = signals.get(match.request_signal_id) | |
| if not offer or not request or offer["type"] != "offer" or request["type"] != "request" or offer["sku"] != request["sku"]: | |
| self._invalid(components, "offer_match", "missing_or_incompatible_offer_request", -3.0, {"offer_signal_id": match.offer_signal_id, "request_signal_id": match.request_signal_id}) | |
| continue | |
| units = min(match.units, offer["units"], request["units"]) | |
| if units <= 0 or self.inventory[offer["warehouse_id"]].get(offer["sku"], 0) < units: | |
| self._invalid(components, "offer_match", "invalid_units_or_offer_stock", -3.0, {"offer_signal_id": match.offer_signal_id, "request_signal_id": match.request_signal_id, "units": match.units}) | |
| continue | |
| moved_age = self.inventory_age[offer["warehouse_id"]].get(offer["sku"], 0.0) | |
| self._remove_warehouse_stock(offer["warehouse_id"], offer["sku"], units) | |
| self._add_warehouse_stock(request["warehouse_id"], request["sku"], units, moved_age) | |
| compensation = max(match.compensation, offer["price"] * units) | |
| broker_fee = _broker_fee(units, compensation) | |
| _add(components, "global_transfer_cost", -0.5 * units) | |
| _add(components, "center_transfer_broker_fee", broker_fee) | |
| _add(components, "audit_transfer_count", 1.0) | |
| _add(components, "audit_successful_transfer_units", float(units)) | |
| agent_delta["center"] += broker_fee | |
| agent_delta[offer["warehouse_id"]] += compensation | |
| agent_delta[request["warehouse_id"]] -= compensation | |
| moved += units | |
| events.append(f"center matched {units} {offer['sku']} from {offer['warehouse_id']} to {request['warehouse_id']}") | |
| if moved > self._require_recipe().profile.transfer_cap: | |
| self._invalid(components, "offer_match", "transfer_cap_exceeded", -4.0, {"moved_units": moved}) | |
| def _queue_transfer_proposals(self, action: V2JointAction, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| for proposal in action.central_action.inventory_transfer_proposals: | |
| if proposal.from_warehouse not in self.inventory or proposal.to_warehouse not in self.inventory or proposal.units <= 0: | |
| self._invalid(components, "transfer_proposal", "unknown_warehouse_or_invalid_units", -3.0, {"from_warehouse": proposal.from_warehouse, "to_warehouse": proposal.to_warehouse, "units": proposal.units}) | |
| continue | |
| proposal_id = f"p{self.next_proposal_id}" | |
| self.next_proposal_id += 1 | |
| self.pending_transfers[proposal_id] = PendingTransferProposal(proposal_id=proposal_id, **proposal.model_dump()) | |
| events.append(f"center proposed transfer {proposal_id}: {proposal.from_warehouse}->{proposal.to_warehouse} {proposal.units} {proposal.sku}") | |
| def _fulfill_orders(self, action: V2JointAction, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| recipe = self._require_recipe() | |
| specs_by_id = {spec.warehouse_id: spec for spec in recipe.warehouse_specs} | |
| priority_by_wh: dict[str, dict[str, int]] = { | |
| warehouse_id: {rule.sku: rule.priority for rule in warehouse_action.local_priority} | |
| for warehouse_id, warehouse_action in action.warehouse_actions.items() | |
| } | |
| accepted = [ | |
| order for order in self._visible_orders() | |
| if self.order_status[order.order_id] == "accepted" | |
| ] | |
| for order in sorted(accepted, key=lambda item: (-priority_by_wh.get(item.warehouse_id, {}).get(item.sku, 1), item.deadline_round, -item.units * item.customer_value_per_unit)): | |
| spec = specs_by_id[order.warehouse_id] | |
| if self.inventory[order.warehouse_id].get(order.sku, 0) < order.units or self.drivers_available[order.warehouse_id] <= 0: | |
| continue | |
| delivery_cost = spec.route_costs[spec.region] * order.units | |
| self._remove_warehouse_stock(order.warehouse_id, order.sku, order.units) | |
| self.drivers_available[order.warehouse_id] -= 1 | |
| self.driver_returns[order.warehouse_id].append(self.round_index + spec.route_times[spec.region]) | |
| self.order_status[order.order_id] = "fulfilled" | |
| value = order.units * order.customer_value_per_unit | |
| _add(components, "global_fulfilled_customer_value", value) | |
| _add(components, "global_warehouse_delivery_cost", -delivery_cost) | |
| fulfillment_profit = max(0.0, value - delivery_cost) | |
| center_service_bonus = fulfillment_profit * cfg("center_rewards", "network_service_profit_share") | |
| agent_delta[order.warehouse_id] += value - delivery_cost - center_service_bonus | |
| agent_delta["center"] += center_service_bonus | |
| _add(components, "center_network_service_bonus", center_service_bonus) | |
| center_units = min(order.units, self.center_sourced_inventory[order.warehouse_id].get(order.sku, 0)) | |
| if center_units: | |
| margin = center_units * self.center_margin_per_unit[order.warehouse_id].get(order.sku, 0.0) | |
| self.center_sourced_inventory[order.warehouse_id][order.sku] -= center_units | |
| agent_delta["center"] += margin | |
| _add(components, "center_realized_wholesale_margin", margin) | |
| events.append(f"{order.warehouse_id} fulfilled {order.order_id}") | |
| def _expire_orders(self, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| for order in self._visible_orders(): | |
| if order.deadline_round >= self.round_index: | |
| continue | |
| status = self.order_status[order.order_id] | |
| if status not in {"pending", "accepted"}: | |
| continue | |
| multiplier = cfg("order_rewards", "accepted_missed_penalty_multiplier") if status == "accepted" else cfg("order_rewards", "silent_expiry_penalty_multiplier") | |
| penalty = order.units * order.customer_value_per_unit * multiplier | |
| self.order_status[order.order_id] = "expired" | |
| _add(components, "global_stockout_penalty", -penalty) | |
| agent_delta[order.warehouse_id] -= penalty * cfg("warehouse_rewards", "local_stockout_share") | |
| agent_delta["center"] -= penalty * cfg("center_rewards", "network_stockout_share") | |
| events.append(f"{order.order_id} expired after {status}") | |
| def _apply_terminal_order_cleanup(self, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| accepted_count = 0 | |
| pending_count = 0 | |
| total_penalty = 0.0 | |
| for order in self._visible_orders(): | |
| status = self.order_status[order.order_id] | |
| if status not in {"pending", "accepted"}: | |
| continue | |
| multiplier = ( | |
| cfg("order_rewards", "accepted_missed_penalty_multiplier") | |
| if status == "accepted" | |
| else cfg("global_rewards", "terminal_pending_expiry_multiplier") | |
| ) | |
| penalty = order.units * order.customer_value_per_unit * multiplier | |
| self.order_status[order.order_id] = "expired" | |
| _add(components, "global_terminal_order_penalty", -penalty) | |
| agent_delta[order.warehouse_id] -= penalty * cfg("warehouse_rewards", "local_stockout_share") | |
| agent_delta["center"] -= penalty * cfg("center_rewards", "network_stockout_share") | |
| total_penalty += penalty | |
| if status == "accepted": | |
| accepted_count += 1 | |
| else: | |
| pending_count += 1 | |
| if total_penalty: | |
| events.append(f"terminal closed {accepted_count} accepted and {pending_count} pending visible orders") | |
| def _apply_holding_and_spoilage(self, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| center_holding = sum(self.central_depot_inventory.values()) * cfg("center_rewards", "depot_holding_cost_per_unit") | |
| _add(components, "global_holding_cost", -center_holding) | |
| agent_delta["center"] -= center_holding | |
| shelf_life = int(cfg("global_rewards", "fresh_milk_shelf_life_rounds")) | |
| expiry_fraction = cfg("global_rewards", "fresh_milk_expiry_fraction_per_round") | |
| disposal_multiplier = cfg("global_rewards", "fresh_milk_disposal_cost_multiplier") | |
| self._age_fresh_milk() | |
| for sku, units in list(self.central_depot_inventory.items()): | |
| if sku == "fresh_milk" and units > 0 and self.central_depot_inventory_age.get(sku, 0.0) > shelf_life: | |
| spoiled = min(units, max(1, ceil(units * expiry_fraction))) | |
| penalty = spoiled * price_band(sku)["procurement_cost"] * disposal_multiplier | |
| self._remove_depot_stock(sku, spoiled) | |
| _add(components, "global_spoilage_cost", -penalty) | |
| agent_delta["center"] -= penalty * cfg("center_rewards", "depot_spoilage_share") | |
| events.append(f"depot spoiled {spoiled} {sku}") | |
| for warehouse_id, inventory in self.inventory.items(): | |
| holding = sum(inventory.values()) * cfg("warehouse_rewards", "holding_cost_per_unit") | |
| _add(components, "global_holding_cost", -holding) | |
| agent_delta[warehouse_id] -= holding | |
| fresh = inventory.get("fresh_milk", 0) | |
| if fresh > 0 and self.inventory_age[warehouse_id].get("fresh_milk", 0.0) > shelf_life: | |
| spoiled = min(fresh, max(1, ceil(fresh * expiry_fraction))) | |
| penalty = spoiled * price_band("fresh_milk")["procurement_cost"] * disposal_multiplier | |
| self._remove_warehouse_stock(warehouse_id, "fresh_milk", spoiled) | |
| _add(components, "global_spoilage_cost", -penalty) | |
| agent_delta[warehouse_id] -= penalty * cfg("warehouse_rewards", "warehouse_spoilage_share") | |
| agent_delta["center"] -= penalty * cfg("center_rewards", "warehouse_spoilage_share") | |
| events.append(f"{warehouse_id} spoiled {spoiled} fresh_milk") | |
| def _apply_terminal_inventory_penalty(self, components: dict[str, float], agent_delta: dict[str, float], events: list[str]) -> None: | |
| multiplier = cfg("global_rewards", "terminal_inventory_penalty_multiplier") | |
| depot_value = sum(units * price_band(sku)["procurement_cost"] for sku, units in self.central_depot_inventory.items()) | |
| depot_penalty = depot_value * multiplier | |
| if depot_penalty: | |
| _add(components, "global_terminal_inventory_penalty", -depot_penalty) | |
| agent_delta["center"] -= depot_penalty * cfg("center_rewards", "terminal_depot_inventory_share") | |
| warehouse_penalty_total = 0.0 | |
| leftover_units = 0 | |
| for warehouse_id, inventory in self.inventory.items(): | |
| value = sum(units * price_band(sku)["procurement_cost"] for sku, units in inventory.items()) | |
| penalty = value * multiplier | |
| if not penalty: | |
| continue | |
| warehouse_penalty_total += penalty | |
| leftover_units += sum(inventory.values()) | |
| _add(components, "global_terminal_inventory_penalty", -penalty) | |
| agent_delta[warehouse_id] -= penalty * cfg("warehouse_rewards", "terminal_inventory_share") | |
| agent_delta["center"] -= penalty * cfg("center_rewards", "terminal_warehouse_inventory_share") | |
| if depot_penalty or warehouse_penalty_total: | |
| depot_units = sum(self.central_depot_inventory.values()) | |
| events.append(f"terminal leftover inventory penalty on {depot_units} depot units and {leftover_units} warehouse units") | |
| def _apply_terminal_fairness_penalty(self, components: dict[str, float], events: list[str]) -> None: | |
| service_rates = self._service_rates() | |
| if not service_rates: | |
| return | |
| avg_rate = mean(service_rates.values()) | |
| mad = mean(abs(rate - avg_rate) for rate in service_rates.values()) | |
| penalty = min(cfg("global_rewards", "terminal_fairness_weight") * mad, cfg("global_rewards", "terminal_fairness_cap")) | |
| if penalty: | |
| _add(components, "global_fairness_penalty", -penalty) | |
| _add(components, "audit_fairness_penalty", penalty) | |
| events.append(f"terminal fairness penalty {penalty:.2f} from service-rate spread") | |
| self.audit_metrics["per_warehouse_service_rate"] = {key: round(value, 4) for key, value in service_rates.items()} | |
| def _service_rates(self) -> dict[str, float]: | |
| recipe = self._require_recipe() | |
| fulfilled: dict[str, float] = {spec.warehouse_id: 0.0 for spec in recipe.warehouse_specs} | |
| total: dict[str, float] = {spec.warehouse_id: 0.0 for spec in recipe.warehouse_specs} | |
| for order in recipe.orders: | |
| value = order.units * order.customer_value_per_unit | |
| total[order.warehouse_id] += value | |
| if self.order_status.get(order.order_id) == "fulfilled": | |
| fulfilled[order.warehouse_id] += value | |
| return { | |
| warehouse_id: (fulfilled[warehouse_id] / value if value else 1.0) | |
| for warehouse_id, value in total.items() | |
| } | |
| def _update_audit_metrics(self, components: dict[str, float]) -> None: | |
| mapping = { | |
| "audit_transfer_count": "transfer_count", | |
| "audit_successful_transfer_units": "successful_transfer_units", | |
| "global_transfer_cost": "transfer_cost", | |
| "center_transfer_broker_fee": "center_broker_fee", | |
| "global_liquidation_recovery": "liquidation_recovery", | |
| "global_fulfilled_customer_value": "fulfilled_customer_value", | |
| "global_stockout_penalty": "stockout_penalty", | |
| "global_terminal_order_penalty": "terminal_order_penalty", | |
| "global_holding_cost": "holding_cost", | |
| "global_spoilage_cost": "spoilage_cost", | |
| "global_terminal_inventory_penalty": "terminal_leftover_penalty", | |
| "audit_fairness_penalty": "fairness_penalty", | |
| "global_invalid_action_penalty": "invalid_action_penalty", | |
| "global_rejected_transfer_penalty": "rejected_transfer_penalty", | |
| "audit_rejected_transfer_count": "rejected_transfer_count", | |
| } | |
| for component, metric in mapping.items(): | |
| value = components.get(component, 0.0) | |
| if value: | |
| self.audit_metrics[metric] = round(float(self.audit_metrics.get(metric, 0.0)) + float(value), 3) | |
| def _empty_audit_metrics() -> dict[str, Any]: | |
| return { | |
| "transfer_count": 0.0, | |
| "successful_transfer_units": 0.0, | |
| "rejected_transfer_count": 0, | |
| "transfer_cost": 0.0, | |
| "center_broker_fee": 0.0, | |
| "liquidation_recovery": 0.0, | |
| "fulfilled_customer_value": 0.0, | |
| "stockout_penalty": 0.0, | |
| "terminal_order_penalty": 0.0, | |
| "holding_cost": 0.0, | |
| "spoilage_cost": 0.0, | |
| "terminal_leftover_penalty": 0.0, | |
| "fairness_penalty": 0.0, | |
| "invalid_action_penalty": 0.0, | |
| "per_warehouse_service_rate": {}, | |
| } | |
| def _order_action_context(self, order: OrderTemplate, warehouse_id: str) -> dict[str, Any]: | |
| stock = self.inventory[warehouse_id].get(order.sku, 0) | |
| value = order.units * order.customer_value_per_unit | |
| return { | |
| "order_id": order.order_id, | |
| "status": self.order_status[order.order_id], | |
| "sku": order.sku, | |
| "units": order.units, | |
| "order_value": value, | |
| "stock_available": stock, | |
| "stock_gap": max(0, order.units - stock), | |
| "driver_available": self.drivers_available[warehouse_id] > 0, | |
| "rounds_until_deadline": order.deadline_round - self.round_index, | |
| "can_fulfill_now": stock >= order.units and self.drivers_available[warehouse_id] > 0, | |
| "reject_penalty": round(value * cfg("order_rewards", "reject_penalty_multiplier"), 3), | |
| "accepted_missed_penalty": round(value * cfg("order_rewards", "accepted_missed_penalty_multiplier"), 3), | |
| "silent_expiry_penalty": round(value * cfg("order_rewards", "silent_expiry_penalty_multiplier"), 3), | |
| } | |
| def _invalid(self, components: dict[str, float], action_type: str, reason: str, penalty: float, context: dict[str, Any]) -> None: | |
| _add(components, "global_invalid_action_penalty", penalty) | |
| if not hasattr(self, "_current_invalid_action_details"): | |
| self._current_invalid_action_details = [] | |
| self._current_invalid_action_details.append( | |
| { | |
| "action_type": action_type, | |
| "reason": reason, | |
| "penalty": penalty, | |
| "context": context, | |
| } | |
| ) | |
| def _add_depot_stock(self, sku: str, units: int, age: float = 0.0) -> None: | |
| current_units = self.central_depot_inventory.get(sku, 0) | |
| if sku == "fresh_milk" and units > 0: | |
| current_age = self.central_depot_inventory_age.get(sku, 0.0) | |
| total_units = current_units + units | |
| self.central_depot_inventory_age[sku] = ((current_units * current_age) + (units * age)) / total_units | |
| self.central_depot_inventory[sku] = current_units + units | |
| def _remove_depot_stock(self, sku: str, units: int) -> None: | |
| self.central_depot_inventory[sku] = max(0, self.central_depot_inventory.get(sku, 0) - units) | |
| if self.central_depot_inventory[sku] == 0: | |
| self.central_depot_inventory_age[sku] = 0.0 | |
| def _add_warehouse_stock(self, warehouse_id: str, sku: str, units: int, age: float = 0.0) -> None: | |
| current_units = self.inventory[warehouse_id].get(sku, 0) | |
| if sku == "fresh_milk" and units > 0: | |
| current_age = self.inventory_age[warehouse_id].get(sku, 0.0) | |
| total_units = current_units + units | |
| self.inventory_age[warehouse_id][sku] = ((current_units * current_age) + (units * age)) / total_units | |
| self.inventory[warehouse_id][sku] = current_units + units | |
| def _remove_warehouse_stock(self, warehouse_id: str, sku: str, units: int) -> None: | |
| self.inventory[warehouse_id][sku] = max(0, self.inventory[warehouse_id].get(sku, 0) - units) | |
| if self.inventory[warehouse_id][sku] == 0: | |
| self.inventory_age[warehouse_id][sku] = 0.0 | |
| def _age_fresh_milk(self) -> None: | |
| if self.central_depot_inventory.get("fresh_milk", 0) > 0: | |
| self.central_depot_inventory_age["fresh_milk"] = self.central_depot_inventory_age.get("fresh_milk", 0.0) + 1.0 | |
| for warehouse_id, inventory in self.inventory.items(): | |
| if inventory.get("fresh_milk", 0) > 0: | |
| self.inventory_age[warehouse_id]["fresh_milk"] = self.inventory_age[warehouse_id].get("fresh_milk", 0.0) + 1.0 | |
| def _shipment_invalid_reason(self, shipment: Any, used_units: int, cap: int, band: dict[str, float]) -> str: | |
| if shipment.to_warehouse not in self.inventory: | |
| return "unknown_destination_warehouse" | |
| if shipment.units <= 0: | |
| return "invalid_units" | |
| if shipment.unit_price > band["max_wholesale_price"]: | |
| return "unit_price_above_max_wholesale" | |
| if shipment.units > self.central_depot_inventory.get(shipment.sku, 0): | |
| return "insufficient_depot_stock" | |
| if used_units + shipment.units > cap: | |
| return "replenishment_cap_exceeded" | |
| if self.depot_trucks_available <= 0: | |
| return "depot_truck_unavailable" | |
| return "invalid_replenishment" | |
| def _receive_returns(self) -> None: | |
| for arrival, sku, units in [item for item in self.depot_procurements if item[0] <= self.round_index]: | |
| self._add_depot_stock(sku, units, 0.0) | |
| self.depot_procurements = [item for item in self.depot_procurements if item[0] > self.round_index] | |
| returned = [round_index for round_index in self.depot_truck_returns if round_index <= self.round_index] | |
| if returned: | |
| self.depot_trucks_available += len(returned) | |
| self.depot_truck_returns = [round_index for round_index in self.depot_truck_returns if round_index > self.round_index] | |
| for warehouse_id, return_rounds in self.driver_returns.items(): | |
| done = [round_index for round_index in return_rounds if round_index <= self.round_index] | |
| if done: | |
| self.drivers_available[warehouse_id] += len(done) | |
| self.driver_returns[warehouse_id] = [round_index for round_index in return_rounds if round_index > self.round_index] | |
| def _visible_orders(self) -> list[OrderTemplate]: | |
| recipe = self._require_recipe() | |
| return [order for order in recipe.orders if order.created_round <= self.round_index and self.order_status[order.order_id] not in {"fulfilled", "rejected", "expired"}] | |
| def _require_recipe(self) -> HiddenRecipe: | |
| if self.recipe is None: | |
| self.reset(self.default_task_id, 101) | |
| assert self.recipe is not None | |
| return self.recipe | |
| def _add(components: dict[str, float], key: str, value: float) -> None: | |
| components[key] = components.get(key, 0.0) + value | |
| def _broker_fee(units: int, compensation: float) -> float: | |
| per_unit_fee = cfg("center_rewards", "transfer_broker_fee_per_unit") * units | |
| compensation_fee = cfg("center_rewards", "transfer_broker_fee_compensation_share") * compensation | |
| return min(per_unit_fee, compensation_fee) | |