Data Module¶

Loaders¶

signalflow.data.source.binance.BinanceSpotLoader `dataclass` ¶

BinanceSpotLoader(store: DuckDbSpotStore = (lambda: DuckDbSpotStore(db_path=(Path('raw_data.duckdb'))))(), timeframe: str = '1m')

Bases: RawDataLoader

Downloads and stores Binance spot OHLCV data for fixed timeframe.

Combines BinanceClient (source) and DuckDbSpotStore (storage) to provide complete data pipeline with gap filling and incremental updates.

Attributes:

Name	Type	Description
`store`	`DuckDbSpotStore`	Storage backend. Default: raw_data.duckdb.
`timeframe`	`str`	Fixed timeframe for all data. Default: "1m".

download `async` ¶

download(pairs: list[str], days: Optional[int] = None, start: Optional[datetime] = None, end: Optional[datetime] = None, fill_gaps: bool = True) -> None

Download historical data with intelligent range detection.

Automatically determines what to download

If no existing data: download full range
If data exists: download before/after existing range
If fill_gaps=True: detect and fill gaps in existing range

Parameters:

Name	Type	Description	Default
`pairs`	`list[str]`	Trading pairs to download.	required
`days`	`int \| None`	Number of days back from end. Default: 7.	`None`
`start`	`datetime \| None`	Range start (overrides days).	`None`
`end`	`datetime \| None`	Range end. Default: now.	`None`
`fill_gaps`	`bool`	Detect and fill gaps. Default: True.	`True`

Note

Runs async download for all pairs concurrently. Logs progress for large downloads. Errors logged but don't stop other pairs.

Source code in src/signalflow/data/source/binance.py

async def download(
    self,
    pairs: list[str],
    days: Optional[int] = None,
    start: Optional[datetime] = None,
    end: Optional[datetime] = None,
    fill_gaps: bool = True,
) -> None:
    """Download historical data with intelligent range detection.

    Automatically determines what to download:
        - If no existing data: download full range
        - If data exists: download before/after existing range
        - If fill_gaps=True: detect and fill gaps in existing range

    Args:
        pairs (list[str]): Trading pairs to download.
        days (int | None): Number of days back from end. Default: 7.
        start (datetime | None): Range start (overrides days).
        end (datetime | None): Range end. Default: now.
        fill_gaps (bool): Detect and fill gaps. Default: True.

    Note:
        Runs async download for all pairs concurrently.
        Logs progress for large downloads.
        Errors logged but don't stop other pairs.
    """

    now = datetime.now(timezone.utc).replace(tzinfo=None)
    if end is None:
        end = now
    else:
        end = _ensure_utc_naive(end)

    if start is None:
        start = end - timedelta(days=days if days else 7)
    else:
        start = _ensure_utc_naive(start)

    tf_minutes = {
        "1m": 1,
        "3m": 3,
        "5m": 5,
        "15m": 15,
        "30m": 30,
        "1h": 60,
        "2h": 120,
        "4h": 240,
        "6h": 360,
        "8h": 480,
        "12h": 720,
        "1d": 1440,
    }.get(self.timeframe, 1)

    async def download_pair(client: BinanceClient, pair: str) -> None:
        logger.info(f"Processing {pair} from {start} to {end}")

        db_min, db_max = self.store.get_time_bounds(pair)
        ranges_to_download: list[tuple[datetime, datetime]] = []

        if db_min is None:
            ranges_to_download.append((start, end))
        else:
            if start < db_min:
                ranges_to_download.append((start, db_min - timedelta(minutes=tf_minutes)))
            if end > db_max:
                ranges_to_download.append((db_max + timedelta(minutes=tf_minutes), end))

            if fill_gaps:
                overlap_start = max(start, db_min)
                overlap_end = min(end, db_max)
                if overlap_start < overlap_end:
                    gaps = self.store.find_gaps(pair, overlap_start, overlap_end, tf_minutes)
                    ranges_to_download.extend(gaps)

        for range_start, range_end in ranges_to_download:
            if range_start >= range_end:
                continue

            logger.info(f"{pair}: downloading {range_start} -> {range_end}")

            try:
                klines = await client.get_klines_range(
                    pair=pair,
                    timeframe=self.timeframe,
                    start_time=range_start,
                    end_time=range_end,
                )
                self.store.insert_klines(pair, klines)
            except Exception as e:  
                logger.error(f"Error downloading {pair}: {e}")

    async with BinanceClient() as client:
        await asyncio.gather(*[download_pair(client, pair) for pair in pairs])

    self.store.close()

sync `async` ¶

sync(pairs: list[str], update_interval_sec: int = 60) -> None

Real-time sync - continuously update with latest data.

Runs indefinitely, fetching latest candles at specified interval. Useful for live trading or monitoring.

Parameters:

Name	Type	Description	Default
`pairs`	`list[str]`	Trading pairs to sync.	required
`update_interval_sec`	`int`	Update interval in seconds. Default: 60.	`60`

Note

Runs forever - use Ctrl+C to stop or run in background task. Fetches last 5 candles per update (ensures no gaps). Errors logged but sync continues.

Source code in src/signalflow/data/source/binance.py

async def sync(
    self,
    pairs: list[str],
    update_interval_sec: int = 60,
) -> None:
    """Real-time sync - continuously update with latest data.

    Runs indefinitely, fetching latest candles at specified interval.
    Useful for live trading or monitoring.

    Args:
        pairs (list[str]): Trading pairs to sync.
        update_interval_sec (int): Update interval in seconds. Default: 60.

    Note:
        Runs forever - use Ctrl+C to stop or run in background task.
        Fetches last 5 candles per update (ensures no gaps).
        Errors logged but sync continues.
    """

    logger.info(f"Starting real-time sync for {pairs}")
    logger.info(f"Update interval: {update_interval_sec}s (timeframe={self.timeframe})")

    async def fetch_and_store(client: BinanceClient, pair: str) -> None:
        try:
            klines = await client.get_klines(pair=pair, timeframe=self.timeframe, limit=5)
            self.store.insert_klines(pair, klines)
        except Exception as e:
            logger.error(f"Error syncing {pair}: {e}")

    async with BinanceClient() as client:
        while True:
            await asyncio.gather(*[fetch_and_store(client, pair) for pair in pairs])
            logger.debug(f"Synced {len(pairs)} pairs")
            await asyncio.sleep(update_interval_sec)

Storage¶

signalflow.data.raw_store.duckdb_stores.DuckDbSpotStore `dataclass` ¶

DuckDbSpotStore(db_path: Path, timeframe: str = '1m')

Bases: RawDataStore

DuckDB storage backend for OHLCV spot data.

Provides efficient storage and retrieval of candlestick (OHLCV) data using DuckDB as the backend. Designed for fixed-timeframe storage (timeframe not stored per-row, configured at database level).

Key features

Automatic schema migration from legacy formats
Efficient batch inserts with upsert (INSERT OR REPLACE)
Gap detection for data continuity checks
Multi-pair batch loading
Indexed queries for fast retrieval

Schema

pair (VARCHAR): Trading pair
timestamp (TIMESTAMP): Bar open time (timezone-naive)
open, high, low, close (DOUBLE): OHLC prices
volume (DOUBLE): Trading volume
trades (INTEGER): Number of trades

Attributes:

Name	Type	Description
`db_path`	`Path`	Path to DuckDB file.
`timeframe`	`str`	Fixed timeframe for all data (e.g., "1m", "5m"). Default: "1m".
`_con`	`DuckDBPyConnection`	Database connection (initialized in post_init).

Example

from signalflow.data.raw_store import DuckDbSpotStore
from pathlib import Path
from datetime import datetime

# Create store
store = DuckDbSpotStore(
    db_path=Path("data/binance_spot.duckdb"),
    timeframe="1m"
)

try:
    # Insert data
    klines = [
        {
            "timestamp": datetime(2024, 1, 1, 10, 0),
            "open": 45000.0,
            "high": 45100.0,
            "low": 44900.0,
            "close": 45050.0,
            "volume": 100.5,
            "trades": 150
        }
    ]
    store.insert_klines("BTCUSDT", klines)

    # Load data
    df = store.load("BTCUSDT", hours=24)

    # Check data bounds
    min_ts, max_ts = store.get_time_bounds("BTCUSDT")
    print(f"Data range: {min_ts} to {max_ts}")

    # Get statistics
    stats = store.get_stats()
    print(stats)

finally:
    store.close()

Note

Timeframe is fixed per database, not per row. Automatically migrates from legacy schema (open_time, timeframe columns). Always call close() to cleanup database connection.

__post_init__ ¶

__post_init__() -> None

Initialize database connection and ensure schema.

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def __post_init__(self) -> None:
    """Initialize database connection and ensure schema."""
    self._con = duckdb.connect(str(self.db_path))
    self._ensure_tables()

close ¶

close() -> None

Close database connection and cleanup resources.

Always call in finally block or use context manager to ensure cleanup.

Example

store = DuckDbSpotStore(Path("data/binance.duckdb"))
try:
    df = store.load("BTCUSDT", hours=24)
finally:
    store.close()

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def close(self) -> None:
    """Close database connection and cleanup resources.

    Always call in finally block or use context manager to ensure cleanup.

    Example:
        ```python
        store = DuckDbSpotStore(Path("data/binance.duckdb"))
        try:
            df = store.load("BTCUSDT", hours=24)
        finally:
            store.close()
        ```
    """
    self._con.close()

find_gaps ¶

find_gaps(pair: str, start: datetime, end: datetime, tf_minutes: int) -> list[tuple[datetime, datetime]]

Find gaps in data coverage for a pair.

Detects missing bars in expected continuous sequence based on timeframe. Useful for data quality checks and incremental backfilling.

Parameters:

Name	Type	Description	Default
`pair`	`str`	Trading pair (e.g., "BTCUSDT").	required
`start`	`datetime`	Start of expected range.	required
`end`	`datetime`	End of expected range.	required
`tf_minutes`	`int`	Timeframe in minutes (e.g., 1 for 1m, 5 for 5m).	required

Returns:

Type	Description
`list[tuple[datetime, datetime]]`	list[tuple[datetime, datetime]]: List of (gap_start, gap_end) tuples. Empty list if no gaps found.

Example

from datetime import datetime

# Check for gaps in January 2024
gaps = store.find_gaps(
    pair="BTCUSDT",
    start=datetime(2024, 1, 1),
    end=datetime(2024, 1, 31),
    tf_minutes=1
)

if gaps:
    print(f"Found {len(gaps)} gaps:")
    for gap_start, gap_end in gaps:
        duration = gap_end - gap_start
        print(f"  {gap_start} to {gap_end} ({duration})")

        # Backfill gaps
        backfill_data(pair="BTCUSDT", start=gap_start, end=gap_end)
else:
    print("No gaps found - data is continuous")

# Data quality report
gaps = store.find_gaps("BTCUSDT", start, end, tf_minutes=1)
total_expected = int((end - start).total_seconds() / 60)
total_missing = sum((g[1] - g[0]).total_seconds() / 60 for g in gaps)
coverage = (1 - total_missing / total_expected) * 100
print(f"Data coverage: {coverage:.2f}%")

Note

Returns full range [(start, end)] if no data exists. Computationally expensive for large date ranges - use sparingly.

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def find_gaps(
    self,
    pair: str,
    start: datetime,
    end: datetime,
    tf_minutes: int,
) -> list[tuple[datetime, datetime]]:
    """Find gaps in data coverage for a pair.

    Detects missing bars in expected continuous sequence based on timeframe.
    Useful for data quality checks and incremental backfilling.

    Args:
        pair (str): Trading pair (e.g., "BTCUSDT").
        start (datetime): Start of expected range.
        end (datetime): End of expected range.
        tf_minutes (int): Timeframe in minutes (e.g., 1 for 1m, 5 for 5m).

    Returns:
        list[tuple[datetime, datetime]]: List of (gap_start, gap_end) tuples.
            Empty list if no gaps found.

    Example:
        ```python
        from datetime import datetime

        # Check for gaps in January 2024
        gaps = store.find_gaps(
            pair="BTCUSDT",
            start=datetime(2024, 1, 1),
            end=datetime(2024, 1, 31),
            tf_minutes=1
        )

        if gaps:
            print(f"Found {len(gaps)} gaps:")
            for gap_start, gap_end in gaps:
                duration = gap_end - gap_start
                print(f"  {gap_start} to {gap_end} ({duration})")

                # Backfill gaps
                backfill_data(pair="BTCUSDT", start=gap_start, end=gap_end)
        else:
            print("No gaps found - data is continuous")

        # Data quality report
        gaps = store.find_gaps("BTCUSDT", start, end, tf_minutes=1)
        total_expected = int((end - start).total_seconds() / 60)
        total_missing = sum((g[1] - g[0]).total_seconds() / 60 for g in gaps)
        coverage = (1 - total_missing / total_expected) * 100
        print(f"Data coverage: {coverage:.2f}%")
        ```

    Note:
        Returns full range [(start, end)] if no data exists.
        Computationally expensive for large date ranges - use sparingly.
    """
    existing = self._con.execute("""
        SELECT timestamp
        FROM ohlcv
        WHERE pair = ? AND timestamp BETWEEN ? AND ?
        ORDER BY timestamp
    """, [pair, start, end]).fetchall()

    if not existing:
        return [(start, end)]

    existing_times = {row[0] for row in existing}
    gaps: list[tuple[datetime, datetime]] = []

    gap_start: Optional[datetime] = None
    current = start

    while current <= end:
        if current not in existing_times:
            if gap_start is None:
                gap_start = current
        else:
            if gap_start is not None:
                gaps.append((gap_start, current - timedelta(minutes=tf_minutes)))
                gap_start = None
        current += timedelta(minutes=tf_minutes)

    if gap_start is not None:
        gaps.append((gap_start, end))

    return gaps

get_stats ¶

get_stats() -> pl.DataFrame

Get database statistics per pair.

Returns summary statistics for all pairs in database.

Returns:

Type	Description
`DataFrame`	pl.DataFrame: Statistics with columns: - pair (str): Trading pair - rows (int): Number of bars - first_candle (datetime): Earliest timestamp - last_candle (datetime): Latest timestamp - total_volume (float): Sum of volume

Example

# Get overview
stats = store.get_stats()
print(stats)

# Check coverage
for row in stats.iter_rows(named=True):
    pair = row["pair"]
    days = (row["last_candle"] - row["first_candle"]).days
    print(f"{pair}: {row['rows']:,} bars over {days} days")

# Identify incomplete data
min_rows = stats["rows"].min()
incomplete = stats.filter(pl.col("rows") < min_rows * 0.9)
print(f"Pairs with <90% coverage: {incomplete['pair'].to_list()}")

Note

Timeframe not included in output (stored in meta table). Sorted alphabetically by pair.

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def get_stats(self) -> pl.DataFrame:
    """Get database statistics per pair.

    Returns summary statistics for all pairs in database.

    Returns:
        pl.DataFrame: Statistics with columns:
            - pair (str): Trading pair
            - rows (int): Number of bars
            - first_candle (datetime): Earliest timestamp
            - last_candle (datetime): Latest timestamp
            - total_volume (float): Sum of volume

    Example:
        ```python
        # Get overview
        stats = store.get_stats()
        print(stats)

        # Check coverage
        for row in stats.iter_rows(named=True):
            pair = row["pair"]
            days = (row["last_candle"] - row["first_candle"]).days
            print(f"{pair}: {row['rows']:,} bars over {days} days")

        # Identify incomplete data
        min_rows = stats["rows"].min()
        incomplete = stats.filter(pl.col("rows") < min_rows * 0.9)
        print(f"Pairs with <90% coverage: {incomplete['pair'].to_list()}")
        ```

    Note:
        Timeframe not included in output (stored in meta table).
        Sorted alphabetically by pair.
    """
    return self._con.execute("""
        SELECT
            pair,
            COUNT(*) as rows,
            MIN(timestamp) as first_candle,
            MAX(timestamp) as last_candle,
            ROUND(SUM(volume), 2) as total_volume
        FROM ohlcv
        GROUP BY pair
        ORDER BY pair
    """).pl()

get_time_bounds ¶

get_time_bounds(pair: str) -> tuple[Optional[datetime], Optional[datetime]]

Get earliest and latest timestamps for a pair.

Useful for

Checking data availability
Planning data updates
Validating date ranges

Parameters:

Name	Type	Description	Default
`pair`	`str`	Trading pair (e.g., "BTCUSDT").	required

Returns:

Type	Description
`tuple[Optional[datetime], Optional[datetime]]`	tuple[datetime \| None, datetime \| None]: (min_timestamp, max_timestamp). Both None if no data exists for pair.

Example

# Check data availability
min_ts, max_ts = store.get_time_bounds("BTCUSDT")

if min_ts and max_ts:
    print(f"Data available: {min_ts} to {max_ts}")
    days = (max_ts - min_ts).days
    print(f"Total days: {days}")
else:
    print("No data available")

# Plan incremental update
_, max_ts = store.get_time_bounds("BTCUSDT")
if max_ts:
    # Fetch data from max_ts to now
    fetch_data(start=max_ts, end=datetime.now())

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def get_time_bounds(self, pair: str) -> tuple[Optional[datetime], Optional[datetime]]:
    """Get earliest and latest timestamps for a pair.

    Useful for:
        - Checking data availability
        - Planning data updates
        - Validating date ranges

    Args:
        pair (str): Trading pair (e.g., "BTCUSDT").

    Returns:
        tuple[datetime | None, datetime | None]: (min_timestamp, max_timestamp).
            Both None if no data exists for pair.

    Example:
        ```python
        # Check data availability
        min_ts, max_ts = store.get_time_bounds("BTCUSDT")

        if min_ts and max_ts:
            print(f"Data available: {min_ts} to {max_ts}")
            days = (max_ts - min_ts).days
            print(f"Total days: {days}")
        else:
            print("No data available")

        # Plan incremental update
        _, max_ts = store.get_time_bounds("BTCUSDT")
        if max_ts:
            # Fetch data from max_ts to now
            fetch_data(start=max_ts, end=datetime.now())
        ```
    """
    result = self._con.execute("""
        SELECT MIN(timestamp), MAX(timestamp)
        FROM ohlcv
        WHERE pair = ?
    """, [pair]).fetchone()
    return (result[0], result[1]) if result and result[0] else (None, None)

insert_klines ¶

insert_klines(pair: str, klines: list[dict]) -> None

Upsert klines (INSERT OR REPLACE).

Efficient batch insertion with automatic upsert on (pair, timestamp) conflict. Uses Arrow-based bulk insert for >10 rows for better performance.

Timestamp normalization

Removes timezone info
Rounds to minute (removes seconds/microseconds)
If second != 0, rounds up to next minute

Parameters:

Name	Type	Description	Default
`pair`	`str`	Trading pair (e.g., "BTCUSDT").	required
`klines`	`list[dict]`	List of kline dictionaries. Each must contain: - timestamp (datetime): Bar open time - open (float): Open price - high (float): High price - low (float): Low price - close (float): Close price - volume (float): Trading volume - trades (int, optional): Number of trades	required

Example

from datetime import datetime

# Insert single kline
store.insert_klines("BTCUSDT", [
    {
        "timestamp": datetime(2024, 1, 1, 10, 0),
        "open": 45000.0,
        "high": 45100.0,
        "low": 44900.0,
        "close": 45050.0,
        "volume": 100.5,
        "trades": 150
    }
])

# Batch insert (efficient for >10 rows)
klines = [
    {
        "timestamp": datetime(2024, 1, 1, 10, i),
        "open": 45000.0 + i,
        "high": 45100.0 + i,
        "low": 44900.0 + i,
        "close": 45050.0 + i,
        "volume": 100.0,
        "trades": 150
    }
    for i in range(100)
]
store.insert_klines("BTCUSDT", klines)

# Upsert - updates existing rows
store.insert_klines("BTCUSDT", [
    {
        "timestamp": datetime(2024, 1, 1, 10, 0),
        "open": 45010.0,  # Updated price
        "high": 45110.0,
        "low": 44910.0,
        "close": 45060.0,
        "volume": 101.0,
        "trades": 152
    }
])

Note

Empty klines list is silently ignored. Uses executemany for ≤10 rows, Arrow bulk insert for >10 rows. Automatically logs insert count at debug level.

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def insert_klines(self, pair: str, klines: list[dict]) -> None:
    """Upsert klines (INSERT OR REPLACE).

    Efficient batch insertion with automatic upsert on (pair, timestamp) conflict.
    Uses Arrow-based bulk insert for >10 rows for better performance.

    Timestamp normalization:
        - Removes timezone info
        - Rounds to minute (removes seconds/microseconds)
        - If second != 0, rounds up to next minute

    Args:
        pair (str): Trading pair (e.g., "BTCUSDT").
        klines (list[dict]): List of kline dictionaries. Each must contain:
            - timestamp (datetime): Bar open time
            - open (float): Open price
            - high (float): High price
            - low (float): Low price
            - close (float): Close price
            - volume (float): Trading volume
            - trades (int, optional): Number of trades

    Example:
        ```python
        from datetime import datetime

        # Insert single kline
        store.insert_klines("BTCUSDT", [
            {
                "timestamp": datetime(2024, 1, 1, 10, 0),
                "open": 45000.0,
                "high": 45100.0,
                "low": 44900.0,
                "close": 45050.0,
                "volume": 100.5,
                "trades": 150
            }
        ])

        # Batch insert (efficient for >10 rows)
        klines = [
            {
                "timestamp": datetime(2024, 1, 1, 10, i),
                "open": 45000.0 + i,
                "high": 45100.0 + i,
                "low": 44900.0 + i,
                "close": 45050.0 + i,
                "volume": 100.0,
                "trades": 150
            }
            for i in range(100)
        ]
        store.insert_klines("BTCUSDT", klines)

        # Upsert - updates existing rows
        store.insert_klines("BTCUSDT", [
            {
                "timestamp": datetime(2024, 1, 1, 10, 0),
                "open": 45010.0,  # Updated price
                "high": 45110.0,
                "low": 44910.0,
                "close": 45060.0,
                "volume": 101.0,
                "trades": 152
            }
        ])
        ```

    Note:
        Empty klines list is silently ignored.
        Uses executemany for ≤10 rows, Arrow bulk insert for >10 rows.
        Automatically logs insert count at debug level.
    """
    if not klines:
        return

    if len(klines) <= 10:
        self._con.executemany(
            "INSERT OR REPLACE INTO ohlcv VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            [
                (
                    pair,
                    k["timestamp"],
                    k["open"],
                    k["high"],
                    k["low"],
                    k["close"],
                    k["volume"],
                    k.get("trades"),
                )
                for k in klines
            ],
        )
    else:
        df = pl.DataFrame(
            {
                "pair": [pair] * len(klines),
                "timestamp": [
                    k["timestamp"]
                    .replace(tzinfo=None)
                    .replace(second=0, microsecond=0)
                    + timedelta(minutes=1)
                    if k["timestamp"].second != 0 or k["timestamp"].microsecond != 0
                    else k["timestamp"].replace(tzinfo=None)
                    for k in klines
                ],
                "open": [k["open"] for k in klines],
                "high": [k["high"] for k in klines],
                "low": [k["low"] for k in klines],
                "close": [k["close"] for k in klines],
                "volume": [k["volume"] for k in klines],
                "trades": [k.get("trades") for k in klines],
            }
        )
        self._con.register("temp_klines", df.to_arrow())
        self._con.execute("INSERT OR REPLACE INTO ohlcv SELECT * FROM temp_klines")
        self._con.unregister("temp_klines")

    logger.debug(f"Inserted {len(klines):,} rows for {pair}")

load ¶

load(pair: str, hours: Optional[int] = None, start: Optional[datetime] = None, end: Optional[datetime] = None) -> pl.DataFrame

Load data for a single trading pair.

Output columns: pair, timestamp, open, high, low, close, volume, trades

Parameters:

Name	Type	Description	Default
`pair`	`str`	Trading pair (e.g., "BTCUSDT").	required
`hours`	`int \| None`	Load last N hours of data. Mutually exclusive with start/end.	`None`
`start`	`datetime \| None`	Start datetime (inclusive). Requires end parameter.	`None`
`end`	`datetime \| None`	End datetime (inclusive). Requires start parameter.	`None`

Returns:

Type	Description
`DataFrame`	pl.DataFrame: OHLCV data sorted by timestamp. Timezone-naive timestamps.

Example

# Load last 24 hours
df = store.load("BTCUSDT", hours=24)

# Load specific range
df = store.load(
    "BTCUSDT",
    start=datetime(2024, 1, 1),
    end=datetime(2024, 1, 31)
)

# Check loaded data
print(df.select(["timestamp", "close"]).head())

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def load(
    self,
    pair: str,
    hours: Optional[int] = None,
    start: Optional[datetime] = None,
    end: Optional[datetime] = None,
) -> pl.DataFrame:
    """Load data for a single trading pair.

    Output columns: pair, timestamp, open, high, low, close, volume, trades

    Args:
        pair (str): Trading pair (e.g., "BTCUSDT").
        hours (int | None): Load last N hours of data. Mutually exclusive with start/end.
        start (datetime | None): Start datetime (inclusive). Requires end parameter.
        end (datetime | None): End datetime (inclusive). Requires start parameter.

    Returns:
        pl.DataFrame: OHLCV data sorted by timestamp. Timezone-naive timestamps.

    Example:
        ```python
        # Load last 24 hours
        df = store.load("BTCUSDT", hours=24)

        # Load specific range
        df = store.load(
            "BTCUSDT",
            start=datetime(2024, 1, 1),
            end=datetime(2024, 1, 31)
        )

        # Check loaded data
        print(df.select(["timestamp", "close"]).head())
        ```
    """
    query = """
        SELECT
            ? AS pair,
            timestamp, open, high, low, close, volume, trades
        FROM ohlcv
        WHERE pair = ?
    """
    params: list[object] = [pair, pair]

    if hours is not None:
        query += f" AND timestamp > NOW() - INTERVAL '{int(hours)}' HOUR"
    elif start and end:
        query += " AND timestamp BETWEEN ? AND ?"
        params.extend([start, end])
    elif start:
        query += " AND timestamp >= ?"
        params.append(start)
    elif end:
        query += " AND timestamp <= ?"
        params.append(end)

    query += " ORDER BY timestamp"
    df = self._con.execute(query, params).pl()

    if 'timestamp' in df.columns:
        df = df.with_columns(
            pl.col('timestamp').dt.replace_time_zone(None)
        )

    return df

load_many ¶

load_many(pairs: Iterable[str], hours: Optional[int] = None, start: Optional[datetime] = None, end: Optional[datetime] = None) -> pl.DataFrame

Batch load for multiple pairs.

Output columns: pair, timestamp, open, high, low, close, volume, trades

More efficient than multiple load() calls due to single query.

Parameters:

Name	Type	Description	Default
`pairs`	`Iterable[str]`	Trading pairs to load.	required
`hours`	`int \| None`	Load last N hours of data.	`None`
`start`	`datetime \| None`	Start datetime (inclusive).	`None`
`end`	`datetime \| None`	End datetime (inclusive).	`None`

Returns:

Type	Description
`DataFrame`	pl.DataFrame: Combined OHLCV data sorted by (pair, timestamp). Empty DataFrame with correct schema if no pairs provided.

Example

# Load multiple pairs
df = store.load_many(
    pairs=["BTCUSDT", "ETHUSDT", "BNBUSDT"],
    start=datetime(2024, 1, 1),
    end=datetime(2024, 1, 31)
)

# Analyze by pair
for pair in df["pair"].unique():
    pair_df = df.filter(pl.col("pair") == pair)
    print(f"{pair}: {len(pair_df)} bars")

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def load_many(
    self,
    pairs: Iterable[str],
    hours: Optional[int] = None,
    start: Optional[datetime] = None,
    end: Optional[datetime] = None,
) -> pl.DataFrame:
    """Batch load for multiple pairs.

    Output columns: pair, timestamp, open, high, low, close, volume, trades

    More efficient than multiple load() calls due to single query.

    Args:
        pairs (Iterable[str]): Trading pairs to load.
        hours (int | None): Load last N hours of data.
        start (datetime | None): Start datetime (inclusive).
        end (datetime | None): End datetime (inclusive).

    Returns:
        pl.DataFrame: Combined OHLCV data sorted by (pair, timestamp).
            Empty DataFrame with correct schema if no pairs provided.

    Example:
        ```python
        # Load multiple pairs
        df = store.load_many(
            pairs=["BTCUSDT", "ETHUSDT", "BNBUSDT"],
            start=datetime(2024, 1, 1),
            end=datetime(2024, 1, 31)
        )

        # Analyze by pair
        for pair in df["pair"].unique():
            pair_df = df.filter(pl.col("pair") == pair)
            print(f"{pair}: {len(pair_df)} bars")
        ```
    """
    pairs = list(pairs)
    if not pairs:
        return pl.DataFrame(
            schema={
                "pair": pl.Utf8,
                "timestamp": pl.Datetime,
                "open": pl.Float64,
                "high": pl.Float64,
                "low": pl.Float64,
                "close": pl.Float64,
                "volume": pl.Float64,
                "trades": pl.Int64,
            }
        )

    placeholders = ",".join(["?"] * len(pairs))
    query = f"""
        SELECT
            pair,
            timestamp, open, high, low, close, volume, trades
        FROM ohlcv
        WHERE pair IN ({placeholders})
    """
    params: list[object] = [*pairs]

    if hours is not None:
        query += f" AND timestamp > NOW() - INTERVAL '{int(hours)}' HOUR"
    elif start and end:
        query += " AND timestamp BETWEEN ? AND ?"
        params.extend([start, end])
    elif start:
        query += " AND timestamp >= ?"
        params.append(start)
    elif end:
        query += " AND timestamp <= ?"
        params.append(end)

    query += " ORDER BY pair, timestamp"

    df = self._con.execute(query, params).pl()

    if 'timestamp' in df.columns:
        df = df.with_columns(
            pl.col('timestamp').dt.replace_time_zone(None)
        )

    return df

load_many_pandas ¶

load_many_pandas(pairs: list[str], start: datetime | None = None, end: datetime | None = None) -> pd.DataFrame

Load data for multiple pairs as Pandas DataFrame.

Convenience wrapper around load_many() for Pandas compatibility.

Parameters:

Name	Type	Description	Default
`pairs`	`list[str]`	List of trading pairs.	required
`start`	`datetime \| None`	Start datetime (inclusive).	`None`
`end`	`datetime \| None`	End datetime (inclusive).	`None`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: Combined OHLCV data as Pandas DataFrame.

Example

df = store.load_many_pandas(
    pairs=["BTCUSDT", "ETHUSDT"],
    start=datetime(2024, 1, 1),
    end=datetime(2024, 1, 31)
)

# Use with pandas
df["returns"] = df.groupby("pair")["close"].pct_change()

Source code in src/signalflow/data/raw_store/duckdb_stores.py

def load_many_pandas(
    self,
    pairs: list[str],
    start: datetime | None = None,
    end: datetime | None = None,
) -> pd.DataFrame:
    """Load data for multiple pairs as Pandas DataFrame.

    Convenience wrapper around load_many() for Pandas compatibility.

    Args:
        pairs (list[str]): List of trading pairs.
        start (datetime | None): Start datetime (inclusive).
        end (datetime | None): End datetime (inclusive).

    Returns:
        pd.DataFrame: Combined OHLCV data as Pandas DataFrame.

    Example:
        ```python
        df = store.load_many_pandas(
            pairs=["BTCUSDT", "ETHUSDT"],
            start=datetime(2024, 1, 1),
            end=datetime(2024, 1, 31)
        )

        # Use with pandas
        df["returns"] = df.groupby("pair")["close"].pct_change()
        ```
    """
    df_pl = self.load_many(pairs=pairs, start=start, end=end)
    return df_pl.to_pandas()

Factory¶

signalflow.data.raw_data_factory.RawDataFactory ¶

Factory for creating RawData instances from various sources.

Provides static methods to construct RawData objects from different storage backends (DuckDB, Parquet, etc.) with proper validation and schema normalization.

Key features

Automatic schema validation
Duplicate detection
Timezone normalization
Column cleanup (remove unnecessary columns)
Proper sorting by (pair, timestamp)

Example

from signalflow.data import RawDataFactory
from pathlib import Path
from datetime import datetime

# Load spot data from DuckDB
raw_data = RawDataFactory.from_duckdb_spot_store(
    spot_store_path=Path("data/binance_spot.duckdb"),
    pairs=["BTCUSDT", "ETHUSDT"],
    start=datetime(2024, 1, 1),
    end=datetime(2024, 12, 31),
    data_types=["spot"]
)

# Access loaded data
spot_df = raw_data["spot"]
print(f"Loaded {len(spot_df)} bars")
print(f"Pairs: {raw_data.pairs}")
print(f"Date range: {raw_data.datetime_start} to {raw_data.datetime_end}")

# Use in detector
from signalflow.detector import SmaCrossSignalDetector

detector = SmaCrossSignalDetector(fast_window=10, slow_window=20)
signals = detector.detect(raw_data)

from_duckdb_spot_store `staticmethod` ¶

from_duckdb_spot_store(spot_store_path: Path, pairs: list[str], start: datetime, end: datetime, data_types: list[str] | None = None) -> RawData

Create RawData from DuckDB spot store.

Loads spot trading data from DuckDB storage with validation, deduplication checks, and schema normalization.

Processing steps

Load data from DuckDB for specified pairs and date range
Validate required columns (pair, timestamp)
Remove unnecessary columns (timeframe)
Normalize timestamps (microseconds, timezone-naive)
Check for duplicates (pair, timestamp)
Sort by (pair, timestamp)
Package into RawData container

Parameters:

Name	Type	Description	Default
`spot_store_path`	`Path`	Path to DuckDB file.	required
`pairs`	`list[str]`	List of trading pairs to load (e.g., ["BTCUSDT", "ETHUSDT"]).	required
`start`	`datetime`	Start datetime (inclusive).	required
`end`	`datetime`	End datetime (inclusive).	required
`data_types`	`list[str] \| None`	Data types to load. Default: None. Currently supports: ["spot"].	`None`

Returns:

Name	Type	Description
`RawData`	`RawData`	Immutable container with loaded and validated data.

Raises:

Type	Description
`ValueError`	If required columns missing (pair, timestamp).
`ValueError`	If duplicate (pair, timestamp) combinations detected.

Example

from pathlib import Path
from datetime import datetime
from signalflow.data import RawDataFactory

# Load single pair
raw_data = RawDataFactory.from_duckdb_spot_store(
    spot_store_path=Path("data/binance.duckdb"),
    pairs=["BTCUSDT"],
    start=datetime(2024, 1, 1),
    end=datetime(2024, 1, 31),
    data_types=["spot"]
)

# Load multiple pairs
raw_data = RawDataFactory.from_duckdb_spot_store(
    spot_store_path=Path("data/binance.duckdb"),
    pairs=["BTCUSDT", "ETHUSDT", "BNBUSDT"],
    start=datetime(2024, 1, 1),
    end=datetime(2024, 12, 31),
    data_types=["spot"]
)

# Check loaded data
spot_df = raw_data["spot"]
print(f"Shape: {spot_df.shape}")
print(f"Columns: {spot_df.columns}")
print(f"Pairs: {spot_df['pair'].unique().to_list()}")

# Verify no duplicates
dup_check = (
    spot_df.group_by(["pair", "timestamp"])
    .len()
    .filter(pl.col("len") > 1)
)
assert dup_check.is_empty()

# Use in pipeline
from signalflow.core import RawDataView
view = RawDataView(raw=raw_data)
spot_pandas = view.to_pandas("spot")

Example

# Handle missing data gracefully
try:
    raw_data = RawDataFactory.from_duckdb_spot_store(
        spot_store_path=Path("data/binance.duckdb"),
        pairs=["BTCUSDT"],
        start=datetime(2024, 1, 1),
        end=datetime(2024, 1, 31),
        data_types=["spot"]
    )
except ValueError as e:
    if "missing columns" in str(e):
        print("Data schema invalid")
    elif "Duplicate" in str(e):
        print("Data contains duplicates")
    raise

# Validate date range
assert raw_data.datetime_start == datetime(2024, 1, 1)
assert raw_data.datetime_end == datetime(2024, 1, 31)

# Check data quality
spot_df = raw_data["spot"]

# Verify timestamps are sorted
assert spot_df["timestamp"].is_sorted()

# Verify timezone-naive
assert spot_df["timestamp"].dtype == pl.Datetime("us")

# Verify no nulls in key columns
assert spot_df["pair"].null_count() == 0
assert spot_df["timestamp"].null_count() == 0

Note

Store connection is automatically closed via finally block. Timestamps are normalized to timezone-naive microseconds. Duplicate detection shows first 10 examples if found. All data sorted by (pair, timestamp) for consistent ordering.

Source code in src/signalflow/data/raw_data_factory.py

@staticmethod
def from_duckdb_spot_store(
    spot_store_path: Path,
    pairs: list[str],
    start: datetime,
    end: datetime,
    data_types: list[str] | None = None,
) -> RawData:
    """Create RawData from DuckDB spot store.

    Loads spot trading data from DuckDB storage with validation,
    deduplication checks, and schema normalization.

    Processing steps:
        1. Load data from DuckDB for specified pairs and date range
        2. Validate required columns (pair, timestamp)
        3. Remove unnecessary columns (timeframe)
        4. Normalize timestamps (microseconds, timezone-naive)
        5. Check for duplicates (pair, timestamp)
        6. Sort by (pair, timestamp)
        7. Package into RawData container

    Args:
        spot_store_path (Path): Path to DuckDB file.
        pairs (list[str]): List of trading pairs to load (e.g., ["BTCUSDT", "ETHUSDT"]).
        start (datetime): Start datetime (inclusive).
        end (datetime): End datetime (inclusive).
        data_types (list[str] | None): Data types to load. Default: None.
            Currently supports: ["spot"].

    Returns:
        RawData: Immutable container with loaded and validated data.

    Raises:
        ValueError: If required columns missing (pair, timestamp).
        ValueError: If duplicate (pair, timestamp) combinations detected.

    Example:
        ```python
        from pathlib import Path
        from datetime import datetime
        from signalflow.data import RawDataFactory

        # Load single pair
        raw_data = RawDataFactory.from_duckdb_spot_store(
            spot_store_path=Path("data/binance.duckdb"),
            pairs=["BTCUSDT"],
            start=datetime(2024, 1, 1),
            end=datetime(2024, 1, 31),
            data_types=["spot"]
        )

        # Load multiple pairs
        raw_data = RawDataFactory.from_duckdb_spot_store(
            spot_store_path=Path("data/binance.duckdb"),
            pairs=["BTCUSDT", "ETHUSDT", "BNBUSDT"],
            start=datetime(2024, 1, 1),
            end=datetime(2024, 12, 31),
            data_types=["spot"]
        )

        # Check loaded data
        spot_df = raw_data["spot"]
        print(f"Shape: {spot_df.shape}")
        print(f"Columns: {spot_df.columns}")
        print(f"Pairs: {spot_df['pair'].unique().to_list()}")

        # Verify no duplicates
        dup_check = (
            spot_df.group_by(["pair", "timestamp"])
            .len()
            .filter(pl.col("len") > 1)
        )
        assert dup_check.is_empty()

        # Use in pipeline
        from signalflow.core import RawDataView
        view = RawDataView(raw=raw_data)
        spot_pandas = view.to_pandas("spot")
        ```

    Example:
        ```python
        # Handle missing data gracefully
        try:
            raw_data = RawDataFactory.from_duckdb_spot_store(
                spot_store_path=Path("data/binance.duckdb"),
                pairs=["BTCUSDT"],
                start=datetime(2024, 1, 1),
                end=datetime(2024, 1, 31),
                data_types=["spot"]
            )
        except ValueError as e:
            if "missing columns" in str(e):
                print("Data schema invalid")
            elif "Duplicate" in str(e):
                print("Data contains duplicates")
            raise

        # Validate date range
        assert raw_data.datetime_start == datetime(2024, 1, 1)
        assert raw_data.datetime_end == datetime(2024, 1, 31)

        # Check data quality
        spot_df = raw_data["spot"]

        # Verify timestamps are sorted
        assert spot_df["timestamp"].is_sorted()

        # Verify timezone-naive
        assert spot_df["timestamp"].dtype == pl.Datetime("us")

        # Verify no nulls in key columns
        assert spot_df["pair"].null_count() == 0
        assert spot_df["timestamp"].null_count() == 0
        ```

    Note:
        Store connection is automatically closed via finally block.
        Timestamps are normalized to timezone-naive microseconds.
        Duplicate detection shows first 10 examples if found.
        All data sorted by (pair, timestamp) for consistent ordering.
    """
    data: dict[str, pl.DataFrame] = {}
    store = DuckDbSpotStore(spot_store_path)
    try:
        if "spot" in data_types:
            spot = store.load_many(pairs=pairs, start=start, end=end)

            required = {"pair", "timestamp"}
            missing = required - set(spot.columns)
            if missing:
                raise ValueError(f"Spot df missing columns: {sorted(missing)}")

            if "timeframe" in spot.columns:
                spot = spot.drop("timeframe")

            spot = spot.with_columns(
                pl.col("timestamp").cast(pl.Datetime("us")).dt.replace_time_zone(None)
            )

            dup_count = (
                spot.group_by(["pair", "timestamp"]).len()
                .filter(pl.col("len") > 1)
            )
            if dup_count.height > 0:
                dups = (
                    spot.join(
                        dup_count.select(["pair", "timestamp"]),
                        on=["pair", "timestamp"],
                    )
                    .select(["pair", "timestamp"])
                    .head(10)
                )
                raise ValueError(
                    f"Duplicate (pair, timestamp) detected. Examples:\n{dups}"
                )

            spot = spot.sort(["pair", "timestamp"])
            data["spot"] = spot

        return RawData(
            datetime_start=start,
            datetime_end=end,
            pairs=pairs,
            data=data,
        )
    finally:
        store.close()

Data Module¶

Loaders¶

signalflow.data.source.binance.BinanceSpotLoader dataclass ¶

download async ¶

sync async ¶

Storage¶

signalflow.data.raw_store.duckdb_stores.DuckDbSpotStore dataclass ¶

__post_init__ ¶

close ¶

find_gaps ¶

get_stats ¶

get_time_bounds ¶

insert_klines ¶

load ¶

load_many ¶

load_many_pandas ¶

Factory¶

signalflow.data.raw_data_factory.RawDataFactory ¶

from_duckdb_spot_store staticmethod ¶

signalflow.data.source.binance.BinanceSpotLoader `dataclass` ¶

download `async` ¶

sync `async` ¶

signalflow.data.raw_store.duckdb_stores.DuckDbSpotStore `dataclass` ¶

from_duckdb_spot_store `staticmethod` ¶