Building a Grid Trading Bot with Technical Indicators

import requests
from dataclasses import dataclass

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://tickatlas.com/v1"

@dataclass
class GridLevel:
    price: float
    side: str  # "buy" or "sell"
    level: int
    filled: bool = False

class SmartGrid:
    def __init__(self, symbol: str, num_levels: int = 5,
                 atr_spacing_factor: float = 0.5):
        self.symbol = symbol
        self.num_levels = num_levels
        self.atr_factor = atr_spacing_factor
        self.levels: list[GridLevel] = []

    def calculate_grid(self, timeframe: str = "H4") -> list[GridLevel]:
        """Build grid levels using ATR for spacing."""
        # Fetch ATR for dynamic spacing
        atr_resp = requests.get(
            f"{BASE_URL}/indicator",
            headers={"X-API-Key": API_KEY},
            params={"symbol": self.symbol, "indicator": "ATR_14",
                    "timeframe": timeframe},
        )
        atr_data = atr_resp.json()["data"]
        atr = atr_data["values"]["atr"]
        center = atr_data["ohlcv"]["close"]

        spacing = atr * self.atr_factor
        self.levels = []

        for i in range(1, self.num_levels + 1):
            # Buy levels below center
            self.levels.append(GridLevel(
                price=round(center - (spacing * i), 5),
                side="buy", level=-i,
            ))
            # Sell levels above center
            self.levels.append(GridLevel(
                price=round(center + (spacing * i), 5),
                side="sell", level=i,
            ))

        self.levels.sort(key=lambda x: x.price, reverse=True)
        return self.levels

# Example
grid = SmartGrid("EURUSD", num_levels=5, atr_spacing_factor=0.5)
levels = grid.calculate_grid("H4")
for level in levels:
    print(f"  {level.side.upper()} @ {level.price} [Level {level.level}]")

def is_ranging_market(symbol: str, timeframe: str = "D1") -> dict:
    """Detect if market is in a range using Bollinger Band width."""
    bb_resp = requests.get(
        f"{BASE_URL}/indicator",
        headers={"X-API-Key": API_KEY},
        params={"symbol": symbol, "indicator": "BOLLINGER_BANDS",
                "timeframe": timeframe},
    )
    data = bb_resp.json()["data"]["values"]

    upper = data["upper"]
    lower = data["lower"]
    middle = data["middle"]

    # Band width as percentage of middle
    width_pct = ((upper - lower) / middle) * 100

    # Narrow bands suggest ranging market
    is_ranging = width_pct < 2.0  # Adjust threshold per symbol

    return {
        "symbol": symbol,
        "is_ranging": is_ranging,
        "band_width_pct": round(width_pct, 3),
        "upper": upper,
        "lower": lower,
        "grid_suitable": is_ranging,
    }

# Only run grid bot on ranging pairs
for sym in ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD"]:
    result = is_ranging_market(sym)
    if result["grid_suitable"]:
        print(f"{sym}: RANGING (width={result['band_width_pct']}%) -- grid OK")
    else:
        print(f"{sym}: TRENDING (width={result['band_width_pct']}%) -- skip")

import time

def run_grid_bot(symbol: str, timeframe: str = "H4"):
    """Main grid bot loop."""
    grid = SmartGrid(symbol, num_levels=5, atr_spacing_factor=0.5)
    levels = grid.calculate_grid(timeframe)

    print(f"Grid initialized for {symbol}")
    for level in levels:
        print(f"  {level.side.upper()} @ {level.price}")

    while True:
        # Check if market is still ranging
        range_check = is_ranging_market(symbol, "D1")
        if not range_check["grid_suitable"]:
            print(f"Market trending -- pausing grid for {symbol}")
            time.sleep(3600)
            continue

        # Get current price
        price_resp = requests.get(
            f"{BASE_URL}/indicator",
            headers={"X-API-Key": API_KEY},
            params={"symbol": symbol, "indicator": "RSI_14",
                    "timeframe": "M5"},
        )
        current = price_resp.json()["data"]["ohlcv"]["close"]

        # Check if any grid levels are hit
        for level in levels:
            if level.filled:
                continue
            if level.side == "buy" and current <= level.price:
                print(f"BUY triggered @ {level.price} (current: {current})")
                level.filled = True
                # Place buy order via broker API
            elif level.side == "sell" and current >= level.price:
                print(f"SELL triggered @ {level.price} (current: {current})")
                level.filled = True
                # Place sell order via broker API

        # Recalculate grid every 4 hours
        time.sleep(300)  # Check every 5 minutes