data-store

Canonical market data archive library. The only interface through which the filesystem may be modified.

PyPI package: tt-data-store · GitHub: Tiny-Trader/data-store

Parquet is the source of truth. metadata.db (SQLite) catalogues instruments, files, ingestions, and validation — it is not a substitute for the archive.

Responsibilities

• Push new data into the FS and maintain metadata
  • Normalize
  • Validate
  • Store
• Get a slice of data from the FS

Callers own acquisition formats (e.g. CSV). Ingress is candles.write.

Mental model

Your script / collector
        ↓
  store.candles.write(...)
        ↓
normalize → validate → merge → atomic Parquet write → update metadata.db

Acquisition (CSV download, API, manual script) is the caller's job. The store only accepts normalized candle data via candles.write.

On disk, an archive looks like:

data/
├── market/          # Parquet files (human-readable layout)
├── reference/       # calendars, etc.
└── metadata.db      # catalogue

Canonical layout and semantics live in docs/.

Install

pip install tt-data-store
# or from source:
uv pip install -e .

from tt_data_store import MarketStore

Opening an archive

from tt_data_store import MarketStore

store = MarketStore("/path/to/data")

Optional S3 config (or env vars MARKETSTORE_S3_BUCKET, MARKETSTORE_S3_PREFIX, MARKETSTORE_S3_REGION):

from tt_data_store import MarketStore, RemoteConfig

store = MarketStore(
    "./data",
    remote=RemoteConfig(bucket="my-bucket", prefix="archive", region="ap-south-1"),
)

Usage

Register instruments

# NIFTY index
spot = store.instruments.create(
    exchange="NSE",
    instrument_type="INDEX",
    symbol="NIFTY",
)
# → instrument_key: "NSE:INDEX:NIFTY"

# NIFTY option
opt = store.instruments.create(
    exchange="NSE",
    instrument_type="OPTION",
    symbol="NIFTY",
    underlying_id=spot.id,
    expiry_date="2026-08-27",
    strike=25000,
    option_type="CE",
)
# → "NSE:OPTION:NIFTY:2026-08-27:25000:CE"

Look up later by key:

inst = store.instruments.get("NSE:INDEX:NIFTY")
all_nifty = store.instruments.list(symbol="NIFTY")

Write candles (main ingress)

Pass a list of dicts or a PyArrow table. Each candle has:

timestamp, open, high, low, close, volume (optional), open_interest (optional)

from datetime import datetime
from zoneinfo import ZoneInfo

IST = ZoneInfo("Asia/Kolkata")

rows = [
    {
        "timestamp": datetime(2026, 8, 11, 9, 15, tzinfo=IST),
        "open": 100.0,
        "high": 101.0,
        "low": 99.0,
        "close": 100.5,
        "volume": 10,
        "open_interest": 100,
    },
]

result = store.candles.write(spot, rows, source="upstox")

write handles normalization, dedup, merge with existing data, gap detection, atomic file replacement, and ingestion provenance. It returns a WriteResult with row counts, gaps, quality status (VALID / PARTIAL / etc.), and the file path.

Files land in deterministic locations — e.g. spot → market/nifty/spot/2026.parquet, options → market/nifty/options/2026-08-27/25000_CE.parquet.

Read candles

table = store.candles.read(
    spot,
    start="2026-08-01",
    end="2026-08-31",
)

Returns a PyArrow table, filtered and sorted by timestamp. The store resolves which Parquet files to read.

Inspect files and quality

file = store.files.get("market/nifty/spot/2026.parquet")

result = store.validate(spot)          # or store.validate(file.path)

report = store.inspect()

Sync with S3

Local writes go to disk first; S3 is separate:

store.remote.push(files=["market/nifty/spot/2026.parquet", "metadata.db"])
store.remote.pull(files=["market/nifty/options/2026-08-27/25000_CE.parquet"])
store.remote.sync(files=[...])

End-to-end producer flow

store = MarketStore("./data")

# 1. Ensure instrument exists
inst = store.instruments.get("NSE:INDEX:NIFTY")
if inst is None:
    inst = store.instruments.create(
        exchange="NSE", instrument_type="INDEX", symbol="NIFTY"
    )

# 2. Fetch from Upstox/NSE/CSV — convert to candle dicts yourself
candles = parse_my_csv("NIFTY.csv")

# 3. Single write call — store does the rest
result = store.candles.write(
    inst,
    candles,
    source="upstox",
    source_instrument_id="NSE_INDEX|Nifty 50",
    requested_start="2026-08-01",
    requested_end="2026-08-31",
)

# 4. Optionally push to S3
store.remote.push()

What not to do

• Don't write Parquet files directly
• Don't run SQL against metadata.db in normal workflows
• Don't parse CSV inside tt_data_store — convert to candles first
• Don't use paths as the primary interface (use instrument keys and date ranges)

Public API

store.instruments.get(...) / .list(...) / .create(...)
store.candles.read(...) / .write(...)
store.files.get(...) / .list(...)
store.validate(...)
store.inspect()
store.remote.pull(...) / .push(...) / .sync(...)

Repository layout

data-store/
├── docs/
├── pyproject.toml
├── src/            # tt_data_store package modules
└── tests/