Signal calculation examples¶

Compare moving-average, RSI and Bollinger-band calculations using historical data. These indicators transform observed prices; they do not establish that a trade should be placed or that combining indicators improves predictions. Inspect order methods and their call sites before enabling execution.

Setup¶

Use a dedicated OANDA practice account and set both FIVETWENTY_OANDA_TOKEN and FIVETWENTY_OANDA_ACCOUNT in the kernel's environment. Review the configuration cell before making requests. Install the packages imported by the setup cell, select that Python environment as the Jupyter kernel, and run cells in order.

The repository's uv run poe docs-validate-notebooks command executes a temporary copy with synthetic HTTP responses. Running this notebook normally uses its configured API credentials; the offline validation result is not a live account test.

Setup and Imports¶

In [ ]:

import os
from decimal import Decimal

import pandas as pd

from fivetwenty import AsyncClient, Environment
from fivetwenty.exceptions import FiveTwentyError
from fivetwenty.models import CandlestickGranularity

# Jupyter async support
try:
    import nest_asyncio

    nest_asyncio.apply()
except ImportError:
    print("Install nest_asyncio for Jupyter: uv add nest_asyncio")

# Configuration
TOKEN = os.getenv("FIVETWENTY_OANDA_TOKEN", "your-token-here")
ENVIRONMENT = Environment.PRACTICE
ACCOUNT_ID = os.getenv("FIVETWENTY_OANDA_ACCOUNT", "your-account-id-here")

print("✅ Setup complete" if TOKEN != "your-token-here" and ACCOUNT_ID != "your-account-id-here" else "⚠️ Set FIVETWENTY_OANDA_TOKEN and FIVETWENTY_OANDA_ACCOUNT environment variables")

import os from decimal import Decimal import pandas as pd from fivetwenty import AsyncClient, Environment from fivetwenty.exceptions import FiveTwentyError from fivetwenty.models import CandlestickGranularity # Jupyter async support try: import nest_asyncio nest_asyncio.apply() except ImportError: print("Install nest_asyncio for Jupyter: uv add nest_asyncio") # Configuration TOKEN = os.getenv("FIVETWENTY_OANDA_TOKEN", "your-token-here") ENVIRONMENT = Environment.PRACTICE ACCOUNT_ID = os.getenv("FIVETWENTY_OANDA_ACCOUNT", "your-account-id-here") print("✅ Setup complete" if TOKEN != "your-token-here" and ACCOUNT_ID != "your-account-id-here" else "⚠️ Set FIVETWENTY_OANDA_TOKEN and FIVETWENTY_OANDA_ACCOUNT environment variables")

Strategy Base Class¶

Let's create a base class for our trading strategies:

In [ ]:

class TradingStrategy:
    """Base class for trading strategies."""

    def __init__(self, client: AsyncClient, account_id: str, instrument: str):
        self.client = client
        self.account_id = account_id
        self.instrument = instrument
        self.position_size = 1000  # Default position size

    async def get_candles(self, count: int = 100, granularity: CandlestickGranularity = CandlestickGranularity.M5) -> pd.DataFrame:
        """Get historical candlestick data."""
        try:
            candles = await self.client.instruments.get_instrument_candles(instrument=self.instrument, count=count, granularity=granularity)

            # Convert to pandas DataFrame
            data = []
            for candle in candles["candles"]:
                if candle.mid:
                    data.append({"time": pd.to_datetime(candle.time), "open": float(candle.mid.o), "high": float(candle.mid.h), "low": float(candle.mid.l), "close": float(candle.mid.c), "volume": int(candle.volume)})

            df = pd.DataFrame(data)
            df.set_index("time", inplace=True)
            return df

        except FiveTwentyError as e:
            print(f"Error getting candles: {e.message}")
            return pd.DataFrame()

    async def place_order(self, units: int, stop_loss: float | None = None, take_profit: float | None = None):
        """Place a market order with optional stop loss and take profit."""
        try:
            order_kwargs = {"account_id": self.account_id, "instrument": self.instrument, "units": units}

            if stop_loss:
                order_kwargs["stop_loss"] = Decimal(str(stop_loss))

            if take_profit:
                order_kwargs["take_profit"] = Decimal(str(take_profit))

            response = await self.client.orders.post_market_order(**order_kwargs)

            if response.order_fill_transaction:
                fill = response.order_fill_transaction
                direction = "BUY" if int(units) > 0 else "SELL"
                print(f"✅ {direction} order filled: {units} units at {fill.price}")
                return fill
            print("❌ Order not filled")
            return None

        except FiveTwentyError as e:
            print(f"❌ Order error: {e.message}")
            return None

    async def analyze(self) -> dict:
        """Analyze market conditions. Override in subclasses."""
        raise NotImplementedError

    async def execute(self) -> bool:
        """Execute trading logic. Override in subclasses."""
        raise NotImplementedError


print("✅ Base strategy class defined")

class TradingStrategy: """Base class for trading strategies.""" def __init__(self, client: AsyncClient, account_id: str, instrument: str): self.client = client self.account_id = account_id self.instrument = instrument self.position_size = 1000 # Default position size async def get_candles(self, count: int = 100, granularity: CandlestickGranularity = CandlestickGranularity.M5) -> pd.DataFrame: """Get historical candlestick data.""" try: candles = await self.client.instruments.get_instrument_candles(instrument=self.instrument, count=count, granularity=granularity) # Convert to pandas DataFrame data = [] for candle in candles["candles"]: if candle.mid: data.append({"time": pd.to_datetime(candle.time), "open": float(candle.mid.o), "high": float(candle.mid.h), "low": float(candle.mid.l), "close": float(candle.mid.c), "volume": int(candle.volume)}) df = pd.DataFrame(data) df.set_index("time", inplace=True) return df except FiveTwentyError as e: print(f"Error getting candles: {e.message}") return pd.DataFrame() async def place_order(self, units: int, stop_loss: float | None = None, take_profit: float | None = None): """Place a market order with optional stop loss and take profit.""" try: order_kwargs = {"account_id": self.account_id, "instrument": self.instrument, "units": units} if stop_loss: order_kwargs["stop_loss"] = Decimal(str(stop_loss)) if take_profit: order_kwargs["take_profit"] = Decimal(str(take_profit)) response = await self.client.orders.post_market_order(**order_kwargs) if response.order_fill_transaction: fill = response.order_fill_transaction direction = "BUY" if int(units) > 0 else "SELL" print(f"✅ {direction} order filled: {units} units at {fill.price}") return fill print("❌ Order not filled") return None except FiveTwentyError as e: print(f"❌ Order error: {e.message}") return None async def analyze(self) -> dict: """Analyze market conditions. Override in subclasses.""" raise NotImplementedError async def execute(self) -> bool: """Execute trading logic. Override in subclasses.""" raise NotImplementedError print("✅ Base strategy class defined")

1. Moving Average Crossover Strategy¶

This example compares short and long moving averages. Inspect whether its condition detects a new crossover or simply an existing above/below relationship before connecting it to repeated execution.

In [ ]:

class MovingAverageCrossover(TradingStrategy):
    """Moving Average Crossover Strategy."""

    def __init__(self, client: AsyncClient, account_id: str, instrument: str, short_period: int = 10, long_period: int = 20):
        super().__init__(client, account_id, instrument)
        self.short_period = short_period
        self.long_period = long_period

    async def analyze(self) -> dict:
        """Calculate moving averages and determine signal."""
        # Get enough data for the longer moving average
        df = await self.get_candles(count=self.long_period + 10)

        if df.empty:
            return {"signal": "NO_DATA", "short_ma": None, "long_ma": None}

        # Calculate moving averages
        df["short_ma"] = df["close"].rolling(window=self.short_period).mean()
        df["long_ma"] = df["close"].rolling(window=self.long_period).mean()

        # Get current and previous values
        current_short = df["short_ma"].iloc[-1]
        current_long = df["long_ma"].iloc[-1]
        prev_short = df["short_ma"].iloc[-2]
        prev_long = df["long_ma"].iloc[-2]

        # Determine signal
        signal = "HOLD"
        if prev_short <= prev_long and current_short > current_long:
            signal = "BUY"  # Golden cross
        elif prev_short >= prev_long and current_short < current_long:
            signal = "SELL"  # Death cross

        return {"signal": signal, "short_ma": current_short, "long_ma": current_long, "current_price": df["close"].iloc[-1], "dataframe": df}

    async def execute(self) -> bool:
        """Execute the moving average crossover strategy."""
        analysis = await self.analyze()

        print(f"📊 MA Analysis for {self.instrument}:")
        print(f"  Short MA ({self.short_period}): {analysis['short_ma']:.5f}")
        print(f"  Long MA ({self.long_period}): {analysis['long_ma']:.5f}")
        print(f"  Current Price: {analysis['current_price']:.5f}")
        print(f"  Signal: {analysis['signal']}")

        if analysis["signal"] == "BUY":
            # Calculate stop loss and take profit
            current_price = analysis["current_price"]
            stop_loss = current_price * 0.995  # 0.5% stop loss
            take_profit = current_price * 1.01  # 1% take profit

            await self.place_order(self.position_size, stop_loss, take_profit)
            return True

        if analysis["signal"] == "SELL":
            # Calculate stop loss and take profit for short position
            current_price = analysis["current_price"]
            stop_loss = current_price * 1.005  # 0.5% stop loss
            take_profit = current_price * 0.99  # 1% take profit

            await self.place_order(-self.position_size, stop_loss, take_profit)
            return True

        return False


print("✅ Moving Average Crossover strategy defined")

class MovingAverageCrossover(TradingStrategy): """Moving Average Crossover Strategy.""" def __init__(self, client: AsyncClient, account_id: str, instrument: str, short_period: int = 10, long_period: int = 20): super().__init__(client, account_id, instrument) self.short_period = short_period self.long_period = long_period async def analyze(self) -> dict: """Calculate moving averages and determine signal.""" # Get enough data for the longer moving average df = await self.get_candles(count=self.long_period + 10) if df.empty: return {"signal": "NO_DATA", "short_ma": None, "long_ma": None} # Calculate moving averages df["short_ma"] = df["close"].rolling(window=self.short_period).mean() df["long_ma"] = df["close"].rolling(window=self.long_period).mean() # Get current and previous values current_short = df["short_ma"].iloc[-1] current_long = df["long_ma"].iloc[-1] prev_short = df["short_ma"].iloc[-2] prev_long = df["long_ma"].iloc[-2] # Determine signal signal = "HOLD" if prev_short <= prev_long and current_short > current_long: signal = "BUY" # Golden cross elif prev_short >= prev_long and current_short < current_long: signal = "SELL" # Death cross return {"signal": signal, "short_ma": current_short, "long_ma": current_long, "current_price": df["close"].iloc[-1], "dataframe": df} async def execute(self) -> bool: """Execute the moving average crossover strategy.""" analysis = await self.analyze() print(f"📊 MA Analysis for {self.instrument}:") print(f" Short MA ({self.short_period}): {analysis['short_ma']:.5f}") print(f" Long MA ({self.long_period}): {analysis['long_ma']:.5f}") print(f" Current Price: {analysis['current_price']:.5f}") print(f" Signal: {analysis['signal']}") if analysis["signal"] == "BUY": # Calculate stop loss and take profit current_price = analysis["current_price"] stop_loss = current_price * 0.995 # 0.5% stop loss take_profit = current_price * 1.01 # 1% take profit await self.place_order(self.position_size, stop_loss, take_profit) return True if analysis["signal"] == "SELL": # Calculate stop loss and take profit for short position current_price = analysis["current_price"] stop_loss = current_price * 1.005 # 0.5% stop loss take_profit = current_price * 0.99 # 1% take profit await self.place_order(-self.position_size, stop_loss, take_profit) return True return False print("✅ Moving Average Crossover strategy defined")

2. RSI Mean Reversion Strategy¶

A strategy that uses the Relative Strength Index to identify overbought/oversold conditions.

In [ ]:

class RSIMeanReversion(TradingStrategy):
    """RSI Mean Reversion Strategy."""

    def __init__(self, client: AsyncClient, account_id: str, instrument: str, rsi_period: int = 14, oversold: float = 30, overbought: float = 70):
        super().__init__(client, account_id, instrument)
        self.rsi_period = rsi_period
        self.oversold = oversold
        self.overbought = overbought

    def calculate_rsi(self, df: pd.DataFrame) -> pd.Series:
        """Calculate RSI (Relative Strength Index)."""
        delta = df["close"].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=self.rsi_period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=self.rsi_period).mean()

        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        return rsi

    async def analyze(self) -> dict:
        """Calculate RSI and determine signal."""
        df = await self.get_candles(count=self.rsi_period + 20)

        if df.empty:
            return {"signal": "NO_DATA", "rsi": None}

        # Calculate RSI
        df["rsi"] = self.calculate_rsi(df)

        current_rsi = df["rsi"].iloc[-1]

        # Determine signal
        signal = "HOLD"
        if current_rsi < self.oversold:
            signal = "BUY"  # Oversold - expect price to rise
        elif current_rsi > self.overbought:
            signal = "SELL"  # Overbought - expect price to fall

        return {"signal": signal, "rsi": current_rsi, "current_price": df["close"].iloc[-1], "dataframe": df}

    async def execute(self) -> bool:
        """Execute the RSI mean reversion strategy."""
        analysis = await self.analyze()

        print(f"📊 RSI Analysis for {self.instrument}:")
        print(f"  RSI: {analysis['rsi']:.2f}")
        print(f"  Current Price: {analysis['current_price']:.5f}")
        print(f"  Signal: {analysis['signal']}")

        if analysis["signal"] == "BUY":
            current_price = analysis["current_price"]
            stop_loss = current_price * 0.99  # 1% stop loss
            take_profit = current_price * 1.02  # 2% take profit

            await self.place_order(self.position_size, stop_loss, take_profit)
            return True

        if analysis["signal"] == "SELL":
            current_price = analysis["current_price"]
            stop_loss = current_price * 1.01  # 1% stop loss
            take_profit = current_price * 0.98  # 2% take profit

            await self.place_order(-self.position_size, stop_loss, take_profit)
            return True

        return False


print("✅ RSI Mean Reversion strategy defined")

class RSIMeanReversion(TradingStrategy): """RSI Mean Reversion Strategy.""" def __init__(self, client: AsyncClient, account_id: str, instrument: str, rsi_period: int = 14, oversold: float = 30, overbought: float = 70): super().__init__(client, account_id, instrument) self.rsi_period = rsi_period self.oversold = oversold self.overbought = overbought def calculate_rsi(self, df: pd.DataFrame) -> pd.Series: """Calculate RSI (Relative Strength Index).""" delta = df["close"].diff() gain = (delta.where(delta > 0, 0)).rolling(window=self.rsi_period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=self.rsi_period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi async def analyze(self) -> dict: """Calculate RSI and determine signal.""" df = await self.get_candles(count=self.rsi_period + 20) if df.empty: return {"signal": "NO_DATA", "rsi": None} # Calculate RSI df["rsi"] = self.calculate_rsi(df) current_rsi = df["rsi"].iloc[-1] # Determine signal signal = "HOLD" if current_rsi < self.oversold: signal = "BUY" # Oversold - expect price to rise elif current_rsi > self.overbought: signal = "SELL" # Overbought - expect price to fall return {"signal": signal, "rsi": current_rsi, "current_price": df["close"].iloc[-1], "dataframe": df} async def execute(self) -> bool: """Execute the RSI mean reversion strategy.""" analysis = await self.analyze() print(f"📊 RSI Analysis for {self.instrument}:") print(f" RSI: {analysis['rsi']:.2f}") print(f" Current Price: {analysis['current_price']:.5f}") print(f" Signal: {analysis['signal']}") if analysis["signal"] == "BUY": current_price = analysis["current_price"] stop_loss = current_price * 0.99 # 1% stop loss take_profit = current_price * 1.02 # 2% take profit await self.place_order(self.position_size, stop_loss, take_profit) return True if analysis["signal"] == "SELL": current_price = analysis["current_price"] stop_loss = current_price * 1.01 # 1% stop loss take_profit = current_price * 0.98 # 2% take profit await self.place_order(-self.position_size, stop_loss, take_profit) return True return False print("✅ RSI Mean Reversion strategy defined")

3. Bollinger Bands Strategy¶

This example compares prices with bands derived from recent data; a band touch does not confirm a reversal.

In [ ]:

class BollingerBands(TradingStrategy):
    """Bollinger Bands Strategy."""

    def __init__(self, client: AsyncClient, account_id: str, instrument: str, period: int = 20, std_dev: float = 2.0):
        super().__init__(client, account_id, instrument)
        self.period = period
        self.std_dev = std_dev

    async def analyze(self) -> dict:
        """Calculate Bollinger Bands and determine signal."""
        df = await self.get_candles(count=self.period + 10)

        if df.empty:
            return {"signal": "NO_DATA"}

        # Calculate Bollinger Bands
        df["sma"] = df["close"].rolling(window=self.period).mean()
        df["std"] = df["close"].rolling(window=self.period).std()
        df["upper_band"] = df["sma"] + (df["std"] * self.std_dev)
        df["lower_band"] = df["sma"] - (df["std"] * self.std_dev)

        # Current values
        current_price = df["close"].iloc[-1]
        upper_band = df["upper_band"].iloc[-1]
        lower_band = df["lower_band"].iloc[-1]
        sma = df["sma"].iloc[-1]

        # Determine signal
        signal = "HOLD"
        if current_price <= lower_band:
            signal = "BUY"  # Price touched lower band - potential bounce
        elif current_price >= upper_band:
            signal = "SELL"  # Price touched upper band - potential reversal

        return {"signal": signal, "current_price": current_price, "upper_band": upper_band, "lower_band": lower_band, "sma": sma, "band_width": upper_band - lower_band, "dataframe": df}

    async def execute(self) -> bool:
        """Execute the Bollinger Bands strategy."""
        analysis = await self.analyze()

        print(f"📊 Bollinger Bands Analysis for {self.instrument}:")
        print(f"  Current Price: {analysis['current_price']:.5f}")
        print(f"  Upper Band: {analysis['upper_band']:.5f}")
        print(f"  SMA: {analysis['sma']:.5f}")
        print(f"  Lower Band: {analysis['lower_band']:.5f}")
        print(f"  Band Width: {analysis['band_width']:.5f}")
        print(f"  Signal: {analysis['signal']}")

        if analysis["signal"] == "BUY":
            # Target the SMA as take profit, stop loss below lower band
            stop_loss = analysis["lower_band"] * 0.999
            take_profit = analysis["sma"]

            await self.place_order(self.position_size, stop_loss, take_profit)
            return True

        if analysis["signal"] == "SELL":
            # Target the SMA as take profit, stop loss above upper band
            stop_loss = analysis["upper_band"] * 1.001
            take_profit = analysis["sma"]

            await self.place_order(-self.position_size, stop_loss, take_profit)
            return True

        return False


print("✅ Bollinger Bands strategy defined")

class BollingerBands(TradingStrategy): """Bollinger Bands Strategy.""" def __init__(self, client: AsyncClient, account_id: str, instrument: str, period: int = 20, std_dev: float = 2.0): super().__init__(client, account_id, instrument) self.period = period self.std_dev = std_dev async def analyze(self) -> dict: """Calculate Bollinger Bands and determine signal.""" df = await self.get_candles(count=self.period + 10) if df.empty: return {"signal": "NO_DATA"} # Calculate Bollinger Bands df["sma"] = df["close"].rolling(window=self.period).mean() df["std"] = df["close"].rolling(window=self.period).std() df["upper_band"] = df["sma"] + (df["std"] * self.std_dev) df["lower_band"] = df["sma"] - (df["std"] * self.std_dev) # Current values current_price = df["close"].iloc[-1] upper_band = df["upper_band"].iloc[-1] lower_band = df["lower_band"].iloc[-1] sma = df["sma"].iloc[-1] # Determine signal signal = "HOLD" if current_price <= lower_band: signal = "BUY" # Price touched lower band - potential bounce elif current_price >= upper_band: signal = "SELL" # Price touched upper band - potential reversal return {"signal": signal, "current_price": current_price, "upper_band": upper_band, "lower_band": lower_band, "sma": sma, "band_width": upper_band - lower_band, "dataframe": df} async def execute(self) -> bool: """Execute the Bollinger Bands strategy.""" analysis = await self.analyze() print(f"📊 Bollinger Bands Analysis for {self.instrument}:") print(f" Current Price: {analysis['current_price']:.5f}") print(f" Upper Band: {analysis['upper_band']:.5f}") print(f" SMA: {analysis['sma']:.5f}") print(f" Lower Band: {analysis['lower_band']:.5f}") print(f" Band Width: {analysis['band_width']:.5f}") print(f" Signal: {analysis['signal']}") if analysis["signal"] == "BUY": # Target the SMA as take profit, stop loss below lower band stop_loss = analysis["lower_band"] * 0.999 take_profit = analysis["sma"] await self.place_order(self.position_size, stop_loss, take_profit) return True if analysis["signal"] == "SELL": # Target the SMA as take profit, stop loss above upper band stop_loss = analysis["upper_band"] * 1.001 take_profit = analysis["sma"] await self.place_order(-self.position_size, stop_loss, take_profit) return True return False print("✅ Bollinger Bands strategy defined")

Initialize Connection and Test Strategies¶

Now let's connect to OANDA and test our strategies:

In [ ]:

async def initialize_connection():
    """Initialize connection and get account ID."""
    global ACCOUNT_ID

    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        try:
            accounts = await client.accounts.get_accounts()
            if accounts:
                ACCOUNT_ID = accounts[0].id
                print(f"✅ Connected to account: {ACCOUNT_ID}")
                return ACCOUNT_ID
            print("❌ No accounts found")
            return None
        except FiveTwentyError as e:
            print(f"❌ Connection error: {e.message}")
            return None


# Initialize connection
account_id = await initialize_connection()

async def initialize_connection(): """Initialize connection and get account ID.""" global ACCOUNT_ID async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: try: accounts = await client.accounts.get_accounts() if accounts: ACCOUNT_ID = accounts[0].id print(f"✅ Connected to account: {ACCOUNT_ID}") return ACCOUNT_ID print("❌ No accounts found") return None except FiveTwentyError as e: print(f"❌ Connection error: {e.message}") return None # Initialize connection account_id = await initialize_connection()

Test Moving Average Strategy¶

In [ ]:

if account_id:
    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        # Test Moving Average Crossover
        ma_strategy = MovingAverageCrossover(client, account_id, "EUR_USD", short_period=5, long_period=15)

        print("🔄 Testing Moving Average Crossover Strategy...")
        executed = await ma_strategy.execute()

        if not executed:
            print("📈 No trade signal - market analysis complete")
else:
    print("❌ No account connection - cannot test strategies")

if account_id: async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: # Test Moving Average Crossover ma_strategy = MovingAverageCrossover(client, account_id, "EUR_USD", short_period=5, long_period=15) print("🔄 Testing Moving Average Crossover Strategy...") executed = await ma_strategy.execute() if not executed: print("📈 No trade signal - market analysis complete") else: print("❌ No account connection - cannot test strategies")

Test RSI Strategy¶

In [ ]:

if account_id:
    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        # Test RSI Mean Reversion
        rsi_strategy = RSIMeanReversion(client, account_id, "GBP_USD", oversold=25, overbought=75)

        print("🔄 Testing RSI Mean Reversion Strategy...")
        executed = await rsi_strategy.execute()

        if not executed:
            print("📈 No trade signal - market analysis complete")
else:
    print("❌ No account connection - cannot test strategies")

if account_id: async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: # Test RSI Mean Reversion rsi_strategy = RSIMeanReversion(client, account_id, "GBP_USD", oversold=25, overbought=75) print("🔄 Testing RSI Mean Reversion Strategy...") executed = await rsi_strategy.execute() if not executed: print("📈 No trade signal - market analysis complete") else: print("❌ No account connection - cannot test strategies")

Test Bollinger Bands Strategy¶

In [ ]:

if account_id:
    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        # Test Bollinger Bands
        bb_strategy = BollingerBands(client, account_id, "USD_JPY", period=15, std_dev=1.5)

        print("🔄 Testing Bollinger Bands Strategy...")
        executed = await bb_strategy.execute()

        if not executed:
            print("📈 No trade signal - market analysis complete")
else:
    print("❌ No account connection - cannot test strategies")

if account_id: async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: # Test Bollinger Bands bb_strategy = BollingerBands(client, account_id, "USD_JPY", period=15, std_dev=1.5) print("🔄 Testing Bollinger Bands Strategy...") executed = await bb_strategy.execute() if not executed: print("📈 No trade signal - market analysis complete") else: print("❌ No account connection - cannot test strategies")

Strategy Performance Analysis¶

Let's create a function to analyze multiple strategies simultaneously:

In [ ]:

async def analyze_multiple_strategies(account_id: str, instruments: list[str]):
    """Analyze multiple strategies across different instruments."""

    if not account_id:
        print("❌ No account connection")
        return None

    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        results = []

        for instrument in instruments:
            print(f"\n📊 Analyzing {instrument}...")

            # Moving Average analysis
            ma_strategy = MovingAverageCrossover(client, account_id, instrument)
            ma_analysis = await ma_strategy.analyze()

            # RSI analysis
            rsi_strategy = RSIMeanReversion(client, account_id, instrument)
            rsi_analysis = await rsi_strategy.analyze()

            # Bollinger Bands analysis
            bb_strategy = BollingerBands(client, account_id, instrument)
            bb_analysis = await bb_strategy.analyze()

            result = {"instrument": instrument, "ma_signal": ma_analysis.get("signal", "NO_DATA"), "rsi_signal": rsi_analysis.get("signal", "NO_DATA"), "bb_signal": bb_analysis.get("signal", "NO_DATA"), "current_price": ma_analysis.get("current_price", 0), "rsi": rsi_analysis.get("rsi", 0)}

            results.append(result)

            print(f"  MA Signal: {result['ma_signal']}")
            print(f"  RSI Signal: {result['rsi_signal']} (RSI: {result['rsi']:.1f})")
            print(f"  BB Signal: {result['bb_signal']}")

        # Create summary DataFrame
        df_results = pd.DataFrame(results)

        print("\n📈 Strategy Signals Summary:")
        print(df_results[["instrument", "ma_signal", "rsi_signal", "bb_signal"]].to_string(index=False))

        return df_results


# Analyze multiple currency pairs
instruments_to_analyze = ["EUR_USD", "GBP_USD", "USD_JPY", "AUD_USD"]
strategy_results = await analyze_multiple_strategies(account_id, instruments_to_analyze)

async def analyze_multiple_strategies(account_id: str, instruments: list[str]): """Analyze multiple strategies across different instruments.""" if not account_id: print("❌ No account connection") return None async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: results = [] for instrument in instruments: print(f"\n📊 Analyzing {instrument}...") # Moving Average analysis ma_strategy = MovingAverageCrossover(client, account_id, instrument) ma_analysis = await ma_strategy.analyze() # RSI analysis rsi_strategy = RSIMeanReversion(client, account_id, instrument) rsi_analysis = await rsi_strategy.analyze() # Bollinger Bands analysis bb_strategy = BollingerBands(client, account_id, instrument) bb_analysis = await bb_strategy.analyze() result = {"instrument": instrument, "ma_signal": ma_analysis.get("signal", "NO_DATA"), "rsi_signal": rsi_analysis.get("signal", "NO_DATA"), "bb_signal": bb_analysis.get("signal", "NO_DATA"), "current_price": ma_analysis.get("current_price", 0), "rsi": rsi_analysis.get("rsi", 0)} results.append(result) print(f" MA Signal: {result['ma_signal']}") print(f" RSI Signal: {result['rsi_signal']} (RSI: {result['rsi']:.1f})") print(f" BB Signal: {result['bb_signal']}") # Create summary DataFrame df_results = pd.DataFrame(results) print("\n📈 Strategy Signals Summary:") print(df_results[["instrument", "ma_signal", "rsi_signal", "bb_signal"]].to_string(index=False)) return df_results # Analyze multiple currency pairs instruments_to_analyze = ["EUR_USD", "GBP_USD", "USD_JPY", "AUD_USD"] strategy_results = await analyze_multiple_strategies(account_id, instruments_to_analyze)

Interpreting the results¶

Check the printed inputs, timestamps and handled errors before interpreting output. A completed cell can still report an API failure. Example calculations and simulations do not establish a profitable strategy, a guaranteed loss cap or readiness for unattended execution.

For further work, test boundary cases, currency conversion, incomplete data and recovery after a disconnect. Keep signal evaluation separate from order submission and reconcile any account changes.

• SDK lifecycle and data types
• Error handling
• Streaming behavior