Trade Signals API Examples

This page provides comprehensive examples of how to use the AlgoTest Trade Signals API to build and execute automated trading strategies. The examples demonstrate authentication, real-time data handling, strategy implementation, and trade signal execution.

Quick Start

Here's a minimal example to get you started with the Trade Signals API:

from algotest_login import AlgoTestClient
from trade_signals import TradeSignals
import os
from dotenv import load_dotenv

load_dotenv()

# Initialize client and authenticate
client = AlgoTestClient(
    phone_number=os.getenv("PHONE_NUMBER"),
    main_url="https://api.algotest.in"
)

# Create trade signals instance
trade = TradeSignals(
    main_url="https://api.algotest.in",
    order_url="https://orders.algotest.in",
    access_token=os.getenv("ACCESS_TOKEN"),
    broker_id=os.getenv("BROKER_ID"),
    token=client.get_tokens()
)

# Create a signal
signal_payload = {
    "signal_name": "My First Signal",
    "signal_type": "paper",
    "brokers": []
}

signal_tag = trade.create_trade_signals(signal_payload)

# Send a trade
trade_payload = "BTCUSD.P buy 10"
trade.send_trade_signals(
    tag=signal_tag,
    payload=trade_payload,
    execution_type='paper'
)

Authentication

The AlgoTestClient class handles authentication and session management:

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_tokens(self):
        return {
            "X-CSRF-TOKEN-ACCESS": self.csrf_token,
            "Authorization": self.jwt_token,
        }

Usage Example

# Initialize the client
client = AlgoTestClient(
    phone_number="+1234567890",
    main_url="https://api.algotest.in"
)

# Get authentication tokens for API calls
tokens = client.get_tokens()

Fetching Contracts

Before trading, you need to fetch available contracts for your underlying instrument:

import requests

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.contract_count = 0

        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()
            self.contract_count = len(self.contracts)
            print(f"Fetched {self.contract_count} contracts for {self.underlying}")
        else:
            raise Exception(f"Failed to fetch contracts: {response.status_code}")

Usage Example

# Fetch contracts for BTCUSD
contracts = ContractFetcher(
    token=client.get_tokens(),
    underlying="DELTA_BTCUSD",
    prices_url="https://prices.algotest.in"
)

print(f"Available contracts: {contracts.contract_count}")

WebSocket Connection for Live Data

Establish a WebSocket connection to receive real-time market data:

import json
import ssl
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):
        print(f"Received: {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(f"[ERROR] {error}")

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

        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")
        self.ws.run_forever(sslopt={"context": ssl_context})

Subscription Message Example

subscription_message = {
    "msg": {
        "type": "subscribe",
        "datatypes": ["candle"],
        "underlyings": [
            {
                "underlying": "DELTA_BTCUSD",
                "cash": False,
                "options": [],
                "futures": [],
            }
        ],
        "tokens": ["DELTA_27"],
    }
}

# Create WebSocket client
ws_client = OptionChainWebSocketClient(
    url="wss://prices.algotest.in/updates?structured=true",
    jwt_token=client.jwt_token,
    subscription_message=subscription_message,
    on_data_callback=your_callback_function
)

ws_client.start()

Building a Trading Strategy

EMA-Based Strategy Example

Here's a complete strategy implementation using Exponential Moving Average:

import json
import os
import time
from datetime import datetime, timedelta
from collections import deque
from trade_signals import TradeSignals

class Strategy:
    def __init__(self, underlying: str, main_url: str, order_url: str, 
                 access_token: str, broker_id: str, token: dict, 
                 persist_file="candle_store.txt"):

        # Strategy configuration
        self.EMA_PERIOD = 3
        self.MAX_CANDLES = 60
        self.QUANTITY = 10
        self.TRADING_SYMBOL = "BTCUSD.P"

        # Data storage
        self.candles = deque(maxlen=self.MAX_CANDLES)
        self.underlying = underlying
        self.last_timestamp = None
        self.persist_file = persist_file
        self.last_dump_time = time.time()

        # Trading state
        self.trade_flag = 0  # 0: no position, 1: long position
        self.trade_signal_tag = None
        self.open_trades = []

        # Initialize trade signals client
        self.trade = TradeSignals(
            main_url=main_url,
            order_url=order_url,
            access_token=access_token,
            broker_id=broker_id,
            token=token
        )

        self.load_from_file()

    def calculate_ema(self, period: int):
        """Calculate Exponential Moving Average"""
        if len(self.candles) < period:
            return None

        prices = [c["close"] for c in self.candles]
        multiplier = 2 / (period + 1)
        ema = prices[0]

        for price in prices[1:]:
            ema = (price - ema) * multiplier + ema

        return round(ema, 2)

    def check_entry_condition(self, candle):
        """Check if entry conditions are met"""
        return (candle["ema"] < candle["close"] and 
                self.trade_flag == 0)

    def check_exit_condition(self, candle):
        """Check if exit conditions are met"""
        return (candle["ema"] > candle["close"] and 
                self.trade_flag == 1)

    def execute_trade(self, action: str):
        """Execute a trade signal"""
        trade_payload = f"{self.TRADING_SYMBOL} {action} {self.QUANTITY}"

        # Create signal if it doesn't exist
        if not self.trade_signal_tag:
            signal_payload = {
                "signal_name": "EMA Strategy Signal",
                "signal_type": "paper",
                "brokers": []
            }

            self.trade_signal_tag = self.trade.create_trade_signals(signal_payload)

            if not self.trade_signal_tag:
                raise Exception("Failed to create signal")

        # Send trade signal
        trade_completed = self.trade.send_trade_signals(
            tag=self.trade_signal_tag,
            payload=trade_payload,
            execution_type='paper'
        )

        return trade_completed

    def handle_price_update(self, raw_message):
        """Process incoming price data"""
        try:
            message = json.loads(raw_message)
            fut_data = message.get("candle", {}).get(self.underlying, {}).get("FUT", {})

            for contract_key, candle in fut_data.items():
                if contract_key is None or contract_key == "null":
                    timestamp = candle.get("timestamp")
                    if not timestamp:
                        return

                    dt = datetime.fromisoformat(timestamp)

                    # Calculate EMA
                    ema = self.calculate_ema(self.EMA_PERIOD)

                    # Create OHLC data structure
                    ohlc = {
                        "timestamp": timestamp,
                        "open": candle["open"],
                        "high": candle["high"],
                        "low": candle["low"],
                        "close": candle["close"],
                        "ema": ema,
                    }

                    self.candles.append(ohlc)
                    self.last_timestamp = dt

                    # Check trading conditions
                    if ema:  # Only trade if EMA is available
                        if self.check_entry_condition(ohlc):
                            print(f"Entry condition met at {timestamp}")
                            if self.execute_trade("buy"):
                                self.trade_flag = 1
                                print("Long position opened")

                        elif self.check_exit_condition(ohlc):
                            print(f"Exit condition met at {timestamp}")
                            if self.execute_trade("sell"):
                                self.trade_flag = 0
                                print("Position closed")

                    # Periodic data persistence
                    if time.time() - self.last_dump_time >= 300:  # Every 5 minutes
                        self.save_to_file()
                        self.last_dump_time = time.time()

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

    def save_to_file(self):
        """Save candle data to file"""
        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")

    def load_from_file(self):
        """Load candle data from file"""
        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}")

Creating and Sending Trade Signals

TradeSignals Class

import requests

class TradeSignals:
    def __init__(self, main_url: str, order_url: str, access_token: str, 
                 broker_id: str, token: dict):
        self.main_url = main_url
        self.order_url = order_url
        self.access_token = access_token
        self.broker_id = broker_id

        self.headers = {
            "Content-Type": "application/json",
            "X-CSRF-TOKEN-ACCESS": token.get('X-CSRF-TOKEN-ACCESS'),
            "Cookie": f"access_token_cookie={token.get('Authorization')}"
        }

    def create_trade_signals(self, payload: dict):
        """Create a new trade signal"""
        response = requests.post(
            f"{self.main_url}/trade-signal/create", 
            json=payload, 
            headers=self.headers
        )

        if response.status_code == 200:
            content = response.json()
            print(f"Signal created successfully: {content}")
            return content.get("id")
        else:
            raise Exception(f"Failed to create signal: {response.status_code}, {response.text}")

    def send_trade_signals(self, tag: str, payload: str, execution_type: str = "paper"):
        """Send a trade signal"""
        if execution_type == "paper":
            url = f"{self.order_url}/webhook/tv/tk-trade?token={self.access_token}&tag={tag}"
        elif execution_type == "live":
            url = f"{self.order_url}/webhook/tv/tk-trade?token={self.access_token}&tag={tag}&brokers={self.broker_id}"

        response = requests.post(url, data=payload, headers=self.headers)

        if response.status_code == 200:
            print(f"Signal sent successfully: {response.json()}")
            return True
        else:
            raise Exception(f"Failed to send signal: {response.status_code}, {response.text}")

Usage Examples

Creating a Signal

# Create a paper trading signal
signal_payload = {
    "signal_name": "My Strategy Signal",
    "signal_type": "paper",
    "brokers": []
}

signal_tag = trade.create_trade_signals(signal_payload)

Sending Trade Orders

# Buy order
buy_order = "BTCUSD.P buy 10"
trade.send_trade_signals(
    tag=signal_tag,
    payload=buy_order,
    execution_type='paper'
)

# Sell order
sell_order = "BTCUSD.P sell 10"
trade.send_trade_signals(
    tag=signal_tag,
    payload=sell_order,
    execution_type='paper'
)

# Live trading (requires broker setup)
trade.send_trade_signals(
    tag=signal_tag,
    payload=buy_order,
    execution_type='live'
)

Complete Example: EMA-Based Strategy

Here's the complete main script that ties everything together:

from algotest_login import AlgoTestClient
from contracts_fetch import ContractFetcher
from option_chain_websocket import OptionChainWebSocketClient
from strategy import Strategy
from dotenv import load_dotenv
import os

# Load environment variables
load_dotenv()

# Configuration
PHONE_NUMBER = os.getenv("PHONE_NUMBER")
ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")
BROKER_ID = os.getenv("BROKER_ID")

UNDERLYING = "DELTA_BTCUSD"
PRICES_URL = "https://prices.algotest.in"
MAIN_URL = "https://api.algotest.in"
ORDERS_URL = "https://orders.algotest.in"
OPTION_CHAIN_WS = "wss://prices.algotest.in/updates?structured=true"

def main():
    try:
        # Step 1: Authenticate
        print("Authenticating...")
        client = AlgoTestClient(phone_number=PHONE_NUMBER, main_url=MAIN_URL)

        # Step 2: Fetch contracts
        print("Fetching contracts...")
        contracts = ContractFetcher(
            token=client.get_tokens(),
            underlying=UNDERLYING,
            prices_url=PRICES_URL
        )
        print(f"Found {contracts.contract_count} contracts")

        # Step 3: Set up WebSocket subscription
        subscription_message = {
            "msg": {
                "type": "subscribe",
                "datatypes": ["candle"],
                "underlyings": [
                    {
                        "underlying": UNDERLYING,
                        "cash": False,
                        "options": [],
                        "futures": [],
                    }
                ],
                "tokens": ["DELTA_27"],
            }
        }

        # Step 4: Initialize strategy
        print("Initializing strategy...")
        strategy = Strategy(
            underlying=UNDERLYING,
            main_url=MAIN_URL,
            order_url=ORDERS_URL,
            access_token=ACCESS_TOKEN,
            broker_id=BROKER_ID,
            token=client.get_tokens()
        )

        # Step 5: Start WebSocket connection
        print("Starting WebSocket connection...")
        ws_client = OptionChainWebSocketClient(
            OPTION_CHAIN_WS,
            client.jwt_token,
            subscription_message,
            on_data_callback=strategy.handle_price_update
        )

        # This will run indefinitely
        ws_client.start()

    except Exception as e:
        print(f"Error in main: {e}")

if __name__ == "__main__":
    main()

Environment Setup

Required Environment Variables

Create a .env file in your project root:

PHONE_NUMBER=+1234567890
PASSWORD=your_algotest_password
ACCESS_TOKEN=your_access_token
BROKER_ID=your_broker_id

Dependencies

Install required packages:

pip install -r requirements.txt

requirements.txt:

certifi==2025.7.9
charset-normalizer==3.4.2
python-dotenv==1.1.1
requests==2.32.4
websocket-client
gevent==25.5.1
greenlet==3.2.3

Error Handling

Robust Error Handling Example

import logging
from functools import wraps
import time

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def handle_api_errors(func):
    """Decorator for handling API errors"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.ConnectionError as e:
            logger.error(f"Connection error in {func.__name__}: {e}")
            raise
        except requests.exceptions.Timeout as e:
            logger.error(f"Timeout error in {func.__name__}: {e}")
            raise
        except requests.exceptions.RequestException as e:
            logger.error(f"Request error in {func.__name__}: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error in {func.__name__}: {e}")
            raise
    return wrapper

class RobustTradeSignals(TradeSignals):
    @handle_api_errors
    def create_trade_signals(self, payload: dict, max_retries: int = 3):
        """Create trade signals with retry logic"""
        for attempt in range(max_retries):
            try:
                response = requests.post(
                    f"{self.main_url}/trade-signal/create",
                    json=payload,
                    headers=self.headers,
                    timeout=30
                )

                if response.status_code == 200:
                    content = response.json()
                    logger.info(f"Signal created successfully: {content}")
                    return content.get("id")
                elif response.status_code == 429:  # Rate limit
                    wait_time = 2 ** attempt
                    logger.warning(f"Rate limited, waiting {wait_time}s...")
                    time.sleep(wait_time)
                    continue
                else:
                    raise Exception(f"API error: {response.status_code}, {response.text}")

            except requests.exceptions.Timeout:
                if attempt == max_retries - 1:
                    raise
                logger.warning(f"Timeout on attempt {attempt + 1}, retrying...")
                time.sleep(2 ** attempt)

Best Practices

1. Rate Limiting

import time
from functools import wraps

class RateLimiter:
    def __init__(self, max_calls: int, time_window: int):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = []

    def wait_if_needed(self):
        now = time.time()
        # Remove old calls outside the time window
        self.calls = [call_time for call_time in self.calls 
                     if now - call_time < self.time_window]

        if len(self.calls) >= self.max_calls:
            sleep_time = self.time_window - (now - self.calls[0])
            if sleep_time > 0:
                time.sleep(sleep_time)

        self.calls.append(now)

# Usage
rate_limiter = RateLimiter(max_calls=10, time_window=60)  # 10 calls per minute

def rate_limited_api_call():
    rate_limiter.wait_if_needed()
    # Make your API call here

2. Position Management

class PositionManager:
    def __init__(self):
        self.positions = {}
        self.max_position_size = 100
        self.max_daily_trades = 50
        self.daily_trade_count = 0
        self.last_trade_date = None

    def can_open_position(self, symbol: str, quantity: int) -> bool:
        """Check if position can be opened"""
        current_position = self.positions.get(symbol, 0)

        # Check position size limits
        if abs(current_position + quantity) > self.max_position_size:
            return False

        # Check daily trade limits
        today = datetime.now().date()
        if self.last_trade_date != today:
            self.daily_trade_count = 0
            self.last_trade_date = today

        return self.daily_trade_count < self.max_daily_trades

    def update_position(self, symbol: str, quantity: int):
        """Update position after trade execution"""
        if symbol not in self.positions:
            self.positions[symbol] = 0

        self.positions[symbol] += quantity
        self.daily_trade_count += 1

        if self.positions[symbol] == 0:
            del self.positions[symbol]

3. Strategy Validation

def validate_strategy_signals(strategy_instance):
    """Validate strategy before live trading"""
    test_data = generate_test_candles()  # Your test data

    signals = []
    for candle in test_data:
        signal = strategy_instance.process_candle(candle)
        if signal:
            signals.append(signal)

    # Check for excessive trading
    if len(signals) > len(test_data) * 0.1:  # More than 10% of candles
        raise ValueError("Strategy generates too many signals")

    # Check for minimum holding period
    holding_periods = calculate_holding_periods(signals)
    if min(holding_periods) < 5:  # Less than 5 minutes
        raise ValueError("Strategy has insufficient holding periods")

    return True

4. Configuration Management

from dataclasses import dataclass
from typing import Optional

@dataclass
class StrategyConfig:
    # Trading parameters
    symbol: str = "BTCUSD.P"
    quantity: int = 10
    max_position: int = 100

    # Technical indicators
    ema_period: int = 21
    rsi_period: int = 14
    bb_period: int = 20

    # Risk management
    stop_loss_pct: float = 2.0
    take_profit_pct: float = 4.0
    max_daily_loss: float = 1000.0

    # API settings
    execution_type: str = "paper"  # "paper" or "live"
    rate_limit_calls: int = 10
    rate_limit_window: int = 60

    @classmethod
    def from_file(cls, config_file: str) -> 'StrategyConfig':
        """Load configuration from JSON file"""
        import json
        with open(config_file, 'r') as f:
            config_dict = json.load(f)
        return cls(**config_dict)

This comprehensive guide provides everything you need to build sophisticated trading strategies using the AlgoTest Trade Signals API. Remember to always test your strategies thoroughly in paper trading mode before deploying them live.

Additional Resources

AlgoTest Community: https://t.me/algotest_in
Support: https://algotest.in/contact
API Documentation: Refer to the other sections of this documentation

⚠️ Disclaimer: This code is for educational and development purposes only. Trading in derivatives involves risk. Always test your strategies in paper trading mode before going live.

Trade Signals API Examples

Table of Contents

Quick Start

Authentication

AlgoTest Login Client

Usage Example

Fetching Contracts

Usage Example

WebSocket Connection for Live Data

Subscription Message Example

Building a Trading Strategy

EMA-Based Strategy Example

Creating and Sending Trade Signals

TradeSignals Class

Usage Examples

Creating a Signal

Sending Trade Orders

Complete Example: EMA-Based Strategy

Environment Setup

Required Environment Variables

Dependencies

Error Handling

Robust Error Handling Example

Best Practices

1. Rate Limiting

2. Position Management

3. Strategy Validation

4. Configuration Management

Additional Resources