AlgoTest Signals API Examples

This page provides comprehensive examples of how to use the AlgoTest Signals API to build and execute automated options trading strategies.
The examples demonstrate authentication, real-time prices handling, strategy execution from JSON payloads, and signal creation.

Authentication

The AlgoTestClient class manages session tokens and authentication headers.

import os
import requests
from dotenv import load_dotenv

class AlgoTestClient:
    def __init__(self, phone_number: str, main_url: str):
        load_dotenv()
        self.base_url = main_url
        self.phone_number = phone_number
        self.password = os.getenv("PASSWORD", "default_password")
        self.session = requests.Session()
        self.csrf_token = None
        self.jwt_token = None

        self.login()

    def login(self):
        login_payload = {
            "phoneNumber": self.phone_number,
            "password": self.password,
        }

        headers = {"Content-Type": "application/json"}
        response = self.session.post(f"{self.base_url}/login", json=login_payload, headers=headers)

        if response.status_code == 200:
            self.csrf_token = self.session.cookies.get("csrf_access_token")
            self.jwt_token = self.session.cookies.get("access_token_cookie")

            self.session.headers.update({
                "X-CSRF-TOKEN-ACCESS": self.csrf_token,
                "Authorization": self.jwt_token,
            })

            print("Login successful!")
        else:
            raise Exception(f"Login failed: {response.status_code}, {response.text}")

    def get_session(self):
        return self.session

    def get_tokens(self):
        return {
            "X-CSRF-TOKEN-ACCESS": self.csrf_token,
            "Authorization": self.jwt_token,
        }

Fetching Contracts

Before executing a strategy, fetch available contracts for the chosen underlying:

import requests
from datetime import datetime

class ContractFetcher:
    def __init__(self, token, underlying : str, prices_url : str):
        self.underlying = underlying
        self.contracts_url = f"{prices_url}/contracts?underlying={underlying}"
        self.contracts = None

        self.headers = {"Content-Type": "application/json"}
        self.headers.update(token)

        self.fetch_contracts()

    def fetch_contracts(self):
        response = requests.get(self.contracts_url, headers=self.headers)

        if response.status_code == 200:
            self.contracts = response.json()
        else:
            raise Exception(f"Failed to fetch contracts: {response.status_code}, {response.text}")

    def get_latest_expiry(self):
        opt = self.contracts[self.underlying]["OPT"]
        dates = []
        for exp_str in opt.keys():
            try:
                dates.append(datetime.fromisoformat(exp_str))
            except ValueError:
                continue

        if dates:
            latest_date = min(dates)
            latest_expiry = latest_date.date().isoformat()   
        else:
            latest_expiry = None

        return latest_expiry

WebSocket Connection for Live Data

Use the WebSocket client to subscribe to live candles:

import json
import ssl
import time
from websocket import WebSocketApp

class OptionChainWebSocketClient:
    def __init__(self, url, jwt_token, subscription_payload, on_data_callback=None):
        self.url = url
        self.jwt_token = jwt_token
        self.subscription_payload = subscription_payload
        self.ws = None
        self.on_data_callback = on_data_callback

    def on_message(self, ws, message):
        if self.on_data_callback:
            self.on_data_callback(message) 

    def on_open(self, ws):
        print("WebSocket connection opened, sending subscription…")
        ws.send(json.dumps(self.subscription_payload))
        print("Subscription sent")

    def on_close(self, ws, *args):
        print("WebSocket connection closed:")

    def on_error(self, ws, error):
        print("[ERROR]", error)

    def start(self):
        ssl_context = ssl._create_unverified_context()

        while True:
            self.ws = WebSocketApp(
                self.url,
                on_open=self.on_open,
                on_message=self.on_message,
                on_close=self.on_close,
                on_error=self.on_error,
                header=[f"Cookie: access_token_cookie={self.jwt_token}"],
            )
            print("Starting WebSocket loop")
            try:
                self.ws.run_forever(sslopt={"context": ssl_context})
            except Exception as e:
                print(f"[ERROR] Exception in run_forever: {e}")

            print("WebSocket disconnected, reconnecting in 5 seconds…")
            time.sleep(5)

Building a Trading Strategy

Strategies are defined in strategy_payload.json, making them fully configurable. This defines a weekly ATM straddle with 2 lots CE Sell + 2 lots PE Sell, each with a 50% SL.

Example: ATM Straddle Sell

{
  "access_token": "anything",
  "alert_name": "SENSEX_14:41",
  "exit_time": "2025-10-16T15:15",
  "strategy": {
    "Ticker": "SENSEX",
    "Legs": [
      {
        "PositionConfig": {
          "PositionType": "PositionType.Sell",
          "Lots": 2,
          "LegStopLoss": {
            "Type": "LegTgtSLType.Percentage",
            "Value": 50
          },
          "LegTarget": {
            "Type": "None",
            "Value": 0
          },
          "LegTrailSL": {
            "Type": "None",
            "Value": {}
          },
          "ExpiryKind": "ExpiryType.Weekly",
          "EntryType": "EntryType.EntryByStrikeType",
          "StrikeParameter": "StrikeType.ATM",
          "InstrumentKind": "LegType.CE"
        },
        "ExecutionConfig": {
          "ProductType": "ProductType.NRML"
        }
      },
      {
        "PositionConfig": {
          "PositionType": "PositionType.Sell",
          "Lots": 2,
          "LegStopLoss": {
            "Type": "LegTgtSLType.Percentage",
            "Value": 50
          },
          "LegTarget": {
            "Type": "None",
            "Value": 0
          },
          "LegTrailSL": {
            "Type": "None",
            "Value": {}
          },
          "ExpiryKind": "ExpiryType.Weekly",
          "EntryType": "EntryType.EntryByStrikeType",
          "StrikeParameter": "StrikeType.ATM",
          "InstrumentKind": "LegType.PE"
        },
        "ExecutionConfig": {
          "ProductType": "ProductType.NRML"
        }
      }
    ]
  }
}

Strategy Logic

Core strategy logic. Reads strategy payload, maintains candle store, evaluates rules, and triggers signals.

import json
import os
import time
from datetime import datetime, timedelta
from collections import deque
from signals_api import AlgoTestSignals

MAX_NUMBER_OF_CANDLES = 375

class Strategy:
    def __init__(self, underlying : str, main_url : str, access_token :str, broker_id : str, expiry : datetime, 
                 supertrend_length: int, supertrend_multiplier : float, persist_file="candle_store.txt"):
        self.candles = deque(maxlen=MAX_NUMBER_OF_CANDLES)
        self.spot_candles = deque(maxlen=MAX_NUMBER_OF_CANDLES)
        self.underlying = underlying
        self.last_timestamp = None
        self.supertrend_length = supertrend_length
        self.supertrend_multiplier = supertrend_multiplier
        self.main_url = main_url
        self.expiry = expiry
        self.persist_file = persist_file
        self.last_dump_time = time.time()
        self.tradeflag = 0
        self.current_atm = None
        self.strike_step = 100 

        self.trade = AlgoTestSignals(base_url=self.main_url, access_token=access_token, broker_id=broker_id)
        self.load_from_file()

    def load_from_file(self):
        if os.path.exists(self.persist_file):
            with open(self.persist_file, "r") as f:
                try:
                    lines = f.readlines()
                    for line in lines:
                        data = json.loads(line.strip())
                        self.candles.append(data)
                    if self.candles:
                        self.last_timestamp = datetime.fromisoformat(self.candles[-1]["timestamp"])
                    print(f"Loaded {len(self.candles)} candles from file.")
                except Exception as e:
                    print(f"Failed to load from file: {e}")

    def save_to_file(self):
        with open(self.persist_file, "w") as f:
            for candle in self.candles:
                f.write(json.dumps(candle) + "\n")
        print(f"{len(self.candles)} candles saved at {datetime.now().isoformat()}")

    def calculate_supertrend(self, period, multiplier):

        if len(self.candles) < period + 2:
            print("Not enough data to calculate Supertrend")
            return None  

        closes = [c["close"] for c in self.candles]

        spot_highs = [c["high"] for c in self.spot_candles]
        spot_lows = [c["low"] for c in self.spot_candles]
        spot_closes = [c["close"] for c in self.spot_candles]

        trs = []
        for i in range(1, len(spot_highs)):
            tr = max(spot_highs[i] - spot_lows[i],
                     abs(spot_highs[i] - spot_closes[i-1]),
                     abs(spot_lows[i] - spot_closes[i-1]))
            trs.append(tr)
        atr = sum(trs[-period:]) / period

        upperband = closes[-1] + multiplier * atr
        lowerband = closes[-1] - multiplier * atr

        if len(self.candles) < period + 3:
            final_upperband = upperband
            final_lowerband = lowerband
        else:
            prev = self.candles[-2]
            final_upperband = min(upperband, prev.get("final_upperband", upperband))
            final_lowerband = max(lowerband, prev.get("final_lowerband", lowerband))

        close = closes[-1]
        if close > final_upperband:
            supertrend = final_lowerband
            direction = "up"
        elif close < final_lowerband:
            supertrend = final_upperband
            direction = "down"
        else:
            prev = self.candles[-2]
            direction = prev.get("supertrend_direction", "up")
            supertrend = prev.get("supertrend", (final_upperband + final_lowerband) / 2)


        self.candles[-1]["atr"] = atr
        self.candles[-1]["supertrend"] = supertrend
        self.candles[-1]["supertrend_direction"] = direction
        self.candles[-1]["final_upperband"] = final_upperband
        self.candles[-1]["final_lowerband"] = final_lowerband

        return supertrend, direction

    def check_condition(self): 
        if len(self.candles) < 2:
            return  

        prev_candle = self.candles[-2]
        last_candle = self.candles[-1]

        prev_dir = prev_candle.get("supertrend_direction")
        curr_dir = last_candle.get("supertrend_direction")

        if not prev_dir or not curr_dir:
            return

        if prev_dir != curr_dir:
            print()
            print(f"[TREND CHANGE] at {last_candle['timestamp']}")
            print(f"Previous Direction: {prev_dir}, Current Direction: {curr_dir}")
            print(f"Candle: {last_candle}")
            print()

            if curr_dir == "up" and self.tradeflag == 1:
                print("Supertrend Green → Buy Straddle")
                self.trade.stop_signal(position_id=self.current_position_id)
                self.tradeflag = 0

            elif curr_dir == "down" and self.tradeflag == 0:
                print("Supertrend Red → Sell Straddle")
                self.current_position_id = self.trade.start_signal(execution_type="paper")
                self.tradeflag = 1


    def get_nearest_strike(self, price: float, step: int = 100) -> int:
        return int(round(price / step) * step)

    def format_expiry(self) -> str:
        return f"{self.expiry}T15:30:00"


    def handle_price_update(self, raw_message):
        try:
            message = json.loads(raw_message)

            cash_data = message.get("candle", {}).get(self.underlying, {}).get("CASH", {})
            if not cash_data:
                return

            cash_close = cash_data.get("close")
            cash_timestamp = cash_data.get("timestamp")

            if not cash_close or not cash_timestamp:
                return

            atm_strike = self.get_nearest_strike(cash_close, self.strike_step)
            self.current_atm = atm_strike 

            opt_data = message.get("candle", {}).get(self.underlying, {}).get("OPT", {})
            if not opt_data:
                return

            expiry_key = self.format_expiry()
            expiry_data = opt_data.get(expiry_key, {})
            strike_data = expiry_data.get(str(atm_strike) + ".0", {})

            if not strike_data:
                print(f"[INFO] ATM {atm_strike} not found in option chain")
                return

            ce = strike_data.get("CE", {})
            pe = strike_data.get("PE", {})

            print(f"[INFO] ATM Strike: {atm_strike}, CE: {ce.get('close')}, PE: {pe.get('close')}")

            if ce and pe:
                timestamp = ce.get("timestamp") or pe.get("timestamp")
                dt = datetime.fromisoformat(timestamp)

                spot_ohlc = {
                    "timestamp": cash_timestamp,
                    "open": cash_data.get("open"),
                    "high": cash_data.get("high"),
                    "low": cash_data.get("low"),
                    "close": cash_close,
                }

                combined_ohlc = {
                    "timestamp": timestamp,
                    "open": (ce["open"] + pe["open"]),
                    "high": max((ce["open"] + pe["open"]),(ce["close"] + pe["close"])),
                    "low": min((ce["open"] + pe["open"]),(ce["close"] + pe["close"])),
                    "close": (ce["close"] + pe["close"]),
                    "atm_strike": atm_strike,
                    "expiry": expiry_key,
                }

                if self.last_timestamp and (dt - self.last_timestamp) > timedelta(minutes=1):
                    print(f"[WARNING] Skipped candle. Last: {self.last_timestamp}, New: {dt}")

                self.candles.append(combined_ohlc)
                self.spot_candles.append(spot_ohlc)
                self.last_timestamp = dt

                print(f"Spot Candle -> {spot_ohlc}")
                print(f"[ATM {atm_strike}] Combined Candle -> {combined_ohlc}")

                try:
                    supertrend, direction = self.calculate_supertrend(period=self.supertrend_length, multiplier=self.supertrend_multiplier)
                    if supertrend:
                        print(f"[ATM {atm_strike}] Supertrend: {supertrend}, Direction: {direction}")
                except Exception as e:
                    print(f"[ERROR] Supertrend calculation failed: {e}")

                self.check_condition()

                if time.time() - self.last_dump_time >= 300:
                    self.save_to_file()
                    self.last_dump_time = time.time()

        except Exception as e:
            print(f"[ERROR] Failed to process message: {e}")

Creating and Sending Signals

The signals_api.py module handles trade execution:

import requests
import json
import copy
from pathlib import Path
from datetime import datetime, time

class AlgoTestSignals:
    def __init__(self, base_url: str, access_token: str, broker_id: str, strategy_file: str = "strategy_payload.json"):
        self.access_token = access_token
        self.broker_id = broker_id
        self.base_url = base_url
        self.strategy_file = Path(strategy_file)

        self.headers = {
            "Content-Type": "application/json",
        }

        self.strategy_payload = self._load_strategy_payload()


    def _load_strategy_payload(self) -> dict:
        try:
            if not self.strategy_file.exists():
                raise FileNotFoundError(f"Strategy file not found: {self.strategy_file}")

            if self.strategy_file.stat().st_size == 0:
                raise ValueError(f"Strategy file is empty: {self.strategy_file}")

            with open(self.strategy_file, "r") as f:
                data = json.load(f)

            print(f"Strategy payload loaded from {self.strategy_file}")
            return data

        except FileNotFoundError as e:
            print(str(e))
            raise
        except ValueError as e:
            print(str(e))
            raise
        except json.JSONDecodeError:
            print(f"Strategy file is not valid JSON: {self.strategy_file}")
            raise


    def update_exit_time_to_today(self, payload: dict) -> dict:
        updated_payload = payload.copy()
        exit_dt = datetime.combine(datetime.today().date(), time(15, 15, 0))
        updated_payload["exit_time"] = exit_dt.strftime("%Y-%m-%dT%H:%M:%S")
        return updated_payload


    def start_signal(self, execution_type: str = "paper") -> str:
        if execution_type == "paper":
            url = f"{self.base_url}/webhook/custom-position/execution/start/paper"
        elif execution_type == "live":
            url = f"{self.base_url}/webhook/custom-position/execution/start/live?broker_id={self.broker_id}"

        body = self.strategy_payload.copy()
        body["access_token"] = self.access_token
        body = self.update_exit_time_to_today(body)
        print(f"[AlgoTest] Starting signal with payload: {json.dumps(body, indent=2)}")

        response = requests.post(url, json=body, headers=self.headers)
        if response.status_code == 200:  
            data = response.json()
            print(f"[AlgoTest] Start signal successful: {data}")
            position_id = data.get("position_id") or data.get("id")
            if position_id is None:
                raise Exception("Position ID not returned in start_signal response")
            return position_id
        else:
            raise Exception(f"Failed to start signal: {response.status_code}, {response.text}")


    def stop_signal(self, position_id: str) -> bool:
        if position_id is None:
            raise ValueError("Position ID must be provided to stop a signal")

        url = f"{self.base_url}/webhook/custom-position/execution/square-off/{position_id}"
        body = {
            "access_token": self.access_token,
        }

        response = requests.post(url, json=body, headers=self.headers)
        if response.status_code == 200:
            print(f"[AlgoTest] Stop signal successful: {response.json()}")
            return True
        else:
            raise Exception(f"Failed to stop signal: {response.status_code}, {response.text}")

Complete Example: ATM Straddle Strategy

The main.py ties everything together:

Authenticate
Fetch contracts
Load strategy JSON
Start WebSocket
Pass prices to Strategy
Place trades via signals_api.py

Environment Setup

Create .env file:

PHONE_NUMBER=+911234567890
PASSWORD=your_password
ACCESS_TOKEN=your_access_token
BROKER_ID=your_broker_id
SUPERTREND_MULTIPLIER=your_supertrend_multiplier
SUPERTREND_LENGTH=your_supertrend_length

Install dependencies:

pip install -r requirements.txt

Disclaimer

This code is for educational purposes only. Options trading carries significant risk. Always test strategies in paper trading mode before deploying live.

Resources

Community: AlgoTest Telegram
Support: AlgoTest Email