We are building a real-time market-monitoring agent that ingests a stream of simulated price ticks, evaluates them against a watchlist, and emits structured trade alerts. This pattern applies anywhere you need an LLM to reason over live events, from IoT sensors to application logs. I run it on Oxlo.ai because its request-based pricing removes the token anxiety that usually comes with stuffing large rolling windows into a prompt every few seconds.
What you'll need
- Python 3.10 or newer
- The OpenAI SDK:
pip install openai - An Oxlo.ai API key from https://portal.oxlo.ai
I will use llama-3.3-70b because it is fast and handles structured tool calls well, but Oxlo.ai also offers qwen-3-32b and kimi-k2.6 if you need deeper reasoning. Because Oxlo.ai charges one flat rate per request, you can stuff a large rolling window of ticks into the prompt without watching token meters climb. That matters when you are firing a request every few seconds. See https://oxlo.ai/pricing for plan details.
Step 1: Set up the Oxlo.ai client
We only need the standard OpenAI SDK pointed at Oxlo.ai. I keep my key in an environment variable so it never sits in source control.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
print("Client ready:", client.base_url)
Step 2: Simulate a live tick stream
Rather than pull live market data, I built a small generator that yields realistic ticks and injects a volatility spike every so often so we have something to detect.
import random
import time
from dataclasses import dataclass
@dataclass
class Tick:
symbol: str
price: float
volume: int
ts: float
def tick_stream(symbols=("AAPL", "TSLA", "NVDA"), interval=0.5):
prices = {s: random.uniform(150, 600) for s in symbols}
while True:
symbol = random.choice(symbols)
jump = random.choice([0.98, 1.0, 1.0, 1.0, 1.02, 1.05])
prices[symbol] = round(prices[symbol] * jump, 2)
tick = Tick(symbol, prices[symbol], random.randint(100, 10000), time.time())
yield tick
time.sleep(interval)
Step 3: Define the agent prompt and tool schema
The agent receives a block of recent ticks and decides whether to alert. I force it to reply via function calling so the downstream code stays deterministic.
SYSTEM_PROMPT = """You are a real-time market monitor.
Analyze the provided tick window and decide if any symbol shows unusual activity that warrants an alert.
Emit an alert only if a price moved more than 2% within the window or if volume is anomalously high.
Respond with the alert_trade tool. If nothing is notable, set alert to false."""TOOLS = [
{
"type": "function",
"function": {
"name": "alert_trade",
"description": "Emit a structured alert",
"parameters": {
"type": "object",
"properties": {
"symbol": {"type": "string"},
"alert": {"type": "boolean"},
"reason": {"type": "string"},
"suggested_action": {"enum": ["BUY", "SELL", "HOLD", "WATCH"], "type": "string"}
},
"required": ["symbol", "alert", "reason", "suggested_action"]
}
}
}
]Step 4: Build the analysis worker
I batch ticks into a rolling window and hand them to a worker thread. Oxlo.ai has no cold starts on popular models, so the first inference call returns as fast as the rest, which matters when the agent starts up mid-stream.
import json
import queue
import threading
import time
class RealTimeAgent:
def __init__(self, client, max_queue=2):
self.client = client
self.q = queue.Queue(maxsize=max_queue)
self.worker = threading.Thread(target=self._loop, daemon=True)
self._stop = False
def start(self):
self.worker.start()
def push(self, ticks_batch):
try:
self.q.put_nowait(ticks_batch)
except queue.Full:
print("Backpressure: skipping a window")
def _loop(self):
while not self._stop:
try:
batch = self.q.get(timeout=1)
except queue.Empty:
continue
self._analyze(batch)
def _analyze(self, batch):
payload = "\n".join(
f"{t.symbol} | ${t.price} | vol {t.volume}" for t in batch
)
try:
response = self.client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Ticks:\n{payload}"},
],
tools=TOOLS,
tool_choice={"type": "function", "function": {"name": "alert_trade"}},
temperature=0.1,
)
tool_call = response.choices[0].message.tool_calls[0]
result = json.loads(tool_call.function.arguments)
if result.get("alert"):
print(f"ALERT: {result}")
else:
print(f"quiet: {result['symbol']} {result['suggested_action']}")
except Exception as e:
print("Analysis failed:", e)
def stop(self):
self._stop = True
self.worker.join(timeout=5)
Step 5: Wire the producer loop
The producer gathers ticks into a small buffer and flushes it to the agent every four seconds or when eight ticks are ready. Keeping the buffer small limits latency, while the flat per-request cost on Oxlo.ai means I do not have to truncate price history to save tokens.
def run_agent(duration_seconds=20):
agent = RealTimeAgent(client)
agent.start()
stream = tick_stream()
buffer = []
last_flush = time.time()
start = time.time()
for tick in stream:
buffer.append(tick)
if len(buffer) >= 8 or (time.time() - last_flush) >= 4:
agent.push(buffer)
buffer = []
last_flush = time.time()
if time.time() - start > duration_seconds:
break
time.sleep(3)
agent.stop()
print("Done")
if __name__ == "__main__":
run_agent()
Step 6: Run it
Run the script and watch stdout. Most windows return HOLD, but injected volatility should trigger a WATCH or SELL alert.
$ python agent.py
Client ready: https://api.oxlo.ai/v1
quiet: AAPL HOLD
quiet: TSLA HOLD
ALERT: {'symbol': 'NVDA', 'alert': True, 'reason': 'Price jumped 5.2% in window', 'suggested_action': 'WATCH'}
quiet: NVDA WATCH
Backpressure: skipping a window
quiet: AAPL HOLD
ALERT: {'symbol': 'TSLA', 'alert': True, 'reason': 'Volume spike detected', 'suggested_action': 'SELL'}
DoneNext steps
Swap the simulator for a real WebSocket feed and add Pydantic validation on the tool-call arguments. If you need deeper reasoning over larger windows, try qwen-3-32b or kimi-k2.6 on Oxlo.ai without worrying about token costs scaling with window size.
Top comments (0)