Historical data analysis¶
Retrieve candles, calculate indicators, compare illustrative backtests and plot results. This is an analysis exercise, not a validated execution simulator. Inspect candle completeness, timing, currency units and cost assumptions before interpreting any reported return.
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 [ ]:
Copied!
import os
from datetime import datetime
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
# Try to import plotly for interactive charts
try:
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
PLOTLY_AVAILABLE = True
except ImportError:
print("Install plotly for interactive charts: uv add plotly")
PLOTLY_AVAILABLE = False
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")
# Set style for plots
plt.style.use("seaborn-v0_8")
sns.set_palette("husl")
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 datetime import datetime
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
# Try to import plotly for interactive charts
try:
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
PLOTLY_AVAILABLE = True
except ImportError:
print("Install plotly for interactive charts: uv add plotly")
PLOTLY_AVAILABLE = False
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")
# Set style for plots
plt.style.use("seaborn-v0_8")
sns.set_palette("husl")
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")
Data Collection Framework¶
Let's create a data collection class:
In [ ]:
Copied!
class FiveTwentyDataCollector:
"""Comprehensive data collection and analysis framework."""
def __init__(self, client: AsyncClient, account_id: str):
self.client = client
self.account_id = account_id
async def get_historical_data(self, instrument: str, granularity: CandlestickGranularity, count: int | None = None, from_time: str | None = None, to_time: str | None = None) -> pd.DataFrame:
"""Get historical candlestick data."""
try:
kwargs = {"instrument": instrument, "granularity": granularity}
if count:
kwargs["count"] = count
if from_time:
kwargs["from_time"] = from_time
if to_time:
kwargs["to_time"] = to_time
candles = await self.client.instruments.get_instrument_candles(**kwargs)
# 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), "complete": candle.complete})
df = pd.DataFrame(data)
if not df.empty:
df.set_index("time", inplace=True)
df.sort_index(inplace=True)
print(f"✅ Retrieved {len(df)} candles for {instrument}")
return df
except FiveTwentyError as e:
print(f"❌ Error getting data: {e.message}")
return pd.DataFrame()
async def get_multiple_instruments(self, instruments: list[str], granularity: CandlestickGranularity, count: int = 500) -> dict[str, pd.DataFrame]:
"""Get data for multiple instruments."""
data_dict = {}
for instrument in instruments:
print(f"Fetching data for {instrument}...")
df = await self.get_historical_data(instrument, granularity, count=count)
if not df.empty:
data_dict[instrument] = df
return data_dict
def add_technical_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
"""Add common technical indicators to the dataframe."""
df = df.copy()
# Moving Averages
df["sma_20"] = df["close"].rolling(window=20).mean()
df["sma_50"] = df["close"].rolling(window=50).mean()
df["ema_12"] = df["close"].ewm(span=12).mean()
df["ema_26"] = df["close"].ewm(span=26).mean()
# MACD
df["macd"] = df["ema_12"] - df["ema_26"]
df["macd_signal"] = df["macd"].ewm(span=9).mean()
df["macd_histogram"] = df["macd"] - df["macd_signal"]
# RSI
delta = df["close"].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df["rsi"] = 100 - (100 / (1 + rs))
# Bollinger Bands
bb_period = 20
df["bb_middle"] = df["close"].rolling(window=bb_period).mean()
bb_std = df["close"].rolling(window=bb_period).std()
df["bb_upper"] = df["bb_middle"] + (bb_std * 2)
df["bb_lower"] = df["bb_middle"] - (bb_std * 2)
df["bb_width"] = df["bb_upper"] - df["bb_lower"]
df["bb_position"] = (df["close"] - df["bb_lower"]) / (df["bb_upper"] - df["bb_lower"])
# Average True Range (ATR)
high_low = df["high"] - df["low"]
high_close = np.abs(df["high"] - df["close"].shift())
low_close = np.abs(df["low"] - df["close"].shift())
true_range = np.maximum(high_low, np.maximum(high_close, low_close))
df["atr"] = true_range.rolling(window=14).mean()
# Price changes
df["price_change"] = df["close"].diff()
df["price_change_pct"] = df["close"].pct_change() * 100
# Support and Resistance levels (simplified)
df["resistance"] = df["high"].rolling(window=20).max()
df["support"] = df["low"].rolling(window=20).min()
return df
print("✅ Data collection framework defined")
class FiveTwentyDataCollector:
"""Comprehensive data collection and analysis framework."""
def __init__(self, client: AsyncClient, account_id: str):
self.client = client
self.account_id = account_id
async def get_historical_data(self, instrument: str, granularity: CandlestickGranularity, count: int | None = None, from_time: str | None = None, to_time: str | None = None) -> pd.DataFrame:
"""Get historical candlestick data."""
try:
kwargs = {"instrument": instrument, "granularity": granularity}
if count:
kwargs["count"] = count
if from_time:
kwargs["from_time"] = from_time
if to_time:
kwargs["to_time"] = to_time
candles = await self.client.instruments.get_instrument_candles(**kwargs)
# 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), "complete": candle.complete})
df = pd.DataFrame(data)
if not df.empty:
df.set_index("time", inplace=True)
df.sort_index(inplace=True)
print(f"✅ Retrieved {len(df)} candles for {instrument}")
return df
except FiveTwentyError as e:
print(f"❌ Error getting data: {e.message}")
return pd.DataFrame()
async def get_multiple_instruments(self, instruments: list[str], granularity: CandlestickGranularity, count: int = 500) -> dict[str, pd.DataFrame]:
"""Get data for multiple instruments."""
data_dict = {}
for instrument in instruments:
print(f"Fetching data for {instrument}...")
df = await self.get_historical_data(instrument, granularity, count=count)
if not df.empty:
data_dict[instrument] = df
return data_dict
def add_technical_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
"""Add common technical indicators to the dataframe."""
df = df.copy()
# Moving Averages
df["sma_20"] = df["close"].rolling(window=20).mean()
df["sma_50"] = df["close"].rolling(window=50).mean()
df["ema_12"] = df["close"].ewm(span=12).mean()
df["ema_26"] = df["close"].ewm(span=26).mean()
# MACD
df["macd"] = df["ema_12"] - df["ema_26"]
df["macd_signal"] = df["macd"].ewm(span=9).mean()
df["macd_histogram"] = df["macd"] - df["macd_signal"]
# RSI
delta = df["close"].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df["rsi"] = 100 - (100 / (1 + rs))
# Bollinger Bands
bb_period = 20
df["bb_middle"] = df["close"].rolling(window=bb_period).mean()
bb_std = df["close"].rolling(window=bb_period).std()
df["bb_upper"] = df["bb_middle"] + (bb_std * 2)
df["bb_lower"] = df["bb_middle"] - (bb_std * 2)
df["bb_width"] = df["bb_upper"] - df["bb_lower"]
df["bb_position"] = (df["close"] - df["bb_lower"]) / (df["bb_upper"] - df["bb_lower"])
# Average True Range (ATR)
high_low = df["high"] - df["low"]
high_close = np.abs(df["high"] - df["close"].shift())
low_close = np.abs(df["low"] - df["close"].shift())
true_range = np.maximum(high_low, np.maximum(high_close, low_close))
df["atr"] = true_range.rolling(window=14).mean()
# Price changes
df["price_change"] = df["close"].diff()
df["price_change_pct"] = df["close"].pct_change() * 100
# Support and Resistance levels (simplified)
df["resistance"] = df["high"].rolling(window=20).max()
df["support"] = df["low"].rolling(window=20).min()
return df
print("✅ Data collection framework defined")
Backtesting Framework¶
The following engine simulates trades under the assumptions implemented in its methods. Check price timing, costs and sizing before interpreting its output.
In [ ]:
Copied!
class BacktestEngine:
"""Comprehensive backtesting framework."""
def __init__(self, initial_balance: float = 10000):
self.initial_balance = initial_balance
self.reset()
def reset(self):
"""Reset backtest to initial state."""
self.balance = self.initial_balance
self.equity = self.initial_balance
self.trades = []
self.positions = []
self.equity_curve = []
self.drawdown_curve = []
self.peak_balance = self.initial_balance
def add_trade(self, entry_time: datetime, exit_time: datetime, entry_price: float, exit_price: float, units: int, instrument: str, strategy: str = "Unknown"):
"""Add a completed trade to the backtest."""
# Calculate P/L
if units > 0: # Long position
pnl = (exit_price - entry_price) * units
else: # Short position
pnl = (entry_price - exit_price) * abs(units)
# Calculate percentage return
position_value = abs(units * entry_price)
pnl_percentage = (pnl / position_value) * 100 if position_value > 0 else 0
# Update balance
self.balance += pnl
self.equity = self.balance
# Track peak for drawdown calculation
self.peak_balance = max(self.peak_balance, self.equity)
# Calculate drawdown
drawdown = ((self.peak_balance - self.equity) / self.peak_balance) * 100
trade = {
"entry_time": entry_time,
"exit_time": exit_time,
"instrument": instrument,
"strategy": strategy,
"entry_price": entry_price,
"exit_price": exit_price,
"units": units,
"direction": "LONG" if units > 0 else "SHORT",
"pnl": pnl,
"pnl_percentage": pnl_percentage,
"balance_after": self.balance,
"drawdown": drawdown,
"duration_hours": (exit_time - entry_time).total_seconds() / 3600,
}
self.trades.append(trade)
self.equity_curve.append({"time": exit_time, "equity": self.equity, "drawdown": drawdown})
def calculate_performance_metrics(self) -> dict:
"""Calculate comprehensive performance metrics."""
if not self.trades:
return {"error": "No trades to analyze"}
trades_df = pd.DataFrame(self.trades)
# Basic metrics
total_trades = len(self.trades)
winning_trades = len(trades_df[trades_df["pnl"] > 0])
losing_trades = len(trades_df[trades_df["pnl"] < 0])
win_rate = (winning_trades / total_trades * 100) if total_trades > 0 else 0
# P/L metrics
total_pnl = trades_df["pnl"].sum()
avg_win = trades_df[trades_df["pnl"] > 0]["pnl"].mean() if winning_trades > 0 else 0
avg_loss = trades_df[trades_df["pnl"] < 0]["pnl"].mean() if losing_trades > 0 else 0
profit_factor = abs(avg_win * winning_trades / (avg_loss * losing_trades)) if avg_loss != 0 and losing_trades > 0 else float("inf")
# Return metrics
total_return = ((self.balance - self.initial_balance) / self.initial_balance) * 100
# Drawdown metrics
drawdowns = trades_df["drawdown"]
max_drawdown = drawdowns.max() if not drawdowns.empty else 0
# Sharpe ratio (simplified - assuming risk-free rate of 0)
returns = trades_df["pnl_percentage"]
sharpe_ratio = returns.mean() / returns.std() if returns.std() > 0 else 0
# Average trade duration
avg_duration = trades_df["duration_hours"].mean()
return {
"total_trades": total_trades,
"winning_trades": winning_trades,
"losing_trades": losing_trades,
"win_rate": win_rate,
"total_pnl": total_pnl,
"total_return": total_return,
"avg_win": avg_win,
"avg_loss": avg_loss,
"profit_factor": profit_factor,
"max_drawdown": max_drawdown,
"sharpe_ratio": sharpe_ratio,
"avg_duration_hours": avg_duration,
"initial_balance": self.initial_balance,
"final_balance": self.balance,
}
def get_equity_curve_df(self) -> pd.DataFrame:
"""Get equity curve as DataFrame."""
if not self.equity_curve:
return pd.DataFrame()
df = pd.DataFrame(self.equity_curve)
df.set_index("time", inplace=True)
return df
print("✅ Backtesting framework defined")
class BacktestEngine:
"""Comprehensive backtesting framework."""
def __init__(self, initial_balance: float = 10000):
self.initial_balance = initial_balance
self.reset()
def reset(self):
"""Reset backtest to initial state."""
self.balance = self.initial_balance
self.equity = self.initial_balance
self.trades = []
self.positions = []
self.equity_curve = []
self.drawdown_curve = []
self.peak_balance = self.initial_balance
def add_trade(self, entry_time: datetime, exit_time: datetime, entry_price: float, exit_price: float, units: int, instrument: str, strategy: str = "Unknown"):
"""Add a completed trade to the backtest."""
# Calculate P/L
if units > 0: # Long position
pnl = (exit_price - entry_price) * units
else: # Short position
pnl = (entry_price - exit_price) * abs(units)
# Calculate percentage return
position_value = abs(units * entry_price)
pnl_percentage = (pnl / position_value) * 100 if position_value > 0 else 0
# Update balance
self.balance += pnl
self.equity = self.balance
# Track peak for drawdown calculation
self.peak_balance = max(self.peak_balance, self.equity)
# Calculate drawdown
drawdown = ((self.peak_balance - self.equity) / self.peak_balance) * 100
trade = {
"entry_time": entry_time,
"exit_time": exit_time,
"instrument": instrument,
"strategy": strategy,
"entry_price": entry_price,
"exit_price": exit_price,
"units": units,
"direction": "LONG" if units > 0 else "SHORT",
"pnl": pnl,
"pnl_percentage": pnl_percentage,
"balance_after": self.balance,
"drawdown": drawdown,
"duration_hours": (exit_time - entry_time).total_seconds() / 3600,
}
self.trades.append(trade)
self.equity_curve.append({"time": exit_time, "equity": self.equity, "drawdown": drawdown})
def calculate_performance_metrics(self) -> dict:
"""Calculate comprehensive performance metrics."""
if not self.trades:
return {"error": "No trades to analyze"}
trades_df = pd.DataFrame(self.trades)
# Basic metrics
total_trades = len(self.trades)
winning_trades = len(trades_df[trades_df["pnl"] > 0])
losing_trades = len(trades_df[trades_df["pnl"] < 0])
win_rate = (winning_trades / total_trades * 100) if total_trades > 0 else 0
# P/L metrics
total_pnl = trades_df["pnl"].sum()
avg_win = trades_df[trades_df["pnl"] > 0]["pnl"].mean() if winning_trades > 0 else 0
avg_loss = trades_df[trades_df["pnl"] < 0]["pnl"].mean() if losing_trades > 0 else 0
profit_factor = abs(avg_win * winning_trades / (avg_loss * losing_trades)) if avg_loss != 0 and losing_trades > 0 else float("inf")
# Return metrics
total_return = ((self.balance - self.initial_balance) / self.initial_balance) * 100
# Drawdown metrics
drawdowns = trades_df["drawdown"]
max_drawdown = drawdowns.max() if not drawdowns.empty else 0
# Sharpe ratio (simplified - assuming risk-free rate of 0)
returns = trades_df["pnl_percentage"]
sharpe_ratio = returns.mean() / returns.std() if returns.std() > 0 else 0
# Average trade duration
avg_duration = trades_df["duration_hours"].mean()
return {
"total_trades": total_trades,
"winning_trades": winning_trades,
"losing_trades": losing_trades,
"win_rate": win_rate,
"total_pnl": total_pnl,
"total_return": total_return,
"avg_win": avg_win,
"avg_loss": avg_loss,
"profit_factor": profit_factor,
"max_drawdown": max_drawdown,
"sharpe_ratio": sharpe_ratio,
"avg_duration_hours": avg_duration,
"initial_balance": self.initial_balance,
"final_balance": self.balance,
}
def get_equity_curve_df(self) -> pd.DataFrame:
"""Get equity curve as DataFrame."""
if not self.equity_curve:
return pd.DataFrame()
df = pd.DataFrame(self.equity_curve)
df.set_index("time", inplace=True)
return df
print("✅ Backtesting framework defined")
Initialize Connection and Get Data¶
In [ ]:
Copied!
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()
Historical Data Analysis¶
In [ ]:
Copied!
if account_id:
async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
collector = FiveTwentyDataCollector(client, account_id)
print("📊 Fetching historical data for EUR/USD...")
# Get 1000 hours of hourly data
eur_usd_data = await collector.get_historical_data(instrument="EUR_USD", granularity=CandlestickGranularity.H1, count=1000)
if not eur_usd_data.empty:
# Add technical indicators
eur_usd_data = collector.add_technical_indicators(eur_usd_data)
print("\n📈 Data Summary for EUR/USD:")
print(f" Period: {eur_usd_data.index[0]} to {eur_usd_data.index[-1]}")
print(f" Total candles: {len(eur_usd_data)}")
print(f" Price range: {eur_usd_data['low'].min():.5f} - {eur_usd_data['high'].max():.5f}")
print(f" Average volume: {eur_usd_data['volume'].mean():.0f}")
# Display first few rows
print("\n📋 Sample Data:")
print(eur_usd_data[["open", "high", "low", "close", "volume"]].head())
else:
print("❌ No account connection - cannot fetch data")
eur_usd_data = pd.DataFrame()
if account_id:
async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
collector = FiveTwentyDataCollector(client, account_id)
print("📊 Fetching historical data for EUR/USD...")
# Get 1000 hours of hourly data
eur_usd_data = await collector.get_historical_data(instrument="EUR_USD", granularity=CandlestickGranularity.H1, count=1000)
if not eur_usd_data.empty:
# Add technical indicators
eur_usd_data = collector.add_technical_indicators(eur_usd_data)
print("\n📈 Data Summary for EUR/USD:")
print(f" Period: {eur_usd_data.index[0]} to {eur_usd_data.index[-1]}")
print(f" Total candles: {len(eur_usd_data)}")
print(f" Price range: {eur_usd_data['low'].min():.5f} - {eur_usd_data['high'].max():.5f}")
print(f" Average volume: {eur_usd_data['volume'].mean():.0f}")
# Display first few rows
print("\n📋 Sample Data:")
print(eur_usd_data[["open", "high", "low", "close", "volume"]].head())
else:
print("❌ No account connection - cannot fetch data")
eur_usd_data = pd.DataFrame()
Market Analysis and Visualization¶
In [ ]:
Copied!
if not eur_usd_data.empty:
# Create comprehensive market analysis plots
fig, axes = plt.subplots(4, 1, figsize=(15, 16))
# 1. Price and Moving Averages
axes[0].plot(eur_usd_data.index, eur_usd_data["close"], label="Close Price", linewidth=1, alpha=0.8)
axes[0].plot(eur_usd_data.index, eur_usd_data["sma_20"], label="SMA 20", alpha=0.7)
axes[0].plot(eur_usd_data.index, eur_usd_data["sma_50"], label="SMA 50", alpha=0.7)
axes[0].fill_between(eur_usd_data.index, eur_usd_data["bb_lower"], eur_usd_data["bb_upper"], alpha=0.2, label="Bollinger Bands")
axes[0].set_title("EUR/USD Price Action with Technical Indicators")
axes[0].set_ylabel("Price")
axes[0].legend()
axes[0].grid(True, alpha=0.3)
# 2. RSI
axes[1].plot(eur_usd_data.index, eur_usd_data["rsi"], label="RSI", color="orange")
axes[1].axhline(y=70, color="r", linestyle="--", alpha=0.7, label="Overbought (70)")
axes[1].axhline(y=30, color="g", linestyle="--", alpha=0.7, label="Oversold (30)")
axes[1].fill_between(eur_usd_data.index, 30, 70, alpha=0.1, color="gray")
axes[1].set_title("Relative Strength Index (RSI)")
axes[1].set_ylabel("RSI")
axes[1].set_ylim(0, 100)
axes[1].legend()
axes[1].grid(True, alpha=0.3)
# 3. MACD
axes[2].plot(eur_usd_data.index, eur_usd_data["macd"], label="MACD", linewidth=1)
axes[2].plot(eur_usd_data.index, eur_usd_data["macd_signal"], label="Signal", linewidth=1)
axes[2].bar(eur_usd_data.index, eur_usd_data["macd_histogram"], label="Histogram", alpha=0.6, width=0.8)
axes[2].axhline(y=0, color="black", linestyle="-", alpha=0.3)
axes[2].set_title("MACD (Moving Average Convergence Divergence)")
axes[2].set_ylabel("MACD")
axes[2].legend()
axes[2].grid(True, alpha=0.3)
# 4. Volume and ATR
ax4_twin = axes[3].twinx()
axes[3].bar(eur_usd_data.index, eur_usd_data["volume"], alpha=0.6, label="Volume")
ax4_twin.plot(eur_usd_data.index, eur_usd_data["atr"], color="red", label="ATR", linewidth=2)
axes[3].set_title("Volume and Average True Range (ATR)")
axes[3].set_ylabel("Volume")
ax4_twin.set_ylabel("ATR")
axes[3].legend(loc="upper left")
ax4_twin.legend(loc="upper right")
axes[3].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Statistical analysis
print("\n📊 Statistical Analysis:")
returns = eur_usd_data["price_change_pct"].dropna()
print(f" Average daily return: {returns.mean():.4f}%")
print(f" Daily volatility: {returns.std():.4f}%")
print(f" Sharpe ratio: {returns.mean() / returns.std():.4f}")
print(f" Max daily gain: {returns.max():.4f}%")
print(f" Max daily loss: {returns.min():.4f}%")
print(f" Current RSI: {eur_usd_data['rsi'].iloc[-1]:.2f}")
print(f" Current MACD: {eur_usd_data['macd'].iloc[-1]:.6f}")
else:
print("❌ No data available for analysis")
if not eur_usd_data.empty:
# Create comprehensive market analysis plots
fig, axes = plt.subplots(4, 1, figsize=(15, 16))
# 1. Price and Moving Averages
axes[0].plot(eur_usd_data.index, eur_usd_data["close"], label="Close Price", linewidth=1, alpha=0.8)
axes[0].plot(eur_usd_data.index, eur_usd_data["sma_20"], label="SMA 20", alpha=0.7)
axes[0].plot(eur_usd_data.index, eur_usd_data["sma_50"], label="SMA 50", alpha=0.7)
axes[0].fill_between(eur_usd_data.index, eur_usd_data["bb_lower"], eur_usd_data["bb_upper"], alpha=0.2, label="Bollinger Bands")
axes[0].set_title("EUR/USD Price Action with Technical Indicators")
axes[0].set_ylabel("Price")
axes[0].legend()
axes[0].grid(True, alpha=0.3)
# 2. RSI
axes[1].plot(eur_usd_data.index, eur_usd_data["rsi"], label="RSI", color="orange")
axes[1].axhline(y=70, color="r", linestyle="--", alpha=0.7, label="Overbought (70)")
axes[1].axhline(y=30, color="g", linestyle="--", alpha=0.7, label="Oversold (30)")
axes[1].fill_between(eur_usd_data.index, 30, 70, alpha=0.1, color="gray")
axes[1].set_title("Relative Strength Index (RSI)")
axes[1].set_ylabel("RSI")
axes[1].set_ylim(0, 100)
axes[1].legend()
axes[1].grid(True, alpha=0.3)
# 3. MACD
axes[2].plot(eur_usd_data.index, eur_usd_data["macd"], label="MACD", linewidth=1)
axes[2].plot(eur_usd_data.index, eur_usd_data["macd_signal"], label="Signal", linewidth=1)
axes[2].bar(eur_usd_data.index, eur_usd_data["macd_histogram"], label="Histogram", alpha=0.6, width=0.8)
axes[2].axhline(y=0, color="black", linestyle="-", alpha=0.3)
axes[2].set_title("MACD (Moving Average Convergence Divergence)")
axes[2].set_ylabel("MACD")
axes[2].legend()
axes[2].grid(True, alpha=0.3)
# 4. Volume and ATR
ax4_twin = axes[3].twinx()
axes[3].bar(eur_usd_data.index, eur_usd_data["volume"], alpha=0.6, label="Volume")
ax4_twin.plot(eur_usd_data.index, eur_usd_data["atr"], color="red", label="ATR", linewidth=2)
axes[3].set_title("Volume and Average True Range (ATR)")
axes[3].set_ylabel("Volume")
ax4_twin.set_ylabel("ATR")
axes[3].legend(loc="upper left")
ax4_twin.legend(loc="upper right")
axes[3].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Statistical analysis
print("\n📊 Statistical Analysis:")
returns = eur_usd_data["price_change_pct"].dropna()
print(f" Average daily return: {returns.mean():.4f}%")
print(f" Daily volatility: {returns.std():.4f}%")
print(f" Sharpe ratio: {returns.mean() / returns.std():.4f}")
print(f" Max daily gain: {returns.max():.4f}%")
print(f" Max daily loss: {returns.min():.4f}%")
print(f" Current RSI: {eur_usd_data['rsi'].iloc[-1]:.2f}")
print(f" Current MACD: {eur_usd_data['macd'].iloc[-1]:.6f}")
else:
print("❌ No data available for analysis")
Simple Moving Average Strategy Backtest¶
In [ ]:
Copied!
def backtest_sma_strategy(df: pd.DataFrame, short_period: int = 20, long_period: int = 50) -> BacktestEngine:
"""Backtest a simple moving average crossover strategy."""
backtest = BacktestEngine(initial_balance=10000)
position = None # Current position: None, 'LONG', or 'SHORT'
entry_price = None
entry_time = None
# Calculate signals
df = df.copy()
df["short_ma"] = df["close"].rolling(window=short_period).mean()
df["long_ma"] = df["close"].rolling(window=long_period).mean()
# Generate signals
df["signal"] = 0
df.loc[df.index[short_period:], "signal"] = np.where(df["short_ma"].iloc[short_period:] > df["long_ma"].iloc[short_period:], 1, -1)
df["signal_change"] = df["signal"].diff()
for i, (timestamp, row) in enumerate(df.iterrows()):
if i < long_period: # Not enough data for signal
continue
current_signal = row["signal"]
signal_change = row["signal_change"]
current_price = row["close"]
# Entry signals
if position is None:
if signal_change == 2: # Short MA crosses above Long MA
position = "LONG"
entry_price = current_price
entry_time = timestamp
elif signal_change == -2: # Short MA crosses below Long MA
position = "SHORT"
entry_price = current_price
entry_time = timestamp
# Exit signals
elif position == "LONG" and signal_change == -2:
# Close long position
backtest.add_trade(
entry_time=entry_time,
exit_time=timestamp,
entry_price=entry_price,
exit_price=current_price,
units=1000, # Fixed position size
instrument="EUR_USD",
strategy="SMA_Crossover",
)
# Enter short position
position = "SHORT"
entry_price = current_price
entry_time = timestamp
elif position == "SHORT" and signal_change == 2:
# Close short position
backtest.add_trade(
entry_time=entry_time,
exit_time=timestamp,
entry_price=entry_price,
exit_price=current_price,
units=-1000, # Fixed position size
instrument="EUR_USD",
strategy="SMA_Crossover",
)
# Enter long position
position = "LONG"
entry_price = current_price
entry_time = timestamp
# Close any remaining position at the end
if position is not None:
final_price = df["close"].iloc[-1]
final_time = df.index[-1]
units = 1000 if position == "LONG" else -1000
backtest.add_trade(entry_time=entry_time, exit_time=final_time, entry_price=entry_price, exit_price=final_price, units=units, instrument="EUR_USD", strategy="SMA_Crossover")
return backtest
if not eur_usd_data.empty:
print("🔄 Running SMA Crossover Strategy Backtest...")
# Run backtest
sma_backtest = backtest_sma_strategy(eur_usd_data, short_period=20, long_period=50)
# Get performance metrics
metrics = sma_backtest.calculate_performance_metrics()
print("\n📊 SMA Crossover Strategy Results:")
print(f" Total Trades: {metrics['total_trades']}")
print(f" Win Rate: {metrics['win_rate']:.2f}%")
print(f" Total Return: {metrics['total_return']:.2f}%")
print(f" Profit Factor: {metrics['profit_factor']:.2f}")
print(f" Average Win: ${metrics['avg_win']:.2f}")
print(f" Average Loss: ${metrics['avg_loss']:.2f}")
print(f" Max Drawdown: {metrics['max_drawdown']:.2f}%")
print(f" Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f" Average Trade Duration: {metrics['avg_duration_hours']:.1f} hours")
print(f" Final Balance: ${metrics['final_balance']:.2f}")
else:
print("❌ No data available for backtesting")
def backtest_sma_strategy(df: pd.DataFrame, short_period: int = 20, long_period: int = 50) -> BacktestEngine:
"""Backtest a simple moving average crossover strategy."""
backtest = BacktestEngine(initial_balance=10000)
position = None # Current position: None, 'LONG', or 'SHORT'
entry_price = None
entry_time = None
# Calculate signals
df = df.copy()
df["short_ma"] = df["close"].rolling(window=short_period).mean()
df["long_ma"] = df["close"].rolling(window=long_period).mean()
# Generate signals
df["signal"] = 0
df.loc[df.index[short_period:], "signal"] = np.where(df["short_ma"].iloc[short_period:] > df["long_ma"].iloc[short_period:], 1, -1)
df["signal_change"] = df["signal"].diff()
for i, (timestamp, row) in enumerate(df.iterrows()):
if i < long_period: # Not enough data for signal
continue
current_signal = row["signal"]
signal_change = row["signal_change"]
current_price = row["close"]
# Entry signals
if position is None:
if signal_change == 2: # Short MA crosses above Long MA
position = "LONG"
entry_price = current_price
entry_time = timestamp
elif signal_change == -2: # Short MA crosses below Long MA
position = "SHORT"
entry_price = current_price
entry_time = timestamp
# Exit signals
elif position == "LONG" and signal_change == -2:
# Close long position
backtest.add_trade(
entry_time=entry_time,
exit_time=timestamp,
entry_price=entry_price,
exit_price=current_price,
units=1000, # Fixed position size
instrument="EUR_USD",
strategy="SMA_Crossover",
)
# Enter short position
position = "SHORT"
entry_price = current_price
entry_time = timestamp
elif position == "SHORT" and signal_change == 2:
# Close short position
backtest.add_trade(
entry_time=entry_time,
exit_time=timestamp,
entry_price=entry_price,
exit_price=current_price,
units=-1000, # Fixed position size
instrument="EUR_USD",
strategy="SMA_Crossover",
)
# Enter long position
position = "LONG"
entry_price = current_price
entry_time = timestamp
# Close any remaining position at the end
if position is not None:
final_price = df["close"].iloc[-1]
final_time = df.index[-1]
units = 1000 if position == "LONG" else -1000
backtest.add_trade(entry_time=entry_time, exit_time=final_time, entry_price=entry_price, exit_price=final_price, units=units, instrument="EUR_USD", strategy="SMA_Crossover")
return backtest
if not eur_usd_data.empty:
print("🔄 Running SMA Crossover Strategy Backtest...")
# Run backtest
sma_backtest = backtest_sma_strategy(eur_usd_data, short_period=20, long_period=50)
# Get performance metrics
metrics = sma_backtest.calculate_performance_metrics()
print("\n📊 SMA Crossover Strategy Results:")
print(f" Total Trades: {metrics['total_trades']}")
print(f" Win Rate: {metrics['win_rate']:.2f}%")
print(f" Total Return: {metrics['total_return']:.2f}%")
print(f" Profit Factor: {metrics['profit_factor']:.2f}")
print(f" Average Win: ${metrics['avg_win']:.2f}")
print(f" Average Loss: ${metrics['avg_loss']:.2f}")
print(f" Max Drawdown: {metrics['max_drawdown']:.2f}%")
print(f" Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f" Average Trade Duration: {metrics['avg_duration_hours']:.1f} hours")
print(f" Final Balance: ${metrics['final_balance']:.2f}")
else:
print("❌ No data available for backtesting")
RSI Mean Reversion Strategy Backtest¶
In [ ]:
Copied!
def backtest_rsi_strategy(df: pd.DataFrame, oversold: float = 30, overbought: float = 70) -> BacktestEngine:
"""Backtest an RSI mean reversion strategy."""
backtest = BacktestEngine(initial_balance=10000)
position = None
entry_price = None
entry_time = None
for i, (timestamp, row) in enumerate(df.iterrows()):
if i < 20: # Not enough data for RSI
continue
current_rsi = row["rsi"]
current_price = row["close"]
# Skip if RSI is NaN
if pd.isna(current_rsi):
continue
# Entry signals
if position is None:
if current_rsi < oversold: # Oversold - buy signal
position = "LONG"
entry_price = current_price
entry_time = timestamp
elif current_rsi > overbought: # Overbought - sell signal
position = "SHORT"
entry_price = current_price
entry_time = timestamp
# Exit signals
elif position == "LONG" and current_rsi > 50: # Exit long when RSI returns to middle
backtest.add_trade(entry_time=entry_time, exit_time=timestamp, entry_price=entry_price, exit_price=current_price, units=1000, instrument="EUR_USD", strategy="RSI_MeanReversion")
position = None
elif position == "SHORT" and current_rsi < 50: # Exit short when RSI returns to middle
backtest.add_trade(entry_time=entry_time, exit_time=timestamp, entry_price=entry_price, exit_price=current_price, units=-1000, instrument="EUR_USD", strategy="RSI_MeanReversion")
position = None
# Close any remaining position
if position is not None:
final_price = df["close"].iloc[-1]
final_time = df.index[-1]
units = 1000 if position == "LONG" else -1000
backtest.add_trade(entry_time=entry_time, exit_time=final_time, entry_price=entry_price, exit_price=final_price, units=units, instrument="EUR_USD", strategy="RSI_MeanReversion")
return backtest
if not eur_usd_data.empty:
print("🔄 Running RSI Mean Reversion Strategy Backtest...")
# Run backtest
rsi_backtest = backtest_rsi_strategy(eur_usd_data, oversold=25, overbought=75)
# Get performance metrics
rsi_metrics = rsi_backtest.calculate_performance_metrics()
print("\n📊 RSI Mean Reversion Strategy Results:")
print(f" Total Trades: {rsi_metrics['total_trades']}")
print(f" Win Rate: {rsi_metrics['win_rate']:.2f}%")
print(f" Total Return: {rsi_metrics['total_return']:.2f}%")
print(f" Profit Factor: {rsi_metrics['profit_factor']:.2f}")
print(f" Average Win: ${rsi_metrics['avg_win']:.2f}")
print(f" Average Loss: ${rsi_metrics['avg_loss']:.2f}")
print(f" Max Drawdown: {rsi_metrics['max_drawdown']:.2f}%")
print(f" Sharpe Ratio: {rsi_metrics['sharpe_ratio']:.2f}")
print(f" Average Trade Duration: {rsi_metrics['avg_duration_hours']:.1f} hours")
print(f" Final Balance: ${rsi_metrics['final_balance']:.2f}")
else:
print("❌ No data available for backtesting")
def backtest_rsi_strategy(df: pd.DataFrame, oversold: float = 30, overbought: float = 70) -> BacktestEngine:
"""Backtest an RSI mean reversion strategy."""
backtest = BacktestEngine(initial_balance=10000)
position = None
entry_price = None
entry_time = None
for i, (timestamp, row) in enumerate(df.iterrows()):
if i < 20: # Not enough data for RSI
continue
current_rsi = row["rsi"]
current_price = row["close"]
# Skip if RSI is NaN
if pd.isna(current_rsi):
continue
# Entry signals
if position is None:
if current_rsi < oversold: # Oversold - buy signal
position = "LONG"
entry_price = current_price
entry_time = timestamp
elif current_rsi > overbought: # Overbought - sell signal
position = "SHORT"
entry_price = current_price
entry_time = timestamp
# Exit signals
elif position == "LONG" and current_rsi > 50: # Exit long when RSI returns to middle
backtest.add_trade(entry_time=entry_time, exit_time=timestamp, entry_price=entry_price, exit_price=current_price, units=1000, instrument="EUR_USD", strategy="RSI_MeanReversion")
position = None
elif position == "SHORT" and current_rsi < 50: # Exit short when RSI returns to middle
backtest.add_trade(entry_time=entry_time, exit_time=timestamp, entry_price=entry_price, exit_price=current_price, units=-1000, instrument="EUR_USD", strategy="RSI_MeanReversion")
position = None
# Close any remaining position
if position is not None:
final_price = df["close"].iloc[-1]
final_time = df.index[-1]
units = 1000 if position == "LONG" else -1000
backtest.add_trade(entry_time=entry_time, exit_time=final_time, entry_price=entry_price, exit_price=final_price, units=units, instrument="EUR_USD", strategy="RSI_MeanReversion")
return backtest
if not eur_usd_data.empty:
print("🔄 Running RSI Mean Reversion Strategy Backtest...")
# Run backtest
rsi_backtest = backtest_rsi_strategy(eur_usd_data, oversold=25, overbought=75)
# Get performance metrics
rsi_metrics = rsi_backtest.calculate_performance_metrics()
print("\n📊 RSI Mean Reversion Strategy Results:")
print(f" Total Trades: {rsi_metrics['total_trades']}")
print(f" Win Rate: {rsi_metrics['win_rate']:.2f}%")
print(f" Total Return: {rsi_metrics['total_return']:.2f}%")
print(f" Profit Factor: {rsi_metrics['profit_factor']:.2f}")
print(f" Average Win: ${rsi_metrics['avg_win']:.2f}")
print(f" Average Loss: ${rsi_metrics['avg_loss']:.2f}")
print(f" Max Drawdown: {rsi_metrics['max_drawdown']:.2f}%")
print(f" Sharpe Ratio: {rsi_metrics['sharpe_ratio']:.2f}")
print(f" Average Trade Duration: {rsi_metrics['avg_duration_hours']:.1f} hours")
print(f" Final Balance: ${rsi_metrics['final_balance']:.2f}")
else:
print("❌ No data available for backtesting")
Strategy Comparison and Equity Curves¶
In [ ]:
Copied!
if not eur_usd_data.empty and "sma_backtest" in locals() and "rsi_backtest" in locals():
# Get equity curves
sma_equity = sma_backtest.get_equity_curve_df()
rsi_equity = rsi_backtest.get_equity_curve_df()
# Plot comparison
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(15, 10))
# Equity curves
ax1.plot(sma_equity.index, sma_equity["equity"], label="SMA Crossover", linewidth=2)
ax1.plot(rsi_equity.index, rsi_equity["equity"], label="RSI Mean Reversion", linewidth=2)
ax1.axhline(y=10000, color="gray", linestyle="--", alpha=0.7, label="Initial Balance")
ax1.set_title("Strategy Comparison - Equity Curves")
ax1.set_ylabel("Equity ($)")
ax1.legend()
ax1.grid(True, alpha=0.3)
# Drawdown comparison
ax2.fill_between(sma_equity.index, 0, -sma_equity["drawdown"], alpha=0.7, label="SMA Drawdown")
ax2.fill_between(rsi_equity.index, 0, -rsi_equity["drawdown"], alpha=0.7, label="RSI Drawdown")
ax2.set_title("Strategy Drawdowns")
ax2.set_ylabel("Drawdown (%)")
ax2.set_xlabel("Time")
ax2.legend()
ax2.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Strategy comparison table
comparison_data = {
"Metric": ["Total Return (%)", "Win Rate (%)", "Total Trades", "Profit Factor", "Max Drawdown (%)", "Sharpe Ratio", "Avg Duration (hrs)"],
"SMA Crossover": [f"{metrics['total_return']:.2f}", f"{metrics['win_rate']:.2f}", f"{metrics['total_trades']}", f"{metrics['profit_factor']:.2f}", f"{metrics['max_drawdown']:.2f}", f"{metrics['sharpe_ratio']:.2f}", f"{metrics['avg_duration_hours']:.1f}"],
"RSI Mean Reversion": [f"{rsi_metrics['total_return']:.2f}", f"{rsi_metrics['win_rate']:.2f}", f"{rsi_metrics['total_trades']}", f"{rsi_metrics['profit_factor']:.2f}", f"{rsi_metrics['max_drawdown']:.2f}", f"{rsi_metrics['sharpe_ratio']:.2f}", f"{rsi_metrics['avg_duration_hours']:.1f}"],
}
comparison_df = pd.DataFrame(comparison_data)
print("\n📊 Strategy Comparison:")
print(comparison_df.to_string(index=False))
# Determine best strategy
best_return = "SMA Crossover" if metrics["total_return"] > rsi_metrics["total_return"] else "RSI Mean Reversion"
best_sharpe = "SMA Crossover" if metrics["sharpe_ratio"] > rsi_metrics["sharpe_ratio"] else "RSI Mean Reversion"
best_drawdown = "SMA Crossover" if metrics["max_drawdown"] < rsi_metrics["max_drawdown"] else "RSI Mean Reversion"
print("\n🏆 Best Performance:")
print(f" Highest Return: {best_return}")
print(f" Best Risk-Adjusted Return: {best_sharpe}")
print(f" Lowest Drawdown: {best_drawdown}")
else:
print("❌ Backtest data not available for comparison")
if not eur_usd_data.empty and "sma_backtest" in locals() and "rsi_backtest" in locals():
# Get equity curves
sma_equity = sma_backtest.get_equity_curve_df()
rsi_equity = rsi_backtest.get_equity_curve_df()
# Plot comparison
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(15, 10))
# Equity curves
ax1.plot(sma_equity.index, sma_equity["equity"], label="SMA Crossover", linewidth=2)
ax1.plot(rsi_equity.index, rsi_equity["equity"], label="RSI Mean Reversion", linewidth=2)
ax1.axhline(y=10000, color="gray", linestyle="--", alpha=0.7, label="Initial Balance")
ax1.set_title("Strategy Comparison - Equity Curves")
ax1.set_ylabel("Equity ($)")
ax1.legend()
ax1.grid(True, alpha=0.3)
# Drawdown comparison
ax2.fill_between(sma_equity.index, 0, -sma_equity["drawdown"], alpha=0.7, label="SMA Drawdown")
ax2.fill_between(rsi_equity.index, 0, -rsi_equity["drawdown"], alpha=0.7, label="RSI Drawdown")
ax2.set_title("Strategy Drawdowns")
ax2.set_ylabel("Drawdown (%)")
ax2.set_xlabel("Time")
ax2.legend()
ax2.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Strategy comparison table
comparison_data = {
"Metric": ["Total Return (%)", "Win Rate (%)", "Total Trades", "Profit Factor", "Max Drawdown (%)", "Sharpe Ratio", "Avg Duration (hrs)"],
"SMA Crossover": [f"{metrics['total_return']:.2f}", f"{metrics['win_rate']:.2f}", f"{metrics['total_trades']}", f"{metrics['profit_factor']:.2f}", f"{metrics['max_drawdown']:.2f}", f"{metrics['sharpe_ratio']:.2f}", f"{metrics['avg_duration_hours']:.1f}"],
"RSI Mean Reversion": [f"{rsi_metrics['total_return']:.2f}", f"{rsi_metrics['win_rate']:.2f}", f"{rsi_metrics['total_trades']}", f"{rsi_metrics['profit_factor']:.2f}", f"{rsi_metrics['max_drawdown']:.2f}", f"{rsi_metrics['sharpe_ratio']:.2f}", f"{rsi_metrics['avg_duration_hours']:.1f}"],
}
comparison_df = pd.DataFrame(comparison_data)
print("\n📊 Strategy Comparison:")
print(comparison_df.to_string(index=False))
# Determine best strategy
best_return = "SMA Crossover" if metrics["total_return"] > rsi_metrics["total_return"] else "RSI Mean Reversion"
best_sharpe = "SMA Crossover" if metrics["sharpe_ratio"] > rsi_metrics["sharpe_ratio"] else "RSI Mean Reversion"
best_drawdown = "SMA Crossover" if metrics["max_drawdown"] < rsi_metrics["max_drawdown"] else "RSI Mean Reversion"
print("\n🏆 Best Performance:")
print(f" Highest Return: {best_return}")
print(f" Best Risk-Adjusted Return: {best_sharpe}")
print(f" Lowest Drawdown: {best_drawdown}")
else:
print("❌ Backtest data not available for comparison")
Multi-Instrument Analysis¶
In [ ]:
Copied!
if account_id:
async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
collector = FiveTwentyDataCollector(client, account_id)
print("📊 Fetching data for multiple instruments...")
instruments = ["EUR_USD", "GBP_USD", "USD_JPY", "AUD_USD"]
multi_data = await collector.get_multiple_instruments(
instruments=instruments,
granularity=CandlestickGranularity.H4, # 4-hour data
count=200,
)
if multi_data:
# Create correlation matrix
price_data = {}
return_data = {}
for instrument, df in multi_data.items():
price_data[instrument] = df["close"]
return_data[instrument] = df["close"].pct_change() * 100
price_df = pd.DataFrame(price_data)
return_df = pd.DataFrame(return_data)
# Calculate correlations
price_corr = price_df.corr()
return_corr = return_df.corr()
# Plot correlation heatmaps
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
sns.heatmap(price_corr, annot=True, cmap="coolwarm", center=0, square=True, ax=ax1, fmt=".3f")
ax1.set_title("Price Correlation Matrix")
sns.heatmap(return_corr, annot=True, cmap="coolwarm", center=0, square=True, ax=ax2, fmt=".3f")
ax2.set_title("Return Correlation Matrix")
plt.tight_layout()
plt.show()
# Volatility analysis
volatility_stats = {}
for instrument, df in multi_data.items():
returns = df["close"].pct_change().dropna() * 100
volatility_stats[instrument] = {"daily_vol": returns.std(), "max_gain": returns.max(), "max_loss": returns.min(), "avg_return": returns.mean(), "skewness": returns.skew(), "kurtosis": returns.kurtosis()}
vol_df = pd.DataFrame(volatility_stats).T
print("\n📊 Multi-Instrument Volatility Analysis:")
print(vol_df.round(4))
# Risk-return scatter plot
plt.figure(figsize=(10, 6))
for instrument in instruments:
if instrument in volatility_stats:
vol = volatility_stats[instrument]["daily_vol"]
ret = volatility_stats[instrument]["avg_return"]
plt.scatter(vol, ret, s=100, alpha=0.7, label=instrument)
plt.annotate(instrument, (vol, ret), xytext=(5, 5), textcoords="offset points")
plt.xlabel("Volatility (Daily Return Std %)")
plt.ylabel("Average Return (%)")
plt.title("Risk-Return Profile by Instrument")
plt.grid(True, alpha=0.3)
plt.axhline(y=0, color="black", linestyle="-", alpha=0.3)
plt.axvline(x=0, color="black", linestyle="-", alpha=0.3)
plt.legend()
plt.show()
else:
print("❌ No account connection - cannot fetch multi-instrument data")
if account_id:
async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
collector = FiveTwentyDataCollector(client, account_id)
print("📊 Fetching data for multiple instruments...")
instruments = ["EUR_USD", "GBP_USD", "USD_JPY", "AUD_USD"]
multi_data = await collector.get_multiple_instruments(
instruments=instruments,
granularity=CandlestickGranularity.H4, # 4-hour data
count=200,
)
if multi_data:
# Create correlation matrix
price_data = {}
return_data = {}
for instrument, df in multi_data.items():
price_data[instrument] = df["close"]
return_data[instrument] = df["close"].pct_change() * 100
price_df = pd.DataFrame(price_data)
return_df = pd.DataFrame(return_data)
# Calculate correlations
price_corr = price_df.corr()
return_corr = return_df.corr()
# Plot correlation heatmaps
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
sns.heatmap(price_corr, annot=True, cmap="coolwarm", center=0, square=True, ax=ax1, fmt=".3f")
ax1.set_title("Price Correlation Matrix")
sns.heatmap(return_corr, annot=True, cmap="coolwarm", center=0, square=True, ax=ax2, fmt=".3f")
ax2.set_title("Return Correlation Matrix")
plt.tight_layout()
plt.show()
# Volatility analysis
volatility_stats = {}
for instrument, df in multi_data.items():
returns = df["close"].pct_change().dropna() * 100
volatility_stats[instrument] = {"daily_vol": returns.std(), "max_gain": returns.max(), "max_loss": returns.min(), "avg_return": returns.mean(), "skewness": returns.skew(), "kurtosis": returns.kurtosis()}
vol_df = pd.DataFrame(volatility_stats).T
print("\n📊 Multi-Instrument Volatility Analysis:")
print(vol_df.round(4))
# Risk-return scatter plot
plt.figure(figsize=(10, 6))
for instrument in instruments:
if instrument in volatility_stats:
vol = volatility_stats[instrument]["daily_vol"]
ret = volatility_stats[instrument]["avg_return"]
plt.scatter(vol, ret, s=100, alpha=0.7, label=instrument)
plt.annotate(instrument, (vol, ret), xytext=(5, 5), textcoords="offset points")
plt.xlabel("Volatility (Daily Return Std %)")
plt.ylabel("Average Return (%)")
plt.title("Risk-Return Profile by Instrument")
plt.grid(True, alpha=0.3)
plt.axhline(y=0, color="black", linestyle="-", alpha=0.3)
plt.axvline(x=0, color="black", linestyle="-", alpha=0.3)
plt.legend()
plt.show()
else:
print("❌ No account connection - cannot fetch multi-instrument data")
Advanced Analytics: Monte Carlo Simulation¶
In [ ]:
Copied!
def monte_carlo_simulation(returns: pd.Series, initial_balance: float = 10000, days: int = 252, simulations: int = 1000) -> dict:
"""Run Monte Carlo simulation based on historical returns."""
# Calculate statistics from historical returns
mean_return = returns.mean() / 100 # Convert percentage to decimal
std_return = returns.std() / 100
# Run simulations
simulation_results = []
for _ in range(simulations):
# Generate random returns based on historical distribution
random_returns = np.random.normal(mean_return, std_return, days)
# Calculate cumulative performance
cumulative_returns = np.cumprod(1 + random_returns)
final_balance = initial_balance * cumulative_returns[-1]
# Calculate maximum drawdown
cumulative_balance = initial_balance * cumulative_returns
peak = np.maximum.accumulate(cumulative_balance)
drawdown = (cumulative_balance - peak) / peak
max_drawdown = drawdown.min() * 100
simulation_results.append({"final_balance": final_balance, "total_return": (final_balance - initial_balance) / initial_balance * 100, "max_drawdown": max_drawdown, "path": cumulative_balance})
# Calculate statistics
final_balances = [sim["final_balance"] for sim in simulation_results]
total_returns = [sim["total_return"] for sim in simulation_results]
max_drawdowns = [sim["max_drawdown"] for sim in simulation_results]
results = {
"simulations": simulation_results,
"statistics": {
"mean_final_balance": np.mean(final_balances),
"median_final_balance": np.median(final_balances),
"std_final_balance": np.std(final_balances),
"mean_return": np.mean(total_returns),
"median_return": np.median(total_returns),
"return_5th_percentile": np.percentile(total_returns, 5),
"return_95th_percentile": np.percentile(total_returns, 95),
"probability_of_loss": len([r for r in total_returns if r < 0]) / len(total_returns) * 100,
"mean_max_drawdown": np.mean(max_drawdowns),
"worst_drawdown": min(max_drawdowns),
},
}
return results
if not eur_usd_data.empty:
print("🎲 Running Monte Carlo Simulation...")
# Get returns for simulation
returns = eur_usd_data["price_change_pct"].dropna()
# Run simulation
mc_results = monte_carlo_simulation(returns, initial_balance=10000, days=252, simulations=1000)
# Display results
stats = mc_results["statistics"]
print("\n🎯 Monte Carlo Simulation Results (1000 simulations, 1 year):")
print(f" Expected Return: {stats['mean_return']:.2f}%")
print(f" Median Return: {stats['median_return']:.2f}%")
print(f" 5th Percentile Return: {stats['return_5th_percentile']:.2f}%")
print(f" 95th Percentile Return: {stats['return_95th_percentile']:.2f}%")
print(f" Probability of Loss: {stats['probability_of_loss']:.1f}%")
print(f" Expected Final Balance: ${stats['mean_final_balance']:.2f}")
print(f" Average Max Drawdown: {stats['mean_max_drawdown']:.2f}%")
print(f" Worst Case Drawdown: {stats['worst_drawdown']:.2f}%")
# Plot simulation results
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
# Sample paths
for i in range(min(100, len(mc_results["simulations"]))):
path = mc_results["simulations"][i]["path"]
ax1.plot(path, alpha=0.1, color="blue")
ax1.axhline(y=10000, color="red", linestyle="--", label="Initial Balance")
ax1.set_title("Monte Carlo Simulation Paths (100 samples)")
ax1.set_xlabel("Days")
ax1.set_ylabel("Portfolio Value ($)")
ax1.legend()
ax1.grid(True, alpha=0.3)
# Return distribution
total_returns = [sim["total_return"] for sim in mc_results["simulations"]]
ax2.hist(total_returns, bins=50, alpha=0.7, edgecolor="black")
ax2.axvline(x=0, color="red", linestyle="--", label="Break-even")
ax2.axvline(x=stats["mean_return"], color="green", linestyle="--", label="Mean Return")
ax2.set_title("Distribution of Returns")
ax2.set_xlabel("Total Return (%)")
ax2.set_ylabel("Frequency")
ax2.legend()
ax2.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
else:
print("❌ No data available for Monte Carlo simulation")
def monte_carlo_simulation(returns: pd.Series, initial_balance: float = 10000, days: int = 252, simulations: int = 1000) -> dict:
"""Run Monte Carlo simulation based on historical returns."""
# Calculate statistics from historical returns
mean_return = returns.mean() / 100 # Convert percentage to decimal
std_return = returns.std() / 100
# Run simulations
simulation_results = []
for _ in range(simulations):
# Generate random returns based on historical distribution
random_returns = np.random.normal(mean_return, std_return, days)
# Calculate cumulative performance
cumulative_returns = np.cumprod(1 + random_returns)
final_balance = initial_balance * cumulative_returns[-1]
# Calculate maximum drawdown
cumulative_balance = initial_balance * cumulative_returns
peak = np.maximum.accumulate(cumulative_balance)
drawdown = (cumulative_balance - peak) / peak
max_drawdown = drawdown.min() * 100
simulation_results.append({"final_balance": final_balance, "total_return": (final_balance - initial_balance) / initial_balance * 100, "max_drawdown": max_drawdown, "path": cumulative_balance})
# Calculate statistics
final_balances = [sim["final_balance"] for sim in simulation_results]
total_returns = [sim["total_return"] for sim in simulation_results]
max_drawdowns = [sim["max_drawdown"] for sim in simulation_results]
results = {
"simulations": simulation_results,
"statistics": {
"mean_final_balance": np.mean(final_balances),
"median_final_balance": np.median(final_balances),
"std_final_balance": np.std(final_balances),
"mean_return": np.mean(total_returns),
"median_return": np.median(total_returns),
"return_5th_percentile": np.percentile(total_returns, 5),
"return_95th_percentile": np.percentile(total_returns, 95),
"probability_of_loss": len([r for r in total_returns if r < 0]) / len(total_returns) * 100,
"mean_max_drawdown": np.mean(max_drawdowns),
"worst_drawdown": min(max_drawdowns),
},
}
return results
if not eur_usd_data.empty:
print("🎲 Running Monte Carlo Simulation...")
# Get returns for simulation
returns = eur_usd_data["price_change_pct"].dropna()
# Run simulation
mc_results = monte_carlo_simulation(returns, initial_balance=10000, days=252, simulations=1000)
# Display results
stats = mc_results["statistics"]
print("\n🎯 Monte Carlo Simulation Results (1000 simulations, 1 year):")
print(f" Expected Return: {stats['mean_return']:.2f}%")
print(f" Median Return: {stats['median_return']:.2f}%")
print(f" 5th Percentile Return: {stats['return_5th_percentile']:.2f}%")
print(f" 95th Percentile Return: {stats['return_95th_percentile']:.2f}%")
print(f" Probability of Loss: {stats['probability_of_loss']:.1f}%")
print(f" Expected Final Balance: ${stats['mean_final_balance']:.2f}")
print(f" Average Max Drawdown: {stats['mean_max_drawdown']:.2f}%")
print(f" Worst Case Drawdown: {stats['worst_drawdown']:.2f}%")
# Plot simulation results
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
# Sample paths
for i in range(min(100, len(mc_results["simulations"]))):
path = mc_results["simulations"][i]["path"]
ax1.plot(path, alpha=0.1, color="blue")
ax1.axhline(y=10000, color="red", linestyle="--", label="Initial Balance")
ax1.set_title("Monte Carlo Simulation Paths (100 samples)")
ax1.set_xlabel("Days")
ax1.set_ylabel("Portfolio Value ($)")
ax1.legend()
ax1.grid(True, alpha=0.3)
# Return distribution
total_returns = [sim["total_return"] for sim in mc_results["simulations"]]
ax2.hist(total_returns, bins=50, alpha=0.7, edgecolor="black")
ax2.axvline(x=0, color="red", linestyle="--", label="Break-even")
ax2.axvline(x=stats["mean_return"], color="green", linestyle="--", label="Mean Return")
ax2.set_title("Distribution of Returns")
ax2.set_xlabel("Total Return (%)")
ax2.set_ylabel("Frequency")
ax2.legend()
ax2.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
else:
print("❌ No data available for Monte Carlo simulation")
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.