ML Probability Trading Signals

Article Strategy library · Author: hummingbot

Summary

This Hummingbot controller listens for machine-learning signals delivered through an MQTT topic. Each signal provides probabilities for short, neutral, and long outcomes; configurable short and long thresholds determine whether the controller records a short signal, a long signal, or no directional signal. The received signal is also retained as feature data for status reporting and executor configuration.

When creating a position executor, the controller passes through the connector, trading pair, side, price, amount, and leverage. It adjusts the executor's triple-barrier volatility setting using a target percentage supplied with the signal, falling back to 0.01 when that feature is absent. The document is implementation code rather than a strategy evaluation: it gives no model-training details, validation method, trading results, or guidance on probability calibration. The listener's signal format and thresholds therefore describe the interface and decision rule, not evidence that the predictions are reliable or profitable.

Key ideas

The controller receives short, neutral, and long probabilities from an external machine-learning signal feed.
Separate configurable thresholds map sufficiently high short or long probabilities to directional signals.
Incoming signal data is retained for status display and position-executor setup.
A supplied target percentage adjusts the executor's volatility setting, with a fallback value when absent.
The document provides no model validation or trading performance evidence.

Tags

Machine learning Execution Risk management Crypto

Full text

# AILivestreamController


# AILivestreamController









## Source (Apache-2.0)

```python
from decimal import Decimal
from typing import List

import pandas_ta as ta  # noqa: F401
from pydantic import Field

from hummingbot.core.data_type.common import TradeType
from hummingbot.remote_iface.mqtt import ExternalTopicFactory
from hummingbot.strategy_v2.controllers.directional_trading_controller_base import (
    DirectionalTradingControllerBase,
    DirectionalTradingControllerConfigBase,
)
from hummingbot.strategy_v2.executors.position_executor.data_types import PositionExecutorConfig


class AILivestreamControllerConfig(DirectionalTradingControllerConfigBase):
    controller_name: str = "ai_livestream"
    long_threshold: float = Field(default=0.5, json_schema_extra={"is_updatable": True})
    short_threshold: float = Field(default=0.5, json_schema_extra={"is_updatable": True})
    topic: str = "hbot/predictions"


class AILivestreamController(DirectionalTradingControllerBase):
    def __init__(self, config: AILivestreamControllerConfig, *args, **kwargs):
        self.config = config
        super().__init__(config, *args, **kwargs)
        # Start ML signal listener
        self._init_ml_signal_listener()

    def _init_ml_signal_listener(self):
        """Initialize a listener for ML signals from the MQTT broker"""
        try:
            normalized_pair = self.config.trading_pair.replace("-", "_").lower()
            topic = f"{self.config.topic}/{normalized_pair}/ML_SIGNALS"
            self._ml_signal_listener = ExternalTopicFactory.create_async(
                topic=topic,
                callback=self._handle_ml_signal,
                use_bot_prefix=False,
            )
            self.logger().info("ML signal listener initialized successfully")
        except Exception as e:
            self.logger().error(f"Failed to initialize ML signal listener: {str(e)}")
            self._ml_signal_listener = None

    def _handle_ml_signal(self, signal: dict, topic: str):
        """Handle incoming ML signal"""
        # self.logger().info(f"Received ML signal: {signal}")
        short, neutral, long = signal["probabilities"]
        if short > self.config.short_threshold:
            self.processed_data["signal"] = -1
        elif long > self.config.long_threshold:
            self.processed_data["signal"] = 1
        else:
            self.processed_data["signal"] = 0
        self.processed_data["features"] = signal

    async def update_processed_data(self):
        pass

    def get_executor_config(self, trade_type: TradeType, price: Decimal, amount: Decimal):
        """
        Get the executor config based on the trade_type, price and amount. This method can be overridden by the
        subclasses if required.
        """
        return PositionExecutorConfig(
            timestamp=self.market_data_provider.time(),
            connector_name=self.config.connector_name,
            trading_pair=self.config.trading_pair,
            side=trade_type,
            entry_price=price,
            amount=amount,
            triple_barrier_config=self.config.triple_barrier_config.new_instance_with_adjusted_volatility(
                volatility_factor=self.processed_data["features"].get("target_pct", 0.01)),
            leverage=self.config.leverage,
        )

    def to_format_status(self) -> List[str]:
        lines = []
        features = self.processed_data.get("features", {})
        lines.append(f"Signal: {self.processed_data.get('signal', 'N/A')}")
        lines.append(f"Timestamp: {features.get('timestamp', 'N/A')}")
        lines.append(f"Probabilities: {features.get('probabilities', 'N/A')}")
        lines.append(f"Target Pct: {features.get('target_pct', 'N/A')}")
        return lines

```

Shown in full with attribution under the source's licence. Licence: Apache-2.0

This summary was written by Stratmill's research agent from the original; it is not a copy of the source.

Thresholding Machine-Learning Probabilities for Directional Trading

Summary

Key ideas

Tags