Configuration Guide

Configure HiveQ Flow backtests, strategies, and data sources.

Overview

Flow uses configuration objects and dictionaries to control backtest behavior, strategy parameters, and data sources. Configuration is passed directly to hf.run_backtest().

Basic Backtest

Minimal configuration for running a backtest:

import hiveq.flow as hf
from hiveq.flow import StrategyConfig

hf.init(trader_id='T123', api_key='KEY')

report = hf.run_backtest(
    strategy_configs=[StrategyConfig(name='MyStrat', type='MyStrategy', params={})],
    symbols=['AAPL'],
    start_date='2025-01-01',  # Required
    end_date='2025-01-31',    # Required - must be AFTER start_date (T+1 minimum)
    data_configs=[{
        'type': 'hiveq_historical',
        'dataset': 'HIVEQ_US_EQ',
        'schema': ['bars_1m']
    }]
)

CRITICAL: end_date must be AFTER start_date (minimum T+1). Cannot be equal.

Strategy Configuration

StrategyConfig

from hiveq.flow import StrategyConfig

StrategyConfig(
    name: str,          # Unique strategy name
    type: str,          # Strategy class name (as string)
    params: dict = {},  # Strategy parameters
    symbols: list[str] = None  # Optional symbol list
)

Examples

Single Strategy:

strategy_configs = [
    StrategyConfig(
        name='momentum',
        type='MomentumStrategy',
        params={'lookback': 20, 'threshold': 1.5}
    )
]

Multiple Strategies:

strategy_configs = [
    StrategyConfig(
        name='momentum_fast',
        type='MomentumStrategy',
        params={'lookback': 10, 'threshold': 2.0},
        symbols=['AAPL', 'MSFT']
    ),
    StrategyConfig(
        name='momentum_slow',
        type='MomentumStrategy',
        params={'lookback': 30, 'threshold': 1.0},
        symbols=['GOOGL', 'TSLA']
    )
]

Accessing Parameters in Strategy:

class MyStrategy:
    def __init__(self):
        # Access via ctx.strategy_config.params in callbacks
        pass

    def on_start(self, ctx: hf.Context, event: events.StartEvent) -> None:
        # Access strategy configuration
        params = ctx.strategy_config.params
        self.lookback = params.get('lookback', 20)
        self.threshold = params.get('threshold', 1.5)

Backtest Configuration

BacktestConfig (Optional)

Use BacktestConfig only for custom capital, commissions, slippage, or deploy mode:

from hiveq.flow.config import BacktestConfig

backtest_config = BacktestConfig(
    start_date: str = None,         # 'YYYY-MM-DD' (or pass to run_backtest)
    end_date: str = None,           # 'YYYY-MM-DD' (must be > start_date)
    initial_capital: float = 1000000.0,
    commission: float = 0.001,      # 0.1%
    slippage: float = 0.0,
    benchmark: str = None,          # e.g., 'SPY'
    risk_free_rate: float = 0.02,   # 2%
    deploy: bool = False            # Enable remote monitoring
)

# Pass to run_backtest
report = hf.run_backtest(
    strategy_configs=strategies,
    symbols=symbols,
    data_configs=data_configs,
    backtest_config=backtest_config  # Optional
)

Note: If you only need to set dates, pass them directly to run_backtest() instead of creating BacktestConfig.

Deploy Mode

Set deploy=True to enable:

Remote monitoring of backtest runs
Parameter change tracking
Performance metrics publishing
Event publishing to data API

backtest_config = BacktestConfig(
    initial_capital=100000,
    commission=0.001,
    deploy=True  # Enable remote monitoring
)

Data Configuration

HiveQ Historical Data (Recommended)

data_configs = [{
    'type': 'hiveq_historical',
    'dataset': 'HIVEQ_US_EQ',      # Dataset name
    'schema': ['bars_1m']          # Data schema
}]

Available Datasets:

HIVEQ_US_EQ - US Equities (bars, TBBO)
HIVEQ_US_FUT - US Futures (bars, TBBO)
HIVEQ_US_OPT - US Options (snapshots)
HIVEQ_QUANT_SIGNALS - Quant Signals (custom signal data)

Available Schemas:

Bars: bars_1s, bars_1m, bars_5m, bars_15m, bars_30m, bars_1h, bars_1d
TBBO: tbbo (top of book quotes + trades)
Options: snaps_1s (1-second options snapshots)
Signals: signals (for HIVEQ_QUANT_SIGNALS)

Multiple Data Sources

data_configs = [
    {
        'type': 'hiveq_historical',
        'dataset': 'HIVEQ_US_EQ',
        'schema': ['bars_1m']
    },
    {
        'type': 'hiveq_historical',
        'dataset': 'HIVEQ_US_FUT',
        'schema': ['bars_5m']
    },
    {
        'type': 'csv',
        'data_type': 'custom',
        'id': 'my_signals',
        'path': 'signals/data.csv'
    }
]

CSV Data

data_configs = [{
    'type': 'csv',
    'data_type': 'bars_1m',       # or 'custom'
    'id': '1_MIN_BAR',            # ID to reference
    'path': 'data/bars.csv'       # Relative path
}]

CSV Format (bars): symbol, timestamp, open, high, low, close, volume

CSV Format (custom): symbol, timestamp, [custom columns...]

Accessing CSV Data:

def on_start(self, ctx: hf.Context, event: events.StartEvent) -> None:
    ctx.subscribe_data(data_id='my_signals')

def on_custom_data(self, ctx: hf.Context, event: events.CustomDataEvent) -> None:
    data = event.data()
    signal = data.column_data('signal', default=0)
    value = data.column_data('value', default=None)

HiveQ Quant Signals

data_configs = [
    {
        'type': 'hiveq_historical',
        'dataset': 'HIVEQ_US_FUT',
        'schema': ['bars_1m']
    },
    {
        'type': 'hiveq_historical',
        'dataset': 'HIVEQ_QUANT_SIGNALS',
        'schema': ['signals'],
        'id': 'hiveq_signals',
        'symbols': ['test_signal']  # Signal identifiers
    }
]

Accessing Signal Data:

import json

def on_start(self, ctx: hf.Context, event: events.StartEvent) -> None:
    ctx.subscribe_data(data_id='hiveq_signals')

def on_custom_data(self, ctx: hf.Context, event: events.CustomDataEvent) -> None:
    data = event.data()
    signal_json = data.column_data('signal_json', default=None)

    if signal_json:
        signal_data = json.loads(signal_json)
        ticker = signal_data.get('ticker')
        flag = signal_data.get('flag')  # 'long', 'short', 'flat'
        weight = float(signal_data.get('weight1', 0.0))

Complete Configuration Examples

Example 1: Simple Equity Backtest

import hiveq.flow as hf
from hiveq.flow import StrategyConfig

hf.init(trader_id='T123', api_key='KEY')

report = hf.run_backtest(
    strategy_configs=[
        StrategyConfig(name='SMA', type='SMACrossover', params={'fast': 10, 'slow': 30})
    ],
    symbols=['AAPL', 'MSFT'],
    start_date='2025-01-01',
    end_date='2025-12-31',
    data_configs=[{
        'type': 'hiveq_historical',
        'dataset': 'HIVEQ_US_EQ',
        'schema': ['bars_1d']
    }]
)

Example 2: Futures with Custom Parameters

from hiveq.flow.config import BacktestConfig

hf.init(trader_id='T123', api_key='KEY')

backtest_config = BacktestConfig(
    initial_capital=50000,
    commission=0.002,  # 0.2%
    slippage=0.0005,   # 0.05%
    benchmark='ES.c.0'
)

report = hf.run_backtest(
    strategy_configs=[
        StrategyConfig(
            name='futures_trend',
            type='FuturesTrendStrategy',
            params={'contracts_per_trade': 1, 'lookback': 20}
        )
    ],
    symbols=['ES.c.0', 'NQ.v.0'],
    start_date='2025-01-01',
    end_date='2025-03-31',
    data_configs=[{
        'type': 'hiveq_historical',
        'dataset': 'HIVEQ_US_FUT',
        'schema': ['bars_5m']
    }],
    backtest_config=backtest_config
)

Example 3: Options Trading

hf.init(trader_id='T123', api_key='KEY')

report = hf.run_backtest(
    strategy_configs=[
        StrategyConfig(
            name='options_spread',
            type='OptionsSpreadStrategy',
            params={'target_delta': 0.3, 'max_spread': 0.10}
        )
    ],
    symbols=['SPX'],
    start_date='2025-01-01',
    end_date='2025-01-05',
    data_configs=[{
        'type': 'hiveq_historical',
        'dataset': 'HIVEQ_US_OPT',
        'schema': ['snaps_1s']
    }]
)

Example 4: Multi-Asset with CSV Signals

hf.init(trader_id='T123', api_key='KEY')

report = hf.run_backtest(
    strategy_configs=[
        StrategyConfig(
            name='signal_strategy',
            type='SignalBasedStrategy',
            params={'signal_threshold': 0.7}
        )
    ],
    symbols=['AAPL', 'ES.c.0'],
    start_date='2025-01-01',
    end_date='2025-03-31',
    data_configs=[
        {
            'type': 'hiveq_historical',
            'dataset': 'HIVEQ_US_EQ',
            'schema': ['bars_1m']
        },
        {
            'type': 'hiveq_historical',
            'dataset': 'HIVEQ_US_FUT',
            'schema': ['bars_1m']
        },
        {
            'type': 'csv',
            'data_type': 'custom',
            'id': 'my_signals',
            'path': 'signals/combined_signals.csv'
        }
    ]
)

Configuration Best Practices

Start Simple: Begin with minimal configuration, add complexity as needed
Date Validation: Always ensure end_date > start_date (minimum T+1)
Data Frequency: Start with daily bars for development, then test intraday
Commission Settings: Use realistic commission rates for production testing
Deploy Mode: Use deploy=True for remote monitoring and collaboration
CSV Paths: Use relative paths for portability
Multiple Strategies: Test strategies independently before combining
Symbol Lists: Keep symbol lists manageable for faster backtests

Common Issues

Issue: "end_date must be after start_date"

Cause: end_date == start_date or end_date < start_date
Solution: Ensure end_date is at least T+1 after start_date

Issue: ImportError for OrderType/OrderSide

Cause: Importing from wrong module
Solution: Import from hiveq.flow.trading_types, not hiveq.flow.config

Issue: CSV file not found

Cause: Incorrect path in data_configs
Solution: Use relative paths from working directory

Issue: Strategy class not found

Cause: type parameter doesn't match actual class name
Solution: Ensure type='ClassName' matches the Python class name exactly

Next Steps

Strategy Development: Implement trading strategies

Document Version: 1.0 Last Updated: 2025-11-20

Configuration

On this page