Quant Pulse

Quant Pulse is a lightweight Python library for generating market signals on demand. It supports various targets (asset classes, markets, etc.) and signal operations through a modular architecture that allows for easy extension and customization.

Current Version: 0.1.0 (alpha)

Core Principles

• Scoped to financial domain, whether decentralized or centralized.

• Remain fully open source to promote transparency and wider adoption.

Design Goals

• Offer automatic trading systems raw or processed data to support decisions-making, embedded directly in-process.

• Prototype composite signals from raw ones using Jupyter Notebook or similar environments.

Non-Goals

• Not designed for high‑frequency trading (HFT) or other fast trading systems.

• Direct interaction with markets is explicitly out of scope.

• Not a server or service — no network interface is provided; embed it in your own process.

Further Notes

• The codebase is still stabilizing, and contributions via pull request are not open yet. Key interfaces are subject to refinement; feel free to experiment with modules, but keep in mind that breaking changes may occur.

Installation

pip install quant-pulse

Requires Python 3.10+. TA-Lib must be installed on the system before pip install (the Python package wraps the native library).

Modules

Path:

src/quant_pulse
├── handlers   # signal handlers
├── services   # core services
└── _core      # framework internals (abstractions, registries, config)

Handlers:

• RSISignal: RSI indicator signal handler
• MASignal: MA indicator signal handler (supports sma, ema, wma)
• ATRSignal: ATR indicator signal handler
• DMISignal: DMI indicator signal handler (combines ADX, DI+, DI-)
• PriceStructSignal: Price structure signal handler
• TrendDirectSignal: Trend direction signal handler
• MntmQualitySignal: Momentum Quality signal handler
• MACDSignal: MACD indicator signal handler
• SupertrendSignal: Supertrend signal handler (composes ATRSignal)

Services:

• CacheManager: factory-pattern based caching to create scoped caches
• ExchangeClient: abstract base type for exchange implementations
• BinanceClient: provides access to OHLCV data from Binance CEX
• HyperliquidClient: provides access to OHLCV data from Hyperliquid DEX
• OHLCVFetcher: abstracts OHLCV data fetching and provides caching

Usage

import yaml
from quant_pulse import Context

with open('config/dev.yaml') as f:
    cfg = yaml.safe_load(f)

with Context.from_dict(cfg) as ctx:
    sig = ctx.get_signal_handler('rsi_hyperliquid')
    out = sig.compute_signal(
        target='BTC',
        signal_op='compute_rsi',
        request={'period': 14, 'timeframe': '1h'}
    )

    if out.errors:
        print('errors:', out.errors)
    else:
        print(f"latest_rsi: {out.result['latest_rsi']}")
        print(f"regime: {out.result['regime']}")

out.result is a plain Python object (dict/list/etc.) — no serialization step, nothing to decode.

Config can also be built in code instead of YAML, via ConfigBuilder:

from quant_pulse import Context, ConfigBuilder

cfg = (
    ConfigBuilder()
    .service('hl_client', 'exchange/hyperliquid_client', 'HyperliquidClient',
              _api_url='https://api.hyperliquid.xyz')
    .service('hl_ohlcv', 'exchange/data_providers', 'OHLCVFetcher',
              using=['hl_client'], ttl_1h=3600)
    .signal('rsi_hyperliquid', 'rsi_signal', ohlcv_service='hl_ohlcv')
    .build()
)

ctx = Context.from_dict(cfg)

A runnable version of the first example is at samples/rsi_sample.py.

Configuration

A Context is built from a config dict with two sections: signals (profiles bound to a handler) and services (infrastructure a handler can depend on).

signals:
  - rsi_hyperliquid:
      _binding: rsi_signal
      ohlcv_service: ohlcv_hyperliquid

  - atr_hyperliquid:
      _binding: atr_signal
      ohlcv_service: ohlcv_hyperliquid

  - supertrend_hyperliquid:
      _binding: supertrend_signal
      ohlcv_service: ohlcv_hyperliquid
      atr_signal: atr_hyperliquid

  - macd_hyperliquid:
      _binding: macd_signal
      ma_signal: ma_hyperliquid
      fast_period: 12
      slow_period: 26
      signal_period: 9

services:
  - cache_manager:
      _service: cache
      _binding: CacheManager
      max_items: 10000
      max_bytes: 104857600

  - hyperliquid_client:
      _service: exchange/hyperliquid_client
      _binding: HyperliquidClient
      _api_url: https://api.hyperliquid.xyz
      timeout_sec: 10.0
      max_retries: 1
      retry_delay: 0

  - ohlcv_hyperliquid:
      _service: exchange/data_providers
      _binding: OHLCVFetcher
      _using:
        - hyperliquid_client
        - cache_manager
      ttl_1m: 60
      ttl_5m: 300
      ttl_1h: 3600
      ttl_1d: 86400
      ttl_default: closest # Auto-select nearest TTL

Handlers can invoke other signals as sub-components (e.g. SupertrendSignal calls its configured atr_signal profile), and declare reusable exchange/data integrations as service dependencies — wired via config, no code changes needed to add profiles or tune parameters.

Signal Sample

def _compute_signal_impl(self, target: str, signal_op: str, request: Optional[dict]) -> Outcome:
    """
    Compute RSI signal
    """
    try:
        period = int(request['period'])
        timeframe = request['timeframe']

        ohlcv_fetcher = self._ctx.get_service(self._ohlcv_service_name)
        if not ohlcv_fetcher or not hasattr(ohlcv_fetcher, 'retrieve_ohlcv'):
            return self._error(f"OHLCV service '{self._ohlcv_service_name}' not found or invalid")

        limit = period + 100
        candles = ohlcv_fetcher.retrieve_ohlcv(symbol=self._symbol, timeframe=timeframe, limit=limit)
        if not candles or len(candles) < period:
            return self._error(f"Insufficient candle data: need {period}, got {len(candles) if candles else 0}")

        close_prices = np.array([candle.close for candle in candles])
        rsi_values = talib.RSI(close_prices, timeperiod=period)
        rsi_list = rsi_values[~np.isnan(rsi_values)].round(2).tolist()

        latest_rsi = next((float(v) for v in reversed(rsi_values) if not np.isnan(v)), None)
        if latest_rsi is None:
            return self._error('Failed to compute RSI: all values are NaN')

        regime = 'overbought' if latest_rsi >= 70 else 'oversold' if latest_rsi <= 30 else 'neutral'

        return Outcome(
            result={'symbol': self._symbol, 'rsi': rsi_list, 'latest_rsi': round(latest_rsi, 2), 'regime': regime},
            computation=self.get_metadata()['name']
        )
    except Exception as e:
        self.logger.error(f"Error computing RSI: {e}", exc_info=True)
        return self._error(str(e))

License

MIT — see LICENSE.