velocitatem/PHANTOM
1
0
Fork 0
mirror of https://github.com/velocitatem/PHANTOM.git synced 2026-05-31 08:33:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

naoice COI implementation

Browse Source
This commit is contained in:
Daniel Rosel
2026-02-02 11:18:37 +01:00
parent 4abef97bf7
commit c4fd1352c9
5 changed files with 221 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
engine/wrapper.py
View File

@@ -3,30 +3,42 @@ from gymnasium import spaces
import numpy as np
from .engine import Limbo, MarketEngine, PricingEngine
from .lib.render import DashboardRenderer
from .lib.coi import compute_coi_proxy
from .lib.coi import (
compute_coi_leakage,
compute_erosion_metrics,
compute_agent_probability,
)
from .lib.behavior import get_transition_models, trajectory_to_events
from .lib.wrappers import EconomicMetricsWrapper
class PHANTOM(gym.Env):
"""Gymnasium wrapper for the Limbo pricing-market simulation. Platform sets prices, market responds with demand."""
"""Gymnasium wrapper for Limbo pricing-market simulation implementing thesis COI framework
reward = R(p,d) - λ·COI_leak(p,τ') per thesis Section on DR-RL
COI_leak uses behavioral divergence to estimate agent probability f(τ')
"""
metadata = {"render_modes": ["human", "ansi"]}
def __init__(self,
n_products: int = 10,
alpha: float = 0.3,
N: int = 100,
human_params: tuple = (50.0, 10.0),
agent_params: tuple = (45.0, 15.0),
noise_std: float = 1.0,
price_bounds: tuple = (10.0, 150.0),
lambda_coi: float = 0.1,
coi_window: int = 10,
render_mode: str = None):
def __init__(
self,
n_products: int = 10,
alpha: float = 0.3,
N: int = 100,
human_params: tuple = (50.0, 10.0),
agent_params: tuple = (45.0, 15.0),
noise_std: float = 1.0,
price_bounds: tuple = (10.0, 150.0),
lambda_coi: float = 0.1,
coi_window: int = 10,
render_mode: str = None,
):
super().__init__()
self.n_products = n_products
self.price_bounds = price_bounds
self.lambda_coi = lambda_coi
self.coi_window = coi_window # K steps for rolling COI calculation
self.coi_window = coi_window
self.render_mode = render_mode
self.alpha = alpha
self.N = N
@@ -34,20 +46,34 @@ class PHANTOM(gym.Env):
self.agent_params = agent_params
self.market = MarketEngine(
alpha=alpha, N=N,
human_params=human_params, agent_params=agent_params, noise_std=noise_std
alpha=alpha,
N=N,
human_params=human_params,
agent_params=agent_params,
noise_std=noise_std,
)
self._platform_stub = PricingEngine()
self._limbo = Limbo(self._platform_stub, self.market)
self.action_space = spaces.Box(
low=price_bounds[0], high=price_bounds[1],
shape=(n_products,), dtype=np.float32
low=price_bounds[0],
high=price_bounds[1],
shape=(n_products,),
dtype=np.float32,
)
self.observation_space = spaces.Dict(
{
"demand": spaces.Box(
low=0.0, high=100.0, shape=(n_products,), dtype=np.float32
),
"prices": spaces.Box(
low=price_bounds[0],
high=price_bounds[1],
shape=(n_products,),
dtype=np.float32,
),
}
)
self.observation_space = spaces.Dict({
"demand": spaces.Box(low=0.0, high=100.0, shape=(n_products,), dtype=np.float32),
"prices": spaces.Box(low=price_bounds[0], high=price_bounds[1], shape=(n_products,), dtype=np.float32),
})
self._prices = None
self._demand = None
@@ -56,25 +82,61 @@ class PHANTOM(gym.Env):
self._price_history = []
self._revenue_history = []
self._renderer = None
self._initial_episode_prices = None # prices at episode start for COI calc
self._initial_episode_prices = None
self._trajectories = [] # session trajectories for agent prob calculation
# load behavioral models for agent probability estimation
try:
self._human_trans, self._agent_trans = get_transition_models()
except Exception:
# fallback if behavioral data unavailable
self._human_trans, self._agent_trans = None, None
def _get_obs(self) -> dict:
demand_arr = np.array([self._demand.get(i, 0.0) for i in range(self.n_products)], dtype=np.float32)
demand_arr = np.array(
[self._demand.get(i, 0.0) for i in range(self.n_products)], dtype=np.float32
)
return {"demand": demand_arr, "prices": self._prices.astype(np.float32)}
def _compute_coi_proxy(self):
return compute_coi_proxy(
self._price_history, self._demand_history, self._initial_episode_prices,
self._prices, self.price_bounds, self.alpha, self.coi_window
def _compute_agent_prob(self) -> float:
"""estimate agent probability from accumulated trajectories using KL divergence"""
if (
not self._trajectories
or self._human_trans is None
or self._agent_trans is None
):
return self.alpha # fallback to contamination level
# aggregate all trajectories from this episode
all_events = []
for traj in self._trajectories:
all_events.extend(trajectory_to_events(traj))
if len(all_events) < 2:
return self.alpha
return compute_agent_probability(
all_events, self._human_trans, self._agent_trans
)
def _compute_reward(self, prices: np.ndarray, demand: dict) -> float:
revenue = np.sum(prices * np.array([demand.get(i, 0.0) for i in range(self.n_products)]))
coi_penalty = self.lambda_coi * self._compute_coi_proxy()
revenue = np.sum(
prices * np.array([demand.get(i, 0.0) for i in range(self.n_products)])
)
# compute agent probability from behavioral trajectories
agent_prob = self._compute_agent_prob()
# COI leakage: minimal implementation per thesis
coi_leakage = compute_coi_leakage(prices, agent_prob)
coi_penalty = self.lambda_coi * coi_leakage
return float(revenue - coi_penalty)
def _record_history(self):
demand_arr = np.array([self._demand.get(i, 0.0) for i in range(self.n_products)])
demand_arr = np.array(
[self._demand.get(i, 0.0) for i in range(self.n_products)]
)
self._demand_history.append(demand_arr)
self._price_history.append(self._prices.copy())
self._revenue_history.append(np.sum(self._prices * demand_arr))
@@ -82,10 +144,11 @@ class PHANTOM(gym.Env):
def reset(self, seed=None, options=None):
super().reset(seed=seed)
self._prices = np.random.uniform(*self.price_bounds, size=self.n_products)
self._initial_episode_prices = self._prices.copy() # snapshot for COI calculation
self._initial_episode_prices = self._prices.copy()
self._demand = self.market.act(self._prices)
self._step_count = 0
self._demand_history, self._price_history, self._revenue_history = [], [], []
self._trajectories = []
self._record_history()
return self._get_obs(), {}
@@ -95,15 +158,36 @@ class PHANTOM(gym.Env):
self._step_count += 1
self._record_history()
coi_proxy = self._compute_coi_proxy()
# capture trajectories generated by market for agent prob estimation
if hasattr(self.market, "last_trajectories"):
self._trajectories.extend(self.market.last_trajectories)
agent_prob = self._compute_agent_prob()
coi_leakage = compute_coi_leakage(self._prices, agent_prob)
reward = self._compute_reward(self._prices, self._demand)
terminated = self._step_count >= 100
# legacy erosion metrics for comparison
erosion = compute_erosion_metrics(
self._price_history,
self._demand_history,
self._initial_episode_prices,
self._prices,
self.price_bounds,
self.alpha,
self.coi_window,
)
info = {
"step": self._step_count,
"coi_proxy": coi_proxy,
"coi_penalty": self.lambda_coi * coi_proxy,
"raw_revenue": np.sum(self._prices * np.array([self._demand.get(i, 0.0) for i in range(self.n_products)])),
"agent_prob": agent_prob,
"coi_leakage": coi_leakage,
"coi_penalty": self.lambda_coi * coi_leakage,
"erosion_metrics": erosion,
"raw_revenue": np.sum(
self._prices
* np.array([self._demand.get(i, 0.0) for i in range(self.n_products)])
),
}
return self._get_obs(), reward, terminated, False, info
@@ -114,10 +198,16 @@ class PHANTOM(gym.Env):
p, q = np.array(self._price_history), np.array(self._demand_history)
dp, dq = np.diff(p, axis=0), np.diff(q, axis=0)
valid = np.abs(dp) > 0.5
with np.errstate(divide='ignore', invalid='ignore'):
elasticity = np.where(valid, (dq / dp) * (p[:-1] / np.maximum(q[:-1], 1.0)), 0.0)
with np.errstate(divide="ignore", invalid="ignore"):
elasticity = np.where(
valid, (dq / dp) * (p[:-1] / np.maximum(q[:-1], 1.0)), 0.0
)
elasticity = np.nan_to_num(np.clip(elasticity, -5.0, 5.0), nan=0.0)
return np.mean(elasticity, axis=0) if len(elasticity) > 0 else np.zeros(self.n_products)
return (
np.mean(elasticity, axis=0)
if len(elasticity) > 0
else np.zeros(self.n_products)
)
def render(self):
if self.render_mode == "human":
@@ -125,7 +215,9 @@ class PHANTOM(gym.Env):
self._renderer = DashboardRenderer()
self._renderer.render(self)
elif self.render_mode == "ansi":
return f"step={self._step_count}, prices={self._prices}, demand={self._demand}"
return (
f"step={self._step_count}, prices={self._prices}, demand={self._demand}"
)
return None
def close(self):
@@ -140,6 +232,7 @@ if __name__ == "__main__":
class RandomPolicy:
"""Minimal SB3-compatible random policy for baseline testing."""
def __init__(self, env):
self.env = env
self.num_timesteps = 0