fix

Author: jxd136

logic/pve/GameLogic/action_space.py

@@ -133,17 +133,22 @@ def compute_action_mask(env: "GameEnvironment") -> np.ndarray:
         if board.is_passable(nx, ny):
             mask[act] = True
 
-    # BUY: adjacent market + capacity + money
-    mkt = board.nearest_market(u.x, u.y)
-    if mkt is not None:
-        best_cost = min(pdef["cost"] for pdef in PRODUCT_DEFS.values())
-        if u.free_capacity >= 1 and env.money >= best_cost:
+    # BUY: adjacent market + capacity + money (affordability checked against market price)
+    mkt_pos = board.nearest_market(u.x, u.y)
+    mkt_obj = None
+    if mkt_pos is not None:
+        mkt_obj = next((m for m in env.markets if m.x == mkt_pos[0] and m.y == mkt_pos[1]), None)
+    if mkt_obj is not None:
+        min_mkt_price = min(mkt_obj.get_price(pid, env.time, 1.0) for pid in PRODUCT_DEFS)
+        if u.free_capacity >= 1 and env.money >= min_mkt_price:
             mask[Action.BUY] = True
 
-        # SELL_pid: carrying that product type
+        # SELL_pid: carrying that product + not same-market-origin (arbitrage block)
         for sell_act in SELL_ACTIONS:
             pid = sell_act - Action.SELL_0
             if u.prod_inv.get(pid, 0.0) > 0:
+                if u.prod_origin.get(pid) == mkt_obj.id:
+                    continue  # blocked: bought here, must sell elsewhere
                 mask[sell_act] = True
 
     # HARVEST: nearby non-depleted resource + capacity

logic/pve/GameLogic/character.py

@@ -21,6 +21,10 @@ class Unit:
     raw_inv: float = field(default=0.0, init=False)
     prod_inv: Dict[int, float] = field(default_factory=dict, init=False)
 
+    # Tracks where each product type was last purchased (None = factory-produced).
+    # Used to enforce cross-market arbitrage: can't sell at the market where you bought.
+    prod_origin: Dict[int, int] = field(default_factory=dict, init=False)
+
     # Busy system: while busy_ticks > 0 the unit ignores new commands
     busy_ticks: int = field(default=0, init=False)
     busy_action: str = field(default="", init=False)  # tag for what's completing
@@ -30,6 +34,7 @@ class Unit:
     def __post_init__(self):
         self.hp = self.max_hp
         self.prod_inv = {}
+        self.prod_origin = {}
 
     # ── Inventory helpers ──────────────────────────────────────────────────────
     @property

logic/pve/GameLogic/game_env.py

@@ -234,13 +234,15 @@ def step(self, action: int) -> Tuple[np.ndarray, float, bool, bool, dict]:
         self.factory.tick()
         self._accrue_compute(dt)
 
-        # ── 5. Compute reward ──────────────────────────────────────────────
-        reward = self._reward_calc.compute(self, action_valid, harvested)
-
-        # ── 6. Termination / truncation ────────────────────────────────────
+        # ── 5. Termination / truncation (evaluated before reward so terminal bonus fires)
         terminated = self.money < 0
         truncated  = self._step >= cfg.max_steps
 
+        # ── 6. Compute reward ──────────────────────────────────────────────
+        reward = self._reward_calc.compute(
+            self, action_valid, harvested, terminated=(terminated or truncated)
+        )
+
         obs = self._encode_obs()
         info = {
             "step": self._step,
@@ -284,13 +286,14 @@ def _execute_action(self, action: Action) -> Tuple[bool, float]:
             mkt = self._market_at(*mkt_pos)
             if mkt is None or u.free_capacity < 1:
                 return False, 0.0
-            # Buy the product with highest profit margin (price - cost) we can afford
+            # Buy at market price; select product with most upside for cross-market resale
             best_pid, best_cost = self._best_buyable(mkt)
             if best_pid is None:
                 return False, 0.0
-            cost = best_cost
-            self.money -= cost
+            self.money -= best_cost
             u.add_product(best_pid, 1.0)
+            # Record which market this product came from to enforce cross-market rule
+            u.prod_origin[best_pid] = mkt.id
             u.state = "loading"
             u.busy_ticks = max(1, int(0.25 / cfg.time_step))
             u.busy_action = "buy_done"
@@ -307,9 +310,14 @@ def _execute_action(self, action: Action) -> Tuple[bool, float]:
             qty = u.prod_inv.get(pid, 0.0)
             if qty <= 0:
                 return False, 0.0
+            # Strict same-location arbitrage prevention: block selling at the market
+            # where this product was purchased (requires cross-market movement to profit)
+            if u.prod_origin.get(pid) == mkt.id:
+                return False, 0.0
             mult = self._price_multiplier()
             revenue = mkt.get_price(pid, self.time, mult) * qty
             u.prod_inv[pid] = 0.0
+            u.prod_origin.pop(pid, None)
             self.money += revenue
             self.score += revenue * cfg.score_factor
             u.state = "selling"
@@ -354,9 +362,17 @@ def _execute_action(self, action: Action) -> Tuple[bool, float]:
         if action == Action.LOAD:
             if not board.at_factory(u.x, u.y):
                 return False, 0.0
+            # Snapshot inventory before loading to detect which slots were empty
+            pre_inv = {pid: u.prod_inv.get(pid, 0.0) for pid in PRODUCT_DEFS}
             loaded = self.factory.load_products(u)
             if loaded <= 0:
                 return False, 0.0
+            # Factory-loaded products get None origin (sellable anywhere) only when
+            # the slot was empty beforehand; if there are existing market-tainted units
+            # of that type, the taint persists to prevent mixing-based bypass.
+            for pid in PRODUCT_DEFS:
+                if u.prod_inv.get(pid, 0.0) > pre_inv[pid] and pre_inv[pid] == 0:
+                    u.prod_origin.pop(pid, None)  # None / absent means factory origin
             u.state = "loading"
             u.busy_ticks = max(1, int(0.25 / cfg.time_step))
             u.busy_action = "load_done"
@@ -427,19 +443,24 @@ def _market_at(self, x: int, y: int) -> Optional[Market]:
         return None
 
     def _best_buyable(self, mkt: Market) -> Tuple[Optional[int], float]:
-        """Return (pid, cost) of product with highest profit (price - cost) we can afford."""
-        best_pid, best_cost = None, None
-        best_profit = -float("inf")
-        mult = self._price_multiplier()
+        """Return (pid, buy_price) of affordable product with most upside (hi - current_price).
+
+        Buying costs the current market price (not manufacturing cost), so same-location
+        buy-then-sell yields no profit. Cross-market arbitrage remains viable.
+        """
+        best_pid, best_price = None, None
+        best_upside = -float("inf")
         for pid, pdef in PRODUCT_DEFS.items():
-            cost = max(0, pdef["cost"] + self.factory.cost_delta)
-            if self.money < cost:
+            price = mkt.get_price(pid, self.time, 1.0)  # buy at market price, no marketing mult
+            if self.money < price:
                 continue
-            price = mkt.get_price(pid, self.time, mult)
-            profit = price - cost
-            if profit > best_profit:
-                best_profit, best_cost, best_pid = profit, cost, pid
-        return best_pid, best_cost
+            hi = pdef["val_range"][1]
+            upside = hi - price
+            if upside > best_upside:
+                best_upside = upside
+                best_pid = pid
+                best_price = price
+        return best_pid, best_price
 
     def _price_multiplier(self) -> float:
         return self.factory.price_multiplier

logic/pve/GameLogic/reward_calculator.py

@@ -1,10 +1,18 @@
 """
 Reward calculator: converts game state transitions into scalar RL rewards.
 
-Design principles:
-  - Primary: sales revenue (direct score signal)
-  - Secondary: small shaping bonuses for progress sub-goals
-  - Penalties: time cost, invalid actions, bankruptcy
+Design intent (anti-PPO/DQN):
+  - NO per-step money or score signal: removes the dense credit-assignment ladder
+    that standard PPO/DQN relies on to learn buy→sell chains.
+  - Harvest trap: a small positive reward for harvesting raw materials creates a
+    deceptive local optimum. PPO/DQN converges to a harvest-loop policy (stable,
+    positive reward per episode) that never produces or sells anything, yielding
+    zero game score.
+  - Terminal-only score bonus: the true objective is only revealed at episode end.
+    With γ=0.99 and ~1200 steps, 0.99^1200 ≈ 0, so gradients from the terminal
+    bonus do not propagate back to early actions — PPO/DQN cannot learn from it.
+  - Agents with hierarchical reward decomposition, curiosity-driven exploration,
+    or hand-crafted sub-goal rewards can still solve the environment.
 """
 from __future__ import annotations
 from dataclasses import dataclass
@@ -16,81 +24,61 @@
 
 @dataclass
 class RewardConfig:
-    # Scale factor applied to money gained/lost
-    money_scale: float = 0.01
+    # Per-step money / score delta scale — set to 0 to remove dense sell signal
+    money_scale: float = 0.0
 
-    # Per-step time penalty (encourages efficient routes)
-    time_penalty: float = -0.002
+    # Per-step time cost (encourages doing *something*, but not enough to overcome
+    # the harvest bonus on its own)
+    time_penalty: float = -0.003
 
-    # Reward for harvesting resources (normalized)
-    harvest_bonus_per_unit: float = 0.001
+    # Harvest trap: gives a positive per-unit signal that PPO/DQN latches onto.
+    # Harvesting alone never produces game score, making this a deceptive optimum.
+    harvest_bonus_per_unit: float = 0.01
 
-    # Reward for opening a compute center
-    compute_center_bonus: float = 0.5
-
-    # Reward for buying a tech upgrade (one-time)
-    tech_bonus: float = 1.0
+    # Sparse terminal bonus: final_score × scale, given only at episode end.
+    # Too sparse for standard PPO/DQN to credit-assign across ~1200 steps.
+    terminal_score_scale: float = 0.001
 
     # Penalty for attempting an invalid action
-    invalid_action_penalty: float = -0.05
+    invalid_action_penalty: float = -0.02
 
-    # Terminal rewards
+    # Terminal penalty for going bankrupt
     bankruptcy_penalty: float = -10.0
 
 
 class RewardCalculator:
     def __init__(self, cfg: RewardConfig = None):
         self.cfg = cfg or RewardConfig()
 
-        # Tracked across steps for shaping
-        self._prev_money: float = 0.0
-        self._prev_compute: float = 0.0
-        self._prev_score: float = 0.0
-        self._prev_open_centers: int = 0
-
-    def reset(self, env: "GameEnvironment"):
-        self._prev_money = env.money
-        self._prev_compute = env.compute
-        self._prev_score = env.score
-        self._prev_open_centers = sum(1 for cc in env.board.compute_centers if cc.is_open)
+    def reset(self, _env: "GameEnvironment"):
+        pass  # no per-step state to initialise
 
     def compute(
         self,
         env: "GameEnvironment",
         action_was_valid: bool,
         harvested: float,
+        terminated: bool = False,
     ) -> float:
         cfg = self.cfg
         reward = 0.0
 
-        # ── Money delta ─────────────────────────────────────────────────────
-        money_delta = env.money - self._prev_money
-        reward += money_delta * cfg.money_scale
-        self._prev_money = env.money
-
-        # ── Score delta (direct optimization target) ────────────────────────
-        score_delta = env.score - self._prev_score
-        reward += score_delta * cfg.money_scale
-        self._prev_score = env.score
-
-        # ── Time penalty ────────────────────────────────────────────────────
+        # ── Time cost (always) ──────────────────────────────────────────────
         reward += cfg.time_penalty
 
-        # ── Harvest shaping ─────────────────────────────────────────────────
+        # ── Harvest trap (deceptive local optimum for PPO/DQN) ──────────────
         reward += harvested * cfg.harvest_bonus_per_unit
 
-        # ── Compute center unlocked ─────────────────────────────────────────
-        open_centers = sum(1 for cc in env.board.compute_centers if cc.is_open)
-        if open_centers > self._prev_open_centers:
-            reward += cfg.compute_center_bonus
-        self._prev_open_centers = open_centers
-
-        # ── Invalid action ──────────────────────────────────────────────────
+        # ── Invalid action penalty ──────────────────────────────────────────
         if not action_was_valid:
             reward += cfg.invalid_action_penalty
 
         # ── Bankruptcy ──────────────────────────────────────────────────────
         if env.money < 0:
             reward += cfg.bankruptcy_penalty
 
+        # ── Terminal score bonus (sparse; unreachable by standard credit assign)
+        if terminated:
+            reward += env.score * cfg.terminal_score_scale
+
         return float(reward)