diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index d29c567..76314d0 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -34,6 +34,8 @@ docker version - Make retries, timeouts, and rate limits explicit. - Add offline fixture tests. CI must not require network access or API keys. - Declare provider-specific libraries as optional dependencies. +- Keep local/database adapters read-only, validate mapped identifiers against + reflected metadata, and bind query values instead of accepting raw SQL. - Keep provider-specific request parameters inside the adapter rather than extending the normalized records. - Never make `pineforge-engine` depend on this package. diff --git a/README.md b/README.md index 372e13b..e7a008d 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,8 @@ the engine receives only normalized bars and ordered trades. live trades, macro vintages, and validation rules. - [Using providers](docs/providers.md) — registry, CCXT market discovery, historical bars, live trades, configuration, and errors. +- [Local files and databases](docs/local-data.md) — runtime schema discovery, + arbitrary CSV/SQLite DDL, SQLAlchemy databases, and column mappings. - [Backtesting](docs/backtesting.md) — CLI options, configuration files, runtime channels, report schema, and reproducibility. - [FastAPI server](docs/server.md) — concurrency, authentication, timeouts, @@ -50,6 +52,8 @@ into contiguous C ABI arrays and submitted in one call. `MacroDataProvider` — small structural protocols that community adapters implement. - `ProviderRegistry` — built-in and installed broker adapters selected by name. +- `BarColumnMapping` and `TabularSchema` — runtime discovery and safe OHLCV + mappings for user-owned files, tables, and views. - `PfBar`, `PfTradeTick`, and `EngineStreamSink` — dependency-free `ctypes` interoperability with PineForge strategy libraries. @@ -96,6 +100,40 @@ this bootstrap; a WebSocket transport can implement the same after an engine warmup. `start_sequence` is the last accepted sequence, so the adapter emits `start_sequence + 1` next. +## Local files and databases + +Built-in `csv`, `sqlite`, and `sqlalchemy` providers let users backtest their +own data without adopting a fixed PineForge DDL. They inspect file headers or +database reflection metadata at runtime, infer common OHLCV names, and accept +partial mappings for arbitrary names. Ambiguous schemas fail instead of being +guessed. + +```python +from pineforge_data import SqliteBarProvider + + +provider = SqliteBarProvider( + "warehouse.sqlite3", + table="price candles", + mapping={ + "timestamp": "epoch seconds", + "open": "first px", + "high": "top px", + "low": "bottom px", + "close": "last px", + "volume": "traded qty", + }, + timestamp_unit="seconds", +) +schema = await provider.inspect_schema() +``` + +SQL identifiers are validated against reflected metadata; filter values are +bound parameters. For complex transformations, expose a database view rather +than putting raw SQL in harness configuration. See the complete +[local-data guide](docs/local-data.md) for CSV, native read-only SQLite, +SQLAlchemy URLs, timestamp handling, symbol/timeframe semantics, and CLI JSON. + ## Direct backtest harness The public backtest input is raw PineScript v6. `pineforge-backtest` fetches diff --git a/docs/backtesting.md b/docs/backtesting.md index 3bf9b7c..863e2ea 100644 --- a/docs/backtesting.md +++ b/docs/backtesting.md @@ -95,6 +95,12 @@ Pine input overrides: Input and override values sent to the FastAPI service must be scalar strings, numbers, or booleans. +The same harness accepts `--provider csv`, `--provider sqlite`, or +`--provider sqlalchemy`. Their provider configuration maps arbitrary source +columns to normalized OHLCV, while raw Pine and normalized bars follow the +same local-container or FastAPI path. See +[Local files and databases](local-data.md) for complete configuration examples. + ## Runtime image policy The package default is an exact `pineforge-release` version and OCI digest. The diff --git a/docs/getting-started.md b/docs/getting-started.md index 50a2642..22331ed 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -24,6 +24,7 @@ Available extras: | Extra | Purpose | |---|---| | `ccxt` | CCXT async exchange adapter | +| `database` | SQLAlchemy 2.x database reflection and queries | | `server` | FastAPI and Uvicorn server dependencies | | `dev` | tests, type checking, formatting, and package builds | @@ -103,5 +104,6 @@ cache. - [Learn the normalized data model](data-model.md). - [Search and filter provider markets](providers.md). +- [Connect local CSV, SQLite, or SQLAlchemy data](local-data.md). - [Configure parameters, runtime channels, and reports](backtesting.md). - [Deploy the concurrent server](server.md). diff --git a/docs/index.md b/docs/index.md index e65fc43..68795a9 100644 --- a/docs/index.md +++ b/docs/index.md @@ -24,6 +24,7 @@ exchange / broker / macro API | Install the package and run a first backtest | [Getting started](getting-started.md) | | Understand instruments, contracts, bars, trades, and macro vintages | [Data model](data-model.md) | | Discover markets and use CCXT or another provider | [Using providers](providers.md) | +| Connect a CSV file, SQLite table, or SQLAlchemy database | [Local files and databases](local-data.md) | | Configure local and remote raw-Pine backtests | [Backtesting](backtesting.md) | | Deploy and operate the concurrent FastAPI service | [FastAPI server](server.md) | | Implement a new exchange or broker adapter | [Provider contract](provider-contract.md) | @@ -46,6 +47,8 @@ exchange / broker / macro API - historical bars exclude the currently forming candle when the provider can identify it; - exact catalog resolution distinguishes spot, swaps, futures, and options; +- user-owned tabular schemas are reflected and mapped explicitly rather than + requiring a PineForge-owned DDL; - macro observations retain release and vintage timestamps to avoid revised- data lookahead; - backtest reports identify both the resolved market and runtime versions. diff --git a/docs/local-data.md b/docs/local-data.md new file mode 100644 index 0000000..3537ec6 --- /dev/null +++ b/docs/local-data.md @@ -0,0 +1,296 @@ +# Connect local files and databases + +PineForge Data can read historical OHLCV from a local CSV file, a SQLite table +or view, or any synchronous database supported by SQLAlchemy 2.x. The source +does not need to use PineForge column names or a prescribed DDL. + +The safe boundary is a runtime column mapping: + +```text +your file / table / view + ↓ reflect header or database metadata +timestamp, open, high, low, close, volume, [symbol], [timeframe] + ↓ validate and normalize +PineForge Bar records +``` + +PineForge Data never modifies a connected source. The native SQLite provider +opens the database in read-only mode. The SQLAlchemy provider performs +reflection and `SELECT` statements only; use a database account with read-only +permissions as defense in depth. + +## Runtime schema discovery + +Every tabular provider exposes `inspect_schema()` before data is loaded: + +```python +from pineforge_data import CsvBarProvider + + +async def inspect_file(): + provider = CsvBarProvider("./vendor-export.csv", venue="research") + schema = await provider.inspect_schema() + for column in schema.columns: + print(column.name, column.data_type, column.nullable) +``` + +`TabularSchema.infer_bar_mapping()` recognizes common names such as +`timestamp`, `datetime`, `open_time`, `PX_OPEN`, `qty`, `ticker`, and +`interval`. Supply only the unusual or ambiguous fields: + +```python +mapping = schema.infer_bar_mapping( + { + "timestamp": "Bucket Start", + "volume": "Total Traded Qty", + } +) +``` + +Inference never chooses between multiple plausible columns. A missing or +ambiguous field raises `SchemaMappingError` with the available columns and the +overrides needed to continue. For a completely custom schema, define all six +required fields explicitly: + +```python +from pineforge_data import BarColumnMapping + +mapping = BarColumnMapping( + timestamp="epoch seconds", + open="first px", + high="top px", + low="bottom px", + close="last px", + volume="traded qty", + symbol="security code", # optional + timeframe="bar interval", # optional +) +``` + +Provider constructors also accept a plain mapping. Plain mappings are partial +overrides; any fields not supplied are inferred at runtime. + +## Timestamp values + +Normalized timestamps are always Unix milliseconds. Set `timestamp_unit` for +numeric source values: + +| Value | Meaning | +|---|---| +| `seconds` or `s` | Unix seconds, including exact fractional milliseconds | +| `milliseconds` or `ms` | Unix milliseconds; the default | +| `microseconds` or `us` | Unix microseconds | +| `nanoseconds` or `ns` | Unix nanoseconds | +| `iso8601` | ISO-8601 text or database `datetime` objects | + +ISO-8601 values with an offset or `Z` are unambiguous. Naive text or database +datetimes use `timestamp_timezone`, which defaults to `UTC` and accepts an IANA +zone such as `America/New_York`. Values that cannot be represented exactly at +millisecond precision are rejected rather than rounded silently. + +## CSV + +CSV requires no optional dependency: + +```python +from pineforge_data import BarRequest, CsvBarProvider + + +async def load_csv(): + provider = CsvBarProvider( + "./exports/equity-bars.csv", + venue="research", + mapping={ + "timestamp": "Bucket Start", + "volume": "Total Traded Qty", + }, + timestamp_unit="iso8601", + ) + listing = await provider.resolve_market("AAPL") + return await provider.fetch_bars( + BarRequest( + listing.instrument, + timeframe="1m", + start_ms=1_751_328_000_000, + end_ms=1_751_414_400_000, + ) + ) +``` + +The default format is UTF-8 with a comma delimiter. Use `encoding` and +`delimiter` for other exports. CSV is scanned locally for each request; use a +database backend for large, repeatedly queried datasets. + +Harness configuration: + +```json +{ + "path": "/data/equity-bars.csv", + "timestamp_unit": "iso8601", + "columns": { + "timestamp": "Bucket Start", + "volume": "Total Traded Qty", + "symbol": "Ticker", + "timeframe": "Bar Size" + } +} +``` + +```bash +pineforge-backtest \ + --pine strategy.pine \ + --provider csv \ + --provider-config csv.json \ + --venue research \ + --symbol AAPL \ + --timeframe 1m \ + --start 2025-07-01T00:00:00Z \ + --end 2025-07-02T00:00:00Z +``` + +## SQLite + +The native provider safely quotes table and column identifiers after verifying +them against SQLite schema metadata. Tables and views can contain spaces, +reserved words, or otherwise unconventional names. + +```python +from pineforge_data import SqliteBarProvider + +provider = SqliteBarProvider( + "./warehouse.sqlite3", + table="price candles", + venue="warehouse", + mapping={ + "timestamp": "epoch seconds", + "open": "first px", + "high": "top px", + "low": "bottom px", + "close": "last px", + "volume": "traded qty", + "symbol": "security code", + "timeframe": "bar interval", + }, + timestamp_unit="seconds", +) +``` + +Equivalent harness configuration: + +```json +{ + "path": "/data/warehouse.sqlite3", + "table": "price candles", + "timestamp_unit": "seconds", + "columns": { + "timestamp": "epoch seconds", + "open": "first px", + "high": "top px", + "low": "bottom px", + "close": "last px", + "volume": "traded qty", + "symbol": "security code", + "timeframe": "bar interval" + } +} +``` + +Select it with `--provider sqlite`. Numeric timestamp, symbol, and timeframe +filters are pushed into the query; normalization still checks every returned +record. + +## SQLAlchemy-compatible databases + +Install SQLAlchemy support plus the driver for the target database: + +```bash +pip install 'pineforge-data[database]' psycopg +``` + +The provider uses SQLAlchemy Core reflection, so it works with supported +PostgreSQL, MySQL, MariaDB, Oracle, Microsoft SQL Server, SQLite, and +third-party dialects. It intentionally accepts a table or view, not arbitrary +SQL text: + +```python +import os + +from pineforge_data import SqlAlchemyBarProvider + +provider = SqlAlchemyBarProvider( + os.environ["PINEFORGE_DATABASE_URL"], + table="daily_prices", + schema="market_data", + venue="warehouse", + mapping={"timestamp": "trading_day", "symbol": "security_id"}, + timestamp_unit="iso8601", +) +``` + +For joins, expressions, timezone conversion, or vendor-specific data cleanup, +create a database view and map that view. This preserves parameterized filters +and avoids placing raw SQL in a backtest configuration. + +Keep credentials out of JSON by naming an environment variable: + +```json +{ + "url_env": "PINEFORGE_DATABASE_URL", + "schema": "market_data", + "table": "daily_prices", + "timestamp_unit": "iso8601", + "columns": { + "timestamp": "trading_day", + "symbol": "security_id" + } +} +``` + +Select it with `--provider sqlalchemy`. A literal `url` is also accepted for +local or non-sensitive configurations. `engine_options` passes SQLAlchemy +engine options such as pool settings; driver-specific packages remain the +user's dependency. + +Only synchronous SQLAlchemy engines are supported initially. Provider calls +are moved off the asyncio event loop, so they remain compatible with the +async PineForge harness. + +## Symbol and timeframe behavior + +- When a `symbol` column is mapped, catalog listing and exact resolution use + its distinct values, and every fetch filters it. +- Without a `symbol` column, the source is treated as a single-instrument + dataset. `resolve_market("AAPL")` binds the dataset to that requested name. + Set the provider configuration `symbol` when `list_markets()` must advertise + it before resolution. +- When a `timeframe` column is mapped, every request filters it exactly. +- For a single-timeframe dataset without such a column, set configuration + `timeframe` to reject accidental requests at another resolution. + +All sources return rows in ascending time order, apply the half-open interval +`[start_ms, end_ms)`, and apply `limit` after normalization. Duplicate +timestamps for the same requested symbol and timeframe raise +`TabularDataError`; PineForge Data never silently selects one duplicate. + +Local rows are assumed to be confirmed bars because a file or database usually +does not expose a provider clock or candle-close capability. The user owns that +snapshot guarantee. Invalid OHLC relationships, negative volume, missing +values, non-finite numbers, and sub-millisecond timestamp loss fail explicitly. + +## Configuration reference + +| Key | CSV | SQLite | SQLAlchemy | Purpose | +|---|---:|---:|---:|---| +| `path` | required | required | — | Local file/database path | +| `table` | — | required | required | Reflected table or view | +| `url` / `url_env` | — | — | one required | SQLAlchemy connection URL or its environment variable | +| `schema` | — | — | optional | Database schema name | +| `columns` | optional | optional | optional | Partial or complete canonical-to-source mapping | +| `timestamp_unit` | optional | optional | optional | Numeric unit or `iso8601` | +| `timestamp_timezone` | optional | optional | optional | IANA zone for naive datetime values | +| `symbol` | optional | optional | optional | Fixed identity when no symbol column exists | +| `timeframe` | optional | optional | optional | Fixed resolution when no timeframe column exists | +| `encoding`, `delimiter` | optional | — | — | CSV parser settings | +| `engine_options` | — | — | optional | SQLAlchemy `create_engine()` options | + +Unknown configuration keys fail early to catch misspellings. diff --git a/docs/providers.md b/docs/providers.md index ae8dca5..bcc3f3b 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -7,6 +7,10 @@ They are selected independently from venues: - venue: an exchange or broker environment, such as `kraken`; - symbol: the exact normalized market symbol on that provider. +Built-in `csv`, `sqlite`, and `sqlalchemy` providers expose the same catalog and +historical-bar protocol for user-owned data. They discover and map arbitrary +columns at runtime; see [Local files and databases](local-data.md). + ## Create a registered provider ```python diff --git a/pyproject.toml b/pyproject.toml index c9a4e9b..70be95d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,7 @@ pineforge-backtest-server = "pineforge_data.server:main" [project.optional-dependencies] ccxt = ["ccxt>=4.5.64,<5"] +database = ["sqlalchemy>=2.0,<3"] server = [ "fastapi>=0.139,<0.140", "uvicorn>=0.51,<0.52", @@ -39,6 +40,7 @@ dev = [ "mypy>=1.13", "pytest>=8.3", "ruff>=0.8", + "sqlalchemy>=2.0,<3", "uvicorn>=0.51,<0.52", ] diff --git a/src/pineforge_data/__init__.py b/src/pineforge_data/__init__.py index 18c9c77..7617ce2 100644 --- a/src/pineforge_data/__init__.py +++ b/src/pineforge_data/__init__.py @@ -27,11 +27,13 @@ TradeTick, ) from .providers import ( + BarColumnMapping, CcxtCapabilityError, CcxtDataError, CcxtDependencyError, CcxtError, CcxtProvider, + CsvBarProvider, HistoricalBarProvider, LiveTradeProvider, MacroDataProvider, @@ -42,6 +44,15 @@ ProviderNotFoundError, ProviderRegistry, ProviderRegistryError, + SchemaMappingError, + SourceColumn, + SqlAlchemyBarProvider, + SqlAlchemyDependencyError, + SqliteBarProvider, + TabularBarProvider, + TabularDataError, + TabularSchema, + TimestampUnit, create_provider, default_registry, ) @@ -56,6 +67,7 @@ "BacktestReport", "BacktestServerError", "Bar", + "BarColumnMapping", "BarRequest", "CcxtCapabilityError", "CcxtDataError", @@ -64,6 +76,7 @@ "CcxtProvider", "CompileCache", "ContractSpec", + "CsvBarProvider", "DockerBacktestRuntime", "DockerExecutionError", "DockerPrerequisiteError", @@ -93,6 +106,15 @@ "ProviderRegistry", "ProviderRegistryError", "ReleaseContractError", + "SchemaMappingError", + "SourceColumn", + "SqlAlchemyBarProvider", + "SqlAlchemyDependencyError", + "SqliteBarProvider", + "TabularBarProvider", + "TabularDataError", + "TabularSchema", + "TimestampUnit", "TradeSubscription", "TradeTick", "create_provider", diff --git a/src/pineforge_data/providers/__init__.py b/src/pineforge_data/providers/__init__.py index c8b061a..e960ab3 100644 --- a/src/pineforge_data/providers/__init__.py +++ b/src/pineforge_data/providers/__init__.py @@ -15,6 +15,7 @@ CcxtError, CcxtProvider, ) +from .local import CsvBarProvider, SqliteBarProvider from .registry import ( ENTRY_POINT_GROUP, ProviderFactory, @@ -24,14 +25,26 @@ create_provider, default_registry, ) +from .sqlalchemy import SqlAlchemyBarProvider, SqlAlchemyDependencyError +from .tabular import ( + BarColumnMapping, + SchemaMappingError, + SourceColumn, + TabularBarProvider, + TabularDataError, + TabularSchema, + TimestampUnit, +) __all__ = [ "ENTRY_POINT_GROUP", + "BarColumnMapping", "CcxtCapabilityError", "CcxtDataError", "CcxtDependencyError", "CcxtError", "CcxtProvider", + "CsvBarProvider", "HistoricalBarProvider", "LiveTradeProvider", "MacroDataProvider", @@ -42,6 +55,15 @@ "ProviderNotFoundError", "ProviderRegistry", "ProviderRegistryError", + "SchemaMappingError", + "SourceColumn", + "SqlAlchemyBarProvider", + "SqlAlchemyDependencyError", + "SqliteBarProvider", + "TabularBarProvider", + "TabularDataError", + "TabularSchema", + "TimestampUnit", "create_provider", "default_registry", ] diff --git a/src/pineforge_data/providers/local.py b/src/pineforge_data/providers/local.py new file mode 100644 index 0000000..b385f56 --- /dev/null +++ b/src/pineforge_data/providers/local.py @@ -0,0 +1,207 @@ +"""CSV and SQLite providers for user-owned historical bars.""" + +from __future__ import annotations + +import csv +import sqlite3 +from pathlib import Path +from typing import TextIO + +from ..models import Instrument +from ..requests import BarRequest +from .tabular import ( + BarColumnMapping, + ColumnMappingInput, + SourceColumn, + TabularBarProvider, + TabularDataError, + TabularRow, + TabularSchema, + TimestampUnit, + numeric_source_bounds, +) + + +class CsvBarProvider(TabularBarProvider): + """Read historical bars from a local CSV file with a runtime column mapping.""" + + def __init__( + self, + path: str | Path, + *, + venue: str = "local", + mapping: ColumnMappingInput = None, + timestamp_unit: TimestampUnit | str = TimestampUnit.MILLISECONDS, + timestamp_timezone: str = "UTC", + instrument: Instrument | None = None, + timeframe: str | None = None, + encoding: str = "utf-8-sig", + delimiter: str = ",", + ) -> None: + super().__init__( + venue=venue, + mapping=mapping, + timestamp_unit=timestamp_unit, + timestamp_timezone=timestamp_timezone, + instrument=instrument, + timeframe=timeframe, + ) + if not encoding: + raise ValueError("encoding must not be empty") + if len(delimiter) != 1: + raise ValueError("delimiter must be exactly one character") + self.path = Path(path).expanduser().resolve() + self.encoding = encoding + self.delimiter = delimiter + self.name = f"csv:{venue}" + + def _reader(self) -> tuple[csv.DictReader[str], TextIO]: + if not self.path.is_file(): + raise FileNotFoundError(f"CSV file not found: {self.path}") + handle = self.path.open("r", encoding=self.encoding, newline="") + reader = csv.DictReader(handle, delimiter=self.delimiter) + return reader, handle + + def _inspect_schema_sync(self) -> TabularSchema: + reader, handle = self._reader() + try: + fieldnames = reader.fieldnames + if not fieldnames: + raise TabularDataError(f"CSV file has no header: {self.path}") + if any(not fieldname for fieldname in fieldnames): + raise TabularDataError("CSV header contains an empty column name") + return TabularSchema( + source=str(self.path), + columns=tuple(SourceColumn(fieldname, "text", None) for fieldname in fieldnames), + ) + finally: + handle.close() + + def _all_rows_sync(self) -> tuple[TabularRow, ...]: + reader, handle = self._reader() + try: + rows: list[TabularRow] = [] + for line_number, row in enumerate(reader, start=2): + if None in row: + raise TabularDataError(f"CSV row {line_number} has more values than the header") + rows.append(dict(row)) + return tuple(rows) + finally: + handle.close() + + def _read_rows_sync( + self, mapping: BarColumnMapping, request: BarRequest + ) -> tuple[TabularRow, ...]: + del mapping, request + return self._all_rows_sync() + + def _distinct_values_sync(self, column: str) -> tuple[object, ...]: + return tuple(row[column] for row in self._all_rows_sync()) + + +def _quote_identifier(identifier: str) -> str: + return '"' + identifier.replace('"', '""') + '"' + + +class SqliteBarProvider(TabularBarProvider): + """Read historical bars from one SQLite table or view in read-only mode.""" + + def __init__( + self, + path: str | Path, + table: str, + *, + venue: str = "local", + mapping: ColumnMappingInput = None, + timestamp_unit: TimestampUnit | str = TimestampUnit.MILLISECONDS, + timestamp_timezone: str = "UTC", + instrument: Instrument | None = None, + timeframe: str | None = None, + ) -> None: + super().__init__( + venue=venue, + mapping=mapping, + timestamp_unit=timestamp_unit, + timestamp_timezone=timestamp_timezone, + instrument=instrument, + timeframe=timeframe, + ) + if not table: + raise ValueError("table must not be empty") + self.path = Path(path).expanduser().resolve() + self.table = table + self.name = f"sqlite:{venue}" + + def _connect(self) -> sqlite3.Connection: + if not self.path.is_file(): + raise FileNotFoundError(f"SQLite database not found: {self.path}") + connection = sqlite3.connect(f"{self.path.as_uri()}?mode=ro", uri=True) + connection.row_factory = sqlite3.Row + return connection + + def _require_table(self, connection: sqlite3.Connection) -> None: + found = connection.execute( + "SELECT 1 FROM sqlite_schema WHERE type IN ('table', 'view') AND name = ? LIMIT 1", + (self.table,), + ).fetchone() + if found is None: + raise TabularDataError(f"SQLite table or view not found: {self.table}") + + def _inspect_schema_sync(self) -> TabularSchema: + with self._connect() as connection: + self._require_table(connection) + rows = connection.execute( + f"PRAGMA table_info({_quote_identifier(self.table)})" + ).fetchall() + if not rows: + raise TabularDataError(f"SQLite source has no columns: {self.table}") + return TabularSchema( + source=f"{self.path}#{self.table}", + columns=tuple( + SourceColumn( + name=str(row[1]), + data_type=str(row[2] or ""), + nullable=not bool(row[3]) and not bool(row[5]), + ) + for row in rows + ), + ) + + def _read_rows_sync( + self, mapping: BarColumnMapping, request: BarRequest + ) -> tuple[TabularRow, ...]: + selected = ", ".join(_quote_identifier(column) for column in mapping.columns) + predicates: list[str] = [] + parameters: list[object] = [] + if mapping.symbol is not None: + predicates.append(f"{_quote_identifier(mapping.symbol)} = ?") + parameters.append(request.instrument.symbol) + if mapping.timeframe is not None: + predicates.append(f"{_quote_identifier(mapping.timeframe)} = ?") + parameters.append(request.timeframe) + bounds = numeric_source_bounds(request.start_ms, request.end_ms, self.timestamp_unit) + if bounds is not None: + predicates.append(f"{_quote_identifier(mapping.timestamp)} >= ?") + parameters.append(bounds[0]) + predicates.append(f"{_quote_identifier(mapping.timestamp)} < ?") + parameters.append(bounds[1]) + where = f" WHERE {' AND '.join(predicates)}" if predicates else "" + statement = ( + f"SELECT {selected} FROM {_quote_identifier(self.table)}{where} " + f"ORDER BY {_quote_identifier(mapping.timestamp)}" + ) + with self._connect() as connection: + self._require_table(connection) + rows = connection.execute(statement, parameters).fetchall() + return tuple(dict(row) for row in rows) + + def _distinct_values_sync(self, column: str) -> tuple[object, ...]: + statement = ( + f"SELECT DISTINCT {_quote_identifier(column)} " + f"FROM {_quote_identifier(self.table)} " + f"ORDER BY {_quote_identifier(column)}" + ) + with self._connect() as connection: + self._require_table(connection) + rows = connection.execute(statement).fetchall() + return tuple(row[0] for row in rows) diff --git a/src/pineforge_data/providers/registry.py b/src/pineforge_data/providers/registry.py index 8a8c4fa..3dffdc6 100644 --- a/src/pineforge_data/providers/registry.py +++ b/src/pineforge_data/providers/registry.py @@ -2,12 +2,16 @@ from __future__ import annotations +import os from collections.abc import Mapping from importlib.metadata import EntryPoint, entry_points from typing import Protocol, cast +from ..models import Instrument from .base import MarketDataProvider from .ccxt import CcxtProvider +from .local import CsvBarProvider, SqliteBarProvider +from .sqlalchemy import SqlAlchemyBarProvider ENTRY_POINT_GROUP = "pineforge_data.providers" @@ -28,6 +32,127 @@ def _ccxt_factory(venue: str, config: Mapping[str, object]) -> MarketDataProvide return CcxtProvider(venue, config=config) +def _unknown_config(config: Mapping[str, object], allowed: set[str]) -> None: + unknown = sorted(config.keys() - allowed) + if unknown: + raise ValueError(f"unknown provider configuration key(s): {', '.join(unknown)}") + + +def _required_text(config: Mapping[str, object], key: str) -> str: + value = config.get(key) + if not isinstance(value, str) or not value: + raise ValueError(f"provider configuration {key!r} must be a non-empty string") + return value + + +def _optional_text(config: Mapping[str, object], key: str) -> str | None: + value = config.get(key) + if value is None: + return None + if not isinstance(value, str) or not value: + raise ValueError(f"provider configuration {key!r} must be a non-empty string") + return value + + +def _column_overrides(config: Mapping[str, object]) -> Mapping[str, str] | None: + value = config.get("columns") + if value is None: + return None + if not isinstance(value, Mapping): + raise ValueError("provider configuration 'columns' must be an object") + overrides: dict[str, str] = {} + allowed = {"timestamp", "open", "high", "low", "close", "volume", "symbol", "timeframe"} + for field, column in value.items(): + if not isinstance(field, str) or not isinstance(column, str): + raise ValueError("provider column mappings must have string keys and values") + if field not in allowed: + raise ValueError(f"unknown canonical column field: {field}") + if not column: + raise ValueError(f"provider column mapping for {field!r} must not be empty") + overrides[field] = column + return overrides + + +def _instrument(venue: str, config: Mapping[str, object]) -> Instrument | None: + symbol = _optional_text(config, "symbol") + return None if symbol is None else Instrument(symbol, venue=venue, provider_id=symbol) + + +_COMMON_TABULAR_KEYS = { + "columns", + "timestamp_unit", + "timestamp_timezone", + "symbol", + "timeframe", +} + + +def _csv_factory(venue: str, config: Mapping[str, object]) -> MarketDataProvider: + _unknown_config(config, _COMMON_TABULAR_KEYS | {"path", "encoding", "delimiter"}) + return CsvBarProvider( + _required_text(config, "path"), + venue=venue, + mapping=_column_overrides(config), + timestamp_unit=_optional_text(config, "timestamp_unit") or "milliseconds", + timestamp_timezone=_optional_text(config, "timestamp_timezone") or "UTC", + instrument=_instrument(venue, config), + timeframe=_optional_text(config, "timeframe"), + encoding=_optional_text(config, "encoding") or "utf-8-sig", + delimiter=_optional_text(config, "delimiter") or ",", + ) + + +def _sqlite_factory(venue: str, config: Mapping[str, object]) -> MarketDataProvider: + _unknown_config(config, _COMMON_TABULAR_KEYS | {"path", "table"}) + return SqliteBarProvider( + _required_text(config, "path"), + _required_text(config, "table"), + venue=venue, + mapping=_column_overrides(config), + timestamp_unit=_optional_text(config, "timestamp_unit") or "milliseconds", + timestamp_timezone=_optional_text(config, "timestamp_timezone") or "UTC", + instrument=_instrument(venue, config), + timeframe=_optional_text(config, "timeframe"), + ) + + +def _sqlalchemy_factory(venue: str, config: Mapping[str, object]) -> MarketDataProvider: + _unknown_config( + config, + _COMMON_TABULAR_KEYS | {"url", "url_env", "table", "schema", "engine_options"}, + ) + url = _optional_text(config, "url") + url_env = _optional_text(config, "url_env") + if url is not None and url_env is not None: + raise ValueError("configure only one of 'url' and 'url_env'") + if url_env is not None: + url = os.environ.get(url_env) + if not url: + raise ValueError(f"database URL environment variable is unset: {url_env}") + if url is None: + raise ValueError("provider configuration requires 'url' or 'url_env'") + engine_options = config.get("engine_options") + normalized_engine_options: Mapping[str, object] | None = None + if engine_options is not None: + if not isinstance(engine_options, Mapping) or not all( + isinstance(key, str) for key in engine_options + ): + raise ValueError("provider configuration 'engine_options' must be an object") + normalized_engine_options = cast(Mapping[str, object], engine_options) + return SqlAlchemyBarProvider( + url, + _required_text(config, "table"), + venue=venue, + schema=_optional_text(config, "schema"), + mapping=_column_overrides(config), + timestamp_unit=_optional_text(config, "timestamp_unit") or "milliseconds", + timestamp_timezone=_optional_text(config, "timestamp_timezone") or "UTC", + instrument=_instrument(venue, config), + timeframe=_optional_text(config, "timeframe"), + engine_options=normalized_engine_options, + ) + + class ProviderRegistry: """Resolve built-ins and externally installed provider factories by name.""" @@ -35,6 +160,9 @@ def __init__(self, *, include_builtin: bool = True) -> None: self._factories: dict[str, ProviderFactory] = {} if include_builtin: self.register("ccxt", _ccxt_factory) + self.register("csv", _csv_factory) + self.register("sqlite", _sqlite_factory) + self.register("sqlalchemy", _sqlalchemy_factory) @staticmethod def _normalize_name(name: str) -> str: diff --git a/src/pineforge_data/providers/sqlalchemy.py b/src/pineforge_data/providers/sqlalchemy.py new file mode 100644 index 0000000..81bad7f --- /dev/null +++ b/src/pineforge_data/providers/sqlalchemy.py @@ -0,0 +1,138 @@ +"""SQLAlchemy Core provider for reflected user-owned database tables.""" + +from __future__ import annotations + +from collections.abc import Mapping +from importlib import import_module +from threading import Lock +from types import ModuleType +from typing import Any, cast + +from ..models import Instrument +from ..requests import BarRequest +from .tabular import ( + BarColumnMapping, + ColumnMappingInput, + SourceColumn, + TabularBarProvider, + TabularRow, + TabularSchema, + TimestampUnit, + numeric_source_bounds, +) + + +class SqlAlchemyDependencyError(RuntimeError): + """SQLAlchemy support was requested without the optional dependency.""" + + +def _load_sqlalchemy() -> ModuleType: + try: + return import_module("sqlalchemy") + except ModuleNotFoundError as exc: + if exc.name == "sqlalchemy" or (exc.name and exc.name.startswith("sqlalchemy.")): + raise SqlAlchemyDependencyError( + "SQLAlchemy is not installed; install pineforge-data[database]" + ) from exc + raise + + +class SqlAlchemyBarProvider(TabularBarProvider): + """Reflect and query one table or view through a synchronous SQLAlchemy engine.""" + + def __init__( + self, + url: str, + table: str, + *, + venue: str = "database", + schema: str | None = None, + mapping: ColumnMappingInput = None, + timestamp_unit: TimestampUnit | str = TimestampUnit.MILLISECONDS, + timestamp_timezone: str = "UTC", + instrument: Instrument | None = None, + timeframe: str | None = None, + engine_options: Mapping[str, object] | None = None, + ) -> None: + super().__init__( + venue=venue, + mapping=mapping, + timestamp_unit=timestamp_unit, + timestamp_timezone=timestamp_timezone, + instrument=instrument, + timeframe=timeframe, + ) + if not url.strip(): + raise ValueError("SQLAlchemy URL must not be empty") + if not table: + raise ValueError("table must not be empty") + if schema is not None and not schema: + raise ValueError("schema must be None or a non-empty string") + + self._sa = _load_sqlalchemy() + options = dict(engine_options or {}) + options.setdefault("hide_parameters", True) + self._engine: Any = self._sa.create_engine(url, **options) + self._table: Any | None = None + self._reflection_lock = Lock() + self.table = table + self.schema_name = schema + self.name = f"sqlalchemy:{venue}" + + def _reflected_table(self) -> Any: + with self._reflection_lock: + if self._table is None: + metadata = self._sa.MetaData() + self._table = self._sa.Table( + self.table, + metadata, + schema=self.schema_name, + autoload_with=self._engine, + ) + return self._table + + def _inspect_schema_sync(self) -> TabularSchema: + table = self._reflected_table() + qualified_table = f"{self.schema_name}.{self.table}" if self.schema_name else self.table + return TabularSchema( + source=f"sqlalchemy:{self.venue}#{qualified_table}", + columns=tuple( + SourceColumn( + name=str(column.name), + data_type=str(column.type), + nullable=bool(column.nullable), + ) + for column in table.columns + ), + ) + + def _read_rows_sync( + self, mapping: BarColumnMapping, request: BarRequest + ) -> tuple[TabularRow, ...]: + table = self._reflected_table() + columns = [table.c[column] for column in mapping.columns] + statement = self._sa.select(*columns) + if mapping.symbol is not None: + statement = statement.where(table.c[mapping.symbol] == request.instrument.symbol) + if mapping.timeframe is not None: + statement = statement.where(table.c[mapping.timeframe] == request.timeframe) + bounds = numeric_source_bounds(request.start_ms, request.end_ms, self.timestamp_unit) + if bounds is not None: + statement = statement.where(table.c[mapping.timestamp] >= bounds[0]) + statement = statement.where(table.c[mapping.timestamp] < bounds[1]) + statement = statement.order_by(table.c[mapping.timestamp]) + + with self._engine.connect() as connection: + result = connection.execute(statement).mappings().all() + return tuple( + cast(TabularRow, {str(key): value for key, value in row.items()}) for row in result + ) + + def _distinct_values_sync(self, column: str) -> tuple[object, ...]: + table = self._reflected_table() + statement = self._sa.select(table.c[column]).distinct().order_by(table.c[column]) + with self._engine.connect() as connection: + return tuple(connection.execute(statement).scalars().all()) + + def _close_sync(self) -> None: + self._engine.dispose() diff --git a/src/pineforge_data/providers/tabular.py b/src/pineforge_data/providers/tabular.py new file mode 100644 index 0000000..fece53c --- /dev/null +++ b/src/pineforge_data/providers/tabular.py @@ -0,0 +1,530 @@ +"""Schema discovery and normalization shared by user-owned tabular sources.""" + +from __future__ import annotations + +import asyncio +from abc import ABC, abstractmethod +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, replace +from datetime import date, datetime, time +from decimal import Decimal, InvalidOperation +from enum import StrEnum +from math import isfinite +from types import MappingProxyType +from typing import TypeAlias +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from ..models import Bar, Instrument, MarketListing +from ..requests import BarRequest, MarketQuery +from .base import MarketNotFoundError + +TabularRow: TypeAlias = Mapping[str, object] + + +class TabularDataError(RuntimeError): + """A user-owned tabular source cannot be normalized safely.""" + + +class SchemaMappingError(TabularDataError): + """Source columns cannot be mapped unambiguously to OHLCV fields.""" + + +class TimestampUnit(StrEnum): + """Storage unit used by numeric timestamps.""" + + SECONDS = "seconds" + MILLISECONDS = "milliseconds" + MICROSECONDS = "microseconds" + NANOSECONDS = "nanoseconds" + ISO8601 = "iso8601" + + @classmethod + def parse(cls, value: TimestampUnit | str) -> TimestampUnit: + if isinstance(value, cls): + return value + aliases = { + "s": cls.SECONDS, + "sec": cls.SECONDS, + "second": cls.SECONDS, + "seconds": cls.SECONDS, + "ms": cls.MILLISECONDS, + "millisecond": cls.MILLISECONDS, + "milliseconds": cls.MILLISECONDS, + "us": cls.MICROSECONDS, + "microsecond": cls.MICROSECONDS, + "microseconds": cls.MICROSECONDS, + "ns": cls.NANOSECONDS, + "nanosecond": cls.NANOSECONDS, + "nanoseconds": cls.NANOSECONDS, + "iso": cls.ISO8601, + "iso8601": cls.ISO8601, + "datetime": cls.ISO8601, + } + normalized = value.strip().casefold() + try: + return aliases[normalized] + except KeyError as exc: + choices = ", ".join(unit.value for unit in cls) + raise ValueError( + f"unknown timestamp unit {value!r}; expected one of: {choices}" + ) from exc + + +@dataclass(frozen=True, slots=True) +class SourceColumn: + """One column discovered from a file header or database reflection.""" + + name: str + data_type: str = "" + nullable: bool | None = None + + def __post_init__(self) -> None: + if not self.name: + raise ValueError("source column name must not be empty") + + +_REQUIRED_FIELDS = ("timestamp", "open", "high", "low", "close", "volume") +_OPTIONAL_FIELDS = ("symbol", "timeframe") +_MAPPING_FIELDS = _REQUIRED_FIELDS + _OPTIONAL_FIELDS +_ALIASES: Mapping[str, tuple[str, ...]] = MappingProxyType( + { + "timestamp": ( + "timestamp", + "timestamp_ms", + "time", + "datetime", + "date", + "ts", + "open_time", + "bar_time", + ), + "open": ("open", "open_price", "price_open", "px_open", "o"), + "high": ("high", "high_price", "price_high", "px_high", "h"), + "low": ("low", "low_price", "price_low", "px_low", "l"), + "close": ("close", "close_price", "price_close", "px_close", "c"), + "volume": ("volume", "vol", "base_volume", "quantity", "qty", "v"), + "symbol": ("symbol", "ticker", "instrument", "market", "security"), + "timeframe": ("timeframe", "interval", "resolution", "bar_size"), + } +) + + +def _normalized_column_name(value: str) -> str: + return "".join(character for character in value.casefold() if character.isalnum()) + + +def _validate_overrides(overrides: Mapping[str, str] | None) -> dict[str, str]: + normalized = dict(overrides or {}) + unknown = sorted(normalized.keys() - set(_MAPPING_FIELDS)) + if unknown: + raise SchemaMappingError( + f"unknown canonical column field(s): {', '.join(unknown)}; " + f"expected: {', '.join(_MAPPING_FIELDS)}" + ) + for field, column in normalized.items(): + if not isinstance(column, str) or not column: + raise SchemaMappingError(f"column override for {field!r} must be a non-empty string") + return normalized + + +@dataclass(frozen=True, slots=True) +class BarColumnMapping: + """Map canonical PineForge bar fields to arbitrary source column names.""" + + timestamp: str + open: str + high: str + low: str + close: str + volume: str + symbol: str | None = None + timeframe: str | None = None + + def __post_init__(self) -> None: + for field in _REQUIRED_FIELDS: + column = getattr(self, field) + if not isinstance(column, str) or not column: + raise ValueError(f"{field} column must be a non-empty string") + for field in _OPTIONAL_FIELDS: + column = getattr(self, field) + if column is not None and not column: + raise ValueError(f"{field} column must be None or a non-empty string") + + selected = [ + column for field in _MAPPING_FIELDS if (column := getattr(self, field)) is not None + ] + duplicates = sorted({column for column in selected if selected.count(column) > 1}) + if duplicates: + raise ValueError( + "one source column cannot map to multiple fields: " + ", ".join(duplicates) + ) + + @property + def columns(self) -> tuple[str, ...]: + """Return mapped source columns in canonical order without duplicates.""" + + return tuple( + column for field in _MAPPING_FIELDS if (column := getattr(self, field)) is not None + ) + + @classmethod + def infer( + cls, + columns: Sequence[str], + overrides: Mapping[str, str] | None = None, + ) -> BarColumnMapping: + """Infer common OHLCV names while requiring explicit ambiguous choices.""" + + available = tuple(columns) + if not available: + raise SchemaMappingError("source exposes no columns") + if len(set(available)) != len(available): + duplicates = sorted({name for name in available if available.count(name) > 1}) + raise SchemaMappingError(f"source has duplicate column names: {', '.join(duplicates)}") + + selected = _validate_overrides(overrides) + missing_overrides = sorted(set(selected.values()) - set(available)) + if missing_overrides: + raise SchemaMappingError( + "mapped column(s) not found: " + f"{', '.join(missing_overrides)}; available columns: {', '.join(available)}" + ) + + used = set(selected.values()) + ambiguous: dict[str, list[str]] = {} + for field in _MAPPING_FIELDS: + if field in selected: + continue + aliases = {_normalized_column_name(alias) for alias in _ALIASES[field]} + matches = [ + column + for column in available + if column not in used and _normalized_column_name(column) in aliases + ] + if len(matches) == 1: + selected[field] = matches[0] + used.add(matches[0]) + elif len(matches) > 1: + ambiguous[field] = matches + + missing = [field for field in _REQUIRED_FIELDS if field not in selected] + if missing or ambiguous: + details: list[str] = [] + if missing: + details.append("missing mappings for " + ", ".join(missing)) + if ambiguous: + choices = "; ".join( + f"{field}: {', '.join(matches)}" for field, matches in sorted(ambiguous.items()) + ) + details.append("ambiguous mappings for " + choices) + details.append("available columns: " + ", ".join(available)) + details.append( + "pass mapping={'timestamp': 'your_time', ...} to override only unusual names" + ) + raise SchemaMappingError("; ".join(details)) + + return cls( + timestamp=selected["timestamp"], + open=selected["open"], + high=selected["high"], + low=selected["low"], + close=selected["close"], + volume=selected["volume"], + symbol=selected.get("symbol"), + timeframe=selected.get("timeframe"), + ) + + def validate(self, columns: Sequence[str]) -> None: + """Require every selected name to exist in the reflected source schema.""" + + missing = sorted(set(self.columns) - set(columns)) + if missing: + raise SchemaMappingError( + f"mapped column(s) not found: {', '.join(missing)}; " + f"available columns: {', '.join(columns)}" + ) + + +ColumnMappingInput: TypeAlias = BarColumnMapping | Mapping[str, str] | None + + +@dataclass(frozen=True, slots=True) +class TabularSchema: + """Runtime description of a user-owned file, table, or view.""" + + source: str + columns: tuple[SourceColumn, ...] + + def __post_init__(self) -> None: + names = self.column_names + if not self.source: + raise ValueError("schema source must not be empty") + if len(set(names)) != len(names): + duplicates = sorted({name for name in names if names.count(name) > 1}) + raise SchemaMappingError(f"source has duplicate column names: {', '.join(duplicates)}") + + @property + def column_names(self) -> tuple[str, ...]: + return tuple(column.name for column in self.columns) + + def infer_bar_mapping(self, overrides: Mapping[str, str] | None = None) -> BarColumnMapping: + """Build a validated mapping from reflected columns and optional overrides.""" + + return BarColumnMapping.infer(self.column_names, overrides) + + +def _numeric_timestamp_ms(value: object, unit: TimestampUnit) -> int: + try: + numeric = Decimal(str(value).strip()) + except (InvalidOperation, AttributeError) as exc: + raise ValueError(f"invalid numeric timestamp: {value!r}") from exc + if not numeric.is_finite(): + raise ValueError("timestamp must be finite") + scale = { + TimestampUnit.SECONDS: Decimal(1_000), + TimestampUnit.MILLISECONDS: Decimal(1), + TimestampUnit.MICROSECONDS: Decimal("0.001"), + TimestampUnit.NANOSECONDS: Decimal("0.000001"), + }[unit] + milliseconds = numeric * scale + if milliseconds != milliseconds.to_integral_value(): + raise ValueError("timestamp cannot be represented exactly at millisecond precision") + normalized = int(milliseconds) + if normalized < 0 or normalized > 2**63 - 1: + raise ValueError("timestamp must fit a non-negative signed 64-bit millisecond value") + return normalized + + +def _timestamp_ms(value: object, unit: TimestampUnit, timezone: ZoneInfo) -> int: + if isinstance(value, bool) or value is None: + raise ValueError("timestamp must be numeric, ISO-8601 text, or a datetime") + + if isinstance(value, datetime): + parsed = value + elif isinstance(value, date): + parsed = datetime.combine(value, time.min) + elif isinstance(value, str): + text = value.strip() + if not text: + raise ValueError("timestamp must not be empty") + if unit is not TimestampUnit.ISO8601: + try: + Decimal(text) + except InvalidOperation: + pass + else: + return _numeric_timestamp_ms(text, unit) + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"invalid ISO-8601 timestamp: {value!r}") from exc + else: + if unit is TimestampUnit.ISO8601: + raise ValueError("timestamp_unit=iso8601 requires text or datetime values") + return _numeric_timestamp_ms(value, unit) + + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone) + normalized = int(parsed.timestamp() * 1_000) + if normalized < 0 or normalized > 2**63 - 1: + raise ValueError("timestamp must fit a non-negative signed 64-bit millisecond value") + return normalized + + +def _number(value: object, field: str) -> float: + if isinstance(value, bool) or value is None: + raise ValueError(f"{field} must be numeric") + try: + normalized = float(value if isinstance(value, (int, float, str)) else str(value)) + except (TypeError, ValueError) as exc: + raise ValueError(f"{field} must be numeric") from exc + if not isfinite(normalized): + raise ValueError(f"{field} must be finite") + return normalized + + +def numeric_source_bounds( + start_ms: int, end_ms: int, unit: TimestampUnit +) -> tuple[int | float, int | float] | None: + """Translate millisecond request bounds for safe numeric SQL pushdown.""" + + if unit is TimestampUnit.ISO8601: + return None + if unit is TimestampUnit.SECONDS: + return start_ms / 1_000, end_ms / 1_000 + if unit is TimestampUnit.MILLISECONDS: + return start_ms, end_ms + if unit is TimestampUnit.MICROSECONDS: + return start_ms * 1_000, end_ms * 1_000 + return start_ms * 1_000_000, end_ms * 1_000_000 + + +class TabularBarProvider(ABC): + """Common catalog and OHLCV behavior for schema-mapped tabular sources.""" + + name: str + venue: str + + def __init__( + self, + *, + venue: str, + mapping: ColumnMappingInput = None, + timestamp_unit: TimestampUnit | str = TimestampUnit.MILLISECONDS, + timestamp_timezone: str = "UTC", + instrument: Instrument | None = None, + timeframe: str | None = None, + ) -> None: + if not venue.strip(): + raise ValueError("venue must not be empty") + if timeframe is not None and not timeframe.strip(): + raise ValueError("timeframe must not be empty") + if instrument is not None and instrument.venue and instrument.venue != venue: + raise ValueError( + f"instrument venue {instrument.venue!r} does not match provider venue {venue!r}" + ) + try: + self._timestamp_timezone = ZoneInfo(timestamp_timezone) + except ZoneInfoNotFoundError as exc: + raise ValueError(f"unknown IANA timestamp timezone: {timestamp_timezone}") from exc + + self.venue = venue + self.timestamp_unit = TimestampUnit.parse(timestamp_unit) + self.timestamp_timezone = timestamp_timezone + self._instrument_template = instrument + self._timeframe = timeframe + self._explicit_mapping: BarColumnMapping | None = None + self._mapping_overrides: dict[str, str] = {} + if isinstance(mapping, BarColumnMapping): + self._explicit_mapping = mapping + elif mapping is not None: + self._mapping_overrides = _validate_overrides(mapping) + + async def inspect_schema(self) -> TabularSchema: + """Reflect columns without reading or normalizing the full dataset.""" + + return await asyncio.to_thread(self._inspect_schema_sync) + + def _resolved_mapping_sync(self) -> BarColumnMapping: + schema = self._inspect_schema_sync() + if self._explicit_mapping is not None: + self._explicit_mapping.validate(schema.column_names) + return self._explicit_mapping + return schema.infer_bar_mapping(self._mapping_overrides) + + def _instrument(self, symbol: str) -> Instrument: + template = self._instrument_template + if template is not None: + if template.symbol != symbol: + raise MarketNotFoundError( + f"{self.name} is bound to {template.symbol!r}, not {symbol!r}" + ) + return replace( + template, + venue=template.venue or self.venue, + provider_id=template.provider_id or symbol, + ) + return Instrument(symbol=symbol, venue=self.venue, provider_id=symbol) + + def _symbols_sync(self, mapping: BarColumnMapping) -> tuple[str, ...]: + if mapping.symbol is None: + if self._instrument_template is None: + return () + return (self._instrument_template.symbol,) + values = self._distinct_values_sync(mapping.symbol) + symbols = { + str(value).strip() for value in values if value is not None and str(value).strip() + } + return tuple(sorted(symbols)) + + async def list_markets(self, query: MarketQuery | None = None) -> Sequence[MarketListing]: + mapping = await asyncio.to_thread(self._resolved_mapping_sync) + symbols = await asyncio.to_thread(self._symbols_sync, mapping) + listings = [MarketListing(self._instrument(symbol)) for symbol in symbols] + return listings if query is None else [item for item in listings if query.matches(item)] + + async def resolve_market(self, symbol: str) -> MarketListing: + if not symbol.strip(): + raise ValueError("symbol must not be empty") + mapping = await asyncio.to_thread(self._resolved_mapping_sync) + if mapping.symbol is not None: + symbols = await asyncio.to_thread(self._symbols_sync, mapping) + if symbol not in symbols: + raise MarketNotFoundError(f"{self.name} has no exact symbol {symbol!r}") + return MarketListing(self._instrument(symbol)) + + def _fetch_bars_sync(self, request: BarRequest) -> Sequence[Bar]: + if request.instrument.venue and request.instrument.venue != self.venue: + raise ValueError( + f"instrument venue {request.instrument.venue!r} does not match provider " + f"venue {self.venue!r}" + ) + if self._instrument_template is not None: + self._instrument(request.instrument.symbol) + if self._timeframe is not None and request.timeframe != self._timeframe: + raise ValueError( + f"source timeframe {self._timeframe!r} does not match request {request.timeframe!r}" + ) + + mapping = self._resolved_mapping_sync() + rows = self._read_rows_sync(mapping, request) + by_timestamp: dict[int, Bar] = {} + for index, row in enumerate(rows, start=1): + try: + if mapping.symbol is not None: + row_symbol = str(row[mapping.symbol]).strip() + if row_symbol != request.instrument.symbol: + continue + if mapping.timeframe is not None: + row_timeframe = str(row[mapping.timeframe]).strip() + if row_timeframe != request.timeframe: + continue + timestamp_ms = _timestamp_ms( + row[mapping.timestamp], self.timestamp_unit, self._timestamp_timezone + ) + if not request.start_ms <= timestamp_ms < request.end_ms: + continue + if timestamp_ms in by_timestamp: + raise TabularDataError( + f"duplicate bar timestamp {timestamp_ms} for " + f"{request.instrument.symbol!r} and {request.timeframe!r}" + ) + by_timestamp[timestamp_ms] = Bar( + instrument=request.instrument, + timestamp_ms=timestamp_ms, + open=_number(row[mapping.open], "open"), + high=_number(row[mapping.high], "high"), + low=_number(row[mapping.low], "low"), + close=_number(row[mapping.close], "close"), + volume=_number(row[mapping.volume], "volume"), + source=self.name, + ) + except TabularDataError: + raise + except (KeyError, TypeError, ValueError) as exc: + raise TabularDataError(f"cannot normalize source row {index}: {exc}") from exc + + bars = [by_timestamp[timestamp] for timestamp in sorted(by_timestamp)] + return bars if request.limit is None else bars[: request.limit] + + async def fetch_bars(self, request: BarRequest) -> Sequence[Bar]: + """Read, validate, sort, and normalize bars for one exact request.""" + + return await asyncio.to_thread(self._fetch_bars_sync, request) + + async def close(self) -> None: + await asyncio.to_thread(self._close_sync) + + @abstractmethod + def _inspect_schema_sync(self) -> TabularSchema: ... + + @abstractmethod + def _read_rows_sync( + self, mapping: BarColumnMapping, request: BarRequest + ) -> Sequence[TabularRow]: ... + + @abstractmethod + def _distinct_values_sync(self, column: str) -> Sequence[object]: ... + + def _close_sync(self) -> None: + return None diff --git a/tests/test_tabular_providers.py b/tests/test_tabular_providers.py new file mode 100644 index 0000000..c5f0369 --- /dev/null +++ b/tests/test_tabular_providers.py @@ -0,0 +1,312 @@ +from __future__ import annotations + +import asyncio +import os +import sqlite3 +from pathlib import Path + +import pytest + +from pineforge_data import ( + BarColumnMapping, + BarRequest, + CsvBarProvider, + Instrument, + MarketNotFoundError, + SchemaMappingError, + SqlAlchemyBarProvider, + SqliteBarProvider, + TabularDataError, + TimestampUnit, + create_provider, +) + + +def test_mapping_infers_common_names_and_accepts_partial_overrides() -> None: + mapping = BarColumnMapping.infer( + ("Bucket Start", "PX_OPEN", "high", "low", "close", "qty", "ticker", "interval"), + {"timestamp": "Bucket Start"}, + ) + + assert mapping.timestamp == "Bucket Start" + assert mapping.open == "PX_OPEN" + assert mapping.volume == "qty" + assert mapping.symbol == "ticker" + assert mapping.timeframe == "interval" + + +def test_mapping_rejects_ambiguous_or_missing_columns() -> None: + with pytest.raises(SchemaMappingError, match="ambiguous mappings for timestamp"): + BarColumnMapping.infer(("timestamp", "open_time", "open", "high", "low", "close", "volume")) + + with pytest.raises(SchemaMappingError, match="missing mappings for volume"): + BarColumnMapping.infer(("timestamp", "open", "high", "low", "close")) + + +def _write_csv(path: Path) -> None: + path.write_text( + "Bucket Start,PX_OPEN,PX_HIGH,PX_LOW,PX_CLOSE,Total Qty,Ticker,Bar Size\n" + "2025-07-01T00:01:00Z,11,13,10,12,6,AAPL,1m\n" + "2025-07-01T00:00:00Z,10,12,9,11,5,AAPL,1m\n" + "2025-07-01T00:00:00Z,20,22,19,21,8,MSFT,1m\n", + encoding="utf-8", + ) + + +def _arbitrary_mapping() -> dict[str, str]: + return { + "timestamp": "Bucket Start", + "open": "PX_OPEN", + "high": "PX_HIGH", + "low": "PX_LOW", + "close": "PX_CLOSE", + "volume": "Total Qty", + "symbol": "Ticker", + "timeframe": "Bar Size", + } + + +def test_csv_provider_inspects_arbitrary_columns_and_normalizes_rows(tmp_path: Path) -> None: + async def run() -> None: + path = tmp_path / "research bars.csv" + _write_csv(path) + provider = CsvBarProvider( + path, + venue="research", + mapping=_arbitrary_mapping(), + timestamp_unit="iso8601", + ) + + schema = await provider.inspect_schema() + assert schema.column_names[0] == "Bucket Start" + assert [item.instrument.symbol for item in await provider.list_markets()] == [ + "AAPL", + "MSFT", + ] + listing = await provider.resolve_market("AAPL") + bars = await provider.fetch_bars( + BarRequest( + listing.instrument, + "1m", + 1_751_328_000_000, + 1_751_328_120_000, + ) + ) + + assert [bar.timestamp_ms for bar in bars] == [ + 1_751_328_000_000, + 1_751_328_060_000, + ] + assert [bar.close for bar in bars] == [11.0, 12.0] + assert all(bar.source == "csv:research" for bar in bars) + with pytest.raises(MarketNotFoundError, match="no exact symbol"): + await provider.resolve_market("NVDA") + + asyncio.run(run()) + + +def test_csv_single_instrument_dataset_binds_symbol_at_resolution(tmp_path: Path) -> None: + async def run() -> None: + path = tmp_path / "single.csv" + path.write_text( + "timestamp,open,high,low,close,volume\n0,1,2,1,2,10\n", + encoding="utf-8", + ) + provider = CsvBarProvider(path, venue="research") + + assert await provider.list_markets() == [] + listing = await provider.resolve_market("PRIVATE_SERIES") + bars = await provider.fetch_bars(BarRequest(listing.instrument, "1m", 0, 60_000)) + + assert listing.instrument.symbol == "PRIVATE_SERIES" + assert len(bars) == 1 + + asyncio.run(run()) + + +def _write_sqlite(path: Path) -> None: + with sqlite3.connect(path) as connection: + connection.execute( + 'CREATE TABLE "price candles" (' + '"epoch seconds" INTEGER, "first px" REAL, "top px" REAL, ' + '"bottom px" REAL, "last px" REAL, "traded qty" REAL, ' + '"security code" TEXT, "bar interval" TEXT)' + ) + connection.executemany( + 'INSERT INTO "price candles" VALUES (?, ?, ?, ?, ?, ?, ?, ?)', + [ + (1_751_328_000, 10, 12, 9, 11, 5, "AAPL", "1m"), + (1_751_328_060, 11, 13, 10, 12, 6, "AAPL", "1m"), + (1_751_328_000, 20, 22, 19, 21, 8, "MSFT", "1m"), + ], + ) + + +def _sqlite_mapping() -> BarColumnMapping: + return BarColumnMapping( + timestamp="epoch seconds", + open="first px", + high="top px", + low="bottom px", + close="last px", + volume="traded qty", + symbol="security code", + timeframe="bar interval", + ) + + +def test_sqlite_provider_reflects_and_queries_arbitrary_ddl(tmp_path: Path) -> None: + async def run() -> None: + path = tmp_path / "market.sqlite3" + _write_sqlite(path) + provider = SqliteBarProvider( + path, + "price candles", + venue="warehouse", + mapping=_sqlite_mapping(), + timestamp_unit=TimestampUnit.SECONDS, + ) + + schema = await provider.inspect_schema() + assert schema.column_names == ( + "epoch seconds", + "first px", + "top px", + "bottom px", + "last px", + "traded qty", + "security code", + "bar interval", + ) + listing = await provider.resolve_market("AAPL") + bars = await provider.fetch_bars( + BarRequest( + listing.instrument, + "1m", + 1_751_328_030_000, + 1_751_328_120_000, + ) + ) + assert [bar.timestamp_ms for bar in bars] == [1_751_328_060_000] + + asyncio.run(run()) + + +def test_sqlalchemy_provider_uses_reflection_and_bound_filters(tmp_path: Path) -> None: + pytest.importorskip("sqlalchemy") + + async def run() -> None: + path = tmp_path / "sqlalchemy.sqlite3" + _write_sqlite(path) + provider = SqlAlchemyBarProvider( + f"sqlite:///{path}", + "price candles", + venue="analytics", + mapping=_sqlite_mapping(), + timestamp_unit="seconds", + ) + try: + schema = await provider.inspect_schema() + assert "first px" in schema.column_names + listing = await provider.resolve_market("MSFT") + bars = await provider.fetch_bars( + BarRequest( + listing.instrument, + "1m", + 1_751_328_000_000, + 1_751_328_060_000, + ) + ) + assert len(bars) == 1 + assert bars[0].close == 21.0 + assert bars[0].source == "sqlalchemy:analytics" + finally: + await provider.close() + + asyncio.run(run()) + + +def test_registry_builds_csv_provider_from_runtime_json_shape(tmp_path: Path) -> None: + path = tmp_path / "bars.csv" + _write_csv(path) + + provider = create_provider( + "csv", + "research", + config={ + "path": str(path), + "timestamp_unit": "iso8601", + "columns": _arbitrary_mapping(), + }, + ) + + assert isinstance(provider, CsvBarProvider) + assert provider.name == "csv:research" + + +def test_registry_reads_sqlalchemy_url_from_environment( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + path = tmp_path / "database.sqlite3" + _write_sqlite(path) + monkeypatch.setenv("TEST_PINEFORGE_DATABASE_URL", f"sqlite:///{path}") + + provider = create_provider( + "sqlalchemy", + "analytics", + config={ + "url_env": "TEST_PINEFORGE_DATABASE_URL", + "table": "price candles", + "timestamp_unit": "seconds", + "columns": { + "timestamp": "epoch seconds", + "open": "first px", + "high": "top px", + "low": "bottom px", + "close": "last px", + "volume": "traded qty", + "symbol": "security code", + "timeframe": "bar interval", + }, + }, + ) + + assert isinstance(provider, SqlAlchemyBarProvider) + assert "TEST_PINEFORGE_DATABASE_URL" in os.environ + asyncio.run(provider.close()) + + +def test_duplicate_timestamp_is_rejected_instead_of_silently_overwritten(tmp_path: Path) -> None: + async def run() -> None: + path = tmp_path / "duplicates.csv" + path.write_text( + "timestamp,open,high,low,close,volume\n0,1,2,1,2,10\n0,2,3,1,2,20\n", + encoding="utf-8", + ) + provider = CsvBarProvider(path) + listing = await provider.resolve_market("TEST") + + with pytest.raises(TabularDataError, match="duplicate bar timestamp"): + await provider.fetch_bars(BarRequest(listing.instrument, "1m", 0, 60_000)) + + asyncio.run(run()) + + +def test_fixed_timeframe_is_validated_when_no_timeframe_column(tmp_path: Path) -> None: + async def run() -> None: + path = tmp_path / "bars.csv" + path.write_text( + "timestamp,open,high,low,close,volume\n0,1,2,1,2,10\n", + encoding="utf-8", + ) + provider = CsvBarProvider( + path, + instrument=Instrument("TEST"), + timeframe="1m", + ) + listing = await provider.resolve_market("TEST") + + with pytest.raises(ValueError, match="source timeframe"): + await provider.fetch_bars(BarRequest(listing.instrument, "5m", 0, 60_000)) + + asyncio.run(run())