Spaces:
Sleeping
Sleeping
| """ | |
| Grader for inventory optimization tasks. | |
| Scores agent performance on a 0.0-1.0 scale using floor/ceiling approach. | |
| - floor: passive agent (no buys, just sells initial stock until empty) | |
| - ceiling: theoretical max profit with perfect demand knowledge | |
| """ | |
| from server.inventory_env import InventoryEnvironment | |
| from models import InventoryAction | |
| from server.constants import ( | |
| TASKS, BASE_PRICES, COST_PRICES, SHIPPING_COST, EVENT_EFFECTS, | |
| WEEKEND_MULTIPLIER, EVENT_DURATION, | |
| ) | |
| import random | |
| def _run_passive(task_name): | |
| """Floor baseline: do nothing, just sell whatever initial stock covers.""" | |
| env = InventoryEnvironment(task_name) | |
| obs = env.reset() | |
| while not obs.done: | |
| action = InventoryAction( | |
| buy_quantities={}, | |
| delivery_method="slow", | |
| liquidate={}, | |
| ) | |
| obs = env.step(action) | |
| return obs.total_profit | |
| def _run_heuristic(task_name): | |
| task = TASKS[task_name] | |
| events = dict(task["events"]) | |
| total_demand = {p: 0 for p in task["base_demand"]} | |
| for day in range(1, task["max_days"] + 1): | |
| # tick events | |
| for event_name in events: | |
| events[event_name] -= 1 | |
| rng = random.Random(task["seed"] * 1000 + day) | |
| for product, (lo, hi) in task["base_demand"].items(): | |
| demand = rng.randint(lo, hi) | |
| # weekend boost | |
| if day % 7 == 5 or day % 7 == 6: | |
| demand = int(WEEKEND_MULTIPLIER * demand) | |
| # event multipliers | |
| for event_name, days_left in events.items(): | |
| if -EVENT_DURATION < days_left <= 0 and event_name in EVENT_EFFECTS: | |
| mult = EVENT_EFFECTS[event_name].get(product, 1.0) | |
| demand = int(demand * mult) | |
| total_demand[product] += demand | |
| total_profit = 0.0 | |
| # sell the initial stock first | |
| initial_stock = task["initial_stock"] | |
| for product in task["base_demand"]: | |
| total_profit += min(initial_stock.get(product, 0), total_demand[product]) * BASE_PRICES[product] | |
| total_demand[product] = max(0, total_demand[product] - initial_stock.get(product, 0)) | |
| # cost price and shipping cost applies after initial stock | |
| total_profit += total_demand[product] * (BASE_PRICES[product] - COST_PRICES[product] - SHIPPING_COST["slow"]) | |
| return total_profit | |
| def compute_baselines(task_name): | |
| """Pre-compute floor and ceiling for a task.""" | |
| floor = _run_passive(task_name) | |
| ceiling = _run_heuristic(task_name) | |
| return floor, ceiling | |
| def grade(task_name, agent_profit): | |
| """ | |
| Grade agent performance on 0.0-1.0 scale. | |
| Args: | |
| task_name: "easy", "medium", or "hard" | |
| agent_profit: total profit achieved by the agent | |
| Returns: | |
| float score between 0.0 and 1.0 | |
| """ | |
| floor, ceiling = compute_baselines(task_name) | |
| if ceiling <= floor: | |
| return 1.0 if agent_profit >= ceiling else 0.0 | |
| score = (agent_profit - floor) / (ceiling - floor) | |
| return max(0.002, min(0.998, score)) | |
| def grade_all(results): | |
| """ | |
| Grade all 3 tasks. | |
| Args: | |
| results: dict of {task_name: agent_profit} | |
| Returns: | |
| dict of {task_name: score} | |
| """ | |
| scores = {} | |
| for task_name, agent_profit in results.items(): | |
| scores[task_name] = grade(task_name, agent_profit) | |
| return scores | |
| if __name__ == "__main__": | |
| print("Computing baselines for all tasks...") | |
| for task_name in ["easy", "medium", "hard"]: | |
| floor, ceiling = compute_baselines(task_name) | |
| print(f" {task_name}: floor={floor:.2f}, ceiling={ceiling:.2f}") |