RohanExploit/Meta-hackathon
0
1"""Multi-channel retail environment with disruption recovery mechanics."""2from __future__ import annotations3 4import random5from typing import Any, Dict, List, Optional, Tuple6 7import numpy as np8 9from .grader import score_episode10from .models import (11 AllocateAction,12 ActionType,13 CompositeAction,14 DisruptionEvent,15 NoOpAction,16 OrderAction,17 PromoteAction,18 RetailAction,19 RetailObservation,20 RetailState,21 SetPriceAction,22)23 24 25def _dump_model(model: Any) -> Dict[str, Any]:26 """Convert Pydantic model to dict."""27 if hasattr(model, "model_dump"):28 return model.model_dump()29 return model.dict()30 31 32class MultiChannelRetailEnv:33 """34 Dynamic multi-channel retail with disruption recovery.35 36 Agents manage:37 - Multi-segment pricing (luxury/budget customers)38 - Inventory allocation across segments39 - Supply chain disruptions (lead time, demand shocks)40 - Recovery strategies (promotions, dynamic pricing)41 - Multi-action timesteps (CompositeAction)42 - Full pipeline visibility (pending_orders in observation)43 """44 45 def __init__(self, seed: Optional[int] = None):46 self.seed = seed47 if seed is not None:48 np.random.seed(seed)49 random.seed(seed)50 51 self.state: Optional[RetailState] = None52 self.initial_cash: float = 0.053 self._horizon: int = 3054 self.episode_metrics: Dict[str, float] = {}55 56 # Disruption parameters57 self._disruption_probability: float = 0.1558 self._disruption_recovery_days: int = 559 60 def reset(self, task_config: Dict[str, Any]) -> RetailObservation:61 """Reset environment with task configuration."""62 if self.seed is not None:63 np.random.seed(self.seed)64 random.seed(self.seed)65 66 products = list(task_config["products"])67 self._horizon = int(task_config.get("horizon", 30))68 69 # Initialize state70 inventory = {p: int(task_config["initial_inventory"].get(p, 5)) for p in products}71 cash = float(task_config["initial_cash"])72 self.initial_cash = cash73 74 # Demand patterns (hidden from agent)75 base_demand_luxury = {p: float(task_config.get("base_demand_luxury", {}).get(p, 1.5)) for p in products}76 base_demand_budget = {p: float(task_config.get("base_demand_budget", {}).get(p, 3.0)) for p in products}77 78 demand_elasticity = {p: float(task_config.get("demand_elasticity", {}).get(p, 1.2)) for p in products}79 80 # Economics81 product_costs = {p: float(task_config["product_costs"][p]) for p in products}82 holding_costs = {p: float(task_config.get("holding_costs", {}).get(p, 0.1)) for p in products}83 max_inventory = {p: float(task_config.get("max_inventory", {}).get(p, 50.0)) for p in products}84 85 # Pricing bounds86 price_bounds: Dict[str, Dict[str, float]] = {}87 for p in products:88 cost = product_costs[p]89 bounds_cfg = task_config.get("price_bounds", {}).get(p, {})90 price_bounds[p] = {91 "min": float(bounds_cfg.get("min", cost * 1.1)),92 "max": float(bounds_cfg.get("max", cost * 3.0)),93 }94 95 # Initial prices96 prices_luxury = {p: float(task_config.get("initial_prices_luxury", {}).get(p, product_costs[p] * 2.0)) for p in products}97 prices_budget = {p: float(task_config.get("initial_prices_budget", {}).get(p, product_costs[p] * 1.3)) for p in products}98 competitor_prices = {p: prices_budget[p] * 0.95 for p in products}99 100 # Supply chain101 lead_time_mean = int(task_config.get("lead_time_mean", 2))102 lead_time_variance = int(task_config.get("lead_time_variance", 1))103 supplier_reliability = float(task_config.get("supplier_reliability", 0.9))104 105 self.state = RetailState(106 day=0,107 cash=cash,108 inventory=inventory,109 base_demand_luxury=base_demand_luxury,110 base_demand_budget=base_demand_budget,111 demand_elasticity=demand_elasticity,112 prices_luxury=prices_luxury,113 prices_budget=prices_budget,114 competitor_prices=competitor_prices,115 price_bounds=price_bounds,116 product_costs=product_costs,117 holding_costs=holding_costs,118 max_inventory=max_inventory,119 lead_time_mean=lead_time_mean,120 lead_time_variance=lead_time_variance,121 pending_orders={p: 0 for p in products},122 pending_order_queue={},123 supplier_reliability=supplier_reliability,124 active_disruptions=[],125 disruption_history=[],126 next_disruption_day=None,127 cumulative_sales_luxury={p: 0.0 for p in products},128 cumulative_sales_budget={p: 0.0 for p in products},129 cumulative_revenue=0.0,130 cumulative_holding_cost=0.0,131 cumulative_stockouts=0,132 cumulative_demand_lost=0.0,133 cumulative_order_cost=0.0,134 demand_history_luxury={p: [] for p in products},135 demand_history_budget={p: [] for p in products},136 stockouts_per_product={p: 0 for p in products},137 )138 139 self.episode_metrics = {140 "horizon": float(self._horizon),141 "num_products": float(len(products)),142 "baseline_profit": float(task_config.get("baseline_profit", 100.0)),143 "total_steps": 0.0,144 "total_revenue": 0.0,145 "total_cost": 0.0,146 "total_holding_cost": 0.0,147 "total_order_cost": 0.0,148 "profit": 0.0,149 "total_demand": 0.0,150 "total_sales": 0.0,151 "fill_rate": 0.0,152 "stockout_count": 0,153 "disruption_events": 0,154 "recovery_success_rate": 0.0,155 "price_efficiency": 0.0,156 "max_possible_holding_cost": float(157 sum(float(max_inventory[p]) * float(holding_costs[p]) for p in products) * self._horizon158 ),159 }160 161 return self._get_observation()162 163 # ── Step (supports both single and composite actions) ───────────164 165 def step(self, action: RetailAction) -> Tuple[RetailObservation, float, bool, Dict[str, Any]]:166 """Execute one step in the environment.167 168 Supports both legacy single-action and new CompositeAction payloads.169 For CompositeAction, sub-actions are resolved in priority order:170 1. price_changes (before demand calculation)171 2. orders (supply chain)172 3. allocations (inventory split)173 4. promotions (demand boost)174 """175 if self.state is None:176 raise RuntimeError("Environment not initialized. Call reset() first.")177 178 info: Dict[str, Any] = {179 "action_taken": _dump_model(action),180 "valid_action": True,181 "disruption_event": None,182 "composite": isinstance(action, CompositeAction),183 }184 185 # Check for new disruptions186 self._check_disruptions()187 188 # Apply disruption effects to demand189 disruption_multiplier = self._compute_disruption_multiplier()190 191 # Update competitor prices192 for p in self.state.inventory.keys():193 my_bdg = self.state.prices_budget[p]194 if self.state.competitor_prices[p] > my_bdg:195 self.state.competitor_prices[p] = max(self.state.product_costs[p] * 1.05, my_bdg * 0.95)196 else:197 self.state.competitor_prices[p] *= random.uniform(0.95, 1.05)198 199 # ── Process action(s) ───────────────────────────────────────200 action_reward = 0.0201 202 if isinstance(action, CompositeAction):203 # Multi-action timestep: execute in priority order204 sub_infos: List[Dict[str, Any]] = []205 206 # 1. Price changes first (affect demand calculation)207 for sub in (action.price_changes or []):208 sub_info: Dict[str, Any] = {"valid_action": True}209 action_reward += self._apply_set_price(sub, sub_info)210 sub_infos.append({"type": "set_price", **sub_info})211 212 # 2. Orders (supply chain)213 for sub in (action.orders or []):214 sub_info = {"valid_action": True}215 action_reward += self._apply_order(sub, sub_info)216 sub_infos.append({"type": "order", **sub_info})217 218 # 3. Allocations219 for sub in (action.allocations or []):220 sub_info = {"valid_action": True}221 action_reward += self._apply_allocate(sub, sub_info)222 sub_infos.append({"type": "allocate", **sub_info})223 224 # 4. Promotions225 for sub in (action.promotions or []):226 sub_info = {"valid_action": True}227 action_reward += self._apply_promote(sub, sub_info)228 sub_infos.append({"type": "promote", **sub_info})229 230 info["sub_actions"] = sub_infos231 else:232 # Legacy single-action path233 if action.action == ActionType.ALLOCATE:234 action_reward = self._apply_allocate(action, info)235 elif action.action == ActionType.SET_PRICE:236 action_reward = self._apply_set_price(action, info)237 elif action.action == ActionType.ORDER:238 action_reward = self._apply_order(action, info)239 elif action.action == ActionType.PROMOTE:240 action_reward = self._apply_promote(action, info)241 elif action.action == ActionType.NOOP:242 pass243 else:244 info["valid_action"] = False245 246 # Process pending order arrivals247 self._realize_arrivals()248 249 # Simulate market (demand and sales)250 market_reward = self._simulate_market(disruption_multiplier)251 252 # Apply holding costs253 holding_cost = self._apply_holding_costs()254 255 # Total reward (step-wise signal aligned with profit)256 total_reward = action_reward + market_reward - holding_cost257 258 # Update state259 self.state.day += 1260 done = self.state.day >= self._horizon or self.state.cash < -100.0261 262 # Finalize metrics263 if done:264 self._finalize_episode()265 grader = score_episode(self.episode_metrics)266 info["grader"] = grader267 info["terminal_summary"] = {268 "cash": float(self.state.cash),269 "profit": float(self.episode_metrics.get("profit", 0.0)),270 "metrics": {k: float(v) for k, v in self.episode_metrics.items()},271 "grader": grader,272 }273 274 self.episode_metrics["total_steps"] += 1.0275 276 observation = self._get_observation()277 return observation, total_reward, done, info278 279 # ── Disruption mechanics ────────────────────────────────────────280 281 def _check_disruptions(self) -> None:282 """Randomly trigger disruptions (supply/demand shocks)."""283 assert self.state is not None284 285 # Check if any active disruptions should end286 for disp in self.state.active_disruptions[:]:287 if self.state.day >= disp.day_triggered + disp.duration_days:288 self.state.active_disruptions.remove(disp)289 290 # Chance of new disruption291 if random.random() < self._disruption_probability:292 event_types = ["demand_collapse", "supply_delay", "demand_spike"]293 event_type = random.choice(event_types)294 product = random.choice(list(self.state.inventory.keys()))295 severity = random.uniform(0.3, 0.9)296 duration = random.randint(2, 5)297 298 disruption = DisruptionEvent(299 event_type=event_type,300 product=product,301 severity=severity,302 duration_days=duration,303 day_triggered=self.state.day,304 )305 self.state.active_disruptions.append(disruption)306 self.state.disruption_history.append(disruption)307 self.episode_metrics["disruption_events"] += 1.0308 309 def _compute_disruption_multiplier(self) -> float:310 """Compute demand multiplier based on active disruptions."""311 if not self.state.active_disruptions:312 return 1.0313 314 # Demand collapses reduce demand, spikes increase it315 multiplier = 1.0316 for disp in self.state.active_disruptions:317 if disp.event_type == "demand_collapse":318 multiplier *= (1.0 - disp.severity * 0.8)319 elif disp.event_type == "demand_spike":320 multiplier *= (1.0 + disp.severity * 1.5)321 322 return max(0.1, min(3.0, multiplier)) # Clamp extremes323 324 # ── Action handlers ─────────────────────────────────────────────325 326 def _apply_allocate(self, action: RetailAction, info: Dict[str, Any]) -> float:327 """Allocate inventory to luxury and budget segments."""328 assert self.state is not None329 330 if action.product not in self.state.inventory:331 info["valid_action"] = False332 return -1.0333 334 total_allocated = action.luxury_units + action.budget_units335 if total_allocated > self.state.inventory[action.product]:336 info["valid_action"] = False337 return -1.0338 339 # Store allocation for market simulation340 info["allocation_luxury"] = action.luxury_units341 info["allocation_budget"] = action.budget_units342 343 return 0.5 # Small reward for proactive allocation344 345 def _apply_set_price(self, action: RetailAction, info: Dict[str, Any]) -> float:346 """Set price for a segment."""347 assert self.state is not None348 349 if action.product not in self.state.price_bounds:350 info["valid_action"] = False351 return -1.0352 353 bounds = self.state.price_bounds[action.product]354 if action.new_price < bounds["min"] or action.new_price > bounds["max"]:355 info["valid_action"] = False356 return -1.0357 358 if action.segment == "luxury":359 self.state.prices_luxury[action.product] = action.new_price360 elif action.segment == "budget":361 self.state.prices_budget[action.product] = action.new_price362 else:363 info["valid_action"] = False364 return -1.0365 366 info["price_set"] = action.new_price367 return 0.0 # Neutral reward; actual benefit comes from sales368 369 def _apply_order(self, action: RetailAction, info: Dict[str, Any]) -> float:370 """Place an order with supplier (subject to reliability and lead time)."""371 assert self.state is not None372 373 if action.product not in self.state.inventory:374 info["valid_action"] = False375 return -2.0376 377 supplier_choice = getattr(action, "supplier", "A").upper()378 if supplier_choice == "B":379 # Fast, reliable, expensive380 unit_cost = self.state.product_costs[action.product] * 1.25381 lead_time_mean = 1382 lead_time_variance = 0383 reliability = 1.0384 else:385 # Slow, cheap, unreliable386 unit_cost = self.state.product_costs[action.product]387 lead_time_mean = self.state.lead_time_mean388 lead_time_variance = self.state.lead_time_variance389 reliability = self.state.supplier_reliability390 391 cost = float(action.quantity) * unit_cost392 if cost > self.state.cash:393 info["valid_action"] = False394 return -2.0395 396 # Deduct cost immediately397 self.state.cash -= cost398 self.state.cumulative_order_cost += cost399 self.episode_metrics["total_order_cost"] = self.state.cumulative_order_cost400 401 # Stochastic lead time402 lead_time = max(0, int(np.random.normal(lead_time_mean, lead_time_variance)))403 arrival_day = self.state.day + lead_time404 405 # Chance order doesn't arrive (supplier unreliability)406 if random.random() > reliability:407 info["order_lost"] = True408 return -1.0 # Penalty for lost order409 410 arrival_bucket = self.state.pending_order_queue.setdefault(arrival_day, {})411 arrival_bucket[action.product] = int(arrival_bucket.get(action.product, 0)) + action.quantity412 self.state.pending_orders[action.product] += action.quantity413 414 info["order_enqueued"] = True415 info["arrival_day"] = arrival_day416 info["lead_time"] = lead_time417 info["supplier"] = supplier_choice418 return 0.0 # Neutral; benefit comes later419 420 def _apply_promote(self, action: RetailAction, info: Dict[str, Any]) -> float:421 """Run a promotional campaign (increases demand at cost)."""422 assert self.state is not None423 424 if action.budget_allocated > self.state.cash:425 info["valid_action"] = False426 return -1.0427 428 self.state.cash -= action.budget_allocated429 info["promotion_budget"] = action.budget_allocated430 info["promotion_multiplier"] = 1.0 + (action.budget_allocated / 100.0) # ROI estimate431 432 return -action.budget_allocated / 10.0 # Small upfront cost433 434 # ── Supply chain pipeline ───────────────────────────────────────435 436 def _realize_arrivals(self) -> None:437 """Process pending orders that arrive today."""438 assert self.state is not None439 440 due = self.state.pending_order_queue.pop(self.state.day, {})441 for product, qty in due.items():442 self.state.inventory[product] += int(qty)443 self.state.pending_orders[product] = max(444 0, int(self.state.pending_orders[product]) - int(qty)445 )446 447 def _build_pending_orders_observation(self) -> Dict[str, list]:448 """Build agent-visible pending order pipeline from internal queue.449 450 Transforms the internal {arrival_day: {product: qty}} queue into a451 per-product list of {quantity, days_to_arrival} dicts that satisfy452 the Markov property — the agent can now see its own pipeline.453 """454 assert self.state is not None455 456 products = list(self.state.inventory.keys())457 result: Dict[str, list] = {p: [] for p in products}458 459 for arrival_day, bucket in sorted(self.state.pending_order_queue.items()):460 days_to_arrival = max(0, arrival_day - self.state.day)461 for product, qty in bucket.items():462 if product in result:463 result[product].append({464 "quantity": int(qty),465 "days_to_arrival": days_to_arrival,466 })467 468 return result469 470 # ── Market simulation ───────────────────────────────────────────471 472 _DEMAND_HISTORY_WINDOW = 3 # rolling window size for demand history473 474 def _simulate_market(self, disruption_multiplier: float) -> float:475 """Simulate demand, allocations, and sales for each segment."""476 assert self.state is not None477 478 total_revenue = 0.0479 total_demand = 0.0480 total_sales = 0.0481 step_unmet = 0 # Track unmet demand for THIS step only482 483 seasonality_trend = 1.0 + 0.5 * np.sin(self.state.day / self._horizon * np.pi)484 485 for product in self.state.inventory.keys():486 comp_price = self.state.competitor_prices[product]487 # Base demand488 demand_lux = self._sample_demand(489 self.state.base_demand_luxury[product],490 self.state.prices_luxury[product],491 self.state.product_costs[product],492 self.state.demand_elasticity[product],493 disruption_multiplier * seasonality_trend,494 comp_price * 1.3,495 )496 demand_bdg = self._sample_demand(497 self.state.base_demand_budget[product],498 self.state.prices_budget[product],499 self.state.product_costs[product],500 self.state.demand_elasticity[product],501 disruption_multiplier * seasonality_trend,502 comp_price,503 )504 505 # Record actual demand in rolling history for realistic observations506 hist_lux = self.state.demand_history_luxury.setdefault(product, [])507 hist_lux.append(float(demand_lux))508 if len(hist_lux) > self._DEMAND_HISTORY_WINDOW:509 hist_lux.pop(0)510 511 hist_bdg = self.state.demand_history_budget.setdefault(product, [])512 hist_bdg.append(float(demand_bdg))513 if len(hist_bdg) > self._DEMAND_HISTORY_WINDOW:514 hist_bdg.pop(0)515 516 total_demand += demand_lux + demand_bdg517 518 # Allocate inventory (simple heuristic: luxury gets priority if priced higher)519 available = int(self.state.inventory[product])520 lux_price = self.state.prices_luxury[product]521 bdg_price = self.state.prices_budget[product]522 523 # Luxury segment gets preference if higher margin524 if lux_price > bdg_price:525 sales_lux = min(int(demand_lux), available // 2)526 sales_bdg = min(int(demand_bdg), available - sales_lux)527 else:528 sales_bdg = min(int(demand_bdg), available // 2)529 sales_lux = min(int(demand_lux), available - sales_bdg)530 531 # Revenue532 revenue_lux = sales_lux * lux_price533 revenue_bdg = sales_bdg * bdg_price534 total_revenue += revenue_lux + revenue_bdg535 536 # Update inventory and metrics537 self.state.inventory[product] -= sales_lux + sales_bdg538 self.state.cumulative_sales_luxury[product] += sales_lux539 self.state.cumulative_sales_budget[product] += sales_bdg540 total_sales += sales_lux + sales_bdg541 542 # Stockout tracking — per product543 unmet_lux = max(0, int(demand_lux) - sales_lux)544 unmet_bdg = max(0, int(demand_bdg) - sales_bdg)545 unmet = unmet_lux + unmet_bdg546 step_unmet += unmet547 548 if unmet > 0:549 self.episode_metrics["stockout_count"] += 1.0550 self.state.cumulative_stockouts += 1551 self.state.cumulative_demand_lost += float(unmet)552 self.state.stockouts_per_product[product] = (553 self.state.stockouts_per_product.get(product, 0) + 1554 )555 556 self.state.cash += total_revenue557 self.state.cumulative_revenue += total_revenue558 559 self.episode_metrics["total_revenue"] += total_revenue560 self.episode_metrics["total_demand"] += total_demand561 self.episode_metrics["total_sales"] += total_sales562 563 # Return reward signal: per-step revenue minus per-step stockout penalty564 stockout_penalty = step_unmet * 2.0565 return total_revenue - stockout_penalty566 567 def _sample_demand(568 self,569 base_demand: float,570 price: float,571 product_cost: float,572 elasticity: float,573 disruption_multiplier: float,574 competitor_price: float,575 ) -> int:576 """Sample demand with price elasticity and disruptions."""577 rel_price = price / max(0.01, competitor_price)578 price_effect = rel_price ** (-elasticity * 1.5)579 580 effective_demand = base_demand * price_effect * disruption_multiplier581 effective_demand = max(0.1, effective_demand)582 583 return int(np.random.poisson(effective_demand))584 585 def _apply_holding_costs(self) -> float:586 """Apply inventory holding costs."""587 assert self.state is not None588 589 total_cost = 0.0590 for product, qty in self.state.inventory.items():591 cost = float(qty) * self.state.holding_costs[product]592 total_cost += cost593 594 self.state.cash -= total_cost595 self.state.cumulative_holding_cost += total_cost596 self.episode_metrics["total_cost"] += total_cost597 self.episode_metrics["total_holding_cost"] += total_cost598 599 return total_cost600 601 def _finalize_episode(self) -> None:602 """Finalize episode metrics."""603 assert self.state is not None604 605 profit = self.state.cash - self.initial_cash606 self.episode_metrics["profit"] = profit607 608 if self.episode_metrics["total_demand"] > 0:609 self.episode_metrics["fill_rate"] = self.episode_metrics["total_sales"] / self.episode_metrics["total_demand"]610 else:611 self.episode_metrics["fill_rate"] = 1.0612 613 if self.episode_metrics["disruption_events"] > 0:614 recovery_count = len([d for d in self.state.disruption_history if d.duration_days < 5])615 self.episode_metrics["recovery_success_rate"] = recovery_count / self.episode_metrics["disruption_events"]616 else:617 self.episode_metrics["recovery_success_rate"] = 0.0618 619 # ── Observation builder ─────────────────────────────────────────620 621 def _get_observation(self) -> RetailObservation:622 """Get partial state observation for agent (Markov-compliant)."""623 if self.state is None:624 raise RuntimeError("Environment not initialized")625 626 # Use actual rolling demand history when available; fall back to base constants627 # on day 0 before any market simulation has run.628 recent_demand_luxury: Dict[str, float] = {}629 recent_demand_budget: Dict[str, float] = {}630 for p in self.state.inventory.keys():631 hist_lux = self.state.demand_history_luxury.get(p, [])632 recent_demand_luxury[p] = (633 float(sum(hist_lux) / len(hist_lux))634 if hist_lux635 else float(max(0.1, self.state.base_demand_luxury[p]))636 )637 hist_bdg = self.state.demand_history_budget.get(p, [])638 recent_demand_budget[p] = (639 float(sum(hist_bdg) / len(hist_bdg))640 if hist_bdg641 else float(max(0.1, self.state.base_demand_budget[p]))642 )643 644 # Per-product stockout counts (how many steps this product had unmet demand)645 recent_stockouts = {646 p: int(self.state.stockouts_per_product.get(p, 0))647 for p in self.state.inventory.keys()648 }649 650 disruption_active = len(self.state.active_disruptions) > 0651 disruption_severity = max([d.severity for d in self.state.active_disruptions], default=0.0)652 market_confidence = max(0.0, 0.9 - (self.state.day / self._horizon) * 0.3)653 654 # Build pipeline visibility (Markov property compliance)655 pending_orders = self._build_pending_orders_observation()656 657 return RetailObservation(658 day=self.state.day,659 cash=self.state.cash,660 inventory=self.state.inventory.copy(),661 pending_orders=pending_orders,662 recent_demand_luxury=recent_demand_luxury,663 recent_demand_budget=recent_demand_budget,664 recent_stockouts=recent_stockouts,665 prices_luxury=self.state.prices_luxury.copy(),666 prices_budget=self.state.prices_budget.copy(),667 competitor_prices=self.state.competitor_prices.copy(),668 disruption_active=disruption_active,669 disruption_severity=disruption_severity,670 market_confidence=market_confidence,671 seasonality_multiplier=1.0 + 0.5 * np.sin(self.state.day / self._horizon * np.pi),672 )673 674 def get_state(self) -> Dict[str, Any]:675 """Get full internal state."""676 if self.state is None:677 raise RuntimeError("Environment not initialized")678 679 return _dump_model(self.state)680