pulse-core 0.1.3

Pure computation library: strategy engines, indicators, and shared types for Pulse.

pulse-core

Shared models, strategy framework, and DB session for Pulse.

This package is the foundation layer consumed by both pulse-backend (FastAPI) and Airflow DAGs. It owns the SQLModel/SQLAlchemy models, the strategy engine abstraction, and a lightweight DB session factory.

Package structure

pulse_core/
├── db.py                   # Singleton engine + context-managed Session
├── models/
│   ├── engine.py           # Engine table (registry metadata)
│   ├── market.py           # Exchange, Instrument, Listing, DataSource
│   ├── mixins.py           # TimestampMixin, SoftDeleteMixin
│   ├── prices.py           # TimeFrame, PriceOHLCV
│   └── signals.py          # Strategy, Signal, Subscription, Delivery*
└── strategies/
    ├── base.py             # BaseStrategy ABC, SignalOutput, UniverseFilter
    ├── indicators.py       # Pure-numpy indicator library (SMA, ADX, BB, …)
    ├── registry.py         # StrategyRegistry (decorator-based)
    ├── sync.py             # sync_engine_registry() — Python ↔ DB reconciliation
    └── engines/
        ├── mean_reversion.py          # Dual MA mean-reversion
        └── regime_mean_reversion.py   # Regime-based probabilistic mean-reversion

Installation

# from the repo root
uv pip install -e pulse-core

Or as a dependency in another package's pyproject.toml:

dependencies = [
    "pulse-core @ file:///${PROJECT_ROOT}/pulse-core",
]

Usage

Database session

from pulse_core.db import get_session

# Reads PULSE_DB_URL from the environment
with get_session() as session:
    ...

Models

from pulse_core.models import Listing, PriceOHLCV, Signal, Engine

Writing a strategy engine

Subclass BaseStrategy, implement validate_parameters() and compute(), then register it with the @StrategyRegistry.register decorator:

from pulse_core.strategies import BaseStrategy, SignalOutput, StrategyRegistry

@StrategyRegistry.register("my_engine")
class MyEngine(BaseStrategy):
    display_name = "My Engine"
    description = "Short description of the strategy."

    def validate_parameters(self) -> None:
        # raise ValueError if self.parameters is invalid
        ...

    def compute(self, listing, candles, as_of) -> SignalOutput | None:
        # return a SignalOutput or None
        ...

The engine is auto-discovered at import time via pulse_core.strategies.engines.__init__.

Syncing the registry to the database

from pulse_core.db import get_session
from pulse_core.strategies.sync import sync_engine_registry

with get_session() as session:
    result = sync_engine_registry(session)
    print(f"upserted={result.upserted}, deactivated={result.deactivated}")

Requirements

• Python >= 3.12
• PostgreSQL (TimescaleDB) — connected via PULSE_DB_URL

Development

# lint
uv run --group dev ruff check .

# tests
uv run --group test pytest