Python API Reference
This page documents every function available to your Python strategy code in the Algo Playground. These functions are defined in the auto-generated script header and are available in all templates and custom strategies.
api()
The generic API caller. All other functions use this internally. You can call any TopStep API endpoint directly.
def api(path, body=None)Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
path | str | Yes | API endpoint path (e.g., "Order/place", "Account/search") |
body | dict or None | No | Request body as a Python dictionary. Automatically serialized to JSON. |
Returns: A Python dictionary parsed from the JSON response.
Error handling: On HTTP errors, returns {"success": False, "errorCode": <http_status>, "errorMessage": "<body>"}. On connection errors, returns {"success": False, "errorCode": -1, "errorMessage": "<error>"}. Errors are also printed to stderr with the [ERROR] prefix.
Examples:
# Search for active accountsresult = api("Account/search", {"onlyActiveAccounts": True})accounts = result.get("accounts", [])
# Get trade historytrades = api("Trade/search", { "accountId": ACCOUNT_ID, "startDate": "2025-01-15"})
# End a replay sessionresult = api("Replay/end", {"accountId": ACCOUNT_ID})place_order()
Places a trading order. Supports market, limit, and stop orders.
def place_order(account_id, contract_id, side, size, order_type=2, limit_price=None, stop_price=None)Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
account_id | int | Yes | — | The account ID (use ACCOUNT_ID from session setup) |
contract_id | str | Yes | — | The contract ID (use CONTRACT_ID from session setup) |
side | int | Yes | — | 0 = Buy, 1 = Sell |
size | int | Yes | — | Number of contracts |
order_type | int | No | 2 | 1 = Limit, 2 = Market, 4 = Stop |
limit_price | float or None | No | None | Required for limit orders (type 1) |
stop_price | float or None | No | None | Required for stop orders (type 4) |
Returns: The API response dictionary from the Order/place endpoint.
Examples:
# Market buy 1 contractplace_order(ACCOUNT_ID, CONTRACT_ID, 0, 1)
# Market sell 2 contracts (e.g., close long + open short)place_order(ACCOUNT_ID, CONTRACT_ID, 1, 2)
# Limit buy at 21500.00place_order(ACCOUNT_ID, CONTRACT_ID, 0, 1, order_type=1, limit_price=21500.00)
# Stop sell at 21480.00 (stop-loss)place_order(ACCOUNT_ID, CONTRACT_ID, 1, 1, order_type=4, stop_price=21480.00)Side values
| Value | Meaning |
|---|---|
0 | Buy (go long or close short) |
1 | Sell (go short or close long) |
Order type values
| Value | Meaning | Required price params |
|---|---|---|
1 | Limit order | limit_price |
2 | Market order | None |
4 | Stop order | stop_price |
get_positions()
Returns a list of currently open positions for the account.
def get_positions(account_id)Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
account_id | int | Yes | The account ID |
Returns: A list of position dictionaries. Each position has:
| Field | Type | Description |
|---|---|---|
type | int | 1 = Long, 2 = Short |
size | int | Number of contracts |
averagePrice | float | Average entry price |
contractId | str | Contract identifier |
Returns an empty list [] if there are no open positions.
Example:
positions = get_positions(ACCOUNT_ID)for pos in positions: direction = "LONG" if pos["type"] == 1 else "SHORT" print(f"[INFO] Position: {direction} {pos['size']} @ {pos['averagePrice']}")
# Check if currently in a positionif len(get_positions(ACCOUNT_ID)) > 0: print("[INFO] Have open positions")close_all_positions()
Closes all open positions for the account by placing opposite-side market orders. Also known as “flattening.”
def close_all_positions(account_id, contract_id)Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
account_id | int | Yes | The account ID |
contract_id | str | Yes | The contract ID |
Returns: Nothing (implicitly returns None).
Behavior: Queries all open positions, then for each position places a market order on the opposite side with the same size. If there are no open positions, does nothing.
Example:
# Flatten before end of sessionclose_all_positions(ACCOUNT_ID, CONTRACT_ID)print("[INFO] All positions closed")
# Use as an emergency exitif some_risk_condition: close_all_positions(ACCOUNT_ID, CONTRACT_ID) position = None print("[INFO] Emergency flatten — risk limit hit")get_account()
Returns account information including balance and equity.
def get_account(account_id=None)Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
account_id | int or None | No | None | Specific account ID to look up. If None, returns the first active account. |
Returns: An account dictionary with fields including:
| Field | Type | Description |
|---|---|---|
id | int | Account ID |
balance | float | Current account balance |
name | str | Account name |
Returns None if no matching account is found.
Example:
# Get current account infoaccount = get_account(ACCOUNT_ID)if account: balance = account.get("balance", 0) print(f"[INFO] Current balance: ${balance:,.2f}")
# Check balance before placing a tradeaccount = get_account(ACCOUNT_ID)if account and account.get("balance", 0) < 48000: print("[INFO] Balance below $48,000 — stopping strategy") breakread_speed()
Reads the current playback speed from the Playground UI. This allows the user to change the backtest speed while it is running.
def read_speed()Parameters: None.
Returns: The current STEP_DELAY value (a float representing the delay in seconds between bars).
Behavior: Reads the speed value from a file on disk (specified by the SPEED_FILE environment variable). Updates the global STEP_DELAY variable. If the file does not exist or cannot be read, the current STEP_DELAY is left unchanged.
Example:
# Standard usage — call on every barfor i in range(MAX_BARS): bar = get_next_bar() if bar is None: break
read_speed() # Update STEP_DELAY from UI if STEP_DELAY > 0: time.sleep(STEP_DELAY) # Pause between bars
# ... strategy logic ...get_next_bar()
Returns the next OHLCV bar from the replay session. This is how your strategy receives market data.
def get_next_bar()Parameters: None.
Returns: A bar dictionary (see Bar Data Format), or None if there are no more bars available or the end date has been reached.
Behavior: Calls the Replay/step API endpoint to advance the replay by one bar. If the end date is set and the bar’s timestamp exceeds it, returns None.
Example:
bar = get_next_bar()if bar is None: print("[INFO] No more bars available") break
# Access bar fieldstimestamp = bar["t"] # "2025-01-15T14:30:00Z"open_price = bar["o"] # 21500.25high_price = bar["h"] # 21505.50low_price = bar["l"] # 21498.75close_price = bar["c"] # 21503.00volume = bar["v"] # 150Function reference summary
| Function | Purpose | Returns |
|---|---|---|
api(path, body) | Call any API endpoint | Response dict |
place_order(acct, contract, side, size, ...) | Place a trade | Response dict |
get_positions(acct) | Get open positions | List of position dicts |
close_all_positions(acct, contract) | Flatten all positions | None |
get_account(acct) | Get account info | Account dict or None |
read_speed() | Read UI speed setting | Current STEP_DELAY float |
get_next_bar() | Get next market data bar | Bar dict or None |
Global variables available
In addition to the functions above, your strategy has access to these global variables set up by the auto-generated sections:
| Variable | Type | Description |
|---|---|---|
API_URL | str | Base URL for API calls |
TOKEN | str | Authentication token |
INSTRUMENT | str | Instrument symbol (e.g., "NQ") |
START_DATE | str | Backtest start date |
END_DATE | str | Backtest end date |
TIMEFRAME | str | Bar timeframe (e.g., "1m") |
STEP_DELAY | float | Current delay between bars (updated by read_speed()) |
ACCOUNT_ID | int | Active account ID |
CONTRACT_ID | str | Active contract ID |
TOTAL_BARS | int | Total bars in the session |
MAX_BARS | int | Maximum bars to process (set by your strategy) |