"""Observation -> prompt. One frozen system prompt per (prompt_version, prompt_id) and a canonical, deterministic user message built from the Observation. Nothing in the prompt comes from latent state.""" from __future__ import annotations from ..agents.base import Observation PROMPT_VERSION = "v1" ROLE_PERSONAS: dict[str, str] = { "generalist": "You are a professional trader managing your own account in a single-stock electronic market.", "value": "You are a value investor: you compare the price to the present value of expected dividends and trade mispricings patiently.", "momentum": "You are a momentum trader: you follow recent price trends and cut positions when the trend reverses.", "contrarian": "You are a contrarian trader: you fade sharp moves that are not justified by dividend news.", "liquidity": "You are a liquidity provider: you prefer resting limit orders near the best quotes and keep your inventory small.", "risk_averse": "You are a risk-averse trader: you keep positions small, avoid market orders and prefer to hold when uncertain.", "macro": "You are a news-driven trader: you weigh public announcements and dividend surprises and act when the market has not yet reacted.", "quant": "You are a quantitative trader: you reason about expected value per share, spread costs and inventory before each order.", } RULES = """ Market mechanics (read carefully): - One risky stock paying a cash dividend every {div_interval} ticks; the long-run mean dividend is {mu:.2f} per share per period and the per-period discount rate is {r_pct:.2f}%, so a share paying the mean dividend forever is worth about {pv_mean:.2f}. - Dividends deviate persistently from the mean (autocorrelation {rho:.2f} per period). The deviation of the NEXT dividend is not yet announced; public news and, when you have one, your private signal are noisy estimates of it (in cents). The present value of a share rises by about 1/(1+r-rho) ≈ 44 cents per cent of next-dividend deviation. - Prices are in currency units with a tick of {tick:.2f}. Limit orders rest in the book; market orders execute immediately against the best quotes (taker fee {taker_bps:.1f} bps). Orders may be partially filled. Unfilled limit orders expire after {ttl} ticks. - You cannot short sell and cannot spend more than your free cash. Maximum order size: {max_qty} shares. - You decide every {interval} ticks. Between decisions the market keeps trading. Respond with ONLY a JSON object with exactly these keys: {{"action": "hold"|"place"|"cancel", "side": "buy"|"sell"|null, "order_type": "limit"|"market"|null, "quantity": integer|null, "limit_price": number|null, "cancel_order_ids": [integers], "confidence": number in [0,1]|null, "rationale": "one sentence, max 200 characters"}} "place" requires side, order_type and quantity (and limit_price for a limit order). "cancel" requires cancel_order_ids. Anything else is treated as hold. """.strip() def system_prompt(prompt_id: str, obs: Observation, mu_cents: float, taker_bps: float, ttl: int | None, decision_interval: int | None = None) -> str: persona = ROLE_PERSONAS.get(prompt_id, ROLE_PERSONAS["generalist"]) tick = obs.tick_cents[0] / 100.0 pv_mean = (mu_cents / obs.r_period) / 100.0 return persona + "\n\n" + RULES.format( div_interval=obs.div_interval, mu=mu_cents / 100.0, r_pct=obs.r_period * 100.0, pv_mean=pv_mean, rho=0.98, tick=tick, taker_bps=taker_bps, ttl=ttl if ttl is not None else 10_000, max_qty=obs.max_order_qty, interval=max(1, decision_interval or obs.div_interval)) def user_message(obs: Observation, asset: int = 0, adaptive: bool = False) -> str: b = obs.books[asset] tc = obs.tick_cents[asset] u = lambda ticks: f"{ticks * tc / 100.0:.2f}" # noqa: E731 lines = [f"Time: tick {obs.t}. Next dividend in {obs.ticks_to_dividend} ticks."] if b.halted: lines.append("TRADING IS HALTED on this stock (orders will be rejected).") bb = u(b.best_bid) if b.best_bid is not None else "none" ba = u(b.best_ask) if b.best_ask is not None else "none" lines.append(f"Best bid {bb} / best ask {ba} / mid {u(b.mid)} / last trade {u(b.last_price) if b.last_price else 'none'}.") depth_b = ", ".join(f"{u(p)}x{q}" for p, q in b.bids[:5]) or "empty" depth_a = ", ".join(f"{u(p)}x{q}" for p, q in b.asks[:5]) or "empty" lines.append(f"Bid depth (price x shares): {depth_b}. Ask depth: {depth_a}.") h = obs.mid_history[asset] if len(h) >= 2: pts = [u(x) for x in h[::-max(1, len(h) // 10)][::-1]] lines.append(f"Mid-price history (oldest to newest, {len(h)} ticks sampled): " + ", ".join(pts) + ".") r = (h[-1] / h[0] - 1.0) * 100.0 lines.append(f"Change over the window: {r:+.2f}%. Recent volatility (std of tick returns): {100 * (h[1:] / h[:-1] - 1).std():.3f}%.") tr = obs.recent_trades.get(asset, []) if tr: vol = sum(q for _, _, q, _ in tr) buyv = sum(q for _, _, q, ag in tr if ag > 0) lines.append(f"Last {len(tr)} trades: volume {vol} shares, {100 * buyv / max(vol, 1):.0f}% buyer-initiated.") lines.append(f"Last dividend paid: {obs.last_dividend[asset] / 100.0:.2f} per share. Public estimate of the next dividend's deviation " f"from its mean: public estimate {obs.x_hat_public.get(asset, 0.0):+.1f} cents. " f"Present value of expected dividends using only public information: {obs.public_value[asset] / 100.0:.2f}.") if obs.news: for n in obs.news[-3:]: lines.append(f"Public news (tick {n.time}): {n.text}") if asset in obs.private_signals: lines.append(f"Your private signal: next dividend deviation estimated at {obs.private_signals[asset]:+.1f} cents " f"(signal precision {obs.signal_precision.get(asset, 0):.1f}; higher is more reliable).") else: lines.append("You have no private signal this period.") lines.append(f"Your account: cash {obs.cash / 100.0:.2f} (free {obs.free_cash / 100.0:.2f}), " f"holdings {obs.positions.get(asset, 0)} shares, equity {obs.equity / 100.0:.2f}.") if obs.open_orders: oo = "; ".join(f"id {o.order_id}: {'buy' if o.side > 0 else 'sell'} {o.remaining}/{o.qty} @ {u(o.price) if o.price else 'mkt'} (age {o.age:.0f})" for o in obs.open_orders[:6]) lines.append(f"Your open orders: {oo}.") else: lines.append("You have no open orders.") if adaptive and obs.realized_pnl_history: lines.append(f"Feedback: your equity changed by {obs.realized_pnl_history[-1]:+.2f}% since your last decision " f"(cumulative {sum(obs.realized_pnl_history):+.2f}%).") if obs.memory: lines.append("Your recent decisions and outcomes:") lines += [f" - {m}" for m in obs.memory] if obs.channel: lines.append("Messages from other traders in your group (unverified opinions):") lines += [f" - {m}" for m in obs.channel] lines.append("Decide now. Reply with the JSON object only.") return "\n".join(lines)