Sampled price streams and analysis¶

Consume sampled price updates, keep a rolling buffer, calculate indicators and inspect a local paper-trading simulation. OANDA price streaming is not a lossless tick feed. Simulation results depend on the simplified execution and cost assumptions in the code; they do not reproduce broker fills.

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
import time
from collections import deque
from collections.abc import Callable
from datetime import datetime, timezone

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

from fivetwenty import AsyncClient, Environment
from fivetwenty.exceptions import FiveTwentyError, StreamStall
from fivetwenty.models import ReconnectionPolicy, StreamingConfiguration, StreamState

# 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 import time from collections import deque from collections.abc import Callable from datetime import datetime, timezone import matplotlib.pyplot as plt import numpy as np import pandas as pd from fivetwenty import AsyncClient, Environment from fivetwenty.exceptions import FiveTwentyError, StreamStall from fivetwenty.models import ReconnectionPolicy, StreamingConfiguration, StreamState # 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")

Real-time Data Buffer System¶

The following class stores recent observations and exposes them for analysis. Review its retention limits before increasing the run duration.

In [ ]:

class RealTimeDataBuffer:
    """High-performance real-time data buffer with analytics."""

    def __init__(self, max_size: int = 1000):
        self.max_size = max_size
        self.prices = {}
        self.price_history = {}
        self.timestamps = {}
        self.statistics = {}
        self.callbacks = []

    def add_price(self, instrument: str, bid: float, ask: float, timestamp: datetime = None):
        """Add a new price tick to the buffer."""
        if timestamp is None:
            timestamp = datetime.now(timezone.utc)

        # Initialize instrument data structures if needed
        if instrument not in self.prices:
            self.prices[instrument] = {"bid": bid, "ask": ask, "mid": (bid + ask) / 2}
            self.price_history[instrument] = deque(maxlen=self.max_size)
            self.timestamps[instrument] = deque(maxlen=self.max_size)
            self.statistics[instrument] = {}

        # Update current prices
        mid_price = (bid + ask) / 2
        spread = ask - bid

        self.prices[instrument] = {"bid": bid, "ask": ask, "mid": mid_price, "spread": spread, "timestamp": timestamp}

        # Add to history
        self.price_history[instrument].append(mid_price)
        self.timestamps[instrument].append(timestamp)

        # Update statistics
        self._update_statistics(instrument)

        # Trigger callbacks
        for callback in self.callbacks:
            try:
                callback(instrument, self.prices[instrument])
            except Exception as e:
                print(f"Callback error: {e}")

    def _update_statistics(self, instrument: str):
        """Update real-time statistics for an instrument."""
        history = list(self.price_history[instrument])

        if len(history) < 2:
            return

        # Calculate statistics
        prices = np.array(history)
        returns = np.diff(prices) / prices[:-1] * 100  # Percentage returns

        self.statistics[instrument] = {
            "count": len(history),
            "current_price": history[-1],
            "min_price": prices.min(),
            "max_price": prices.max(),
            "price_range": prices.max() - prices.min(),
            "mean_price": prices.mean(),
            "std_price": prices.std(),
            "last_return": returns[-1] if len(returns) > 0 else 0,
            "volatility": returns.std() if len(returns) > 1 else 0,
            "mean_return": returns.mean() if len(returns) > 0 else 0,
            "price_change": history[-1] - history[0] if len(history) > 1 else 0,
            "price_change_pct": ((history[-1] - history[0]) / history[0] * 100) if len(history) > 1 and history[0] != 0 else 0,
        }

    def get_recent_prices(self, instrument: str, count: int = 100) -> pd.DataFrame:
        """Get recent price history as DataFrame."""
        if instrument not in self.price_history:
            return pd.DataFrame()

        history = list(self.price_history[instrument])[-count:]
        timestamps = list(self.timestamps[instrument])[-count:]

        if not history:
            return pd.DataFrame()

        df = pd.DataFrame({"price": history, "timestamp": timestamps})
        df.set_index("timestamp", inplace=True)
        return df

    def add_callback(self, callback: Callable[[str, dict], None]):
        """Add a callback function to be called on price updates."""
        self.callbacks.append(callback)

    def get_statistics(self, instrument: str) -> dict:
        """Get current statistics for an instrument."""
        return self.statistics.get(instrument, {})

    def get_current_price(self, instrument: str) -> dict:
        """Get current price information for an instrument."""
        return self.prices.get(instrument, {})


print("✅ Real-time data buffer system defined")

class RealTimeDataBuffer: """High-performance real-time data buffer with analytics.""" def __init__(self, max_size: int = 1000): self.max_size = max_size self.prices = {} self.price_history = {} self.timestamps = {} self.statistics = {} self.callbacks = [] def add_price(self, instrument: str, bid: float, ask: float, timestamp: datetime = None): """Add a new price tick to the buffer.""" if timestamp is None: timestamp = datetime.now(timezone.utc) # Initialize instrument data structures if needed if instrument not in self.prices: self.prices[instrument] = {"bid": bid, "ask": ask, "mid": (bid + ask) / 2} self.price_history[instrument] = deque(maxlen=self.max_size) self.timestamps[instrument] = deque(maxlen=self.max_size) self.statistics[instrument] = {} # Update current prices mid_price = (bid + ask) / 2 spread = ask - bid self.prices[instrument] = {"bid": bid, "ask": ask, "mid": mid_price, "spread": spread, "timestamp": timestamp} # Add to history self.price_history[instrument].append(mid_price) self.timestamps[instrument].append(timestamp) # Update statistics self._update_statistics(instrument) # Trigger callbacks for callback in self.callbacks: try: callback(instrument, self.prices[instrument]) except Exception as e: print(f"Callback error: {e}") def _update_statistics(self, instrument: str): """Update real-time statistics for an instrument.""" history = list(self.price_history[instrument]) if len(history) < 2: return # Calculate statistics prices = np.array(history) returns = np.diff(prices) / prices[:-1] * 100 # Percentage returns self.statistics[instrument] = { "count": len(history), "current_price": history[-1], "min_price": prices.min(), "max_price": prices.max(), "price_range": prices.max() - prices.min(), "mean_price": prices.mean(), "std_price": prices.std(), "last_return": returns[-1] if len(returns) > 0 else 0, "volatility": returns.std() if len(returns) > 1 else 0, "mean_return": returns.mean() if len(returns) > 0 else 0, "price_change": history[-1] - history[0] if len(history) > 1 else 0, "price_change_pct": ((history[-1] - history[0]) / history[0] * 100) if len(history) > 1 and history[0] != 0 else 0, } def get_recent_prices(self, instrument: str, count: int = 100) -> pd.DataFrame: """Get recent price history as DataFrame.""" if instrument not in self.price_history: return pd.DataFrame() history = list(self.price_history[instrument])[-count:] timestamps = list(self.timestamps[instrument])[-count:] if not history: return pd.DataFrame() df = pd.DataFrame({"price": history, "timestamp": timestamps}) df.set_index("timestamp", inplace=True) return df def add_callback(self, callback: Callable[[str, dict], None]): """Add a callback function to be called on price updates.""" self.callbacks.append(callback) def get_statistics(self, instrument: str) -> dict: """Get current statistics for an instrument.""" return self.statistics.get(instrument, {}) def get_current_price(self, instrument: str) -> dict: """Get current price information for an instrument.""" return self.prices.get(instrument, {}) print("✅ Real-time data buffer system defined")

Live Trading Signal Generator¶

Let's create a real-time signal generation system:

In [ ]:

class LiveSignalGenerator:
    """Real-time trading signal generator."""

    def __init__(self, data_buffer: RealTimeDataBuffer):
        self.data_buffer = data_buffer
        self.signals = {}
        self.signal_history = {}
        self.moving_averages = {}

        # Signal parameters
        self.ma_short_period = 10
        self.ma_long_period = 20
        self.rsi_period = 14
        self.volatility_threshold = 0.5  # Percentage

        # Register callback
        self.data_buffer.add_callback(self._on_price_update)

    def _on_price_update(self, instrument: str, price_data: dict):
        """Process new price data and generate signals."""
        try:
            # Get recent price history
            df = self.data_buffer.get_recent_prices(instrument, self.ma_long_period + 5)

            if len(df) < self.ma_long_period:
                return  # Not enough data

            # Calculate indicators
            signals = self._calculate_signals(instrument, df, price_data)

            # Store signals
            self.signals[instrument] = signals

            # Add to signal history
            if instrument not in self.signal_history:
                self.signal_history[instrument] = deque(maxlen=100)

            self.signal_history[instrument].append({"timestamp": price_data["timestamp"], "price": price_data["mid"], "signals": signals.copy()})

        except Exception as e:
            print(f"Signal generation error for {instrument}: {e}")

    def _calculate_signals(self, instrument: str, df: pd.DataFrame, current_price_data: dict) -> dict:
        """Calculate trading signals based on current data."""
        signals = {"timestamp": current_price_data["timestamp"], "price": current_price_data["mid"], "spread": current_price_data["spread"]}

        prices = df["price"].values

        # Moving averages
        if len(prices) >= self.ma_long_period:
            short_ma = prices[-self.ma_short_period :].mean()
            long_ma = prices[-self.ma_long_period :].mean()

            signals["short_ma"] = short_ma
            signals["long_ma"] = long_ma
            signals["ma_signal"] = "BUY" if short_ma > long_ma else "SELL"
            signals["ma_strength"] = abs(short_ma - long_ma) / long_ma * 100

        # RSI calculation
        if len(prices) >= self.rsi_period + 1:
            rsi = self._calculate_rsi(prices, self.rsi_period)
            signals["rsi"] = rsi

            if rsi < 30:
                signals["rsi_signal"] = "BUY"  # Oversold
            elif rsi > 70:
                signals["rsi_signal"] = "SELL"  # Overbought
            else:
                signals["rsi_signal"] = "HOLD"

        # Volatility analysis
        if len(prices) >= 10:
            returns = np.diff(prices[-10:]) / prices[-10:-1] * 100
            volatility = returns.std()
            signals["volatility"] = volatility
            signals["high_volatility"] = volatility > self.volatility_threshold

        # Momentum
        if len(prices) >= 5:
            momentum = (prices[-1] - prices[-5]) / prices[-5] * 100
            signals["momentum"] = momentum
            signals["momentum_signal"] = "BUY" if momentum > 0.1 else "SELL" if momentum < -0.1 else "HOLD"

        # Composite signal
        signals["composite_signal"] = self._generate_composite_signal(signals)

        return signals

    def _calculate_rsi(self, prices: np.ndarray, period: int) -> float:
        """Calculate RSI for the given prices."""
        deltas = np.diff(prices)
        gains = np.where(deltas > 0, deltas, 0)
        losses = np.where(deltas < 0, -deltas, 0)

        avg_gain = gains[-period:].mean()
        avg_loss = losses[-period:].mean()

        if avg_loss == 0:
            return 100

        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        return rsi

    def _generate_composite_signal(self, signals: dict) -> str:
        """Generate a composite signal based on multiple indicators."""
        buy_votes = 0
        sell_votes = 0
        total_votes = 0

        # MA signal
        if "ma_signal" in signals:
            if signals["ma_signal"] == "BUY":
                buy_votes += 1
            elif signals["ma_signal"] == "SELL":
                sell_votes += 1
            total_votes += 1

        # RSI signal
        if "rsi_signal" in signals:
            if signals["rsi_signal"] == "BUY":
                buy_votes += 1
            elif signals["rsi_signal"] == "SELL":
                sell_votes += 1
            total_votes += 1

        # Momentum signal
        if "momentum_signal" in signals:
            if signals["momentum_signal"] == "BUY":
                buy_votes += 1
            elif signals["momentum_signal"] == "SELL":
                sell_votes += 1
            total_votes += 1

        # Determine composite signal
        if total_votes == 0:
            return "HOLD"

        buy_ratio = buy_votes / total_votes
        sell_ratio = sell_votes / total_votes

        if buy_ratio >= 0.6:  # 60% consensus for buy
            return "BUY"
        if sell_ratio >= 0.6:  # 60% consensus for sell
            return "SELL"
        return "HOLD"

    def get_current_signals(self, instrument: str) -> dict:
        """Get current signals for an instrument."""
        return self.signals.get(instrument, {})

    def get_signal_history(self, instrument: str) -> list[dict]:
        """Get signal history for an instrument."""
        return list(self.signal_history.get(instrument, []))


print("✅ Live signal generator defined")

class LiveSignalGenerator: """Real-time trading signal generator.""" def __init__(self, data_buffer: RealTimeDataBuffer): self.data_buffer = data_buffer self.signals = {} self.signal_history = {} self.moving_averages = {} # Signal parameters self.ma_short_period = 10 self.ma_long_period = 20 self.rsi_period = 14 self.volatility_threshold = 0.5 # Percentage # Register callback self.data_buffer.add_callback(self._on_price_update) def _on_price_update(self, instrument: str, price_data: dict): """Process new price data and generate signals.""" try: # Get recent price history df = self.data_buffer.get_recent_prices(instrument, self.ma_long_period + 5) if len(df) < self.ma_long_period: return # Not enough data # Calculate indicators signals = self._calculate_signals(instrument, df, price_data) # Store signals self.signals[instrument] = signals # Add to signal history if instrument not in self.signal_history: self.signal_history[instrument] = deque(maxlen=100) self.signal_history[instrument].append({"timestamp": price_data["timestamp"], "price": price_data["mid"], "signals": signals.copy()}) except Exception as e: print(f"Signal generation error for {instrument}: {e}") def _calculate_signals(self, instrument: str, df: pd.DataFrame, current_price_data: dict) -> dict: """Calculate trading signals based on current data.""" signals = {"timestamp": current_price_data["timestamp"], "price": current_price_data["mid"], "spread": current_price_data["spread"]} prices = df["price"].values # Moving averages if len(prices) >= self.ma_long_period: short_ma = prices[-self.ma_short_period :].mean() long_ma = prices[-self.ma_long_period :].mean() signals["short_ma"] = short_ma signals["long_ma"] = long_ma signals["ma_signal"] = "BUY" if short_ma > long_ma else "SELL" signals["ma_strength"] = abs(short_ma - long_ma) / long_ma * 100 # RSI calculation if len(prices) >= self.rsi_period + 1: rsi = self._calculate_rsi(prices, self.rsi_period) signals["rsi"] = rsi if rsi < 30: signals["rsi_signal"] = "BUY" # Oversold elif rsi > 70: signals["rsi_signal"] = "SELL" # Overbought else: signals["rsi_signal"] = "HOLD" # Volatility analysis if len(prices) >= 10: returns = np.diff(prices[-10:]) / prices[-10:-1] * 100 volatility = returns.std() signals["volatility"] = volatility signals["high_volatility"] = volatility > self.volatility_threshold # Momentum if len(prices) >= 5: momentum = (prices[-1] - prices[-5]) / prices[-5] * 100 signals["momentum"] = momentum signals["momentum_signal"] = "BUY" if momentum > 0.1 else "SELL" if momentum < -0.1 else "HOLD" # Composite signal signals["composite_signal"] = self._generate_composite_signal(signals) return signals def _calculate_rsi(self, prices: np.ndarray, period: int) -> float: """Calculate RSI for the given prices.""" deltas = np.diff(prices) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) avg_gain = gains[-period:].mean() avg_loss = losses[-period:].mean() if avg_loss == 0: return 100 rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return rsi def _generate_composite_signal(self, signals: dict) -> str: """Generate a composite signal based on multiple indicators.""" buy_votes = 0 sell_votes = 0 total_votes = 0 # MA signal if "ma_signal" in signals: if signals["ma_signal"] == "BUY": buy_votes += 1 elif signals["ma_signal"] == "SELL": sell_votes += 1 total_votes += 1 # RSI signal if "rsi_signal" in signals: if signals["rsi_signal"] == "BUY": buy_votes += 1 elif signals["rsi_signal"] == "SELL": sell_votes += 1 total_votes += 1 # Momentum signal if "momentum_signal" in signals: if signals["momentum_signal"] == "BUY": buy_votes += 1 elif signals["momentum_signal"] == "SELL": sell_votes += 1 total_votes += 1 # Determine composite signal if total_votes == 0: return "HOLD" buy_ratio = buy_votes / total_votes sell_ratio = sell_votes / total_votes if buy_ratio >= 0.6: # 60% consensus for buy return "BUY" if sell_ratio >= 0.6: # 60% consensus for sell return "SELL" return "HOLD" def get_current_signals(self, instrument: str) -> dict: """Get current signals for an instrument.""" return self.signals.get(instrument, {}) def get_signal_history(self, instrument: str) -> list[dict]: """Get signal history for an instrument.""" return list(self.signal_history.get(instrument, [])) print("✅ Live signal generator defined")

Real-time Market Monitor¶

Let's create a market monitoring system:

In [ ]:

class RealTimeMarketMonitor:
    """Real-time market monitoring and alerting system."""

    def __init__(self, client: AsyncClient, account_id: str, instruments: list[str]):
        self.client = client
        self.account_id = account_id
        self.instruments = instruments
        self.data_buffer = RealTimeDataBuffer(max_size=1000)
        self.signal_generator = LiveSignalGenerator(self.data_buffer)

        # Monitoring state
        self.is_running = False
        self.start_time = None
        self.price_count = 0
        self.heartbeat_count = 0
        self.last_heartbeat = None

        # Alert settings
        self.price_alerts = {}  # {instrument: {'upper': price, 'lower': price}}
        self.volatility_alerts = {}  # {instrument: threshold}
        self.signal_alerts = True  # Alert on strong signals

        # Performance tracking
        self.performance_stats = {"total_ticks": 0, "ticks_per_second": 0, "instruments_active": set(), "connection_issues": 0, "last_update": None}

    async def start_monitoring(self, duration_seconds: int | None = None):
        """Start real-time market monitoring."""
        print(f"🚀 Starting real-time monitoring for {len(self.instruments)} instruments...")
        print(f"Instruments: {', '.join(self.instruments)}")

        self.is_running = True
        self.start_time = datetime.now(timezone.utc)

        try:
            # Configure streaming with reconnection policy
            config = StreamingConfiguration(
                stall_timeout=60.0,
                reconnection_policy=ReconnectionPolicy(max_attempts=10, delay_seconds=1.0),
            )

            # Start streaming with timeout
            start_time = time.time()

            async for event, state in self.client.pricing.stream_pricing_with_retries(account_id=self.account_id, instruments=self.instruments, config=config):
                # Check if duration limit reached
                if duration_seconds and (time.time() - start_time) > duration_seconds:
                    print(f"\n⏰ Monitoring duration ({duration_seconds}s) completed")
                    break

                if state == StreamState.RECONNECTING:
                    self.performance_stats["connection_issues"] += 1

                await self._process_streaming_event(event)

                # Stop if requested
                if not self.is_running:
                    break

        except StreamStall as e:
            print(f"\n🚨 Stream stalled: {e}")
            self.performance_stats["connection_issues"] += 1
        except FiveTwentyError as e:
            print(f"\n❌ OANDA API Error: {e.message}")
            self.performance_stats["connection_issues"] += 1
        except Exception as e:
            print(f"\n❌ Unexpected error: {e}")
        finally:
            self.is_running = False
            self._print_final_statistics()

    async def _process_streaming_event(self, event):
        """Process a streaming event (price or heartbeat)."""
        current_time = datetime.now(timezone.utc)

        if event.type == "PRICE":
            self.price_count += 1
            self.performance_stats["total_ticks"] += 1
            self.performance_stats["instruments_active"].add(event.instrument)
            self.performance_stats["last_update"] = current_time

            # Extract price data
            if event.bids and event.asks:
                bid = float(event.bids[0].price)
                ask = float(event.asks[0].price)

                # Add to data buffer
                self.data_buffer.add_price(instrument=event.instrument, bid=bid, ask=ask, timestamp=current_time)

                # Check alerts
                await self._check_alerts(event.instrument, bid, ask)

                # Print periodic updates
                if self.price_count % 50 == 0:
                    await self._print_status_update()

        elif event.type == "HEARTBEAT":
            self.heartbeat_count += 1
            self.last_heartbeat = current_time

            if self.heartbeat_count % 10 == 0:
                print(f"💓 Heartbeat #{self.heartbeat_count} at {current_time.strftime('%H:%M:%S')}")

    async def _check_alerts(self, instrument: str, bid: float, ask: float):
        """Check for price and signal alerts."""
        mid_price = (bid + ask) / 2

        # Price alerts
        if instrument in self.price_alerts:
            alerts = self.price_alerts[instrument]
            if "upper" in alerts and mid_price >= alerts["upper"]:
                print(f"🚨 PRICE ALERT: {instrument} hit upper limit {alerts['upper']:.5f} (current: {mid_price:.5f})")
            elif "lower" in alerts and mid_price <= alerts["lower"]:
                print(f"🚨 PRICE ALERT: {instrument} hit lower limit {alerts['lower']:.5f} (current: {mid_price:.5f})")

        # Signal alerts
        if self.signal_alerts:
            signals = self.signal_generator.get_current_signals(instrument)
            if signals and signals.get("composite_signal") in ["BUY", "SELL"] and signals.get("ma_strength", 0) > 0.05:  # Strong signal
                print(f"📈 SIGNAL ALERT: {instrument} - {signals['composite_signal']} (Price: {mid_price:.5f}, Strength: {signals['ma_strength']:.3f}%)")

    async def _print_status_update(self):
        """Print current monitoring status."""
        elapsed = (datetime.now(timezone.utc) - self.start_time).total_seconds()
        ticks_per_second = self.price_count / elapsed if elapsed > 0 else 0

        print(f"\n📊 Status Update (Tick #{self.price_count}):")
        print(f"  Elapsed: {elapsed:.1f}s | Rate: {ticks_per_second:.2f} ticks/sec")
        print(f"  Heartbeats: {self.heartbeat_count} | Active Instruments: {len(self.performance_stats['instruments_active'])}")

        # Show current prices
        print("  Current Prices:")
        for instrument in self.instruments:
            price_data = self.data_buffer.get_current_price(instrument)
            if price_data:
                signals = self.signal_generator.get_current_signals(instrument)
                signal_str = f" [{signals.get('composite_signal', 'N/A')}]" if signals else ""
                print(f"    {instrument}: {price_data['bid']:.5f}/{price_data['ask']:.5f} (Spread: {price_data['spread']:.5f}){signal_str}")

    def _print_final_statistics(self):
        """Print final monitoring statistics."""
        if not self.start_time:
            return

        elapsed = (datetime.now(timezone.utc) - self.start_time).total_seconds()
        avg_ticks_per_second = self.price_count / elapsed if elapsed > 0 else 0

        print("\n📊 MONITORING SUMMARY:")
        print(f"  Duration: {elapsed:.1f} seconds")
        print(f"  Total Price Ticks: {self.price_count:,}")
        print(f"  Total Heartbeats: {self.heartbeat_count}")
        print(f"  Average Rate: {avg_ticks_per_second:.2f} ticks/second")
        print(f"  Active Instruments: {len(self.performance_stats['instruments_active'])}")
        print(f"  Connection Issues: {self.performance_stats['connection_issues']}")

        # Show final statistics for each instrument
        print("\n📈 Final Statistics by Instrument:")
        for instrument in self.instruments:
            stats = self.data_buffer.get_statistics(instrument)
            if stats:
                print(f"  {instrument}:")
                print(f"    Ticks: {stats['count']} | Range: {stats['price_range']:.5f}")
                print(f"    Change: {stats['price_change']:+.5f} ({stats['price_change_pct']:+.3f}%)")
                print(f"    Volatility: {stats['volatility']:.4f}%")

    def stop_monitoring(self):
        """Stop the monitoring process."""
        print("\n🛑 Stopping monitoring...")
        self.is_running = False

    def set_price_alert(self, instrument: str, upper: float | None = None, lower: float | None = None):
        """Set price alerts for an instrument."""
        if instrument not in self.price_alerts:
            self.price_alerts[instrument] = {}

        if upper is not None:
            self.price_alerts[instrument]["upper"] = upper
        if lower is not None:
            self.price_alerts[instrument]["lower"] = lower

        print(f"🔔 Price alert set for {instrument}: Upper={upper}, Lower={lower}")


print("✅ Real-time market monitor defined")

class RealTimeMarketMonitor: """Real-time market monitoring and alerting system.""" def __init__(self, client: AsyncClient, account_id: str, instruments: list[str]): self.client = client self.account_id = account_id self.instruments = instruments self.data_buffer = RealTimeDataBuffer(max_size=1000) self.signal_generator = LiveSignalGenerator(self.data_buffer) # Monitoring state self.is_running = False self.start_time = None self.price_count = 0 self.heartbeat_count = 0 self.last_heartbeat = None # Alert settings self.price_alerts = {} # {instrument: {'upper': price, 'lower': price}} self.volatility_alerts = {} # {instrument: threshold} self.signal_alerts = True # Alert on strong signals # Performance tracking self.performance_stats = {"total_ticks": 0, "ticks_per_second": 0, "instruments_active": set(), "connection_issues": 0, "last_update": None} async def start_monitoring(self, duration_seconds: int | None = None): """Start real-time market monitoring.""" print(f"🚀 Starting real-time monitoring for {len(self.instruments)} instruments...") print(f"Instruments: {', '.join(self.instruments)}") self.is_running = True self.start_time = datetime.now(timezone.utc) try: # Configure streaming with reconnection policy config = StreamingConfiguration( stall_timeout=60.0, reconnection_policy=ReconnectionPolicy(max_attempts=10, delay_seconds=1.0), ) # Start streaming with timeout start_time = time.time() async for event, state in self.client.pricing.stream_pricing_with_retries(account_id=self.account_id, instruments=self.instruments, config=config): # Check if duration limit reached if duration_seconds and (time.time() - start_time) > duration_seconds: print(f"\n⏰ Monitoring duration ({duration_seconds}s) completed") break if state == StreamState.RECONNECTING: self.performance_stats["connection_issues"] += 1 await self._process_streaming_event(event) # Stop if requested if not self.is_running: break except StreamStall as e: print(f"\n🚨 Stream stalled: {e}") self.performance_stats["connection_issues"] += 1 except FiveTwentyError as e: print(f"\n❌ OANDA API Error: {e.message}") self.performance_stats["connection_issues"] += 1 except Exception as e: print(f"\n❌ Unexpected error: {e}") finally: self.is_running = False self._print_final_statistics() async def _process_streaming_event(self, event): """Process a streaming event (price or heartbeat).""" current_time = datetime.now(timezone.utc) if event.type == "PRICE": self.price_count += 1 self.performance_stats["total_ticks"] += 1 self.performance_stats["instruments_active"].add(event.instrument) self.performance_stats["last_update"] = current_time # Extract price data if event.bids and event.asks: bid = float(event.bids[0].price) ask = float(event.asks[0].price) # Add to data buffer self.data_buffer.add_price(instrument=event.instrument, bid=bid, ask=ask, timestamp=current_time) # Check alerts await self._check_alerts(event.instrument, bid, ask) # Print periodic updates if self.price_count % 50 == 0: await self._print_status_update() elif event.type == "HEARTBEAT": self.heartbeat_count += 1 self.last_heartbeat = current_time if self.heartbeat_count % 10 == 0: print(f"💓 Heartbeat #{self.heartbeat_count} at {current_time.strftime('%H:%M:%S')}") async def _check_alerts(self, instrument: str, bid: float, ask: float): """Check for price and signal alerts.""" mid_price = (bid + ask) / 2 # Price alerts if instrument in self.price_alerts: alerts = self.price_alerts[instrument] if "upper" in alerts and mid_price >= alerts["upper"]: print(f"🚨 PRICE ALERT: {instrument} hit upper limit {alerts['upper']:.5f} (current: {mid_price:.5f})") elif "lower" in alerts and mid_price <= alerts["lower"]: print(f"🚨 PRICE ALERT: {instrument} hit lower limit {alerts['lower']:.5f} (current: {mid_price:.5f})") # Signal alerts if self.signal_alerts: signals = self.signal_generator.get_current_signals(instrument) if signals and signals.get("composite_signal") in ["BUY", "SELL"] and signals.get("ma_strength", 0) > 0.05: # Strong signal print(f"📈 SIGNAL ALERT: {instrument} - {signals['composite_signal']} (Price: {mid_price:.5f}, Strength: {signals['ma_strength']:.3f}%)") async def _print_status_update(self): """Print current monitoring status.""" elapsed = (datetime.now(timezone.utc) - self.start_time).total_seconds() ticks_per_second = self.price_count / elapsed if elapsed > 0 else 0 print(f"\n📊 Status Update (Tick #{self.price_count}):") print(f" Elapsed: {elapsed:.1f}s | Rate: {ticks_per_second:.2f} ticks/sec") print(f" Heartbeats: {self.heartbeat_count} | Active Instruments: {len(self.performance_stats['instruments_active'])}") # Show current prices print(" Current Prices:") for instrument in self.instruments: price_data = self.data_buffer.get_current_price(instrument) if price_data: signals = self.signal_generator.get_current_signals(instrument) signal_str = f" [{signals.get('composite_signal', 'N/A')}]" if signals else "" print(f" {instrument}: {price_data['bid']:.5f}/{price_data['ask']:.5f} (Spread: {price_data['spread']:.5f}){signal_str}") def _print_final_statistics(self): """Print final monitoring statistics.""" if not self.start_time: return elapsed = (datetime.now(timezone.utc) - self.start_time).total_seconds() avg_ticks_per_second = self.price_count / elapsed if elapsed > 0 else 0 print("\n📊 MONITORING SUMMARY:") print(f" Duration: {elapsed:.1f} seconds") print(f" Total Price Ticks: {self.price_count:,}") print(f" Total Heartbeats: {self.heartbeat_count}") print(f" Average Rate: {avg_ticks_per_second:.2f} ticks/second") print(f" Active Instruments: {len(self.performance_stats['instruments_active'])}") print(f" Connection Issues: {self.performance_stats['connection_issues']}") # Show final statistics for each instrument print("\n📈 Final Statistics by Instrument:") for instrument in self.instruments: stats = self.data_buffer.get_statistics(instrument) if stats: print(f" {instrument}:") print(f" Ticks: {stats['count']} | Range: {stats['price_range']:.5f}") print(f" Change: {stats['price_change']:+.5f} ({stats['price_change_pct']:+.3f}%)") print(f" Volatility: {stats['volatility']:.4f}%") def stop_monitoring(self): """Stop the monitoring process.""" print("\n🛑 Stopping monitoring...") self.is_running = False def set_price_alert(self, instrument: str, upper: float | None = None, lower: float | None = None): """Set price alerts for an instrument.""" if instrument not in self.price_alerts: self.price_alerts[instrument] = {} if upper is not None: self.price_alerts[instrument]["upper"] = upper if lower is not None: self.price_alerts[instrument]["lower"] = lower print(f"🔔 Price alert set for {instrument}: Upper={upper}, Lower={lower}") print("✅ Real-time market monitor defined")

Initialize Connection¶

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()

Basic Streaming Example¶

Let's start with a simple streaming example:

In [ ]:

async def basic_streaming_demo(account_id: str, duration_seconds: int = 30):
    """Basic streaming demonstration."""

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

    print(f"📡 Starting basic streaming demo for {duration_seconds} seconds...")

    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        try:
            start_time = time.time()
            price_count = 0
            heartbeat_count = 0

            async for event in client.pricing.get_pricing_stream(account_id=account_id, instruments=["EUR_USD", "GBP_USD"]):
                elapsed = time.time() - start_time

                if elapsed > duration_seconds:
                    print(f"\n⏰ Demo completed after {duration_seconds} seconds")
                    break

                if event.type == "PRICE":
                    price_count += 1

                    if event.bids and event.asks:
                        bid = float(event.bids[0].price)
                        ask = float(event.asks[0].price)
                        spread = ask - bid

                        if price_count <= 10 or price_count % 10 == 0:
                            print(f"📈 {event.instrument}: {bid:.5f}/{ask:.5f} (Spread: {spread:.5f}) [{price_count}]")

                elif event.type == "HEARTBEAT":
                    heartbeat_count += 1
                    if heartbeat_count % 5 == 0:
                        print(f"💓 Heartbeat #{heartbeat_count} at {event.time}")

            print("\n📊 Demo Summary:")
            print(f"  Duration: {elapsed:.1f} seconds")
            print(f"  Price updates: {price_count}")
            print(f"  Heartbeats: {heartbeat_count}")
            print(f"  Average rate: {price_count / elapsed:.2f} ticks/second")

        except Exception as e:
            print(f"❌ Streaming error: {e}")


# Run basic demo
print("🎬 Running Basic Streaming Demo...")
print("(This will stream live prices for 15 seconds)")
await basic_streaming_demo(account_id, duration_seconds=15)

async def basic_streaming_demo(account_id: str, duration_seconds: int = 30): """Basic streaming demonstration.""" if not account_id: print("❌ No account connection") return print(f"📡 Starting basic streaming demo for {duration_seconds} seconds...") async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: try: start_time = time.time() price_count = 0 heartbeat_count = 0 async for event in client.pricing.get_pricing_stream(account_id=account_id, instruments=["EUR_USD", "GBP_USD"]): elapsed = time.time() - start_time if elapsed > duration_seconds: print(f"\n⏰ Demo completed after {duration_seconds} seconds") break if event.type == "PRICE": price_count += 1 if event.bids and event.asks: bid = float(event.bids[0].price) ask = float(event.asks[0].price) spread = ask - bid if price_count <= 10 or price_count % 10 == 0: print(f"📈 {event.instrument}: {bid:.5f}/{ask:.5f} (Spread: {spread:.5f}) [{price_count}]") elif event.type == "HEARTBEAT": heartbeat_count += 1 if heartbeat_count % 5 == 0: print(f"💓 Heartbeat #{heartbeat_count} at {event.time}") print("\n📊 Demo Summary:") print(f" Duration: {elapsed:.1f} seconds") print(f" Price updates: {price_count}") print(f" Heartbeats: {heartbeat_count}") print(f" Average rate: {price_count / elapsed:.2f} ticks/second") except Exception as e: print(f"❌ Streaming error: {e}") # Run basic demo print("🎬 Running Basic Streaming Demo...") print("(This will stream live prices for 15 seconds)") await basic_streaming_demo(account_id, duration_seconds=15)

Advanced Real-time Monitoring¶

In [ ]:

if account_id:
    # Create advanced monitor
    instruments_to_monitor = ["EUR_USD", "GBP_USD", "USD_JPY"]

    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        monitor = RealTimeMarketMonitor(client, account_id, instruments_to_monitor)

        # Set some price alerts (example prices - adjust based on current market)
        monitor.set_price_alert("EUR_USD", upper=1.2000, lower=1.0500)
        monitor.set_price_alert("GBP_USD", upper=1.3500, lower=1.2000)

        print("🚀 Starting Advanced Market Monitor...")
        print("This will monitor live market data with real-time signals and alerts")
        print("Duration: 30 seconds")
        print("\nReal-time data will appear below:")
        print("=" * 60)

        # Run monitoring for 30 seconds
        await monitor.start_monitoring(duration_seconds=30)
else:
    print("❌ No account connection - cannot run advanced monitoring")

if account_id: # Create advanced monitor instruments_to_monitor = ["EUR_USD", "GBP_USD", "USD_JPY"] async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: monitor = RealTimeMarketMonitor(client, account_id, instruments_to_monitor) # Set some price alerts (example prices - adjust based on current market) monitor.set_price_alert("EUR_USD", upper=1.2000, lower=1.0500) monitor.set_price_alert("GBP_USD", upper=1.3500, lower=1.2000) print("🚀 Starting Advanced Market Monitor...") print("This will monitor live market data with real-time signals and alerts") print("Duration: 30 seconds") print("\nReal-time data will appear below:") print("=" * 60) # Run monitoring for 30 seconds await monitor.start_monitoring(duration_seconds=30) else: print("❌ No account connection - cannot run advanced monitoring")

Real-time Signal Analysis¶

In [ ]:

def analyze_recent_signals(monitor: RealTimeMarketMonitor, instrument: str):
    """Analyze recent signals for an instrument."""

    print(f"\n📊 Signal Analysis for {instrument}:")

    # Get current signals
    current_signals = monitor.signal_generator.get_current_signals(instrument)

    if not current_signals:
        print("  No signals available yet")
        return

    print(f"  Current Price: {current_signals.get('price', 'N/A'):.5f}")
    print(f"  Spread: {current_signals.get('spread', 'N/A'):.5f}")
    print(f"  Short MA: {current_signals.get('short_ma', 'N/A'):.5f}")
    print(f"  Long MA: {current_signals.get('long_ma', 'N/A'):.5f}")
    print(f"  RSI: {current_signals.get('rsi', 'N/A'):.2f}")
    print(f"  Volatility: {current_signals.get('volatility', 'N/A'):.4f}%")
    print(f"  Momentum: {current_signals.get('momentum', 'N/A'):+.4f}%")
    print("  ")
    print("  🎯 Signals:")
    print(f"    MA Signal: {current_signals.get('ma_signal', 'N/A')}")
    print(f"    RSI Signal: {current_signals.get('rsi_signal', 'N/A')}")
    print(f"    Momentum Signal: {current_signals.get('momentum_signal', 'N/A')}")
    print(f"    🔥 COMPOSITE: {current_signals.get('composite_signal', 'N/A')}")

    # Get signal history
    signal_history = monitor.signal_generator.get_signal_history(instrument)

    if len(signal_history) > 5:
        print("\n  📈 Recent Signal Changes:")
        for i, signal_data in enumerate(signal_history[-5:]):
            timestamp = signal_data["timestamp"].strftime("%H:%M:%S")
            composite = signal_data["signals"].get("composite_signal", "N/A")
            price = signal_data["price"]
            print(f"    {timestamp}: {composite} @ {price:.5f}")

    # Get price statistics
    stats = monitor.data_buffer.get_statistics(instrument)
    if stats:
        print("\n  📊 Price Statistics:")
        print(f"    Ticks Received: {stats['count']}")
        print(f"    Price Range: {stats['price_range']:.5f}")
        print(f"    Total Change: {stats['price_change']:+.5f} ({stats['price_change_pct']:+.3f}%)")
        print(f"    Current Volatility: {stats['volatility']:.4f}%")


# Analyze signals if we have monitor data
if "monitor" in locals() and account_id:
    for instrument in instruments_to_monitor:
        analyze_recent_signals(monitor, instrument)
else:
    print("⚠️ Run the advanced monitoring cell first to analyze signals")

def analyze_recent_signals(monitor: RealTimeMarketMonitor, instrument: str): """Analyze recent signals for an instrument.""" print(f"\n📊 Signal Analysis for {instrument}:") # Get current signals current_signals = monitor.signal_generator.get_current_signals(instrument) if not current_signals: print(" No signals available yet") return print(f" Current Price: {current_signals.get('price', 'N/A'):.5f}") print(f" Spread: {current_signals.get('spread', 'N/A'):.5f}") print(f" Short MA: {current_signals.get('short_ma', 'N/A'):.5f}") print(f" Long MA: {current_signals.get('long_ma', 'N/A'):.5f}") print(f" RSI: {current_signals.get('rsi', 'N/A'):.2f}") print(f" Volatility: {current_signals.get('volatility', 'N/A'):.4f}%") print(f" Momentum: {current_signals.get('momentum', 'N/A'):+.4f}%") print(" ") print(" 🎯 Signals:") print(f" MA Signal: {current_signals.get('ma_signal', 'N/A')}") print(f" RSI Signal: {current_signals.get('rsi_signal', 'N/A')}") print(f" Momentum Signal: {current_signals.get('momentum_signal', 'N/A')}") print(f" 🔥 COMPOSITE: {current_signals.get('composite_signal', 'N/A')}") # Get signal history signal_history = monitor.signal_generator.get_signal_history(instrument) if len(signal_history) > 5: print("\n 📈 Recent Signal Changes:") for i, signal_data in enumerate(signal_history[-5:]): timestamp = signal_data["timestamp"].strftime("%H:%M:%S") composite = signal_data["signals"].get("composite_signal", "N/A") price = signal_data["price"] print(f" {timestamp}: {composite} @ {price:.5f}") # Get price statistics stats = monitor.data_buffer.get_statistics(instrument) if stats: print("\n 📊 Price Statistics:") print(f" Ticks Received: {stats['count']}") print(f" Price Range: {stats['price_range']:.5f}") print(f" Total Change: {stats['price_change']:+.5f} ({stats['price_change_pct']:+.3f}%)") print(f" Current Volatility: {stats['volatility']:.4f}%") # Analyze signals if we have monitor data if "monitor" in locals() and account_id: for instrument in instruments_to_monitor: analyze_recent_signals(monitor, instrument) else: print("⚠️ Run the advanced monitoring cell first to analyze signals")

Real-time Data Visualization¶

In [ ]:

def plot_real_time_data(monitor: RealTimeMarketMonitor, instrument: str):
    """Create plots of real-time data."""

    # Get recent price data
    df = monitor.data_buffer.get_recent_prices(instrument, count=200)

    if df.empty:
        print(f"No data available for {instrument}")
        return

    # Get signal history
    signal_history = monitor.signal_generator.get_signal_history(instrument)

    # Create plots
    fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(15, 12))

    # Price chart with signals
    ax1.plot(df.index, df["price"], label="Price", linewidth=1, alpha=0.8)

    # Add signal markers
    if signal_history:
        buy_signals = []
        sell_signals = []
        buy_times = []
        sell_times = []

        for signal_data in signal_history:
            composite = signal_data["signals"].get("composite_signal")
            if composite == "BUY":
                buy_signals.append(signal_data["price"])
                buy_times.append(signal_data["timestamp"])
            elif composite == "SELL":
                sell_signals.append(signal_data["price"])
                sell_times.append(signal_data["timestamp"])

        if buy_signals:
            ax1.scatter(buy_times, buy_signals, color="green", marker="^", s=50, alpha=0.7, label="BUY Signals")
        if sell_signals:
            ax1.scatter(sell_times, sell_signals, color="red", marker="v", s=50, alpha=0.7, label="SELL Signals")

    ax1.set_title(f"{instrument} - Real-time Price with Signals")
    ax1.set_ylabel("Price")
    ax1.legend()
    ax1.grid(True, alpha=0.3)

    # Price returns distribution
    returns = df["price"].pct_change().dropna() * 100
    ax2.hist(returns, bins=30, alpha=0.7, edgecolor="black")
    ax2.axvline(x=returns.mean(), color="red", linestyle="--", label=f"Mean: {returns.mean():.4f}%")
    ax2.set_title("Price Returns Distribution")
    ax2.set_xlabel("Return (%)")
    ax2.set_ylabel("Frequency")
    ax2.legend()
    ax2.grid(True, alpha=0.3)

    # Rolling volatility
    rolling_vol = returns.rolling(window=20).std()
    ax3.plot(df.index[1:], rolling_vol, label="Rolling Volatility (20 periods)", color="orange")
    ax3.axhline(y=rolling_vol.mean(), color="red", linestyle="--", label=f"Average: {rolling_vol.mean():.4f}%")
    ax3.set_title("Rolling Volatility")
    ax3.set_xlabel("Time")
    ax3.set_ylabel("Volatility (%)")
    ax3.legend()
    ax3.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

    # Print statistics
    print(f"\n📊 Real-time Data Summary for {instrument}:")
    print(f"  Data Points: {len(df)}")
    print(f"  Time Span: {df.index[0].strftime('%H:%M:%S')} to {df.index[-1].strftime('%H:%M:%S')}")
    print(f"  Price Range: {df['price'].min():.5f} - {df['price'].max():.5f}")
    print(f"  Total Change: {(df['price'].iloc[-1] - df['price'].iloc[0]):.5f}")
    print(f"  Average Return: {returns.mean():.4f}%")
    print(f"  Volatility: {returns.std():.4f}%")
    print(f"  Signal Count: BUY={len(buy_signals) if 'buy_signals' in locals() else 0}, SELL={len(sell_signals) if 'sell_signals' in locals() else 0}")


# Create visualizations if we have monitor data
if "monitor" in locals() and account_id:
    # Plot data for the first instrument
    plot_real_time_data(monitor, instruments_to_monitor[0])
else:
    print("⚠️ Run the advanced monitoring cell first to visualize data")

def plot_real_time_data(monitor: RealTimeMarketMonitor, instrument: str): """Create plots of real-time data.""" # Get recent price data df = monitor.data_buffer.get_recent_prices(instrument, count=200) if df.empty: print(f"No data available for {instrument}") return # Get signal history signal_history = monitor.signal_generator.get_signal_history(instrument) # Create plots fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(15, 12)) # Price chart with signals ax1.plot(df.index, df["price"], label="Price", linewidth=1, alpha=0.8) # Add signal markers if signal_history: buy_signals = [] sell_signals = [] buy_times = [] sell_times = [] for signal_data in signal_history: composite = signal_data["signals"].get("composite_signal") if composite == "BUY": buy_signals.append(signal_data["price"]) buy_times.append(signal_data["timestamp"]) elif composite == "SELL": sell_signals.append(signal_data["price"]) sell_times.append(signal_data["timestamp"]) if buy_signals: ax1.scatter(buy_times, buy_signals, color="green", marker="^", s=50, alpha=0.7, label="BUY Signals") if sell_signals: ax1.scatter(sell_times, sell_signals, color="red", marker="v", s=50, alpha=0.7, label="SELL Signals") ax1.set_title(f"{instrument} - Real-time Price with Signals") ax1.set_ylabel("Price") ax1.legend() ax1.grid(True, alpha=0.3) # Price returns distribution returns = df["price"].pct_change().dropna() * 100 ax2.hist(returns, bins=30, alpha=0.7, edgecolor="black") ax2.axvline(x=returns.mean(), color="red", linestyle="--", label=f"Mean: {returns.mean():.4f}%") ax2.set_title("Price Returns Distribution") ax2.set_xlabel("Return (%)") ax2.set_ylabel("Frequency") ax2.legend() ax2.grid(True, alpha=0.3) # Rolling volatility rolling_vol = returns.rolling(window=20).std() ax3.plot(df.index[1:], rolling_vol, label="Rolling Volatility (20 periods)", color="orange") ax3.axhline(y=rolling_vol.mean(), color="red", linestyle="--", label=f"Average: {rolling_vol.mean():.4f}%") ax3.set_title("Rolling Volatility") ax3.set_xlabel("Time") ax3.set_ylabel("Volatility (%)") ax3.legend() ax3.grid(True, alpha=0.3) plt.tight_layout() plt.show() # Print statistics print(f"\n📊 Real-time Data Summary for {instrument}:") print(f" Data Points: {len(df)}") print(f" Time Span: {df.index[0].strftime('%H:%M:%S')} to {df.index[-1].strftime('%H:%M:%S')}") print(f" Price Range: {df['price'].min():.5f} - {df['price'].max():.5f}") print(f" Total Change: {(df['price'].iloc[-1] - df['price'].iloc[0]):.5f}") print(f" Average Return: {returns.mean():.4f}%") print(f" Volatility: {returns.std():.4f}%") print(f" Signal Count: BUY={len(buy_signals) if 'buy_signals' in locals() else 0}, SELL={len(sell_signals) if 'sell_signals' in locals() else 0}") # Create visualizations if we have monitor data if "monitor" in locals() and account_id: # Plot data for the first instrument plot_real_time_data(monitor, instruments_to_monitor[0]) else: print("⚠️ Run the advanced monitoring cell first to visualize data")

Live Trading Simulation¶

Let's create a paper trading simulation using real-time data:

In [ ]:

class LiveTradingSimulator:
    """Paper trading simulator using real-time data."""

    def __init__(self, initial_balance: float = 10000):
        self.initial_balance = initial_balance
        self.balance = initial_balance
        self.positions = {}  # {instrument: {'units': int, 'entry_price': float, 'entry_time': datetime}}
        self.trades = []
        self.equity_history = []

        # Trading parameters
        self.max_position_size = 1000
        self.min_signal_strength = 0.05  # Minimum MA strength for trading

    def process_signal(self, instrument: str, price_data: dict, signals: dict):
        """Process a trading signal and execute trades."""

        current_price = price_data["mid"]
        current_time = price_data["timestamp"]
        composite_signal = signals.get("composite_signal", "HOLD")
        ma_strength = signals.get("ma_strength", 0)

        # Check if we have a position
        has_position = instrument in self.positions

        # Entry logic
        if not has_position and composite_signal in ["BUY", "SELL"]:
            # Check signal strength
            if ma_strength >= self.min_signal_strength:
                units = self.max_position_size if composite_signal == "BUY" else -self.max_position_size

                self.positions[instrument] = {"units": units, "entry_price": current_price, "entry_time": current_time}

                print(f"🎯 TRADE ENTRY: {composite_signal} {abs(units)} {instrument} @ {current_price:.5f}")

        # Exit logic
        elif has_position:
            position = self.positions[instrument]

            # Check for opposite signal or weak signal
            should_exit = False

            if (position["units"] > 0 and composite_signal == "SELL") or (position["units"] < 0 and composite_signal == "BUY"):
                should_exit = True
            elif composite_signal == "HOLD" and ma_strength < self.min_signal_strength / 2:
                should_exit = True  # Exit on weak signals

            if should_exit:
                self._close_position(instrument, current_price, current_time)

        # Update equity
        self._update_equity(current_time)

    def _close_position(self, instrument: str, exit_price: float, exit_time: datetime):
        """Close a position and record the trade."""

        if instrument not in self.positions:
            return

        position = self.positions[instrument]
        entry_price = position["entry_price"]
        units = position["units"]
        entry_time = position["entry_time"]

        # Calculate P/L
        if units > 0:  # Long position
            pnl = (exit_price - entry_price) * units
        else:  # Short position
            pnl = (entry_price - exit_price) * abs(units)

        # Update balance
        self.balance += pnl

        # Record trade
        trade = {"instrument": instrument, "entry_time": entry_time, "exit_time": exit_time, "entry_price": entry_price, "exit_price": exit_price, "units": units, "pnl": pnl, "duration_minutes": (exit_time - entry_time).total_seconds() / 60}

        self.trades.append(trade)

        # Remove position
        del self.positions[instrument]

        direction = "LONG" if units > 0 else "SHORT"
        pnl_str = f"{pnl:+.2f}"
        print(f"🎯 TRADE EXIT: Close {direction} {abs(units)} {instrument} @ {exit_price:.5f} | P/L: ${pnl_str}")

    def _update_equity(self, timestamp: datetime):
        """Update equity including unrealized P/L."""
        # For simplicity, we'll just track balance (realized P/L)
        # In a real implementation, you'd calculate unrealized P/L for open positions

        self.equity_history.append({"timestamp": timestamp, "balance": self.balance, "open_positions": len(self.positions)})

    def get_performance_summary(self) -> dict:
        """Get trading performance summary."""

        if not self.trades:
            return {"error": "No trades executed"}

        total_trades = len(self.trades)
        winning_trades = len([t for t in self.trades if t["pnl"] > 0])
        total_pnl = sum(t["pnl"] for t in self.trades)

        return {
            "total_trades": total_trades,
            "winning_trades": winning_trades,
            "win_rate": (winning_trades / total_trades * 100) if total_trades > 0 else 0,
            "total_pnl": total_pnl,
            "total_return": (total_pnl / self.initial_balance * 100),
            "average_trade": total_pnl / total_trades if total_trades > 0 else 0,
            "current_balance": self.balance,
            "open_positions": len(self.positions),
        }


# Create simulator and add callback if we have monitor
if "monitor" in locals() and account_id:
    simulator = LiveTradingSimulator(initial_balance=10000)

    # Add trading callback
    def trading_callback(instrument: str, price_data: dict):
        signals = monitor.signal_generator.get_current_signals(instrument)
        if signals:
            simulator.process_signal(instrument, price_data, signals)

    monitor.data_buffer.add_callback(trading_callback)

    print("✅ Live trading simulator connected to real-time data feed")
    print("Trading signals will be automatically processed during streaming")
else:
    print("⚠️ Advanced monitoring needs to be running to use the trading simulator")

class LiveTradingSimulator: """Paper trading simulator using real-time data.""" def __init__(self, initial_balance: float = 10000): self.initial_balance = initial_balance self.balance = initial_balance self.positions = {} # {instrument: {'units': int, 'entry_price': float, 'entry_time': datetime}} self.trades = [] self.equity_history = [] # Trading parameters self.max_position_size = 1000 self.min_signal_strength = 0.05 # Minimum MA strength for trading def process_signal(self, instrument: str, price_data: dict, signals: dict): """Process a trading signal and execute trades.""" current_price = price_data["mid"] current_time = price_data["timestamp"] composite_signal = signals.get("composite_signal", "HOLD") ma_strength = signals.get("ma_strength", 0) # Check if we have a position has_position = instrument in self.positions # Entry logic if not has_position and composite_signal in ["BUY", "SELL"]: # Check signal strength if ma_strength >= self.min_signal_strength: units = self.max_position_size if composite_signal == "BUY" else -self.max_position_size self.positions[instrument] = {"units": units, "entry_price": current_price, "entry_time": current_time} print(f"🎯 TRADE ENTRY: {composite_signal} {abs(units)} {instrument} @ {current_price:.5f}") # Exit logic elif has_position: position = self.positions[instrument] # Check for opposite signal or weak signal should_exit = False if (position["units"] > 0 and composite_signal == "SELL") or (position["units"] < 0 and composite_signal == "BUY"): should_exit = True elif composite_signal == "HOLD" and ma_strength < self.min_signal_strength / 2: should_exit = True # Exit on weak signals if should_exit: self._close_position(instrument, current_price, current_time) # Update equity self._update_equity(current_time) def _close_position(self, instrument: str, exit_price: float, exit_time: datetime): """Close a position and record the trade.""" if instrument not in self.positions: return position = self.positions[instrument] entry_price = position["entry_price"] units = position["units"] entry_time = position["entry_time"] # Calculate P/L if units > 0: # Long position pnl = (exit_price - entry_price) * units else: # Short position pnl = (entry_price - exit_price) * abs(units) # Update balance self.balance += pnl # Record trade trade = {"instrument": instrument, "entry_time": entry_time, "exit_time": exit_time, "entry_price": entry_price, "exit_price": exit_price, "units": units, "pnl": pnl, "duration_minutes": (exit_time - entry_time).total_seconds() / 60} self.trades.append(trade) # Remove position del self.positions[instrument] direction = "LONG" if units > 0 else "SHORT" pnl_str = f"{pnl:+.2f}" print(f"🎯 TRADE EXIT: Close {direction} {abs(units)} {instrument} @ {exit_price:.5f} | P/L: ${pnl_str}") def _update_equity(self, timestamp: datetime): """Update equity including unrealized P/L.""" # For simplicity, we'll just track balance (realized P/L) # In a real implementation, you'd calculate unrealized P/L for open positions self.equity_history.append({"timestamp": timestamp, "balance": self.balance, "open_positions": len(self.positions)}) def get_performance_summary(self) -> dict: """Get trading performance summary.""" if not self.trades: return {"error": "No trades executed"} total_trades = len(self.trades) winning_trades = len([t for t in self.trades if t["pnl"] > 0]) total_pnl = sum(t["pnl"] for t in self.trades) return { "total_trades": total_trades, "winning_trades": winning_trades, "win_rate": (winning_trades / total_trades * 100) if total_trades > 0 else 0, "total_pnl": total_pnl, "total_return": (total_pnl / self.initial_balance * 100), "average_trade": total_pnl / total_trades if total_trades > 0 else 0, "current_balance": self.balance, "open_positions": len(self.positions), } # Create simulator and add callback if we have monitor if "monitor" in locals() and account_id: simulator = LiveTradingSimulator(initial_balance=10000) # Add trading callback def trading_callback(instrument: str, price_data: dict): signals = monitor.signal_generator.get_current_signals(instrument) if signals: simulator.process_signal(instrument, price_data, signals) monitor.data_buffer.add_callback(trading_callback) print("✅ Live trading simulator connected to real-time data feed") print("Trading signals will be automatically processed during streaming") else: print("⚠️ Advanced monitoring needs to be running to use the trading simulator")

Trading Performance Analysis¶

In [ ]:

if "simulator" in locals():
    # Get performance summary
    performance = simulator.get_performance_summary()

    print("🏆 LIVE TRADING SIMULATION RESULTS:")
    print("=" * 50)

    if "error" in performance:
        print(f"  {performance['error']}")
        print("  No trades were executed during the simulation.")
        print("  This could be due to:")
        print("    - Short monitoring duration")
        print("    - No strong trading signals")
        print("    - Conservative signal thresholds")
    else:
        print("  📊 Trading Statistics:")
        print(f"    Total Trades: {performance['total_trades']}")
        print(f"    Winning Trades: {performance['winning_trades']}")
        print(f"    Win Rate: {performance['win_rate']:.1f}%")
        print(f"    Total P/L: ${performance['total_pnl']:+.2f}")
        print(f"    Total Return: {performance['total_return']:+.2f}%")
        print(f"    Average Trade: ${performance['average_trade']:+.2f}")
        print(f"    Final Balance: ${performance['current_balance']:.2f}")
        print(f"    Open Positions: {performance['open_positions']}")

        # Show individual trades
        if simulator.trades:
            print("\n  📋 Individual Trades:")
            for i, trade in enumerate(simulator.trades, 1):
                direction = "LONG" if trade["units"] > 0 else "SHORT"
                pnl_str = f"${trade['pnl']:+.2f}"
                duration = f"{trade['duration_minutes']:.1f}min"
                print(f"    {i}. {trade['instrument']} {direction} {abs(trade['units'])} units: {trade['entry_price']:.5f} → {trade['exit_price']:.5f} ({duration}) = {pnl_str}")

        # Plot equity curve if we have trade data
        if simulator.equity_history:
            equity_df = pd.DataFrame(simulator.equity_history)

            plt.figure(figsize=(12, 6))
            plt.plot(equity_df["timestamp"], equity_df["balance"], linewidth=2, label="Account Balance")
            plt.axhline(y=simulator.initial_balance, color="gray", linestyle="--", alpha=0.7, label="Initial Balance")

            plt.title("Live Trading Simulation - Equity Curve")
            plt.xlabel("Time")
            plt.ylabel("Balance ($)")
            plt.legend()
            plt.grid(True, alpha=0.3)
            plt.tight_layout()
            plt.show()
else:
    print("⚠️ No trading simulator data available")
    print("Run the Live Trading Simulation cell first")

if "simulator" in locals(): # Get performance summary performance = simulator.get_performance_summary() print("🏆 LIVE TRADING SIMULATION RESULTS:") print("=" * 50) if "error" in performance: print(f" {performance['error']}") print(" No trades were executed during the simulation.") print(" This could be due to:") print(" - Short monitoring duration") print(" - No strong trading signals") print(" - Conservative signal thresholds") else: print(" 📊 Trading Statistics:") print(f" Total Trades: {performance['total_trades']}") print(f" Winning Trades: {performance['winning_trades']}") print(f" Win Rate: {performance['win_rate']:.1f}%") print(f" Total P/L: ${performance['total_pnl']:+.2f}") print(f" Total Return: {performance['total_return']:+.2f}%") print(f" Average Trade: ${performance['average_trade']:+.2f}") print(f" Final Balance: ${performance['current_balance']:.2f}") print(f" Open Positions: {performance['open_positions']}") # Show individual trades if simulator.trades: print("\n 📋 Individual Trades:") for i, trade in enumerate(simulator.trades, 1): direction = "LONG" if trade["units"] > 0 else "SHORT" pnl_str = f"${trade['pnl']:+.2f}" duration = f"{trade['duration_minutes']:.1f}min" print(f" {i}. {trade['instrument']} {direction} {abs(trade['units'])} units: {trade['entry_price']:.5f} → {trade['exit_price']:.5f} ({duration}) = {pnl_str}") # Plot equity curve if we have trade data if simulator.equity_history: equity_df = pd.DataFrame(simulator.equity_history) plt.figure(figsize=(12, 6)) plt.plot(equity_df["timestamp"], equity_df["balance"], linewidth=2, label="Account Balance") plt.axhline(y=simulator.initial_balance, color="gray", linestyle="--", alpha=0.7, label="Initial Balance") plt.title("Live Trading Simulation - Equity Curve") plt.xlabel("Time") plt.ylabel("Balance ($)") plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() else: print("⚠️ No trading simulator data available") print("Run the Live Trading Simulation cell first")

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