"""LLMPolicy: Observation -> prompt -> provider -> validated Decision -> Action. Validation is local and strict: quantities within [1, max_order_qty], limit prices within ±20% of the mid and on the tick grid, cash/holdings feasibility, cancel ids owned by the agent. Invalid decisions become `hold` with meta.valid=False and a reason; they are counted in the reliability statistics. Decisions are made in batches (one per decision round) with a bounded thread pool.""" from __future__ import annotations import json import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field import numpy as np from ..agents.base import Observation from ..exchange.orders import Action, CancelIntent, OrderIntent, OrderType, Side from .budget import BudgetExceeded, BudgetTracker from .cache import DecisionCache from .prompting import PROMPT_VERSION, system_prompt, user_message from .providers import ProviderResult, get_provider from .schema import Decision, parse_decision @dataclass(slots=True) class LLMDecisionRecord: t: int agent_id: int provider: str model: str prompt_id: str valid: bool reason: str action: str side: str order_type: str quantity: int limit_price: float confidence: float rationale: str input_tokens: int cached_input_tokens: int output_tokens: int reasoning_tokens: int latency_s: float usd: float cached_hit: bool error: str prompt_sha: str request_id: str = "" features: str = "" # JSON feature vector (for the frozen library) rel_price: float = float("nan") @dataclass class LLMPolicyConfig: models: dict[str, str] prompt_version: str = PROMPT_VERSION reasoning_effort: str | None = "low" temperature: float | None = None max_output_tokens: int = 1200 concurrency: int = 8 timeout_s: float = 60.0 max_retries: int = 3 usd_cap: float = 5.0 token_cap: int | None = None cache_path: str | None = None use_cache: bool = True frozen_only: bool = False # never call a provider; cache misses -> hold (frozen library replay) adaptive_prompt: bool = False mu_cents: float = 50.0 taker_bps: float = 2.0 order_ttl: int | None = 50 decision_interval: int = 20 price_assumptions: dict[str, dict[str, float]] = field(default_factory=dict) mock_kwargs: dict = field(default_factory=dict) provider_kwargs: dict = field(default_factory=dict) library_path: str | None = None class LLMPolicy: name = "llm" def __init__(self, cfg: LLMPolicyConfig) -> None: self.cfg = cfg self.budget = BudgetTracker(cfg.usd_cap, cfg.token_cap, cfg.price_assumptions) self.cache = DecisionCache(cfg.cache_path) if (cfg.use_cache and cfg.cache_path) else None self._providers: dict[str, object] = {} self.records: list[LLMDecisionRecord] = [] self.halted_for_budget = False def provider(self, name: str): if name not in self._providers: if name == "mock": self._providers[name] = get_provider("mock", **self.cfg.mock_kwargs) elif name == "library": from .library import LibraryProvider self._providers[name] = LibraryProvider(self.cfg.library_path, seed=self.cfg.mock_kwargs.get("seed", 0)) else: self._providers[name] = get_provider(name, timeout_s=self.cfg.timeout_s, max_retries=self.cfg.max_retries, **self.cfg.provider_kwargs) return self._providers[name] # ---- one decision ------------------------------------------------------------------ def _call(self, provider: str, model: str, prompt_id: str, system: str, user: str) -> tuple[ProviderResult, bool]: key = DecisionCache.key(provider, model, self.cfg.prompt_version, system, user) if self.cache is not None: hit = self.cache.get(key) if hit is not None: text, usage = hit return ProviderResult(text=text, model=model, raw=usage), True if self.cfg.frozen_only or self.halted_for_budget: return ProviderResult(text="", error="frozen_miss" if self.cfg.frozen_only else "budget_halt", model=model), False try: self.budget.check() except BudgetExceeded as e: self.halted_for_budget = True return ProviderResult(text="", error=str(e), model=model), False res = self.provider(provider.split('_')[0] if provider.startswith('openai') else provider).complete(system, user, model, reasoning_effort=self.cfg.reasoning_effort, temperature=self.cfg.temperature, max_output_tokens=self.cfg.max_output_tokens, cache_key=f"ma-{self.cfg.prompt_version}-{prompt_id}") if self.cache is not None and res.text and not res.error: self.cache.put(key, provider, model, self.cfg.prompt_version, prompt_id, system, user, res.text, {"input_tokens": res.input_tokens, "output_tokens": res.output_tokens, "cached": res.cached_input_tokens}) return res, False def decide_one(self, obs: Observation, provider: str, model: str, prompt_id: str) -> tuple[Action, LLMDecisionRecord]: system = system_prompt(prompt_id, obs, self.cfg.mu_cents, self.cfg.taker_bps, self.cfg.order_ttl, self.cfg.decision_interval) user = user_message(obs, 0, adaptive=self.cfg.adaptive_prompt) if provider == "library": mid_units = obs.books[0].mid * obs.tick_cents[0] / 100.0 res, cached_hit = ProviderResult(text=self.provider("library").complete_obs(obs, mid_units), model=model), True else: res, cached_hit = self._call(provider, model, prompt_id, system, user) cost = self.budget.record(model, res.input_tokens, res.cached_input_tokens, res.output_tokens, res.reasoning_tokens, res.latency_s, cached_hit=cached_hit) if res.error: self.budget.usage.failures += 1 dec, perr = (None, res.error) if res.error else parse_decision(res.text) action, valid, reason = self.to_action(dec, obs) if dec is not None else (Action(), False, perr) if not valid: self.budget.usage.invalid += 1 import hashlib rec = LLMDecisionRecord( t=obs.t, agent_id=obs.agent_id, provider=provider, model=model, prompt_id=prompt_id, valid=valid, reason=reason, action=dec.action if dec else "hold", side=(dec.side or "") if dec else "", order_type=(dec.order_type or "") if dec else "", quantity=int(dec.quantity or 0) if dec else 0, limit_price=float(dec.limit_price or 0.0) if dec else 0.0, confidence=float(dec.confidence) if (dec and dec.confidence is not None) else float("nan"), rationale=dec.rationale if dec else "", input_tokens=res.input_tokens, cached_input_tokens=res.cached_input_tokens, output_tokens=res.output_tokens, reasoning_tokens=res.reasoning_tokens, latency_s=res.latency_s, usd=cost, cached_hit=cached_hit, error=res.error, prompt_sha=hashlib.sha256((system + user).encode()).hexdigest()[:16], request_id=res.request_id) from .library import features as _feat rec.features = json.dumps([round(float(x), 6) for x in _feat(obs)]) if dec is not None and dec.action == "place" and dec.order_type == "limit" and dec.limit_price: rec.rel_price = float(dec.limit_price * 100.0 / obs.tick_cents[0] / obs.books[0].mid - 1.0) action.meta.update({"valid": valid, "reason": reason, "rationale": rec.rationale, "confidence": rec.confidence, "llm_action": rec.action, "provider": provider, "model": model}) self.records.append(rec) return action, rec def decide_batch(self, items: list[tuple[Observation, str, str, str]]) -> list[tuple[Action, LLMDecisionRecord]]: if not items: return [] if len(items) == 1 or self.cfg.concurrency <= 1: return [self.decide_one(*it) for it in items] with ThreadPoolExecutor(max_workers=self.cfg.concurrency) as ex: return list(ex.map(lambda it: self.decide_one(*it), items)) # ---- validation -------------------------------------------------------------------- def to_action(self, dec: Decision, obs: Observation) -> tuple[Action, bool, str]: a = 0 b = obs.books[a] tc = obs.tick_cents[a] act = Action() if dec.action == "hold": return act, True, "" if dec.action == "cancel": mine = {o.order_id for o in obs.open_orders} ids = [i for i in dec.cancel_order_ids if i in mine] if not ids: return act, False, "cancel: no valid order ids" act.cancels = [CancelIntent(i) for i in ids] return act, True, "" # place if dec.side is None or dec.order_type is None or dec.quantity is None: return act, False, "place: missing side/order_type/quantity" q = int(dec.quantity) if q <= 0: return act, False, "place: quantity <= 0" if q > obs.max_order_qty: return act, False, f"place: quantity {q} > max {obs.max_order_qty}" side = Side.BUY if dec.side == "buy" else Side.SELL if b.halted: return act, False, "place: market halted" if dec.order_type == "limit": if dec.limit_price is None: return act, False, "place: limit without price" px = int(round(dec.limit_price * 100.0 / tc)) if px <= 0 or abs(px - b.mid) > 0.2 * b.mid: return act, False, "place: limit price outside ±20% of mid" otype = OrderType.LIMIT else: px = None otype = OrderType.MARKET if side is Side.BUY and b.best_ask is None or side is Side.SELL and b.best_bid is None: return act, False, "place: market order with empty opposite side" if side is Side.BUY: ref = px if px is not None else (b.best_ask or b.mid) need = int(ref * tc * q * 1.001) if need > obs.free_cash: return act, False, f"place: cost {need / 100:.2f} > free cash {obs.free_cash / 100:.2f}" else: held = obs.positions.get(a, 0) - sum(o.remaining for o in obs.open_orders if o.side < 0) if q > held: if held <= 0: return act, False, "place: no free shares to sell" q = held # partial validity: clip to available (logged as valid but clipped) act.orders.append(OrderIntent(a, side, otype, q, px, tag="llm")) return act, True, ""