Live Paper Trading¶

Test strategies in real-time with paper trading.

---

📋 Notebook Information

• RustyBT Version: 0.1.2+
• Last Validated: 2025-11-07
• API Compatibility: Verified ✅
• Documentation: API Reference

---

In [1]:

from rustybt.analytics import setup_notebook

setup_notebook()

from rustybt.analytics import setup_notebook setup_notebook()

✅ Notebook environment configured successfully
   - Async/await support enabled
   - Pandas display options optimized
   - Progress bars configured

Paper Trading Setup¶

In [2]:

# Note: Live trading requires an async environment and proper data portal setup
# This example shows the basic structure

import asyncio
from rustybt import TradingAlgorithm
from rustybt.api import order, symbol, set_commission, set_slippage
from rustybt.finance.commission import PerShare
from rustybt.finance.slippage import VolumeShareSlippage
from rustybt.live import LiveTradingEngine
from rustybt.live.brokers import PaperBroker


class SimpleMovingAverage(TradingAlgorithm):
    """Simple moving average strategy for live trading."""
    
    def initialize(self, context) -> None:
        """Initialize strategy - runs once at start."""
        # Configure trading costs
        set_commission(PerShare(cost=0.001, min_trade_cost=1.0))
        set_slippage(VolumeShareSlippage(volume_limit=0.025, price_impact=0.1))
        
        # Set up strategy parameters
        context.asset = symbol("SPY")
        context.window_size = 20
        context.price_history = []
        
    def handle_data(self, context, data) -> None:
        """Called on each new bar of data."""
        # Get current price
        price = data.current(context.asset, "close")
        
        # Update price history
        context.price_history.append(price)
        if len(context.price_history) > context.window_size:
            context.price_history.pop(0)
        
        # Wait for enough history
        if len(context.price_history) < context.window_size:
            return
        
        # Calculate moving average
        moving_avg = sum(context.price_history) / len(context.price_history)
        
        # Simple trading logic
        current_position = context.portfolio.positions.get(context.asset, 0)
        
        if price > moving_avg and current_position == 0:
            # Price above MA and no position - buy
            cash = context.portfolio.cash
            shares = int(cash * 0.95 / price)  # Use 95% of cash
            if shares > 0:
                order(context.asset, shares)
                
        elif price < moving_avg and current_position > 0:
            # Price below MA and have position - sell
            order(context.asset, -current_position)


# Example: Initialize live trading engine
# 
# Note: This requires proper setup of data_portal (data bundle) and async execution
# In a real scenario, you would:
#
# 1. Create a data portal with your data bundle
# 2. Initialize the paper broker
# 3. Create the live trading engine
# 4. Run the engine with asyncio
#
# Example code structure:
#
# async def run_live_trading():
#     # Initialize components
#     broker = PaperBroker(starting_cash=100000.0)
#     
#     # Create data portal (requires bundle setup - see 02_data_ingestion.ipynb)
#     # from rustybt.data.polars_data_portal import PolarsDataPortal
#     # data_portal = PolarsDataPortal(...)
#     
#     # Initialize engine
#     # engine = LiveTradingEngine(
#     #     strategy=SimpleMovingAverage(),
#     #     broker_adapter=broker,
#     #     data_portal=data_portal,
#     # )
#     
#     # Start the engine
#     # await engine.run()
#
# # Run in notebook (requires nest_asyncio for Jupyter)
# # import nest_asyncio
# # nest_asyncio.apply()
# # asyncio.run(run_live_trading())

print("✓ Live trading strategy defined")
print("✓ To run live trading, set up a data portal and use the async pattern above")

# Note: Live trading requires an async environment and proper data portal setup # This example shows the basic structure import asyncio from rustybt import TradingAlgorithm from rustybt.api import order, symbol, set_commission, set_slippage from rustybt.finance.commission import PerShare from rustybt.finance.slippage import VolumeShareSlippage from rustybt.live import LiveTradingEngine from rustybt.live.brokers import PaperBroker class SimpleMovingAverage(TradingAlgorithm): """Simple moving average strategy for live trading.""" def initialize(self, context) -> None: """Initialize strategy - runs once at start.""" # Configure trading costs set_commission(PerShare(cost=0.001, min_trade_cost=1.0)) set_slippage(VolumeShareSlippage(volume_limit=0.025, price_impact=0.1)) # Set up strategy parameters context.asset = symbol("SPY") context.window_size = 20 context.price_history = [] def handle_data(self, context, data) -> None: """Called on each new bar of data.""" # Get current price price = data.current(context.asset, "close") # Update price history context.price_history.append(price) if len(context.price_history) > context.window_size: context.price_history.pop(0) # Wait for enough history if len(context.price_history) < context.window_size: return # Calculate moving average moving_avg = sum(context.price_history) / len(context.price_history) # Simple trading logic current_position = context.portfolio.positions.get(context.asset, 0) if price > moving_avg and current_position == 0: # Price above MA and no position - buy cash = context.portfolio.cash shares = int(cash * 0.95 / price) # Use 95% of cash if shares > 0: order(context.asset, shares) elif price < moving_avg and current_position > 0: # Price below MA and have position - sell order(context.asset, -current_position) # Example: Initialize live trading engine # # Note: This requires proper setup of data_portal (data bundle) and async execution # In a real scenario, you would: # # 1. Create a data portal with your data bundle # 2. Initialize the paper broker # 3. Create the live trading engine # 4. Run the engine with asyncio # # Example code structure: # # async def run_live_trading(): # # Initialize components # broker = PaperBroker(starting_cash=100000.0) # # # Create data portal (requires bundle setup - see 02_data_ingestion.ipynb) # # from rustybt.data.polars_data_portal import PolarsDataPortal # # data_portal = PolarsDataPortal(...) # # # Initialize engine # # engine = LiveTradingEngine( # # strategy=SimpleMovingAverage(), # # broker_adapter=broker, # # data_portal=data_portal, # # ) # # # Start the engine # # await engine.run() # # # Run in notebook (requires nest_asyncio for Jupyter) # # import nest_asyncio # # nest_asyncio.apply() # # asyncio.run(run_live_trading()) print("✓ Live trading strategy defined") print("✓ To run live trading, set up a data portal and use the async pattern above")

✓ Live trading strategy defined
✓ To run live trading, set up a data portal and use the async pattern above