data.storage.store_market_data

data.storage.store_market_data

DataStorage

DataStorage(database_path: str | Path | None = None)

Allows for the storage of OHLCV data to prevent re-polling API data that has already been received before.

Initializes DataStorage.

Parameters:

Name	Type	Description	Default
database_path	str | Path | None	Where to store the database. If not provided, falls back to the OS-standard user data directory %LOCALAPPDATA%\contango\contango on Windows).	None

Source code in src\contango\data\storage\store_market_data.py

def __init__(self, database_path: str | Path | None = None) -> None:
    """
    Initializes `DataStorage`.

    Args:
        database_path: Where to store the database. If not provided, falls
                       back to the OS-standard user data directory %LOCALAPPDATA%\\contango\\contango on Windows).
    """
    resolved_path = Path(database_path) if database_path is not None else _default_database_path()
    resolved_path.parent.mkdir(parents=True, exist_ok=True)

    self._connection = sqlite3.connect(resolved_path)
    self.database_path = resolved_path

    self._create_tables()

add_data_to_storage

add_data_to_storage(interval: str, data: list[MarketDataEvent]) -> None

Adds market data to storage. Existing candles with the same symbol/interval/timestamp are ignored.

Parameters:

Name	Type	Description	Default
interval	str	The interval to save the data as (i.e. 1m, 5m, etc).	required
data	list[MarketDataEvent]	The data to save into the database.	required

Source code in src\contango\data\storage\store_market_data.py

def add_data_to_storage(
    self,
    interval: str,
    data: list[MarketDataEvent],
) -> None:
    """
    Adds market data to storage. Existing candles with the same symbol/interval/timestamp are ignored.

    Args:
        interval: The interval to save the data as (i.e. `1m`, `5m`, etc).
        data: The data to save into the database.
    """
    self._connection.executemany(
        """
        INSERT OR IGNORE INTO market_data
        (
            symbol,
            interval,
            timestamp,
            open,
            high,
            low,
            close,
            volume
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """,
        [
            (
                event.symbol,
                interval,
                event.timestamp,
                event.open,
                event.high,
                event.low,
                event.close,
                event.volume,
            )
            for event in data
        ],
    )

    self._connection.commit()

close

close() -> None

Closes the connection to the database.

Source code in src\contango\data\storage\store_market_data.py

def close(self) -> None:
    """
    Closes the connection to the database.
    """
    self._connection.close()

get_data

get_data(symbol: str, interval: str, start_timestamp: int, end_timestamp: int) -> list[MarketDataEvent]

Retrieves market data in chronological order by timestamp.

Parameters:

Name	Type	Description	Default
symbol	str	The symbol to derive data from.	required
interval	str	The interval to retrieve data from for the symbol (i.e. 1m, 5m, etc).	required
start_timestamp	int	The start timestamp in unix ms.	required
end_timestamp	int	The end timestamp in unix ms.	required

Returns:

Type	Description
list[MarketDataEvent]	list[MarketDataEvent]: The data for the provided parameters, if any.

Source code in src\contango\data\storage\store_market_data.py

def get_data(
    self,
    symbol: str,
    interval: str,
    start_timestamp: int,
    end_timestamp: int,
) -> list[MarketDataEvent]:
    """
    Retrieves market data in chronological order by timestamp.

    Args:
        symbol: The symbol to derive data from.
        interval: The interval to retrieve data from for the symbol (i.e. `1m`, `5m`, etc).
        start_timestamp: The start timestamp in unix ms.
        end_timestamp: The end timestamp in unix ms.

    Returns:
        list[MarketDataEvent]: The data for the provided parameters, if any.
    """
    cursor = self._connection.execute(
        """
        SELECT
            timestamp,
            symbol,
            open,
            high,
            low,
            close,
            volume
        FROM market_data
        WHERE symbol = ?
          AND interval = ?
          AND timestamp BETWEEN ? AND ?
        ORDER BY timestamp ASC
        """,
        (
            symbol,
            interval,
            start_timestamp,
            end_timestamp,
        ),
    )

    return [
        MarketDataEvent(
            timestamp=row[0],
            symbol=row[1],
            open=row[2],
            high=row[3],
            low=row[4],
            close=row[5],
            volume=row[6],
        )
        for row in cursor.fetchall()
    ]

get_missing_timestamps

get_missing_timestamps(symbol: str, interval: str, expected_timestamps: Iterable[datetime]) -> list[int]

Returns the timestamps that are NOT currently stored for the given symbol/interval, out of a caller-supplied set of expected timestamps. Callers are responsible for generating the correct expected schedule.

Parameters:

Name	Type	Description	Default
symbol	str	The symbol to check.	required
interval	str	The interval to check (i.e. 1m, 5m, etc).	required
expected_timestamps	Iterable[datetime]	The datetimes candles are expected to exist for. Must be timezone-aware (or assumed UTC if naive) since stored timestamps are unix ms in UTC.	required

Returns:

Type	Description
list[int]	list[int]: Sorted list of missing timestamps (unix ms). Empty if fully cached.

Source code in src\contango\data\storage\store_market_data.py

def get_missing_timestamps(
    self,
    symbol: str,
    interval: str,
    expected_timestamps: Iterable[datetime],
) -> list[int]:
    """
    Returns the timestamps that are NOT currently stored for the given
    symbol/interval, out of a caller-supplied set of expected timestamps.
    Callers are responsible for generating the correct expected schedule.

    Args:
        symbol: The symbol to check.
        interval: The interval to check (i.e. `1m`, `5m`, etc).
        expected_timestamps: The datetimes candles are expected to exist
                             for. Must be timezone-aware (or assumed UTC if naive) since
                             stored timestamps are unix ms in UTC.

    Returns:
        list[int]: Sorted list of missing timestamps (unix ms). Empty if
                   fully cached.
    """
    expected_ms = {
        int(
            (ts if ts.tzinfo else ts.replace(tzinfo=timezone.utc))
            .timestamp() * 1000
        )
        for ts in expected_timestamps
    }

    if not expected_ms:
        return []

    cursor = self._connection.execute(
        """
        SELECT timestamp
        FROM market_data
        WHERE symbol = ?
        AND interval = ?
        AND timestamp BETWEEN ? AND ?
        """,
        (
            symbol,
            interval,
            min(expected_ms),
            max(expected_ms),
        ),
    )

    existing = {row[0] for row in cursor.fetchall()}

    return sorted(expected_ms - existing)