# Investing Algorithm Framework Investing Algorithm Framework is a comprehensive Python library for creating, backtesting, comparing, and deploying algorithmic trading strategies. It provides a complete ecosystem for quantitative trading, supporting both event-driven and vector backtesting modes, with built-in support for technical analysis, risk management (stop losses, take profits, position sizing), and professional-grade HTML reporting dashboards. The framework integrates with multiple cryptocurrency exchanges via CCXT, enabling strategies to be tested against historical data and deployed for live trading. The framework emphasizes a signal-based approach to strategy development, where users implement `generate_buy_signals` and `generate_sell_signals` methods that return pandas Series with boolean values. This design enables clean separation between signal generation logic and trade execution, while the framework handles portfolio management, order execution, and performance tracking automatically. Advanced features include position scaling (pyramiding), trailing stop losses, multiple data source support, parallel processing for parameter optimization, and cloud deployment to AWS Lambda or Azure Functions. ## Creating the Application The `create_app` function initializes the main application instance that orchestrates strategies, markets, and backtesting. ```python from investing_algorithm_framework import create_app, PortfolioConfiguration # Create the main application app = create_app() # Add market configuration app.add_market( market="bitvavo", trading_symbol="EUR" ) # Add portfolio configuration app.add_portfolio_configuration( PortfolioConfiguration( market="BITVAVO", trading_symbol="EUR", initial_balance=10000, fee_percentage=0.001, # 0.1% trading fee slippage_percentage=0.0005 # 0.05% slippage ) ) # Run the application (live trading) app.run() ``` ## TradingStrategy Class The `TradingStrategy` class is the base class for implementing trading logic. It supports signal-based trading with automatic position sizing, stop losses, and take profits. ```python from investing_algorithm_framework import ( TradingStrategy, TimeUnit, DataSource, DataType, PositionSize, StopLossRule, TakeProfitRule ) import pandas as pd class MovingAverageCrossover(TradingStrategy): # Required configuration algorithm_id = "MA-Crossover-Strategy" time_unit = TimeUnit.HOUR interval = 4 # Run every 4 hours symbols = ["BTC", "ETH"] trading_symbol = "EUR" # Data sources for market data data_sources = [ DataSource( identifier="btc_data", symbol="BTC/EUR", data_type=DataType.OHLCV, time_frame="4h", market="BITVAVO", warmup_window=100, # 100 candles for indicator calculation pandas=True ), DataSource( identifier="eth_data", symbol="ETH/EUR", data_type=DataType.OHLCV, time_frame="4h", market="BITVAVO", warmup_window=100, pandas=True ) ] # Position sizing (% of portfolio per symbol) position_sizes = [ PositionSize(symbol="BTC", percentage_of_portfolio=30), PositionSize(symbol="ETH", percentage_of_portfolio=20) ] # Risk management stop_losses = [ StopLossRule(symbol="BTC", percentage_threshold=5, trailing=False, sell_percentage=100), StopLossRule(symbol="ETH", percentage_threshold=5, trailing=True, sell_percentage=100) ] take_profits = [ TakeProfitRule(symbol="BTC", percentage_threshold=15, trailing=True, sell_percentage=50), TakeProfitRule(symbol="ETH", percentage_threshold=10, trailing=False, sell_percentage=100) ] def __init__(self, short_window=20, long_window=50): super().__init__() self.short_window = short_window self.long_window = long_window def generate_buy_signals(self, data): """Generate buy signals for each symbol.""" signals = {} for symbol, identifier in [("BTC", "btc_data"), ("ETH", "eth_data")]: df = data[identifier] close = df["Close"] short_ma = close.rolling(window=self.short_window).mean() long_ma = close.rolling(window=self.long_window).mean() # Golden cross: short MA crosses above long MA buy_signal = (short_ma > long_ma) & (short_ma.shift(1) <= long_ma.shift(1)) signals[symbol] = buy_signal.fillna(False) return signals def generate_sell_signals(self, data): """Generate sell signals for each symbol.""" signals = {} for symbol, identifier in [("BTC", "btc_data"), ("ETH", "eth_data")]: df = data[identifier] close = df["Close"] short_ma = close.rolling(window=self.short_window).mean() long_ma = close.rolling(window=self.long_window).mean() # Death cross: short MA crosses below long MA sell_signal = (short_ma < long_ma) & (short_ma.shift(1) >= long_ma.shift(1)) signals[symbol] = sell_signal.fillna(False) return signals # Register strategy with the app app.add_strategy(MovingAverageCrossover(short_window=20, long_window=50)) ``` ## Downloading Market Data The `download` function fetches historical OHLCV data from supported exchanges via CCXT. ```python from investing_algorithm_framework import download, download_v2, create_data_storage_path from datetime import datetime, timezone # Basic data download data = download( symbol="BTC/EUR", market="bitvavo", time_frame="1h", start_date="2024-01-01", end_date="2024-06-01", warmup_window=200, # Extra candles before start_date pandas=True, # Return as pandas DataFrame save=True, # Save to disk storage_path="./data/" ) print(f"Downloaded {len(data)} candles") print(data.head()) # Output columns: Open, High, Low, Close, Volume (indexed by Datetime) # Download with result object (returns both data and file path) result = download_v2( symbol="ETH/EUR", market="bitvavo", time_frame="4h", start_date="2024-01-01", end_date="2024-06-01", storage_path="./data/", filename="eth_eur_4h" # Custom filename ) print(f"Data shape: {result.data.shape}") print(f"Saved to: {result.path}") # Get the expected storage path without downloading path = create_data_storage_path( symbol="BTC/EUR", market="bitvavo", time_frame="1h", start_date=datetime(2024, 1, 1, tzinfo=timezone.utc), end_date=datetime(2024, 6, 1, tzinfo=timezone.utc), storage_path="./data/" ) # Output: ./data/OHLCV_BTC-EUR_BITVAVO_1h_2024-01-01-00-00_2024-06-01-00-00.csv ``` ## Backtesting Strategies Run event-driven or vector backtests to evaluate strategy performance with the `run_backtest` method. ```python from investing_algorithm_framework import ( create_app, BacktestDateRange, BacktestReport ) from datetime import datetime, timezone app = create_app() app.add_strategy(MovingAverageCrossover()) app.add_market(market="bitvavo", trading_symbol="EUR") # Define backtest period (dates must be rounded to the hour) backtest_range = BacktestDateRange( start_date=datetime(2023, 1, 1, 0, 0, tzinfo=timezone.utc), end_date=datetime(2024, 1, 1, 0, 0, tzinfo=timezone.utc), name="Full Year 2023" ) # Run event-driven backtest backtest = app.run_backtest( backtest_date_range=backtest_range, initial_amount=10000, risk_free_rate=0.04 # 4% for Sharpe ratio calculation ) # Access backtest metrics for run in backtest.get_all_backtest_runs(): metrics = run.backtest_metrics print(f"Total Return: {metrics.total_net_gain_percentage:.2f}%") print(f"CAGR: {metrics.cagr:.2f}") print(f"Sharpe Ratio: {metrics.sharpe_ratio:.2f}") print(f"Sortino Ratio: {metrics.sortino_ratio:.2f}") print(f"Max Drawdown: {metrics.max_drawdown:.2f}%") print(f"Win Rate: {metrics.win_rate:.2f}%") print(f"Total Trades: {metrics.number_of_trades_closed}") # Access individual trades for trade in run.trades: print(f" {trade.target_symbol}: {trade.net_gain:.2f} ({trade.opened_at} -> {trade.closed_at})") # Generate and display HTML report report = BacktestReport(backtest) report.show(browser=True) # Opens in default browser report.save("backtest_report.html") # Save to file ``` ## Multiple Backtest Date Ranges Test strategies across different market conditions using multiple date ranges. ```python from investing_algorithm_framework import BacktestDateRange # Define multiple test periods date_ranges = [ BacktestDateRange( start_date=datetime(2022, 1, 1, 0, 0, tzinfo=timezone.utc), end_date=datetime(2022, 12, 31, 0, 0, tzinfo=timezone.utc), name="Bear Market 2022" ), BacktestDateRange( start_date=datetime(2023, 1, 1, 0, 0, tzinfo=timezone.utc), end_date=datetime(2023, 12, 31, 0, 0, tzinfo=timezone.utc), name="Recovery 2023" ), BacktestDateRange( start_date=datetime(2024, 1, 1, 0, 0, tzinfo=timezone.utc), end_date=datetime(2024, 6, 1, 0, 0, tzinfo=timezone.utc), name="H1 2024" ) ] # Run backtests across all date ranges backtests = app.run_backtests( backtest_date_ranges=date_ranges, initial_amount=10000, backtest_storage_directory="./backtests/" # Save results ) # Compare strategies with BacktestReport report = BacktestReport.open(directory_path="./backtests/", show_progress=True) report.show(browser=True) ``` ## Order Creation Create market and limit orders with various sizing options using strategy methods. ```python from investing_algorithm_framework import TradingStrategy, TimeUnit, OrderSide class CustomOrderStrategy(TradingStrategy): time_unit = TimeUnit.HOUR interval = 1 symbols = ["BTC"] def apply_strategy(self, context, data): """Custom strategy logic with manual order management.""" symbol = "BTC" full_symbol = f"{symbol}/EUR" current_price = context.get_latest_price(full_symbol) # Market buy order - various sizing options self.create_market_buy_order( target_symbol="BTC", amount=0.01, # Buy 0.01 BTC ) self.create_market_buy_order( target_symbol="BTC", amount_trading_symbol=500, # Spend 500 EUR ) self.create_market_buy_order( target_symbol="BTC", percentage_of_portfolio=10, # Use 10% of portfolio ) # Market sell order self.create_market_sell_order( target_symbol="BTC", percentage_of_position=50, # Sell 50% of position ) # Limit order with explicit price self.create_limit_order( target_symbol="BTC", order_side=OrderSide.BUY, price=current_price * 0.98, # 2% below current price amount_trading_symbol=1000, execute=True, validate=True ) # Close entire position if self.has_position(symbol="BTC", amount_gt=0): self.close_position(symbol="BTC") ``` ## Position and Trade Management Access positions, trades, and orders within strategy methods. ```python class PositionManagementStrategy(TradingStrategy): time_unit = TimeUnit.HOUR interval = 1 symbols = ["BTC", "ETH"] def apply_strategy(self, context, data): # Check if position exists if self.has_position(symbol="BTC", amount_gt=0): position = self.get_position(symbol="BTC") print(f"BTC Position: {position.amount} @ avg cost {position.cost}") # Get all positions with non-zero amounts positions = self.get_positions(amount_gt=0) for pos in positions: print(f"{pos.symbol}: {pos.amount}") # Check for open orders if self.has_open_orders(target_symbol="BTC"): print("BTC has pending orders") # Access trades open_trades = self.get_open_trades(target_symbol="BTC") closed_trades = self.get_closed_trades() for trade in open_trades: print(f"Open trade: {trade.target_symbol}") print(f" Entry: {trade.open_price}") print(f" Current P/L: {trade.net_gain}") # Close a specific trade self.close_trade(trade=trade) # Trade event callbacks def on_trade_created(self, context, trade): print(f"Trade created: {trade.target_symbol}") def on_trade_opened(self, context, trade): print(f"Trade opened: {trade.target_symbol} @ {trade.open_price}") def on_trade_closed(self, context, trade): print(f"Trade closed: {trade.target_symbol}, P/L: {trade.net_gain}") def on_trade_stop_loss_triggered(self, context, trade): print(f"Stop loss triggered: {trade.target_symbol}") def on_trade_take_profit_triggered(self, context, trade): print(f"Take profit triggered: {trade.target_symbol}") ``` ## Position Scaling (Pyramiding) Scale into and out of positions using `ScalingRule` for advanced position management. ```python from investing_algorithm_framework import TradingStrategy, TimeUnit, DataSource, PositionSize, ScalingRule class PyramidingStrategy(TradingStrategy): time_unit = TimeUnit.HOUR interval = 4 symbols = ["BTC"] data_sources = [ DataSource( identifier="btc_data", symbol="BTC/EUR", time_frame="4h", warmup_window=50, market="BITVAVO" ) ] position_sizes = [ PositionSize(symbol="BTC", percentage_of_portfolio=20), ] scaling_rules = [ ScalingRule( symbol="BTC", max_entries=3, # Initial + 2 scale-ins scale_in_percentage=[50, 25], # 1st add 50%, 2nd add 25% of initial size scale_out_percentage=[25, 50], # 1st trim 25%, 2nd trim 50% max_position_percentage=50, # Never exceed 50% of portfolio cooldown_in_bars=3 # Wait 3 bars between actions ) ] def generate_buy_signals(self, data): """Buy signal also used for scale-in by default.""" df = data["btc_data"] close = df["Close"] ma20 = close.rolling(20).mean() return {"BTC": close > ma20} def generate_sell_signals(self, data): """Full exit signal.""" df = data["btc_data"] close = df["Close"] ma50 = close.rolling(50).mean() return {"BTC": close < ma50} def generate_scale_in_signals(self, data): """Optional: separate scale-in logic (defaults to buy signals).""" df = data["btc_data"] close = df["Close"] rolling_high = close.rolling(20).max() return {"BTC": close >= rolling_high} def generate_scale_out_signals(self, data): """Optional: partial close logic.""" df = data["btc_data"] close = df["Close"] volatility = close.pct_change().rolling(20).std() avg_vol = volatility.rolling(50).mean() return {"BTC": volatility > avg_vol * 2} ``` ## External Data Sources Load CSV, JSON, or Parquet data from URLs for sentiment analysis, macro indicators, or other external datasets. ```python from investing_algorithm_framework import TradingStrategy, DataSource, TimeUnit import polars as pl class SentimentStrategy(TradingStrategy): time_unit = TimeUnit.DAY interval = 1 symbols = ["BTC"] data_sources = [ # CSV from URL DataSource.from_csv( identifier="sentiment", url="https://example.com/crypto_sentiment.csv", date_column="date", date_format="%Y-%m-%d", cache=True, refresh_interval="1d", post_process=lambda df: df.with_columns( ((pl.col("score") - pl.col("score").mean()) / pl.col("score").std()).alias("z_score") ) ), # JSON from API DataSource.from_json( identifier="earnings", url="https://api.example.com/earnings.json", date_column="report_date", refresh_interval="1d" ), # Parquet from cloud storage DataSource.from_parquet( identifier="features", url="https://storage.example.com/features.parquet", date_column="date", refresh_interval="1W" ) ] def generate_buy_signals(self, data): sentiment_df = data["sentiment"] # Use sentiment z-score > 1 as buy signal buy_condition = sentiment_df["z_score"] > 1.0 return {"BTC": buy_condition} def generate_sell_signals(self, data): sentiment_df = data["sentiment"] # Use sentiment z-score < -1 as sell signal sell_condition = sentiment_df["z_score"] < -1.0 return {"BTC": sell_condition} ``` ## Backtest Report API Generate comprehensive HTML reports for single or multiple backtests with the `BacktestReport` class. ```python from investing_algorithm_framework import BacktestReport, Backtest # Single backtest report backtest = app.run_backtest(backtest_date_range=date_range, initial_amount=10000) report = BacktestReport(backtest) report.show(browser=True) # Display in browser report.save("report.html") # Save to file # Multi-strategy comparison strategy_a_backtest = app.run_backtest(...) strategy_b_backtest = app.run_backtest(...) comparison_report = BacktestReport(backtests=[strategy_a_backtest, strategy_b_backtest]) comparison_report.show() # Load backtests from directory report = BacktestReport.open( directory_path="./my_backtests/", show_progress=True # Show loading progress bar ) # Load specific backtests report = BacktestReport.open( backtests=[backtest_a, backtest_b], directory_path=["./backtests/strategy_a/", "./backtests/strategy_b/"] ) # Save and reload backtest backtest.save(directory_path="./saved_backtest/") loaded_backtest = Backtest.open(directory_path="./saved_backtest/") ``` ## Vector Backtesting Run fast vectorized backtests for parameter optimization and large-scale testing. ```python from investing_algorithm_framework import create_app, BacktestDateRange import os app = create_app() # Create multiple strategy variations strategies = [ MovingAverageCrossover(short_window=10, long_window=30), MovingAverageCrossover(short_window=20, long_window=50), MovingAverageCrossover(short_window=30, long_window=100), ] date_ranges = [ BacktestDateRange(start_date=datetime(2023, 1, 1, tzinfo=timezone.utc), end_date=datetime(2023, 6, 1, tzinfo=timezone.utc)), BacktestDateRange(start_date=datetime(2023, 6, 1, tzinfo=timezone.utc), end_date=datetime(2024, 1, 1, tzinfo=timezone.utc)), ] # Filter functions for early termination def window_filter(backtest_run): """Filter after each date range.""" return backtest_run.backtest_metrics.total_return > 0 def final_filter(backtest): """Filter at the end.""" return backtest.backtest_summary.sharpe_ratio > 1.0 # Run parallel vector backtests backtests = app.run_vector_backtests( strategies=strategies, backtest_date_ranges=date_ranges, initial_amount=10000, n_workers=os.cpu_count() - 1, # Parallel processing use_checkpoints=True, # Resume interrupted backtests backtest_storage_directory="./optimization_results/", window_filter_function=window_filter, final_filter_function=final_filter ) # Display results report = BacktestReport(backtests=backtests) report.show() ``` ## Summary Investing Algorithm Framework provides a comprehensive platform for algorithmic trading strategy development, from initial concept through production deployment. The signal-based architecture enables clean, testable strategy code while the framework handles complex tasks like order execution, position management, and risk control automatically. Key use cases include systematic trading strategy development, historical performance analysis through backtesting, strategy comparison and optimization, and live trading deployment on cryptocurrency exchanges. The framework excels at bridging the gap between research and production. Strategies can be rapidly prototyped using vector backtesting (10-100x faster than event-driven), refined with realistic event-driven simulation that includes stop losses and take profits, and then deployed to cloud environments with minimal code changes. The integrated HTML reporting dashboard enables visual comparison of multiple strategies across different market conditions, with over 30 performance metrics including CAGR, Sharpe ratio, Sortino ratio, max drawdown, win rate, and detailed trade-level analysis. Whether building simple moving average crossovers or complex multi-asset strategies with position scaling and external data integration, the framework provides the tools needed for professional quantitative trading.