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 price_assumptions: dict[str, dict[str, float]] = field(default_factory=dict)75 mock_kwargs: dict = field(default_factory=dict)76 provider_kwargs: dict = field(default_factory=dict)77 library_path: str | None = None787980class LLMPolicy:81 name = "llm"8283 def __init__(self, cfg: LLMPolicyConfig) -> None:84 self.cfg = cfg85 self.budget = BudgetTracker(cfg.usd_cap, cfg.token_cap, cfg.price_assumptions)86 self.cache = DecisionCache(cfg.cache_path) if (cfg.use_cache and cfg.cache_path) else None87 self._providers: dict[str, object] = {}88 self.records: list[LLMDecisionRecord] = []89 self.halted_for_budget = False9091 def provider(self, name: str):92 if name not in self._providers:93 if name == "mock":94 self._providers[name] = get_provider("mock", **self.cfg.mock_kwargs)95 elif name == "library":96 from .library import LibraryProvider97 self._providers[name] = LibraryProvider(self.cfg.library_path, seed=self.cfg.mock_kwargs.get("seed", 0))98 else:99 self._providers[name] = get_provider(name, timeout_s=self.cfg.timeout_s, max_retries=self.cfg.max_retries,100 **self.cfg.provider_kwargs)101 return self._providers[name]102103 # ---- one decision ------------------------------------------------------------------104 def _call(self, provider: str, model: str, prompt_id: str, system: str, user: str) -> tuple[ProviderResult, bool]:105 key = DecisionCache.key(provider, model, self.cfg.prompt_version, system, user)106 if self.cache is not None:107 hit = self.cache.get(key)108 if hit is not None:109 text, usage = hit110 return ProviderResult(text=text, model=model, raw=usage), True111 if self.cfg.frozen_only or self.halted_for_budget:112 return ProviderResult(text="", error="frozen_miss" if self.cfg.frozen_only else "budget_halt", model=model), False113 try:114 self.budget.check()115 except BudgetExceeded as e:116 self.halted_for_budget = True117 return ProviderResult(text="", error=str(e), model=model), False118 res = self.provider(provider.split('_')[0] if provider.startswith('openai') else provider).complete(system, user, model, reasoning_effort=self.cfg.reasoning_effort,119 temperature=self.cfg.temperature, max_output_tokens=self.cfg.max_output_tokens,120 cache_key=f"ma-{self.cfg.prompt_version}-{prompt_id}")121 if self.cache is not None and res.text and not res.error:122 self.cache.put(key, provider, model, self.cfg.prompt_version, prompt_id, system, user, res.text,123 {"input_tokens": res.input_tokens, "output_tokens": res.output_tokens, "cached": res.cached_input_tokens})124 return res, False125126 def decide_one(self, obs: Observation, provider: str, model: str, prompt_id: str) -> tuple[Action, LLMDecisionRecord]:127 system = system_prompt(prompt_id, obs, self.cfg.mu_cents, self.cfg.taker_bps, self.cfg.order_ttl)128 user = user_message(obs, 0, adaptive=self.cfg.adaptive_prompt)129 if provider == "library":130 mid_units = obs.books[0].mid * obs.tick_cents[0] / 100.0131 res, cached_hit = ProviderResult(text=self.provider("library").complete_obs(obs, mid_units), model=model), True132 else:133 res, cached_hit = self._call(provider, model, prompt_id, system, user)134 cost = self.budget.record(model, res.input_tokens, res.cached_input_tokens, res.output_tokens, res.reasoning_tokens,135 res.latency_s, cached_hit=cached_hit)136 if res.error:137 self.budget.usage.failures += 1138 dec, perr = (None, res.error) if res.error else parse_decision(res.text)139 action, valid, reason = self.to_action(dec, obs) if dec is not None else (Action(), False, perr)140 if not valid:141 self.budget.usage.invalid += 1142 import hashlib143 rec = LLMDecisionRecord(144 t=obs.t, agent_id=obs.agent_id, provider=provider, model=model, prompt_id=prompt_id, valid=valid, reason=reason,145 action=dec.action if dec else "hold", side=(dec.side or "") if dec else "", order_type=(dec.order_type or "") if dec else "",146 quantity=int(dec.quantity or 0) if dec else 0, limit_price=float(dec.limit_price or 0.0) if dec else 0.0,147 confidence=float(dec.confidence) if (dec and dec.confidence is not None) else float("nan"),148 rationale=dec.rationale if dec else "", input_tokens=res.input_tokens, cached_input_tokens=res.cached_input_tokens,149 output_tokens=res.output_tokens, reasoning_tokens=res.reasoning_tokens, latency_s=res.latency_s, usd=cost,150 cached_hit=cached_hit, error=res.error, prompt_sha=hashlib.sha256((system + user).encode()).hexdigest()[:16],151 request_id=res.request_id)152 from .library import features as _feat153 rec.features = json.dumps([round(float(x), 6) for x in _feat(obs)])154 if dec is not None and dec.action == "place" and dec.order_type == "limit" and dec.limit_price:155 rec.rel_price = float(dec.limit_price * 100.0 / obs.tick_cents[0] / obs.books[0].mid - 1.0)156 action.meta.update({"valid": valid, "reason": reason, "rationale": rec.rationale, "confidence": rec.confidence,157 "llm_action": rec.action, "provider": provider, "model": model})158 self.records.append(rec)159 return action, rec160161 def decide_batch(self, items: list[tuple[Observation, str, str, str]]) -> list[tuple[Action, LLMDecisionRecord]]:162 if not items:163 return []164 if len(items) == 1 or self.cfg.concurrency <= 1:165 return [self.decide_one(*it) for it in items]166 with ThreadPoolExecutor(max_workers=self.cfg.concurrency) as ex:167 return list(ex.map(lambda it: self.decide_one(*it), items))168169 # ---- validation --------------------------------------------------------------------170 def to_action(self, dec: Decision, obs: Observation) -> tuple[Action, bool, str]:171 a = 0172 b = obs.books[a]173 tc = obs.tick_cents[a]174 act = Action()175 if dec.action == "hold":176 return act, True, ""177 if dec.action == "cancel":178 mine = {o.order_id for o in obs.open_orders}179 ids = [i for i in dec.cancel_order_ids if i in mine]180 if not ids:181 return act, False, "cancel: no valid order ids"182 act.cancels = [CancelIntent(i) for i in ids]183 return act, True, ""184 # place185 if dec.side is None or dec.order_type is None or dec.quantity is None:186 return act, False, "place: missing side/order_type/quantity"187 q = int(dec.quantity)188 if q <= 0:189 return act, False, "place: quantity <= 0"190 if q > obs.max_order_qty:191 return act, False, f"place: quantity {q} > max {obs.max_order_qty}"192 side = Side.BUY if dec.side == "buy" else Side.SELL193 if b.halted:194 return act, False, "place: market halted"195 if dec.order_type == "limit":196 if dec.limit_price is None:197 return act, False, "place: limit without price"198 px = int(round(dec.limit_price * 100.0 / tc))199 if px <= 0 or abs(px - b.mid) > 0.2 * b.mid:200 return act, False, "place: limit price outside ±20% of mid"201 otype = OrderType.LIMIT202 else:203 px = None204 otype = OrderType.MARKET205 if side is Side.BUY and b.best_ask is None or side is Side.SELL and b.best_bid is None:206 return act, False, "place: market order with empty opposite side"207 if side is Side.BUY:208 ref = px if px is not None else (b.best_ask or b.mid)209 need = int(ref * tc * q * 1.001)210 if need > obs.free_cash:211 return act, False, f"place: cost {need / 100:.2f} > free cash {obs.free_cash / 100:.2f}"212 else:213 held = obs.positions.get(a, 0) - sum(o.remaining for o in obs.open_orders if o.side < 0)214 if q > held:215 if held <= 0:216 return act, False, "place: no free shares to sell"217 q = held # partial validity: clip to available (logged as valid but clipped)218 act.orders.append(OrderIntent(a, side, otype, q, px, tag="llm"))219 return act, True, ""220