Skip to content
Back to App

Strategy Structure

Every TestMax strategy — whether built from a template or written from scratch — follows the same five-section structure. Understanding these sections is essential for writing and debugging custom strategies.

The five sections

A complete strategy script is assembled from these parts:

  1. Script header — Imports, configuration, and API helper functions
  2. Strategy parameters — Template-specific variables read from environment variables
  3. Backtest setup — Session initialization and the get_next_bar() function
  4. Strategy logic — Your main trading loop (this is the part you write)
  5. Cleanup — Position flattening, session end, and results output

When you select a template and click Run, TestMax concatenates these sections into a single Python script and executes it on the server. When you write a custom strategy, you only need to write section 4 — the other sections are auto-generated.

Section 1: Script header

The header is identical for all strategies. It handles imports, reads global configuration from environment variables, and defines all the API helper functions.

#!/usr/bin/env python3
"""
TestMax Algo Playground — Portable Trading Script
=================================================
This script is compatible with both:
- TestMax Simulator API (for backtesting)
- Real TopStep ProjectX Gateway API (for live trading)
"""
import os, sys, json, time, urllib.request, urllib.error
# ─── CONFIG ──────────────────────────────────────────────
API_URL = os.environ.get("TESTMAX_API_URL", "http://localhost:8087/api/topstep-sim")
TOKEN = os.environ.get("TESTMAX_TOKEN", "")
INSTRUMENT = os.environ.get("INSTRUMENT", "NQ")
START_DATE = os.environ.get("START_DATE", "2025-01-15")
TIMEFRAME = os.environ.get("TIMEFRAME", "1s")
STEP_DELAY = float(os.environ.get("STEP_DELAY", "0.02"))
SPEED_FILE = os.environ.get("SPEED_FILE", "")

The header also defines these functions (see the API Reference for full details):

  • api(path, body=None) — Generic API caller
  • place_order(account_id, contract_id, side, size, order_type=2, limit_price=None, stop_price=None) — Place a trade
  • get_positions(account_id) — Get open positions
  • close_all_positions(account_id, contract_id) — Flatten all positions
  • get_account(account_id=None) — Get account info
  • read_speed() — Read current playback speed from UI

Section 2: Strategy parameters

Each template reads its specific parameters from environment variables. These are set by the Playground UI based on the parameter values you enter in the sidebar.

For the SMA Crossover template:

FAST_PERIOD = int(os.environ.get("FAST_PERIOD", "10"))
SLOW_PERIOD = int(os.environ.get("SLOW_PERIOD", "30"))
MAX_BARS = int(os.environ.get("MAX_BARS", "0")) or int(os.environ.get("TOTAL_BARS", "999999"))

For a custom strategy, you can define your own parameters by reading environment variables with sensible defaults:

MY_THRESHOLD = float(os.environ.get("MY_THRESHOLD", "0.5"))
LOOKBACK = int(os.environ.get("LOOKBACK", "20"))
MAX_BARS = int(os.environ.get("MAX_BARS", "0")) or int(os.environ.get("TOTAL_BARS", "999999"))

The MAX_BARS pattern

All templates use this pattern:

MAX_BARS = int(os.environ.get("MAX_BARS", "0")) or int(os.environ.get("TOTAL_BARS", "999999"))

This means: if the user set MAX_BARS to a non-zero value, use that. Otherwise, use the total number of bars in the session (set by the Playground), or fall back to 999999 (effectively unlimited).

Section 3: Backtest setup

The setup section handles session initialization and defines the get_next_bar() function. It works in two modes:

Playground mode (normal usage): When running inside the Playground, the session is pre-created by the server. The ACCOUNT_ID, CONTRACT_ID, and TOTAL_BARS environment variables are already set.

Standalone mode: If those environment variables are not set, the script creates its own session via the API. This allows you to run the script outside of TestMax (useful for advanced users or live trading).

# ─── BACKTEST SESSION SETUP ─────────────────────────────
END_DATE = os.environ.get("END_DATE", "")
_env_account = os.environ.get("ACCOUNT_ID", "")
_env_contract = os.environ.get("CONTRACT_ID", "")
_env_total = os.environ.get("TOTAL_BARS", "")
if _env_account and _env_contract:
# Pre-created session (running inside playground)
ACCOUNT_ID = int(_env_account)
CONTRACT_ID = _env_contract
TOTAL_BARS = int(_env_total) if _env_total else 3600
print(f"[INFO] Using pre-created session: {INSTRUMENT} ({CONTRACT_ID})")
else:
# Standalone mode — create session via API
print("[INFO] Starting backtest session via API...")
session = api("Replay/start", {
"instrumentSymbol": INSTRUMENT,
"startDate": START_DATE,
"timeframe": TIMEFRAME,
})
if not session.get("success", True) or not session.get("accountId"):
print(f"[ERROR] Failed to start session: {session}")
sys.exit(1)
ACCOUNT_ID = session["accountId"]
CONTRACT_ID = session["instrument"]["contractId"]
TOTAL_BARS = session.get("totalBars", 3600)
print(f"[INFO] Account: {ACCOUNT_ID}, Contract: {CONTRACT_ID}, Bars: {TOTAL_BARS}")

The setup also defines get_next_bar():

def get_next_bar():
"""Get next bar from replay. Replace with SignalR for live trading."""
result = api("Replay/step", {"accountId": ACCOUNT_ID, "steps": 1})
bars = result.get("bars", [])
if not bars:
return None
bar = bars[0]
# Stop if past end date
if END_DATE and bar.get("t", "") > END_DATE + "T23:59:59":
return None
return bar

Section 4: Strategy logic (what you write)

This is the core of your strategy — the part that makes trading decisions. Every strategy follows the same pattern:

# State variables
closes = []
position = None # "LONG", "SHORT", or None
trade_count = 0
wins = 0
entry_price = 0
# Optional: define helper functions (indicators, etc.)
def my_indicator(data, period):
# Your calculation here
pass
print("[INFO] Strategy: My Custom Strategy")
try:
for i in range(MAX_BARS):
# 1. Get the next bar
bar = get_next_bar()
if bar is None:
break
# 2. Update data arrays
closes.append(bar["c"])
# 3. Handle speed control
read_speed()
if STEP_DELAY > 0:
time.sleep(STEP_DELAY)
# 4. Skip if not enough data yet
if len(closes) < REQUIRED_BARS:
continue
# 5. Compute indicators
price = bar["c"]
signal = my_indicator(closes, PERIOD)
# 6. Generate trading signals
if buy_condition and position != "LONG":
# Enter long (or reverse from short)
if position == "SHORT":
place_order(ACCOUNT_ID, CONTRACT_ID, 0, 2) # size 2 to close + reverse
# Track P&L
trade_count += 1
else:
place_order(ACCOUNT_ID, CONTRACT_ID, 0, 1)
position = "LONG"
entry_price = price
trade_count += 1
print(f"[TRADE] BUY @ {price:.2f}")
elif sell_condition and position != "SHORT":
# Enter short (or reverse from long)
if position == "LONG":
place_order(ACCOUNT_ID, CONTRACT_ID, 1, 2)
trade_count += 1
else:
place_order(ACCOUNT_ID, CONTRACT_ID, 1, 1)
position = "SHORT"
entry_price = price
trade_count += 1
print(f"[TRADE] SELL @ {price:.2f}")

Key patterns

The main loop: for i in range(MAX_BARS) iterates through bars. get_next_bar() returns None when there are no more bars, which triggers the break.

Speed control: read_speed() and time.sleep(STEP_DELAY) must be called on every bar to respect the UI speed control. Without these, the backtest runs as fast as possible with no chart updates.

Position tracking: The position variable tracks whether you are long, short, or flat. Always check the current position before placing an order to avoid doubling up.

Order sizing for reversals: When flipping from long to short (or vice versa), use size 2 instead of size 1. Size 1 closes the existing position, and the additional size 1 opens the new position — a single order of size 2 accomplishes both.

Win tracking: The wins and trade_count variables are used by the cleanup section to calculate the win rate. Update them on every trade closure.

Section 5: Cleanup

The cleanup section runs inside a finally block so it executes even if the strategy throws an error. It closes all open positions, ends the replay session, and prints the final results.

# ─── CLEANUP ─────────────────────────────────────────────
finally:
# Always close positions and end session
try:
close_all_positions(ACCOUNT_ID, CONTRACT_ID)
except:
pass
try:
end = api("Replay/end", {"accountId": ACCOUNT_ID})
account = end.get("account", {})
balance = account.get("balance", 0)
pnl = balance - 50000
print("-" * 60)
print(f"[DONE] Backtest Complete!")
print(f"[DONE] Final Balance: ${balance:,.2f}")
print(f"[DONE] Net P&L: ${pnl:+,.2f}")
print(f"[DONE] Total Trades: {trade_count}")
if trade_count > 0:
print(f"[DONE] Win Rate: {wins / trade_count * 100:.1f}%")
print(f"\n[INFO] View full analytics in TestMax → Analytics page")
except Exception as e:
print(f"[ERROR] Cleanup failed: {e}")

Important notes about cleanup

  • The cleanup uses finally: which corresponds to the try: at the beginning of section 4. Your strategy logic runs inside this try block.
  • close_all_positions() is called first to flatten any open position before ending the session.
  • api("Replay/end", ...) ends the replay session and returns the final account state.
  • The trade_count and wins variables must be defined and updated in your strategy logic — the cleanup references them.
  • If your strategy has a bug that prevents trade_count from being defined, the cleanup will throw a NameError. Always define these variables at the top of your strategy section.

Full assembled example

Here is a minimal complete strategy showing all five sections together:

#!/usr/bin/env python3
import os, sys, json, time, urllib.request, urllib.error
# ─── Section 1: CONFIG + API HELPERS ─────────────────────
API_URL = os.environ.get("TESTMAX_API_URL", "...")
TOKEN = os.environ.get("TESTMAX_TOKEN", "")
# ... (api helpers defined here) ...
# ─── Section 2: PARAMETERS ──────────────────────────────
THRESHOLD = float(os.environ.get("THRESHOLD", "0.5"))
MAX_BARS = int(os.environ.get("MAX_BARS", "0")) or int(os.environ.get("TOTAL_BARS", "999999"))
# ─── Section 3: BACKTEST SETUP ──────────────────────────
# ... (session init + get_next_bar defined here) ...
# ─── Section 4: YOUR STRATEGY ───────────────────────────
closes = []
position = None
trade_count = 0
wins = 0
entry_price = 0
try:
for i in range(MAX_BARS):
bar = get_next_bar()
if bar is None:
break
closes.append(bar["c"])
read_speed()
if STEP_DELAY > 0:
time.sleep(STEP_DELAY)
# Your logic here...
# ─── Section 5: CLEANUP ────────────────────────────────
finally:
close_all_positions(ACCOUNT_ID, CONTRACT_ID)
end = api("Replay/end", {"accountId": ACCOUNT_ID})
# ... (print results) ...