Development Guide

Overview

This section is designed for quantitative developers and strategy creators. It will walk you through the framework provided for building, testing, and deploying your own automated trading strategies on the platform.

The Foundation: The BaseStrategy Framework

All strategies inherit from the BaseStrategy abstract class. This class provides the core structure and lifecycle methods that the platform’s runtime uses to manage and interact with your strategy.

Code Reference: platform/strategies/base.py


class BaseStrategy(ABC):
    def __init__(self, strategy_id: UUID, manifest: StrategyManifest, config: Dict[str, Any]):
        # ... initialization ...
 
    @abstractmethod
    async def generate_intents(self, market_data: Dict[str, Any]) -> List[Intent]:
        """
        The core logic of the strategy. This method is called periodically
        by the platform with the latest market data. It should return a list
        of Intents to be executed.
        """
        pass
 
    async def _on_initialize(self) -> None:
        """Called once when the strategy is started."""
        pass
 
    async def _on_shutdown(self) -> None:
        """Called once when the strategy is stopped."""
        pass
 
    # ... other helper methods for state, performance tracking, etc. ...

Key Components:

StrategyManifest: A dataclass that provides metadata about your strategy to the platform, such as its name, version, and any ml_models it requires
_on_initialize(): This is your strategy’s setup method. Use it to load historical data, initialize machine learning models, or set up any initial state
generate_intents(market_data): This is the heart of your strategy. The platform will call this method on a schedule or trigger. You are provided with the latest market_data and are expected to return a list of Intent objects

Example Strategy: MomentumStrategy

The codebase includes a sophisticated MomentumStrategy that serves as an excellent reference implementation.

Code Reference: platform/strategies/examples/momentum.py

How It Works

Feature Calculation (MomentumFeatures): It uses the Polars library for high-performance, in-memory data manipulation. The calculate_momentum_features static method takes a DataFrame of price history and computes technical indicators like moving averages, RSI, and volatility ratios.
ML Integration (_predict_signal): The strategy uses a pre-trained ONNX model to generate a “signal strength” between 0 and 1. It calls self.predict_with_model(...), a helper method provided by BaseStrategy, which handles the complexities of running the ONNX inference session.
Risk Management (RiskScorer): Before deciding on a trade size, it calculates a risk score, blending ML predictions with heuristics. This score is then used to scale down the position size.
Intent Generation (_generate_asset_intent): This is where everything comes together:
- It takes the signal strength and risk score
- It determines the intent_type (ACQUIRE for a strong signal, DISPOSE for a weak one)
- It calculates the trade amount
- Finally, it constructs and returns a fully-formed Intent object

Tutorial: Building a Simple DCA Strategy

Let’s create a new strategy from scratch. Our goal is simple: every day, we will buy $100 worth of WETH on Ethereum.

Step 1: Create the Strategy File

Create a new file: platform/strategies/examples/dca.py

Step 2: Define the Strategy Class and Manifest


import uuid
from datetime import datetime, timedelta
from typing import Any, Dict, List
from decimal import Decimal
 
from platform.strategies.base import BaseStrategy, StrategyManifest
from platform.types.common import get_asset_by_key
from platform.types.intent import (
    AssetSpec,
    Intent,
    IntentConstraints,
    IntentType,
    ExecutionStyle,
)
 
# A simple state class to keep track of our last purchase
class DcaState:
    last_purchase_time: datetime = datetime.min
 
class DcaStrategy(BaseStrategy):
    """
    A simple strategy that performs a daily Dollar-Cost Averaging
    purchase of a specific asset.
    """
 
    def __init__(self, strategy_id: uuid.UUID, config: Dict[str, Any]):
        manifest = StrategyManifest(
            name="SimpleDCA",
            version="1.0.0",
            description="Performs a daily purchase of a configured asset.",
            # No ML models needed for this simple strategy
            ml_models=[],
            # Define the configuration options this strategy accepts
            config_schema={
                "purchase_asset_key": {"type": "string", "required": True},
                "quote_asset_key": {"type": "string", "required": True},
                "purchase_amount_quote": {"type": "number", "required": True},
            },
        )
        super().__init__(strategy_id, manifest, config)
        self.state = DcaState()
 
    async def _on_initialize(self) -> None:
        """
        Called when the strategy starts. We'll just log a message.
        """
        print(f"DCA Strategy {self.strategy_id} initialized.")
 
    async def generate_intents(self, market_data: Dict[str, Any]) -> List[Intent]:
        """
        This is our core logic. We check if 24 hours have passed
        since our last purchase. If so, we create a new ACQUIRE intent.
        """
        time_since_last_purchase = datetime.utcnow() - self.state.last_purchase_time
        if time_since_last_purchase < timedelta(days=1):
            # Not time to buy yet, do nothing.
            return []
 
        print("DCA trigger! Generating a new purchase intent.")
 
        # Get asset details from our config
        purchase_asset = get_asset_by_key(self.config["purchase_asset_key"])
        quote_asset = get_asset_by_key(self.config["quote_asset_key"])
        purchase_amount = Decimal(str(self.config["purchase_amount_quote"]))
 
        if not purchase_asset or not quote_asset:
            print("Error: Could not find assets in registry.")
            return []
 
        # Construct the Intent object
        dca_intent = Intent(
            strategy_id=self.strategy_id,
            type=IntentType.ACQUIRE,
            assets=[
                # We want to acquire the target asset, but we don't specify the amount
                # because we are defining our size by the quote asset.
                AssetSpec(asset=purchase_asset),
                # And we will spend a fixed amount of the quote asset.
                AssetSpec(asset=quote_asset, amount=purchase_amount),
            ],
            constraints=IntentConstraints(
                max_slippage=Decimal("0.01"),  # 1% slippage tolerance
                time_window_ms=600000, # 10 minute execution window
                execution_style=ExecutionStyle.ADAPTIVE,
            ),
        )
 
        # Update our state and return the intent
        self.state.last_purchase_time = datetime.utcnow()
        return [dca_intent]

Step 3: Run the Strategy

While the full strategy runtime is managed by the platform, you could test the generate_intents logic in a simple test script. The platform would discover this new strategy, and a user could activate it via the API, providing a configuration like:


{
  "purchase_asset_key": "WETH_ETHEREUM",
  "quote_asset_key": "USDC_ETHEREUM",
  "purchase_amount_quote": 100.0
}

Once activated, the platform’s scheduler would call generate_intents periodically, and once every 24 hours, your strategy would submit a new intent to the system, which would then be executed automatically.

Strategy Development Best Practices

1. State Management

Immutable State: Keep strategy state immutable and use copy-on-write patterns
Persistence: Use the platform’s state management for any data that needs to survive restarts
Validation: Always validate configuration parameters in _on_initialize()

2. Error Handling

Graceful Degradation: If your strategy encounters an error, return an empty list of intents rather than crashing
Logging: Use structured logging to track strategy behavior and debug issues
Monitoring: Implement health checks and performance metrics

3. Performance Optimization

Efficient Data Processing: Use libraries like Polars for high-performance data manipulation
Caching: Cache expensive computations and reuse results when possible
Async Operations: Use async/await for I/O operations to avoid blocking

4. Risk Management

Position Sizing: Always implement position sizing logic based on risk parameters
Stop Losses: Consider implementing stop-loss mechanisms for risk control
Diversification: Avoid over-concentration in single assets or strategies

Testing Your Strategy

Unit Testing


# tests/test_dca_strategy.py
import pytest
from datetime import datetime, timedelta
from platform.strategies.examples.dca import DcaStrategy
 
@pytest.fixture
def dca_strategy():
    config = {
        "purchase_asset_key": "WETH_ETHEREUM",
        "quote_asset_key": "USDC_ETHEREUM",
        "purchase_amount_quote": 100.0
    }
    return DcaStrategy(uuid.uuid4(), config)
 
async def test_dca_strategy_generates_intent_on_first_run(dca_strategy):
    # First run should generate an intent
    intents = await dca_strategy.generate_intents({})
    assert len(intents) == 1
    assert intents[0].type == IntentType.ACQUIRE
 
async def test_dca_strategy_does_not_generate_intent_too_soon(dca_strategy):
    # Generate first intent
    await dca_strategy.generate_intents({})
    
    # Second run should not generate another intent
    intents = await dca_strategy.generate_intents({})
    assert len(intents) == 0

Integration Testing


# tests/integration/test_strategy_integration.py
import pytest
from platform.strategies.base import BaseStrategy
 
async def test_strategy_lifecycle():
    # Test strategy initialization
    strategy = DcaStrategy(uuid.uuid4(), config)
    await strategy._on_initialize()
    
    # Test intent generation
    intents = await strategy.generate_intents(market_data)
    assert len(intents) > 0
    
    # Test strategy shutdown
    await strategy._on_shutdown()

Advanced Strategy Patterns

1. Multi-Asset Strategies


async def generate_intents(self, market_data: Dict[str, Any]) -> List[Intent]:
    intents = []
    
    # Generate intents for multiple assets
    for asset_config in self.config["assets"]:
        intent = await self._generate_asset_intent(asset_config, market_data)
        if intent:
            intents.append(intent)
    
    return intents

2. Conditional Execution


async def generate_intents(self, market_data: Dict[str, Any]) -> List[Intent]:
    # Check market conditions
    if not self._should_trade(market_data):
        return []
    
    # Check risk limits
    if not self._check_risk_limits():
        return []
    
    # Generate intents
    return await self._generate_trading_intents(market_data)

3. ML-Enhanced Strategies


async def _predict_signal(self, features: np.ndarray) -> float:
    # Use the platform's ML infrastructure
    prediction = await self.predict_with_model(
        model_name="momentum_predictor",
        input_data=features
    )
    return prediction.signal_strength

Strategy Configuration

Configuration Schema

The config_schema in your StrategyManifest defines what configuration options your strategy accepts:


config_schema={
    "purchase_asset_key": {
        "type": "string", 
        "required": True,
        "description": "Asset to purchase (e.g., WETH_ETHEREUM)"
    },
    "purchase_amount_quote": {
        "type": "number", 
        "required": True,
        "min": 0.01,
        "max": 10000,
        "description": "Amount to spend in quote currency"
    },
    "execution_frequency": {
        "type": "string",
        "enum": ["daily", "weekly", "monthly"],
        "default": "daily",
        "description": "How often to execute purchases"
    }
}

Environment-Specific Configuration


async def _on_initialize(self) -> None:
    # Load environment-specific configuration
    env = os.getenv("ENVIRONMENT", "development")
    if env == "production":
        self.risk_multiplier = 0.5  # More conservative in production
    else:
        self.risk_multiplier = 1.0  # Full risk in development

Monitoring and Observability

Performance Metrics


async def generate_intents(self, market_data: Dict[str, Any]) -> List[Intent]:
    start_time = time.time()
    
    try:
        intents = await self._generate_intents_internal(market_data)
        
        # Record performance metrics
        execution_time = time.time() - start_time
        self.record_metric("intent_generation_time", execution_time)
        self.record_metric("intents_generated", len(intents))
        
        return intents
    except Exception as e:
        self.record_metric("intent_generation_errors", 1)
        raise

Health Checks


async def health_check(self) -> Dict[str, Any]:
    return {
        "status": "healthy",
        "last_intent_generation": self.state.last_intent_time,
        "total_intents_generated": self.state.total_intents,
        "errors_last_24h": self.state.errors_last_24h
    }

Deployment and Operations

Strategy Registration

Strategies are automatically discovered by the platform when placed in the platform/strategies/ directory. The platform will:

Load the strategy class
Validate the configuration schema
Register the strategy with the runtime
Make it available for activation via the API

Strategy Activation

Users can activate strategies through the API:


curl -X POST http://localhost:8000/strategies/activate \
  -H "Content-Type: application/json" \
  -d '{
    "strategy_name": "SimpleDCA",
    "config": {
      "purchase_asset_key": "WETH_ETHEREUM",
      "quote_asset_key": "USDC_ETHEREUM",
      "purchase_amount_quote": 100.0
    }
  }'

Monitoring Active Strategies


# List active strategies
curl http://localhost:8000/strategies/active
 
# Get strategy status
curl http://localhost:8000/strategies/{strategy_id}/status
 
# Deactivate strategy
curl -X POST http://localhost:8000/strategies/{strategy_id}/deactivate

Next Steps

Backend Engineers: Return to Backend Engine to understand how intents are processed
Frontend Developers: Check Frontend Dashboard to see how strategies are managed in the UI
System Architects: Explore Integration & Operations for advanced deployment patterns
Quantitative Developers: Start building your own strategies using the examples and patterns provided