#!/usr/bin/env python3
"""tfrun: run a trading bot you write as one function against tradefloor.

    curl -O https://app.tradefloor.dev/agents/examples/tfrun.py
    export TF_KEY=tfk_...                               # from https://app.tradefloor.dev/keys

    python tfrun.py my_bot.py --days 60                 # one simulation
    python tfrun.py my_bot.py --suite tf-quick-2026.2   # a published suite, against the baselines
    python tfrun.py my_bot.py --days 60 --offline       # on your machine; not available yet
    python tfrun.py my_bot.py --check "peak_leverage<=1.5,max_drawdown<=20%"   # CI checks
    python tfrun.py my_bot.py --shared MARKET_ID --days 20  # take part in a shared market
    python tfrun.py my_bot.py --join K7QM4XTR --days 20     # take a seat with a market's join code
    python tfrun.py my_bot.py --line LINE_ID --days 120      # drive a line of the Simulator
    python tfrun.py --strategy st_x:3 --line LINE_ID --days 120   # a saved strategy's code
    python tfrun.py --strategy st_x:3 --simulation SIM_ID --days 46   # every line it trades

my_bot.py is one function. tfrun runs the loop, moves time, retries safely
and prints what happened with a link to watch it:

    from tfrun import buy, sell, target

    def on_step(market):
        if market.change_pct("AAA") < -2 and not market.position("AAA"):
            return [buy("AAA", 100)]
        return []

Optional: `setup(config)` runs once per session before the first step (a
suite runs it again for each market), and `on_day_end(market)` runs after
each close. Both on_step and on_day_end may return orders; orders returned
after the close fill at the next open.

When on_step runs: at every step while the market is open (09:30, then every
`--step` minutes; 30 by default), or once a day at the open with `--every
day`. A market order fills at the start of the next step, when time moves.

What `market` has (the compact observation, plus helpers):

    market.day, .time ("10:30"), .tick, .is_open, .last_day
    market.tickers, .prices, .price(t), .change_pct(t), .quote(t)
    market.cash, .net_worth, .pnl, .buying_power, .leverage
    market.positions ({ticker: shares}), .position(t), .weight(t), .open_orders
    market.events (today's events), .events_for(t), .vix, .economy
    market.fills (fills since the last call), .refused (orders refused last time)
    market.closes(t, days=20), .bars(t, days=20)   daily history (one call a day); with
                                                    --history-days N it reaches back
                                                    before day 0
    market.obs                                      the raw compact observation

Orders:

    buy(ticker, shares, limit=None, tif="day")      a market order, or a limit order at `limit`
    sell(ticker, shares, limit=None, tif="day")     selling what you do not hold opens a short
    target(ticker, weight)                          trade to weight x net worth (0 closes)
    target(ticker, shares=n)                        trade to hold n shares (negative is short)
    cancel(order_id)                                cancel an open order
    twap(ticker, shares, side="buy", steps=13)      a parent order the server slices evenly
    vwap(ticker, shares, side="buy", minutes=390)   the same along the engine's volume profile;
                                                    both take max_participation= and limit=

    market.algo_status(algo_id=None)                your TWAP/VWAP orders, or one with its
                                                    children and fills (one call)
    market.lots(ticker=None)                        your positions as tax lots, with realised
                                                    gains by term and harvestable losses

Scenarios: --scenario takes a packaged name (rate_shock), a crash replay
(replay_2008) or a saved scenario's id (cs_...); --scenario-file runs a
scenario document from a JSON file. save_scenario(document) and
scale_scenario("rate_shock", 2) save one and return it with its id.

buy, sell and target take note="why": up to 500 characters, kept with the
order and shown beside its fills (only you see it).

Checks for CI: `--check EXPR` (repeat it, or comma-separate) reads the
session's behaviour metrics at the end, such as peak_leverage<=1.5,
turnover<=20 or max_drawdown<=20% (a %% value is a fraction), prints each
with its value, and exits 4 if any fails. With --suite, every market is
checked. Checks need a full or read-only key: set TF_CHECK_KEY to one when
TF_KEY is a trader key.

Every write carries an Idempotency-Key, so a retry after a timeout never
trades twice. tfrun keeps to your plan's calls a minute and waits out 429
and 503 answers as the server says; a daily limit stops the run cleanly and
says when it resets. If your function raises, the run stops, prints the
traceback and leaves the session open; `--resume SESSION_ID` carries on
after a fix. A suite steps at its own step size unless `--step` says
otherwise.

A shared market: `--shared MARKET_ID` plays your bot in a market several
agents trade at once, through the seat its host gave TF_KEY (`--join CODE`
takes the seat an invitation's code offers first). A market made in the
Simulator needs only its join code: `--join CODE` takes a seat as code
(POST /v1/markets/join) and plays it. Time moves when every
participant is ready or the host's timeout runs out, so tfrun waits between
steps as the server says; market.fills has what your orders filled in the
meantime. At the end your bot leaves the market (--keep-open stays in it).

`--line LINE_ID` drives a line of the Simulator from where it is for --days
days, prints its Simulator link and leaves it open, so the line keeps going
with the rest of its simulation. The Simulator's driver strip shows the
command. Time on the line moves when your code moves it. If your function
raises, the error shows in the line's Activity and the run stops.
`--strategy ST[:V]` runs version V (the latest without it) of a strategy
saved on tradefloor, its code and entry function, on this machine, so no bot
file is needed. `--simulation SIM_ID` with --strategy drives every line of
that simulation traded by that strategy version, one after another, each
for --days days (a Scenario test). `--strategy ST[:V] --resume RUN_ID` plays
a benchmark run's markets with the strategy.

Settings: TF_KEY (your API key; TRADEFLOOR_KEY works too) and TF_URL (default
https://app.tradefloor.dev).
`--template NAME` opens the session from a named template on a fresh seed
(the only way a trader key opens one). `--json` prints a summary for CI.
Exit codes: 0 finished, 1 your code raised or the server refused, 2 bad
arguments, 3 stopped at a plan limit, 4 a --check failed, 130 interrupted.

Standard library only; Python 3.9 or later.
"""

from __future__ import annotations

import argparse
import contextlib
import csv
import http.client
import importlib.util
import json
import math
import os
import re
import socket
import subprocess
import sys
import tempfile
import threading
import time
import traceback
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional

__version__ = "0.3.0"
__all__ = ["buy", "sell", "target", "cancel", "twap", "vwap", "save_scenario",
           "scale_scenario", "Market", "Order", "main"]

DEFAULT_URL = "https://app.tradefloor.dev"
TICKS_PER_DAY = 390
BATCH = 50                     # orders per batch call, the API's cap
MAX_REFUSALS_SHOWN = 20
HERE = os.path.abspath(__file__)

EXIT_OK, EXIT_ERROR, EXIT_USAGE, EXIT_LIMIT, EXIT_INTERRUPTED = 0, 1, 2, 3, 130
EXIT_CHECK = 4
MAX_NOTE = 500
#: How often tfrun tells tradefloor it is still running (POST
#: /v1/trader/heartbeat); after 60 s without a call the Simulator says it
#: lost contact.
HEARTBEAT_EVERY = 15.0
#: The metrics a --check may name (GET /v1/sessions/{id}/behaviour), so a
#: typo is refused before the run and not after it.
CHECK_METRICS = ("turnover", "peak_leverage", "peak_leverage_at_closes", "peak_net_leverage",
                 "peak_weight", "concentration", "max_drawdown", "max_drawdown_at_closes",
                 "refused_orders", "order_to_fill", "exposure_after_losses", "orders", "fills")
_CHECK = re.compile(r"^\s*([a-z_]+)\s*(<=|>=|==|<|>|=)\s*-?[0-9]*\.?[0-9]+(?:e-?[0-9]+)?\s*%?\s*$",
                    re.I)


# -- orders -------------------------------------------------------------------------------


class Order:
    """What a bot returns. Made by buy(), sell(), target(), cancel(), twap()
    and vwap()."""

    __slots__ = ("kind", "ticker", "shares", "weight", "limit", "tif", "order_id", "note", "algo")

    def __init__(self, kind: str, ticker: Optional[str] = None, shares: Optional[float] = None,
                 weight: Optional[float] = None, limit: Optional[float] = None, tif: str = "day",
                 order_id: Optional[str] = None, note: Optional[str] = None,
                 algo: Optional[Dict[str, Any]] = None) -> None:
        self.kind, self.ticker, self.shares, self.weight = kind, ticker, shares, weight
        self.limit, self.tif, self.order_id, self.note = limit, tif, order_id, note
        self.algo = algo

    def __repr__(self) -> str:
        if self.kind == "cancel":
            return f"cancel({self.order_id!r})"
        if self.kind in ("twap", "vwap"):
            a = self.algo or {}
            return f"{self.kind}({self.ticker!r}, {self.shares!r}, side={a.get('side')!r})"
        if self.kind == "target":
            what = f"{self.weight!r}" if self.weight is not None else f"shares={self.shares!r}"
            return f"target({self.ticker!r}, {what})"
        extra = f", limit={self.limit!r}" if self.limit is not None else ""
        return f"{self.kind}({self.ticker!r}, {self.shares!r}{extra})"


def _number(value: Any, what: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
        raise ValueError(f"{what} must be a finite number, not {value!r}")
    return float(value)


def _note(note: Optional[str]) -> Optional[str]:
    if note is None:
        return None
    if not isinstance(note, str) or len(note) > MAX_NOTE:
        raise ValueError(f"note must be text of at most {MAX_NOTE} characters")
    return note.strip() or None


def _order(side: str, ticker: str, shares: float, limit: Optional[float], tif: str,
           note: Optional[str] = None) -> Order:
    if not isinstance(ticker, str) or not ticker:
        raise ValueError(f"the ticker must be text such as 'AAA', not {ticker!r}")
    if _number(shares, "shares") <= 0:
        raise ValueError(f"{side}() needs more than 0 shares, not {shares!r}")
    if limit is not None and _number(limit, "limit") <= 0:
        raise ValueError(f"a limit price must be above 0, not {limit!r}")
    if tif not in ("day", "gtc"):
        raise ValueError(f"tif must be 'day' or 'gtc', not {tif!r}")
    return Order(side, ticker, shares=shares, limit=limit, tif=tif, note=_note(note))


def buy(ticker: str, shares: float, limit: Optional[float] = None, tif: str = "day",
        note: Optional[str] = None) -> Order:
    """Buy `shares` of `ticker`: at the market, or at `limit` or better.
    `note`: why, kept with the order."""
    return _order("buy", ticker, shares, limit, tif, note)


def sell(ticker: str, shares: float, limit: Optional[float] = None, tif: str = "day",
         note: Optional[str] = None) -> Order:
    """Sell `shares` of `ticker`. Selling what you do not hold opens a short."""
    return _order("sell", ticker, shares, limit, tif, note)


def target(ticker: str, weight: Optional[float] = None, *, shares: Optional[float] = None,
           limit: Optional[float] = None, tif: str = "day", note: Optional[str] = None) -> Order:
    """Trade `ticker` to `weight` x net worth (0.1 is 10%; negative is short;
    0 closes the position), or to hold `shares` shares. Worked out from the
    last price and your position when the order is sent; open orders are not
    counted. Nothing is sent when the difference is under one share."""
    if (weight is None) == (shares is None):
        raise ValueError("target() takes a weight, or shares=, but not both")
    if not isinstance(ticker, str) or not ticker:
        raise ValueError(f"the ticker must be text such as 'AAA', not {ticker!r}")
    if weight is not None:
        _number(weight, "weight")
    else:
        _number(shares, "shares")
    if tif not in ("day", "gtc"):
        raise ValueError(f"tif must be 'day' or 'gtc', not {tif!r}")
    if limit is not None and _number(limit, "limit") <= 0:
        raise ValueError(f"a limit price must be above 0, not {limit!r}")
    return Order("target", ticker, shares=shares, weight=weight, limit=limit, tif=tif,
                 note=_note(note))


def cancel(order_id: str) -> Order:
    """Cancel an open order (its `order_id` from market.open_orders)."""
    if not isinstance(order_id, str) or not order_id:
        raise ValueError(f"cancel() needs an order id, not {order_id!r}")
    return Order("cancel", order_id=order_id)


def _algo(kind: str, ticker: str, shares: float, side: str, steps: Optional[int],
          minutes: Optional[int], max_participation: Optional[float],
          limit: Optional[float], note: Optional[str] = None) -> Order:
    if not isinstance(ticker, str) or not ticker:
        raise ValueError(f"the ticker must be text such as 'AAA', not {ticker!r}")
    if _number(shares, "shares") <= 0:
        raise ValueError(f"{kind}() needs more than 0 shares, not {shares!r}")
    if side not in ("buy", "sell"):
        raise ValueError(f"side must be 'buy' or 'sell', not {side!r}")
    if (steps is None) == (minutes is None):
        raise ValueError(f"{kind}() takes steps= or minutes= for its horizon, one of them")
    if max_participation is not None and not 0 < _number(max_participation,
                                                         "max_participation") <= 1:
        raise ValueError(f"max_participation must be above 0 and at most 1, not "
                         f"{max_participation!r}")
    if limit is not None and _number(limit, "limit") <= 0:
        raise ValueError(f"a limit price must be above 0, not {limit!r}")
    return Order(kind, ticker, shares=shares, limit=limit, note=_note(note), algo={
        "side": side, "horizon_steps": steps, "horizon_minutes": minutes,
        "max_participation": max_participation})


def twap(ticker: str, shares: float, side: str = "buy", *, steps: Optional[int] = None,
         minutes: Optional[int] = None, max_participation: Optional[float] = None,
         limit: Optional[float] = None, note: Optional[str] = None) -> Order:
    """A TWAP parent order: the server sends a child market order before each
    step of the horizon (`steps` steps, or `minutes` rounded up to steps),
    spreading `shares` evenly in time. `max_participation` (0 to 1] caps a
    child at that share of the step's expected volume; `limit` stops children
    trading past that price; `note` is kept with the parent and each child.
    market.algo_status() shows how it is going."""
    return _algo("twap", ticker, shares, side, steps, minutes, max_participation, limit, note)


def vwap(ticker: str, shares: float, side: str = "buy", *, steps: Optional[int] = None,
         minutes: Optional[int] = None, max_participation: Optional[float] = None,
         limit: Optional[float] = None, note: Optional[str] = None) -> Order:
    """A VWAP parent order: as twap(), with each step's share following the
    engine's intraday volume profile (more at the open and the close)."""
    return _algo("vwap", ticker, shares, side, steps, minutes, max_participation, limit, note)


# -- the market a bot sees ------------------------------------------------------------------


class Market:
    """The compact observation as attributes, with helpers. `obs` is the raw
    dict (docs: /agents.md, "observe")."""

    def __init__(self, obs: Dict[str, Any], *, last_day: Optional[int] = None,
                 fills: Optional[List[dict]] = None, refused: Optional[List[dict]] = None,
                 history: Optional[Callable[[int, int], Dict[str, List[dict]]]] = None,
                 read: Optional[Callable[[str, Optional[dict]], Any]] = None) -> None:
        self.obs = obs
        self._read = read
        clock, account = obs.get("clock", {}), obs.get("account", {})
        self.session_id: str = obs.get("session_id", "")
        self.day: int = clock.get("day", 0)
        self.tick: int = clock.get("tick", 0)
        self.time: str = clock.get("time", "")
        self.step: int = clock.get("step", 0)
        self.is_open: bool = bool(clock.get("market_open"))
        self.last_day = last_day
        self.cash: float = account.get("cash", 0.0)
        self.net_worth: float = account.get("net_worth", 0.0)
        self.pnl: float = account.get("pnl", 0.0)
        self.buying_power = account.get("buying_power")
        self.leverage = account.get("leverage")
        market = dict(obs.get("market") or {})
        self.vix = market.pop("vix", None)
        self.economy: Dict[str, Any] = market
        table = obs.get("quotes") or {}
        columns = table.get("columns") or []
        self.quotes: Dict[str, Dict[str, Any]] = {row[0]: dict(zip(columns, row))
                                                  for row in table.get("rows") or []}
        self.tickers: List[str] = list(self.quotes)
        self.prices: Dict[str, float] = {t: q.get("last") for t, q in self.quotes.items()}
        self.positions: Dict[str, float] = {p["ticker"]: p["quantity"]
                                            for p in obs.get("positions") or []}
        self.open_orders: List[dict] = list(obs.get("open_orders") or [])
        self.events: List[dict] = list(obs.get("events") or [])
        self.fills: List[dict] = list(fills or [])
        self.refused: List[dict] = list(refused or [])
        self._history = history

    def __repr__(self) -> str:
        return (f"<Market day {self.day} {self.time} {'open' if self.is_open else 'closed'}, "
                f"{len(self.tickers)} tickers, net worth {self.net_worth:,.2f}>")

    def _known(self, ticker: str) -> Dict[str, Any]:
        try:
            return self.quotes[ticker]
        except KeyError:
            raise KeyError(f"{ticker!r} is not in this market; market.tickers has "
                           f"{', '.join(self.tickers[:10])}"
                           f"{', ...' if len(self.tickers) > 10 else ''}") from None

    def price(self, ticker: str) -> float:
        """The last traded price."""
        return self._known(ticker)["last"]

    def change_pct(self, ticker: str) -> float:
        """The change since the previous close, in percent (2.5 is 2.5%)."""
        return self._known(ticker).get("chg_pct") or 0.0

    def quote(self, ticker: str) -> Dict[str, Any]:
        """last, chg_pct, bid, ask, high, low and volume."""
        return dict(self._known(ticker))

    def position(self, ticker: str) -> float:
        """Shares held (negative when short, 0 when none)."""
        return self.positions.get(ticker, 0)

    def weight(self, ticker: str) -> float:
        """The position's value over net worth."""
        if not self.net_worth:
            return 0.0
        return self.position(ticker) * self.price(ticker) / self.net_worth

    def events_for(self, ticker: str) -> List[dict]:
        """Today's company events for `ticker`: each has `direction`
        ("positive" or "negative") and `text`, never a size."""
        return [e for e in self.events if e.get("ticker") == ticker]

    def bars(self, ticker: str, days: int = 20) -> List[dict]:
        """Daily bars (day, open, high, low, close, volume) of the last `days`
        finished trading days, oldest first (today's counts once it has
        closed). Fetched once a day."""
        self._known(ticker)
        if self._history is None or days <= 0:
            return []
        got = self._history(self.day, days).get(ticker) or []
        done = [b for b in got
                if b.get("day", 0) < self.day or (not self.is_open and b.get("day") == self.day)]
        return done[-days:]

    def closes(self, ticker: str, days: int = 20) -> List[float]:
        """Closing prices of the last `days` finished trading days, oldest first."""
        return [b["close"] for b in self.bars(ticker, days)]

    def _get(self, path: str, query: Optional[dict] = None) -> Any:
        if self._read is None:
            raise RuntimeError("this market has no server to ask")
        return self._read(path, query)

    def algo_status(self, algo_id: Optional[str] = None) -> Any:
        """Your TWAP and VWAP orders in this session, oldest first: status
        (working, filled, expired or cancelled), filled_quantity,
        avg_fill_price and slippage_bps against arrival_price (positive is a
        cost). With `algo_id`: that one as {algo, children, fills}. One call."""
        if algo_id is None:
            return self._get("/algo-orders")
        return self._get(f"/algo-orders/{urllib.parse.quote(algo_id)}")

    def lots(self, ticker: Optional[str] = None) -> Dict[str, Any]:
        """Your positions as tax lots: `lots` (open, with cost, day opened,
        days held and unrealised P&L), `realised` (each closed part with its
        gain and term: long when held more than 252 trading days),
        realised_short_term, realised_long_term and harvestable_losses. Lots
        close first in, first out. One call."""
        return self._get("/lots", {"ticker": ticker} if ticker else None)


# -- talking to the server ------------------------------------------------------------------


class ApiError(Exception):
    """A refusal from the server: its error body (code, message, hint)."""

    def __init__(self, status: int, body: Any, headers: Any = None) -> None:
        body = body if isinstance(body, dict) else {}
        self.status = status
        self.code: str = str(body.get("code") or ("http_%d" % status))
        self.message: str = str(body.get("message") or "the server refused the request")
        self.hint: str = str(body.get("hint") or "")
        retry = body.get("retry_after")
        if retry is None and headers is not None and headers.get("Retry-After"):
            try:
                retry = float(headers.get("Retry-After"))
            except ValueError:
                retry = None
        self.retry_after: Optional[float] = None if retry is None else max(0.0, float(retry))
        self.body = body
        super().__init__(f"{self.code}: {self.message}")


def _short(text: str, n: int = 110) -> str:
    text = " ".join(str(text).split())
    return text if len(text) <= n else text[:n - 3] + "..."


class Api:
    """JSON over HTTP with the retry rules in the module docstring."""

    TRANSPORT_TRIES = 6
    SERVER_TRIES = 8
    #: How long one call keeps waiting out 429s (and 409s that say when to
    #: come back) before it gives up: a busy minute is not a reason to stop.
    WAIT_SECONDS = 900.0

    def __init__(self, base: str, key: Optional[str], *, say: Callable[[str], None],
                 sleep: Callable[[float], None] = time.sleep, timeout: float = 90.0) -> None:
        self.base = base.rstrip("/")
        self.key = key
        self.say = say
        self.sleep = sleep
        self.timeout = timeout
        self.pause = 0.0               # before the next call, when the call bucket is empty
        self.told: set = set()         # what has been said once already
        self.driver: Optional[str] = None   # X-TF-Driver: what runs, for the Simulator's strip

    def get(self, path: str, query: Optional[dict] = None) -> Any:
        return self.call("GET", path, query=query)

    def post(self, path: str, body: Any = None, *, key: Optional[str] = None,
             query: Optional[dict] = None) -> Any:
        return self.call("POST", path, body, key=key, query=query)

    def delete(self, path: str, *, key: str) -> Any:
        return self.call("DELETE", path, key=key)

    def optional(self, path: str, query: Optional[dict] = None) -> Any:
        """A GET that is None when the server does not have it (404) or the
        key's scope does not cover it (a trader key and the report card)."""
        try:
            return self.get(path, query)
        except ApiError as e:
            if e.status == 404 or e.code == "forbidden":
                return None
            raise

    def call(self, method: str, path: str, body: Any = None, *, key: Optional[str] = None,
             query: Optional[dict] = None) -> Any:
        url = self.base + path
        if query:
            url += "?" + urllib.parse.urlencode({k: v for k, v in query.items() if v is not None})
        headers = {"Accept": "application/json", "User-Agent": f"tfrun/{__version__}"}
        if self.key:
            headers["Authorization"] = f"Bearer {self.key}"
        if self.driver:
            headers["X-TF-Driver"] = self.driver
        data = None
        if body is not None:
            data = json.dumps(body).encode("utf-8")
            headers["Content-Type"] = "application/json"
        elif method == "POST":
            data = b""
        if key is not None:
            headers["Idempotency-Key"] = key
        transport = waits = server = 0
        waited = 0.0
        while True:
            if self.pause:
                self.sleep(self.pause)
                self.pause = 0.0
            request = urllib.request.Request(url, data=data, headers=headers, method=method)
            try:
                with urllib.request.urlopen(request, timeout=self.timeout) as r:
                    status, raw, got = r.status, r.read(), r.headers
            except urllib.error.HTTPError as e:
                status, raw, got = e.code, e.read(), e.headers
            except (urllib.error.URLError, http.client.HTTPException, OSError) as e:
                transport += 1
                reason = getattr(e, "reason", None) or e
                if transport >= self.TRANSPORT_TRIES:
                    raise ApiError(0, {"code": "unreachable", "message": (
                        f"no answer from {self.base} after {transport} tries ({reason})"),
                        "hint": "Check TF_URL and your connection, then run again."}) from None
                wait = min(30.0, 2.0 ** (transport - 1))
                what = "sending it again" if key is None else "sending it again with the same key"
                self.say(f"no answer ({_short(str(reason), 60)}); {what} in {wait:.0f}s")
                self.sleep(wait)
                continue
            self._pace(got)
            try:
                answer = json.loads(raw.decode("utf-8")) if raw else None
            except ValueError:
                answer = {"message": _short(raw.decode("utf-8", "replace") or f"HTTP {status}")}
            if 200 <= status < 300:
                return answer
            err = ApiError(status, answer, got)
            if err.code == "quota_exceeded" or status in (400, 401, 403, 404, 405, 422):
                raise err
            if err.code == "rate_limited" or status == 429:
                waits += 1
                wait = self._backoff(err.retry_after, waits)
                if waited + wait > self.WAIT_SECONDS:
                    raise err
                if wait >= 5 or "429" not in self.told:
                    self.told.add("429")
                    self.say(f"rate limited; waiting {wait:.1f}s ({_short(err.message, 80)})")
            elif status == 409 and err.retry_after is not None:
                waits += 1
                wait = self._backoff(err.retry_after, waits)
                if waited + wait > self.WAIT_SECONDS:
                    raise err
            elif status >= 500:
                server += 1
                tries = 2 if status == 500 else self.SERVER_TRIES
                if server >= tries:
                    raise err
                wait = err.retry_after if err.retry_after is not None else min(30.0, 2.0 ** server)
                self.say(f"server busy ({status}); waiting {wait:.1f}s, then sending the same "
                         "request again")
            else:
                raise err
            wait = min(float(wait), 300.0)
            waited += wait
            self.sleep(wait)

    @staticmethod
    def _backoff(retry_after: Optional[float], waits: int) -> float:
        """How long to wait before sending a refused call again: what the
        server said, and after a few refusals in a row (something else is
        spending the same calls, such as the Simulator open on this run)
        at least a growing pause, up to 30 seconds."""
        wait = retry_after if retry_after is not None else 1.0
        if waits > 3:
            wait = max(wait, min(30.0, 0.5 * 2.0 ** (waits - 4)))
        return wait

    def _pace(self, headers: Any) -> None:
        """Keep to the plan's calls a minute: when X-RateLimit-Remaining is
        0, wait one refill interval before the next call instead of being
        refused."""
        try:
            limit = int(headers.get("X-RateLimit-Limit") or 0)
            remaining = int(headers.get("X-RateLimit-Remaining") or 1)
        except ValueError:
            return
        if limit > 0 and remaining <= 0:
            self.pause = 60.0 / limit
            if "pace" not in self.told:
                self.told.add("pace")
                self.say(f"keeping to your plan's {limit} calls a minute")


# -- the bot -------------------------------------------------------------------------------


class UsageError(Exception):
    pass


def read_start_prices(path: str) -> Dict[str, float]:
    """--start-prices FILE: "TICKER,price" lines, a header line allowed.
    The prices are yours; the session reports each one's gap to the model's
    fair value (start_gaps)."""
    try:
        text = Path(path).read_text(encoding="utf-8")
    except OSError as e:
        raise UsageError(f"--start-prices {path}: {e}") from None
    prices: Dict[str, float] = {}
    rows = [r for r in csv.reader(text.splitlines()) if r and any(c.strip() for c in r)]
    for n, row in enumerate(rows, 1):
        cells = [c.strip() for c in row]
        try:
            price = float(cells[1].lstrip("$")) if len(cells) >= 2 else None
        except ValueError:
            price = None
        if price is None and n == 1:
            continue                                   # a header line
        if price is None or not price > 0 or not cells[0]:
            raise UsageError(f"--start-prices {path}, line {n}: write TICKER,price with a "
                             f"price above 0, such as NVDA,181.20")
        prices[cells[0].upper()] = price
    if not prices:
        raise UsageError(f"--start-prices {path} has no TICKER,price lines")
    return prices


class BotError(Exception):
    """The user's function raised, or returned something that is not an order."""

    def __init__(self, where: str, exc: BaseException) -> None:
        self.where = where
        self.exc = exc
        super().__init__(f"{where}: {exc}")


class Bot:
    def __init__(self, path: str, *, entry: str = "on_step", name: Optional[str] = None) -> None:
        file = Path(path)
        if not file.is_file():
            raise UsageError(f"no file {path}")
        self.path = file.resolve()
        self.name = name or file.name
        self.entry = entry
        sys.path.insert(0, str(self.path.parent))
        spec = importlib.util.spec_from_file_location(f"tfrun_bot_{file.stem}", self.path)
        if spec is None or spec.loader is None:
            raise UsageError(f"{path} is not a Python file")
        module = importlib.util.module_from_spec(spec)
        for name in ("buy", "sell", "target", "cancel", "twap", "vwap"):   # without the import
            setattr(module, name, globals()[name])
        sys.modules[spec.name] = module
        try:
            spec.loader.exec_module(module)
        except Exception as e:  # noqa: BLE001 - the user's module failed to load
            raise BotError(f"loading {self.name}", e) from None
        self.module = module
        self.on_step = getattr(module, entry, None)
        if not callable(self.on_step):
            raise UsageError(f"{self.name} has no {entry}(market) function")
        self.setup = getattr(module, "setup", None)
        self.on_day_end = getattr(module, "on_day_end", None)
        for name in ("setup", "on_day_end"):
            if getattr(self, name) is not None and not callable(getattr(self, name)):
                raise UsageError(f"{self.name}: {name} must be a function")

    def call(self, name: str, arg: Any) -> List[Order]:
        fn = getattr(self, name)
        where = self.entry if name == "on_step" else name
        try:
            out = fn(arg)
        except Exception as e:  # noqa: BLE001 - reported with its traceback
            raise BotError(where, e) from None
        if name == "setup":
            return []
        if out is None:
            return []
        if isinstance(out, Order):
            return [out]
        try:
            orders = list(out)
        except TypeError:
            orders = [out]
        for o in orders:
            if not isinstance(o, Order):
                raise BotError(name, TypeError(
                    f"{name} returned {o!r}, which is not an order; return a list of buy(...), "
                    "sell(...), target(...), cancel(...), twap(...) or vwap(...), or []"))
        return orders


def parse_strategy(ref: str) -> tuple:
    """"st_x:3" -> ("st_x", 3); "st_x" -> ("st_x", None)."""
    sid, _, v = (ref or "").partition(":")
    if not re.fullmatch(r"st_[0-9a-z]+", sid) or (v and not v.isdigit()):
        raise UsageError(f"--strategy {ref!r} is not a strategy such as st_0123456789ab:3")
    return sid, (int(v) if v else None)


def fetch_strategy(api: "Api", ref: str) -> Dict[str, Any]:
    """Version V (the latest without one) of a saved strategy, from GET
    /v1/strategies/{id}: {"strategy_id", "version", "name", "code", "entry"}."""
    sid, want = parse_strategy(ref)
    try:
        body = api.get(f"/v1/strategies/{urllib.parse.quote(sid)}",
                       {"version": want} if want is not None else None)
    except ApiError as e:
        if e.status == 404:
            raise UsageError(f"no strategy {sid} on {api.base} for this key") from None
        raise
    st = body.get("strategy") if isinstance(body.get("strategy"), dict) else body
    versions = [v for v in (st.get("versions") or []) if isinstance(v, dict)]
    if not versions and isinstance(body.get("version"), dict):
        versions = [body["version"]]
    if not versions and body.get("code") is not None:
        versions = [body]
    if want is None:
        want = st.get("latest_version") or max((int(v.get("version") or 0) for v in versions),
                                               default=None)
    got = next((v for v in versions if v.get("version") is not None
                and int(v["version"]) == int(want or 0)), None)
    if got is None:
        raise UsageError(f"strategy {sid} has no version {want}")
    if not got.get("code"):
        raise UsageError(f"{sid}:{want} has no code to run here (it is a "
                         f"{st.get('kind') or 'rule or agent'} strategy)")
    return {"strategy_id": sid, "version": int(want), "name": st.get("name") or sid,
            "code": got["code"], "entry": got.get("entry") or "on_step"}


def strategy_bot(api: "Api", ref: str) -> Bot:
    """The saved strategy as a Bot: its code in a file of its own, so a
    traceback shows its lines."""
    st = fetch_strategy(api, ref)
    folder = Path(tempfile.mkdtemp(prefix="tfrun-"))
    file = folder / f"{st['strategy_id']}_v{st['version']}.py"
    file.write_bytes(st["code"].encode("utf-8"))
    bot = Bot(str(file), entry=st["entry"], name=f"{st['strategy_id']}:{st['version']}")
    bot.strategy = st
    return bot


def user_traceback(exc: BaseException) -> str:
    """The traceback without tfrun's own frames in front."""
    tb = exc.__traceback__
    while tb is not None and (os.path.abspath(tb.tb_frame.f_code.co_filename) == HERE
                              or tb.tb_frame.f_code.co_filename.startswith("<frozen")):
        tb = tb.tb_next
    return "".join(traceback.format_exception(type(exc), exc, tb))


# -- running -------------------------------------------------------------------------------


class Stop(Exception):
    """Stop the run: a plan limit or a refusal, already explained to the user."""

    def __init__(self, code: int, status: str) -> None:
        self.exit_code = code
        self.status = status
        super().__init__(status)


def _pct(x: Optional[float]) -> str:
    return "n/a" if x is None else f"{x:+.2f}%"


def _money(x: Optional[float]) -> str:
    return "n/a" if x is None else f"{x:,.2f}"


def _host() -> str:
    """This computer's short name, as the Simulator's strip shows it."""
    name = socket.gethostname() or "this computer"
    return name.split(".")[0][:80] or "this computer"


def error_line(exc: BaseException, path: Optional[str]) -> Optional[int]:
    """The line of the bot's own file the exception came from."""
    tb = exc.__traceback__
    line = None
    while tb is not None:
        if path and os.path.abspath(tb.tb_frame.f_code.co_filename) == path:
            line = tb.tb_lineno
        tb = tb.tb_next
    return line


class Beat:
    """Tells tradefloor that this run is alive: a heartbeat when it starts
    and every HEARTBEAT_EVERY seconds from a thread of its own, and a
    goodbye when it ends. Best effort: a missed beat never stops the run,
    and a server without /v1/trader stops the beats."""

    def __init__(self, api: "Api", target_type: str, target_id: str, *,
                 target_day: Optional[int] = None, day: Optional[int] = None,
                 days: Optional[int] = None, every: float = HEARTBEAT_EVERY) -> None:
        self.api = api
        self.target_type = target_type
        self.target_id = target_id
        self.target_day = target_day
        self.day = day
        self.days = days        # --days, which the strip says this run played
        self.line_id: Optional[str] = None
        self.last_error: Optional[dict] = None
        self.every = every
        self.off = False
        self._stop = threading.Event()
        self._thread: Optional[threading.Thread] = None

    def _post(self, path: str, body: dict) -> None:
        if self.off or not self.api.key:
            return
        req = urllib.request.Request(
            self.api.base + path, data=json.dumps(body).encode("utf-8"), method="POST",
            headers={"Content-Type": "application/json", "Authorization": f"Bearer {self.api.key}",
                     "Idempotency-Key": f"tfrun-beat-{uuid.uuid4().hex}",
                     "User-Agent": f"tfrun/{__version__}"})
        try:
            with urllib.request.urlopen(req, timeout=10) as r:
                r.read()
        except urllib.error.HTTPError as e:
            if e.code in (404, 405):        # a server without the Simulator
                self.off = True
        except (OSError, ValueError, http.client.HTTPException):
            pass

    def send(self) -> None:
        body: Dict[str, Any] = {"target_type": self.target_type, "target_id": self.target_id,
                                "day": self.day, "target_day": self.target_day,
                                "host": _host(), "client_version": __version__}
        if self.line_id:
            body["line_id"] = self.line_id
        if self.days:
            body["days"] = self.days
        if self.last_error is not None:
            body["last_error"] = self.last_error
        self._post("/v1/trader/heartbeat", body)

    def _loop(self) -> None:
        while not self._stop.wait(self.every):
            self.send()

    def start(self) -> "Beat":
        self.send()
        self._thread = threading.Thread(target=self._loop, name="tfrun-heartbeat", daemon=True)
        self._thread.start()
        return self

    def stop(self, reason: str = "finished") -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=2)
        self._post("/v1/trader/goodbye", {"target_type": self.target_type,
                                          "target_id": self.target_id, "day": self.day,
                                          "reason": reason})


class Runner:
    def __init__(self, api: Api, bot: Bot, args: argparse.Namespace, *, site: str,
                 offline: bool, out: Any = None, err: Any = None) -> None:
        self.api = api
        self.bot = bot
        self.args = args
        self.site = site.rstrip("/")
        self.offline = offline
        self.out = out or sys.stdout
        self.err = err or sys.stderr
        self.run_id = uuid.uuid4().hex[:12]
        self.fills = 0
        self.refusals = 0
        self.open_session: Optional[str] = None      # left open if the run stops
        self.suite_run: Optional[str] = None
        self.check_api: Optional[Api] = None         # TF_CHECK_KEY's, for --check
        self.last_obs: Optional[dict] = None
        self._bars: Dict[str, Any] = {}
        self._progress = False
        self._hosted: Optional[bool] = None
        self.beats: List[Beat] = []                  # what tells the Simulator it runs
        self.carry_on: Optional[str] = None          # a line: on_step errors skip the day
        self.error_days: List[int] = []              # the days on_step raised on (a line)
        self._errored: set = set()                   # (line, day) pairs already reported

    # -- output ---------------------------------------------------------------------------

    def say(self, text: str) -> None:
        self._end_progress()
        print(text, file=self.err, flush=True)

    def result(self, text: str) -> None:
        self._end_progress()
        if not self.args.json:
            print(text, file=self.out, flush=True)

    def progress(self, obs: dict, last_day: int, label: str = "") -> None:
        clock = obs["clock"]
        done = clock["day"] + (0 if clock["market_open"] else 1)
        worth = obs["account"]["net_worth"]
        text = (f"{label}{done} of {last_day + 1} days  net worth {worth:,.2f}  "
                f"fills {self.fills}")
        if self.args.verbose:
            self.say(text)
        elif self.err.isatty():
            self.err.write("\r" + text.ljust(70))
            self.err.flush()
            self._progress = True

    def _end_progress(self) -> None:
        if self._progress:
            self.err.write("\n")
            self.err.flush()
            self._progress = False

    def usage(self) -> Optional[dict]:
        """GET /v1/usage (it spends no call); None on a server of your own."""
        usage = self.api.optional("/v1/usage")
        self._hosted = isinstance(usage, dict)
        return usage if self._hosted else None

    def link(self, session_id: str) -> str:
        """Where to watch the session: the Simulator on the hosted app."""
        if self.offline:
            return "stored in ~/.tradefloor/sessions"
        if self._hosted is None:
            try:
                self.usage()
            except ApiError:
                self._hosted = not _is_local(self.site)
        if not self._hosted:
            return f"{self.site}/v1/sessions/{session_id}"
        return f"{self.site}/simulator?s={session_id}"

    def key(self, *parts: Any) -> str:
        return "-".join(["tfrun", self.run_id] + [str(p) for p in parts])

    # -- a plan's allowance ----------------------------------------------------------------

    def check_allowance(self, sim_days: float, what: str) -> None:
        """Refuse up front a run that cannot fit in what is left today, or
        whose --check this key could not read at the end."""
        usage = self.usage()
        if usage is None:
            return
        if (self.args.check and self.check_api is None
                and (usage.get("key") or {}).get("scope") == "trader"):
            raise UsageError("--check reads GET /v1/sessions/{id}/behaviour, which a trader key "
                             "cannot. Set TF_CHECK_KEY to a full or read-only key.")
        left = (usage.get("remaining_today") or {}).get("sim_days")
        if left is None or left >= sim_days:
            return
        resets = (usage.get("now") or {}).get("resets_at", "00:00 UTC")
        self.say(f"{what} needs {sim_days:g} simulated days and your plan has {left:g} left "
                 f"today. The allowance resets at {resets}.")
        raise Stop(EXIT_LIMIT, "stopped")

    def explain_limit(self, e: ApiError) -> None:
        self.say(f"stopped: {e.message}")
        if e.hint:
            self.say(e.hint)
        usage = self.usage() if e.status != 0 else None
        if usage is not None:
            plan, today = usage.get("plan") or {}, usage.get("today") or {}
            parts = []
            if plan.get("sim_days_per_day") is not None:
                parts.append(f"{today.get('sim_days', 0):g} of {plan['sim_days_per_day']:g} "
                             "simulated days")
            if plan.get("compute_seconds_per_day") is not None:
                parts.append(f"{today.get('compute_seconds', 0):g} of "
                             f"{plan['compute_seconds_per_day']:g} compute seconds")
            resets = (usage.get("now") or {}).get("resets_at")
            if parts:
                self.say("used today: " + ", ".join(parts)
                         + (f"; resets at {resets}" if resets else ""))

    # -- one session ----------------------------------------------------------------------

    def history(self, sid: str) -> Callable[[int, int], Dict[str, List[dict]]]:
        def fetch(day: int, days: int) -> Dict[str, List[dict]]:
            want = max(days, 20)
            have = self._bars.get(sid)
            if have is None or have[0] != day or have[1] < want:
                # Days below 0 are history the session opened with (history_days);
                # the server starts at the first day it has.
                got = self.api.get(f"/v1/sessions/{sid}/bars",
                                   {"resolution": "day", "since_day": day - want})
                self._bars[sid] = (day, want, got)
                have = self._bars[sid]
            return have[2]
        return fetch

    def next_move(self, obs: dict, last_day: int) -> dict:
        clock = obs["clock"]
        if not clock.get("market_open"):
            return {"until": "next_open"}
        finishing = clock["day"] >= last_day or self.bot.on_day_end is not None
        if self.args.every == "day":
            return {"until": "close"} if finishing else {"until": "next_open"}
        tps = clock.get("ticks_per_step") or 30
        if clock["tick"] + tps >= TICKS_PER_DAY and not finishing:
            return {"until": "next_open"}
        return {"steps": 1}

    def submit(self, sid: str, orders: List[Order], market: Market) -> List[dict]:
        """Send what the bot returned; the orders the server refused."""
        requests: List[dict] = []
        refused: List[dict] = []
        where = f"d{market.day}-t{market.tick}"
        for o in orders:
            if o.kind == "cancel":
                try:
                    self.api.delete(f"/v1/sessions/{sid}/orders/{urllib.parse.quote(o.order_id)}",
                                    key=self.key(sid[:8], where, "cancel", o.order_id))
                except ApiError as e:
                    if e.code == "quota_exceeded" or e.status in (0, 401):
                        raise
                    refused.append({"cancel": o.order_id, "code": e.code, "message": e.message,
                                    "hint": e.hint})
                continue
            if o.kind in ("twap", "vwap"):
                body = {"ticker": o.ticker, "quantity": o.shares, "algo": o.kind,
                        **{k: v for k, v in (o.algo or {}).items() if v is not None}}
                if o.limit is not None:
                    body["limit_price"] = o.limit
                if o.note:
                    body["note"] = o.note
                try:
                    self.api.post(f"/v1/sessions/{sid}/algo-orders", body,
                                  key=self.key(sid[:8], where, o.kind, o.ticker, o.shares))
                except ApiError as e:
                    if e.code == "quota_exceeded" or e.status in (0, 401):
                        raise
                    refused.append({"ticker": o.ticker, "side": body.get("side", "buy"),
                                    "quantity": o.shares, "code": e.code, "message": e.message,
                                    "hint": e.hint})
                continue
            if o.kind == "target":
                price = market.prices.get(o.ticker)
                if not price:
                    refused.append({"ticker": o.ticker, "side": "target", "quantity": 0,
                                    "code": "invalid_order", "message": (
                                        f"{o.ticker!r} has no price in this market"
                                        if o.ticker not in market.prices else "no last price"),
                                    "hint": "Use a ticker from market.tickers."})
                    continue
                have = market.position(o.ticker)
                want = o.shares if o.weight is None else o.weight * market.net_worth / price
                delta = int(want - have)            # whole shares, toward zero
                if delta == 0:
                    continue
                side, qty = ("buy", delta) if delta > 0 else ("sell", -delta)
            else:
                side, qty = o.kind, o.shares
            req = {"ticker": o.ticker, "side": side, "quantity": qty, "time_in_force": o.tif}
            if o.limit is not None:
                req.update(type="limit", limit_price=o.limit)
            if o.note:
                req["note"] = o.note
            requests.append(req)
        for i in range(0, len(requests), BATCH):
            chunk = requests[i:i + BATCH]
            got = self.api.post(f"/v1/sessions/{sid}/orders/batch", {"orders": chunk},
                                key=self.key(sid[:8], where, "orders", i // BATCH))
            for req, row in zip(chunk, got.get("results") or []):
                if "error" in row:
                    e = row["error"]
                    refused.append({**req, "code": e.get("code"), "message": e.get("message"),
                                    "hint": e.get("hint")})
        for r in refused:
            self.refusals += 1
            if self.refusals <= MAX_REFUSALS_SHOWN:
                what = (f"cancel {r['cancel']}" if "cancel" in r
                        else f"{r['side']} {r['quantity']:g} {r['ticker']}")
                self.say(f"day {market.day} {market.time}: {what} refused: {r['code']}: "
                         f"{_short(r['message'], 90)}")
            elif self.refusals == MAX_REFUSALS_SHOWN + 1:
                self.say("more refusals are not printed; market.refused has them")
        return refused

    def play(self, sid: str, last_day: int, config: Dict[str, Any], label: str = "") -> dict:
        """Play an open session until `last_day` has closed. The last observation."""
        self.open_session = sid
        self.fills = 0
        obs = self.api.get(f"/v1/sessions/{sid}/observation", {"view": "compact"})
        self.last_obs = obs
        history = self.history(sid)
        if self.bot.setup is not None:
            self.bot.call("setup", dict(config))
        fills: List[dict] = []
        refused: List[dict] = []
        while obs.get("status", "open") == "open" and obs["clock"]["day"] <= last_day:
            clock = obs["clock"]
            self.beat_day(clock)
            hook = "on_step" if clock["market_open"] else "on_day_end"
            if clock["market_open"] or self.bot.on_day_end is not None:
                market = Market(obs, last_day=last_day, fills=fills, refused=refused,
                                history=history,
                                read=lambda path, query=None: self.api.get(
                                    f"/v1/sessions/{sid}{path}", query))
                try:
                    orders = self.bot.call(hook, market)
                except BotError as e:
                    if self.carry_on is None:
                        raise
                    # A line goes on: the day plays with no orders from the
                    # bot, and the error is one red row for the day.
                    self.day_failed(e, int(clock["day"]))
                    orders = []
                refused = self.submit(sid, orders, market) if orders else []
                fills = []
            if not clock["market_open"] and clock["day"] >= last_day:
                break
            move = self.next_move(obs, last_day)
            res = self.api.post(f"/v1/sessions/{sid}/advance", move, query={"view": "compact"},
                                key=self.key(sid[:8], f"d{clock['day']}-t{clock['tick']}", "adv"))
            if res.get("shared") and res["shared"].get("state") in ("waiting", "lobby", "paused"):
                res = self.wait_turn(sid, clock, res)
            fills += res.get("fills") or []
            self.fills += len(res.get("fills") or [])
            obs = res["observation"]
            self.last_obs = obs
            now = obs["clock"]
            if now["day"] != clock["day"] or (not now["market_open"] and now["day"] >= last_day):
                self.progress(obs, last_day, label)
        self._end_progress()
        self.beat_day(obs["clock"])
        return obs

    def beat_day(self, clock: dict) -> None:
        """The day the heartbeats report: the one on_step is playing."""
        day = clock.get("day")
        if day is None:
            return
        for b in self.beats:
            b.day = int(day)
            # A heartbeat carries only the error of the day it reports.
            if b.last_error is not None and b.last_error.get("day") != int(day):
                b.last_error = None

    def day_failed(self, e: BotError, day: int) -> None:
        """on_step raised while driving a line: tell the line once a day (one
        red row, one notification), show the traceback, and carry on."""
        if (self.carry_on, day) in self._errored:
            return
        self._errored.add((self.carry_on, day))
        self.error_days.append(day)
        line = error_line(e.exc, str(self.bot.path))
        if len(self.error_days) == 1:
            self.say(user_traceback(e.exc).rstrip())
        self.say(f"day {day}: {e.where} raised {type(e.exc).__name__}: {e.exc}"
                 + (f" (line {line})" if line else "") + "; the day went ahead with no orders")
        err = {"day": day, "type": type(e.exc).__name__, "message": str(e.exc)[:200],
               "line": line}
        # The heartbeat carries the error: the server writes the day's red
        # row and notification from it. Without one, report it directly.
        if self.carry_on and not self.beats:
            self.report_error(self.carry_on, e, day=day, line=line)
        for b in self.beats:
            b.last_error = err
            b.send()

    # -- a shared market (W21) ----------------------------------------------------------------

    def wait_turn(self, sid: str, clock: dict, res: dict) -> dict:
        """In a shared market the step runs when every participant is ready
        (or at the host's timeout), maybe in another's call: look again as
        the server says until the clock moves, then collect what filled."""
        shared = res["shared"]
        said = False
        while True:
            if not said:
                self.say(f"day {clock['day']}: {_short(shared.get('message') or 'waiting', 100)}")
                said = True
            self.api.sleep(float(shared.get("retry_after") or 2.0))
            obs = self.api.get(f"/v1/sessions/{sid}/observation", {"view": "compact"})
            now = obs["clock"]
            moved = (now["day"], now["tick"], now.get("market_open")) != \
                (clock["day"], clock["tick"], clock.get("market_open"))
            if moved or obs.get("status", "open") != "open":
                break
            info = self.api.get(f"/v1/sessions/{sid}")
            shared = info.get("shared") or {}
            if shared.get("status") == "closed":
                break
        got = self.api.get(f"/v1/sessions/{sid}/fills", {"since_day": clock["day"]})
        after = (clock["day"], clock["tick"])
        fills = [{"order_id": f["order_id"], "ticker": f["ticker"], "side": f["side"],
                  "quantity": f["quantity"], "price": f["price"], "day": f["at"]["day"],
                  "tick": f["at"]["tick"], "liquidity": f.get("liquidity"),
                  **({"counterparty": f["counterparty"]} if f.get("counterparty") else {})}
                 for f in got if (f["at"]["day"], f["at"]["tick"]) >= after]
        return {**res, "observation": obs, "fills": fills}

    def shared(self) -> dict:
        """Take part in a shared market through this key's seat."""
        a = self.args
        seat_line = bool(a.join and not a.shared)
        if seat_line:
            # A market made in the Simulator: its join code alone takes a
            # seat, held by this key as code; the seat is the line it gives.
            got = self.api.post("/v1/markets/join", {"code": a.join, "kind": "code"},
                                key=self.key("join"))
            sid, market = got["line_id"], got.get("simulation_id")
        else:
            if a.join:
                seat = self.api.post("/v1/shared/join", {"code": a.join}, key=self.key("join"))
            else:
                seat = self.api.get(f"/v1/shared/{urllib.parse.quote(a.shared)}/seat")
            sid, market = seat["seat_id"], seat.get("market_id")
        info = self.api.get(f"/v1/sessions/{sid}")
        sh = info.get("shared") or {}
        last_day = info["clock"]["day"] + a.days - 1
        self.say(f"taking part in shared market {sh.get('title') or market} as "
                 f"{sh.get('label')}: {sh.get('participants')} participants, "
                 f"{len(info['tickers'])} companies, {a.days} days; the clock "
                 + ("moves when everyone is ready"
                    + (f" or after {sh['timeout_s']:g} s" if sh.get("timeout_s") else "")
                    if sh.get("mode") == "all_ready" else "is the host's"))
        cfg = info["config"]
        config = {"session_id": sid, "tickers": info["tickers"], "days": a.days,
                  "every": a.every, "shared": sh, **cfg}
        if seat_line:
            # The seat is a line driven as code: the strip hears from this
            # run (heartbeats, a goodbye) and an on_step error is a red row
            # for the day, as on any line tfrun drives.
            config.update(line_id=sid, simulation_id=market)
            beat = Beat(self.api, "line", sid, target_day=last_day,
                        day=max(0, int(info["clock"]["day"])), days=a.days)
            self.beats.append(beat)
            beat.start()
            self.carry_on = sid
            reason = "interrupted"
            try:
                self.play(sid, last_day, config)
                reason = "finished"
            except BotError as e:
                self.report_error(sid, e)
                raise
            finally:
                self.carry_on = None
                self.beats.remove(beat)
                beat.stop(reason)
        else:
            self.play(sid, last_day, config)
        report = None
        if not a.keep_open and self.api.get(f"/v1/sessions/{sid}")["status"] == "open":
            report = self.api.post(f"/v1/sessions/{sid}/close", key=self.key(sid[:8], "leave"))
            self.open_session = None
        return self.summary(self.api.get(f"/v1/sessions/{sid}"), report, days=a.days)

    # -- lines of the Simulator -----------------------------------------------------------

    def line_link(self, simulation_id: Optional[str], line_id: str) -> str:
        if not simulation_id:
            return f"{self.site}/simulator?s={urllib.parse.quote(line_id)}"
        return (f"{self.site}/simulator?sim={urllib.parse.quote(simulation_id)}"
                f"&line={urllib.parse.quote(line_id)}")

    def report_error(self, line_id: str, e: BotError, *, day: Optional[int] = None,
                     line: Optional[int] = None) -> None:
        """Tell the line: a red row in its Activity. Best effort."""
        body: Dict[str, Any] = {"where": e.where, "error": f"{type(e.exc).__name__}: {e.exc}"}
        if day is not None:
            body.update(day=day, type=type(e.exc).__name__, message=str(e.exc)[:200], line=line)
        try:
            self.api.post(f"/v1/lines/{urllib.parse.quote(line_id)}/errors", body,
                          key=self.key(line_id[:8], "error", day if day is not None else "x"))
        except ApiError:
            pass

    def drive(self, line_id: str, simulation_id: Optional[str], label: str = "") -> dict:
        """Play a line's session from where it is for --days days, and leave
        it open: the line goes on with its simulation."""
        a = self.args
        info = self.api.get(f"/v1/sessions/{urllib.parse.quote(line_id)}")
        link = self.line_link(simulation_id, line_id)
        if info["status"] != "open":
            self.say(f"line {line_id} is closed already: {link}")
            out = self.summary(info, None, days=a.days, link=link)
        else:
            clock = info["clock"]
            first = clock["day"] if clock["market_open"] else clock["day"] + 1
            last_day = first + a.days - 1
            self.check_allowance(a.days, f"{a.days} days on this line")
            self.say(f"{label}driving line {line_id} from day {first} to day {last_day}: {link}")
            config = {"session_id": line_id, "line_id": line_id, "simulation_id": simulation_id,
                      "tickers": info["tickers"], "days": a.days, "every": a.every,
                      **info["config"]}
            beat = Beat(self.api, "line", line_id, target_day=last_day, day=max(0, first - 1),
                        days=a.days)
            for b in self.beats:
                b.line_id = line_id
                b.last_error = None     # the last line's error is not this line's
            self.beats.append(beat)
            beat.start()
            self.carry_on = line_id
            reason = "interrupted"
            try:
                self.play(line_id, last_day, config, label)
                reason = "finished"
            except BotError as e:
                self.report_error(line_id, e)
                raise
            finally:
                self.carry_on = None
                self.beats.remove(beat)
                beat.stop(reason)
            self.open_session = None
            out = self.summary(self.api.get(f"/v1/sessions/{urllib.parse.quote(line_id)}"),
                               None, days=a.days, link=link)
        out.update(mode="line", line_id=line_id, simulation_id=simulation_id)
        return out

    def line(self) -> dict:
        try:
            got = self.api.get(f"/v1/lines/{urllib.parse.quote(self.args.line)}")
        except ApiError as e:
            if e.code != "forbidden":
                raise
            # A trader key reads only the simulations whose lines name it;
            # it still trades its own session, and the link finds the line.
            return self.drive(self.args.line, None)
        return self.drive(got["line"]["line_id"], got["simulation_id"])

    def lines(self) -> dict:
        """--simulation: each code line the strategy version trades."""
        a = self.args
        st = getattr(self.bot, "strategy", None) or {}
        fam = self.api.get(f"/v1/simulations/{urllib.parse.quote(a.simulation)}")
        if fam.get("status") == "draft":
            raise UsageError(f"simulation {a.simulation} has not started yet; start it in the "
                             "Simulator, then run this again")
        mine = [x for x in fam.get("lines") or []
                if (x.get("trader") or {}).get("kind") == "code"
                and (x.get("trader") or {}).get("strategy_id") == st.get("strategy_id")
                and int((x.get("trader") or {}).get("version") or 0) == int(st.get("version") or 0)
                and x.get("status", "open") == "open"]
        if not mine:
            raise UsageError(f"no open line of simulation {a.simulation} is traded by "
                             f"{st.get('strategy_id')}:{st.get('version')}")
        self.check_allowance(a.days * len(mine), f"{len(mine)} lines of {a.days} days")
        done = []
        beat = Beat(self.api, "simulation", a.simulation, target_day=a.days)
        self.beats.append(beat)
        beat.start()
        reason = "interrupted"
        try:
            for i, x in enumerate(mine):
                beat.line_id = x["line_id"]
                beat.last_error = None
                done.append(self.drive(x["line_id"], a.simulation,
                                       label=f"line {i + 1} of {len(mine)}: "))
            reason = "finished"
        finally:
            self.beats.remove(beat)
            beat.stop(reason)
        out = {"tfrun": __version__, "mode": "lines", "status": "done",
               "simulation_id": a.simulation, "lines": done, "orders_refused": self.refusals,
               "link": f"{self.site}/simulator?sim={urllib.parse.quote(a.simulation)}",
               "error": None}
        self.result(f"simulation    {a.simulation}: {len(done)} lines played  {out['link']}")
        return out

    # -- one simulation --------------------------------------------------------------------

    def session_config(self) -> Dict[str, Any]:
        a = self.args
        body: Dict[str, Any] = {}
        if a.config:
            try:
                loaded = json.loads(Path(a.config).read_text(encoding="utf-8"))
            except (OSError, ValueError) as e:
                raise UsageError(f"--config {a.config}: {e}") from None
            if not isinstance(loaded, dict):
                raise UsageError(f"--config {a.config} must hold a JSON object of session settings")
            body.update(loaded)
        for field, value in (("preset", a.preset), ("seed", a.seed), ("universe_size", a.companies),
                             ("scenario", a.scenario), ("history_days", a.history_days)):
            if value is not None:
                body[field] = value
        if a.step is not None:
            body["ticks_per_step"] = a.step
        if a.roster or a.tickers:
            companies = dict(body.get("companies") or {})
            if a.roster:
                companies["roster"] = a.roster
            if a.tickers:
                companies["tickers"] = [t.strip() for t in a.tickers.split(",") if t.strip()]
            body["companies"] = companies
        if a.start_prices:
            body["start_prices"] = read_start_prices(a.start_prices)
            body["price_mode"] = a.price_mode
        if a.scenario_file:
            try:
                doc = json.loads(Path(a.scenario_file).read_text(encoding="utf-8"))
            except (OSError, ValueError) as e:
                raise UsageError(f"--scenario-file {a.scenario_file}: {e}") from None
            if not isinstance(doc, dict):
                raise UsageError(f"--scenario-file {a.scenario_file} must hold a scenario "
                                 "document, a JSON object")
            body["scenario"] = doc
        body.setdefault("label", f"tfrun {self.bot.name}")
        return body

    def simulate(self) -> dict:
        a = self.args
        if a.resume:
            sid = a.resume
            info = self.api.get(f"/v1/sessions/{sid}")
            if info["status"] != "open":
                self.say(f"session {sid} is closed already")
                return self.summary(info, None, days=a.days)
            left = a.days - info["clock"]["day"]
            if left > 0:
                self.check_allowance(left, f"The rest of this run ({left} days)")
            self.say(f"resuming {sid} at day {info['clock']['day']}")
        else:
            history = a.history_days or 0
            self.check_allowance(a.days + history, f"A {a.days}-day run" + (
                f" with {history} days of history" if history else ""))
            if a.template:
                # A fresh seed on the template's settings; a trader key sees no seed.
                body = {"template": a.template, "label": f"tfrun {self.bot.name}"}
                info = self.api.post("/v1/sessions/from-template", body, key=self.key("open"))
            else:
                info = self.api.post("/v1/sessions", self.session_config(),
                                     key=self.key("open"))
            sid = info["session_id"]
            cfg = info["config"]
            what = ("each day at the open" if a.every == "day"
                    else f"every {cfg.get('ticks_per_step', 30)}-minute step")
            past = (f" after {cfg['history_days']} days of history" if cfg.get("history_days")
                    else "")
            seed = cfg.get("seed")
            self.say(f"opened {sid}: {cfg.get('preset')}, "
                     + (f"seed {seed}, " if seed is not None else "a seed you do not see, ")
                     + f"{len(info['tickers'])} companies, {a.days} days{past}; on_step runs {what}")
        cfg = info["config"]
        config = {"session_id": sid, "tickers": info["tickers"], "days": a.days,
                  "every": a.every, **cfg}
        self.play(sid, a.days - 1, config)
        report = None
        if not a.keep_open:
            report = self.api.post(f"/v1/sessions/{sid}/close", key=self.key(sid[:8], "close"))
            self.open_session = None
        return self.summary(self.api.get(f"/v1/sessions/{sid}"), report, days=a.days)

    def summary(self, info: dict, report: Optional[dict], *, days: int,
                link: Optional[str] = None) -> dict:
        """Work out the result, print it, and return it for --json."""
        sid = info["session_id"]
        cash = float(info["config"].get("cash") or 1_000_000.0)
        series = self.api.optional(f"/v1/sessions/{sid}/series",
                                   {"fields": "net_worth,index", "since_day": 0}) or {}
        worths = [(d, v) for d, v in zip(series.get("days") or [], series.get("net_worth") or [])
                  if v is not None]
        index = [v for v in series.get("index") or [] if v is not None]
        if report is not None:
            worth = report["account"]["net_worth"]
        elif self.last_obs is not None and self.last_obs.get("session_id") == sid:
            worth = self.last_obs["account"]["net_worth"]
        elif worths:
            worth = worths[-1][1]
        else:
            worth = cash
        peak, peak_day, drawdown, dd_from, dd_to = cash, 0, 0.0, None, None
        for d, v in worths:
            if v > peak:
                peak, peak_day = v, d
            if peak > 0 and v / peak - 1 < drawdown:
                drawdown, dd_from, dd_to = v / peak - 1, peak_day, d
        fills = report["fills"] if report is not None else self.fills
        out: Dict[str, Any] = {
            "tfrun": __version__, "mode": "simulation", "status": "done", "session_id": sid,
            "session_status": info.get("status"), "preset": info["config"].get("preset"),
            "seed": info["config"].get("seed"), "companies": len(info.get("tickers") or []),
            "days": days, "day_reached": info["clock"]["day"],
            "net_worth": round(worth, 2), "return_pct": round((worth / cash - 1) * 100, 4),
            "max_drawdown_pct": round(drawdown * 100, 4), "fills": fills,
            "index_return_pct": round(index[-1] - 100.0, 4) if index else None,
            "buy_and_hold_return_pct": None, "vs_buy_and_hold_points": None,
            "orders_refused": self.refusals, "link": link or self.link(sid),
            "state_hash": (report or {}).get("state_hash") or (
                self.last_obs or {}).get("state_hash"),
            "caveats": (report or {}).get("caveats") or [], "error": None,
        }
        if not self.offline and not self.args.no_report:
            card = self.report_card(sid)
            if card:
                out.update(card)
        if self.args.check:
            out.update(self.checks([sid]))
        self.result(f"session       {sid} ({out['preset']}, seed {out['seed']}, "
                    f"{out['companies']} companies)")
        self.result(f"return        {_pct(out['return_pct'])}   net worth "
                    f"{_money(out['net_worth'])}")
        if dd_from is not None:
            self.result(f"drawdown      {_pct(out['max_drawdown_pct'])}   worst fall from a "
                        f"closing high, day {dd_from} to day {dd_to}")
        else:
            self.result("drawdown      none at any close")
        self.result(f"trades        {fills} fills" + (f", {self.refusals} orders refused"
                                                     if self.refusals else ""))
        if out["index_return_pct"] is not None:
            self.result(f"index         {_pct(out['index_return_pct'])}   the market's "
                        "cap-weighted index")
        if out["vs_buy_and_hold_points"] is not None:
            ahead = out["vs_buy_and_hold_points"]
            self.result(f"buy-and-hold  {_pct(out['buy_and_hold_return_pct'])}   you are "
                        f"{abs(ahead):.2f} points {'ahead' if ahead >= 0 else 'behind'}")
        if self.args.verbose and out["caveats"]:
            self.result("caveats:")
            for c in out["caveats"]:
                self.result(f"  - {c}")
        if self.offline:
            self.result("stored        ~/.tradefloor/sessions")
        elif not (getattr(self.args, "mode", None) == "simulation" and practice(self.args)):
            self.result(("watch it      " if self._hosted else "see it        ") + out["link"])
        return out

    def checks(self, sessions: List[str]) -> dict:
        """--check: each session's behaviour checks, printed; passed is
        true when every check of every session passed."""
        api = self.check_api or self.api
        query = {"checks": ",".join(self.args.check)}
        rows, passed = [], True
        for sid in sessions:
            try:
                got = api.get(f"/v1/sessions/{sid}/behaviour", query)
            except ApiError as e:
                if e.code == "forbidden":
                    raise UsageError("--check reads GET /v1/sessions/{id}/behaviour, which this "
                                     "key's scope does not cover (a trader key). Set "
                                     "TF_CHECK_KEY to a full or read-only key.") from None
                raise
            for c in got.get("checks") or []:
                ok = bool(c.get("passed"))
                passed = passed and ok
                value = c.get("value")
                shown = "-" if value is None else f"{value:.4g}"
                where = f" {sid[:8]}" if len(sessions) > 1 else ""
                self.result(f"check{where}   {c['check']:<28} {shown:>10}   "
                            f"{'pass' if ok else 'FAIL'}" + (f" ({c['why']})" if c.get("why")
                                                              else ""))
                rows.append({"session_id": sid, **c})
        return {"checks": rows, "checks_passed": passed}

    def report_card(self, sid: str) -> Optional[dict]:
        """vs buy-and-hold from the report card, when the server has one."""
        try:
            body = self.api.optional(f"/v1/sessions/{sid}/report")
        except ApiError:
            return None
        facts = (body or {}).get("facts") or body or {}
        compare = facts.get("compare") or {}
        bh = next((r for r in compare.get("reference_agents") or []
                   if r.get("agent") == "buy_and_hold"), None)
        if bh is None or compare.get("vs_buy_and_hold_return_pct") is None:
            return None
        return {"buy_and_hold_return_pct": bh.get("return_pct"),
                "vs_buy_and_hold_points": compare["vs_buy_and_hold_return_pct"]}

    # -- a suite ---------------------------------------------------------------------------

    def suite(self) -> dict:
        a = self.args
        if a.resume:
            view = self.api.get(f"/v1/suite-runs/{a.resume}")
            suite = self.api.get(f"/v1/suites/{view['suite']}")
            self.say(f"resuming suite run {view['run_id']} ({suite['name']})")
        else:
            suite = self.api.get(f"/v1/suites/{a.suite}")
            markets = len(suite.get("cells") or [])
            self.check_allowance(suite["days"] * markets,
                                 f"{suite['name']} ({markets} markets of {suite['days']} days)")
            body: Dict[str, Any] = {"suite": suite["name"], "mode": "agent",
                                    "label": f"tfrun {self.bot.name}"}
            if a.step is not None:
                body["ticks_per_step"] = a.step
            view = self.api.post("/v1/suite-runs", body, key=self.key("suite"))
            self.say(f"suite run {view['run_id']}: {suite['name']}, {markets} markets of "
                     f"{suite['days']} days")
        self.suite_run = view["run_id"]
        run_id = view["run_id"]
        last_day = int(suite["days"]) - 1
        beat = Beat(self.api, "benchmark_run", run_id, target_day=last_day)
        self.beats.append(beat)
        beat.start()
        reason = "interrupted"
        try:
            view = self._suite_markets(view, suite, run_id, last_day)
            reason = "finished"
        finally:
            self.beats.remove(beat)
            beat.stop(reason)
        return self.suite_summary(view, suite)

    def _suite_markets(self, view: dict, suite: dict, run_id: str, last_day: int) -> dict:
        a = self.args
        while view.get("status") == "running":
            cells = view.get("cells") or []
            open_cells = [c for c in cells if c.get("status") == "open" and c.get("session_id")]
            if open_cells:
                c = open_cells[0]
                sid = c["session_id"]
                n = len(cells)
                label = f"market {c['cell'] + 1} of {n}: "
                info = self.api.get(f"/v1/sessions/{sid}")
                config = {"session_id": sid, "tickers": info["tickers"], "days": suite["days"],
                          "every": a.every, "suite": suite["name"], "market": c["cell"] + 1,
                          "markets": n, **info["config"]}
                if info["status"] == "open":
                    self.play(sid, last_day, config, label)
                    try:
                        self.api.post(f"/v1/sessions/{sid}/close", key=self.key(sid[:8], "close"))
                    except ApiError as e:
                        if e.code != "session_closed":
                            raise
                    self.open_session = None
                view = self.api.get(f"/v1/suite-runs/{run_id}")
                done = next((x for x in view.get("cells") or [] if x.get("cell") == c["cell"]), {})
                if view.get("busy") and done.get("status") == "open":
                    # Another request holds the run for a moment (the run
                    # page, or the server's own job): ask again shortly.
                    self.api.sleep(2.0)
                    continue
                mine = ((done.get("results") or {}).get("agent") or {}).get("return_pct")
                self.say(f"market {c['cell'] + 1} of {n}: "
                         + (_pct(mine) if mine is not None else done.get("status", "?")))
                continue
            if any(c.get("status") == "waiting" for c in cells):
                self.say(f"stopped: {view.get('note') or 'the next market could not be opened'}")
                raise Stop(EXIT_LIMIT, "stopped")
            self.say("every market is played; working out the baselines and the verdict")
            for _ in range(60):
                view = self.api.post(f"/v1/suite-runs/{run_id}/continue")
                if view.get("status") != "running":
                    break
                if view.get("busy"):
                    self.api.sleep(3.0)
            else:
                self.say("the verdict is not ready yet; the server finishes it on its own")
                break
        return view

    def run_page(self, view: dict) -> str:
        """The run's page: a benchmark run's (/benchmarks?run=sr_...) when
        the run is one, else the suite run's own."""
        try:
            got = self.api.get(f"/v1/benchmark-runs/{view['run_id']}")
            page = (got.get("links") or {}).get("page")
            if page:
                return page
        except ApiError:
            pass
        return (view.get("links") or {}).get("page") or f"/suites/runs/{view['run_id']}"

    def suite_summary(self, view: dict, suite: dict) -> dict:
        link = f"{self.site}{self.run_page(view)}"
        verdict = view.get("verdict") or {}
        ref = verdict.get("reference") or "buy_and_hold"
        mine = next((s for s in verdict.get("strategies") or [] if s.get("strategy") == "agent"),
                    {})
        against = (mine.get("against") or {}).get(ref) or {}
        markets = []
        for c in view.get("cells") or []:
            r = c.get("results") or {}
            markets.append({"market": c.get("cell", 0) + 1, "status": c.get("status"),
                            "session_id": c.get("session_id"), "scenario": c.get("scenario"),
                            "return_pct": (r.get("agent") or {}).get("return_pct"),
                            f"{ref}_return_pct": (r.get(ref) or {}).get("return_pct")})
        out = {"tfrun": __version__, "mode": "suite", "status": view.get("status"),
               "run_id": view["run_id"], "suite": suite["name"],
               "checks": None, "checks_passed": None,
               "headline": verdict.get("headline"),
               "median_difference_points": against.get("median_difference"),
               "wins": against.get("wins"), "losses": against.get("losses"),
               "p_value": against.get("p_value"), "markets": markets, "link": link,
               "verdict": verdict or None, "orders_refused": self.refusals, "error": None}
        label = ref.replace("_", "-")
        self.result(f"suite         {suite['name']} (run {view['run_id']}, {view.get('status')})")
        if verdict:
            self.result(f"verdict       {verdict.get('headline')}")
            if against.get("median_difference") is not None:
                self.result(f"median        {against['median_difference']:+.2f} points a market "
                            f"against {label}")
        if self.args.verbose:
            for m in markets:
                self.result(f"market {m['market']:<6} {_pct(m['return_pct'])}   {label} "
                            f"{_pct(m[f'{ref}_return_pct'])}")
            for c in verdict.get("caveats") or []:
                self.result(f"  - {c}")
        self.result(f"results       {link}")
        if self.args.check:
            played = [m["session_id"] for m in markets if m.get("session_id")]
            out.update(self.checks(played))
        return out


# -- the local server for --offline ---------------------------------------------------------


# -- saving scenarios (for --scenario) ------------------------------------------------------


def env_key() -> Optional[str]:
    """The API key from TF_KEY, or else TRADEFLOOR_KEY (the name New
    simulation used to give), or None."""
    return os.environ.get("TF_KEY") or os.environ.get("TRADEFLOOR_KEY") or None


def _scenario_api(url: Optional[str], key: Optional[str]) -> Api:
    return Api((url or os.environ.get("TF_URL", DEFAULT_URL)).rstrip("/"),
               key if key is not None else env_key(),
               say=lambda text: print(f"tfrun: {text}", file=sys.stderr))


def save_scenario(document: Dict[str, Any], *, url: Optional[str] = None,
                  key: Optional[str] = None) -> Dict[str, Any]:
    """Save a scenario document on the server (TF_URL, with TF_KEY) and
    return it with its id (cs_...), to run with --scenario. A document is
    {"name", "description", "shocks": [...], "transmission": [...]}, each
    change {"target", "operation" (add, multiply or set), "value", "at" (the
    day it lands), "duration", "shape"}; GET /v1/scenarios/targets lists the
    targets and their bounds. Raises ApiError with the reason when refused."""
    if not isinstance(document, dict):
        raise ValueError("save_scenario() takes a scenario document, a dict")
    return _scenario_api(url, key).post("/v1/scenarios/custom", {"document": document},
                                        key=f"tfrun-save-{uuid.uuid4().hex}")


def scale_scenario(source: str, scale: float, *, name: Optional[str] = None,
                   url: Optional[str] = None, key: Optional[str] = None) -> Dict[str, Any]:
    """Save `source` (a packaged scenario, a crash replay or a cs_ id) with
    every change `scale` times its size (2 doubles it), and return the new
    scenario with its id. An add is multiplied, a multiply raised to the
    power, a VIX level scaled as its distance from 15."""
    body: Dict[str, Any] = {"from": source, "scale": _number(scale, "scale")}
    if name is not None:
        body["name"] = name
    return _scenario_api(url, key).post("/v1/scenarios/custom", body,
                                        key=f"tfrun-scale-{uuid.uuid4().hex}")


def _free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port


class LocalServer:
    """`python -m tradefloor_serve` on a free port: the same code the hosted
    service runs, on this machine, with nothing metered."""

    def __init__(self) -> None:
        if importlib.util.find_spec("tradefloor_serve") is None:
            raise UsageError("Offline mode isn't available yet; run without --offline to use "
                             "the hosted service.")
        self.port = _free_port()
        self.url = f"http://127.0.0.1:{self.port}"
        self.log = tempfile.TemporaryFile()
        self.proc = subprocess.Popen(
            [sys.executable, "-m", "tradefloor_serve", "--port", str(self.port), "--no-mcp",
             "--log-level", "warning"], stdout=subprocess.DEVNULL, stderr=self.log)
        deadline = time.time() + 60
        while True:
            if self.proc.poll() is not None:
                self.log.seek(0)
                text = self.log.read().decode("utf-8", "replace").strip().splitlines()
                raise UsageError("the local server did not start: "
                                 + (text[-1] if text else f"exit {self.proc.returncode}"))
            try:
                with urllib.request.urlopen(self.url + "/v1/health", timeout=2) as r:
                    if r.status == 200:
                        return
            except (urllib.error.URLError, OSError):
                pass
            if time.time() > deadline:
                self.stop()
                raise UsageError("the local server did not start within 60 seconds")
            time.sleep(0.2)

    def stop(self) -> None:
        if self.proc.poll() is None:
            self.proc.terminate()
            try:
                self.proc.wait(10)
            except subprocess.TimeoutExpired:
                self.proc.kill()
        self.log.close()


# -- the command line ----------------------------------------------------------------------


def parse_args(argv: Optional[List[str]]) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="tfrun.py", description="Run a trading bot written as one function, on one "
        "tradefloor simulation or a published suite.",
        epilog="Settings: TF_KEY (your API key), TF_URL (default https://app.tradefloor.dev).")
    p.add_argument("bot", nargs="?", help="your bot: a Python file with on_step(market) "
                   "(not needed with --strategy)")
    g = p.add_argument_group("one simulation")
    g.add_argument("--days", type=int, help="trading days to play (default 20)")
    g.add_argument("--preset", help="the market model (default: the recommended one)")
    g.add_argument("--seed", type=int, help="the market's seed (default 1)")
    g.add_argument("--companies", type=int, help="how many companies (default 20)")
    g.add_argument("--scenario", help="a packaged scenario (rate_shock), a crash replay "
                   "(replay_2008) or a saved scenario's id (cs_...)")
    g.add_argument("--scenario-file", metavar="FILE",
                   help="a scenario document in a JSON file (see save_scenario)")
    g.add_argument("--roster", help="real companies from a roster, such as nasdaq-100")
    g.add_argument("--tickers", help="real companies by ticker, such as NVDA,AAPL")
    g.add_argument("--start-prices", metavar="FILE",
                   help='starting prices you choose: a CSV of "TICKER,price" lines')
    g.add_argument("--price-mode", choices=["fair_value", "mispriced"], default="fair_value",
                   help="take those prices as fair value (default), or as a mispricing the "
                        "model pulls back (pt-v19; pt-v20 adds the gap to fair value)")
    g.add_argument("--history-days", type=int, metavar="N",
                   help="days of prices the market runs before day 0, so rules that need "
                        "past prices can trade from the start (costs N simulated days)")
    g.add_argument("--config", help="a JSON file of session settings; the flags above win")
    g.add_argument("--template", metavar="NAME",
                   help="open from a named template on a fresh seed, such as baseline or "
                        "rate_shock (GET /v1/templates); the way a trader key opens a session")
    g = p.add_argument_group("a suite")
    g.add_argument("--suite", help="a published suite, such as tf-quick-2026.2")
    g = p.add_argument_group("a shared market")
    g.add_argument("--shared", metavar="MARKET_ID",
                   help="take part in a shared market through TF_KEY's seat in it")
    g.add_argument("--join", metavar="CODE",
                   help="take a seat with a market's join code (a market made in the "
                        "Simulator); with --shared, first join with the code an invitation sent")
    g = p.add_argument_group("the Simulator")
    g.add_argument("--line", metavar="LINE_ID",
                   help="drive a line of a simulation from where it is, for --days days; the "
                        "Simulator's driver strip shows the command")
    g.add_argument("--strategy", metavar="ST[:V]",
                   help="run a saved strategy's code (version V, or the latest) instead of a "
                        "bot file")
    g.add_argument("--simulation", metavar="SIM_ID",
                   help="with --strategy: drive every line of the simulation that strategy "
                        "version trades, one after another")
    g = p.add_argument_group("time")
    g.add_argument("--every", choices=["step", "day"], default="step",
                   help="call on_step at every step (default) or once a day at the open")
    g.add_argument("--step", type=int, metavar="MINUTES",
                   help="minutes a step (default 30; a suite's own for --suite)")
    g = p.add_argument_group("running")
    g.add_argument("--resume", metavar="ID",
                   help="carry on an open session, or a suite run (sr_...)")
    g.add_argument("--keep-open", action="store_true",
                   help="leave the session open at the end (to fork or look at it)")
    g.add_argument("--keep", action="store_true",
                   help="keep the market of a run without --line (by default it is a "
                        "practice market, removed at the end)")
    g.add_argument("--offline", action="store_true",
                   help="run on this machine instead of the hosted service (not available "
                        "yet)")
    g.add_argument("--no-report", action="store_true",
                   help="skip the report card (it costs a few compute seconds)")
    g.add_argument("--json", action="store_true", help="print a summary as JSON (for CI)")
    g.add_argument("--check", action="append", metavar="EXPR",
                   help="a check on the behaviour metrics at the end, such as "
                        "peak_leverage<=1.5 or max_drawdown<=20%%; repeat it or comma-separate "
                        "several; exits 4 if any fails")
    g.add_argument("-v", "--verbose", action="store_true",
                   help="a line a day, and the caveats at the end")
    a = p.parse_args(argv)
    if a.bot and a.strategy:
        p.error("give a bot file or --strategy, not both")
    if not a.bot and not a.strategy:
        p.error("give your bot's file, such as my_bot.py, or --strategy st_...:V")
    if a.strategy:
        try:
            parse_strategy(a.strategy)
        except UsageError as e:
            p.error(str(e))
    if a.simulation and not a.strategy:
        p.error("--simulation drives the lines a saved strategy trades: give --strategy too")
    if a.line and a.simulation:
        p.error("--line drives one line and --simulation every line of a strategy; use one")
    market_flags = [f for f in ("preset", "seed", "companies", "scenario", "scenario_file",
                                "roster", "tickers", "start_prices", "history_days", "config")
                    if getattr(a, f) is not None]
    if a.suite and (market_flags or a.days is not None):
        flags = ", ".join("--" + f.replace("_", "-")
                          for f in market_flags + (["days"] if a.days else []))
        p.error(f"a suite fixes its own markets, so {flags} cannot be used with --suite")
    if a.resume and market_flags:
        p.error("--resume carries on an existing session, whose market is already set")
    if a.scenario and a.scenario_file:
        p.error("--scenario and --scenario-file are two ways to say the same thing; use one")
    run_id = bool(a.resume and a.resume.startswith("sr_"))
    if a.suite and a.resume and not run_id:
        p.error("--resume with --suite takes the suite run's id (sr_...)")
    a.mode = "suite" if a.suite or run_id else "simulation"
    if a.shared or a.join:
        flag = "--shared" if a.shared else "--join"
        if market_flags or a.suite or a.resume or a.template or a.offline:
            p.error(f"{flag} plays in a market its host has set up, so it cannot be used with "
                    "the market flags, --suite, --resume, --template or --offline")
        a.mode = "shared"
    if a.mode == "suite" and a.offline:
        p.error("suites run on the hosted service; --offline plays one simulation")
    if a.line or a.simulation:
        flag = "--line" if a.line else "--simulation"
        if market_flags or a.suite or a.resume or a.template or a.shared or a.join or a.offline:
            p.error(f"{flag} carries on a line of the Simulator, whose market is already set, so "
                    "it cannot be used with the market flags, --suite, --resume, --template, "
                    "--shared, --join or --offline")
        a.mode = "line" if a.line else "lines"
    if a.template and (market_flags or a.suite):
        p.error("--template fixes the market, so it cannot be used with --suite or the market "
                "flags")
    checks = [c.strip() for x in (a.check or []) for c in x.split(",") if c.strip()]
    for c in checks:
        m = _CHECK.match(c)
        if not m:
            p.error(f"--check {c!r} is not metric<op>number, such as peak_leverage<=1.5")
        if m.group(1).lower() not in CHECK_METRICS:
            p.error(f"--check {c!r}: unknown metric {m.group(1)!r}; the metrics are "
                    + ", ".join(CHECK_METRICS))
    a.check = checks
    if a.days is None:
        a.days = 20
    if a.days < 1:
        p.error("--days must be at least 1")
    if a.history_days is not None and a.history_days < 0:
        p.error("--history-days must be 0 or more")
    return a


def _is_local(url: str) -> bool:
    host = urllib.parse.urlparse(url).hostname or ""
    return host in ("localhost", "127.0.0.1", "::1")


def practice(args: Any) -> bool:
    """A run without --line, --simulation, --suite or --resume that keeps
    nothing: the practice market of Check it (`python tfrun.py my_bot.py
    --days 5`)."""
    return not (args.resume or args.keep_open or getattr(args, "keep", False) or args.template
                or args.check or args.offline or getattr(args, "config", None))


def main(argv: Optional[List[str]] = None, *, out: Any = None, err: Any = None,
         sleep: Callable[[float], None] = time.sleep) -> int:
    """Run tfrun; the exit code."""
    out = out or sys.stdout
    err = err or sys.stderr
    # `from tfrun import buy` in the bot finds this module when run as a script.
    sys.modules.setdefault("tfrun", sys.modules[__name__])
    args = parse_args(argv)
    site = os.environ.get("TF_URL", DEFAULT_URL).rstrip("/")
    key = env_key()
    server: Optional[LocalServer] = None
    runner: Optional[Runner] = None

    def fail(text: str) -> None:
        print(text, file=err, flush=True)

    quiet = contextlib.redirect_stdout(err) if args.json else contextlib.nullcontext()
    with quiet:
        try:
            if not args.offline and not key and not _is_local(site):
                fail(f"Set TF_KEY to an API key from {site}/keys, or run with --offline.")
                return EXIT_USAGE
            if args.offline:
                server = LocalServer()
                site, key = server.url, None
            api = Api(site, key, say=lambda t: None, sleep=sleep)
            bot = strategy_bot(api, args.strategy) if args.strategy else Bot(args.bot)
            api.driver = f"tfrun {__version__}; " + (f"strategy={bot.name}" if args.strategy
                                                    else f"bot={bot.name}")
            runner = Runner(api, bot, args, site=site, offline=args.offline, out=out, err=err)
            api.say = runner.say
            check_key = os.environ.get("TF_CHECK_KEY") or None
            if check_key and not args.offline:
                runner.check_api = Api(site, check_key, say=runner.say, sleep=sleep)
            summary = (runner.suite() if args.mode == "suite" else
                       runner.shared() if args.mode == "shared" else
                       runner.line() if args.mode == "line" else
                       runner.lines() if args.mode == "lines" else runner.simulate())
            if args.mode == "simulation" and practice(args) and summary.get("session_id"):
                # Check it's practice run (Handover 2, 1c): a throwaway market,
                # not saved and not one of your simulations. Its days are used.
                try:
                    api.delete(f"/v1/sessions/{summary['session_id']}",
                               key=runner.key(summary["session_id"][:8], "practice"))
                    summary["practice"] = True
                    summary["link"] = None
                    runner.result("practice      a throwaway market: it is not kept. Run with "
                                  "--keep to keep it")
                except ApiError as e:
                    runner.say(f"could not remove the practice market: {e}")
            if runner.error_days:
                days = ", ".join(str(d) for d in runner.error_days[:10])
                runner.say(f"{bot.name} raised on {len(runner.error_days)} "
                           f"day{'s' if len(runner.error_days) != 1 else ''} ({days}); those days "
                           "played with no orders from it. Fix it and run the same command to "
                           "go on from here.")
                summary["error_days"] = list(runner.error_days)
            if args.json:
                print(json.dumps(summary, indent=2), file=out)
            if summary.get("checks_passed") is False:
                return EXIT_CHECK
            return EXIT_ERROR if runner.error_days else EXIT_OK
        except UsageError as e:
            fail(str(e))
            return EXIT_USAGE
        except BotError as e:
            fail(user_traceback(e.exc).rstrip())
            return _stopped(runner, args, err, out, EXIT_ERROR, "error",
                            {"code": "bot_error", "message": f"{e.where}: {e.exc!r}"},
                            f"{e.where} raised")
        except Stop as e:
            return _stopped(runner, args, err, out, e.exit_code, e.status, None, None)
        except ApiError as e:
            body = {"code": e.code, "message": e.message, "hint": e.hint}
            if e.code == "quota_exceeded":
                if runner is not None:
                    runner.explain_limit(e)
                return _stopped(runner, args, err, out, EXIT_LIMIT, "stopped", body, None)
            fail(f"stopped: {e.code}: {e.message}" + (f"\n{e.hint}" if e.hint else ""))
            return _stopped(runner, args, err, out, EXIT_ERROR, "error", body, None)
        except KeyboardInterrupt:
            return _stopped(runner, args, err, out, EXIT_INTERRUPTED, "interrupted", None,
                            "interrupted")
        finally:
            if server is not None:
                server.stop()


def _stopped(runner: Optional[Runner], args: argparse.Namespace, err: Any, out: Any, code: int,
             status: str, error: Optional[dict], why: Optional[str]) -> int:
    """Say where the run stopped and how to carry on; the exit code."""
    where: Dict[str, Any] = {"tfrun": __version__, "status": status, "error": error}
    if runner is not None and runner.open_session and getattr(args, "mode", None) == \
            "simulation" and practice(args):
        # A practice run that stopped early is thrown away too: it is never
        # one of your simulations, so there is nothing to carry on.
        sid = runner.open_session
        try:
            runner.api.delete(f"/v1/sessions/{sid}", key=runner.key(sid[:8], "practice"))
            print("the practice market is removed; fix the code and run the same command again",
                  file=err, flush=True)
            runner.open_session = None
            where["practice"] = True
        except (ApiError, OSError):
            pass
    if runner is not None and (runner.open_session or runner.suite_run):
        sid = runner.open_session
        lead = f"{why}; " if why else ""
        if sid:
            clock = (runner.last_obs or {}).get("clock") or {}
            at = f" at day {clock.get('day')} {clock.get('time', '')}".rstrip() if clock else ""
            print(f"{lead}session {sid} is left open{at}: {runner.link(sid)}", file=err)
            where.update(session_id=sid, day=clock.get("day"), time=clock.get("time"),
                         link=runner.link(sid))
        elif runner.suite_run:
            print(f"{lead}suite run {runner.suite_run} is saved: "
                  f"{runner.site}/suites/runs/{runner.suite_run}", file=err)
        if runner.suite_run:
            where["suite_run"] = runner.suite_run
        resume = runner.suite_run or sid
        again = "after a fix" if code == EXIT_ERROR else "later"
        days = "" if runner.suite_run else f" --days {args.days}"
        flag = " --offline" if args.offline else ""
        bot = args.bot or f"--strategy {args.strategy}"
        if args.line or args.simulation:
            what = f"--line {sid}" if sid else f"--simulation {args.simulation}"
            print(f"carry on {again}: python tfrun.py {bot} {what} --days {args.days}",
                  file=err, flush=True)
        else:
            print(f"carry on {again}: python tfrun.py {bot}{flag} --resume {resume}{days}",
                  file=err, flush=True)
    if args.json:
        print(json.dumps(where, indent=2), file=out)
    return code


if __name__ == "__main__":
    sys.exit(main())