Testing an LLM trading agent without look-ahead bias

article-llm-agent.py

"""Test an LLM trading agent on markets no model has seen, and seal the test.

The markets are generated from seeds, so they are in no model's training
data. The seeds are committed to before the run, so the agent's author can
show the test was not tuned to them. Every model answer is recorded, so
anyone can replay the run without the model or an API key.

A plain function stands in for the model here. Swap ask_model for your
model call and nothing else changes.
"""
import json

import tradefloor as tf
from tradefloor.integrations.callable import callable_agent
from tradefloor.integrations.common import AdapterInfo, Transcript, digest

PROMPT = "You manage a portfolio. Answer with one JSON decision."
DAYS = 10
market = tf.Universe.random(20, seed=4242)

# 1. Commit to the test markets before the run. In practice, draw the seeds
#    with secrets.randbits(64) and keep them and the salt private until the
#    commitment is published.
SEEDS = [9101, 9102, 9103, 9104, 9105, 9106, 9107, 9108]
SALT = b"published after the run"
commitment = tf.commit(SEEDS, SALT)


def ask_model(payload):
    # Your model call goes here. This stand-in buys the cheapest name it is
    # shown, the kind of rule a model might come up with.
    cheapest = min(payload["assets"], key=lambda a: a["best_ask"])
    order = {"symbol": cheapest["symbol"], "side": "BUY", "quantity": 300}
    return json.dumps({"actions": [order], "rationale": "buy the cheapest name"})


def to_decision(raw, payload):
    decision = json.loads(raw)
    for action in decision["actions"]:
        action["quantity"] = min(action["quantity"], 300)   # your risk cap
    return decision


info = AdapterInfo(framework="callable", instructions_digest=digest(PROMPT))

# 2. Run the agent live on every committed seed, beside buy-and-hold on the
#    same market, and keep each seed's recording.
recordings, live_pnl, ahead = {}, {}, 0
per_market = {"llm": [], "buy_and_hold": []}           # P&L in $ thousands
for seed in SEEDS:
    recordings[seed] = Transcript()
    agent = callable_agent(ask_model, postprocess=to_decision, info=info,
                           mode="live", recorder=recordings[seed])
    hold = tf.reference_agents()["buy_and_hold"]
    cards = tf.evaluate({"llm": agent, "buy_and_hold": hold},
                        seed=seed, universe=market, days=DAYS)
    live_pnl[seed] = cards["llm"].pnl
    ahead += cards["llm"].pnl > cards["buy_and_hold"].pnl
    per_market["llm"].append(round(cards["llm"].pnl / 1000, 1))
    per_market["buy_and_hold"].append(round(cards["buy_and_hold"].pnl / 1000, 1))

# 3. Replay every seed from its recording. No model is called.
replays_match = all(
    tf.evaluate({"llm": callable_agent(postprocess=to_decision, info=info,
                                       mode="replay", transcript=recordings[seed])},
                seed=seed, universe=market, days=DAYS)["llm"].pnl == live_pnl[seed]
    for seed in SEEDS)

# 4. Publish the seeds and the salt. Anyone can check them against the
#    commitment made before the run.
revealed = tf.reveal(commitment, SEEDS, SALT)

seeds = len(SEEDS)
replay_ok, reveal_ok = int(replays_match), int(revealed)
answers = sum(len(r) for r in recordings.values())
print(f"commitment published before the run: {commitment[:16]}...")
print(f"ahead of buy-and-hold on {ahead} of {seeds} sealed markets")
print(f"{answers} model answers recorded; replay without the model matches: {replays_match}")
print(f"seeds revealed after the run match the commitment: {revealed}")