diff --git a/.env.example b/.env.example index 749383dff..67b368726 100644 --- a/.env.example +++ b/.env.example @@ -39,3 +39,13 @@ ADMIN_USER_ID= # Examples: http://myserver.com:8088, https://dashboard.example.com # WEB_URL=http://localhost:8088 + +# Hackathon race monitor (scripts/hackathon_monitor.py) +# HACKATHON_API_URL=https://example.org +# HACKATHON_API_TOKEN= +# HACKATHON_SLUG=agent-builders-cup-1 +# Comma-separated server names; empty means every server in config.yml +# HACKATHON_SERVERS= +# YAML of race agent id -> Condor run keys, for strategies whose slug differs +# (template: scripts/hackathon_agents.example.yml) +# HACKATHON_AGENT_MAP= diff --git a/frontend/src/components/chat/DockExecution.tsx b/frontend/src/components/chat/DockExecution.tsx index 827373344..7230e20fb 100644 --- a/frontend/src/components/chat/DockExecution.tsx +++ b/frontend/src/components/chat/DockExecution.tsx @@ -36,7 +36,12 @@ import { pnlColor, shortBotName, } from "@/lib/formatters"; -import { quoteConverter, runningLeaves } from "@/lib/perf-population"; +import { + botsByController, + quoteConverter, + runningLeaves, + terminatedLeaves, +} from "@/lib/perf-population"; import { UNATTACHED_BOT, controllerNodeId, type PerfLeaf } from "@/lib/perf-tree"; import { agentsQuery } from "@/lib/queryClient"; @@ -177,7 +182,10 @@ export function DockExecution({ // and the caches are shared rather than doubled, and a reader with the // browser open pays nothing for this panel. No history walk: this folds, it // does not chart, and the walk costs a paged request per controller. - const fleet = useFleetData(server, { population: "running", history: false }); + // `terminated: true` adds the runs and the controllers they left behind, under + // the keys `/bots`' Terminated side already holds: an agent row is credited + // with what its stopped bots made, not only with what is trading now. + const fleet = useFleetData(server, { population: "running", history: false, terminated: true }); // The shared roster (key and cadence: `agentsQuery`), so the agent rows' // liveness arrives with a request nobody made for them. @@ -248,12 +256,50 @@ export function DockExecution({ return [...controllers, ...all.filter((leaf) => leaf.kind !== "controller")]; }, [fleet.controllers, fleet.executors, fleet.owners, fleet.deeds]); + /** + * Everything that has finished on this server — `terminatedLeaves`, the + * construction `/bots?population=terminated` folds, with no period window. + * + * `executionRows` keeps only what is credited to an agent and folds it into + * that agent's row, so an agent that stopped a bot or closed an executor keeps + * the money it made. The live half still moves on every socket frame; this + * half changes only when something stops. + */ + const history = useMemo( + () => + terminatedLeaves({ + executors: fleet.executors, + terminatedControllers: fleet.terminatedControllers, + runs: fleet.runs, + owners: fleet.owners, + deeds: fleet.deeds, + botByController: botsByController(fleet.controllers), + }), + [ + fleet.executors, + fleet.terminatedControllers, + fleet.runs, + fleet.owners, + fleet.deeds, + fleet.controllers, + ], + ); + const convert = useMemo(() => quoteConverter(fleet.convert), [fleet.convert]); const currencySymbol = fleet.currencySymbol ?? "$"; const rows = useMemo( - () => executionRows({ leaves, deeds: fleet.deeds, agents, owners: fleet.owners, convert, now }), - [leaves, fleet.deeds, agents, fleet.owners, convert, now], + () => + executionRows({ + leaves, + history, + deeds: fleet.deeds, + agents, + owners: fleet.owners, + convert, + now, + }), + [leaves, history, fleet.deeds, agents, fleet.owners, convert, now], ); // Only what the reader has actually clicked; the default for everything else @@ -334,7 +380,9 @@ export function DockExecution({ ); } - if (!deployedCount && !unattached) { + // An agent that has stopped everything still has a row — its history is on + // this server — so the empty sentence is for a server with no rows at all. + if (!deployedCount && !unattached && rows.length === 0) { return (

@@ -447,6 +495,17 @@ export function DockExecution({ onToggle={() => toggle(row.id)} onOpenAgent={onOpenAgent} /> + ) : row.kind === "history" ? ( + + navigate( + `/bots?population=terminated&scope=${encodeURIComponent(row.parentId ?? "")}`, + ) + } + /> ) : ( @@ -574,7 +633,7 @@ function AgentRow({ */ const total = (value: number, what: string, pnl: boolean) => ( void; +}) { + const cell = "whitespace-nowrap px-1.5 py-0.5 text-right font-mono tabular-nums"; + const pnl = (value: number) => ( + + {formatCurrencyPnl(value, symbol)} + + ); + const described = `${row.label} — realized PnL only: a stopped bot's last open position is not counted${ + capped ? ". Older closed executors are past the loaded page and are missing" : "" + }`; + return ( + { + if (e.key !== "Enter" && e.key !== " ") return; + e.preventDefault(); + onOpen(); + }} + className="cursor-pointer text-[var(--color-text-muted)] transition-colors hover:bg-[var(--color-surface-hover)]" + > + + {row.label} + {capped ? " +" : ""} + + {formatCompactVolume(row.totals.volume, symbol)} + {pnl(row.totals.realized)} + {pnl(row.totals.unrealized)} + {pnl(row.totals.net)} + + + ); +} + /** * One bot, once per group: its whole branch in the browser, and the level the * executors under it are counted at. diff --git a/frontend/src/components/chat/executionTree.test.ts b/frontend/src/components/chat/executionTree.test.ts index bde146c60..b8305e5b7 100644 --- a/frontend/src/components/chat/executionTree.test.ts +++ b/frontend/src/components/chat/executionTree.test.ts @@ -14,6 +14,7 @@ import { describe, expect, it } from "vitest"; import { AUTO_OPEN_AGENTS, + HISTORY_SUFFIX, executionCounts, executionRows, openRows, @@ -296,3 +297,100 @@ describe("the header's counts", () => { expect(executionCounts(rows)).toEqual({ controllers: 3, paused: 1 }); }); }); + +describe("an agent's history", () => { + const live = [ + leaf({ id: "c1", bot: "race-live", agent: "race.mm", how: "namespace", realized: 10, unrealized: 5, net: 15, volume: 1000 }), + ]; + // A stopped bot: its final snapshot still carries a frozen 30 unrealized. + const stoppedCtrl = leaf({ + id: "c0", + bot: "race-old", + agent: "race.mm", + how: "namespace", + realized: 40, + unrealized: 30, + net: 70, + volume: 5000, + running: false, + status: "stopped", + endedAt: NOW - 600_000, + }); + // A standalone executor the agent's session opened and closed. + const closedExec = leaf({ + id: "e9", + kind: "executor", + controllerId: "", + bot: "race.mm_s3", + agent: "race.mm", + how: "session", + realized: -4, + net: -4, + volume: 200, + running: false, + status: "terminated", + endedAt: NOW - 60_000, + }); + const rowsWith = (history: PerfLeaf[], leaves: PerfLeaf[] = live) => + executionRows({ leaves, history, deeds: null, agents: [], owners: [CONDOR_UI], convert, now: NOW }); + + it("credits the agent with its stopped bots and closed executors, under one history row", () => { + const rows = rowsWith([stoppedCtrl, closedExec]); + const agentRow = rows.find((r) => r.kind === "agent")!; + const history = rows.find((r) => r.kind === "history")!; + + expect(history.id).toBe(`agent:race.mm${HISTORY_SUFFIX}`); + expect(history.parentId).toBe("agent:race.mm"); + expect(history.label).toBe("History · 1 stopped bot, 1 closed executor"); + // Realized only for the stopped bot: its last mark is not money. + expect(history.totals.net).toBe(36); + expect(history.totals.unrealized).toBe(0); + // The agent is live + history, and its rows still add up to it. + expect(agentRow.totals.net).toBe(15 + 36); + expect(agentRow.totals.volume).toBe(6200); + const children = rows.filter((r) => r.parentId === agentRow.id); + expect(children.reduce((n, r) => n + r.totals.net, 0)).toBe(agentRow.totals.net); + }); + + it("draws no row for a finished controller or a bot with nothing live under it", () => { + const rows = rowsWith([stoppedCtrl]); + expect(rows.map((r) => r.kind)).toEqual(["agent", "controller", "history"]); + expect(executionCounts(rows)).toEqual({ controllers: 1, paused: 0 }); + }); + + it("keeps an agent that stopped everything on the panel", () => { + const rows = rowsWith([stoppedCtrl], []); + expect(rows.map((r) => [r.kind, r.totals.net])).toEqual([ + ["agent", 40], + ["history", 40], + ]); + }); + + it("does not count a live controller's closed executor twice", () => { + // The controller record already covers every executor it ever ran. + const closedUnderLive = leaf({ + id: "e1", + kind: "executor", + controllerId: "c1", + bot: "race-live", + agent: "race.mm", + how: "namespace", + realized: 3, + net: 3, + running: false, + status: "terminated", + }); + const rows = rowsWith([closedUnderLive]); + expect(rows.find((r) => r.kind === "agent")!.totals.net).toBe(15); + expect(rows.some((r) => r.kind === "history")).toBe(false); + }); + + it("leaves unowned and pseudo-run history out", () => { + const rows = rowsWith([ + leaf({ ...closedExec, id: "e2", agent: "", how: "none", bot: "main" }), + leaf({ ...closedExec, id: "e3", agent: "condor.ui", how: "deed", bot: "main" }), + ]); + expect(rows.map((r) => r.kind)).toEqual(["agent", "controller"]); + expect(rows[0].totals.net).toBe(15); + }); +}); diff --git a/frontend/src/components/chat/executionTree.ts b/frontend/src/components/chat/executionTree.ts index 92f0a4a59..8c4ae0299 100644 --- a/frontend/src/components/chat/executionTree.ts +++ b/frontend/src/components/chat/executionTree.ts @@ -12,6 +12,13 @@ // the ledger* rather than dropped. So the agent rows partition the panel's // total by construction, which is the property the fold discipline rests on. // +// **An agent's money is everything it ever deployed.** The live rows are what +// is trading now, but an agent row's totals also carry the bots it stopped and +// the executors that closed — the same lifetime figure the race monitor posts +// (`scripts/hackathon_monitor.py`) — under one `history` row per agent, so the +// rows beneath an agent still add up to it. Stopping a bot moves its money from +// a controller row to the history row; it never takes it off the agent. +// // Nothing here fetches and nothing here renders (the ARCH-300 split), so every // judgement below — which rows exist, which level collapses, which agent a run // key belongs to — is reachable from a test rather than only from a rendered @@ -50,7 +57,8 @@ const GROUPING = ["agent", "bot"] as const; export const AUTO_OPEN_AGENTS = 3; /** - * One line of the panel — an agent, a bot, or a controller. + * One line of the panel — an agent, a bot, a controller, or what an agent has + * already finished. * * Executors are deliberately not rows: they are counted on the controller that * is running them, exactly as they were before this feature, and a third @@ -58,9 +66,12 @@ export const AUTO_OPEN_AGENTS = 3; * still in every `leaves` above them, so nothing is missing from a total. */ export interface ExecutionRow { - /** `agent:{runKey}` / `bot:{name}` / a controller node id — the browser's own ids. */ + /** + * `agent:{runKey}` / `bot:{name}` / a controller node id — the browser's own + * ids — or `{agent id}::history` for the finished records under an agent. + */ id: string; - kind: "agent" | "bot" | "controller"; + kind: "agent" | "bot" | "controller" | "history"; label: string; depth: 0 | 1 | 2; /** The row this one hangs under, `null` at the top — what {@link visibleRows} walks. */ @@ -76,7 +87,15 @@ export interface ExecutionRow { } export interface ExecutionInput { + /** The running population — what is trading now, and what draws a row. */ leaves: readonly PerfLeaf[]; + /** + * The terminated population (`terminatedLeaves`): stopped bots' controllers + * and closed executors. Only what is credited to a real agent is kept — see + * {@link agentHistory} — and it is folded into that agent's row, never drawn + * as rows of its own. + */ + history?: readonly PerfLeaf[]; deeds: DeedIndex | null; /** `["agents"]`, for turning a run key into an agent somebody can open. */ agents: readonly AgentSummary[]; @@ -95,6 +114,37 @@ export interface ExecutionInput { const AGENT_PREFIX = AXIS_PREFIX.agent; +/** The suffix a history row's id carries after its agent's id. */ +export const HISTORY_SUFFIX = "::history"; + +/** + * The finished records an agent row is credited with. + * + * **Only an agent's.** A pseudo-run (a chat, a delegation, the dashboard) and an + * unowned record keep their live rows and nothing else: the terminated side of + * a real server holds every hand-opened position it has ever had, and this + * panel is read for what the agents did, not for the archive. + * + * **A stopped controller counts what it realized, not its last mark.** Its + * final snapshot outlives the bot, so the unrealized PnL in it is a frozen mark + * on a book nobody is quoting any more, not money ([[CORR-633]]). This is the + * rule the server-side aggregator (`_merge_stopped_instance`) applies, and so + * the race monitor, so the dock and the race board cannot disagree about a + * stopped bot. A closed executor's PnL is already all realized. + */ +export function agentHistory(history: readonly PerfLeaf[]): PerfLeaf[] { + const out: PerfLeaf[] = []; + for (const leaf of history) { + if (!leaf.agent || isPseudoRunKey(leaf.agent)) continue; + out.push( + leaf.kind === "controller" + ? { ...leaf, unrealized: 0, net: leaf.realized, positions: [] } + : leaf, + ); + } + return out; +} + /** * Every row the panel can draw, parents before their children. * @@ -106,9 +156,18 @@ const AGENT_PREFIX = AXIS_PREFIX.agent; */ export function executionRows(input: ExecutionInput): ExecutionRow[] { const { leaves, deeds, agents, owners, convert, now } = input; - const tree = buildTree([...leaves], "All", { grouping: [...GROUPING], deeds }); + const history = agentHistory(input.history ?? []); + // One tree over both populations, live first so live agents keep the top of + // the panel. `buildTree`'s spine rule is what keeps this from double + // counting: a closed executor of a live controller hangs under that + // controller's node, whose spine is the controller record — which already + // covers every executor it ever ran. + const tree = buildTree([...leaves, ...history], "All", { grouping: [...GROUPING], deeds }); const bySlug = new Map(agents.map((agent) => [agent.slug, agent])); - const fold = (node: PerfNode) => foldLeaves(node.leaves, convert, now); + const live = new Set(leaves); + const isLive = (leaf: PerfLeaf) => live.has(leaf); + const foldOf = (spine: PerfLeaf[]) => foldLeaves(spine, convert, now); + const fold = (node: PerfNode) => foldOf(node.leaves.filter(isLive)); /** * A controller node, as a row — or nothing. @@ -125,7 +184,9 @@ export function executionRows(input: ExecutionInput): ExecutionRow[] { parentId: string, ): ExecutionRow | null => { const leaf = node.leaves[0]; - if (!leaf || leaf.kind !== "controller") return null; + // A finished controller is in the tree only to be folded into its agent; + // its row would be a controller nothing is running. + if (!leaf || leaf.kind !== "controller" || !isLive(leaf)) return null; return { id: node.id, kind: "controller", @@ -154,7 +215,9 @@ export function executionRows(input: ExecutionInput): ExecutionRow[] { // — the sidebar's own rule, per owner rather than per fleet: a fleet // running one bot must not spend a chevron saying so, and neither must an // agent that does. - const botLevel = distinguishes(agentNode.leaves, "bot", deeds); + const liveSpine = agentNode.leaves.filter(isLive); + const finished = agentNode.leaves.filter((leaf) => !isLive(leaf)); + const botLevel = distinguishes(liveSpine, "bot", deeds); const under: ExecutionRow[] = []; for (const child of agentNode.children) { @@ -164,6 +227,8 @@ export function executionRows(input: ExecutionInput): ExecutionRow[] { continue; } if (child.kind !== "bot") continue; + // A bot that only has finished records under it is history, not a row. + if (!child.leaves.some(isLive)) continue; const controllers = child.children .map((c) => controllerRow(c, botLevel ? 2 : 1, botLevel ? child.id : agentNode.id)) .filter((row): row is ExecutionRow => row !== null); @@ -179,11 +244,24 @@ export function executionRows(input: ExecutionInput): ExecutionRow[] { parentId: agentNode.id, hasChildren: controllers.length > 0, totals: fold(child), - leaves: child.leaves, + leaves: child.leaves.filter(isLive), }); under.push(...controllers); } + if (finished.length > 0) { + under.push({ + id: `${agentNode.id}${HISTORY_SUFFIX}`, + kind: "history", + label: historyLabel(finished), + depth: 1, + parentId: agentNode.id, + hasChildren: false, + totals: foldOf(finished), + leaves: finished, + }); + } + rows.push({ id: agentNode.id, kind: "agent", @@ -200,7 +278,8 @@ export function executionRows(input: ExecutionInput): ExecutionRow[] { depth: 0, parentId: null, hasChildren: under.length > 0, - totals: fold(agentNode), + // The whole spine, finished records included: what this agent has made. + totals: foldOf(agentNode.leaves), leaves: agentNode.leaves, ...(claimed ? { agent: { slug: claimed.slug, name: claimed.name } } : {}), }); @@ -210,6 +289,26 @@ export function executionRows(input: ExecutionInput): ExecutionRow[] { return rows; } +/** + * What a history row says it holds: `"History · 2 stopped bots, 14 closed executors"`. + * + * Bots rather than controllers, because a stop is something done to a bot — and + * the executors counted are the ones filed under no finished controller (the + * agent's own standalone ones), since a controller's record already covers its. + */ +export function historyLabel(finished: readonly PerfLeaf[]): string { + const bots = new Set(); + let executors = 0; + for (const leaf of finished) { + if (leaf.kind === "controller") bots.add(leaf.bot); + else executors += 1; + } + const parts: string[] = []; + if (bots.size) parts.push(`${bots.size} stopped bot${bots.size === 1 ? "" : "s"}`); + if (executors) parts.push(`${executors} closed executor${executors === 1 ? "" : "s"}`); + return `History · ${parts.join(", ")}`; +} + /** * Which rows are expanded: the default for this fleet's shape, with the * reader's own toggles on top. diff --git a/frontend/src/components/perf/PerfBrowser.tsx b/frontend/src/components/perf/PerfBrowser.tsx index 5f3ebaf1c..ea1198eaf 100644 --- a/frontend/src/components/perf/PerfBrowser.tsx +++ b/frontend/src/components/perf/PerfBrowser.tsx @@ -67,7 +67,6 @@ import { isExecutorActive, pnlColor, shortBotName, - toMs, } from "@/lib/formatters"; import { ancestorChain, @@ -82,8 +81,6 @@ import { emptyScopeNode, foldLeaves, indexTree, - leafFromExecutor, - leafFromTerminatedController, matchesGrain, parsePopulation, resolveScope, @@ -96,7 +93,13 @@ import { type Population, type PerfLeaf, } from "@/lib/perf-tree"; -import { botsByController, quoteConverter, runningLeaves } from "@/lib/perf-population"; +import { + botsByController, + latestRunByBot, + quoteConverter, + runningLeaves, + terminatedLeaves, +} from "@/lib/perf-population"; import { collapseGrouping, DEFAULT_GROUPING, @@ -123,7 +126,6 @@ import type { ConvertFn } from "@/lib/rates"; import { unavailableLabel } from "@/lib/strategy-unavailable"; import { useViewFacts } from "@/lib/viewFacts"; import { - attributionIndex, loopFacts, loopStatus, ownerOf, @@ -826,14 +828,7 @@ export function PerfBrowser({ * practice; where it is not, the most recent run wins, which is the one the * finished controllers on screen belong to. */ - const runByBot = useMemo(() => { - const latest = new Map(); - for (const run of runs) { - const seen = latest.get(run.bot_name); - if (!seen || (run.created_at ?? "") > (seen.created_at ?? "")) latest.set(run.bot_name, run); - } - return latest; - }, [runs]); + const runByBot = useMemo(() => latestRunByBot(runs), [runs]); /** * The instant a terminated fold reaches back to, hoisted out of `leavesFor` @@ -877,80 +872,25 @@ export function PerfBrowser({ */ const leavesFor = useCallback( (which: Population): PerfLeaf[] => { - const all: PerfLeaf[] = []; - /** - * The run a record belongs to **and how we know**: the two links the - * runtime enforces — the bot's namespace, and the session id a standalone - * executor is tagged with — then, only if neither answers, the record - * Condor kept of its own deeds (`lib/agent-attribution`, FEAT-096/106). - * - * Bot name first: a controller is attributed through its bot, and an - * executor working under one inherits that answer, so the `controller_id` - * fallback is left to the executor nobody claims — which is exactly the - * agent-created one, whose `controller_id` *is* its session's agent id. - * - * Built once per fold rather than per record (PERF-331): the owner list - * is fixed for the whole call, and the terminated population below walks - * every executor the fleet has ever had. - */ - const agentOf = attributionIndex(owners, deeds); - /** - * The bot a *closed* executor hung under, by the run that opened it. - * - * Asked at the executor's own start time, which is the instant the - * question is about: a position that outlived its bot's stop still - * belongs to the run that opened it. Falls through to the live fleet for - * an executor of a bot that is still deployed — it has no closed run - * window to sit in — and to `(unattached)` for one that belongs to no - * run at all, which on a real server is every hand-opened position. - */ - const closedBotOf = (ex: ExecutorInfo, startedAt: number | null) => { - if (!ex.controller_id) return UNATTACHED_BOT; - const at = startedAt ?? (ex.close_timestamp > 0 ? toMs(ex.close_timestamp) : null); - const owner = at === null ? null : attribute(ex.controller_id, at); - return owner ?? botByController.get(ex.controller_id) ?? UNATTACHED_BOT; - }; if (which === "running") { - // Shared with the workspace's Money view rather than written twice - // (FEAT-109): that view has to report the same number this browser - // reports at `?scope=agent:{runKey}`, and two copies of the + // Shared with the workspace's Money view and the execution dock rather + // than written again (FEAT-109): each has to report the same number this + // browser reports at `?scope=agent:{runKey}`, and copies of the // construction would make them agree by coincidence. - all.push( - ...runningLeaves({ controllers, executors, owners, deeds, botByController }), - ); - } else { - // The window applies to what has finished, and is measured from each - // record's *end*: a period called "the last week" is the trading that - // stopped in it, not the trading that started in it. - for (const ex of executors) { - if (isExecutorActive(ex.status)) continue; - const started = ex.timestamp > 0 ? toMs(ex.timestamp) : null; - const bot = closedBotOf(ex, started); - const att = agentOf(bot, ex.controller_id); - const leaf = leafFromExecutor(ex, bot, att.runKey, att.how); - if (windowCutoff && leaf.endedAt !== null && leaf.endedAt < windowCutoff) continue; - all.push(leaf); - } - // The controllers those runs left behind. This is the spine of the - // terminated tree, exactly as the live controllers are the spine of the - // running one: a controller record covers every executor it ever ran, - // including the ones that closed long before the bounded executor walk - // reaches, so a finished bot reports what it actually did rather than - // whatever fraction of its executors is still in the table. - for (const ctrl of terminatedControllers) { - const att = agentOf(ctrl.bot_name, ""); - const leaf = leafFromTerminatedController( - ctrl, - runByBot.get(ctrl.bot_name), - att.runKey, - att.how, - ); - if (windowCutoff && leaf.endedAt !== null && leaf.endedAt < windowCutoff) continue; - all.push(leaf); - } + return runningLeaves({ controllers, executors, owners, deeds, botByController }); } - - return all; + // Shared with the execution dock's per-agent history for the same reason. + return terminatedLeaves({ + executors, + terminatedControllers, + runs, + owners, + deeds, + botByController, + windowCutoff, + attribute, + runByBot, + }); }, [ controllers, @@ -959,6 +899,7 @@ export function PerfBrowser({ terminatedControllers, botByController, attribute, + runs, owners, deeds, windowCutoff, diff --git a/frontend/src/hooks/useFleetData.ts b/frontend/src/hooks/useFleetData.ts index 01c63e4c8..0b02f507e 100644 --- a/frontend/src/hooks/useFleetData.ts +++ b/frontend/src/hooks/useFleetData.ts @@ -101,9 +101,19 @@ export function useFleetData( population, enabled = true, history = true, + terminated = population === "terminated", }: { population: Population; enabled?: boolean; + /** + * Whether to read what has finished — the runs and the controllers they + * left behind — alongside the live fleet. + * + * Follows the population by default: `/bots` reads it only on its + * Terminated side. The execution dock folds the running population *and* + * credits each agent with its stopped bots, so it asks for both. + */ + terminated?: boolean; /** * Whether to walk the fleet's performance history (ARCH-324). * @@ -310,7 +320,7 @@ export function useFleetData( const { data: runsData } = useQuery({ queryKey: ["bot-runs", server], queryFn: () => api.getBotRuns(server!, { limit: 200 }), - enabled: on && population === "terminated", + enabled: on && terminated, refetchInterval: 30_000, }); @@ -325,7 +335,7 @@ export function useFleetData( const { data: terminatedData } = useQuery({ queryKey: ["terminated-controllers", server], queryFn: () => api.getTerminatedControllers(server!), - enabled: on && population === "terminated", + enabled: on && terminated, refetchInterval: 60_000, staleTime: 30_000, }); diff --git a/frontend/src/lib/perf-population.ts b/frontend/src/lib/perf-population.ts index cbfb56274..af52c4aae 100644 --- a/frontend/src/lib/perf-population.ts +++ b/frontend/src/lib/perf-population.ts @@ -7,24 +7,27 @@ // construction has to be shared rather than copied. A copy would make the two // screens agree by coincidence and drift the first time a leaf gains a field. // -// The terminated population deliberately stays in the browser: it needs the -// run-window attributor, the period cutoff and the finished-controller records, -// none of which the Money view has or wants. What is lifted is the half that -// two hosts genuinely share. +// The terminated population followed for the same reason: the execution dock +// credits each agent with what its stopped bots and closed executors earned, and +// that has to be the number `/bots?population=terminated` folds for the same +// agent — so `terminatedLeaves` is the browser's construction, lifted, not a +// second one written beside it. // // Nothing here fetches and nothing here renders (the ARCH-300 split). import { attributionIndex, type DeedIndex, type FleetOwner } from "@/lib/agent-attribution"; -import type { ControllerInfo, ExecutorInfo } from "@/lib/api"; -import { isExecutorActive } from "@/lib/formatters"; +import type { BotRunInfo, ControllerInfo, ExecutorInfo } from "@/lib/api"; +import { isExecutorActive, toMs } from "@/lib/formatters"; import { UNATTACHED_BOT, leafFromController, leafFromExecutor, + leafFromTerminatedController, type ConvertQuote, type PerfLeaf, } from "@/lib/perf-tree"; import type { ConvertFn } from "@/lib/rates"; +import { buildAttributor, runWindows, type Attributor } from "@/lib/run-attribution"; /** * `useRates`'s converter, in the shape the fold takes. @@ -107,3 +110,96 @@ export function runningLeaves({ } return all; } + +/** + * The run behind each bot name, for the two things a controller record cannot + * say: when its bot stopped, and which archive it left. + * + * A bot name carries its own deploy timestamp, so it is unique per run in + * practice; where it is not, the most recent run wins, which is the one the + * finished controllers on screen belong to. + */ +export function latestRunByBot(runs: readonly BotRunInfo[]): Map { + const latest = new Map(); + for (const run of runs) { + const seen = latest.get(run.bot_name); + if (!seen || (run.created_at ?? "") > (seen.created_at ?? "")) latest.set(run.bot_name, run); + } + return latest; +} + +/** + * Everything that has finished, in the browser's one vocabulary: every closed + * executor, plus the controllers the finished runs left behind. + * + * A closed executor is filed under the run that **opened** it, asked through + * the run-window attributor at its own start time — a position that outlived + * its bot's stop still belongs to the run that opened it. The live fleet is the + * fallback for an executor of a bot that is still deployed (it has no closed + * window to sit in), and `(unattached)` for one that belongs to no run at all, + * which on a real server is every hand-opened position. + * + * `windowCutoff` (epoch ms, `0` for no window) drops what finished before it: + * a period called "the last week" is the trading that *stopped* in it. + * + * `attribute` and `runByBot` are taken from a caller that already memoised + * them and built from `runs` otherwise. + */ +export function terminatedLeaves({ + executors, + terminatedControllers, + runs, + owners, + deeds, + botByController, + windowCutoff = 0, + attribute = buildAttributor(runWindows(runs)), + runByBot = latestRunByBot(runs), +}: { + executors: readonly ExecutorInfo[]; + terminatedControllers: readonly ControllerInfo[]; + runs: readonly BotRunInfo[]; + owners: FleetOwner[]; + deeds: DeedIndex | null; + /** The live fleet's controller → bot map — see {@link botsByController}. */ + botByController: Map; + windowCutoff?: number; + attribute?: Attributor; + runByBot?: Map; +}): PerfLeaf[] { + const all: PerfLeaf[] = []; + // One index for the whole walk (PERF-331): this population walks every + // executor the fleet has ever had. + const agentOf = attributionIndex(owners, deeds); + const closedBotOf = (ex: ExecutorInfo, startedAt: number | null) => { + if (!ex.controller_id) return UNATTACHED_BOT; + const at = startedAt ?? (ex.close_timestamp > 0 ? toMs(ex.close_timestamp) : null); + const owner = at === null ? null : attribute(ex.controller_id, at); + return owner ?? botByController.get(ex.controller_id) ?? UNATTACHED_BOT; + }; + for (const ex of executors) { + if (isExecutorActive(ex.status)) continue; + const started = ex.timestamp > 0 ? toMs(ex.timestamp) : null; + const bot = closedBotOf(ex, started); + const att = agentOf(bot, ex.controller_id); + const leaf = leafFromExecutor(ex, bot, att.runKey, att.how); + if (windowCutoff && leaf.endedAt !== null && leaf.endedAt < windowCutoff) continue; + all.push(leaf); + } + // The spine of the terminated tree, exactly as the live controllers are the + // spine of the running one: a controller record covers every executor it + // ever ran, including the ones that closed long before the bounded executor + // walk reaches. + for (const ctrl of terminatedControllers) { + const att = agentOf(ctrl.bot_name, ""); + const leaf = leafFromTerminatedController( + ctrl, + runByBot.get(ctrl.bot_name), + att.runKey, + att.how, + ); + if (windowCutoff && leaf.endedAt !== null && leaf.endedAt < windowCutoff) continue; + all.push(leaf); + } + return all; +} diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/scripts/hackathon_agents.example.yml b/scripts/hackathon_agents.example.yml new file mode 100644 index 000000000..9ccb3ca59 --- /dev/null +++ b/scripts/hackathon_agents.example.yml @@ -0,0 +1,43 @@ +# Agent map for scripts/hackathon_monitor.py (HACKATHON_AGENT_MAP). +# +# Race agent id (the strategy slug the leaderboard uses) -> the Condor run keys +# (`agent.strategy`) whose bots and executors make up that agent. Copy this +# file, check every key against the strategies installed on this host: +# +# uv run python -c "from condor.agents.strategy import StrategyStore; [print(s.key) for s in StrategyStore().list_all()]" +# +# and fill in the empty ones. An id mapped to [] is measured as nothing and +# logged every minute until it is filled. +# +# The 14 agents of agent-builders-cup-1 as of 2026-10-03. Keys were read from +# each submission's AGENT.md / loops//loop.md and assume the agent was +# installed under its own folder name. + +# xrpl +aegis-defensive-market-making: [aegis.aegis_operator] +delta-raptor-volume-hunting-market-maker: [delta_raptor.rlusd_xrp_maker] + +# botcamp +churn: [stable_churn_operator.stable_churn_supervisor] +# Submitted as a Hummingbot controller (mm_at_the_touch.py), not a Condor agent. +cohort-13-market-making-at-the-touch: [] + +# bitget +quad-threat: [quad_threat_orchestrator.orchestrate] +midas-market-maker-w-adverse-selection-detection-dynamic-hedging-position-sizing-self-correcting-logic: [midas.midas_hybrid_operator] + +# gate +gateforum-multi-agent-research-decision-trader: [gateforum.gateforum_debate_operator] +self-improving-market-making-fleet: [market_making_cup.mm_cup_operator] + +# orca +adaptive-orca-lp-agent: [adaptive_orca_lp.adaptive_lp_controller_loop] +trend-aware-lp-rebalancer-agent: [trend_aware_lp_rebalancer_agent.orca] + +# meteora +regime-switch-lp: [regime_switch_lp.regime_operator] +jev-decision-maker-at-meteora: [jev.jev_desk] + +# derive +flyby-long-volatility-collector-desk: [flyby.flyby_operator] +theta-covered-call-hedge-desk: [theta.theta_funding_desk] diff --git a/scripts/hackathon_monitor.py b/scripts/hackathon_monitor.py new file mode 100644 index 000000000..3e1137855 --- /dev/null +++ b/scripts/hackathon_monitor.py @@ -0,0 +1,1253 @@ +"""Report every hackathon agent's PnL and volume to the race API, once a minute. + +The race API names its agents by **strategy slug**. Each one is resolved to the +Condor strategies with that slug. Bot PnL and volume are the portfolio page's +Bots tile: every running controller on the live bot status, realized plus +unrealized, and the volume it reports, attributed by the same namespace, deed +and extra-bot bases the bots page uses. Standalone executors its sessions +created (``controller_id = "{agent}.{strategy}_{N}"``) are added from +``fetch_agent_performance_batch``. A non-USD executor quote is restated in USD +there; a live controller is already the quote the portfolio page sums. + +Run it from the repo root, where ``config.yml`` and ``.env`` live:: + + uv run python scripts/hackathon_monitor.py --dry-run --once # look first + uv run python scripts/hackathon_monitor.py --start-race # at the gun + uv run python scripts/hackathon_monitor.py # after a restart + +``--start-race`` records what every agent had already made and traded, and that +baseline is subtracted from every later post, so the race counts from its start +and not from each strategy's first test run. It is kept in the state file and +survives a restart; without one the lifetime figures are posted as they are. + +When a Condor strategy's slug differs from the race id, name it in an agent map: +a YAML file of race agent id → Condor run keys (``agent.strategy``). An id in +the map is measured by those keys only; any other id is matched by slug. A bot +deployed outside that run's namespace, with no deed, is named under ``bots`` +and added to the same total. The file is re-read every minute, so an edit +takes effect without a restart:: + + churn: [stable_churn_operator.stable_churn_supervisor] + adaptive-orca-lp: [alice.orca_lp, bob.orca_lp] + delta-raptor: + runs: [delta_raptor.rlusd_xrp_maker] + bots: [delta_raptor_usd_desk_sit] + +Environment (``.env`` is read): + +- ``HACKATHON_API_URL`` base URL of the race site, e.g. ``https://example.org`` +- ``HACKATHON_API_TOKEN`` bearer token +- ``HACKATHON_SLUG`` defaults to ``agent-builders-cup-1`` +- ``HACKATHON_SERVERS`` comma-separated Condor server names; defaults to + every server in ``config.yml`` +- ``HACKATHON_AGENT_MAP`` path of the agent map; none by default. Start from + ``scripts/hackathon_agents.example.yml``, which lists the field + +An agent's figure is the sum over all of those servers. A server that stops +answering keeps contributing the last figure it gave (remembered in the state +file, so a restart does not lose it); one that has never answered counts as +nothing and is named in the log every minute. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import os +import sys +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable + +# Run as a file (``python scripts/hackathon_monitor.py``) the repo root is not +# on the path, and everything below imports from it. +ROOT = Path(__file__).resolve().parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +log = logging.getLogger("hackathon_monitor") + +DEFAULT_SLUG = "agent-builders-cup-1" +DEFAULT_INTERVAL = 60.0 +DEFAULT_STATE = ".condor/hackathon/{slug}.json" +HTTP_TIMEOUT = 20.0 +# One server's whole fetch. A dead host must cost the minute a bounded wait, not +# the other servers' figures. +SERVER_TIMEOUT = 45.0 + + +@dataclass(frozen=True) +class Totals: + """One agent's cumulative figures, in quote.""" + + pnl: float = 0.0 + volume: float = 0.0 + + def minus(self, other: "Totals") -> "Totals": + return Totals(self.pnl - other.pnl, self.volume - other.volume) + + def plus(self, other: "Totals") -> "Totals": + return Totals(self.pnl + other.pnl, self.volume + other.volume) + + +# ── The race API ── + + +def race_url(base_url: str, slug: str) -> str: + return f"{base_url.rstrip('/')}/api/hackathons/{slug}/race-data" + + +def parse_agent_ids(payload: Any) -> list[str]: + """The agent ids out of the GET response, in race order. + + The site answers ``{"data": {"agents": [{"agent_id": ...}, ...]}}``; any + other shape is an error rather than an empty field. + """ + try: + return [agent["agent_id"] for agent in payload["data"]["agents"]] + except (KeyError, TypeError) as exc: + raise RuntimeError( + f"Unexpected race-data response: {json.dumps(payload)[:300]}" + ) from exc + + +def parse_post_result(payload: Any) -> tuple[int, list[str]]: + """``(accepted, unknown ids)`` out of the POST response.""" + try: + data = payload["data"] + return data["accepted"], data["unknown"] + except (KeyError, TypeError) as exc: + raise RuntimeError( + f"Unexpected race-data POST response: {json.dumps(payload)[:300]}" + ) from exc + + +def build_payload( + agent_ids: Iterable[str], totals: dict[str, Totals], now: float +) -> dict[str, Any]: + """The POST body. An agent with no figure this minute is left out.""" + stamp = datetime.fromtimestamp(now, tz=timezone.utc) + return { + "timestamp": stamp.isoformat(timespec="seconds").replace("+00:00", "Z"), + "agents": [ + { + "agent_id": agent_id, + "pnl_quote": round(totals[agent_id].pnl, 6), + "volume_quote": round(totals[agent_id].volume, 6), + } + for agent_id in agent_ids + if agent_id in totals + ], + } + + +async def fetch_agent_ids(session: Any, url: str) -> list[str]: + async with session.get(url) as response: + response.raise_for_status() + return parse_agent_ids(await response.json()) + + +async def post_race_data( + session: Any, url: str, payload: dict[str, Any] +) -> tuple[int, list[str]]: + async with session.post(url, json=payload) as response: + if response.status >= 400: + body = (await response.text())[:300] + raise RuntimeError(f"POST {url} answered {response.status}: {body}") + return parse_post_result(await response.json()) + + +# ── Who is who ── + + +@dataclass(frozen=True) +class AgentMap: + """Race agent id → the run keys to measure, and bot bases to add. + + ``runs`` is the bots-page join (namespace, declared names, deeds). ``bots`` + names bases that join matches nothing for: deployed beside the run, outside + its namespace, and never written into a deed. + """ + + runs: dict[str, list[str]] + bots: dict[str, list[str]] + + +def _string_list(value: Any, label: str) -> list[str]: + if isinstance(value, str): + return [value] + if isinstance(value, list) and all(isinstance(item, str) for item in value): + return list(value) + raise RuntimeError(f"{label} must be a string or a list of strings") + + +def load_agent_map(path: Path) -> AgentMap: + """Race agent id → Condor run keys and extra bot bases, from the YAML file. + + A string or a list is run keys, as before. A mapping may set ``runs`` and + ``bots``; anything else is rejected so a typo does not silently drop a key. + """ + import yaml + + raw = yaml.safe_load(path.read_text()) + if not isinstance(raw, dict): + raise RuntimeError(f"{path}: expected a mapping of agent id to run keys") + runs: dict[str, list[str]] = {} + bots: dict[str, list[str]] = {} + for agent_id, spec in raw.items(): + aid = str(agent_id) + if isinstance(spec, (str, list)): + runs[aid] = _string_list(spec, f"{path}: {aid}") + bots[aid] = [] + continue + if isinstance(spec, dict) and set(spec).issubset({"runs", "bots"}): + runs[aid] = _string_list(spec.get("runs") or [], f"{path}: {aid}.runs") + bots[aid] = _string_list(spec.get("bots") or [], f"{path}: {aid}.bots") + continue + raise RuntimeError( + f"{path}: {aid} must map to a run key, a list of them, or runs/bots" + ) + return AgentMap(runs=runs, bots=bots) + + +def deed_bases_by_run() -> dict[str, list[str]]: + """Bot bases the bots page credits to each run, from the deed index. + + The same last join ``/bots`` uses: a recorded deploy, not a name. A bot + inside a namespace is already covered by the owner's namespace and is + listed again here only when the index says so; callers dedupe. + """ + from condor.agents.deed_index import build_deed_index + + out: dict[str, list[str]] = {} + for base, ref in build_deed_index().bots.items(): + bases = out.setdefault(ref.run_key, []) + if base not in bases: + bases.append(base) + return out + + +def match_owners( + agent_ids: Iterable[str], + owners: Iterable[Any], + agent_map: dict[str, list[str]] | None = None, +) -> dict[str, list]: + """Each race agent id → the Condor strategies it names. + + An id in ``agent_map`` names exactly the run keys listed there, including a + chat or delegation: the bots page credits a bot to whichever run deployed + it, and a delegation's namespace is empty. Any other id is a strategy slug, + so it can match under more than one agent; all of them count. A full run + key (``agent.strategy``) is accepted too. An unmapped pseudo-run is never a + race agent — slug matching still requires a namespace. + """ + agent_map = agent_map or {} + named = {key for keys in agent_map.values() for key in keys} + owners = [ + owner for owner in owners if owner.namespace or owner.run_key in named + ] + matched: dict[str, list] = {} + for agent_id in agent_ids: + if agent_id in agent_map: + keys = agent_map[agent_id] + matched[agent_id] = [owner for owner in owners if owner.run_key in keys] + else: + matched[agent_id] = [ + owner + for owner in owners + if agent_id in (owner.strategy_slug, owner.run_key) + ] + return matched + + +async def collect_totals( + client: Any, + matched: dict[str, list], + extra_bots: dict[str, list[str]] | None = None, + ledger: dict[str, dict[str, float]] | None = None, + server: str = "", +) -> tuple[dict[str, Totals], set[str], dict[str, dict[str, float]]]: + """Lifetime PnL and volume per race agent, and which ones came back degraded. + + Bot figures are the portfolio page's Bots tile (live controller status). + Executor figures still come from one batch over the session tags: the two + sets are disjoint, so the sum is the bots on screen plus every standalone + executor the sessions created. A client with no bot status — the unit-test + double — keeps the batch's snapshot merge instead. + + Which bots, is the bots page's rule (``attributionIndex``): the namespace, + any declared legacy name, and the bots the deed index recorded for this run + — the only way a delegation's ``quad-e1`` is credited, since the name proves + nothing. ``extra_bots`` adds bases that rule does not see. Every base for one + race agent is partitioned together, so a short extra name such as + ``midas_vol_race`` does not also swallow ``midas_vol_race-…``. An agent with + no matching strategy and no named bots has traded nothing and reads zero. + """ + from condor.agents.performance import fetch_agent_performance_batch + + extra_bots = extra_bots or {} + deeds = deed_bases_by_run() + tags: dict[str, list[str]] = {} + bases: dict[str, list[str]] = {} + for agent_id, owners in matched.items(): + owner_tags: list[str] = [] + seen: list[str] = [] + for owner in owners: + these = list(owner.agent_ids) or [owner.run_key] + owner_tags.extend(these) + for base in ( + owner.namespace, + *owner.declared_bots, + *deeds.get(owner.run_key, ()), + ): + if base and base not in seen: + seen.append(base) + for base in extra_bots.get(agent_id, []): + if base and base not in seen: + seen.append(base) + if not owner_tags and seen: + owner_tags.append(agent_id) + tags[agent_id] = owner_tags + if seen and owner_tags: + bases[owner_tags[0]] = seen + + all_tags = [tag for owner_tags in tags.values() for tag in owner_tags] + failed: set[str] = set() + try: + live = await read_portfolio_controllers(client) + except Exception as exc: # noqa: BLE001 - the snapshot still answers + log.warning("Portfolio bot status failed (%s); using the snapshot", exc) + live = None + # This tick's per-bot snapshot, and the set of bots currently live. Both are + # taken before any retired generation is added, so a bot is never counted as + # both live and retired. + snapshots = bot_rows_by_name(live) if live is not None else {} + # Live status is the portfolio page. Passing the bases into the batch as + # well would add the snapshot on top of it. + perf = await fetch_agent_performance_batch( + client, all_tags, None if live is not None else bases, failed_ids=failed + ) + + totals: dict[str, Totals] = {} + degraded: set[str] = set() + for agent_id, owner_tags in tags.items(): + batch_pnl = sum(perf[tag].total_pnl for tag in owner_tags if tag in perf) + batch_vol = sum(perf[tag].volume for tag in owner_tags if tag in perf) + totals[agent_id] = Totals(pnl=batch_pnl, volume=batch_vol) + live_tot = Totals() + retired_tot = Totals() + if live is not None and owner_tags: + live_tot = portfolio_bot_totals(live, bases.get(owner_tags[0], [])) + totals[agent_id] = totals[agent_id].plus(live_tot) + # ...plus every generation this agent has retired. Wrapped so a bad + # archive can never cost the live figure: the worst case is the old + # understated number, never a failed tick. + try: + retired_tot = await inactive_bot_totals( + client, + bases.get(owner_tags[0], []), + ledger if ledger is not None else {}, + server, + snapshots, + ) + totals[agent_id] = totals[agent_id].plus(retired_tot) + except Exception as exc: # noqa: BLE001 + log.warning( + "Retired-generation accounting failed for %s (%s)", agent_id, exc + ) + if os.environ.get("HACKATHON_DEBUG_RETIRED"): + log.warning( + "[breakdown] %-52s batch(pnl=%+.4f vol=%10.2f) live(pnl=%+.4f " + "vol=%10.2f) retired(pnl=%+.4f vol=%10.2f)", + agent_id, + batch_pnl, + batch_vol, + live_tot.pnl, + live_tot.volume, + retired_tot.pnl, + retired_tot.volume, + ) + if failed.intersection(owner_tags): + degraded.add(agent_id) + return totals, degraded, snapshots + + +# ── Direct-venue agents (no Hummingbot bot/controller/executor at all) ── +# +# theta's order path (``theta_hedge``) signs locally against Derive's v3 API +# and never creates a Hummingbot bot, controller, or executor — so neither of +# ``collect_totals``'s two sources (the Bots tile, the executor-tag batch) can +# ever see its trades, by construction, not by a missing agent-map entry. This +# reads the venue directly for the one subaccount theta trades from and +# returns its lifetime fill volume + realized PnL in the same ``Totals`` shape +# everything else in this file already uses, so it drops into the normal +# baseline/since-start/state-file machinery unmodified. +# +# One venue read, scoped to one agent id. A failure here must not cost the +# other 13 agents their tick (same rule ``_read_server`` already applies per +# server), so the caller wraps this in a try/except and logs rather than +# raises. +DIRECT_VENUE_AGENTS: dict[str, dict[str, Any]] = { + "theta-covered-call-hedge-desk": { + "keystore": "/home/hummingbot/condor/bots/credentials/theta/derive_perpetual.yml", + "subaccount_id": 107967, + }, + # flyby places Derive orders from its own routine, so it has no Condor + # controller or executor for the bot-status path to read: its fills are only + # visible on the venue. Measured here from the same subaccount the routine signs with. + "flyby-long-volatility-collector-desk": { + "keystore": "/home/hummingbot/condor/bots/credentials/master_account/connectors/derive_perpetual.yml", + "subaccount_id": 74536, + }, +} + + +async def fetch_direct_venue_totals() -> dict[str, Totals]: + """Lifetime fill volume + current PnL for agents with no Condor-visible bot. + + Volume is every fill's ``trade_price * trade_amount`` from + ``private/get_trade_history``, summed across pages. PnL is the account's + present figure, not just what has settled: ``realized_pnl`` from the trade + history (closed legs) plus ``unrealized_pnl`` from each OPEN position via + ``private/get_positions`` (mark-to-market on anything still held). + + Funding (``cumulative_funding`` / ``pending_funding``) is deliberately left + out of PnL. Verified against a live position (identity closes to the last + digit): ``unrealized_pnl == unrealized_pnl_excl_fees - total_fees``, with + no funding term anywhere in it — Derive's own schema confirms funding is + tracked separately, as a margin-calculation input, not a PnL component + (``pending_funding`` is "added to the portfolio value for margin + calculations"; ``net_settlements`` is unrelated settlement cashflow, + "subtracted from the portfolio value for margin calculations", not + funding). So funding genuinely sits outside unrealized_pnl rather than + being folded into it — the reason to exclude it here is that it is not, + by itself, the account's P&L: a short can accrue positive funding while + the position itself is underwater on price, which funding-only would + misreport as green. At the size seen on theta's book (sub-cent) it would + not matter either way, but the number must stay right as it scales. + Empty dict (not an exception) if nothing is configured. + """ + if not DIRECT_VENUE_AGENTS: + return {} + + import sys as _sys + from pathlib import Path as _Path + + theta_routines = _Path( + "/home/hummingbot/condor/.condor/agents/theta/routines" + ) + theta_root = theta_routines.parent + for p in (str(theta_routines), str(theta_root)): + if p not in _sys.path: + _sys.path.insert(0, p) + from _theta_derive import from_condor_keystore, _post # noqa: E402 + + out: dict[str, Totals] = {} + for agent_id, spec in DIRECT_VENUE_AGENTS.items(): + creds = from_condor_keystore(keystore=spec["keystore"]) + sub_id = int(spec["subaccount_id"]) + volume = 0.0 + pnl = 0.0 + page = 1 + while True: + resp = await _post( + creds, + "get_trade_history", + {"subaccount_id": sub_id, "page": page, "page_size": 1000}, + ) + trades = resp.get("trades") or [] + for t in trades: + volume += abs(float(t.get("trade_price") or 0)) * abs( + float(t.get("trade_amount") or 0) + ) + pnl += float(t.get("realized_pnl") or 0) + pagination = resp.get("pagination") or {} + num_pages = int(pagination.get("num_pages") or 1) + if page >= num_pages: + break + page += 1 + positions = await _post( + creds, "get_positions", {"subaccount_id": sub_id} + ) + for pos in positions.get("positions") or []: + pnl += float(pos.get("unrealized_pnl") or 0) + out[agent_id] = Totals(pnl=pnl, volume=volume) + return out + + +async def read_portfolio_controllers(client: Any) -> list[dict] | None: + """Controllers the portfolio page's Bots tile sums, or ``None`` if unavailable. + + ``None`` means this client has no bot-status call (the test double). An + empty list means the server has no running bot, which is what the page shows. + """ + getter = getattr( + getattr(client, "bot_orchestration", None), "get_active_bots_status", None + ) + if getter is None: + return None + from condor.fetchers.bots import build_bots_page + + page = build_bots_page(await getter()) + return list(page.get("controllers") or []) + + +def portfolio_bot_totals(controllers: list[dict], bases: list[str]) -> Totals: + """Sum the portfolio page's controller rows that these bases own. + + Longest base wins, the same partition the bots page uses, so a namespace + and a longer tagged sibling of it are not both credited. + """ + from condor.fetchers.bot_performance import partition_instances + + by_name: dict[str, Totals] = {} + for ctrl in controllers: + name = str(ctrl.get("bot_name") or "") + if not name: + continue + row = Totals( + float(ctrl.get("global_pnl_quote") or 0), + float(ctrl.get("volume_traded") or 0), + ) + by_name[name] = by_name.get(name, Totals()).plus(row) + assigned = partition_instances({name: {} for name in by_name}, bases) + total = Totals() + for names in assigned.values(): + for name in names: + total = total.plus(by_name[name]) + return total + + +# -- Bots that are no longer live: archived runs, and stopped-but-unarchived -- +# +# A race agent's figure is not the figure of the bot it happens to be running +# this minute. Every redeploy -- the agent's own cold start, a drawdown kill, an +# operator revision -- starts the new bot's counters at zero, and the retired +# bot's lifetime contribution used to vanish from the total. That dropped the +# reported volume AND hid the loss the retired sleeve had already realised +# (quad-e4 stopped on its own drawdown cap and its -$39.26 disappeared, leaving +# the entry reading +$1.87 profit). So an agent total is now: +# +# live bots + every retired generation +# +# Two sources, and they are not interchangeable: +# +# 1. ARCHIVED runs. Authoritative and immutable: the sqlite the stopped bot +# left behind, read once per process (``fetch_archived_run`` caches it). +# PnL is the sum of each executor's ``net_pnl_quote`` -- the same +# fee-inclusive net the live fleet reports -- and NOT the model's +# ``total_pnl``. ``total_pnl`` is not gross: condor/archived_pnl.py charges +# a fee per fill, but on the sell leg only, so it is net of roughly HALF the +# fee load. On the r4 archive the fills' fees were BUY 0.575221 + SELL +# 0.513049 = 1.088270 and total_pnl read +0.660335 -- gross minus the sell +# legs alone; charging the buy legs too gives +0.085114, which still is not +# the executors' -0.515330. The two are different estimators of one history, +# not two conventions for one, and the executors' net is the one that +# matches what the board already published for that bot while it ran. +# +# Volume comes from the trade walk (``total_volume``); PnL comes from the +# executor rows. That mix is deliberate: each mirrors what the live path +# publishes (``volume_traded`` is trade-derived, the live PnL is fee-net per +# controller) and each is the reliable source for its own metric. The +# executor table's volume column is not usable -- a closed executor was seen +# there with filled_amount_quote 0.0 beside non-zero fees. Do not expect the +# two to agree; HACKATHON_DEBUG_RETIRED prints both side by side. +# +# 2. STOPPED-but-unarchived runs. No route can read their sqlite: the archived +# reader refuses any path outside ``bots/archived/``, and +# ``/bot-orchestration//history`` asks the live container (commlib), +# which answers with empty trades once the strategy has stopped. The only +# surviving record is the last snapshot this monitor took while that bot was +# still live, so ``bot_ledger`` keeps one per bot name in the state file. +# +# UNREALIZED is never taken from a retired generation. The inventory carries +# into the live fleet, so crediting a retired generation's unrealized would count +# one open position twice and double the loss at every generation change. + + +def realized_usd(executor: Any) -> float: + """One retired executor's realized PnL, in USD. + + Two corrections a naive sum gets wrong: + + - Only CLOSED executors are realized. ``net_pnl_quote`` on an executor that + never closed is a mark-to-market, and crediting that as realized would put + unrealized into a retired generation -- the one thing this design must not + do. Filter on ``close_timestamp``: ``status`` is the raw numeric enum out of + the DB (every executor on api-2 reads "4"), so a status check would match + nothing and silently let everything through. + - ``usd_rate``. Every other figure from this fetch is restated in USD -- + ``convert_trades_to_usd`` runs before the volume is totalled -- but the + executors keep their native quote, and upstream's own + ``calculate_pnl_from_executors`` multiplies by the rate. Without it a + generation quoted in anything but USD posts a native-quote PnL beside a USD + volume, off by the FX rate. + """ + if not float(getattr(executor, "close_timestamp", 0.0) or 0.0): + return 0.0 + pnl = float(getattr(executor, "pnl", 0.0) or 0.0) + return pnl * float(getattr(executor, "usd_rate", 1.0) or 1.0) + + +def bot_rows_by_name(controllers: list[dict]) -> dict[str, dict[str, float]]: + """Per live bot name: its rows' {volume, realized} - the ledger snapshot.""" + out: dict[str, dict[str, float]] = {} + for ctrl in controllers: + name = str(ctrl.get("bot_name") or "") + if not name: + continue + row = out.setdefault(name, {"volume": 0.0, "realized": 0.0}) + row["volume"] += float(ctrl.get("volume_traded") or 0) + row["realized"] += float(ctrl.get("realized_pnl_quote") or 0) + return out + + +# Archive listings, held briefly. The listing is one cheap call, but it fired +# once per agent per server per tick (14x); more to the point, its UNION across +# servers is what the ledger is filtered against, so a generation archived on one +# server is never also credited from another server's snapshot in the same tick. +_ARCHIVE_PATHS: dict[str, tuple[float, list[str]]] = {} +_ARCHIVE_PATHS_TTL = 120.0 + + +def clear_archive_paths_cache() -> None: + """Forget the cached listings, so a test (or a re-pointed client) re-lists.""" + _ARCHIVE_PATHS.clear() + + +def fleet_archived_names() -> set[str]: + """Every bot name known to hold an archive, on any server.""" + from condor.fetchers.archived_run import extract_bot_name + + names: set[str] = set() + for _at, paths in _ARCHIVE_PATHS.values(): + names |= {extract_bot_name(p) for p in paths} + return names + + +async def listed_archive_paths(client: Any, server: str) -> list[str]: + """One server's archive db paths, from cache when listed recently.""" + now = time.monotonic() + cached = _ARCHIVE_PATHS.get(server) + if cached is not None and now - cached[0] < _ARCHIVE_PATHS_TTL: + return list(cached[1]) + reader = getattr(client, "archived_bots", None) + if reader is None: + return [] + try: + dbs = await reader.list_databases() + except Exception as exc: # noqa: BLE001 - no archive list is not a failed tick + log.warning("Archived-bot list failed on %s (%s)", server, exc) + # Fall back to the last good listing rather than crediting nothing: a + # failed listing must never make a generation disappear for a tick. + return list(cached[1]) if cached else [] + if not isinstance(dbs, list): + return [] + paths = [d for d in dbs if isinstance(d, str)] + _ARCHIVE_PATHS[server] = (now, paths) + return list(paths) + + +async def archived_bot_rows(client: Any, server: str) -> dict[str, Any]: + """{bot name: ArchivedBotPerformance} for every archive on one server.""" + from condor.fetchers.archived_run import extract_bot_name, fetch_archived_run + + out: dict[str, Any] = {} + for db_path in await listed_archive_paths(client, server): + name = extract_bot_name(db_path) + try: + out[name] = await fetch_archived_run(client, server, db_path) + except Exception as exc: # noqa: BLE001 - one bad archive, not the server + log.warning("Archived run %s on %s unreadable (%s)", name, server, exc) + return out + + +async def inactive_bot_totals( + client: Any, + bases: list[str], + ledger: dict[str, dict[str, float]], + server: str, + live_rows: dict[str, dict[str, float]], +) -> Totals: + """Volume + net realized PnL from this agent's retired generations. + + The ledger is read, never mutated: an entry whose bot also has an archive is + SKIPPED here (the archive supersedes it), and leaving the record in place + means a later agent's failed listing cannot make a generation vanish + mid-tick. + """ + from condor.fetchers.bot_performance import partition_instances + + def _live_volume(rows: dict[str, dict[str, float]], name: str) -> float: + return float((rows.get(name) or {}).get("volume") or 0.0) + + def reporting(name: str) -> bool: + """Is this bot still giving a real live reading? + + A bot the API still lists whose strategy has died reports zeros. That is + the absence of a reading, not a reading of zero, so a retired generation + must not be dismissed as live on the strength of it. + """ + row = live_rows.get(name) or {} + return bool(row.get("volume") or row.get("realized")) + + archived = await archived_bot_rows(client, server) + superseded = set(archived) | fleet_archived_names() + + rows: dict[str, Totals] = {} + for name, perf in archived.items(): + executors = perf.executors or [] + closed = [ + e for e in executors if float(getattr(e, "close_timestamp", 0.0) or 0.0) + ] + total_volume = float(getattr(perf, "total_volume", 0.0) or 0.0) + pnl = sum(realized_usd(e) for e in executors) + if total_volume <= 0.0 and closed: + # A generation that closed executors cannot have traded nothing, so a + # zero here means the trade walk came back empty or truncated -- the + # real way this figure goes wrong. (A divergence from the executors' + # own filled column is NOT a warning sign: the two legitimately span + # 0.46x to 3.7x on the archives we have.) + log.warning( + "Archived run %s on %s credits %d closed executor(s) but reports " + "trade volume 0 (stats_source=%s): the walk looks truncated", + name, + server, + len(closed), + getattr(perf, "stats_source", "?"), + ) + # An archive and a live row sharing a name are two generations IF the live + # counters restarted: cumulative volume cannot fall within one generation, + # so a live reading below the archive's total means a redeploy under a + # reused name and the archive must still be counted. Only once the live + # reading has caught up is "already counted live" the safe read. + if reporting(name) and _live_volume(live_rows, name) >= total_volume: + continue + if reporting(name): + # Credited from the archive AND from the live fleet this tick. This is + # the one double count the design accepts, and it is deliberately + # LOUD: the volume comparison cannot separate a fresh generation that + # has traded less (the reuse case this rule exists for) from a stale + # partial reading of the old one, and no rule is sound in both + # directions. Counting the archive is the point of the change, so the + # live row stands too -- do NOT "fix" it by dropping the name from the + # live sum, which under-counts the genuine reuse case. + log.warning( + "Archived run %s on %s is ALSO live (live volume %.2f < archive " + "%.2f): this name is credited from both the archive and the live " + "fleet this tick", + name, + server, + _live_volume(live_rows, name), + total_volume, + ) + rows[name] = Totals(pnl=pnl, volume=total_volume) + if os.environ.get("HACKATHON_DEBUG_RETIRED"): + log.warning( + "[retired-src] %s %s stats_source=%s executors=%d closed=%d " + "trade_vol=%.2f", + server, + name, + getattr(perf, "stats_source", "?"), + len(executors), + len(closed), + total_volume, + ) + for name, snap in ledger.items(): + if reporting(name) or name in superseded: + continue + rows[name] = Totals( + pnl=float(snap.get("realized") or 0.0), + volume=float(snap.get("volume") or 0.0), + ) + if not rows: + return Totals() + + assigned = partition_instances({name: {} for name in rows}, bases) + total = Totals() + for names in assigned.values(): + for name in names: + total = total.plus(rows[name]) + if os.environ.get("HACKATHON_DEBUG_RETIRED"): + # Audit line: which retired generations were credited, and to whom. + log.warning( + "[retired] %s bases=%s | %s | TOTAL pnl=%+.4f vol=%.2f", + server, + bases, + "; ".join( + "%s(pnl=%+.4f trade_vol=%.2f)" % (n, rows[n].pnl, rows[n].volume) + for n in sorted(rows) + ), + total.pnl, + total.volume, + ) + return total + + +# -- The state file: the baseline, and each server's last figures -- + + +def read_state(path: Path) -> dict[str, Any]: + return json.loads(path.read_text()) if path.exists() else {} + + +def write_state(path: Path, state: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n") + tmp.replace(path) + + +def _dump(totals: dict[str, Totals]) -> dict[str, dict[str, float]]: + return {a: {"pnl": t.pnl, "volume": t.volume} for a, t in totals.items()} + + +def _parse(rows: Any) -> dict[str, Totals]: + return { + agent_id: Totals(float(row["pnl"]), float(row["volume"])) + for agent_id, row in (rows or {}).items() + } + + +def load_baseline(path: Path) -> dict[str, Totals] | None: + """What each agent had at the start of the race, or ``None`` if never taken.""" + state = read_state(path) + return _parse(state["baseline"]) if "baseline" in state else None + + +def save_baseline(path: Path, totals: dict[str, Totals], now: float) -> None: + stamp = datetime.fromtimestamp(now, tz=timezone.utc).isoformat(timespec="seconds") + write_state( + path, {**read_state(path), "started_at": stamp, "baseline": _dump(totals)} + ) + + +def load_last_good(path: Path) -> dict[str, dict[str, Totals]]: + """Per server, the last clean figure each agent had there.""" + servers = read_state(path).get("last_good") or {} + return {server: _parse(rows) for server, rows in servers.items()} + + +def save_last_good(path: Path, last_good: dict[str, dict[str, Totals]]) -> None: + dumped = {server: _dump(rows) for server, rows in last_good.items()} + write_state(path, {**read_state(path), "last_good": dumped}) + + +# Retired generations no source can read any more, seeded by hand. Kept here +# rather than only in the state file, which --dry-run never writes and a reset +# would erase, taking the recorded loss with it. Each figure was measured and the +# comment says from where: do not add one without that evidence. +SEEDED_RETIRED: dict[str, dict[str, dict[str, float]]] = { + "hummingbot-api-1": { + # quad-e4 was killed by its own drawdown cap at 03:21:42 on 2026-10-08 + # with the container still up and reporting zeros. api-1 holds no archive + # for it and no route reads a stopped bot's sqlite, so this is its last + # live snapshot: the board's own drop, 112,495.04 -> 4,330.18, whose + # difference is exactly the volume below, and its executors' + # net_pnl_quote, all realized, no positions left. + "quad-e4-20261007-051441": {"volume": 108164.86, "realized": -39.26}, + }, +} + + +def load_bot_ledger(path: Path) -> dict[str, dict[str, dict[str, float]]]: + """Per server, the last snapshot of every bot this monitor saw live. + + Kept after a bot stops. For a stopped-but-unarchived bot this is the only + surviving record of what it traded, because nothing can read its sqlite. + + Nothing prunes this ledger: every bot name seen since the race started + accumulates, a few dozen rows, and one that is archived is skipped rather than + removed -- bounded by the number of generations, not by time. + """ + servers = read_state(path).get("bot_ledger") or {} + seeded = { + server: {name: dict(snap) for name, snap in rows.items()} + for server, rows in SEEDED_RETIRED.items() + } + ledger: dict[str, dict[str, dict[str, float]]] = { + server: dict(rows) for server, rows in seeded.items() + } + for server, rows in servers.items(): + target = ledger.setdefault(server, {}) + for name, snap in (rows or {}).items(): + if name in seeded.get(server, {}): + # Seeded names are AUTHORITATIVE: they cover bots no source can + # read any more, so no state entry may override one. A live + # reading recorded while the bot ran still can -- that happens in + # the merge, not here. + continue + target[name] = dict(snap) + return ledger + + +def save_bot_ledger(path: Path, ledger: dict[str, dict[str, dict[str, float]]]) -> None: + write_state(path, {**read_state(path), "bot_ledger": ledger}) + + +def distinct_servers(servers: dict[str, dict], wanted: list[str]) -> list[str]: + """The servers to read, each API once. + + ``wanted`` restricts the set and must name configured servers. Two names for + one host and port are one API, and counting it twice would double every + figure on it, so only the first name is kept. + """ + unknown = [name for name in wanted if name not in servers] + if unknown: + raise ValueError(f"Unknown server(s): {', '.join(unknown)}") + names: list[str] = [] + seen: dict[tuple[str, str], str] = {} + for name in wanted or list(servers): + config = servers[name] or {} + key = (str(config.get("host", name)), str(config.get("port", ""))) + if key in seen: + log.warning("%s is the same API as %s; reading it once", name, seen[key]) + continue + seen[key] = name + names.append(name) + return names + + +def combine( + agent_ids: Iterable[str], + answers: dict[str, tuple[dict[str, Totals], set[str]] | None], + last_good: dict[str, dict[str, Totals]], +) -> tuple[dict[str, Totals], dict[str, set[str]]]: + """Sum the field across servers, and say which figures were not fresh. + + ``answers`` maps a server to its ``(totals, degraded)``, or ``None`` when it + did not answer at all. A clean figure is used and remembered in + ``last_good``; anything else falls back to the last clean one from that + server. Returns the sums and ``{server: agents}`` for the pairs that have + never had a clean figure, which count as nothing. + """ + agent_ids = list(agent_ids) + totals = {agent_id: Totals() for agent_id in agent_ids} + unknown: dict[str, set[str]] = {} + for server, answer in answers.items(): + fresh, degraded = answer if answer is not None else ({}, set(agent_ids)) + known = last_good.setdefault(server, {}) + for agent_id in agent_ids: + if agent_id in fresh and agent_id not in degraded: + known[agent_id] = fresh[agent_id] + if agent_id in known: + totals[agent_id] = totals[agent_id].plus(known[agent_id]) + else: + unknown.setdefault(server, set()).add(agent_id) + return totals, unknown + + +def since_start( + totals: dict[str, Totals], baseline: dict[str, Totals] | None +) -> dict[str, Totals]: + """Lifetime figures restated from the race start. + + An agent missing from the baseline had nothing when it was taken. + """ + if baseline is None: + return dict(totals) + return { + agent_id: t.minus(baseline.get(agent_id, Totals())) + for agent_id, t in totals.items() + } + + +# ── The loop ── + + +class Monitor: + def __init__(self, args: argparse.Namespace, session: Any) -> None: + self.args = args + self.session = session + self.url = race_url(args.url, args.slug) + self.state_path = Path(args.state.format(slug=args.slug)) + self.agent_ids: list[str] = [] + # Per server, the last clean figure per agent: a failed fetch is reported + # as the last thing known rather than as an agent that gave its PnL back. + self.last_good = load_last_good(self.state_path) + # Per server, the last snapshot of each bot seen live. Retained when a bot + # stops so a retired generation keeps contributing (see inactive_bot_totals). + self.bot_ledger = load_bot_ledger(self.state_path) + self.posted = False + + async def refresh_agent_ids(self) -> None: + """Re-read the field; keep the last list when the site does not answer.""" + try: + ids = await fetch_agent_ids(self.session, self.url) + except Exception as exc: # noqa: BLE001 - the last list still stands + if not self.agent_ids: + raise + log.warning("Could not refresh the agent list (%s); keeping it", exc) + return + if not ids: + raise RuntimeError(f"GET {self.url} listed no agents") + if ids != self.agent_ids: + log.info("Race agents (%d): %s", len(ids), ", ".join(ids)) + self.agent_ids = ids + + async def _read_server( + self, name: str, matched: dict[str, list], extra_bots: dict[str, list[str]] + ) -> tuple[dict[str, Totals], set[str]] | None: + from config_manager import get_config_manager + + async def read() -> tuple[dict[str, Totals], set[str], dict]: + client = await get_config_manager().get_client(name) + return await collect_totals( + client, matched, extra_bots, self.bot_ledger.get(name, {}), name + ) + + try: + return await asyncio.wait_for(read(), SERVER_TIMEOUT) + except Exception as exc: # noqa: BLE001 - one server down is not the tick + log.warning("Server %s did not answer: %s", name, exc or type(exc).__name__) + return None + + def _merge_ledger(self, names: list[str], answers: list) -> None: + """Fold this tick's live per-bot snapshots into the persistent ledger. + + Entries for bots that have stopped are deliberately NOT removed here: a + stopped bot's last snapshot is what keeps its volume and realized PnL on + the board. Pruning happens only when a bot turns up archived, where the + archive supersedes it. + """ + for name, answer in zip(names, answers): + if not answer or len(answer) < 3: + continue + rows = answer[2] or {} + server_ledger = self.bot_ledger.setdefault(name, {}) + for bot_name, snap in rows.items(): + prior = server_ledger.get(bot_name) + if prior and not snap.get("volume") and not snap.get("realized"): + # Still registered, no longer reporting (its strategy died). + # Keep what it was last worth rather than overwriting the + # record with a zero that is not a reading. + continue + if prior: + # Merge keys, so a seeded note stands beside the reading. + merged = dict(prior) + merged.update(snap) + server_ledger[bot_name] = merged + else: + server_ledger[bot_name] = dict(snap) + + async def measure(self) -> tuple[dict[str, Totals], dict[str, set[str]]]: + """Lifetime totals for the field over every server, and what is unknown.""" + from condor.agents.fleet_map import build_fleet_map + from config_manager import get_config_manager + + agent_map = ( + load_agent_map(self.args.agent_map) + if self.args.agent_map + else AgentMap(runs={}, bots={}) + ) + matched = match_owners(self.agent_ids, build_fleet_map(), agent_map.runs) + for agent_id, owners in matched.items(): + found = {owner.run_key for owner in owners} + if agent_id in agent_map.runs: + missing = [key for key in agent_map.runs[agent_id] if key not in found] + if missing or not owners: + log.warning( + "%s: no Condor strategy with run key %s", + agent_id, + ", ".join(missing) or "(none listed)", + ) + continue + if not owners: + log.warning( + "%s: no Condor strategy with that slug; map it to a run key " + "in the agent map", + agent_id, + ) + + names = distinct_servers(get_config_manager().list_servers(), self.args.servers) + # List every server's archives BEFORE the per-server reads, which are + # gathered concurrently. The retired-generation filter needs the union + # across servers, and building it from inside those reads would leave it + # empty or partial on the first tick after a restart -- a silent + # nondeterminism in a correctness filter. Cheap: the listing is cached, so + # the reads below reuse it. + for name in names: + try: + client = await get_config_manager().get_client(name) + await listed_archive_paths(client, name) + except Exception as exc: # noqa: BLE001 - a failed listing is not fatal + log.warning( + "Archive listing for %s failed during prime (%s)", name, exc + ) + answers = await asyncio.gather( + *(self._read_server(name, matched, agent_map.bots) for name in names) + ) + # Each answer carries this tick's per-bot snapshot as a third element; + # fold those into the ledger, then hand combine() the pair it expects. + self._merge_ledger(names, answers) + totals, unknown = combine( + self.agent_ids, + { + name: (answer[0], answer[1]) if answer else None + for name, answer in zip(names, answers) + }, + self.last_good, + ) + for server, agents in sorted(unknown.items()): + log.warning( + "No figure ever from %s for %d agent(s); counted as nothing", + server, + len(agents), + ) + try: + direct = await fetch_direct_venue_totals() + except Exception as exc: # noqa: BLE001 - a direct-venue miss is not the tick + log.warning("Direct-venue read failed (%s); using Condor figures only", exc) + direct = {} + for agent_id, extra in direct.items(): + if agent_id in totals: + totals[agent_id] = totals[agent_id].plus(extra) + unknown.pop(agent_id, None) + for server_agents in unknown.values(): + server_agents.discard(agent_id) + return totals, unknown + + async def tick(self) -> None: + await self.refresh_agent_ids() + totals, unknown = await self.measure() + now = time.time() + + if self.args.start_race: + if unknown: + raise RuntimeError( + "Cannot take the baseline: no figure from " + + ", ".join(sorted(unknown)) + + ". Fix the server or leave it out with --servers." + ) + save_baseline(self.state_path, totals, now) + log.info("Race started: baseline written to %s", self.state_path) + self.args.start_race = False + if not self.args.dry_run: + save_last_good(self.state_path, self.last_good) + save_bot_ledger(self.state_path, self.bot_ledger) + + baseline = load_baseline(self.state_path) + payload = build_payload(self.agent_ids, since_start(totals, baseline), now) + for row in payload["agents"]: + log.info( + " %-28s pnl %12.2f volume %14.2f", + row["agent_id"], + row["pnl_quote"], + row["volume_quote"], + ) + if self.args.dry_run: + print(json.dumps(payload, indent=2)) + return + accepted, unknown = await post_race_data(self.session, self.url, payload) + self.posted = True + log.info( + "Posted %d agents, %d accepted (%s)", + len(payload["agents"]), + accepted, + "since race start" if baseline is not None else "lifetime, no baseline", + ) + if unknown: + log.warning( + "The site does not know %d agent ids: %s", + len(unknown), + ", ".join(unknown), + ) + + async def run(self) -> None: + while True: + started = time.monotonic() + try: + await self.tick() + except Exception: + # Before the first post there is nothing to protect: fail loudly + # on a bad URL or token. After it, a bad minute is just skipped. + if self.args.once or self.args.start_race or not self.posted: + raise + log.exception("Tick failed; retrying next interval") + if self.args.once: + return + elapsed = time.monotonic() - started + await asyncio.sleep(max(1.0, self.args.interval - elapsed)) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + env = os.environ + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--url", default=env.get("HACKATHON_API_URL", "")) + parser.add_argument("--token", default=env.get("HACKATHON_API_TOKEN", "")) + parser.add_argument("--slug", default=env.get("HACKATHON_SLUG", DEFAULT_SLUG)) + parser.add_argument( + "--servers", + default=env.get("HACKATHON_SERVERS", ""), + help="comma-separated server names; default is every configured server", + ) + parser.add_argument("--interval", type=float, default=DEFAULT_INTERVAL) + parser.add_argument("--state", default=DEFAULT_STATE) + parser.add_argument( + "--agent-map", + type=Path, + default=env.get("HACKATHON_AGENT_MAP") or None, + help="YAML of race agent id -> Condor run keys, for slugs that differ", + ) + parser.add_argument("--once", action="store_true", help="one tick, then exit") + parser.add_argument( + "--dry-run", action="store_true", help="print the payload instead of posting" + ) + parser.add_argument( + "--start-race", + action="store_true", + help="record the current figures as the baseline, then keep running", + ) + args = parser.parse_args(argv) + if not args.url: + parser.error("set HACKATHON_API_URL or pass --url") + if not args.token: + parser.error("set HACKATHON_API_TOKEN or pass --token") + state = Path(args.state.format(slug=args.slug)) + args.servers = [name.strip() for name in args.servers.split(",") if name.strip()] + if args.agent_map: + args.agent_map = Path(args.agent_map) + if not args.agent_map.is_file(): + parser.error(f"agent map {args.agent_map} does not exist") + load_agent_map(args.agent_map) + if args.start_race and load_baseline(state) is not None: + parser.error(f"{state} already holds a baseline; delete it to restart the race") + if args.start_race and args.dry_run: + parser.error("--start-race writes the baseline; it cannot be a dry run") + return args + + +async def main(argv: list[str] | None = None) -> None: + import aiohttp + from dotenv import load_dotenv + + load_dotenv() + args = parse_args(argv) + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + # A strategy that has not deployed yet is warned about on every fetch; the + # monitor's own lines already say who has no figure. + logging.getLogger("condor.agents.performance").setLevel(logging.ERROR) + + from config_manager import get_config_manager + + try: + async with aiohttp.ClientSession( + headers={"Authorization": f"Bearer {args.token}"}, + timeout=aiohttp.ClientTimeout(total=HTTP_TIMEOUT), + ) as session: + await Monitor(args, session).run() + finally: + await get_config_manager().close_all_clients() + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + pass diff --git a/tests/test_hackathon_monitor.py b/tests/test_hackathon_monitor.py new file mode 100644 index 000000000..5784dc609 --- /dev/null +++ b/tests/test_hackathon_monitor.py @@ -0,0 +1,390 @@ +"""The hackathon race monitor posts what the bots page shows, counted from the start.""" + +from __future__ import annotations + +import asyncio +import json + +import pytest + +from condor.agents.fleet_map import FleetOwner +from condor.agents.performance import AgentPerformance +from scripts import hackathon_monitor as monitor +from scripts.hackathon_monitor import Totals + + +def _owner(agent: str, strategy: str, sessions: int = 1, **extra) -> FleetOwner: + run_key = f"{agent}.{strategy}" + return FleetOwner( + run_key=run_key, + agent_slug=agent, + agent_name=agent, + strategy_slug=strategy, + strategy_name=strategy, + namespace=f"{agent}-{strategy}", + agent_ids=[f"{run_key}_{n}" for n in range(1, sessions + 1)], + **extra, + ) + + +def test_the_race_url_is_built_from_the_base_and_the_slug(): + assert ( + monitor.race_url("https://race.example/", "agent-builders-cup-1") + == "https://race.example/api/hackathons/agent-builders-cup-1/race-data" + ) + + +# What GET /api/hackathons/[slug]/race-data answers on the site. +RACE_DATA_RESPONSE = { + "data": { + "hackathon": "agent-builders-cup-1", + "race_start": "2026-10-05T00:00:00.000Z", + "race_end": "2026-10-12T00:00:00.000Z", + "agents": [ + { + "agent_id": "grid_a", + "strategy_title": "Grid A", + "team": {"slug": "orca", "name": "Orca"}, + "driver": "Alice", + }, + { + "agent_id": "mm_b", + "strategy_title": "MM B", + "team": {"slug": "gate", "name": "Gate"}, + "driver": "Bob", + }, + ], + } +} + + +def test_agent_ids_are_read_from_the_race_data_response(): + assert monitor.parse_agent_ids(RACE_DATA_RESPONSE) == ["grid_a", "mm_b"] + + +@pytest.mark.parametrize( + "payload", [{"agents": [{"agent_id": "grid_a"}]}, {"error": "nope"}, None] +) +def test_any_other_agent_list_shape_is_an_error(payload): + with pytest.raises(RuntimeError, match="Unexpected race-data response"): + monitor.parse_agent_ids(payload) + + +def test_the_post_result_gives_accepted_and_unknown_ids(): + assert monitor.parse_post_result( + {"data": {"accepted": 1, "unknown": ["typo_c"]}} + ) == (1, ["typo_c"]) + with pytest.raises(RuntimeError, match="Unexpected race-data POST response"): + monitor.parse_post_result({"accepted": 1}) + + +def test_an_id_matches_its_strategy_slug_under_every_agent(): + owners = [ + _owner("alice", "grid_a"), + _owner("bob", "grid_a"), + _owner("bob", "mm_b"), + # A pseudo-run: no namespace, never a race agent. + FleetOwner("condor.chat", "condor", "Condor", "chat", "Chat", namespace=""), + ] + + matched = monitor.match_owners(["grid_a", "bob.mm_b", "chat", "ghost"], owners) + + assert [o.run_key for o in matched["grid_a"]] == ["alice.grid_a", "bob.grid_a"] + assert [o.run_key for o in matched["bob.mm_b"]] == ["bob.mm_b"] + assert matched["chat"] == [] + assert matched["ghost"] == [] + + +def test_a_mapped_id_names_only_its_run_keys(): + owners = [ + _owner("alice", "grid_a"), + _owner("bob", "grid_a"), + _owner("bob", "mm_b"), + ] + + matched = monitor.match_owners( + ["race-grid", "grid_a", "unfilled"], + owners, + {"race-grid": ["bob.grid_a", "bob.mm_b"], "unfilled": []}, + ) + + assert [o.run_key for o in matched["race-grid"]] == ["bob.grid_a", "bob.mm_b"] + assert [o.run_key for o in matched["grid_a"]] == ["alice.grid_a", "bob.grid_a"] + assert matched["unfilled"] == [] + + +def test_the_agent_map_reads_a_key_or_a_list_and_rejects_anything_else(tmp_path): + path = tmp_path / "agents.yml" + path.write_text("churn: stable.churn_supervisor\nfleet: [a.mm, b.mm]\n") + assert monitor.load_agent_map(path) == { + "churn": ["stable.churn_supervisor"], + "fleet": ["a.mm", "b.mm"], + } + + path.write_text("churn: {key: stable.churn_supervisor}\n") + with pytest.raises(RuntimeError, match="churn must map"): + monitor.load_agent_map(path) + + +def test_totals_sum_an_owners_sessions_and_attach_its_bots_once(monkeypatch): + calls = {} + + async def fake_batch(client, agent_ids, bot_names, failed_ids=None): + calls["agent_ids"] = agent_ids + calls["bot_names"] = bot_names + failed_ids.add("bob.mm_b_1") + return { + "alice.grid_a_1": AgentPerformance( + "alice.grid_a_1", total_pnl=10, volume=100 + ), + "alice.grid_a_2": AgentPerformance( + "alice.grid_a_2", total_pnl=-4, volume=50 + ), + "bob.mm_b_1": AgentPerformance("bob.mm_b_1"), + "carol.new_c": AgentPerformance("carol.new_c", total_pnl=1, volume=2), + } + + monkeypatch.setattr( + "condor.agents.performance.fetch_agent_performance_batch", fake_batch + ) + matched = { + "grid_a": [_owner("alice", "grid_a", sessions=2, declared_bots=["legacy-bot"])], + "mm_b": [_owner("bob", "mm_b")], + # Deployed a bot but never opened a session: its run key carries the bots. + "new_c": [_owner("carol", "new_c", sessions=0)], + "ghost": [], + } + + totals, degraded = asyncio.run(monitor.collect_totals(object(), matched)) + + assert totals["grid_a"] == Totals(pnl=6, volume=150) + assert totals["new_c"] == Totals(pnl=1, volume=2) + assert totals["ghost"] == Totals(), "no strategy has traded nothing" + assert degraded == {"mm_b"} + assert calls["bot_names"] == { + "alice.grid_a_1": ["alice-grid_a", "legacy-bot"], + "bob.mm_b_1": ["bob-mm_b"], + "carol.new_c": ["carol-new_c"], + } + assert "alice.grid_a_2" in calls["agent_ids"] + + +def test_the_baseline_round_trips_and_is_subtracted(tmp_path): + path = tmp_path / "nested" / "race.json" + assert monitor.load_baseline(path) is None + + monitor.save_baseline(path, {"grid_a": Totals(5, 1000)}, now=1_800_000_000) + baseline = monitor.load_baseline(path) + + assert baseline == {"grid_a": Totals(5, 1000)} + now = {"grid_a": Totals(12, 1500), "late": Totals(3, 40)} + assert monitor.since_start(now, baseline) == { + "grid_a": Totals(7, 500), + "late": Totals(3, 40), + } + assert monitor.since_start(now, None) == now, "no baseline posts lifetime" + + +def test_the_payload_keeps_the_race_order_and_drops_agents_with_no_figure(): + payload = monitor.build_payload( + ["mm_b", "grid_a", "down"], + {"grid_a": Totals(1.23456789, 10), "mm_b": Totals(-2, 0)}, + now=1_800_000_000, + ) + + assert payload == { + "timestamp": "2027-01-15T08:00:00Z", + "agents": [ + {"agent_id": "mm_b", "pnl_quote": -2, "volume_quote": 0}, + {"agent_id": "grid_a", "pnl_quote": 1.234568, "volume_quote": 10}, + ], + } + json.dumps(payload) + + +def test_an_agent_is_summed_across_servers(): + last_good: dict = {} + answers = { + "local": ({"grid_a": Totals(10, 100), "mm_b": Totals(1, 5)}, set()), + "remote": ({"grid_a": Totals(-3, 40), "mm_b": Totals(0, 0)}, set()), + } + + totals, unknown = monitor.combine(["grid_a", "mm_b"], answers, last_good) + + assert totals == {"grid_a": Totals(7, 140), "mm_b": Totals(1, 5)} + assert unknown == {} + assert last_good["remote"]["grid_a"] == Totals(-3, 40) + + +def test_a_server_that_stops_answering_keeps_its_last_figure(): + """An outage must not read as an agent that gave its PnL back.""" + last_good: dict = {} + ids = ["grid_a", "mm_b"] + up = {"grid_a": Totals(10, 100), "mm_b": Totals(1, 5)} + remote = {"grid_a": Totals(5, 50), "mm_b": Totals(2, 20)} + monitor.combine(ids, {"local": (up, set()), "remote": (remote, set())}, last_good) + + # The remote is down, and the local one degraded for a single agent. + later = {"grid_a": Totals(0, 0), "mm_b": Totals(3, 9)} + totals, unknown = monitor.combine( + ids, {"local": (later, {"grid_a"}), "remote": None}, last_good + ) + + assert totals == {"grid_a": Totals(15, 150), "mm_b": Totals(5, 29)} + assert unknown == {} + + +def test_a_server_never_heard_from_counts_as_nothing_and_is_named(): + totals, unknown = monitor.combine( + ["grid_a"], {"local": ({"grid_a": Totals(4, 8)}, set()), "dead": None}, {} + ) + + assert totals == {"grid_a": Totals(4, 8)} + assert unknown == {"dead": {"grid_a"}} + + +def test_last_figures_survive_a_restart_beside_the_baseline(tmp_path): + path = tmp_path / "race.json" + monitor.save_last_good(path, {"remote": {"grid_a": Totals(5, 50)}}) + monitor.save_baseline(path, {"grid_a": Totals(1, 2)}, now=1_800_000_000) + monitor.save_last_good(path, {"remote": {"grid_a": Totals(6, 60)}}) + + assert monitor.load_last_good(path) == {"remote": {"grid_a": Totals(6, 60)}} + assert monitor.load_baseline(path) == {"grid_a": Totals(1, 2)} + + +def test_one_api_under_two_names_is_read_once(): + servers = { + "local": {"host": "localhost", "port": 8000}, + "alias": {"host": "localhost", "port": 8000}, + "remote": {"host": "box.example", "port": 8000}, + } + + assert monitor.distinct_servers(servers, []) == ["local", "remote"] + assert monitor.distinct_servers(servers, ["remote"]) == ["remote"] + with pytest.raises(ValueError, match="nope"): + monitor.distinct_servers(servers, ["nope"]) + + +def test_starting_a_race_twice_is_refused(tmp_path, capsys): + state = tmp_path / "race.json" + base = ["--url", "https://x", "--token", "t", "--state", str(state)] + # Remembered figures alone are not a baseline. + monitor.save_last_good(state, {"local": {"grid_a": Totals(1, 1)}}) + args = monitor.parse_args([*base, "--start-race", "--servers", "local, remote"]) + assert args.servers == ["local", "remote"] + + monitor.save_baseline(state, {"grid_a": Totals(1, 1)}, now=1_800_000_000) + with pytest.raises(SystemExit): + monitor.parse_args([*base, "--start-race"]) + + assert "already holds a baseline" in capsys.readouterr().err + + +# ── Stopped trading still counts (the race is cumulative) ── + + +class _StopClient: + """One agent's trading in every stopped and running shape, as the API serves it. + + - ``alice-grid_a-20261003-100000`` is a live bot with two controllers; the + agent paused ``c_killed`` (``manual_kill_switch``), which keeps it listed in + the bot's reports with its final figures. + - ``alice-grid_a-20261002-090000`` is a bot the agent stopped and archived; + its last snapshot holds a frozen unrealized mark nobody owns any more. + - ``alice.grid_a_1`` tagged two standalone executors, one closed, one open. + """ + + base_url = "" # no cross-test caching + + LIVE = "alice-grid_a-20261003-100000" + STOPPED = "alice-grid_a-20261002-090000" + + def __init__(self): + def snap(bot, cid, realized, unrealized, volume): + return { + "bot_name": bot, + "controller_id": cid, + "trading_pair": "BTC-USDT", + "timestamp": "2026-10-03T10:00:00", + "performance": { + "realized_pnl_quote": realized, + "unrealized_pnl_quote": unrealized, + "volume_traded": volume, + }, + } + + snapshots = [ + snap(self.LIVE, "c_live", 5.0, 2.0, 100.0), + snap(self.LIVE, "c_killed", 7.0, 0.0, 300.0), + snap(self.STOPPED, "c_old", -3.0, 4.0, 200.0), + ] + live = self.LIVE + + def executor(status, pnl, volume): + return { + "id": f"ex-{status}", + "status": status, + "net_pnl_quote": pnl, + "filled_amount_quote": volume, + "cum_fees_quote": 0.0, + "config": {"type": "order_executor"}, + } + + rows = { + "alice.grid_a_1": [ + executor("TERMINATED", 1.5, 50.0), + executor("RUNNING", -0.5, 20.0), + ] + } + + class _Orchestration: + async def get_latest_controller_performance(self, bot_name=None): + return {"data": snapshots} + + async def get_active_bots_status(self): + return {"status": "success", "data": {live: {"status": "running"}}} + + class _Executors: + async def search_executors(self, controller_ids, limit, cursor=None): + return {"executors": rows.get(controller_ids[0], [])} + + stopped = self.STOPPED + + class _Archived: + async def list_databases(self): + return [f"bots/archived/{stopped}/data/{stopped}.sqlite"] + + self.bot_orchestration = _Orchestration() + self.executors = _Executors() + self.archived_bots = _Archived() + + +@pytest.fixture +def _clean_perf_caches(): + from condor.fetchers import bot_performance as bp + + bp.clear_snapshot_cache() + bp.clear_history_cache() + bp.clear_archived_cache() + bp.clear_live_names_cache() + yield + bp.clear_snapshot_cache() + bp.clear_history_cache() + bp.clear_archived_cache() + bp.clear_live_names_cache() + + +def test_stopped_controllers_bots_and_executors_stay_in_the_agents_total( + _clean_perf_caches, +): + """Through the real aggregator: stopping anything never takes it off the board.""" + matched = {"grid_a": [_owner("alice", "grid_a")]} + + totals, degraded = asyncio.run(monitor.collect_totals(_StopClient(), matched)) + + # live bot (both controllers, the paused one included) + the stopped bot's + # realized (its frozen unrealized mark is not counted) + both executors. + assert totals["grid_a"].pnl == pytest.approx((5 + 2) + 7 + (-3) + 1.5 + (-0.5)) + assert totals["grid_a"].volume == pytest.approx(100 + 300 + 200 + 50 + 20) + assert degraded == set()