Integration Patterns

Overview

This repository integrates OpenAlgo (live trading platform) with AITRAPP (backtest engine) through adapters and mocks.

Architecture:

OpenAlgo Strategy → Adapter → AITRAPP Backtest Engine ↓ Mock OpenAlgo API

Key Components

Integration Utilities

Location: openalgo/strategies/utils/

• aitrapp_integration.py

• Sets up AITRAPP path and imports

• openalgo_mock.py

• Mocks OpenAlgo API for backtesting

• strategy_adapter.py

• Base adapter for strategy conversion

• aitrapp_utils.py

• Helper functions for integration

AITRAPP Core

Location: AITRAPP/AITRAPP/packages/core/

• backtest.py

• Main backtest engine

• historical_data.py

• Historical data loader

• strategies/base.py

• Strategy interface

• risk.py

• Risk management module

• exits.py

• Exit management

Strategy Adapter Pattern

Base Adapter Structure

from AITRAPP.AITRAPP.packages.core.strategies.base import BaseStrategy from openalgo.strategies.utils.openalgo_mock import MockOpenAlgoAPI

class OpenAlgoStrategyAdapter(BaseStrategy): """ Adapter to convert OpenAlgo strategy to AITRAPP format. """ def init(self, openalgo_strategy_class, symbol, params): super().init(symbol) self.openalgo_strategy = openalgo_strategy_class(symbol, params) self.mock_api = MockOpenAlgoAPI()

    # Initialize OpenAlgo strategy with mock API
    self.openalgo_strategy.client = self.mock_api

def on_bar(self, bar):
    """
    AITRAPP callback for each bar.
    Convert bar to OpenAlgo format and run strategy.
    """
    # Convert AITRAPP bar to OpenAlgo format
    openalgo_data = self._convert_bar(bar)
    
    # Update strategy data
    self.openalgo_strategy.data = openalgo_data
    
    # Run strategy logic
    self.openalgo_strategy.calculate_indicators()
    self.openalgo_strategy.check_signals()
    
    # Extract signals
    return self._extract_signals()

def _convert_bar(self, bar):
    """Convert AITRAPP bar to pandas DataFrame"""
    import pandas as pd
    
    df = pd.DataFrame([{
        'open': bar.open,
        'high': bar.high,
        'low': bar.low,
        'close': bar.close,
        'volume': bar.volume,
        'datetime': bar.timestamp
    }])
    
    return df

def _extract_signals(self):
    """Extract trading signals from OpenAlgo strategy"""
    signals = []
    
    if self.openalgo_strategy.position != 0:
        signal = {
            'side': 'BUY' if self.openalgo_strategy.position > 0 else 'SELL',
            'price': self.openalgo_strategy.data.iloc[-1]['close'],
            'quantity': abs(self.openalgo_strategy.position)
        }
        signals.append(signal)
    
    return signals

OpenAlgo API Mock

Mock Implementation

class MockOpenAlgoAPI: """ Mock OpenAlgo API for backtesting. Mimics OpenAlgo API responses without actual broker calls. """ def init(self): self.orders = [] self.positions = [] self.historical_data = {}

def history(self, symbol, exchange="NSE", interval="5m", start_date=None, end_date=None):
    """
    Mock historical data fetch.
    In real backtest, this would load from AITRAPP historical data.
    """
    # Load from AITRAPP historical data loader
    from AITRAPP.AITRAPP.packages.core.historical_data import HistoricalDataLoader
    
    loader = HistoricalDataLoader()
    data = loader.load(
        symbol=symbol,
        exchange=exchange,
        interval=interval,
        start_date=start_date,
        end_date=end_date
    )
    
    # Convert to OpenAlgo format (pandas DataFrame)
    import pandas as pd
    df = pd.DataFrame(data)
    
    return df

def placesmartorder(self, strategy, symbol, action, exchange, price_type, product, quantity, position_size):
    """
    Mock order placement.
    In backtest, this records the order for execution simulation.
    """
    order = {
        'strategy': strategy,
        'symbol': symbol,
        'action': action,
        'exchange': exchange,
        'price_type': price_type,
        'product': product,
        'quantity': quantity,
        'position_size': position_size,
        'timestamp': datetime.now()
    }
    
    self.orders.append(order)
    
    # Return mock response
    return {
        'status': 'success',
        'order_id': f"MOCK_{len(self.orders)}",
        'message': 'Order placed (mock)'
    }

def get_positions(self):
    """Mock positions fetch"""
    return self.positions

def get_orderbook(self):
    """Mock orderbook fetch"""
    return self.orders

Strategy Conversion Workflow

Step 1: Create Adapter

File: openalgo/strategies/adapters/mcx_momentum_adapter.py

from openalgo.strategies.utils.strategy_adapter import BaseStrategyAdapter from openalgo.strategies.scripts.mcx_commodity_momentum_strategy import MCXMomentumStrategy

class MCXMomentumAdapter(BaseStrategyAdapter): def init(self, symbol, params): super().init(MCXMomentumStrategy, symbol, params)

Step 2: Register with Backtest Engine

File: openalgo/strategies/scripts/run_backtest_ranking.py

from openalgo.strategies.adapters.mcx_momentum_adapter import MCXMomentumAdapter

strategies = [ MCXMomentumAdapter("NATURALGAS24FEB26FUT", PARAMS), # Add more strategies ]

Run backtest

from AITRAPP.AITRAPP.packages.core.backtest import BacktestEngine

engine = BacktestEngine( strategies=strategies, start_date="2025-08-15", end_date="2025-11-10", capital=1000000 )

results = engine.run()

Data Format Conversion

AITRAPP Bar to OpenAlgo DataFrame

def convert_aitrapp_bar_to_openalgo(bar): """ Convert AITRAPP bar object to OpenAlgo DataFrame format. """ import pandas as pd

df = pd.DataFrame([{
    'datetime': bar.timestamp,
    'open': bar.open,
    'high': bar.high,
    'low': bar.low,
    'close': bar.close,
    'volume': bar.volume
}])

df.set_index('datetime', inplace=True)
return df

OpenAlgo Signal to AITRAPP Signal

def convert_openalgo_signal_to_aitrapp(signal): """ Convert OpenAlgo signal format to AITRAPP signal format. """ from AITRAPP.AITRAPP.packages.core.strategies.base import Signal

return Signal(
    symbol=signal['symbol'],
    side=signal['side'],  # 'BUY' or 'SELL'
    quantity=signal['quantity'],
    entry_price=signal['price'],
    stop_loss=signal.get('stop_loss'),
    take_profit=signal.get('take_profit')
)

Historical Data Integration

Loading Historical Data

from AITRAPP.AITRAPP.packages.core.historical_data import HistoricalDataLoader

loader = HistoricalDataLoader()

Load data for backtesting

data = loader.load( symbol="NIFTY", exchange="NSE", interval="5m", start_date="2025-08-15", end_date="2025-11-10" )

Convert to OpenAlgo format

import pandas as pd df = pd.DataFrame(data)

Data Format Requirements

AITRAPP Format:

{ 'timestamp': datetime, 'open': float, 'high': float, 'low': float, 'close': float, 'volume': int }

OpenAlgo Format:

DataFrame with columns:

• datetime (index)
• open
• high
• low
• close
• volume

Risk Management Integration

Using AITRAPP Risk Manager

from AITRAPP.AITRAPP.packages.core.risk import RiskManager

risk_manager = RiskManager( account_size=1000000, max_heat=0.02, # 2% portfolio heat daily_loss_limit=-0.025 # -2.5% daily loss limit )

Check signal before entry

is_approved, reason = risk_manager.check_signal(signal)

if not is_approved: logger.warning(f"[REJECTED] {reason}") return

Exit Management Integration

from AITRAPP.AITRAPP.packages.core.exits import ExitManager

exit_manager = ExitManager( stop_loss_type='ATR', # 'FIXED' or 'ATR' take_profit_levels=[0.5, 1.0, 1.5], # Staged TP trailing_stop=True, eod_exit_time='15:25' # IST )

Check exit conditions

exit_signal = exit_manager.check_exit(position, current_bar)

if exit_signal: # Execute exit pass

Backtest Integration

Running Backtest with Adapter

from AITRAPP.AITRAPP.packages.core.backtest import BacktestEngine from openalgo.strategies.adapters.your_strategy_adapter import YourStrategyAdapter

Create adapter

adapter = YourStrategyAdapter("SYMBOL", PARAMS)

Initialize backtest engine

engine = BacktestEngine( strategies=[adapter], start_date="2025-08-15", end_date="2025-11-10", capital=1000000 )

Run backtest

results = engine.run()

Get metrics

print(f"Total Return: {results['total_return']:.2f}%") print(f"Win Rate: {results['win_rate']:.2f}%") print(f"Max Drawdown: {results['max_drawdown']:.2f}%")

Common Integration Patterns

Pattern 1: Direct Adapter

Simple adapter that wraps OpenAlgo strategy

class SimpleAdapter(BaseStrategy): def init(self, openalgo_strategy): self.strategy = openalgo_strategy

def on_bar(self, bar):
    # Convert and run
    pass

Pattern 2: Enhanced Adapter

Adapter with additional AITRAPP features

class EnhancedAdapter(BaseStrategy): def init(self, openalgo_strategy, risk_manager, exit_manager): self.strategy = openalgo_strategy self.risk_manager = risk_manager self.exit_manager = exit_manager

def on_bar(self, bar):
    # Run strategy with risk and exit management
    pass

Pattern 3: Strategy Wrapper

Wrapper that converts strategy logic

class StrategyWrapper: def init(self, openalgo_strategy_class): self.strategy_class = openalgo_strategy_class

def create_aitrapp_strategy(self, symbol, params):
    # Create AITRAPP-compatible strategy
    pass

Troubleshooting

Import Errors

Issue: Cannot import AITRAPP modules

Solution:

Add AITRAPP to path

import sys from pathlib import Path

aitrapp_path = Path(file).parent.parent.parent / "AITRAPP" / "AITRAPP" sys.path.insert(0, str(aitrapp_path))

Data Format Mismatches

Issue: Data format doesn't match expected format

Solution:

• Check data conversion functions

• Verify column names and types

• Ensure datetime index is set correctly

Mock API Not Working

Issue: Mock API not returning expected data

Solution:

• Verify mock implementation matches real API

• Check data loading from AITRAPP

• Ensure proper error handling

Best Practices

• Separate concerns - Keep adapters focused on conversion only

• Preserve strategy logic - Don't modify original strategy code

• Test adapters - Verify adapter works with sample data

• Handle errors - Graceful fallbacks for missing data

• Document conversions - Clear documentation of format conversions

• Reuse mocks - Share mock implementations across adapters

Additional Resources

• Integration guide: openalgo/strategies/AITRAPP_INTEGRATION_GUIDE.md

• Adapter examples: openalgo/strategies/adapters/

• Mock implementations: openalgo/strategies/utils/openalgo_mock.py

• AITRAPP core: AITRAPP/AITRAPP/packages/core/