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:
- Script header — Imports, configuration, and API helper functions
- Strategy parameters — Template-specific variables read from environment variables
- Backtest setup — Session initialization and the
get_next_bar()function - Strategy logic — Your main trading loop (this is the part you write)
- 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 callerplace_order(account_id, contract_id, side, size, order_type=2, limit_price=None, stop_price=None)— Place a tradeget_positions(account_id)— Get open positionsclose_all_positions(account_id, contract_id)— Flatten all positionsget_account(account_id=None)— Get account inforead_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 barSection 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 variablescloses = []position = None # "LONG", "SHORT", or Nonetrade_count = 0wins = 0entry_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 thetry: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_countandwinsvariables must be defined and updated in your strategy logic — the cleanup references them. - If your strategy has a bug that prevents
trade_countfrom being defined, the cleanup will throw aNameError. 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 python3import 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 = Nonetrade_count = 0wins = 0entry_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) ...