From Starting Credit to Pro: Scaling Your Trading Bot as It Grows

# Stage 1: Simple and direct
import requests

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

def check_signal(symbol: str) -> str:
    resp = requests.get(f"{BASE_URL}/indicators", params={
        "symbol": symbol,
        "timeframe": "H1",
    }, headers={"X-API-Key": API_KEY})

    data = resp.json()["data"]["indicators"]
    rsi = data["RSI_14"]
    macd = data["MACD_hist"]

    if rsi < 30 and macd > 0:
        return "BUY"
    return "HOLD"

# Stage 2: Add caching and logging
import time
import logging

logging.basicConfig(level=logging.INFO, filename="bot.log")
logger = logging.getLogger("trading_bot")

class TradingBot:
    def __init__(self, api_key: str, symbols: list[str]):
        self.api_key = api_key
        self.symbols = symbols
        self.cache = {}
        self.cache_time = {}

    def get_data(self, symbol: str, ttl: int = 300) -> dict:
        """Cached API call — only fetches when stale."""
        now = time.time()
        if symbol in self.cache and now - self.cache_time[symbol] < ttl:
            return self.cache[symbol]

        resp = requests.get(f"{BASE_URL}/indicators", params={
            "symbol": symbol,
            "timeframe": "H1",
        }, headers={"X-API-Key": self.api_key})

        if resp.status_code == 429:
            logger.warning(f"Rate limited on {symbol}. Backing off.")
            time.sleep(5)
            return self.cache.get(symbol, {})

        data = resp.json()
        self.cache[symbol] = data
        self.cache_time[symbol] = now
        return data

# Stage 4: Multi-strategy architecture
from concurrent.futures import ThreadPoolExecutor
from queue import Queue

class MultiStrategyBot:
    def __init__(self, api_key: str):
        self.client = CachedClient(api_key)
        self.strategies = []
        self.signal_queue = Queue()

    def add_strategy(self, strategy, symbols, timeframe):
        self.strategies.append({
            "strategy": strategy,
            "symbols": symbols,
            "timeframe": timeframe
        })

    def run_all(self):
        """Run all strategies in parallel."""
        with ThreadPoolExecutor(max_workers=4) as pool:
            futures = []
            for strat_config in self.strategies:
                for symbol in strat_config["symbols"]:
                    future = pool.submit(
                        self._evaluate,
                        strat_config["strategy"],
                        symbol,
                        strat_config["timeframe"]
                    )
                    futures.append(future)

            for future in futures:
                result = future.result()
                if result and result["signal"] != "HOLD":
                    self.signal_queue.put(result)

    def _evaluate(self, strategy, symbol, timeframe):
        data = self.client.get_indicators(symbol, timeframe)
        return strategy.evaluate(symbol, data)