From TestMax to Live Trading

You have built and backtested a strategy in TestMax. Now it is time to run it on real market data through the TopStep ProjectX API. The good news: most of your code stays the same. The core API functions, order types, and strategy logic are identical. The differences are in how you connect and how you receive data.

Danger

Live trading involves real money. Before connecting to a funded account, test thoroughly on a TopStep practice account. Verify that your strategy handles edge cases: disconnections, partial fills, fast markets, and gap opens.

What Changes

Component	TestMax	Live (TopStep ProjectX)
API URL	http://localhost:8087/api/topstep-sim	https://api.topstepx.com/api
Data Feed	get_next_bar() (pull, one bar at a time)	SignalR subscription (push, real-time)
Speed Control	read_speed(), time.sleep(STEP_DELAY)	Remove — real-time data arrives when it arrives
Session Setup	Replay/start	Account/search (account already exists)
Authentication	TestMax token (auto-set)	TopStep API key + session token
Bar Formation	Pre-built historical bars	You aggregate ticks or subscribe to bar events

What Stays the Same

• place_order() — same parameters, same API endpoint
• get_positions() — same API endpoint
• close_all_positions() — same logic
• get_account() — same API endpoint
• All your strategy logic (indicators, signals, risk management)
• All bracket order functionality (SL/TP)

---

Step-by-Step Migration

Step 1: Change the API URL

# TestMax (backtesting)
API_URL = "http://localhost:8087/api/topstep-sim"

# TopStep ProjectX (live)
API_URL = "https://api.topstepx.com/api"

That single line change routes all your API calls to the real trading engine.

Step 2: Update Authentication

In TestMax, the token is auto-provided. For TopStep, you need to authenticate:

import os, json, urllib.request

API_URL = "https://api.topstepx.com/api"
API_KEY = os.environ.get("TOPSTEP_API_KEY", "")
USERNAME = os.environ.get("TOPSTEP_USERNAME", "")
PASSWORD = os.environ.get("TOPSTEP_PASSWORD", "")

def authenticate():
    """Get a session token from TopStep."""
    body = json.dumps({
        "apiKey": API_KEY,
        "username": USERNAME,
        "password": PASSWORD,
    }).encode()

    req = urllib.request.Request(
        f"{API_URL}/Auth/login",
        data=body,
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as res:
        result = json.loads(res.read())
        return result.get("token", "")

TOKEN = authenticate()
print(f"[INFO] Authenticated: token={TOKEN[:20]}...")

Step 3: Replace get_next_bar() with Real-Time Data

This is the biggest change. In TestMax, you pull bars one at a time. In live trading, market data arrives in real time via SignalR (a WebSocket-based protocol).

Option A: SignalR Subscription

The proper approach — subscribe to real-time bar updates:

# Requires: pip install signalrcore
from signalrcore.hub_connection_builder import HubConnectionBuilder

bars_queue = []

def on_bar(bar_data):
    """Called by SignalR when a new bar completes."""
    bars_queue.append({
        "t": bar_data["timestamp"],
        "o": bar_data["open"],
        "h": bar_data["high"],
        "l": bar_data["low"],
        "c": bar_data["close"],
        "v": bar_data["volume"],
    })

# Connect to TopStep SignalR hub
hub = HubConnectionBuilder() \
    .with_url(f"https://rtc.topstepx.com/hubs/market",
              options={"access_token_factory": lambda: TOKEN}) \
    .build()

hub.on("BarUpdate", on_bar)
hub.start()

# Subscribe to the instrument
hub.send("SubscribeBars", [CONTRACT_ID, "1m"])

# Your main loop now waits for bars instead of pulling them:
while True:
    if bars_queue:
        bar = bars_queue.pop(0)
        # ... your strategy logic here ...
    else:
        time.sleep(0.1)  # Wait for next bar

Option B: Polling (Simpler)

If you prefer not to use SignalR, you can poll for recent bars:

import time

last_bar_time = ""

def get_latest_bar(contract_id):
    """Poll the API for the most recent bar."""
    result = api("MarketData/bars", {
        "contractId": contract_id,
        "timeframe": "1m",
        "count": 1,
    })
    bars = result.get("bars", [])
    return bars[0] if bars else None

while True:
    bar = get_latest_bar(CONTRACT_ID)
    if bar and bar["t"] != last_bar_time:
        last_bar_time = bar["t"]
        # ... your strategy logic here ...
    time.sleep(5)  # Poll every 5 seconds

Caution

Polling is simpler but introduces up to 5 seconds of latency. For scalping strategies on 1s or 1m bars, use SignalR. For 5m+ strategies, polling is acceptable.

Step 4: Remove Speed Controls

Delete or disable the TestMax-specific speed controls:

# Remove these lines (TestMax-only):
# read_speed()
# if STEP_DELAY > 0:
#     time.sleep(STEP_DELAY)

In live trading, bars arrive at the speed of the real market. There is no speed slider.

Step 5: Get Your Live Account

Instead of creating a replay session, find your existing TopStep account:

def get_live_account():
    """Find your active TopStep trading account."""
    result = api("Account/search", {"onlyActiveAccounts": True})
    accounts = result.get("accounts", [])
    if not accounts:
        print("[ERROR] No active accounts found")
        return None
    # Use the first active account
    account = accounts[0]
    print(f"[INFO] Account: {account['id']} | Balance: ${account['balance']:,.2f}")
    return account

account = get_live_account()
ACCOUNT_ID = account["id"]

Step 6: Update the Main Loop Structure

Here is how your main loop changes:

# TestMax version:
for i in range(TOTAL_BARS):
    bar = get_next_bar()
    if bar is None:
        break
    read_speed()
    if STEP_DELAY > 0:
        time.sleep(STEP_DELAY)
    # ... strategy logic ...

# Live version:
import signal
running = True
def handle_shutdown(sig, frame):
    global running
    running = False
    print("[INFO] Shutting down...")
signal.signal(signal.SIGINT, handle_shutdown)

while running:
    if bars_queue:
        bar = bars_queue.pop(0)
        # ... exact same strategy logic ...
    else:
        time.sleep(0.1)

# Cleanup
close_all_positions(ACCOUNT_ID, CONTRACT_ID)

---

Full Migration Example

Here is an EMA crossover strategy showing both versions side-by-side:

TestMax Version

# Strategy logic is in the main loop
for i in range(TOTAL_BARS):
    bar = get_next_bar()
    if bar is None:
        break
    read_speed()
    if STEP_DELAY > 0:
        time.sleep(STEP_DELAY)

    price = bar["c"]
    fast_val = fast_ema.update(price)
    slow_val = slow_ema.update(price)

    if fast_ema.value and slow_ema.value:
        if prev_fast <= prev_slow and fast_val > slow_val:
            place_order(ACCOUNT_ID, CONTRACT_ID, 0, 1,
                        sl_ticks=SL_TICKS, tp_ticks=TP_TICKS)
        # ... rest of logic ...

    prev_fast = fast_val
    prev_slow = slow_val

Live TopStep Version

# Strategy logic in an event-driven loop
while running:
    if not bars_queue:
        time.sleep(0.1)
        continue

    bar = bars_queue.pop(0)
    price = bar["c"]
    fast_val = fast_ema.update(price)
    slow_val = slow_ema.update(price)

    if fast_ema.value and slow_ema.value:
        if prev_fast <= prev_slow and fast_val > slow_val:
            place_order(ACCOUNT_ID, CONTRACT_ID, 0, 1,
                        sl_ticks=SL_TICKS, tp_ticks=TP_TICKS)
        # ... exact same logic ...

    prev_fast = fast_val
    prev_slow = slow_val

Notice: the strategy logic inside the loop is identical. The only difference is how bars arrive (pull vs. push).

---

Testing Before Going Live

1. Test on TestMax first — Backtest extensively across multiple date ranges and market conditions.
2. Validate on TopStep practice account — Connect your live code to a practice account and run it during market hours. Verify orders execute correctly, SL/TP brackets trigger, P&L tracking matches expectations, and no duplicate orders on reconnection.
3. Paper trade for at least 1 week — Run the strategy live but monitor it closely. Check every trade against your backtested expectations.
4. Go live with minimum size — Start with 1 contract. Only scale up after confirming consistent performance.

---

Common Pitfalls

1. Reconnection Handling

WebSocket connections drop. Your strategy must handle this gracefully:

def on_disconnect():
    print("[WARN] Disconnected from SignalR, reconnecting...")
    # Check for open positions before reconnecting
    positions = get_positions(ACCOUNT_ID)
    if positions:
        print(f"[WARN] Open positions found: {len(positions)}")
    hub.start()  # Reconnect

hub.on_close(on_disconnect)

2. Duplicate Orders

If your strategy disconnects and reconnects mid-bar, it might re-process a bar that already triggered an order. Guard against this:

last_order_time = ""

def safe_place_order(side, size, **kwargs):
    global last_order_time
    current_time = bar["t"]

    # Prevent duplicate orders within the same bar
    if current_time == last_order_time:
        print(f"[WARN] Duplicate order prevented at {current_time}")
        return None

    result = place_order(ACCOUNT_ID, CONTRACT_ID, side, size, **kwargs)
    if result.get("orderId"):
        last_order_time = current_time
    return result

3. Market Hours

Unlike backtesting, live markets have specific trading hours. Futures markets have a daily maintenance break (typically 5:00-6:00 PM ET). Your strategy must handle:

def is_market_open(bar_time):
    """Check if futures market is open (simplified)."""
    hour = int(bar_time[11:13])
    # CME futures maintenance break: ~22:00-23:00 UTC (varies)
    if 22 <= hour <= 23:
        return False
    return True

4. Slippage

In backtesting, orders fill at the exact price. In live markets, there is slippage — especially during fast-moving markets. Build in a slippage buffer:

# Add a tick or two of buffer to your stop loss
SLIPPAGE_BUFFER = 2  # ticks
adjusted_sl = SL_TICKS + SLIPPAGE_BUFFER

5. Position State on Startup

When your live strategy starts, it might inherit positions from a previous session. Always check:

# On startup:
positions = get_positions(ACCOUNT_ID)
if positions:
    print(f"[WARN] Found {len(positions)} existing positions")
    # Option 1: Close them
    close_all_positions(ACCOUNT_ID, CONTRACT_ID)
    # Option 2: Track them
    # position = "LONG" if positions[0]["type"] == 1 else "SHORT"

---

Migration Checklist

Before running live, verify each item:

• API URL changed to https://api.topstepx.com/api
• Authentication working (API key + login)
• get_next_bar() replaced with SignalR or polling
• read_speed() and time.sleep(STEP_DELAY) removed
• Account discovery working (Account/search)
• Graceful shutdown handler (SIGINT)
• Reconnection logic for WebSocket drops
• Duplicate order prevention
• Market hours filtering
• Tested on practice account for at least 1 week
• Starting with minimum position size (1 contract)
• Monitoring dashboard set up (log file or alerts)

Tip

Set up logging to a file, not just stdout. When your strategy runs for hours or days, you need to be able to review what happened:

import logging
logging.basicConfig(
    filename="strategy.log",
    level=logging.INFO,
    format="%(asctime)s %(message)s"
)
logging.info("[BUY] @ 21500.25 | SL: 21495.00 TP: 21510.00")

The Path Forward

1. TestMax backtesting — develop and validate your strategy idea
2. TestMax optimization — find robust parameters across multiple periods
3. TopStep practice account — verify live connectivity and execution
4. TopStep funded account — trade with real capital at minimum size
5. Scale up — increase position size as confidence grows

The code changes are minimal. The real work is in ensuring your strategy is robust enough to handle the unpredictability of live markets.