quaq-ingest

name

quaq-ingest

description

Use this skill when the user wants to import, convert, or prepare data from any source for use with quaq backtesting. Triggers include: downloading data from Yahoo Finance, converting TradingView exports, getting data from a custom API, preparing OHLCV data, importing CSV files, converting data formats, "how do I get data into quaq", custom data source integration, or any data preparation task for the quaq engine.

Data Ingestion for Quaq

Target Format: quaq CSV

The quaq engine reads OHLCV data as CSV with this exact format:

Header (case-sensitive):

timestamp,open,high,low,close,volume

Rules:

• timestamp: millisecond epoch integers (Unix time * 1000)
• One row per bar, ascending timestamp order, no duplicates
• All values numeric -- no NaN, null, empty, or string values
• Volume must be non-negative

Example rows:

timestamp,open,high,low,close,volume
1704067200000,42000.50,42100.00,41900.00,42050.00,1234.56
1704070800000,42050.00,42200.00,42000.00,42150.00,987.65

Place data files in core/data/ and reference them in your strategy TOML:

[data]
file = "core/data/BTCUSDT_1h.csv"

---

Binance (Recommended)

Binance is the primary supported data source with two ingestion paths.

Option A: Python Parquet fetcher (preferred for bulk data)

python scripts/fetch_parquet.py --symbol BTCUSDT --interval 1h --start 2024-01-01 --end 2024-12-31

Output: core/data/BTCUSDT_1h.parquet -- ready to use directly.

Option B: Live fetch via TOML

Set dataset = "binance" in your strategy TOML for auto-fetching:

[data]
dataset = "binance"
symbol = "BTCUSDT"
timeframe = "1h"
market = "spot"
start_date = "2024-01-01"
end_date = "2024-12-31"

The engine fetches klines from the Binance REST API, caches as CSV in core/data/, and loads automatically. See the quaq-data skill for full details.

---

Yahoo Finance

Use the yfinance Python library to download and convert data.

import yfinance as yf
import pandas as pd

# Download hourly data
ticker = yf.Ticker("BTC-USD")
df = ticker.history(start="2024-01-01", end="2024-12-31", interval="1h")

# Convert to quaq CSV format
out = pd.DataFrame({
    'timestamp': (df.index.astype('int64') // 10**6).astype(int),  # ns -> ms
    'open': df['Open'],
    'high': df['High'],
    'low': df['Low'],
    'close': df['Close'],
    'volume': df['Volume']
})
out.to_csv('core/data/BTCUSD_1h.csv', index=False)

Notes:

• yfinance returns timestamps as pandas DatetimeIndex in nanoseconds
• Divide by 10**6 to convert ns -> ms epoch
• Column names are capitalized in yfinance -- map to lowercase

---

TradingView Export

TradingView CSV export format:

time,open,high,low,close,Volume
1704067200,42000.5,42100,41900,42050,1234.56

Differences from quaq format:

• time is Unix seconds (10 digits) -- multiply by 1000 for ms
• Volume is capitalized -- rename to volume

Python conversion:

import pandas as pd

df = pd.read_csv('tradingview_export.csv')
df['timestamp'] = (df['time'] * 1000).astype(int)
df = df.rename(columns={'Volume': 'volume'})
df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']]
df.to_csv('core/data/converted.csv', index=False)

One-liner (if columns are already correct except timestamp):

python -c "
import pandas as pd
df = pd.read_csv('tv_export.csv')
df['timestamp'] = (df['time'] * 1000).astype(int)
df.rename(columns={'Volume':'volume'})[['timestamp','open','high','low','close','volume']].to_csv('core/data/out.csv', index=False)
"

---

Pandas DataFrame -> quaq CSV

Generic pattern for any DataFrame with a datetime index:

import pandas as pd

# Assuming `df` has a DatetimeIndex and OHLCV columns
out = pd.DataFrame({
    'timestamp': (df.index.astype('int64') // 10**6).astype(int),
    'open': df['open'],    # adjust column names as needed
    'high': df['high'],
    'low': df['low'],
    'close': df['close'],
    'volume': df['volume']
})

# Sort by timestamp ascending, drop duplicates
out = out.sort_values('timestamp').drop_duplicates(subset='timestamp')
out.to_csv('core/data/output.csv', index=False)

If the DataFrame has a Unix timestamp column instead of DatetimeIndex:

# Detect timestamp magnitude and normalize to milliseconds
sample_ts = df['timestamp'].iloc[0]
if sample_ts < 1e12:        # seconds (10 digits)
    df['timestamp'] = (df['timestamp'] * 1000).astype(int)
elif sample_ts < 1e15:      # milliseconds (13 digits) -- already correct
    df['timestamp'] = df['timestamp'].astype(int)
elif sample_ts < 1e18:      # microseconds (16 digits)
    df['timestamp'] = (df['timestamp'] // 1000).astype(int)
else:                        # nanoseconds (19 digits)
    df['timestamp'] = (df['timestamp'] // 1_000_000).astype(int)

---

Custom API -> quaq CSV

Generic template for fetching from a REST API:

import requests
import pandas as pd
import json

url = "https://api.example.com/ohlcv"
params = {
    "symbol": "BTCUSDT",
    "interval": "1h",
    "start": "2024-01-01",
    "end": "2024-12-31"
}

response = requests.get(url, params=params)
data = response.json()

# Adapt this based on your API's response structure
# Common patterns:
# - Array of arrays: [[ts, o, h, l, c, v], ...]
# - Array of objects: [{"time": ..., "open": ..., ...}, ...]

df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

# Normalize timestamps to milliseconds
# (adjust based on your API's timestamp format)
df['timestamp'] = df['timestamp'].astype(int)

df = df.sort_values('timestamp')
df.to_csv('core/data/custom_data.csv', index=False)

---

Parquet Compatibility Warning

The quaq Parquet reader has strict requirements:

• Encoding: PLAIN only (no DELTA_BINARY_PACKED, RLE_DICTIONARY, etc.)
• Compression: UNCOMPRESSED only (no SNAPPY, GZIP, ZSTD)
• Repetition: REQUIRED fields only (no OPTIONAL/nullable columns)

Do NOT use pandas.to_parquet() or pyarrow directly -- they default to OPTIONAL fields, SNAPPY compression, and dictionary encoding, which are all incompatible.

Safe options:

1. Use scripts/fetch_parquet.py as your Parquet writer (it uses a custom Thrift-based writer)
2. Convert to CSV instead -- CSV always works and there is no performance difference for typical dataset sizes

If you must write Parquet programmatically, use the script's writer as a reference implementation.

---

Timestamp Conversion Reference

Source format	Digits	Conversion to ms epoch
Unix seconds	10	ts * 1000
Milliseconds	13	Use as-is
Microseconds	16	ts // 1000
Nanoseconds	19	ts // 1_000_000
pandas Timestamp	--	.value // 10**6
pandas DatetimeIndex	--	.astype('int64') // 10**6
Python datetime	--	int(dt.timestamp() * 1000)
ISO 8601 string	--	int(pd.Timestamp(s).timestamp() * 1000)

---

Validation Checklist

Before loading data into quaq, verify:

1. Header: exactly timestamp,open,high,low,close,volume (case-sensitive, no extra columns)
2. Timestamps: ascending order, no duplicates, millisecond epoch integers
3. No gaps: gaps should be consistent with the expected timeframe (e.g., 3600000ms = 1h)
4. All numeric: no NaN, null, empty strings, or text values
5. Volume: non-negative floats
6. Reasonable values: open/high/low/close should be positive; high >= low; high >= open,close; low <= open,close

Quick sanity check:

# Check header
head -n 1 core/data/mydata.csv
# Expected: timestamp,open,high,low,close,volume

# Check first few rows
head -n 5 core/data/mydata.csv

# Check row count
wc -l core/data/mydata.csv

# Check timestamp range (first and last)
head -n 2 core/data/mydata.csv | tail -1 | cut -d',' -f1
tail -n 1 core/data/mydata.csv | cut -d',' -f1

# Validate with quaq (catches format errors)
cd core && zig build run -- validate my_strategy.toml

Validation errors you might see:

• E023: Missing symbol when using dataset = "binance"
• E024: Missing timeframe when using dataset = "binance"
• E025: Invalid date format (must be YYYY-MM-DD)
• File not found: Check the file path is relative to the working directory