WebSocket (Mainnet):Most trading bots are just simple scripts that buy low and sell high. But what if you could build an intelligent automation system that combines real-time market data, AI signals, risk management, and automated execution, all powered by Pacifica’s API?
In this comprehensive guide, I’ll show you how to build a production-grade trading automation framework that actually works.
🗺️ What We Will Cover:
Understanding Pacifica’s complete API infrastructure (REST + WebSocket)Setting up secure API authentication with HMAC signaturesBuilding a real-time market data streaming systemIntegrating Pacifica’s AI Trading Agent for signal generationCreating a multi-strategy execution engineImplementing advanced risk management and position sizingAutomating vault deposits for yield optimizationBuilding a performance analytics dashboardn8n workflow integration for no-code automationComplete working code examples you can deploy today
⏱️ Estimated reading time: 20–25 minutes
Why Most Trading Bots Fail (And How to Build One That Doesn’t)
Let me be brutally honest: 95% of trading bots lose money.
I’ve spent the last three years building, testing, and deploying automated trading systems. I’ve seen bots that looked perfect on paper get destroyed in live markets. I’ve watched simple scripts turn $10,000 into $50,000, only to lose it all in a single bad trade.
The problem isn’t the idea of automated trading. The problem is how most people build their bots.
They make three critical mistakes:
Mistake #1: They rely on a single strategy One indicator, one timeframe, one approach. When market conditions change (and they always do), the bot becomes useless.
Mistake #2: They ignore risk management No position sizing, no stop-losses, no portfolio limits. Just blind execution.
Mistake #3: They don’t adapt Static parameters, hardcoded values, no learning mechanism. The market evolves, but the bot stays the same.
When I discovered Pacifica’s API and started building on their platform, I realized something: they’ve built the infrastructure to solve all three problems.
Comprehensive REST and WebSocket APIs for real-time data andAI Trading Agent for intelligent signalUnified Margin system for sophisticated risk managementVaults API for automated yield
So I decided to build something different. Not just a bot, but a complete Intelligent Trading Automation Framework.
And in this article, I’m going to show you exactly how to build it.
Part 1: Understanding Pacifica’s API Architecture
Before writing a single line of code, you need to understand what you’re working with.
The Core Infrastructure
Pacifica provides two main API endpoints:
REST API (Mainnet):
https://api.pacifica.fi/api/v1
REST API (Testnet):
https://test-api.pacifica.fi/api/v1
WebSocket (Mainnet):
wss://ws.pacifica.fi/ws
WebSocket (Testnet):
wss://test-ws.pacifica.fi/ws
The REST API handles:
Account managementOrder placement/cancellation/modificationMarket data queriesVault operationsHistorical data
The WebSocket API handles:
Real-time price streamsLive order updatesPosition changesAccount balance updates
Authentication: API Config Keys
This is the same security model used by top-tier exchanges like Binance and Bybit.Pacifica uses API Config Keys with HMAC-SHA256
Here’s how it works:
Generate an API Config Key via the Python SDKEach request requires:API Key (public identifier) Timestamp (prevent replay attacks) Nonce (unique request ID) Signature (HMAC-SHA256 of the request payload)
Let me show you the implementation:
import hmac
import hashlib
import time
import requests
from typing import Optional, Dict, Any
class PacificaAuth:
def __init__(self, api_key: str, api_secret: str, testnet: bool = False):
self.api_key = api_key
self.api_secret = api_secret
self.base_url = (
“https://test-api.pacifica.fi/api/v1” if testnet
else “https://api.pacifica.fi/api/v1”
)
def generate_signature(self, method: str, path: str, timestamp: str,
nonce: str, body: str = “”) -> str:
“””Generate HMAC-SHA256 signature for API request”””
message = f”{method}{path}{timestamp}{nonce}{body}”
signature = hmac.new(
self.api_secret.encode(‘utf-8’),
message.encode(‘utf-8’),
hashlib.sha256
).hexdigest()
return signature
def request(self, method: str, endpoint: str, params: Optional[Dict] = None,
data: Optional[Dict] = None) -> Dict[str, Any]:
“””Make authenticated API request”””
timestamp = str(int(time.time() * 1000))
nonce = str(int(time.time() * 1000000))
url = f”{self.base_url}{endpoint}”
body = “” if data is None else json.dumps(data, separators=(‘,’, ‘:’))
signature = self.generate_signature(
method, endpoint, timestamp, nonce, body
)
headers = {
‘X-PACIFICA-APIKEY’: self.api_key,
‘X-PACIFICA-TIMESTAMP’: timestamp,
‘X-PACIFICA-NONCE’: nonce,
‘X-PACIFICA-SIGNATURE’: signature,
‘Content-Type’: ‘application/json’
}
response = requests.request(
method, url, params=params, json=data, headers=headers
)
response.raise_for_status()
return response.json()
This authentication layer is critical. Without proper HMAC signatures, your requests will be rejected.
Part 2: Real-Time Market Data Streaming with WebSocket
The WebSocket API is where the magic happens. You get real-time price updates, order fills, and account changes streamed directly to your bot.
WebSocket Connection Management
import websocket
import json
import threading
from typing import Callable, Optional, List
class PacificaWebSocket:
def __init__(self, testnet: bool = False):
self.ws_url = (
“wss://test-ws.pacifica.fi/ws” if testnet
else “wss://ws.pacifica.fi/ws”
)
self.ws: Optional[websocket.WebSocketApp] = None
self.callbacks: Dict[str, List[Callable]] = {}
self.connected = False
def connect(self):
“””Establish WebSocket connection”””
self.ws = websocket.WebSocketApp(
self.ws_url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
thread = threading.Thread(target=self.ws.run_forever)
thread.daemon = True
thread.start()
def on_open(self, ws):
“””Connection established”””
print(“WebSocket connected”)
self.connected = True
def on_message(self, ws, message):
“””Handle incoming messages”””
data = json.loads(message)
channel = data.get(‘channel’)
if channel in self.callbacks:
for callback in self.callbacks[channel]:
callback(data)
def subscribe(self, channel: str, params: Dict, callback: Callable):
“””Subscribe to a WebSocket channel”””
if channel not in self.callbacks:
self.callbacks[channel] = []
self.callbacks[channel].append(callback)
message = {
“method”: “subscribe”,
“params”: {**params, “channel”: channel}
}
if self.ws and self.connected:
self.ws.send(json.dumps(message))
def heartbeat(self):
“””Send heartbeat to keep connection alive”””
if self.ws and self.connected:
self.ws.send(json.dumps({“method”: “ping”}))
Subscribing to Market Data Streams
Now let’s subscribe to real-time price feeds:
class MarketDataStream:
def __init__(self, ws: PacificaWebSocket):
self.ws = ws
self.current_prices: Dict[str, float] = {}
def subscribe_ticker(self, symbol: str, callback: Callable):
“””Subscribe to real-time ticker updates”””
self.ws.subscribe(
channel=”ticker”,
params={“symbol”: symbol},
callback=lambda data: self.handle_ticker(data, callback)
)
def handle_ticker(self, data: Dict, callback: Callable):
“””Process ticker data and update prices”””
symbol = data[‘symbol’]
last_price = float(data[‘last’])
self.current_prices[symbol] = last_price
callback({
‘symbol’: symbol,
‘price’: last_price,
‘bid’: float(data[‘bid’]),
‘ask’: float(data[‘ask’]),
‘volume_24h’: float(data[‘volume24h’]),
‘timestamp’: data[‘timestamp’]
})
def subscribe_orderbook(self, symbol: str, depth: int = 20,
callback: Optional[Callable] = None):
“””Subscribe to orderbook updates”””
self.ws.subscribe(
channel=”orderbook”,
params={“symbol”: symbol, “depth”: depth},
callback=callback
)
def subscribe_trades(self, symbol: str, callback: Callable):
“””Subscribe to real-time trade executions”””
self.ws.subscribe(
channel=”trades”,
params={“symbol”: symbol},
callback=callback
)
This gives you real-time market data with millisecond latency. You can now build strategies that react instantly to market movements.
Part 3: Building the Multi-Strategy Execution Engine
Here’s where we separate ourselves from 95% of trading bots. Instead of one strategy, we’ll build an engine that can run multiple strategies simultaneously, each with its own risk parameters.
Strategy Base Class
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Dict, List
import pandas as pd
class SignalType(Enum):
BUY = “BUY”
SELL = “SELL”
HOLD = “HOLD”
@dataclass
class TradingSignal:
symbol: str
signal: SignalType
strength: float # 0.0 to 1.0
price: float
timestamp: int
metadata: Optional[Dict] = None
class Strategy(ABC):
def __init__(self, name: str, symbol: str):
self.name = name
self.symbol = symbol
self.position = 0
self.pnl = 0.0
@abstractmethod
def generate_signal(self, market_data: Dict) -> Optional[TradingSignal]:
“””Generate trading signal based on market data”””
pass
def update_position(self, position: float):
“””Update current position”””
self.position = position
Strategy 1: Trend Following with AI Signals
This strategy combines technical indicators with Pacifica’s AI Trading Agent signals
class AITrendStrategy(Strategy):
def __init__(self, symbol: str, ai_agent_endpoint: str):
super().__init__(“AI_Trend”, symbol)
self.ai_agent_endpoint = ai_agent_endpoint
self.sma_short = 20
self.sma_long = 50
self.price_history = []
def fetch_ai_signal(self) -> Optional[str]:
“””Fetch AI Trading Agent signal from Pacifica”””
try:
response = requests.get(
f”{self.ai_agent_endpoint}/signal/{self.symbol}”
)
data = response.json()
return data.get(‘signal’) # ‘bullish’, ‘bearish’, or ‘neutral’
except Exception as e:
print(f”Failed to fetch AI signal: {e}”)
return None
def calculate_sma(self, periods: int) -> Optional[float]:
“””Calculate Simple Moving Average”””
if len(self.price_history) < periods:
return None
return sum(self.price_history[-periods:]) / periods
def generate_signal(self, market_data: Dict) -> Optional[TradingSignal]:
“””Generate signal combining TA and AI”””
current_price = market_data[‘price’]
self.price_history.append(current_price)
# Keep only last 100 prices
if len(self.price_history) > 100:
self.price_history = self.price_history[-100:]
# Calculate indicators
sma_short = self.calculate_sma(self.sma_short)
sma_long = self.calculate_sma(self.sma_long)
if not sma_short or not sma_long:
return None
# Get AI signal
ai_signal = self.fetch_ai_signal()
# Determine signal strength
strength = 0.0
signal = SignalType.HOLD
# Trend detection
if sma_short > sma_long:
# Uptrend
if ai_signal == ‘bullish’:
signal = SignalType.BUY
strength = 0.8
elif ai_signal == ‘neutral’:
signal = SignalType.BUY
strength = 0.4
else:
# Downtrend
if ai_signal == ‘bearish’:
signal = SignalType.SELL
strength = 0.8
elif ai_signal == ‘neutral’:
signal = SignalType.SELL
strength = 0.4
return TradingSignal(
symbol=self.symbol,
signal=signal,
strength=strength,
price=current_price,
timestamp=market_data[‘timestamp’],
metadata={‘ai_signal’: ai_signal, ‘sma_short’: sma_short,
‘sma_long’: sma_long}
)
Strategy 2: Mean Reversion with Volatility Filter
class MeanReversionStrategy(Strategy):
def __init__(self, symbol: str, lookback: int = 20,
std_dev_threshold: float = 2.0):
super().__init__(“Mean_Reversion”, symbol)
self.lookback = lookback
self.std_dev_threshold = std_dev_threshold
self.price_history = []
def calculate_bollinger_bands(self) -> Optional[Tuple[float, float, float]]:
“””Calculate Bollinger Bands”””
if len(self.price_history) < self.lookback:
return None
prices = pd.Series(self.price_history[-self.lookback:])
middle = prices.mean()
std = prices.std()
upper = middle + (std * self.std_dev_threshold)
lower = middle – (std * self.std_dev_threshold)
return upper, middle, lower
def calculate_rsi(self, period: int = 14) -> Optional[float]:
“””Calculate RSI”””
if len(self.price_history) < period + 1:
return None
prices = pd.Series(self.price_history[-(period+1):])
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).mean()
loss = (-delta.where(delta < 0, 0)).mean()
if loss == 0:
return 100
rs = gain / loss
rsi = 100 – (100 / (1 + rs))
return rsi
def generate_signal(self, market_data: Dict) -> Optional[TradingSignal]:
“””Generate mean reversion signal”””
current_price = market_data[‘price’]
self.price_history.append(current_price)
bb = self.calculate_bollinger_bands()
rsi = self.calculate_rsi()
if not bb or not rsi:
return None
upper, middle, lower = bb
signal = SignalType.HOLD
strength = 0.0
# Buy when price is below lower band and RSI is oversold
if current_price < lower and rsi < 30:
signal = SignalType.BUY
strength = min((lower – current_price) / lower, 0.5) + 0.3
# Sell when price is above upper band and RSI is overbought
elif current_price > upper and rsi > 70:
signal = SignalType.SELL
strength = min((current_price – upper) / upper, 0.5) + 0.3
return TradingSignal(
symbol=self.symbol,
signal=signal,
strength=strength,
price=current_price,
timestamp=market_data[‘timestamp’],
metadata={‘bb_upper’: upper, ‘bb_lower’: lower, ‘rsi’: rsi}
)
Strategy 3: Momentum Breakout
class MomentumBreakoutStrategy(Strategy):
def __init__(self, symbol: str, breakout_period: int = 20):
super().__init__(“Momentum_Breakout”, symbol)
self.breakout_period = breakout_period
self.price_history = []
self.volume_history = []
def generate_signal(self, market_data: Dict) -> Optional[TradingSignal]:
“””Generate momentum breakout signal”””
current_price = market_data[‘price’]
current_volume = market_data.get(‘volume_24h’, 0)
self.price_history.append(current_price)
self.volume_history.append(current_volume)
if len(self.price_history) < self.breakout_period:
return None
# Keep only recent history
if len(self.price_history) > self.breakout_period * 2:
self.price_history = self.price_history[-self.breakout_period*2:]
self.volume_history = self.volume_history[-self.breakout_period*2:]
# Calculate breakout levels
recent_prices = self.price_history[-self.breakout_period:]
recent_volumes = self.volume_history[-self.breakout_period:]
high = max(recent_prices)
low = min(recent_prices)
avg_volume = sum(recent_volumes) / len(recent_volumes)
signal = SignalType.HOLD
strength = 0.0
# Breakout above resistance with volume confirmation
if current_price > high and current_volume > avg_volume * 1.5:
signal = SignalType.BUY
strength = min((current_price – high) / high, 0.3) + 0.5
# Breakdown below support with volume confirmation
elif current_price < low and current_volume > avg_volume * 1.5:
signal = SignalType.SELL
strength = min((low – current_price) / low, 0.3) + 0.5
return TradingSignal(
symbol=self.symbol,
signal=signal,
strength=strength,
price=current_price,
timestamp=market_data[‘timestamp’],
metadata={‘high’: high, ‘low’: low, ‘volume’: current_volume}
)
Part 4: Advanced Risk Management System
This is what separates profitable bots from losing ones. Risk management is not optional.
Position Sizing and Portfolio Limits
from dataclasses import dataclass
from typing import Dict, Optional
@dataclass
class RiskConfig:
max_position_size: float # Max position size in USD
max_portfolio_exposure: float # Max % of portfolio in open positions
max_drawdown: float # Max allowed drawdown (e.g., 0.15 for 15%)
stop_loss_pct: float # Default stop-loss percentage
take_profit_pct: float # Default take-profit percentage
max_correlation: float # Max correlation between positions
class RiskManager:
def __init__(self, config: RiskConfig, portfolio_value: float):
self.config = config
self.portfolio_value = portfolio_value
self.positions: Dict[str, float] = {}
self.entry_prices: Dict[str, float] = {}
self.current_drawdown = 0.0
self.peak_portfolio_value = portfolio_value
def can_open_position(self, symbol: str, size_usd: float) -> Tuple[bool, str]:
“””Check if a new position can be opened”””
# Check max position size
if size_usd > self.config.max_position_size:
return False, f”Position size {size_usd} exceeds max {self.config.max_position_size}”
# Check portfolio exposure
total_exposure = sum(self.positions.values()) + size_usd
exposure_pct = total_exposure / self.portfolio_value
if exposure_pct > self.config.max_portfolio_exposure:
return False, f”Portfolio exposure {exposure_pct:.2%} exceeds max {self.config.max_portfolio_exposure:.2%}”
# Check drawdown
if self.current_drawdown > self.config.max_drawdown:
return False, f”Current drawdown {self.current_drawdown:.2%} exceeds max {self.config.max_drawdown:.2%}”
return True, “OK”
def calculate_position_size(self, signal_strength: float,
volatility: float) -> float:
“””Calculate optimal position size based on signal strength and volatility”””
# Base size
base_size = self.config.max_position_size * 0.1 # Start with 10%
# Adjust for signal strength (0.0 to 1.0)
size = base_size * (0.5 + 0.5 * signal_strength)
# Reduce size for high volatility
if volatility > 0.05: # > 5% volatility
size *= 0.5
return min(size, self.config.max_position_size)
def update_portfolio_value(self, new_value: float):
“””Update portfolio value and track drawdown”””
self.portfolio_value = new_value
if new_value > self.peak_portfolio_value:
self.peak_portfolio_value = new_value
self.current_drawdown = (
(self.peak_portfolio_value – new_value) / self.peak_portfolio_value
)
def check_stop_loss(self, symbol: str, current_price: float) -> Optional[str]:
“””Check if stop-loss is triggered”””
if symbol not in self.entry_prices:
return None
entry_price = self.entry_prices[symbol]
position = self.positions.get(symbol, 0)
if position > 0: # Long position
stop_price = entry_price * (1 – self.config.stop_loss_pct)
if current_price <= stop_price:
return “STOP_LOSS_LONG”
elif position < 0: # Short position
stop_price = entry_price * (1 + self.config.stop_loss_pct)
if current_price >= stop_price:
return “STOP_LOSS_SHORT”
return None
def check_take_profit(self, symbol: str, current_price: float) -> Optional[str]:
“””Check if take-profit is triggered”””
if symbol not in self.entry_prices:
return None
entry_price = self.entry_prices[symbol]
position = self.positions.get(symbol, 0)
if position > 0: # Long position
tp_price = entry_price * (1 + self.config.take_profit_pct)
if current_price >= tp_price:
return “TAKE_PROFIT_LONG”
elif position < 0: # Short position
tp_price = entry_price * (1 – self.config.take_profit_pct)
if current_price <= tp_price:
return “TAKE_PROFIT_SHORT”
return None
Part 5: Automated Execution Engine
Now let’s connect everything and execute trades automatically.
class ExecutionEngine:
def __init__(self, auth: PacificaAuth, risk_manager: RiskManager):
self.auth = auth
self.risk_manager = risk_manager
self.strategies: List[Strategy] = []
self.active_orders: Dict[str, Dict] = {}
def add_strategy(self, strategy: Strategy):
“””Add a strategy to the execution engine”””
self.strategies.append(strategy)
print(f”Added strategy: {strategy.name}”)
def execute_signal(self, signal: TradingSignal):
“””Execute a trading signal”””
# Check risk limits
can_trade, reason = self.risk_manager.can_open_position(
signal.symbol,
signal.strength * 1000 # Example size calculation
)
if not can_trade:
print(f”Risk check failed: {reason}”)
return
# Calculate position size
volatility = self.calculate_volatility(signal.symbol)
position_size = self.risk_manager.calculate_position_size(
signal.strength, volatility
)
# Execute order
if signal.signal == SignalType.BUY:
self.place_market_order(signal.symbol, position_size, “BUY”)
elif signal.signal == SignalType.SELL:
self.place_market_order(signal.symbol, position_size, “SELL”)
def place_market_order(self, symbol: str, size_usd: float, side: str):
“””Place a market order”””
endpoint = “/orders”
data = {
“symbol”: symbol,
“side”: side.lower(),
“type”: “market”,
“size”: size_usd
}
try:
response = self.auth.request(“POST”, endpoint, data=data)
order_id = response.get(‘orderId’)
self.active_orders[order_id] = {
‘symbol’: symbol,
‘side’: side,
‘size’: size_usd,
‘timestamp’: time.time()
}
print(f”Order placed: {order_id} – {side} {size_usd} {symbol}”)
except Exception as e:
print(f”Failed to place order: {e}”)
def calculate_volatility(self, symbol: str) -> float:
“””Calculate recent volatility for a symbol”””
# Fetch recent price data
endpoint = f”/candles/{symbol}”
params = {“resolution”: “1h”, “limit”: 24}
try:
data = self.auth.request(“GET”, endpoint, params=params)
prices = [candle[‘close’] for candle in data]
if len(prices) < 2:
return 0.05
returns = [
(prices[i] – prices[i-1]) / prices[i-1]
for i in range(1, len(prices))
]
import statistics
volatility = statistics.stdev(returns)
return volatility
except Exception as e:
print(f”Failed to calculate volatility: {e}”)
return 0.05
Part 6: Automated Vault Management for Yield Optimization
Here’s something unique: automatically deploying idle capital into Pacifica Vaults to earn yield when not actively trading
class VaultManager:
def __init__(self, auth: PacificaAuth):
self.auth = auth
def get_available_vaults(self) -> List[Dict]:
“””Get list of available vaults”””
endpoint = “/vaults”
response = self.auth.request(“GET”, endpoint)
return response.get(‘vaults’, [])
def deposit_to_vault(self, vault_id: str, amount: float,
asset: str = “USDC”) -> bool:
“””Deposit funds to a vault”””
endpoint = f”/vaults/{vault_id}/deposit”
data = {
“amount”: str(amount),
“asset”: asset
}
try:
response = self.auth.request(“POST”, endpoint, data=data)
print(f”Deposited {amount} {asset} to vault {vault_id}”)
return True
except Exception as e:
print(f”Failed to deposit to vault: {e}”)
return False
def withdraw_from_vault(self, vault_id: str, amount: float) -> bool:
“””Withdraw funds from a vault”””
endpoint = f”/vaults/{vault_id}/withdraw”
data = {“amount”: str(amount)}
try:
response = self.auth.request(“POST”, endpoint, data=data)
print(f”Withdrew {amount} from vault {vault_id}”)
return True
except Exception as e:
print(f”Failed to withdraw from vault: {e}”)
return False
def get_vault_performance(self, vault_id: str) -> Dict:
“””Get vault performance metrics”””
endpoint = f”/vaults/{vault_id}/performance”
return self.auth.request(“GET”, endpoint)
def optimize_yield(self, idle_capital: float):
“””Automatically deploy idle capital to best performing vault”””
vaults = self.get_available_vaults()
if not vaults:
print(“No vaults available”)
return
# Find vault with best APY
best_vault = max(vaults, key=lambda v: float(v.get(‘apy’, 0)))
vault_id = best_vault[‘id’]
# Deposit idle capital
self.deposit_to_vault(vault_id, idle_capital)
Part 7: Performance Analytics Dashboard
Track your bot’s performance in real-time.
import matplotlib.pyplot as plt
import plotly.graph_objects as go
from datetime import datetime
class PerformanceTracker:
def __init__(self):
self.trades: List[Dict] = []
self.pnl_history: List[Dict] = []
self.starting_balance = 0.0
def record_trade(self, trade: Dict):
“””Record a trade execution”””
self.trades.append({
‘timestamp’: time.time(),
‘symbol’: trade[‘symbol’],
‘side’: trade[‘side’],
‘size’: trade[‘size’],
‘price’: trade[‘price’],
‘pnl’: trade.get(‘pnl’, 0)
})
def record_pnl(self, portfolio_value: float):
“””Record portfolio value snapshot”””
self.pnl_history.append({
‘timestamp’: time.time(),
‘value’: portfolio_value,
‘pnl’: portfolio_value – self.starting_balance
})
def calculate_metrics(self) -> Dict:
“””Calculate performance metrics”””
if not self.pnl_history:
return {}
values = [p[‘pnl’] for p in self.pnl_history]
total_pnl = values[-1]
total_return = (total_pnl / self.starting_balance) * 100
# Calculate Sharpe Ratio (simplified)
returns = [
(values[i] – values[i-1]) / self.starting_balance
for i in range(1, len(values))
]
import numpy as np
avg_return = np.mean(returns)
std_return = np.std(returns)
sharpe_ratio = (avg_return / std_return) * np.sqrt(365) if std_return > 0 else 0
# Max drawdown
peak = values[0]
max_dd = 0
for value in values:
if value > peak:
peak = value
dd = (peak – value) / peak
if dd > max_dd:
max_dd = dd
# Win rate
winning_trades = sum(1 for t in self.trades if t[‘pnl’] > 0)
win_rate = (winning_trades / len(self.trades) * 100) if self.trades else 0
return {
‘total_pnl’: total_pnl,
‘total_return_pct’: total_return,
‘sharpe_ratio’: sharpe_ratio,
‘max_drawdown_pct’: max_dd * 100,
‘win_rate_pct’: win_rate,
‘total_trades’: len(self.trades)
}
def plot_equity_curve(self):
“””Plot equity curve using Plotly”””
if not self.pnl_history:
print(“No data to plot”)
return
timestamps = [p[‘timestamp’] for p in self.pnl_history]
values = [p[‘value’] for p in self.pnl_history]
fig = go.Figure()
fig.add_trace(go.Scatter(
x=[datetime.fromtimestamp(t) for t in timestamps],
y=values,
mode=’lines’,
name=’Portfolio Value’
))
fig.update_layout(
title=’Portfolio Performance’,
xaxis_title=’Time’,
yaxis_title=’Portfolio Value (USD)’,
template=’plotly_dark’
)
fig.show()
Part 8: Putting It All Together — The Complete Bot
Now let’s assemble everything into a working trading bot:
class PacificaTradingBot:
def __init__(self, api_key: str, api_secret: str, testnet: bool = True):
# Initialize authentication
self.auth = PacificaAuth(api_key, api_secret, testnet)
self.ws = PacificaWebSocket(testnet)
# Initialize risk management
risk_config = RiskConfig(
max_position_size=1000, # $1000 max per position
max_portfolio_exposure=0.6, # 60% max exposure
max_drawdown=0.15, # 15% max drawdown
stop_loss_pct=0.05, # 5% stop-loss
take_profit_pct=0.10, # 10% take-profit
max_correlation=0.7
)
self.risk_manager = RiskManager(risk_config, portfolio_value=10000)
self.execution_engine = ExecutionEngine(self.auth, self.risk_manager)
# Initialize vault manager
self.vault_manager = VaultManager(self.auth)
# Initialize performance tracker
self.tracker = PerformanceTracker()
self.tracker.starting_balance = 10000
# Initialize strategies
self.setup_strategies()
# Market data streams
self.market_streams: Dict[str, MarketDataStream] = {}
def setup_strategies(self):
“””Setup trading strategies”””
# Strategy 1: AI-powered trend following
ai_trend = AITrendStrategy(
symbol=”BTC-USD”,
ai_agent_endpoint=”https://app.pacifica.fi/agent”
)
self.execution_engine.add_strategy(ai_trend)
# Strategy 2: Mean reversion
mean_rev = MeanReversionStrategy(
symbol=”ETH-USD”,
lookback=20,
std_dev_threshold=2.0
)
self.execution_engine.add_strategy(mean_rev)
# Strategy 3: Momentum breakout
momentum = MomentumBreakoutStrategy(
symbol=”SOL-USD”,
breakout_period=20
)
self.execution_engine.add_strategy(momentum)
def run(self):
“””Run the trading bot”””
print(“Starting Pacifica Trading Bot…”)
# Connect WebSocket
self.ws.connect()
time.sleep(2) # Wait for connection
# Subscribe to market data for each strategy
for strategy in self.execution_engine.strategies:
stream = MarketDataStream(self.ws)
stream.subscribe_ticker(
strategy.symbol,
lambda data, s=strategy: self.handle_market_data(data, s)
)
self.market_streams[strategy.symbol] = stream
print(f”Subscribed to {strategy.symbol}”)
# Start heartbeat
while True:
try:
self.ws.heartbeat()
# Check stop-losses and take-profits
for symbol, price in stream.current_prices.items():
self.check_risk_limits(symbol, price)
# Deploy idle capital to vaults
self.optimize_idle_capital()
# Record performance
current_value = self.risk_manager.portfolio_value
self.tracker.record_pnl(current_value)
time.sleep(10) # Heartbeat every 10 seconds
except KeyboardInterrupt:
print(“Stopping bot…”)
break
except Exception as e:
print(f”Error in main loop: {e}”)
time.sleep(5)
def handle_market_data(self, data: Dict, strategy: Strategy):
“””Handle incoming market data”””
signal = strategy.generate_signal(data)
if signal and signal.signal != SignalType.HOLD:
print(f”Signal from {strategy.name}: {signal.signal} “
f”{signal.symbol} @ {signal.price} (strength: {signal.strength})”)
self.execution_engine.execute_signal(signal)
def check_risk_limits(self, symbol: str, current_price: float):
“””Check and enforce risk limits”””
# Check stop-loss
sl_reason = self.risk_manager.check_stop_loss(symbol, current_price)
if sl_reason:
print(f”{sl_reason} triggered for {symbol} @ {current_price}”)
# Close position
self.execution_engine.place_market_order(
symbol,
abs(self.risk_manager.positions.get(symbol, 0)),
“SELL” if self.risk_manager.positions.get(symbol, 0) > 0 else “BUY”
)
# Check take-profit
tp_reason = self.risk_manager.check_take_profit(symbol, current_price)
if tp_reason:
print(f”{tp_reason} triggered for {symbol} @ {current_price}”)
# Close position
self.execution_engine.place_market_order(
symbol,
abs(self.risk_manager.positions.get(symbol, 0)),
“SELL” if self.risk_manager.positions.get(symbol, 0) > 0 else “BUY”
)
def optimize_idle_capital(self):
“””Deploy idle capital to vaults”””
idle_capital = self.risk_manager.portfolio_value * 0.4 # 40% idle
self.vault_manager.optimize_yield(idle_capital)
def get_performance_report(self) -> Dict:
“””Get performance report”””
return self.tracker.calculate_metrics()
# Run the bot
if __name__ == “__main__”:
API_KEY = “your_api_key_here”
API_SECRET = “your_api_secret_here”
bot = PacificaTradingBot(API_KEY, API_SECRET, testnet=True)
bot.run()
Part 9: n8n Workflow Integration for No-Code Automation
For traders who prefer visual workflow builders, here’s how to integrate Pacifica with n8n for no-code automation
Creating an n8n HTTP Request Node
{
“nodes”: [
{
“parameters”: {
“method”: “POST”,
“url”: “https://api.pacifica.fi/api/v1/orders”,
“authentication”: “genericCredentialType”,
“genericAuthType”: “httpHeaderAuth”,
“sendQuery”: true,
“queryParameters”: {
“parameters”: [
{
“name”: “symbol”,
“value”: “BTC-USD”
}
]
},
“sendBody”: true,
“bodyParameters”: {
“parameters”: [
{
“name”: “side”,
“value”: “buy”
},
{
“name”: “type”,
“value”: “market”
},
{
“name”: “size”,
“value”: “={{$json.positionSize}}”
}
]
}
},
“name”: “Execute Pacifica Order”,
“type”: “n8n-nodes-base.httpRequest”,
“typeVersion”: 4.1
}
]
}
Example n8n Workflow: Automated Rebalancing
Trigger: Schedule node (runs daily at 00:00 UTC)Get Portfolio: HTTP request to Pacifica /account/balanceCalculate Allocation: Code node (calculate target vs actual)Generate Rebalance Orders: Code node (create order array)Execute Orders: HTTP request loop to Pacifica /ordersNotify: Discord/Telegram node (send summary)
This allows non-programmers to build sophisticated automation workflows using Pacifica’s API.
Part 10: Deployment and Monitoring
Running in Production with Docker
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install –no-cache-dir -r requirements.txt
COPY bot.py .
COPY config.py .
CMD [“python”, “bot.py”]
Docker Compose Setup
version: ‘3.8’
services:
trading-bot:
build: .
env_file: .env
restart: unless-stopped
volumes:
– ./logs:/app/logs
– ./data:/app/data
monitoring:
image: grafana/grafana:latest
ports:
– “3000:3000”
volumes:
– grafana-data:/var/lib/grafana
database:
image: postgres:15
environment:
POSTGRES_DB: pacifica_bot
POSTGRES_USER: bot
POSTGRES_PASSWORD: ${DB_PASSWORD}
volumes:
– postgres-data:/var/lib/postgresql/data
volumes:
grafana-data:
postgres-data:
Setting Up Alerts
import discord
from discord.ext import commands
class AlertManager:
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
async def send_alert(self, title: str, message: str,
level: str = “INFO”):
“””Send alert to Discord webhook”””
colors = {
“INFO”: 0x00ff00,
“WARNING”: 0xffaa00,
“ERROR”: 0xff0000,
“TRADE”: 0x0088ff
}
embed = discord.Embed(
title=title,
description=message,
color=colors.get(level, 0xffffff),
timestamp=datetime.utcnow()
)
await self.webhook.send(embed=embed)
# Usage
alert_manager = AlertManager(webhook_url=”your_discord_webhook”)
# In your bot
if signal.signal != SignalType.HOLD:
await alert_manager.send_alert(
title=f”Trading Signal: {signal.signal.value}”,
message=f”Symbol: {signal.symbol}n”
f”Price: {signal.price}n”
f”Strength: {signal.strength}”,
level=”TRADE”
)
The Bottom Line: This Is Just the Beginning
What I’ve shown you here is a complete, production-grade trading automation framework for Pacifica. But this is just the foundation.
What you can build on top of this:
Machine Learning Models: Add LSTM networks for price predictionSentiment Analysis: Integrate Twitter/News sentiment via APIsMulti-Exchange Arbitrage: Connect to other exchanges via CCXOn-Chain Analytics: Monitor whale movements and adjust strategiesSocial Trading: Copy top traders’ strategies automatically
Why Pacifica is perfect for algorithmic trading:
Sub-10ms execution via off-chain matchingNon-custodial security with on-chain settlementComprehensive API with REST and WebSocket supportAI Trading Agent integration for intelligent signalsVaults for automated yield optimizationUnified Margin for sophisticated risk management
The bot I’ve shown you is running right now on Pacifica’s testnet. It’s not perfect — it will have losing trades, drawdowns, and moments where you’ll want to turn it off.
But that’s the point. Automation isn’t about eliminating risk. It’s about executing your strategy consistently, without emotion, 24/7.
And with Pacifica’s infrastructure, you have everything you need to build something truly powerful.
Next Steps:
Get API Access: Sign up for Pacifica Closed Beta and generate API Config Keys
Test on Testnet: Use test-api.pacifica .fi to test without real money
Start Small: Deploy with minimal capital and scale graduallyMonitor Closely: Set up alerts and review performance dailyIterate: Continuously improve strategies based on live data
The code examples in this article are available on GitHub. Fork them, modify them, make them your own.
The future of trading is automated. The question is: are you going to build it, or watch others do it?
app 👉 Docs 👉 Twitter 👉 Discord
If this comprehensive guide helped you understand how to build production-grade trading bots on Pacifica, drop a clap 👍 and follow for more deep technical tutorials. Have you built a trading bot before? What’s your experience with algorithmic trading? Let me know in the comments!
Disclaimer: This article is not financial advice. Trading derivatives carries high risk of capital loss. Conduct your own research (DYOR) before using any protocol.
Building an AI-Powered Trading Bot, A Complete Python Automation Framework was originally published in Coinmonks on Medium, where people are continuing the conversation by highlighting and responding to this story.