deltao 0.3.1

Algorithmic trading framework for Bittensor dTao

deltao

Bittensor dTao Trading Framework - Algorithmic trading strategies for dTao subnet tokens.

Installation

# Clone the repository
git clone <repo-url>
cd deltao

# Install with uv (recommended)
uv sync

# For backtesting support (pandas, numpy)
uv sync --extra backtest

Quick Start

1. Basic Strategy Usage

from decimal import Decimal
from deltao.strategy import MomentumStrategy, BasicRiskManager
from deltao.core import Engine, EngineState, MarketEvent
from deltao.core.clock import HistoricalClock
from deltao.connector import MockConnector

# Create strategy and risk manager
strategy = MomentumStrategy(
    fast_period=10,
    slow_period=30,
    min_momentum_pct=Decimal("0.02"),
)
risk = BasicRiskManager(
    max_position_size=Decimal("50"),
    max_exposure_pct=Decimal("0.3"),
)

# Initialize engine
clock = HistoricalClock(start_time=datetime.now())
state = EngineState(
    tao_balance=Decimal("1000"),
    alpha_balance=Decimal("0"),
    trading_enabled=True,
)
connector = MockConnector()

engine = Engine(
    clock=clock,
    state=state,
    strategy=strategy,
    risk=risk,
    connector=connector,
)

# Process market events
event = MarketEvent(
    netuid=8,
    price=Decimal("1.5"),
    tao_in=Decimal("10000"),
    alpha_in=Decimal("6666"),
    k=Decimal("66660000"),
    block_number=1000000,
    timestamp=datetime.now(),
)

audit = await engine.process(event)
print(f"Orders generated: {len(audit.orders_generated)}")
print(f"Orders approved: {len(audit.orders_approved)}")

2. Running a Backtest

import asyncio
from decimal import Decimal
from deltao.backtest import BacktestingEngine, BacktestConfig, create_sample_data, save_block_data
from deltao.strategy import MomentumStrategy, BasicRiskManager
from pathlib import Path

async def run_backtest():
    # Create sample data (or use your own parquet files)
    data = create_sample_data(
        num_blocks=10000,
        start_block=1_000_000,
        base_price=1.0,
        volatility=0.02,
    )

    data_dir = Path("data/candles")
    save_block_data(data, netuid=8, output_dir=data_dir)

    # Configure backtest
    config = BacktestConfig(
        netuid=8,
        start_block=1_000_000,
        end_block=1_010_000,
        initial_tao=Decimal("1000"),
        initial_alpha=Decimal("0"),
        trade_fee=Decimal("0.001"),  # 0.1% fee
        data_dir=data_dir,
    )

    # Create strategy
    strategy = MomentumStrategy(
        fast_period=10,
        slow_period=30,
        min_momentum_pct=Decimal("0.01"),
    )
    risk = BasicRiskManager(max_position_size=Decimal("50"))

    # Run backtest
    engine = BacktestingEngine(data_dir=data_dir)
    result = await engine.run(config, strategy, risk)

    # Print results
    print(result.summary())

asyncio.run(run_backtest())

3. Using Different Strategies

Momentum Strategy

Follows trends using moving average crossovers.

from deltao.strategy import MomentumStrategy

strategy = MomentumStrategy(
    fast_period=10,          # Fast MA period
    slow_period=30,          # Slow MA period
    min_momentum_pct=Decimal("0.02"),  # Min momentum threshold
    base_order_size=Decimal("10"),
    momentum_multiplier=Decimal("1.0"),
)

Mean Reversion Strategy

Trades against extreme moves using RSI and price deviation.

from deltao.strategy import MeanReversionStrategy

strategy = MeanReversionStrategy(
    rsi_period=14,
    sma_period=20,
    oversold_threshold=Decimal("30"),   # Buy when RSI < 30
    overbought_threshold=Decimal("70"), # Sell when RSI > 70
    min_deviation=Decimal("0.05"),      # 5% min deviation from SMA
    base_order_size=Decimal("10"),
)

4. Risk Management

from deltao.strategy import BasicRiskManager, PositionLimitRiskManager, NoOpRiskManager

# Basic risk checks
risk = BasicRiskManager(
    max_position_size=Decimal("50"),    # Max TAO per order
    max_exposure_pct=Decimal("0.3"),    # Max 30% of balance per trade
    min_balance_reserve=Decimal("0.1"), # Keep 10% as reserve
)

# Position-based limits
risk = PositionLimitRiskManager(
    max_total_exposure=Decimal("500"),  # Max total exposure
    max_single_order=Decimal("50"),     # Max per order
)

# No risk checks (for testing)
risk = NoOpRiskManager()

5. Live Trading with DTaoConnector

from deltao.connector import DTaoConnector
from bittensor import wallet, subtensor

# Initialize connector
connector = DTaoConnector(
    wallet=wallet(name="my_wallet"),
    netuid=8,
    network="finney",
)

# Use with engine
engine = Engine(
    clock=LiveClock(),
    state=state,
    strategy=strategy,
    risk=risk,
    connector=connector,
)

Project Structure

deltao/
├── src/deltao/
│   ├── core/           # Core engine, state, events
│   │   ├── engine.py   # Event processor
│   │   ├── state.py    # Engine state
│   │   ├── clock.py    # Live/Historical clocks
│   │   ├── event.py    # Event types
│   │   └── agent.py    # Strategy agent wrapper
│   │
│   ├── strategy/       # Trading strategies
│   │   ├── base.py     # Base strategy + indicators
│   │   ├── risk.py     # Risk managers
│   │   └── directional/
│   │       ├── momentum.py
│   │       └── mean_reversion.py
│   │
│   ├── backtest/       # Backtesting engine
│   │   ├── engine.py   # Backtest runner
│   │   ├── recorder.py # Metrics calculation
│   │   ├── config.py   # Configuration
│   │   └── data.py     # Data loading/saving
│   │
│   └── connector/      # Chain connectors
│       ├── mock.py     # Mock for testing
│       └── dtao.py     # Bittensor connector
│
└── tests/              # Test suite

Data Format

Historical data uses parquet files with naming convention:

dtao|subnet{netuid}|1block.parquet

Required columns:

• timestamp: Unix timestamp
• block_number: Block number
• price: TAO/Alpha price
• tao_in: TAO in pool
• alpha_in: Alpha in pool
• k: Constant product (tao_in * alpha_in)

Backtest Metrics

The backtest result includes:

Metric	Description
sharpe_ratio	Risk-adjusted return (annualized)
sortino_ratio	Downside risk-adjusted return
max_drawdown	Maximum peak-to-trough decline
win_rate	Percentage of winning trades
profit_factor	Gross profit / Gross loss
total_trades	Number of trades executed
net_pnl	Net profit/loss after fees

Custom Strategy

Create your own strategy by implementing the AlgoStrategy protocol:

from deltao.core import AlgoStrategy, EngineState
from deltao.core.types import OrderRequest, Side

class MyStrategy:
    def generate_orders(
        self, state: EngineState
    ) -> tuple[list[OrderRequest], list[OrderRequest]]:
        """
        Returns: (orders_to_cancel, orders_to_open)
        """
        cancels = []
        opens = []

        # Your logic here
        if should_buy(state):
            opens.append(OrderRequest(
                side=Side.BUY,
                amount=Decimal("10"),
                slippage_tolerance=Decimal("0.02"),
            ))

        return (cancels, opens)

Development

# Run tests
uv run pytest

# Run with backtest dependencies
uv run --extra backtest pytest

# Lint
uv run ruff check .

# Type check
uv run mypy src/deltao

License

MIT