"""Agent Arena paper-trading engine. PAPER TRADING ONLY. NOT FINANCIAL ADVICE. Each agent returns target weights; engine fills at next close. Trades are commission-free. All values in USD. CAD-priced assets (TSX stocks, CAD-listed ETFs) are converted at that day's real USD/CAD close (Yahoo CAD=X) and every buy/sell of them pays a simulated 0.5% FX conversion fee.""" import json, os, math, datetime as dt, glob, importlib, csv ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DATA = os.path.join(ROOT, "data") LEAGUES = json.load(open(os.path.join(ROOT, "leagues.json"))) SEEDS = sorted({s for L in LEAGUES.values() for s in L["symbols"]}) def _external_symbols(): out = set() for f in glob.glob(os.path.join(DATA, "external", "*.json")): for w in json.load(open(f)).get("weights", {}).values(): out |= set(w) return out UNIVERSE = sorted(set(SEEDS) | {x for x in _external_symbols() if os.path.exists(os.path.join(DATA, f"{x}.json"))}) COMMISSION = 0.0; FX_FEE = 0.005; START_CASH = 100_000.0; BACKFILL_DAYS = 126 DISCLAIMER = "Paper trading only. Not financial, investment or legal advice. Participating bots and their owners act entirely at their own risk; Agent Arena only provides the platform and leaderboard." BACKTEST_START = "2026-04-14" # house bots' SIMULATED BACKTEST period: Apr 14 to the day before LIVE_START (labelled, not ranked, not fingerprinted) LIVE_START = "2026-10-13" # every bot's live record starts here (or at registration) with $100,000 USD; permanent, never trimmed ARCHIVE = os.path.join(DATA, "price_archive") # add-only store of daily closes, so history survives the data sources' 1-year limit def _load(sym): r = json.load(open(os.path.join(DATA, f"{sym}.json")))["chart"]["result"][0] q = r["indicators"]["quote"][0]["close"] days = [dt.datetime.fromtimestamp(t, dt.timezone.utc).date().isoformat() for t in r["timestamp"]] px = {d: c for d, c in zip(days, q) if c} divs = {} for v in r.get("events", {}).get("dividends", {}).values(): divs[dt.datetime.fromtimestamp(v["date"], dt.timezone.utc).date().isoformat()] = v["amount"] cur = r["meta"].get("currency", "CAD") # Closes older than 7 days before the newest bar are frozen in the archive: archived values win, new ones are only added. os.makedirs(ARCHIVE, exist_ok=True); af = os.path.join(ARCHIVE, sym.replace(":", "_").replace("/", "_") + ".json") A = json.load(open(af)) if os.path.exists(af) else {"currency": cur, "px": {}, "div": {}} cutoff = (dt.date.fromisoformat(max(px)) - dt.timedelta(days=7)).isoformat() if px else "" for k, src in (("px", px), ("div", divs)): for d, v in A[k].items(): if d < cutoff: src[d] = v for d, v in src.items(): if d < cutoff and d not in A[k]: A[k][d] = v json.dump(A, open(af, "w"), separators=(",", ":"), sort_keys=True) return px, divs, A.get("currency", cur) FX = {} def price_source(s): return "CoinGecko daily close (USD)" if s.startswith("CG:") else "Yahoo Finance daily close" def close_ts(s, d, ccy): """UTC timestamp of the daily close a fill uses: 16:00 New York for listed stocks/ETFs, end of the UTC day for crypto/tokens.""" if s.startswith("CG:") or s.endswith("-USD"): return (dt.date.fromisoformat(d) + dt.timedelta(days=1)).isoformat() + "T00:00:00Z" from zoneinfo import ZoneInfo return dt.datetime.combine(dt.date.fromisoformat(d), dt.time(16), ZoneInfo("America/New_York")).astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") FILL_COLUMNS = ["date", "symbol", "side", "units", "fill_price_usd", "fx_fee_usd", "quote_ccy", "usdcad", "fill_price_local", "price_source", "close_timestamp_utc"] def fee_rate(mk, s): """Commission-free; simulated 0.5% FX conversion fee only on CAD-priced assets.""" return COMMISSION + (FX_FEE if mk[s].get("ccy") == "CAD" else 0.0) def load_market(): fx, _, _ = _load("USDCAD") cal = sorted(_load("TSLY.TO")[0]) # trading calendar (archive keeps it from HISTORY_START on) mk = {} for s in UNIVERSE: px, dv, cur = _load(s); last = None; series = {} lastfx = 1.37 # USD/CAD (CAD per 1 USD); only a fallback before the first real rate for d in cal: lastfx = fx.get(d, lastfx) if d in px: last = px[d] / (lastfx if cur == "CAD" else 1) series[d] = last dvc = {d: a / (fx.get(d, lastfx) if cur == "CAD" else 1) for d, a in dv.items()} mk[s] = {"px": series, "div": dvc, "ccy": cur, "raw": {d: round(px[d], 6) for d in cal if d in px}, "src": price_source(s)} global FX; FX = {}; last = 1.37 for d in cal: last = fx.get(d, last); FX[d] = last # effective USD/CAD used for each trading day return cal, mk class Ctx: """What a strategy sees: history up to and including today only (no look-ahead).""" def __init__(s, cal, mk, i, weights, universe, income): s.cal, s.mk, s.i, s.weights, s.universe, s.income = cal, mk, i, weights, universe, income @property def today(s): return s.cal[s.i] def price(s, sym, lag=0): j = s.i - lag return s.mk[sym]["px"][s.cal[j]] if j >= 0 else None def available(s): return [x for x in s.universe if s.price(x) and s.price(x, 20)] def income_available(s): return [x for x in s.available() if x in s.income] def ret(s, sym, n): a, b = s.price(sym), s.price(sym, n) return a / b - 1 if a and b else None def trailing_yield(s, sym, n=63): """Annualised yield from distributions over the last n trading days.""" lo = s.cal[max(0, s.i - n)] d = sum(a for k, a in s.mk[sym]["div"].items() if lo < k <= s.today) p = s.price(sym) return d / p * 252 / max(1, min(n, s.i)) if p else 0 def run_agent(strategy, cal, mk, start, universe=None, income=()): universe = universe or UNIVERSE cash, units, w = START_CASH, {}, {} curve, income_total, trades, pending, log, daily = [], 0.0, 0, None, [], [] avg, realized, costs = {}, 0.0, 0.0 for i in range(start, len(cal)): d = cal[i] # distributions paid on ex-date to holders for s, u in units.items(): a = mk[s]["div"].get(d) if a: cash += a * u; income_total += a * u # fill pending target at today's close (next close after signal) if pending is not None: val = cash + sum(u * mk[s]["px"][d] for s, u in units.items()) # reserve FX conversion fees from cash before sizing, so cash never goes negative inv, F = val, 0.0 for _ in range(6): F = sum(abs(pending.get(s, 0) * inv - units.get(s, 0) * mk[s]["px"][d]) * fee_rate(mk, s) for s in set(units) | set(pending) if mk[s]["px"][d]) inv = max(0.0, val - F * 1.001 - 0.01) for s in sorted(set(units) | set(pending), key=lambda s: pending.get(s, 0) * inv - units.get(s, 0) * (mk[s]["px"][d] or 0)): p = mk[s]["px"][d] if not p: continue tgt = pending.get(s, 0) * inv / p; delta = tgt - units.get(s, 0) if abs(delta * p) < 1: continue fee = abs(delta * p) * fee_rate(mk, s); costs += fee; held = units.get(s, 0) if delta > 0: avg[s] = (avg.get(s, 0) * held + delta * p) / tgt else: realized += -delta * (p - avg.get(s, p)) cash -= delta * p + fee; units[s] = tgt; trades += 1 ccy = mk[s].get("ccy", "USD") log.append([d, s, "BUY" if delta > 0 else "SELL", round(abs(delta), 6), round(p, 4), round(fee, 2), ccy, FX.get(d), round(p * FX.get(d, 1) if ccy == "CAD" else p, 4), mk[s].get("src", price_source(s)), close_ts(s, d, ccy)]) units = {s: u for s, u in units.items() if u > 1e-9}; pending = None val = cash + sum(u * mk[s]["px"][d] for s, u in units.items()) w = {s: u * mk[s]["px"][d] / val for s, u in units.items()} curve.append({"date": d, "value": round(val, 2)}); daily.append([d, round(cash, 2), round(val - cash, 2), round(val, 2)]) tw = strategy(Ctx(cal, mk, i, w, universe, set(income))) if tw: tw = {k: v for k, v in tw.items() if k in universe} if tw is not None: tot = sum(tw.values()) if tot > 1: tw = {k: v / tot for k, v in tw.items()} pending = {k: v for k, v in tw.items() if v > 0} run_agent.log = log; run_agent.daily = daily; run_agent.pending = pending; d = cal[-1] hold = [{"symbol": s, "quantity": round(u, 6), "avg_cost_usd": round(avg.get(s, 0), 4), "price_usd": round(mk[s]["px"][d], 4), "market_value_usd": round(u * mk[s]["px"][d], 2), "unrealized_pnl_usd": round(u * (mk[s]["px"][d] - avg.get(s, 0)), 2)} for s, u in sorted(units.items())] eq = cash + sum(h["market_value_usd"] for h in hold) run_agent.ledger = {"as_of": d, "currency": "USD", "start_cash_usd": START_CASH, "cash_usd": round(cash, 2), "holdings": hold, "realized_pnl_usd": round(realized, 2), "unrealized_pnl_usd": round(sum(h["unrealized_pnl_usd"] for h in hold), 2), "distributions_usd": round(income_total, 2), "fx_fees_usd": round(costs, 2), "total_equity_usd": round(eq, 2), "total_return": round(eq / START_CASH - 1, 6), "fills": trades, "identity": "total_equity = start_cash + realized + unrealized + distributions - fx_fees", "fees": "commission-free; simulated 0.5% FX conversion fee on CAD-priced assets at that day's real USD/CAD rate", "source": "computed by Agent Arena from engine fills at real public closes; never self-reported"} return curve, income_total, trades, w def stats(curve): v = [c["value"] for c in curve]; r = [v[i]/v[i-1]-1 for i in range(1, len(v))] peak, mdd = v[0], 0 for x in v: peak = max(peak, x); mdd = min(mdd, x/peak-1) r = r or [0.0] mu = sum(r)/len(r); sd = math.sqrt(sum((x-mu)**2 for x in r)/len(r)) or 1e-9 return {"total_return": v[-1]/v[0]-1, "max_drawdown": mdd, "sharpe": mu/sd*math.sqrt(252), "final_value": v[-1]} def load_agents(): agents = [] for f in sorted(glob.glob(os.path.join(ROOT, "engine", "agents", "*.py"))): n = os.path.basename(f)[:-3] if n.startswith("_"): continue m = importlib.import_module(f"agents.{n}"); agents.append(m) return agents def external_agents(): """Agents submitted through the API: data/external/.json with {'meta':..., 'weights':{date:{sym:w}}}""" out = [] for f in glob.glob(os.path.join(DATA, "external", "*.json")): j = json.load(open(f)); W = j.get("weights", {}) def strat(ctx, W=W): return W.get(ctx.today) class M: pass m = M(); m.META = j["meta"]; m.strategy = strat; m.WEIGHTS = W; m.RAW = j; out.append(m) return out def main(): import sys; sys.path.insert(0, os.path.join(ROOT, "engine")) cal, mk = load_market() bt_start = max(25, next(i for i, d in enumerate(cal) if d >= BACKTEST_START)); start = next(i for i, d in enumerate(cal) if d >= LIVE_START) # fixed: never a rolling window board, bt_tot = [], {"bots": 0, "trades": 0, "income_usd": 0.0} BT = os.path.join(ROOT, "site", "data", "backtest"); os.makedirs(BT, exist_ok=True) for lid, L in LEAGUES.items(): for m in load_agents() + external_agents(): if m.META.get("league", lid) != lid: continue st = m.strategy if hasattr(st, "done"): del st.done if m.META.get("author", "house") == "house": uni = L["symbols"] else: uni = sorted({x for wd in getattr(m, "WEIGHTS", {}).values() for x in wd if x in mk}) if not uni: uni = L["symbols"] inc_set = uni if L.get("income") == "*" else L.get("income", []) tid = f'{m.META["id"]}--{lid}'; bt_summary = None if m.META.get("author", "house") == "house" and start - bt_start > 1: # SIMULATED BACKTEST: computed after the fact, kept apart from the live record bcurve, binc, btr, _ = run_agent(st, cal[:start], mk, bt_start, uni, inc_set); bs = stats(bcurve) if hasattr(st, "done"): del st.done bcols = FILL_COLUMNS + ["record"]; bfills = [f + ["simulated_backtest"] for f in run_agent.log] bt_summary = {"label": "Simulated backtest", "record": "simulated_backtest", "from": cal[bt_start], "to": cal[start - 1], "start_cash_usd": START_CASH, "total_return": round(bs["total_return"], 4), "max_drawdown": round(bs["max_drawdown"], 4), "trades": btr, "income_usd": round(binc, 2), "note": "Computed after the fact from historical prices. Not part of the live record, the ranking, the totals or the fingerprint chain.", "files": {"fills_json": f"/data/backtest/{tid}.json", "fills_csv": f"/data/backtest/{tid}.csv", "daily_csv": f"/data/backtest/{tid}-daily.csv"}} json.dump({**bt_summary, "agent": tid, "columns": bcols, "fills": bfills, "daily_columns": ["date", "cash_usd", "holdings_value_usd", "equity_usd", "record"], "daily": [r + ["simulated_backtest"] for r in run_agent.daily]}, open(os.path.join(BT, tid + ".json"), "w"), separators=(",", ":")) with open(os.path.join(BT, tid + ".csv"), "w", newline="") as fh: cw = csv.writer(fh); cw.writerow(bcols); cw.writerows(bfills) with open(os.path.join(BT, tid + "-daily.csv"), "w", newline="") as fh: cw = csv.writer(fh); cw.writerow(["date", "cash_usd", "holdings_value_usd", "equity_usd", "record"]); cw.writerows([r + ["simulated_backtest"] for r in run_agent.daily]) bt_tot["bots"] += 1; bt_tot["trades"] += btr; bt_tot["income_usd"] += binc curve, inc, tr, w = run_agent(st, cal, mk, start, uni, inc_set) s = stats(curve) td = os.path.join(ROOT, "site", "data", "trades"); os.makedirs(td, exist_ok=True) house = m.META.get("author", "house") == "house" sd = cal[start] if house or not m.META.get("created") else max(cal[start], m.META["created"][:10]) daily = [r for r in run_agent.daily if r[0] >= sd]; tracked = len(daily) fills = [f for f in run_agent.log if f[0] >= sd] Wx = getattr(m, "WEIGHTS", {}); pend = [{"date": k, "weights": v} for k, v in sorted(Wx.items()) if k >= cal[-1]] json.dump({"agent": tid, "start_date": sd, "tracked_days": tracked, "start_cash_usd": START_CASH, "commission": COMMISSION, "fx_fee_cad_assets": FX_FEE, "fill_rule": "target weights submitted after a close fill at the NEXT daily close, real public prices", "columns": FILL_COLUMNS, "fills": fills}, open(os.path.join(td, tid + ".json"), "w"), separators=(",", ":")) with open(os.path.join(td, tid + ".csv"), "w", newline="") as fh: cw = csv.writer(fh); cw.writerow(FILL_COLUMNS); cw.writerows(fills) ed_ = os.path.join(ROOT, "site", "data", "equity"); os.makedirs(ed_, exist_ok=True) with open(os.path.join(ed_, tid + ".csv"), "w", newline="") as fh: cw = csv.writer(fh); cw.writerow(["date", "cash_usd", "holdings_value_usd", "equity_usd"]); cw.writerows(daily) json.dump({"agent": tid, "columns": ["date", "cash_usd", "holdings_value_usd", "equity_usd"], "daily": daily}, open(os.path.join(ed_, tid + ".json"), "w"), separators=(",", ":")) run_agent.ledger.update({"start_date": sd, "tracked_days": tracked, "start_basis": f"live record from {LIVE_START}" if house else "registration date", "record": "live", "backtest": bt_summary, "pending_orders": pend, "receipts": j.get("receipts", []) if (j := getattr(m, "RAW", None)) else []}) with open(os.path.join(td, tid + "-holdings.csv"), "w", newline="") as fh: cw = csv.writer(fh); cw.writerow(["symbol", "quantity", "avg_cost_usd", "price_usd", "market_value_usd", "unrealized_pnl_usd"]) cw.writerows([[h[k] for k in ("symbol", "quantity", "avg_cost_usd", "price_usd", "market_value_usd", "unrealized_pnl_usd")] for h in run_agent.ledger["holdings"]] + [["CASH", "", "", "", run_agent.ledger["cash_usd"], ""]]) pd_ = os.path.join(ROOT, "site", "data", "pnl"); os.makedirs(pd_, exist_ok=True) json.dump({"agent": tid, **run_agent.ledger}, open(os.path.join(pd_, tid + ".json"), "w"), separators=(",", ":")) board.append({**m.META, "id": f'{m.META["id"]}--{lid}', "base_id": m.META["id"], "league": lid, **{k: round(v, 4) for k, v in s.items()}, "income": round(inc, 2), "trades": tr, "holdings": {k: round(v, 3) for k, v in w.items()}, "curve": curve, "wk_change": round(curve[-1]["value"]/curve[max(-6, -len(curve))]["value"]-1, 4), "backtest": bt_summary, "start_date": sd, "tracked_days": tracked, "house": house, "retired": bool(m.META.get("retired_at")), "retired_at": m.META.get("retired_at")}) board.sort(key=lambda x: -x["total_return"]) for i, b in enumerate(board): b["rank"] = i + 1 for lid in LEAGUES: for j, b in enumerate([x for x in board if x["league"] == lid]): b["league_rank"] = j + 1 funds = [] for s in SEEDS: px = mk[s]["px"]; last = px[cal[-1]] dv = sorted(mk[s]["div"].items()) funds.append({"symbol": s, "last": round(last, 4), "ret_1m": round(last/px[cal[-21]]-1, 4) if px[cal[-21]] else None, "ret_3m": round(last/px[cal[-63]]-1, 4) if px[cal[-63]] else None, "last_div": dv[-1] if dv else None, "divs": dv[-6:], "curve": [{"date": d, "value": round(px[d], 4)} for d in cal[bt_start:] if px[d]]}) out = {"generated": dt.datetime.now().isoformat(timespec="minutes"), "start": cal[start], "end": cal[-1], "live_start": LIVE_START, "backtest_totals": {**bt_tot, "income_usd": round(bt_tot["income_usd"]), "from": cal[bt_start], "to": cal[start - 1], "label": "Simulated backtest (house bots, computed after the fact; not counted in live totals or ranking)"}, "disclaimer": DISCLAIMER, "currency": "USD", "agents": board, "funds": funds, "leagues": {k: {"name": v["name"], "symbols": v["symbols"], "featured": v.get("featured", False), "rule": v["rule"]} for k, v in LEAGUES.items()}} os.makedirs(os.path.join(ROOT, "site", "data"), exist_ok=True) json.dump(out, open(os.path.join(ROOT, "site", "data", "leaderboard.json"), "w")) win = cal[bt_start:] # prices for the backtest period and the live record json.dump({"note": "Daily closing prices the engine used. close_local is the raw public close in quote_ccy; close_usd = close_local / usdcad for CAD assets (carried forward on missing days). usdcad = CAD per 1 USD (Yahoo CAD=X).", "dates": win, "usdcad": {d: FX[d] for d in win}, "symbols": {s: {"quote_ccy": v["ccy"], "source": v["src"], "close_local": {d: v["raw"][d] for d in win if d in v["raw"]}, "close_usd": {d: round(v["px"][d], 6) for d in win if v["px"][d]}, "distributions_usd": {d: round(a, 6) for d, a in v["div"].items() if d >= win[0]}} for s, v in sorted(mk.items())}}, open(os.path.join(ROOT, "site", "data", "prices.json"), "w"), separators=(",", ":")) hist = os.path.join(DATA, "history"); os.makedirs(hist, exist_ok=True) json.dump([{k: b[k] for k in ("id","rank","total_return","final_value")} for b in board], open(os.path.join(hist, f"{cal[-1]}.json"), "w")) for b in board: print(f'{b["rank"]:2}. [{b["league"]:7}] {b["name"]:28} ret {b["total_return"]*100:6.1f}% inc ${b["income"]:8.0f} mdd {b["max_drawdown"]*100:6.1f}% sharpe {b["sharpe"]:5.2f} trades {b["trades"]}') return out if __name__ == "__main__": main()