1"""LLMPolicy: Observation -> prompt -> provider -> validated Decision -> Action.23Validation is local and strict: quantities within [1, max_order_qty], limit prices within ±20% of the4mid and on the tick grid, cash/holdings feasibility, cancel ids owned by the agent. Invalid decisions5become `hold` with meta.valid=False and a reason; they are counted in the reliability statistics.6Decisions are made in batches (one per decision round) with a bounded thread pool."""7from __future__ import annotations89import json10import time11from concurrent.futures import ThreadPoolExecutor12from dataclasses import dataclass, field1314import numpy as np1516from ..agents.base import Observation17from ..exchange.orders import Action, CancelIntent, OrderIntent, OrderType, Side18from .budget import BudgetExceeded, BudgetTracker19from .cache import DecisionCache20from .prompting import PROMPT_VERSION, system_prompt, user_message21from .providers import ProviderResult, get_provider22from .schema import Decision, parse_decision232425@dataclass(slots=True)26class LLMDecisionRecord:27 t: int28 agent_id: int29 provider: str30 model: str31 prompt_id: str32 valid: bool33 reason: str34 action: str35 side: str36 order_type: str37 quantity: int38 limit_price: float39 confidence: float40 rationale: str41 input_tokens: int42 cached_input_tokens: int43 output_tokens: int44 reasoning_tokens: int45 latency_s: float46 usd: float47 cached_hit: bool48 error: str49 prompt_sha: str50 request_id: str = ""51 features: str = "" # JSON feature vector (for the frozen library)52 rel_price: float = float("nan")535455@dataclass56class LLMPolicyConfig:57 models: dict[str, str]58 prompt_version: str = PROMPT_VERSION59 reasoning_effort: str | None = "low"60 temperature: float | None = None61 max_output_tokens: int = 120062 concurrency: int = 863 timeout_s: float = 60.064 max_retries: int = 365 usd_cap: float = 5.066 token_cap: int | None = None67 cache_path: str | None = None68 use_cache: bool = True69 frozen_only: bool = False # never call a provider; cache misses -> hold (frozen library replay)70 adaptive_prompt: bool = False71 mu_cents: float = 50.072 taker_bps: float = 2.073 order_ttl: int | None = 5074 decision_interval: int = 2075 price_assumptions: dict[str, dict[str, float]] = field(default_factory=dict)76 mock_kwargs: dict = field(default_factory=dict)77 provider_kwargs: dict = field(default_factory=dict)78 library_path: str | None = None798081class LLMPolicy:82 name = "llm"8384 def __init__(self, cfg: LLMPolicyConfig) -> None:85 self.cfg = cfg86 self.budget = BudgetTracker(cfg.usd_cap, cfg.token_cap, cfg.price_assumptions)87 self.cache = DecisionCache(cfg.cache_path) if (cfg.use_cache and cfg.cache_path) else None88 self._providers: dict[str, object] = {}89 self.records: list[LLMDecisionRecord] = []90 self.halted_for_budget = False9192 def provider(self, name: str):93 if name not in self._providers:94 if name == "mock":95 self._providers[name] = get_provider("mock", **self.cfg.mock_kwargs)96 elif name == "library":97 from .library import LibraryProvider98 self._providers[name] = LibraryProvider(self.cfg.library_path, seed=self.cfg.mock_kwargs.get("seed", 0))99 else:100 self._providers[name] = get_provider(name, timeout_s=self.cfg.timeout_s, max_retries=self.cfg.max_retries,101 **self.cfg.provider_kwargs)102 return self._providers[name]103104 # ---- one decision ------------------------------------------------------------------105 def _call(self, provider: str, model: str, prompt_id: str, system: str, user: str) -> tuple[ProviderResult, bool]:106 key = DecisionCache.key(provider, model, self.cfg.prompt_version, system, user)107 if self.cache is not None:108 hit = self.cache.get(key)109 if hit is not None:110 text, usage = hit111 return ProviderResult(text=text, model=model, raw=usage), True112 if self.cfg.frozen_only or self.halted_for_budget:113 return ProviderResult(text="", error="frozen_miss" if self.cfg.frozen_only else "budget_halt", model=model), False114 try:115 self.budget.check()116 except BudgetExceeded as e:117 self.halted_for_budget = True118 return ProviderResult(text="", error=str(e), model=model), False119 res = self.provider(provider.split('_')[0] if provider.startswith('openai') else provider).complete(system, user, model, reasoning_effort=self.cfg.reasoning_effort,120 temperature=self.cfg.temperature, max_output_tokens=self.cfg.max_output_tokens,121 cache_key=f"ma-{self.cfg.prompt_version}-{prompt_id}")122 if self.cache is not None and res.text and not res.error:123 self.cache.put(key, provider, model, self.cfg.prompt_version, prompt_id, system, user, res.text,124 {"input_tokens": res.input_tokens, "output_tokens": res.output_tokens, "cached": res.cached_input_tokens})125 return res, False126127 def decide_one(self, obs: Observation, provider: str, model: str, prompt_id: str) -> tuple[Action, LLMDecisionRecord]:128 system = system_prompt(prompt_id, obs, self.cfg.mu_cents, self.cfg.taker_bps, self.cfg.order_ttl, self.cfg.decision_interval)129 user = user_message(obs, 0, adaptive=self.cfg.adaptive_prompt)130 if provider == "library":131 mid_units = obs.books[0].mid * obs.tick_cents[0] / 100.0132 res, cached_hit = ProviderResult(text=self.provider("library").complete_obs(obs, mid_units), model=model), True133 else:134 res, cached_hit = self._call(provider, model, prompt_id, system, user)135 cost = self.budget.record(model, res.input_tokens, res.cached_input_tokens, res.output_tokens, res.reasoning_tokens,136 res.latency_s, cached_hit=cached_hit)137 if res.error:138 self.budget.usage.failures += 1139 dec, perr = (None, res.error) if res.error else parse_decision(res.text)140 action, valid, reason = self.to_action(dec, obs) if dec is not None else (Action(), False, perr)141 if not valid:142 self.budget.usage.invalid += 1143 import hashlib144 rec = LLMDecisionRecord(145 t=obs.t, agent_id=obs.agent_id, provider=provider, model=model, prompt_id=prompt_id, valid=valid, reason=reason,146 action=dec.action if dec else "hold", side=(dec.side or "") if dec else "", order_type=(dec.order_type or "") if dec else "",147 quantity=int(dec.quantity or 0) if dec else 0, limit_price=float(dec.limit_price or 0.0) if dec else 0.0,148 confidence=float(dec.confidence) if (dec and dec.confidence is not None) else float("nan"),149 rationale=dec.rationale if dec else "", input_tokens=res.input_tokens, cached_input_tokens=res.cached_input_tokens,150 output_tokens=res.output_tokens, reasoning_tokens=res.reasoning_tokens, latency_s=res.latency_s, usd=cost,151 cached_hit=cached_hit, error=res.error, prompt_sha=hashlib.sha256((system + user).encode()).hexdigest()[:16],152 request_id=res.request_id)153 from .library import features as _feat154 rec.features = json.dumps([round(float(x), 6) for x in _feat(obs)])155 if dec is not None and dec.action == "place" and dec.order_type == "limit" and dec.limit_price:156 rec.rel_price = float(dec.limit_price * 100.0 / obs.tick_cents[0] / obs.books[0].mid - 1.0)157 action.meta.update({"valid": valid, "reason": reason, "rationale": rec.rationale, "confidence": rec.confidence,158 "llm_action": rec.action, "provider": provider, "model": model})159 self.records.append(rec)160 return action, rec161162 def decide_batch(self, items: list[tuple[Observation, str, str, str]]) -> list[tuple[Action, LLMDecisionRecord]]:163 if not items:164 return []165 if len(items) == 1 or self.cfg.concurrency <= 1:166 return [self.decide_one(*it) for it in items]167 with ThreadPoolExecutor(max_workers=self.cfg.concurrency) as ex:168 return list(ex.map(lambda it: self.decide_one(*it), items))169170 # ---- validation --------------------------------------------------------------------171 def to_action(self, dec: Decision, obs: Observation) -> tuple[Action, bool, str]:172 a = 0173 b = obs.books[a]174 tc = obs.tick_cents[a]175 act = Action()176 if dec.action == "hold":177 return act, True, ""178 if dec.action == "cancel":179 mine = {o.order_id for o in obs.open_orders}180 ids = [i for i in dec.cancel_order_ids if i in mine]181 if not ids:182 return act, False, "cancel: no valid order ids"183 act.cancels = [CancelIntent(i) for i in ids]184 return act, True, ""185 # place186 if dec.side is None or dec.order_type is None or dec.quantity is None:187 return act, False, "place: missing side/order_type/quantity"188 q = int(dec.quantity)189 if q <= 0:190 return act, False, "place: quantity <= 0"191 if q > obs.max_order_qty:192 return act, False, f"place: quantity {q} > max {obs.max_order_qty}"193 side = Side.BUY if dec.side == "buy" else Side.SELL194 if b.halted:195 return act, False, "place: market halted"196 if dec.order_type == "limit":197 if dec.limit_price is None:198 return act, False, "place: limit without price"199 px = int(round(dec.limit_price * 100.0 / tc))200 if px <= 0 or abs(px - b.mid) > 0.2 * b.mid:201 return act, False, "place: limit price outside ±20% of mid"202 otype = OrderType.LIMIT203 else:204 px = None205 otype = OrderType.MARKET206 if side is Side.BUY and b.best_ask is None or side is Side.SELL and b.best_bid is None:207 return act, False, "place: market order with empty opposite side"208 if side is Side.BUY:209 ref = px if px is not None else (b.best_ask or b.mid)210 need = int(ref * tc * q * 1.001)211 if need > obs.free_cash:212 return act, False, f"place: cost {need / 100:.2f} > free cash {obs.free_cash / 100:.2f}"213 else:214 held = obs.positions.get(a, 0) - sum(o.remaining for o in obs.open_orders if o.side < 0)215 if q > held:216 if held <= 0:217 return act, False, "place: no free shares to sell"218 q = held # partial validity: clip to available (logged as valid but clipped)219 act.orders.append(OrderIntent(a, side, otype, q, px, tag="llm"))220 return act, True, ""221