Examples¶
The examples below are taken from the examples/ directory in the
repository. Each is a standalone script that you can run with:
uv run python examples/<script>.py
Make sure you have ETORO_API_KEY and ETORO_USER_KEY set in your
environment (or in a .env file).
Basic usage¶
Fetch live rates and place a market order.
examples/basic_usage.py¶
"""Basic usage of the etoropy SDK."""
import asyncio
from etoropy import EToroTrading, OrderOptions
async def main() -> None:
# Config is loaded from environment variables (ETORO_API_KEY, ETORO_USER_KEY, etc.)
# Or pass them directly:
async with EToroTrading() as etoro:
# Load bundled instrument mappings for fast symbol resolution
etoro.resolver.load_bundled_csv()
# Get live rates for some instruments
rates = await etoro.get_rates(["AAPL", "TSLA", "BTC"])
for rate in rates:
symbol = etoro.resolver.get_symbol(rate.instrument_id) or str(rate.instrument_id)
print(f"{symbol}: bid={rate.bid}, ask={rate.ask}")
# Get portfolio
portfolio = await etoro.get_portfolio()
print(f"Positions: {len(portfolio.client_portfolio.positions)}")
print(f"Credit: {portfolio.client_portfolio.credit}")
# Place a market order (demo mode)
result = await etoro.buy_by_amount(
"AAPL",
amount=100.0,
options=OrderOptions(leverage=1, stop_loss=100.0, take_profit=200.0),
)
print(f"Order placed: {result.order_for_open.order_id}")
if __name__ == "__main__":
asyncio.run(main())
Stream prices¶
Subscribe to real-time WebSocket price updates.
examples/stream_prices.py¶
"""Stream live prices via WebSocket."""
import asyncio
from etoropy import EToroTrading
from etoropy.models.websocket import WsInstrumentRate
async def main() -> None:
async with EToroTrading() as etoro:
etoro.resolver.load_bundled_csv()
# Listen for price updates
def on_price(symbol: str, instrument_id: int, rate: WsInstrumentRate) -> None:
print(f"{symbol} ({instrument_id}): bid={rate.bid}, ask={rate.ask}")
etoro.on("price", on_price)
# Connect WebSocket
await etoro.connect()
# Start streaming prices
await etoro.stream_prices(["AAPL", "TSLA", "BTC"])
# Stream for 60 seconds
await asyncio.sleep(60)
# Stop streaming
await etoro.stop_streaming_prices(["AAPL", "TSLA", "BTC"])
if __name__ == "__main__":
asyncio.run(main())
Algo bot skeleton¶
A minimal framework for building an event-driven trading bot.
examples/algo_bot_skeleton.py¶
"""Skeleton for an algorithmic trading bot using etoropy."""
import asyncio
import logging
from etoropy import EToroTrading, OrderOptions
from etoropy.models.websocket import WsInstrumentRate, WsPrivateEvent
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s: %(message)s")
logger = logging.getLogger("algo-bot")
class SimpleBot:
def __init__(self) -> None:
self.etoro = EToroTrading()
self.etoro.resolver.load_bundled_csv()
self._running = False
async def start(self) -> None:
logger.info("Starting bot...")
self._running = True
# Register event handlers
self.etoro.on("price", self.on_price)
self.etoro.on("order:update", self.on_order_update)
self.etoro.on("error", self.on_error)
# Connect
await self.etoro.connect()
# Subscribe to private events (order updates)
self.etoro.subscribe_to_private_events()
# Stream prices for instruments of interest
await self.etoro.stream_prices(["AAPL", "BTC"])
logger.info("Bot running. Press Ctrl+C to stop.")
# Keep running
try:
while self._running:
await asyncio.sleep(1)
except asyncio.CancelledError:
pass
finally:
await self.stop()
async def stop(self) -> None:
self._running = False
logger.info("Stopping bot...")
await self.etoro.disconnect()
logger.info("Bot stopped.")
def on_price(self, symbol: str, instrument_id: int, rate: WsInstrumentRate) -> None:
mid_price = (rate.bid + rate.ask) / 2
logger.info("Price update: %s = %.4f (bid=%.4f, ask=%.4f)", symbol, mid_price, rate.bid, rate.ask)
# --- YOUR STRATEGY LOGIC HERE ---
# Example: simple threshold trigger
# if symbol == "AAPL" and mid_price < 150.0:
# asyncio.create_task(self.place_buy("AAPL", 100.0))
def on_order_update(self, event: WsPrivateEvent) -> None:
logger.info(
"Order update: OrderID=%d, Status=%d, InstrumentID=%d",
event.order_id,
event.status_id,
event.instrument_id,
)
def on_error(self, error: Exception) -> None:
logger.error("Error: %s", error)
async def place_buy(self, symbol: str, amount: float) -> None:
try:
result = await self.etoro.buy_by_amount(symbol, amount, OrderOptions(leverage=1))
order_id = result.order_for_open.order_id
logger.info("Buy order placed: %d", order_id)
# Wait for execution
event = await self.etoro.wait_for_order(order_id, timeout_s=10.0)
logger.info("Order executed: position_id=%s", event.position_id)
except Exception as e:
logger.error("Failed to place buy order: %s", e)
async def main() -> None:
bot = SimpleBot()
await bot.start()
if __name__ == "__main__":
asyncio.run(main())
Portfolio & market data¶
import asyncio
from etoropy import EToroTrading, CandleInterval, CandleDirection
async def main():
async with EToroTrading() as etoro:
etoro.resolver.load_bundled_csv()
# Portfolio
portfolio = await etoro.get_portfolio()
positions = await etoro.get_positions()
pending = await etoro.get_pending_orders()
pnl = await etoro.get_pnl()
history = await etoro.get_trade_history(
min_date="2025-01-01", page=1, page_size=50,
)
# Historical candles
candles = await etoro.get_candles(
"AAPL",
interval=CandleInterval.ONE_DAY,
count=30,
direction=CandleDirection.DESC,
)
asyncio.run(main())
Wait for order execution¶
The wait_for_order() method combines a WebSocket
listener with a REST polling fallback:
import asyncio
from etoropy import EToroTrading
async def main():
async with EToroTrading() as etoro:
etoro.resolver.load_bundled_csv()
await etoro.connect()
result = await etoro.buy_by_amount("AAPL", 100.0)
order_id = result.order_for_open.order_id
# Blocks until the order executes, fails, or times out
event = await etoro.wait_for_order(order_id, timeout_s=15.0)
print(f"Executed! Position ID: {event.position_id}")
asyncio.run(main())
Low-level REST clients¶
For finer control, access the individual REST sub-clients through
etoro.rest:
import asyncio
from etoropy import EToroTrading
async def main():
async with EToroTrading() as etoro:
rest = etoro.rest
# Market data
result = await rest.market_data.search_instruments(
fields="instrumentId", search_text="Apple",
)
# Watchlists
watchlists = await rest.watchlists.get_user_watchlists()
# Social feeds
feed = await rest.feeds.get_instrument_feed(instrument_id=1001)
# User info
profile = await rest.users_info.get_user_profile(user_id=12345)
asyncio.run(main())