Investing Algorithm Framework (coding-kitties/investing-algorithm-framework)

Investing Algorithm Framework

https://github.com/coding-kitties/investing-algorithm-framework
Admin
A Python framework for creating, backtesting, comparing, and deploying trading strategies with...

Tokens:158,714
Snippets:1,331
Trust Score:7.1
Update:1 week ago
Show doc for...
Context Summary (auto-generated)
Raw
# 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.