An AI agent that trades fake money and remembers.

# The 6 scheduled runs

Each run is a fresh agent with no memory. The prompt below is the same for all six, only the run name changes.
schedule.sh passes this whole file to your agent CLI, with the run name filled in. Times are US market hours (Eastern).

| Run | Eastern time | Job |
|---|---|---|
| premarket | 9:00 | Read the news on the watch list, check the account, build today's watch list. No orders before 9:30. |
| open | 9:45 | Look for entries that match the strategy. |
| midday | 11:30 | Review open positions against their stops and targets. |
| afternoon | 13:30 | Same review, plus any new entries that still fit. |
| late | 15:00 | Tighten stops, no new positions after this run. |
| close | 15:50 | Close anything the strategy says not to hold overnight, then write the day's handoff. |

## The prompt (every run)

You are run "RUN" of a paper-trading desk. You remember nothing from earlier runs. Paper money only.

1. Run `python3 ~/.claude/skills/paper-desk/papertrade.py brief` and read every line. That is your memory.
2. Read STRATEGY.md in ~/.paper-desk (the human's rules). If it is missing, stop and write a handoff that says so.
3. Run `python3 ~/.claude/skills/paper-desk/papertrade.py status`.
4. Do this run's job from the table. Every order goes through papertrade.py with a real --why. Never call a broker any other way.
5. If anything looks wrong (handoff older than 30 hours, a position you can't explain, an error), place no orders and say so in the handoff.
6. Last step, always, even if you did nothing:
   `python3 ~/.claude/skills/paper-desk/papertrade.py handoff --run RUN --did "<what you did and why>" --next "<exactly what the next run should check>" --watch "<tickers>"`

Write the handoff for a stranger. The next run only knows what you put in it.

#!/usr/bin/env python3
"""Paper-trading desk for an AI agent that wakes up on a schedule and remembers nothing.

Every run starts with `brief` (read the handoff), and ends with `handoff` (write it).
Paper money only. It refuses live trading endpoints on purpose.

    python3 papertrade.py selftest                         # offline check, no account needed
    python3 papertrade.py init --cash 10000                # start a local paper ledger (no broker)
    python3 papertrade.py brief                            # what the last run left for this one
    python3 papertrade.py quote AAPL                       # needs Alpaca paper keys (or pass --price)
    python3 papertrade.py buy AAPL 5 --why "breakout above yesterday's high" [--price 231.10]
    python3 papertrade.py sell AAPL 5 --why "stop hit"
    python3 papertrade.py status                           # cash, positions, P&L
    python3 papertrade.py handoff --did "..." --next "..." [--watch "AAPL,MSFT"]
    python3 papertrade.py report                           # one line per day from the journal

Broker: set ALPACA_KEY and ALPACA_SECRET from a free Alpaca PAPER account to trade through
Alpaca's paper API (https://paper-api.alpaca.markets), in your shell or in $PAPER_DIR/.env so
scheduled runs see them too. Without keys it keeps its own ledger
and you pass --price. Data lives in $PAPER_DIR (default ~/.paper-desk).
Rules it enforces (edit RULES below): max 20% of the account in one name, max 6 orders a day,
no shorting, no repeat of the same order on the same day (unless --again),
no orders when PAUSE exists in the data folder.
Python 3.8+, no extra packages. Not financial advice: it's a sandbox for testing agents.
"""
import argparse, contextlib, datetime as dt, fcntl, json, math, os, sys, tempfile, urllib.request

RULES = {"max_position_pct": 20.0, "max_orders_per_day": 6}
PAPER_URL = "https://paper-api.alpaca.markets"
DATA_URL = "https://data.alpaca.markets"
D = os.path.expanduser(os.environ.get("PAPER_DIR", "~/.paper-desk"))


def p(name): return os.path.join(D, name)


def load_env():
    """Scheduled runs don't see your shell's variables, so keys can live in $PAPER_DIR/.env (KEY=value lines)."""
    try:
        for line in open(p(".env")):
            k, _, v = line.strip().partition("=")
            if k and not k.startswith("#") and v:
                os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
    except FileNotFoundError:
        pass
def now(): return dt.datetime.now().astimezone().isoformat(timespec="seconds")
def today(): return dt.date.today().isoformat()


def load(name, default):
    try:
        return json.load(open(p(name)))
    except FileNotFoundError:
        return default


def save(name, obj):
    os.makedirs(D, exist_ok=True)
    tmp = p(name) + ".tmp"
    json.dump(obj, open(tmp, "w"), indent=2)
    os.replace(tmp, p(name))


def journal(kind, **kw):
    os.makedirs(D, exist_ok=True)
    with open(p("journal.jsonl"), "a") as f:
        f.write(json.dumps(dict(ts=now(), kind=kind, **kw)) + "\n")


def keys():
    load_env()
    k, s = os.environ.get("ALPACA_KEY"), os.environ.get("ALPACA_SECRET")
    base = os.environ.get("ALPACA_BASE", PAPER_URL).rstrip("/")
    if base != PAPER_URL:
        sys.exit(f"Refusing: ALPACA_BASE must be exactly {PAPER_URL}. This tool is paper money only.")
    return (k, s, base) if k and s else None


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *a, **kw):
        raise SystemExit("Refusing: the broker tried to redirect. Nothing was sent.")


OPENER = urllib.request.build_opener(NoRedirect)


def api(method, url, body=None):
    if not (url.startswith(PAPER_URL + "/") or url.startswith(DATA_URL + "/")):
        sys.exit(f"Refusing: {url} is not Alpaca's paper or market data API.")
    k, s, _ = keys()
    req = urllib.request.Request(url, method=method, data=json.dumps(body).encode() if body else None,
                                 headers={"APCA-API-KEY-ID": k, "APCA-API-SECRET-KEY": s, "Content-Type": "application/json"})
    with OPENER.open(req, timeout=20) as r:
        return json.loads(r.read().decode() or "{}")


def positive(x, what):
    x = float(x)
    if not math.isfinite(x) or x <= 0:
        sys.exit(f"Refusing: {what} has to be a number above 0 (got {x}).")
    return x


def quote(sym, price=None):
    if price is not None:
        return positive(price, "--price")
    if not keys():
        sys.exit("No price: pass --price, or set ALPACA_KEY / ALPACA_SECRET from a free Alpaca paper account.")
    q = api("GET", f"{DATA_URL}/v2/stocks/{sym.upper()}/trades/latest")
    return positive(q["trade"]["p"], f"the quote for {sym}")


def ledger():
    return load("ledger.json", None)


def account():
    """Cash, positions {sym: {qty, avg}} and equity, from Alpaca paper or the local ledger."""
    if keys():
        a = api("GET", f"{keys()[2]}/v2/account")
        pos = {x["symbol"]: dict(qty=float(x["qty"]), avg=float(x["avg_entry_price"]), last=float(x["current_price"]))
               for x in api("GET", f"{keys()[2]}/v2/positions")}
        st = load("broker_start.json", None)
        if not st:
            st = dict(equity=float(a["equity"]), ts=now())
            save("broker_start.json", st)
        return dict(cash=float(a["cash"]), positions=pos, equity=float(a["equity"]), start=st["equity"], broker="alpaca-paper")
    L = ledger()
    if not L:
        sys.exit("No ledger yet. Run: python3 papertrade.py init --cash 10000")
    eq = L["cash"] + sum(v["qty"] * v.get("last", v["avg"]) for v in L["positions"].values())
    return dict(cash=L["cash"], positions=L["positions"], equity=eq, start=L["start"], broker="local-ledger")


def orders_today():
    """Counts order intents, written before anything is sent, so a crash mid-order still counts."""
    n = 0
    if os.path.exists(p("journal.jsonl")):
        for line in open(p("journal.jsonl")):
            j = json.loads(line)
            n += j["kind"] == "intent" and j["ts"].startswith(today())
    return n


def pending(sym):
    """Open (unfilled) Alpaca paper orders for sym, so caps and repeat checks see them too."""
    if not keys():
        return []
    return [o for o in api("GET", f"{PAPER_URL}/v2/orders?status=open&symbols={sym}") if o.get("symbol") == sym]


def left(o):
    """Shares an open order still has to fill (partial fills count only what's left)."""
    return max(float(o.get("qty") or 0) - float(o.get("filled_qty") or 0), 0.0)


@contextlib.contextmanager
def desk_lock():
    """One order at a time, even if two runs overlap."""
    os.makedirs(D, exist_ok=True)
    with open(p(".lock"), "w") as f:
        fcntl.flock(f, fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(f, fcntl.LOCK_UN)


def same_order_today(side, sym, qty):
    if os.path.exists(p("journal.jsonl")):
        for line in open(p("journal.jsonl")):
            j = json.loads(line)
            if j["kind"] == "intent" and j["ts"].startswith(today()) and (j["side"], j["sym"], j["qty"]) == (side, sym, qty):
                return j
    return None


def trade(side, sym, qty, why, price=None, again=False):
    with desk_lock():
        _trade(side, sym, qty, why, price, again)


def _trade(side, sym, qty, why, price=None, again=False):
    sym = sym.upper()
    qty = positive(qty, "the quantity")
    if keys() and price is not None:
        sys.exit("Refusing --price with a broker connected: the cap check has to use the broker's own quote.")
    if os.path.exists(p("PAUSE")):
        sys.exit("PAUSE file found: no orders. Delete it to resume.")
    if not why:
        sys.exit("Every order needs --why. The next run only knows what you write down.")
    if orders_today() >= RULES["max_orders_per_day"]:
        sys.exit(f"Blocked: {RULES['max_orders_per_day']} orders already today.")
    dup = same_order_today(side, sym, qty)
    if dup and not again:
        sys.exit(f"Blocked: the same order ({side} {qty:g} {sym}) already went in today at {dup['ts'][11:16]}. "
                 "A run that crashed and restarted does this. Check status first, add --again if you really mean it.")
    px = quote(sym, price)
    acct = account()
    held = acct["positions"].get(sym, {}).get("qty", 0)
    open_orders = pending(sym)
    if any(o.get("side") == side and left(o) == qty for o in open_orders) and not again:
        sys.exit(f"Blocked: an identical {side} {qty:g} {sym} order is still open at the broker. Check it before sending another.")
    pend_buy = sum(left(o) for o in open_orders if o.get("side") == "buy")
    pend_sell = sum(left(o) for o in open_orders if o.get("side") == "sell")
    held_after_pending = held - pend_sell
    if side == "sell" and qty > held_after_pending:
        sys.exit(f"Blocked: selling {qty:g} {sym} but only {held_after_pending:g} held and not already being sold (no shorting).")
    if side == "buy":
        after = (held + pend_buy + qty) * px
        if after > acct["equity"] * RULES["max_position_pct"] / 100:
            sys.exit(f"Blocked: {sym} would be {after / acct['equity'] * 100:.1f}% of the account (max {RULES['max_position_pct']:g}%).")
        if qty * px > acct["cash"]:
            sys.exit(f"Blocked: costs {qty * px:,.2f}, cash is {acct['cash']:,.2f}.")
    journal("intent", side=side, sym=sym, qty=qty, price=px, why=why)
    if keys():
        o = api("POST", f"{keys()[2]}/v2/orders", dict(symbol=sym, qty=str(qty), side=side, type="market", time_in_force="day"))
        ref = o.get("id")
    else:
        L = ledger()
        pos = L["positions"].get(sym, dict(qty=0, avg=0))
        if side == "buy":
            pos["avg"] = (pos["avg"] * pos["qty"] + px * qty) / (pos["qty"] + qty)
            pos["qty"] += qty
            L["cash"] -= qty * px
        else:
            pos["qty"] -= qty
            L["cash"] += qty * px
        pos["last"] = px
        if pos["qty"] <= 1e-9:
            L["positions"].pop(sym, None)
        else:
            L["positions"][sym] = pos
        save("ledger.json", L)
        ref = "local"
    journal("order", side=side, sym=sym, qty=qty, price=px, why=why, ref=ref)
    print(f"{side.upper()} {qty:g} {sym} @ {px:,.2f} ({'Alpaca paper order ' + str(ref) if ref != 'local' else 'local ledger'}). Why: {why}")


def status():
    a = account()
    pl = a["equity"] - a["start"]
    print(f"broker {a['broker']} | equity {a['equity']:,.2f} | cash {a['cash']:,.2f} | P&L {pl:+,.2f} ({pl / a['start'] * 100:+.2f}%) since start")
    for s, v in sorted(a["positions"].items()):
        print(f"  {s:6} {v['qty']:g} @ {v['avg']:,.2f} (last {v.get('last', v['avg']):,.2f})")
    if not a["positions"]:
        print("  no positions")
    return a


def brief():
    h = load("handoff.json", None)
    if not h:
        print("No handoff yet. This is the first run: read the strategy file, then trade or watch, then write a handoff.")
        return
    age = (dt.datetime.now().astimezone() - dt.datetime.fromisoformat(h["ts"])).total_seconds() / 3600
    print(f"Last handoff {h['ts']} ({age:.1f} h ago) from run '{h.get('run', '?')}':")
    print(f"  did:   {h['did']}")
    print(f"  next:  {h['next']}")
    if h.get("watch"):
        print(f"  watch: {', '.join(h['watch'])}")
    print(f"  orders today so far: {orders_today()} of {RULES['max_orders_per_day']}")
    if age > 30:
        print("  WARNING: handoff is over 30 h old. A run was missed or crashed. Check the journal before trading.")


def handoff(did, nxt, watch, run):
    a = account()
    h = dict(ts=now(), run=run or os.environ.get("PAPER_RUN", "manual"), did=did, next=nxt,
             watch=[w.strip().upper() for w in (watch or "").split(",") if w.strip()],
             equity=round(a["equity"], 2), positions={k: v["qty"] for k, v in a["positions"].items()})
    save("handoff.json", h)
    journal("handoff", **{k: v for k, v in h.items() if k != "ts"})
    print(f"handoff saved ({len(h['positions'])} positions, equity {h['equity']:,.2f}). The next run starts from this.")


def report():
    days = {}
    if os.path.exists(p("journal.jsonl")):
        for line in open(p("journal.jsonl")):
            j = json.loads(line)
            d = days.setdefault(j["ts"][:10], dict(orders=0, equity=None))
            d["orders"] += j["kind"] == "order"
            if j["kind"] == "handoff":
                d["equity"] = j["equity"]
    for d, v in sorted(days.items()):
        print(f"{d}  orders {v['orders']}  end equity {v['equity'] if v['equity'] is not None else '-'}")
    if not days:
        print("journal is empty")


def selftest():
    global D
    D = tempfile.mkdtemp(prefix="paper-selftest-")
    os.environ.pop("ALPACA_KEY", None)
    save("ledger.json", dict(cash=10000.0, start=10000.0, positions={}, created=now()))
    brief()
    trade("buy", "TEST", 10, "selftest buy", price=100.0)
    try:
        trade("buy", "TEST", 20, "too big", price=100.0)
    except SystemExit as e:
        print(e)
    trade("sell", "TEST", 4, "selftest trim", price=110.0)
    try:
        trade("sell", "TEST", 4, "same order twice", price=110.0)
    except SystemExit as e:
        print(e)
    handoff("bought 10 TEST, trimmed 4", "hold the other 6 unless it closes under 95", "TEST", "selftest")
    brief()
    a = status()
    ok = abs(a["cash"] - 9440.0) < 1e-6 and a["positions"]["TEST"]["qty"] == 6
    print(f"selftest {'PASS' if ok else 'FAIL'}: cash {a['cash']:,.2f}, 6 TEST held, cap blocked the 30% buy, repeat order blocked")
    sys.exit(0 if ok else 1)


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    sp = ap.add_subparsers(dest="cmd", required=True)
    i = sp.add_parser("init"); i.add_argument("--cash", type=float, default=10000.0)
    for side in ("buy", "sell"):
        t = sp.add_parser(side); t.add_argument("sym"); t.add_argument("qty", type=float)
        t.add_argument("--why", default=""); t.add_argument("--price", type=float)
        t.add_argument("--again", action="store_true", help="allow an identical order on the same day")
    q = sp.add_parser("quote"); q.add_argument("sym"); q.add_argument("--price", type=float)
    h = sp.add_parser("handoff"); h.add_argument("--did", required=True); h.add_argument("--next", required=True)
    h.add_argument("--watch", default=""); h.add_argument("--run", default="")
    for c in ("brief", "status", "report", "selftest"):
        sp.add_parser(c)
    a = ap.parse_args()
    if a.cmd == "init":
        if ledger():
            sys.exit(f"A ledger already exists in {D}. Move it away first if you want a fresh start.")
        cash = positive(a.cash, "--cash")
        save("ledger.json", dict(cash=cash, start=cash, positions={}, created=now()))
        journal("init", cash=cash)
        print(f"paper ledger started with {cash:,.2f} in {D}")
    elif a.cmd in ("buy", "sell"):
        trade(a.cmd, a.sym, a.qty, a.why, a.price, a.again)
    elif a.cmd == "quote":
        print(f"{a.sym.upper()} {quote(a.sym, a.price):,.2f}")
    elif a.cmd == "handoff":
        handoff(a.did, a.next, a.watch, a.run)
    else:
        {"brief": brief, "status": status, "report": report, "selftest": selftest}[a.cmd]()


if __name__ == "__main__":
    main()