From 5b55da14ce00638043443f8c75ae3bab90a186b3 Mon Sep 17 00:00:00 2001 From: javier Date: Fri, 23 Jan 2026 19:02:30 +0100 Subject: [PATCH 1/5] pytorch example --- 2026-01-pytorch/.DS_Store | Bin 0 -> 6148 bytes 2026-01-pytorch/README.md | 85 +++++++++++ 2026-01-pytorch/config/__init__.py | 0 2026-01-pytorch/config/settings.py | 23 +++ 2026-01-pytorch/pyproject.toml | 24 +++ 2026-01-pytorch/src/__init__.py | 0 2026-01-pytorch/src/data_loader.py | 141 ++++++++++++++++++ 2026-01-pytorch/src/features.py | 69 +++++++++ 2026-01-pytorch/src/model.py | 133 +++++++++++++++++ 2026-01-pytorch/steps/__init__.py | 0 2026-01-pytorch/steps/method1_direct_query.py | 53 +++++++ .../steps/method2_parquet_export.py | 57 +++++++ .../steps/method3_parquet_partitions.py | 89 +++++++++++ 2026-01-pytorch/steps/train.py | 136 +++++++++++++++++ 14 files changed, 810 insertions(+) create mode 100644 2026-01-pytorch/.DS_Store create mode 100644 2026-01-pytorch/README.md create mode 100644 2026-01-pytorch/config/__init__.py create mode 100644 2026-01-pytorch/config/settings.py create mode 100644 2026-01-pytorch/pyproject.toml create mode 100644 2026-01-pytorch/src/__init__.py create mode 100644 2026-01-pytorch/src/data_loader.py create mode 100644 2026-01-pytorch/src/features.py create mode 100644 2026-01-pytorch/src/model.py create mode 100644 2026-01-pytorch/steps/__init__.py create mode 100644 2026-01-pytorch/steps/method1_direct_query.py create mode 100644 2026-01-pytorch/steps/method2_parquet_export.py create mode 100644 2026-01-pytorch/steps/method3_parquet_partitions.py create mode 100644 2026-01-pytorch/steps/train.py diff --git a/2026-01-pytorch/.DS_Store b/2026-01-pytorch/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..9b8c37044d898b2f7a42959e9d9bbac0d175978f GIT binary patch literal 6148 zcmeHKyH3ME5S)b+ktj%+@_vAVKR899NX-u*Cze36a0sMG=Z?>Y*#`*Aq@_T+((c@? zckbdTyjcLUI__?P1%NqS5$_tNrq9(Ub`cpN(s{-KTRh?hFE~!Be@`fPi9JquLqFoT zcx7&fhUe|^L~qd%pV4l*e%o|?L$3btYxR#XuG3P1R3H^d1yX@j;JgCbduj74$Bd~! zDv%0%E1=(pLRYMTgQI;q7;FR}&X_ji+GYu2(FCyu4vx&w#HmE5N{krdbmmLc)xg2g z=@7e_=fuqt6N=dF%oi(%RL6{|Kq@d+;L^7X?f-Z5GxPtLl)F?Q75G;Q$ZUPTUh$Qp zw@zM8du^dV(7%keR?c9pm}sq-8*RnQA9Y37d|m?wN24=ubYlJyP+ihefxl4T6SJEl AGynhq literal 0 HcmV?d00001 diff --git a/2026-01-pytorch/README.md b/2026-01-pytorch/README.md new file mode 100644 index 0000000..fc503b5 --- /dev/null +++ b/2026-01-pytorch/README.md @@ -0,0 +1,85 @@ +# PyTorch + QuestDB Data Pipeline Tutorial + +This tutorial demonstrates three ways to load time-series data from QuestDB into PyTorch: + +1. **Direct SQL queries** (REST API or ConnectorX) +2. **Parquet export** via REST API +3. **Direct Parquet partition access** (for local QuestDB) + +The example task is volume forecasting, but the data loading patterns apply to any model. + +## Quick Start + +```bash +# Install dependencies +pip install pandas pyarrow requests torch scikit-learn + +# Or with uv: +uv sync + +# Run the demos in order: +python steps/method1_direct_query.py # See direct query methods +python steps/method2_parquet_export.py # Export data to Parquet +python steps/train.py # Train the model +``` + +## What Each Script Does + +| Script | Description | +|--------|-------------| +| `steps/method1_direct_query.py` | Queries QuestDB demo instance, shows REST API and ConnectorX | +| `steps/method2_parquet_export.py` | Exports 7 days of data to `dataset/training_data.parquet` | +| `steps/method3_parquet_partitions.py` | Shows direct partition access (requires local QuestDB) | +| `steps/train.py` | Loads Parquet, engineers features, trains LSTM, saves model | + +## Project Structure + +``` +├── config/ +│ └── settings.py # Configuration (URLs, paths, hyperparameters) +├── src/ +│ ├── data_loader.py # Three data loading methods +│ ├── features.py # Feature engineering +│ └── model.py # PyTorch Dataset and LSTM model +├── steps/ +│ ├── method1_direct_query.py +│ ├── method2_parquet_export.py +│ ├── method3_parquet_partitions.py +│ └── train.py +├── dataset/ # Exported Parquet files +├── models/ # Saved model checkpoints +└── pyproject.toml +``` + +## Data Source + +Uses the `trades` table from QuestDB's public demo: `https://demo.questdb.io` + +The query aggregates raw trades into 1-minute bars: + +```sql +SELECT + timestamp, + symbol, + sum(amount) as volume, + count() as trade_count, + avg(price) as avg_price, + max(price) - min(price) as price_range +FROM trades +WHERE symbol = 'BTC-USD' +SAMPLE BY 1m +ALIGN TO CALENDAR +``` + +## Method Comparison + +| Method | Best For | Requires | +|--------|----------|----------| +| REST API | Exploration, simple queries | Network access | +| ConnectorX | Larger result sets | `pip install connectorx` | +| Parquet Export | Reproducible training data | Network access | +| Direct Partitions | Terabyte-scale data | Local QuestDB with Parquet format | + +## Note on the Model + +The LSTM model is intentionally simple—this tutorial focuses on data pipeline patterns, not forecasting accuracy. In production, you'd benchmark against simpler methods (moving averages, ARIMA) and tune hyperparameters properly. diff --git a/2026-01-pytorch/config/__init__.py b/2026-01-pytorch/config/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/2026-01-pytorch/config/settings.py b/2026-01-pytorch/config/settings.py new file mode 100644 index 0000000..72c3afa --- /dev/null +++ b/2026-01-pytorch/config/settings.py @@ -0,0 +1,23 @@ +from pathlib import Path + +# QuestDB connection +QUESTDB_HTTP = "https://demo.questdb.io" +QUESTDB_PG = "postgresql://admin:quest@demo.questdb.io:8812/qdb" + +# For local QuestDB with Parquet partitions (Method 3) +QUESTDB_DATA_DIR = Path("/var/lib/questdb/db") + +# Paths +PROJECT_ROOT = Path(__file__).parent.parent +DATASET_DIR = PROJECT_ROOT / "dataset" +MODEL_DIR = PROJECT_ROOT / "models" + +# Data settings +TABLE_NAME = "trades" +SYMBOL = "BTC-USDT" + +# Model settings +SEQUENCE_LENGTH = 30 +BATCH_SIZE = 32 +EPOCHS = 20 +LEARNING_RATE = 0.001 diff --git a/2026-01-pytorch/pyproject.toml b/2026-01-pytorch/pyproject.toml new file mode 100644 index 0000000..dbf4b3f --- /dev/null +++ b/2026-01-pytorch/pyproject.toml @@ -0,0 +1,24 @@ +[project] +name = "pytorch-questdb-pipeline" +version = "0.1.0" +description = "Tutorial: Loading QuestDB time-series data into PyTorch" +requires-python = ">=3.11" +dependencies = [ + "pandas>=2.0.0", + "pyarrow>=15.0.0", + "requests>=2.31.0", + "torch>=2.0.0", + "scikit-learn>=1.4.0", +] + +[project.optional-dependencies] +fast = [ + "connectorx>=0.3.0", # Faster queries via PostgreSQL protocol +] + +[tool.poe.tasks] +method1 = { cmd = "python steps/method1_direct_query.py", help = "Demo: Direct SQL queries" } +method2 = { cmd = "python steps/method2_parquet_export.py", help = "Demo: Parquet export via API" } +method3 = { cmd = "python steps/method3_parquet_partitions.py", help = "Demo: Direct partition access" } +train = { cmd = "python steps/train.py", help = "Train the LSTM model" } +all = ["method1", "method2", "train"] diff --git a/2026-01-pytorch/src/__init__.py b/2026-01-pytorch/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/2026-01-pytorch/src/data_loader.py b/2026-01-pytorch/src/data_loader.py new file mode 100644 index 0000000..c7d428e --- /dev/null +++ b/2026-01-pytorch/src/data_loader.py @@ -0,0 +1,141 @@ +""" +Three methods to load data from QuestDB into Python/PyTorch. +""" +import pandas as pd +import requests +from io import StringIO +from pathlib import Path + +from config.settings import QUESTDB_HTTP, QUESTDB_PG + + +# ============================================================================= +# Method 1: Direct Queries +# ============================================================================= + +def query_rest_api(query: str) -> pd.DataFrame: + """ + Query QuestDB via REST API, return DataFrame. + + Simple and works everywhere. Best for exploration and smaller datasets. + """ + response = requests.get( + f"{QUESTDB_HTTP}/exp", + params={"query": query, "fmt": "csv"} + ) + response.raise_for_status() + return pd.read_csv(StringIO(response.text)) + + +def query_connectorx(query: str) -> pd.DataFrame: + """ + Query QuestDB via PostgreSQL wire protocol using ConnectorX. + + Faster than REST for larger results. Requires: pip install connectorx + """ + import connectorx as cx + return cx.read_sql(QUESTDB_PG, query) + + +# ============================================================================= +# Method 2: Parquet Export via REST API +# ============================================================================= + +def export_to_parquet(query: str, output_path: Path) -> Path: + """ + Export query results to Parquet file via QuestDB REST API. + + Falls back to CSV if Parquet export is disabled on the server. + Best for reproducible training datasets. + """ + output_path.parent.mkdir(parents=True, exist_ok=True) + + # Try native Parquet export + response = requests.get( + f"{QUESTDB_HTTP}/exp", + params={"query": query, "fmt": "parquet"}, + stream=True + ) + + content_type = response.headers.get("content-type", "") + + if response.ok and "parquet" in content_type: + with open(output_path, "wb") as f: + for chunk in response.iter_content(chunk_size=8192): + f.write(chunk) + print(f"✓ Exported native Parquet: {output_path}") + else: + # Fallback: CSV export + pandas conversion + print(" Parquet export not available, using CSV fallback...") + csv_response = requests.get( + f"{QUESTDB_HTTP}/exp", + params={"query": query, "fmt": "csv"} + ) + csv_response.raise_for_status() + df = pd.read_csv(StringIO(csv_response.text)) + df.to_parquet(output_path, index=False) + print(f"✓ Exported via CSV→Parquet: {output_path}") + + return output_path + + +# ============================================================================= +# Method 3: Direct Parquet Partition Access +# ============================================================================= + +def load_parquet_partitions(table_name: str, data_dir: Path): + """ + Load QuestDB Parquet partitions directly via PyArrow. + + Requires: + - Local/mounted access to QuestDB data directory + - Parquet partition format enabled (o3PartitionFormat = 'PARQUET') + + Best for terabyte-scale data and streaming processing. + """ + import pyarrow.dataset as ds + + table_dir = data_dir / table_name + parquet_files = list(table_dir.glob("**/*.parquet")) + + if not parquet_files: + raise FileNotFoundError( + f"No Parquet files found in {table_dir}. " + "Ensure Parquet partition format is enabled in QuestDB." + ) + + print(f"✓ Found {len(parquet_files)} partition files") + return ds.dataset(table_dir, format="parquet") + + +def stream_from_partitions(dataset, columns: list[str], batch_size: int = 100_000): + """ + Stream data from Parquet partitions in batches. + + Memory-efficient for large datasets. + """ + scanner = dataset.scanner(columns=columns, batch_size=batch_size) + for batch in scanner.to_batches(): + yield batch.to_pandas() + + +# ============================================================================= +# Query builders +# ============================================================================= + +def get_training_query(symbol: str = "BTC-USDT", days: int = 7) -> str: + """Build query for training data: 1-minute aggregated bars.""" + return f""" + SELECT + timestamp, + symbol, + sum(amount) as volume, + count() as trade_count, + avg(price) as avg_price, + max(price) - min(price) as price_range + FROM trades + WHERE symbol = '{symbol}' + AND timestamp > dateadd('d', -{days}, now()) + SAMPLE BY 1m + ALIGN TO CALENDAR + """ diff --git a/2026-01-pytorch/src/features.py b/2026-01-pytorch/src/features.py new file mode 100644 index 0000000..45a465c --- /dev/null +++ b/2026-01-pytorch/src/features.py @@ -0,0 +1,69 @@ +""" +Feature engineering for time-series forecasting. +""" +import pandas as pd +import numpy as np +from sklearn.preprocessing import StandardScaler + + +FEATURE_COLS = [ + "volume_lag_1", + "volume_lag_5", + "volume_lag_15", + "volume_ma_15", + "volume_std_15", + "trade_count", + "price_range" +] + + +def prepare_features(df: pd.DataFrame) -> pd.DataFrame: + """ + Add features for volume forecasting. + + Returns DataFrame with features and target column. + """ + df = df.sort_values("timestamp").copy() + + # Ensure numeric types + df["volume"] = pd.to_numeric(df["volume"], errors="coerce") + df["trade_count"] = pd.to_numeric(df["trade_count"], errors="coerce") + df["price_range"] = pd.to_numeric(df["price_range"], errors="coerce") + + # Lag features + for lag in [1, 5, 15]: + df[f"volume_lag_{lag}"] = df["volume"].shift(lag) + + # Rolling statistics + df["volume_ma_15"] = df["volume"].rolling(15).mean() + df["volume_std_15"] = df["volume"].rolling(15).std() + + # Target: next period's volume + df["target"] = df["volume"].shift(-1) + + # Drop rows with NaN (from lags/rolling) + df = df.dropna() + + return df + + +def normalize_features( + train_df: pd.DataFrame, + test_df: pd.DataFrame, + feature_cols: list[str] = FEATURE_COLS +) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, StandardScaler]: + """ + Normalize features using StandardScaler. + + Fits on train, transforms both train and test. + Returns: X_train, X_test, y_train, y_test, scaler + """ + scaler = StandardScaler() + + X_train = scaler.fit_transform(train_df[feature_cols]) + X_test = scaler.transform(test_df[feature_cols]) + + y_train = train_df["target"].values + y_test = test_df["target"].values + + return X_train, X_test, y_train, y_test, scaler diff --git a/2026-01-pytorch/src/model.py b/2026-01-pytorch/src/model.py new file mode 100644 index 0000000..01fade5 --- /dev/null +++ b/2026-01-pytorch/src/model.py @@ -0,0 +1,133 @@ +""" +PyTorch model and dataset classes for time-series forecasting. +""" +import torch +import torch.nn as nn +from torch.utils.data import Dataset, IterableDataset +import numpy as np + + +class TimeSeriesDataset(Dataset): + """ + PyTorch Dataset for sequential time-series data. + + Creates overlapping sequences of `sequence_length` to predict a target value. + Use when data fits in memory. + """ + + def __init__( + self, + features: np.ndarray, + targets: np.ndarray, + sequence_length: int = 30 + ): + self.features = torch.tensor(features, dtype=torch.float32) + self.targets = torch.tensor(targets, dtype=torch.float32) + self.sequence_length = sequence_length + + def __len__(self): + return len(self.features) - self.sequence_length + + def __getitem__(self, idx): + X = self.features[idx:idx + self.sequence_length] + y = self.targets[idx + self.sequence_length - 1] # Target at end of sequence + return X, y + + +class StreamingTimeSeriesDataset(IterableDataset): + """ + PyTorch IterableDataset that streams from Parquet. + + Use when data doesn't fit in memory. + Requires a generator function that yields DataFrames. + """ + + def __init__( + self, + batch_generator, # Generator yielding DataFrames + feature_cols: list[str], + target_col: str = "target", + sequence_length: int = 30 + ): + self.batch_generator = batch_generator + self.feature_cols = feature_cols + self.target_col = target_col + self.sequence_length = sequence_length + + def __iter__(self): + for batch_df in self.batch_generator: + features = batch_df[self.feature_cols].values + targets = batch_df[self.target_col].values + + for i in range(len(features) - self.sequence_length): + X = features[i:i + self.sequence_length] + y = targets[i + self.sequence_length - 1] + + yield ( + torch.tensor(X, dtype=torch.float32), + torch.tensor(y, dtype=torch.float32) + ) + + +class SimpleLSTM(nn.Module): + """ + Basic LSTM for sequence-to-value regression. + + This is intentionally simple—a demonstration of the data pipeline, + not a production forecasting model. + """ + + def __init__(self, input_size: int, hidden_size: int = 32, num_layers: int = 1): + super().__init__() + self.lstm = nn.LSTM( + input_size=input_size, + hidden_size=hidden_size, + num_layers=num_layers, + batch_first=True + ) + self.fc = nn.Linear(hidden_size, 1) + + def forward(self, x): + # x shape: (batch, sequence_length, features) + lstm_out, (hidden, _) = self.lstm(x) + # Use final hidden state + out = self.fc(hidden[-1]) + return out.squeeze(-1) + + +def train_epoch(model, dataloader, criterion, optimizer, device): + """Train for one epoch, return average loss.""" + model.train() + total_loss = 0 + n_batches = 0 + + for X, y in dataloader: + X, y = X.to(device), y.to(device) + + optimizer.zero_grad() + pred = model(X) + loss = criterion(pred, y) + loss.backward() + optimizer.step() + + total_loss += loss.item() + n_batches += 1 + + return total_loss / n_batches if n_batches > 0 else 0 + + +def evaluate(model, dataloader, criterion, device): + """Evaluate model, return average loss.""" + model.eval() + total_loss = 0 + n_batches = 0 + + with torch.no_grad(): + for X, y in dataloader: + X, y = X.to(device), y.to(device) + pred = model(X) + loss = criterion(pred, y) + total_loss += loss.item() + n_batches += 1 + + return total_loss / n_batches if n_batches > 0 else 0 diff --git a/2026-01-pytorch/steps/__init__.py b/2026-01-pytorch/steps/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/2026-01-pytorch/steps/method1_direct_query.py b/2026-01-pytorch/steps/method1_direct_query.py new file mode 100644 index 0000000..f961dc3 --- /dev/null +++ b/2026-01-pytorch/steps/method1_direct_query.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +""" +Method 1: Direct SQL Queries + +Demonstrates querying QuestDB directly via REST API and ConnectorX. +Best for exploration and smaller datasets. +""" +import sys +sys.path.insert(0, ".") + +from src.data_loader import query_rest_api, query_connectorx, get_training_query +from config.settings import SYMBOL + + +def main(): + query = get_training_query(symbol=SYMBOL, days=1) # Just 1 day for demo + + print("=" * 60) + print("METHOD 1: Direct SQL Queries") + print("=" * 60) + + # Method 1a: REST API + print("\n1a. Querying via REST API...") + print(f" Query: SELECT ... FROM trades WHERE symbol='{SYMBOL}' SAMPLE BY 1m") + + df_rest = query_rest_api(query) + print(f" ✓ Loaded {len(df_rest)} rows") + print(f" Columns: {list(df_rest.columns)}") + print(f"\n First 5 rows:") + print(df_rest.head().to_string(index=False)) + + # Method 1b: ConnectorX (faster for large results) + print("\n" + "-" * 60) + print("\n1b. Querying via ConnectorX (PostgreSQL wire protocol)...") + + try: + df_cx = query_connectorx(query) + print(f" ✓ Loaded {len(df_cx)} rows") + except ImportError: + print(" ⚠ ConnectorX not installed. Run: pip install connectorx") + except Exception as e: + print(f" ⚠ ConnectorX error: {e}") + + print("\n" + "=" * 60) + print("Summary:") + print(" - REST API works everywhere, simple to use") + print(" - ConnectorX is faster for larger result sets") + print(" - Both return pandas DataFrames ready for analysis") + print("=" * 60) + + +if __name__ == "__main__": + main() diff --git a/2026-01-pytorch/steps/method2_parquet_export.py b/2026-01-pytorch/steps/method2_parquet_export.py new file mode 100644 index 0000000..5ebfaae --- /dev/null +++ b/2026-01-pytorch/steps/method2_parquet_export.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +""" +Method 2: Parquet Export via REST API + +Demonstrates exporting query results to Parquet files. +Best for reproducible training datasets and larger exports. +""" +import sys +sys.path.insert(0, ".") + +import pyarrow.parquet as pq + +from src.data_loader import export_to_parquet, get_training_query +from config.settings import SYMBOL, DATASET_DIR + + +def main(): + print("=" * 60) + print("METHOD 2: Parquet Export via REST API") + print("=" * 60) + + # Export training data + query = get_training_query(symbol=SYMBOL, days=7) + output_path = DATASET_DIR / "training_data.parquet" + + print(f"\nExporting 7 days of {SYMBOL} data to Parquet...") + print(f"Output: {output_path}") + + export_to_parquet(query, output_path) + + # Verify the export + print("\n" + "-" * 60) + print("Verifying exported file...") + + table = pq.read_table(output_path) + print(f" Rows: {table.num_rows:,}") + print(f" Size: {output_path.stat().st_size / 1024:.1f} KB") + print(f" Schema:") + for field in table.schema: + print(f" - {field.name}: {field.type}") + + # Show sample + df = table.to_pandas() + print(f"\nFirst 5 rows:") + print(df.head().to_string(index=False)) + + print("\n" + "=" * 60) + print("Summary:") + print(" - Parquet files are compressed and efficient") + print(" - Schema and types are preserved") + print(" - File can be versioned and shared") + print(f" - Ready for training: {output_path}") + print("=" * 60) + + +if __name__ == "__main__": + main() diff --git a/2026-01-pytorch/steps/method3_parquet_partitions.py b/2026-01-pytorch/steps/method3_parquet_partitions.py new file mode 100644 index 0000000..b66e0bd --- /dev/null +++ b/2026-01-pytorch/steps/method3_parquet_partitions.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +""" +Method 3: Direct Parquet Partition Access + +Demonstrates reading QuestDB's Parquet partitions directly. +Best for terabyte-scale data and streaming processing. + +NOTE: This requires: + 1. Local or mounted access to QuestDB's data directory + 2. Parquet partition format enabled in QuestDB config + +If you're using the remote demo instance, this won't work—use Method 1 or 2 instead. +""" +import sys +sys.path.insert(0, ".") + +from pathlib import Path +from config.settings import QUESTDB_DATA_DIR, TABLE_NAME + + +def main(): + print("=" * 60) + print("METHOD 3: Direct Parquet Partition Access") + print("=" * 60) + + print(f"\nLooking for partitions in: {QUESTDB_DATA_DIR / TABLE_NAME}") + + table_dir = QUESTDB_DATA_DIR / TABLE_NAME + + if not table_dir.exists(): + print(f""" + ⚠ Directory not found: {table_dir} + + This method requires local QuestDB access with Parquet partitions. + + To enable Parquet partitions in QuestDB: + + 1. In server.conf, set: + cairo.o3.partition.format=PARQUET + + 2. Or convert existing table partitions: + ALTER TABLE {TABLE_NAME} SET PARAM o3PartitionFormat = 'PARQUET'; + + For the remote demo instance, use Method 1 (direct query) or + Method 2 (Parquet export via API) instead. + """) + return + + # If we get here, try to load partitions + try: + import pyarrow.dataset as ds + + parquet_files = list(table_dir.glob("**/*.parquet")) + + if not parquet_files: + print(f" ⚠ No .parquet files found in {table_dir}") + print(" Partitions may be in native QuestDB format, not Parquet.") + return + + print(f" ✓ Found {len(parquet_files)} partition files") + + # Load as dataset + dataset = ds.dataset(table_dir, format="parquet") + print(f" ✓ Loaded dataset with schema:") + for field in dataset.schema: + print(f" - {field.name}: {field.type}") + + # Demo streaming + print("\n Streaming first 3 batches...") + scanner = dataset.scanner(batch_size=10000) + + for i, batch in enumerate(scanner.to_batches()): + if i >= 3: + break + print(f" Batch {i}: {batch.num_rows} rows") + + print("\n" + "=" * 60) + print("Summary:") + print(" - Direct partition access bypasses the query engine") + print(" - PyArrow dataset API enables efficient streaming") + print(" - Best for large-scale batch processing") + print("=" * 60) + + except Exception as e: + print(f" ⚠ Error loading partitions: {e}") + + +if __name__ == "__main__": + main() diff --git a/2026-01-pytorch/steps/train.py b/2026-01-pytorch/steps/train.py new file mode 100644 index 0000000..f1bf918 --- /dev/null +++ b/2026-01-pytorch/steps/train.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +""" +Full Training Pipeline + +Loads data from Parquet (exported via Method 2), engineers features, +trains a simple LSTM model, and saves the result. + +Run Method 2 first to create the training data: + python steps/method2_parquet_export.py + +Then run this: + python steps/train.py +""" +import sys +sys.path.insert(0, ".") + +import torch +import torch.nn as nn +from torch.utils.data import DataLoader +import pandas as pd +from pathlib import Path + +from src.features import prepare_features, normalize_features, FEATURE_COLS +from src.model import TimeSeriesDataset, SimpleLSTM, train_epoch, evaluate +from config.settings import ( + DATASET_DIR, MODEL_DIR, + SEQUENCE_LENGTH, BATCH_SIZE, EPOCHS, LEARNING_RATE +) + + +def main(): + print("=" * 60) + print("PyTorch Training Pipeline") + print("=" * 60) + + # Check for training data + parquet_path = DATASET_DIR / "training_data.parquet" + if not parquet_path.exists(): + print(f"\n⚠ Training data not found: {parquet_path}") + print("Run Method 2 first to export the data:") + print(" python steps/method2_parquet_export.py") + return + + # Load data + print(f"\n1. Loading data from {parquet_path}...") + df = pd.read_parquet(parquet_path) + df["timestamp"] = pd.to_datetime(df["timestamp"]) + print(f" Loaded {len(df):,} rows") + + # Feature engineering + print("\n2. Engineering features...") + df = prepare_features(df) + print(f" After feature engineering: {len(df):,} rows") + print(f" Features: {FEATURE_COLS}") + + # Train/test split (respect time order) + print("\n3. Splitting data (80/20, time-ordered)...") + split_idx = int(len(df) * 0.8) + train_df = df.iloc[:split_idx] + test_df = df.iloc[split_idx:] + print(f" Train: {len(train_df):,} rows") + print(f" Test: {len(test_df):,} rows") + + # Normalize + print("\n4. Normalizing features...") + X_train, X_test, y_train, y_test, scaler = normalize_features(train_df, test_df) + + # Create PyTorch datasets + print(f"\n5. Creating PyTorch datasets (sequence_length={SEQUENCE_LENGTH})...") + train_dataset = TimeSeriesDataset(X_train, y_train, SEQUENCE_LENGTH) + test_dataset = TimeSeriesDataset(X_test, y_test, SEQUENCE_LENGTH) + print(f" Train sequences: {len(train_dataset):,}") + print(f" Test sequences: {len(test_dataset):,}") + + train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True) + test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE) + + # Initialize model + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + print(f"\n6. Initializing model on {device}...") + + model = SimpleLSTM( + input_size=len(FEATURE_COLS), + hidden_size=32, + num_layers=1 + ).to(device) + + criterion = nn.MSELoss() + optimizer = torch.optim.Adam(model.parameters(), lr=LEARNING_RATE) + + print(f" Model: {model.__class__.__name__}") + print(f" Parameters: {sum(p.numel() for p in model.parameters()):,}") + + # Training loop + print(f"\n7. Training for {EPOCHS} epochs...") + print("-" * 40) + + best_test_loss = float("inf") + + for epoch in range(EPOCHS): + train_loss = train_epoch(model, train_loader, criterion, optimizer, device) + test_loss = evaluate(model, test_loader, criterion, device) + + if test_loss < best_test_loss: + best_test_loss = test_loss + best_epoch = epoch + 1 + + if (epoch + 1) % 5 == 0 or epoch == 0: + print(f" Epoch {epoch+1:3d}/{EPOCHS}: train_loss={train_loss:.4f}, test_loss={test_loss:.4f}") + + print("-" * 40) + print(f" Best test loss: {best_test_loss:.4f} (epoch {best_epoch})") + + # Save model + print("\n8. Saving model...") + MODEL_DIR.mkdir(parents=True, exist_ok=True) + model_path = MODEL_DIR / "volume_forecast_lstm.pt" + + torch.save({ + "model_state_dict": model.state_dict(), + "scaler_mean": scaler.mean_, + "scaler_scale": scaler.scale_, + "feature_cols": FEATURE_COLS, + "sequence_length": SEQUENCE_LENGTH, + "test_loss": best_test_loss + }, model_path) + + print(f" ✓ Saved to {model_path}") + + print("\n" + "=" * 60) + print("Training complete!") + print("=" * 60) + + +if __name__ == "__main__": + main() From e03f97b002207665f85f19482117eeb5330bc6cd Mon Sep 17 00:00:00 2001 From: javier Date: Mon, 16 Mar 2026 13:08:40 +0100 Subject: [PATCH 2/5] ignoring .DS_Store files --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.gitignore b/.gitignore index 15ac48d..6aabe20 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ +.DS_Store + # Byte-compiled / optimized / DLL files __pycache__/ *.py[codz] From c2783d3668b6485529213f85f64f3d68605f44e5 Mon Sep 17 00:00:00 2001 From: javier Date: Tue, 7 Apr 2026 18:17:46 +0200 Subject: [PATCH 3/5] pytorch timesfm time series forecasting example --- 2026-01-pytorch/.DS_Store | Bin 6148 -> 0 bytes 2026-01-pytorch/README.md | 85 ------- 2026-01-pytorch/config/settings.py | 23 -- 2026-01-pytorch/pyproject.toml | 24 -- 2026-01-pytorch/src/features.py | 69 ------ 2026-01-pytorch/src/model.py | 133 ---------- 2026-01-pytorch/steps/method1_direct_query.py | 53 ---- 2026-01-pytorch/steps/train.py | 136 ---------- .../README.md | 115 +++++++++ .../config/__init__.py | 0 .../config/settings.py | 26 ++ .../dataset/.gitkeep | 0 .../outputs/.gitkeep | 0 .../pyproject.toml | 26 ++ .../requirements.txt | 10 + .../src/__init__.py | 0 .../src/data_loader.py | 42 +++- .../src/forecaster.py | 222 +++++++++++++++++ .../steps/__init__.py | 0 .../steps/forecast.py | 233 ++++++++++++++++++ .../steps/method1_direct_query.py | 82 ++++++ .../steps/method2_parquet_export.py | 23 +- .../steps/method3_parquet_partitions.py | 14 +- 23 files changed, 764 insertions(+), 552 deletions(-) delete mode 100644 2026-01-pytorch/.DS_Store delete mode 100644 2026-01-pytorch/README.md delete mode 100644 2026-01-pytorch/config/settings.py delete mode 100644 2026-01-pytorch/pyproject.toml delete mode 100644 2026-01-pytorch/src/features.py delete mode 100644 2026-01-pytorch/src/model.py delete mode 100644 2026-01-pytorch/steps/method1_direct_query.py delete mode 100644 2026-01-pytorch/steps/train.py create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/README.md rename {2026-01-pytorch => 2026-04-time-series-forecasting-pytorch-timesfm}/config/__init__.py (100%) create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/config/settings.py rename 2026-01-pytorch/src/__init__.py => 2026-04-time-series-forecasting-pytorch-timesfm/dataset/.gitkeep (100%) rename 2026-01-pytorch/steps/__init__.py => 2026-04-time-series-forecasting-pytorch-timesfm/outputs/.gitkeep (100%) create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/pyproject.toml create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/requirements.txt create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/src/__init__.py rename {2026-01-pytorch => 2026-04-time-series-forecasting-pytorch-timesfm}/src/data_loader.py (78%) create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/src/forecaster.py create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/steps/__init__.py create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/steps/forecast.py create mode 100644 2026-04-time-series-forecasting-pytorch-timesfm/steps/method1_direct_query.py rename {2026-01-pytorch => 2026-04-time-series-forecasting-pytorch-timesfm}/steps/method2_parquet_export.py (62%) rename {2026-01-pytorch => 2026-04-time-series-forecasting-pytorch-timesfm}/steps/method3_parquet_partitions.py (83%) diff --git a/2026-01-pytorch/.DS_Store b/2026-01-pytorch/.DS_Store deleted file mode 100644 index 9b8c37044d898b2f7a42959e9d9bbac0d175978f..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 6148 zcmeHKyH3ME5S)b+ktj%+@_vAVKR899NX-u*Cze36a0sMG=Z?>Y*#`*Aq@_T+((c@? zckbdTyjcLUI__?P1%NqS5$_tNrq9(Ub`cpN(s{-KTRh?hFE~!Be@`fPi9JquLqFoT zcx7&fhUe|^L~qd%pV4l*e%o|?L$3btYxR#XuG3P1R3H^d1yX@j;JgCbduj74$Bd~! zDv%0%E1=(pLRYMTgQI;q7;FR}&X_ji+GYu2(FCyu4vx&w#HmE5N{krdbmmLc)xg2g z=@7e_=fuqt6N=dF%oi(%RL6{|Kq@d+;L^7X?f-Z5GxPtLl)F?Q75G;Q$ZUPTUh$Qp zw@zM8du^dV(7%keR?c9pm}sq-8*RnQA9Y37d|m?wN24=ubYlJyP+ihefxl4T6SJEl AGynhq diff --git a/2026-01-pytorch/README.md b/2026-01-pytorch/README.md deleted file mode 100644 index fc503b5..0000000 --- a/2026-01-pytorch/README.md +++ /dev/null @@ -1,85 +0,0 @@ -# PyTorch + QuestDB Data Pipeline Tutorial - -This tutorial demonstrates three ways to load time-series data from QuestDB into PyTorch: - -1. **Direct SQL queries** (REST API or ConnectorX) -2. **Parquet export** via REST API -3. **Direct Parquet partition access** (for local QuestDB) - -The example task is volume forecasting, but the data loading patterns apply to any model. - -## Quick Start - -```bash -# Install dependencies -pip install pandas pyarrow requests torch scikit-learn - -# Or with uv: -uv sync - -# Run the demos in order: -python steps/method1_direct_query.py # See direct query methods -python steps/method2_parquet_export.py # Export data to Parquet -python steps/train.py # Train the model -``` - -## What Each Script Does - -| Script | Description | -|--------|-------------| -| `steps/method1_direct_query.py` | Queries QuestDB demo instance, shows REST API and ConnectorX | -| `steps/method2_parquet_export.py` | Exports 7 days of data to `dataset/training_data.parquet` | -| `steps/method3_parquet_partitions.py` | Shows direct partition access (requires local QuestDB) | -| `steps/train.py` | Loads Parquet, engineers features, trains LSTM, saves model | - -## Project Structure - -``` -├── config/ -│ └── settings.py # Configuration (URLs, paths, hyperparameters) -├── src/ -│ ├── data_loader.py # Three data loading methods -│ ├── features.py # Feature engineering -│ └── model.py # PyTorch Dataset and LSTM model -├── steps/ -│ ├── method1_direct_query.py -│ ├── method2_parquet_export.py -│ ├── method3_parquet_partitions.py -│ └── train.py -├── dataset/ # Exported Parquet files -├── models/ # Saved model checkpoints -└── pyproject.toml -``` - -## Data Source - -Uses the `trades` table from QuestDB's public demo: `https://demo.questdb.io` - -The query aggregates raw trades into 1-minute bars: - -```sql -SELECT - timestamp, - symbol, - sum(amount) as volume, - count() as trade_count, - avg(price) as avg_price, - max(price) - min(price) as price_range -FROM trades -WHERE symbol = 'BTC-USD' -SAMPLE BY 1m -ALIGN TO CALENDAR -``` - -## Method Comparison - -| Method | Best For | Requires | -|--------|----------|----------| -| REST API | Exploration, simple queries | Network access | -| ConnectorX | Larger result sets | `pip install connectorx` | -| Parquet Export | Reproducible training data | Network access | -| Direct Partitions | Terabyte-scale data | Local QuestDB with Parquet format | - -## Note on the Model - -The LSTM model is intentionally simple—this tutorial focuses on data pipeline patterns, not forecasting accuracy. In production, you'd benchmark against simpler methods (moving averages, ARIMA) and tune hyperparameters properly. diff --git a/2026-01-pytorch/config/settings.py b/2026-01-pytorch/config/settings.py deleted file mode 100644 index 72c3afa..0000000 --- a/2026-01-pytorch/config/settings.py +++ /dev/null @@ -1,23 +0,0 @@ -from pathlib import Path - -# QuestDB connection -QUESTDB_HTTP = "https://demo.questdb.io" -QUESTDB_PG = "postgresql://admin:quest@demo.questdb.io:8812/qdb" - -# For local QuestDB with Parquet partitions (Method 3) -QUESTDB_DATA_DIR = Path("/var/lib/questdb/db") - -# Paths -PROJECT_ROOT = Path(__file__).parent.parent -DATASET_DIR = PROJECT_ROOT / "dataset" -MODEL_DIR = PROJECT_ROOT / "models" - -# Data settings -TABLE_NAME = "trades" -SYMBOL = "BTC-USDT" - -# Model settings -SEQUENCE_LENGTH = 30 -BATCH_SIZE = 32 -EPOCHS = 20 -LEARNING_RATE = 0.001 diff --git a/2026-01-pytorch/pyproject.toml b/2026-01-pytorch/pyproject.toml deleted file mode 100644 index dbf4b3f..0000000 --- a/2026-01-pytorch/pyproject.toml +++ /dev/null @@ -1,24 +0,0 @@ -[project] -name = "pytorch-questdb-pipeline" -version = "0.1.0" -description = "Tutorial: Loading QuestDB time-series data into PyTorch" -requires-python = ">=3.11" -dependencies = [ - "pandas>=2.0.0", - "pyarrow>=15.0.0", - "requests>=2.31.0", - "torch>=2.0.0", - "scikit-learn>=1.4.0", -] - -[project.optional-dependencies] -fast = [ - "connectorx>=0.3.0", # Faster queries via PostgreSQL protocol -] - -[tool.poe.tasks] -method1 = { cmd = "python steps/method1_direct_query.py", help = "Demo: Direct SQL queries" } -method2 = { cmd = "python steps/method2_parquet_export.py", help = "Demo: Parquet export via API" } -method3 = { cmd = "python steps/method3_parquet_partitions.py", help = "Demo: Direct partition access" } -train = { cmd = "python steps/train.py", help = "Train the LSTM model" } -all = ["method1", "method2", "train"] diff --git a/2026-01-pytorch/src/features.py b/2026-01-pytorch/src/features.py deleted file mode 100644 index 45a465c..0000000 --- a/2026-01-pytorch/src/features.py +++ /dev/null @@ -1,69 +0,0 @@ -""" -Feature engineering for time-series forecasting. -""" -import pandas as pd -import numpy as np -from sklearn.preprocessing import StandardScaler - - -FEATURE_COLS = [ - "volume_lag_1", - "volume_lag_5", - "volume_lag_15", - "volume_ma_15", - "volume_std_15", - "trade_count", - "price_range" -] - - -def prepare_features(df: pd.DataFrame) -> pd.DataFrame: - """ - Add features for volume forecasting. - - Returns DataFrame with features and target column. - """ - df = df.sort_values("timestamp").copy() - - # Ensure numeric types - df["volume"] = pd.to_numeric(df["volume"], errors="coerce") - df["trade_count"] = pd.to_numeric(df["trade_count"], errors="coerce") - df["price_range"] = pd.to_numeric(df["price_range"], errors="coerce") - - # Lag features - for lag in [1, 5, 15]: - df[f"volume_lag_{lag}"] = df["volume"].shift(lag) - - # Rolling statistics - df["volume_ma_15"] = df["volume"].rolling(15).mean() - df["volume_std_15"] = df["volume"].rolling(15).std() - - # Target: next period's volume - df["target"] = df["volume"].shift(-1) - - # Drop rows with NaN (from lags/rolling) - df = df.dropna() - - return df - - -def normalize_features( - train_df: pd.DataFrame, - test_df: pd.DataFrame, - feature_cols: list[str] = FEATURE_COLS -) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, StandardScaler]: - """ - Normalize features using StandardScaler. - - Fits on train, transforms both train and test. - Returns: X_train, X_test, y_train, y_test, scaler - """ - scaler = StandardScaler() - - X_train = scaler.fit_transform(train_df[feature_cols]) - X_test = scaler.transform(test_df[feature_cols]) - - y_train = train_df["target"].values - y_test = test_df["target"].values - - return X_train, X_test, y_train, y_test, scaler diff --git a/2026-01-pytorch/src/model.py b/2026-01-pytorch/src/model.py deleted file mode 100644 index 01fade5..0000000 --- a/2026-01-pytorch/src/model.py +++ /dev/null @@ -1,133 +0,0 @@ -""" -PyTorch model and dataset classes for time-series forecasting. -""" -import torch -import torch.nn as nn -from torch.utils.data import Dataset, IterableDataset -import numpy as np - - -class TimeSeriesDataset(Dataset): - """ - PyTorch Dataset for sequential time-series data. - - Creates overlapping sequences of `sequence_length` to predict a target value. - Use when data fits in memory. - """ - - def __init__( - self, - features: np.ndarray, - targets: np.ndarray, - sequence_length: int = 30 - ): - self.features = torch.tensor(features, dtype=torch.float32) - self.targets = torch.tensor(targets, dtype=torch.float32) - self.sequence_length = sequence_length - - def __len__(self): - return len(self.features) - self.sequence_length - - def __getitem__(self, idx): - X = self.features[idx:idx + self.sequence_length] - y = self.targets[idx + self.sequence_length - 1] # Target at end of sequence - return X, y - - -class StreamingTimeSeriesDataset(IterableDataset): - """ - PyTorch IterableDataset that streams from Parquet. - - Use when data doesn't fit in memory. - Requires a generator function that yields DataFrames. - """ - - def __init__( - self, - batch_generator, # Generator yielding DataFrames - feature_cols: list[str], - target_col: str = "target", - sequence_length: int = 30 - ): - self.batch_generator = batch_generator - self.feature_cols = feature_cols - self.target_col = target_col - self.sequence_length = sequence_length - - def __iter__(self): - for batch_df in self.batch_generator: - features = batch_df[self.feature_cols].values - targets = batch_df[self.target_col].values - - for i in range(len(features) - self.sequence_length): - X = features[i:i + self.sequence_length] - y = targets[i + self.sequence_length - 1] - - yield ( - torch.tensor(X, dtype=torch.float32), - torch.tensor(y, dtype=torch.float32) - ) - - -class SimpleLSTM(nn.Module): - """ - Basic LSTM for sequence-to-value regression. - - This is intentionally simple—a demonstration of the data pipeline, - not a production forecasting model. - """ - - def __init__(self, input_size: int, hidden_size: int = 32, num_layers: int = 1): - super().__init__() - self.lstm = nn.LSTM( - input_size=input_size, - hidden_size=hidden_size, - num_layers=num_layers, - batch_first=True - ) - self.fc = nn.Linear(hidden_size, 1) - - def forward(self, x): - # x shape: (batch, sequence_length, features) - lstm_out, (hidden, _) = self.lstm(x) - # Use final hidden state - out = self.fc(hidden[-1]) - return out.squeeze(-1) - - -def train_epoch(model, dataloader, criterion, optimizer, device): - """Train for one epoch, return average loss.""" - model.train() - total_loss = 0 - n_batches = 0 - - for X, y in dataloader: - X, y = X.to(device), y.to(device) - - optimizer.zero_grad() - pred = model(X) - loss = criterion(pred, y) - loss.backward() - optimizer.step() - - total_loss += loss.item() - n_batches += 1 - - return total_loss / n_batches if n_batches > 0 else 0 - - -def evaluate(model, dataloader, criterion, device): - """Evaluate model, return average loss.""" - model.eval() - total_loss = 0 - n_batches = 0 - - with torch.no_grad(): - for X, y in dataloader: - X, y = X.to(device), y.to(device) - pred = model(X) - loss = criterion(pred, y) - total_loss += loss.item() - n_batches += 1 - - return total_loss / n_batches if n_batches > 0 else 0 diff --git a/2026-01-pytorch/steps/method1_direct_query.py b/2026-01-pytorch/steps/method1_direct_query.py deleted file mode 100644 index f961dc3..0000000 --- a/2026-01-pytorch/steps/method1_direct_query.py +++ /dev/null @@ -1,53 +0,0 @@ -#!/usr/bin/env python3 -""" -Method 1: Direct SQL Queries - -Demonstrates querying QuestDB directly via REST API and ConnectorX. -Best for exploration and smaller datasets. -""" -import sys -sys.path.insert(0, ".") - -from src.data_loader import query_rest_api, query_connectorx, get_training_query -from config.settings import SYMBOL - - -def main(): - query = get_training_query(symbol=SYMBOL, days=1) # Just 1 day for demo - - print("=" * 60) - print("METHOD 1: Direct SQL Queries") - print("=" * 60) - - # Method 1a: REST API - print("\n1a. Querying via REST API...") - print(f" Query: SELECT ... FROM trades WHERE symbol='{SYMBOL}' SAMPLE BY 1m") - - df_rest = query_rest_api(query) - print(f" ✓ Loaded {len(df_rest)} rows") - print(f" Columns: {list(df_rest.columns)}") - print(f"\n First 5 rows:") - print(df_rest.head().to_string(index=False)) - - # Method 1b: ConnectorX (faster for large results) - print("\n" + "-" * 60) - print("\n1b. Querying via ConnectorX (PostgreSQL wire protocol)...") - - try: - df_cx = query_connectorx(query) - print(f" ✓ Loaded {len(df_cx)} rows") - except ImportError: - print(" ⚠ ConnectorX not installed. Run: pip install connectorx") - except Exception as e: - print(f" ⚠ ConnectorX error: {e}") - - print("\n" + "=" * 60) - print("Summary:") - print(" - REST API works everywhere, simple to use") - print(" - ConnectorX is faster for larger result sets") - print(" - Both return pandas DataFrames ready for analysis") - print("=" * 60) - - -if __name__ == "__main__": - main() diff --git a/2026-01-pytorch/steps/train.py b/2026-01-pytorch/steps/train.py deleted file mode 100644 index f1bf918..0000000 --- a/2026-01-pytorch/steps/train.py +++ /dev/null @@ -1,136 +0,0 @@ -#!/usr/bin/env python3 -""" -Full Training Pipeline - -Loads data from Parquet (exported via Method 2), engineers features, -trains a simple LSTM model, and saves the result. - -Run Method 2 first to create the training data: - python steps/method2_parquet_export.py - -Then run this: - python steps/train.py -""" -import sys -sys.path.insert(0, ".") - -import torch -import torch.nn as nn -from torch.utils.data import DataLoader -import pandas as pd -from pathlib import Path - -from src.features import prepare_features, normalize_features, FEATURE_COLS -from src.model import TimeSeriesDataset, SimpleLSTM, train_epoch, evaluate -from config.settings import ( - DATASET_DIR, MODEL_DIR, - SEQUENCE_LENGTH, BATCH_SIZE, EPOCHS, LEARNING_RATE -) - - -def main(): - print("=" * 60) - print("PyTorch Training Pipeline") - print("=" * 60) - - # Check for training data - parquet_path = DATASET_DIR / "training_data.parquet" - if not parquet_path.exists(): - print(f"\n⚠ Training data not found: {parquet_path}") - print("Run Method 2 first to export the data:") - print(" python steps/method2_parquet_export.py") - return - - # Load data - print(f"\n1. Loading data from {parquet_path}...") - df = pd.read_parquet(parquet_path) - df["timestamp"] = pd.to_datetime(df["timestamp"]) - print(f" Loaded {len(df):,} rows") - - # Feature engineering - print("\n2. Engineering features...") - df = prepare_features(df) - print(f" After feature engineering: {len(df):,} rows") - print(f" Features: {FEATURE_COLS}") - - # Train/test split (respect time order) - print("\n3. Splitting data (80/20, time-ordered)...") - split_idx = int(len(df) * 0.8) - train_df = df.iloc[:split_idx] - test_df = df.iloc[split_idx:] - print(f" Train: {len(train_df):,} rows") - print(f" Test: {len(test_df):,} rows") - - # Normalize - print("\n4. Normalizing features...") - X_train, X_test, y_train, y_test, scaler = normalize_features(train_df, test_df) - - # Create PyTorch datasets - print(f"\n5. Creating PyTorch datasets (sequence_length={SEQUENCE_LENGTH})...") - train_dataset = TimeSeriesDataset(X_train, y_train, SEQUENCE_LENGTH) - test_dataset = TimeSeriesDataset(X_test, y_test, SEQUENCE_LENGTH) - print(f" Train sequences: {len(train_dataset):,}") - print(f" Test sequences: {len(test_dataset):,}") - - train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True) - test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE) - - # Initialize model - device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - print(f"\n6. Initializing model on {device}...") - - model = SimpleLSTM( - input_size=len(FEATURE_COLS), - hidden_size=32, - num_layers=1 - ).to(device) - - criterion = nn.MSELoss() - optimizer = torch.optim.Adam(model.parameters(), lr=LEARNING_RATE) - - print(f" Model: {model.__class__.__name__}") - print(f" Parameters: {sum(p.numel() for p in model.parameters()):,}") - - # Training loop - print(f"\n7. Training for {EPOCHS} epochs...") - print("-" * 40) - - best_test_loss = float("inf") - - for epoch in range(EPOCHS): - train_loss = train_epoch(model, train_loader, criterion, optimizer, device) - test_loss = evaluate(model, test_loader, criterion, device) - - if test_loss < best_test_loss: - best_test_loss = test_loss - best_epoch = epoch + 1 - - if (epoch + 1) % 5 == 0 or epoch == 0: - print(f" Epoch {epoch+1:3d}/{EPOCHS}: train_loss={train_loss:.4f}, test_loss={test_loss:.4f}") - - print("-" * 40) - print(f" Best test loss: {best_test_loss:.4f} (epoch {best_epoch})") - - # Save model - print("\n8. Saving model...") - MODEL_DIR.mkdir(parents=True, exist_ok=True) - model_path = MODEL_DIR / "volume_forecast_lstm.pt" - - torch.save({ - "model_state_dict": model.state_dict(), - "scaler_mean": scaler.mean_, - "scaler_scale": scaler.scale_, - "feature_cols": FEATURE_COLS, - "sequence_length": SEQUENCE_LENGTH, - "test_loss": best_test_loss - }, model_path) - - print(f" ✓ Saved to {model_path}") - - print("\n" + "=" * 60) - print("Training complete!") - print("=" * 60) - - -if __name__ == "__main__": - main() diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/README.md b/2026-04-time-series-forecasting-pytorch-timesfm/README.md new file mode 100644 index 0000000..8f3f6f5 --- /dev/null +++ b/2026-04-time-series-forecasting-pytorch-timesfm/README.md @@ -0,0 +1,115 @@ +# QuestDB → PyTorch: Time-Series Data Loading & Forecasting + +This tutorial demonstrates how to efficiently load time-series data from QuestDB into Python, then run forecasts using Google's TimesFM foundation model. + +**Focus**: Data pipeline patterns. The forecasting is just to show the complete workflow. + +## What's Inside + +| File | Purpose | +|------|---------| +| `steps/method1_direct_query.py` | Load data via REST API and ConnectorX | +| `steps/method2_parquet_export.py` | Export data to Parquet files | +| `steps/method3_parquet_partitions.py` | Read QuestDB partitions directly | +| `steps/forecast.py` | Run TimesFM forecasts on volume & volatility | + +## The Use Case + +We forecast two operationally useful metrics from crypto trade data: + +1. **Volume** – When will liquidity be available? (execution timing) +2. **Volatility** – How much is price moving? (risk management) + +We intentionally avoid price prediction—forecasting "where will BTC be in an hour" isn't credible. Volume and volatility forecasting, however, serve real needs. + +## Quick Start + +> **Requires Python 3.10+** + +```bash +# 1. Clone and setup +git clone https://github.com/questdb/blog-examples.git +cd blog-examples/2025-pytorch-questdb-timesfm + +# 2. Create environment +python3.11 -m venv .venv +source .venv/bin/activate + +# 3. Install base dependencies +pip install pandas pyarrow requests torch + +# 4. Install TimesFM from source (pip package is outdated) +git clone https://github.com/google-research/timesfm.git +cd timesfm +pip install -e ".[torch]" +cd .. + +# 5. Run the pipeline +python steps/method1_direct_query.py # Test QuestDB connection +python steps/method2_parquet_export.py # Export data to Parquet +python steps/forecast.py # Run forecasts +``` + +## Data Source + +Uses the `trades` table from QuestDB's public demo: `https://demo.questdb.io` + +Raw trades are aggregated into 1-minute OHLCV bars: + +```sql +SELECT + timestamp, + first(price) as open, + max(price) as high, + min(price) as low, + last(price) as close, + sum(amount) as volume, + count() as trade_count +FROM trades +WHERE symbol = 'BTC-USDT' +SAMPLE BY 1m +ALIGN TO CALENDAR +``` + +## Three Data Loading Methods + +| Method | Best For | How | +|--------|----------|-----| +| **REST API** | Exploration, small data | `requests.get("/exp?fmt=csv")` | +| **ConnectorX** | Large results (local QuestDB) | PostgreSQL wire protocol | +| **Parquet Export** | Reproducible pipelines | `requests.get("/exp?fmt=parquet")` | +| **Direct Partitions** | Terabyte-scale | Read QuestDB's Parquet files directly | + +> **Note**: ConnectorX requires PostgreSQL port access (8812). The public demo only exposes REST API, so ConnectorX works with local/self-hosted QuestDB only. + +## Project Structure + +``` +├── config/settings.py # Configuration (URLs, symbols, params) +├── src/ +│ ├── data_loader.py # Three data loading methods +│ └── forecaster.py # TimesFM wrapper +├── steps/ +│ ├── method1_direct_query.py +│ ├── method2_parquet_export.py +│ ├── method3_parquet_partitions.py +│ └── forecast.py +├── dataset/ # Exported Parquet files +├── outputs/ # Forecast results (CSV) +└── BLOG_POST.md # Full tutorial writeup +``` + +## Notes + +- **Python 3.10+ required** – TimesFM doesn't support older versions +- **TimesFM must be installed from source** – The PyPI package is outdated +- **Model downloads ~900MB on first run** – Cached in `~/.cache/huggingface/` +- **GPU optional** – CPU inference works fine, just slower + +## Honest Caveats + +1. **Foundation models aren't magic.** TimesFM gives a quick baseline, but tuned domain-specific models may perform better. + +2. **Volume is hard to forecast.** It's driven by news, large orders, and sentiment—all unpredictable. Wide confidence intervals are expected. + +3. **This is a data pipeline tutorial.** Whether these forecasts are useful for your application requires proper backtesting. diff --git a/2026-01-pytorch/config/__init__.py b/2026-04-time-series-forecasting-pytorch-timesfm/config/__init__.py similarity index 100% rename from 2026-01-pytorch/config/__init__.py rename to 2026-04-time-series-forecasting-pytorch-timesfm/config/__init__.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/config/settings.py b/2026-04-time-series-forecasting-pytorch-timesfm/config/settings.py new file mode 100644 index 0000000..a3a435d --- /dev/null +++ b/2026-04-time-series-forecasting-pytorch-timesfm/config/settings.py @@ -0,0 +1,26 @@ +from pathlib import Path + +# QuestDB connection settings +# REST API - points to public demo instance (works out of the box) +QUESTDB_HTTP = "https://demo.questdb.io" + +# PostgreSQL wire protocol - points to local QuestDB (for ConnectorX) +# The demo instance doesn't expose port 8812 externally, so this requires local QuestDB +QUESTDB_PG = "postgresql://admin:quest@localhost:8812/qdb" + +# For local QuestDB with Parquet partitions (Method 3) +QUESTDB_DATA_DIR = Path("/var/lib/questdb/db") + +# Paths +PROJECT_ROOT = Path(__file__).parent.parent +DATASET_DIR = PROJECT_ROOT / "dataset" +OUTPUT_DIR = PROJECT_ROOT / "outputs" + +# Data settings +TABLE_NAME = "trades" +SYMBOL = "BTC-USDT" + +# TimesFM settings +TIMESFM_MODEL = "google/timesfm-2.5-200m-pytorch" +MAX_CONTEXT = 512 # How much history to use +FORECAST_HORIZON = 60 # How far ahead to forecast (in minutes) diff --git a/2026-01-pytorch/src/__init__.py b/2026-04-time-series-forecasting-pytorch-timesfm/dataset/.gitkeep similarity index 100% rename from 2026-01-pytorch/src/__init__.py rename to 2026-04-time-series-forecasting-pytorch-timesfm/dataset/.gitkeep diff --git a/2026-01-pytorch/steps/__init__.py b/2026-04-time-series-forecasting-pytorch-timesfm/outputs/.gitkeep similarity index 100% rename from 2026-01-pytorch/steps/__init__.py rename to 2026-04-time-series-forecasting-pytorch-timesfm/outputs/.gitkeep diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/pyproject.toml b/2026-04-time-series-forecasting-pytorch-timesfm/pyproject.toml new file mode 100644 index 0000000..0cb7dbb --- /dev/null +++ b/2026-04-time-series-forecasting-pytorch-timesfm/pyproject.toml @@ -0,0 +1,26 @@ +[project] +name = "pytorch-questdb-timesfm" +version = "0.1.0" +description = "Tutorial: QuestDB time-series data with Google's TimesFM foundation model" +requires-python = ">=3.10" +dependencies = [ + "pandas>=2.0.0", + "pyarrow>=15.0.0", + "requests>=2.31.0", + "torch>=2.0.0", +] + +[project.optional-dependencies] +# NOTE: TimesFM must be installed from source, not pip +# git clone https://github.com/google-research/timesfm.git +# cd timesfm && pip install -e ".[torch]" +fast = [ + "connectorx>=0.3.0", # Faster queries via PostgreSQL protocol +] + +[tool.poe.tasks] +method1 = { cmd = "python steps/method1_direct_query.py", help = "Demo: Direct SQL queries" } +method2 = { cmd = "python steps/method2_parquet_export.py", help = "Demo: Parquet export via API" } +method3 = { cmd = "python steps/method3_parquet_partitions.py", help = "Demo: Direct partition access" } +forecast = { cmd = "python steps/forecast.py", help = "Run TimesFM forecasting" } +all = ["method1", "method2", "forecast"] diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/requirements.txt b/2026-04-time-series-forecasting-pytorch-timesfm/requirements.txt new file mode 100644 index 0000000..2169c21 --- /dev/null +++ b/2026-04-time-series-forecasting-pytorch-timesfm/requirements.txt @@ -0,0 +1,10 @@ +# Core dependencies +pandas>=2.0.0 +pyarrow>=15.0.0 +requests>=2.31.0 +torch>=2.0.0 +connectorx>=0.3.0 + +# TimesFM must be installed separately from source: +# git clone https://github.com/google-research/timesfm.git +# cd timesfm && pip install -e ".[torch]" diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/src/__init__.py b/2026-04-time-series-forecasting-pytorch-timesfm/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/2026-01-pytorch/src/data_loader.py b/2026-04-time-series-forecasting-pytorch-timesfm/src/data_loader.py similarity index 78% rename from 2026-01-pytorch/src/data_loader.py rename to 2026-04-time-series-forecasting-pytorch-timesfm/src/data_loader.py index c7d428e..2a5e1fd 100644 --- a/2026-01-pytorch/src/data_loader.py +++ b/2026-04-time-series-forecasting-pytorch-timesfm/src/data_loader.py @@ -1,5 +1,5 @@ """ -Three methods to load data from QuestDB into Python/PyTorch. +Three methods to load data from QuestDB into Python. """ import pandas as pd import requests @@ -32,9 +32,14 @@ def query_connectorx(query: str) -> pd.DataFrame: Query QuestDB via PostgreSQL wire protocol using ConnectorX. Faster than REST for larger results. Requires: pip install connectorx + + Note: Uses redshift:// driver because QuestDB's PG wire protocol is + compatible with Redshift's simpler dialect. """ import connectorx as cx - return cx.read_sql(QUESTDB_PG, query) + # Replace postgresql:// with redshift:// for QuestDB compatibility + conn = QUESTDB_PG.replace("postgresql://", "redshift://") + return cx.read_sql(conn, query) # ============================================================================= @@ -46,7 +51,7 @@ def export_to_parquet(query: str, output_path: Path) -> Path: Export query results to Parquet file via QuestDB REST API. Falls back to CSV if Parquet export is disabled on the server. - Best for reproducible training datasets. + Best for reproducible datasets. """ output_path.parent.mkdir(parents=True, exist_ok=True) @@ -123,16 +128,35 @@ def stream_from_partitions(dataset, columns: list[str], batch_size: int = 100_00 # Query builders # ============================================================================= -def get_training_query(symbol: str = "BTC-USDT", days: int = 7) -> str: - """Build query for training data: 1-minute aggregated bars.""" +def get_ohlcv_query(symbol: str = "BTC-USDT", days: int = 7, interval: str = "1m") -> str: + """ + Build query for OHLCV data aggregated at specified interval. + + Returns: timestamp, open, high, low, close, volume + """ return f""" SELECT timestamp, - symbol, + first(price) as open, + max(price) as high, + min(price) as low, + last(price) as close, sum(amount) as volume, - count() as trade_count, - avg(price) as avg_price, - max(price) - min(price) as price_range + count() as trade_count + FROM trades + WHERE symbol = '{symbol}' + AND timestamp > dateadd('d', -{days}, now()) + SAMPLE BY {interval} + ALIGN TO CALENDAR + """ + + +def get_volume_series_query(symbol: str = "BTC-USDT", days: int = 7) -> str: + """Build query for volume time series (1-minute intervals).""" + return f""" + SELECT + timestamp, + sum(amount) as volume FROM trades WHERE symbol = '{symbol}' AND timestamp > dateadd('d', -{days}, now()) diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/src/forecaster.py b/2026-04-time-series-forecasting-pytorch-timesfm/src/forecaster.py new file mode 100644 index 0000000..66371f4 --- /dev/null +++ b/2026-04-time-series-forecasting-pytorch-timesfm/src/forecaster.py @@ -0,0 +1,222 @@ +""" +TimesFM wrapper for easy forecasting. + +TimesFM is a foundation model for time-series forecasting developed by Google Research. +It provides zero-shot forecasting - no training needed, just pass your data. + +Paper: https://arxiv.org/abs/2310.10688 +Model: https://huggingface.co/google/timesfm-2.5-200m-pytorch +""" +import os +import logging +import warnings +import numpy as np +import pandas as pd +from typing import Optional +from dataclasses import dataclass + +# Suppress noisy HuggingFace Hub warnings before importing anything else +os.environ["HF_HUB_DISABLE_PROGRESS_BARS"] = "1" +os.environ["HF_HUB_VERBOSITY"] = "error" +logging.getLogger("huggingface_hub").setLevel(logging.ERROR) +logging.getLogger("huggingface_hub.utils._http").setLevel(logging.ERROR) +warnings.filterwarnings("ignore", message=".*unauthenticated.*") +warnings.filterwarnings("ignore", message=".*HF Hub.*") + + +@dataclass +class ForecastResult: + """Container for forecast results.""" + timestamps: pd.DatetimeIndex + point_forecast: np.ndarray + quantile_forecast: Optional[np.ndarray] # Shape: (horizon, num_quantiles) + quantile_levels: Optional[list[float]] # e.g., [0.1, 0.2, ..., 0.9] + + def to_dataframe(self) -> pd.DataFrame: + """Convert to DataFrame with point forecast and quantile columns.""" + df = pd.DataFrame({ + "timestamp": self.timestamps, + "forecast": self.point_forecast, + }) + + if self.quantile_forecast is not None and self.quantile_levels is not None: + for i, q in enumerate(self.quantile_levels): + df[f"q{int(q*100):02d}"] = self.quantile_forecast[:, i] + + return df + + +class TimesFMForecaster: + """ + Wrapper around Google's TimesFM model for easy forecasting. + + Usage: + forecaster = TimesFMForecaster() + forecaster.load_model() + + result = forecaster.forecast( + series=df["volume"].values, + timestamps=df["timestamp"], + horizon=60 + ) + """ + + def __init__( + self, + model_name: str = "google/timesfm-2.5-200m-pytorch", + max_context: int = 512, + max_horizon: int = 256, + use_quantiles: bool = True, + ): + self.model_name = model_name + self.max_context = max_context + self.max_horizon = max_horizon + self.use_quantiles = use_quantiles + self.model = None + + def load_model(self): + """Load the TimesFM model from HuggingFace.""" + import warnings + import logging + import torch + + # Suppress HuggingFace warnings aggressively + warnings.filterwarnings("ignore", message=".*unauthenticated.*") + warnings.filterwarnings("ignore", message=".*HF_TOKEN.*") + logging.getLogger("huggingface_hub").setLevel(logging.ERROR) + logging.getLogger("huggingface_hub.utils._http").setLevel(logging.ERROR) + + import timesfm + + print(f"Loading TimesFM model: {self.model_name}") + torch.set_float32_matmul_precision("high") + + # Handle different TimesFM versions + if hasattr(timesfm, 'TimesFM_2p5_200M_torch'): + # Source install (recommended) + self.model = timesfm.TimesFM_2p5_200M_torch.from_pretrained(self.model_name) + elif hasattr(timesfm, 'TimesFm'): + # Older API + self.model = timesfm.TimesFm( + hparams=timesfm.TimesFmHparams( + backend="gpu" if torch.cuda.is_available() else "cpu", + per_core_batch_size=32, + horizon_len=self.max_horizon, + context_len=self.max_context, + ), + checkpoint=timesfm.TimesFmCheckpoint( + huggingface_repo_id=self.model_name + ), + ) + print("✓ Model loaded (legacy API)") + return + else: + raise RuntimeError( + "TimesFM not properly installed. Please install from source:\n" + " git clone https://github.com/google-research/timesfm.git\n" + " cd timesfm && pip install -e '.[torch]'" + ) + + self.model.compile( + timesfm.ForecastConfig( + max_context=self.max_context, + max_horizon=self.max_horizon, + normalize_inputs=True, + use_continuous_quantile_head=self.use_quantiles, + force_flip_invariance=True, + infer_is_positive=True, + fix_quantile_crossing=True, + ) + ) + print("✓ Model loaded and compiled") + + def forecast( + self, + series: np.ndarray, + timestamps: pd.DatetimeIndex, + horizon: int, + freq: str = "1min", + ) -> ForecastResult: + """ + Generate forecasts for a time series. + + Args: + series: 1D numpy array of historical values + timestamps: DatetimeIndex corresponding to series + horizon: Number of steps to forecast + freq: Frequency for generating future timestamps + + Returns: + ForecastResult with point and quantile forecasts + """ + if self.model is None: + self.load_model() + + # Ensure we don't exceed max context + if len(series) > self.max_context: + series = series[-self.max_context:] + timestamps = timestamps[-self.max_context:] + + # Run forecast + point_forecast, quantile_forecast = self.model.forecast( + horizon=horizon, + inputs=[series.astype(np.float32)], + ) + + # Generate future timestamps + last_ts = timestamps[-1] + future_timestamps = pd.date_range( + start=last_ts + pd.Timedelta(freq), + periods=horizon, + freq=freq + ) + + # Extract results (model returns batch dimension) + point = point_forecast[0] # Shape: (horizon,) + + if quantile_forecast is not None: + # Shape: (1, horizon, num_quantiles) -> (horizon, num_quantiles) + quantiles = quantile_forecast[0] + # TimesFM 2.5 returns: mean, then 10th to 90th percentiles + quantile_levels = [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9] + else: + quantiles = None + quantile_levels = None + + return ForecastResult( + timestamps=future_timestamps, + point_forecast=point, + quantile_forecast=quantiles, + quantile_levels=quantile_levels, + ) + + def forecast_multiple( + self, + series_list: list[np.ndarray], + horizon: int, + ) -> list[np.ndarray]: + """ + Forecast multiple time series efficiently in a single batch. + + Args: + series_list: List of 1D numpy arrays + horizon: Number of steps to forecast + + Returns: + List of point forecasts (one per input series) + """ + if self.model is None: + self.load_model() + + # Truncate each series to max_context + inputs = [ + s[-self.max_context:].astype(np.float32) + for s in series_list + ] + + point_forecasts, _ = self.model.forecast( + horizon=horizon, + inputs=inputs, + ) + + return [point_forecasts[i] for i in range(len(inputs))] diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/steps/__init__.py b/2026-04-time-series-forecasting-pytorch-timesfm/steps/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/steps/forecast.py b/2026-04-time-series-forecasting-pytorch-timesfm/steps/forecast.py new file mode 100644 index 0000000..4f2bc85 --- /dev/null +++ b/2026-04-time-series-forecasting-pytorch-timesfm/steps/forecast.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python3 +""" +Zero-Shot Forecasting with TimesFM + +This script demonstrates using Google's TimesFM foundation model +for time-series forecasting on QuestDB data. + +We forecast two operationally useful metrics: +1. Volume - When will liquidity be available? Useful for execution timing. +2. Volatility - How much is price moving? Useful for risk management. + +Note: We intentionally avoid price prediction. Forecasting "where will BTC +be in an hour" is essentially claiming to beat the market—not credible. +Volume and volatility forecasting, however, serve real operational needs. +""" +# Suppress HuggingFace warnings before any imports +import os +import logging +import warnings +os.environ["HF_HUB_VERBOSITY"] = "error" +logging.getLogger("huggingface_hub").setLevel(logging.ERROR) +warnings.filterwarnings("ignore", message=".*unauthenticated.*") +warnings.filterwarnings("ignore", message=".*HF_TOKEN.*") + +import sys +sys.path.insert(0, ".") + +import pandas as pd +import numpy as np +from pathlib import Path + +from src.data_loader import query_rest_api, get_ohlcv_query +from src.forecaster import TimesFMForecaster +from config.settings import ( + SYMBOL, DATASET_DIR, OUTPUT_DIR, + MAX_CONTEXT, FORECAST_HORIZON +) + + +def load_data() -> pd.DataFrame: + """ + Load OHLCV data from Parquet (preferred) or live query (fallback). + + Returns DataFrame with columns: timestamp, open, high, low, close, volume, trade_count + """ + parquet_path = DATASET_DIR / "btc_ohlcv_7d.parquet" + + if parquet_path.exists(): + print(f"Loading from Parquet: {parquet_path}") + df = pd.read_parquet(parquet_path) + else: + print(f"Parquet not found, querying live data from QuestDB...") + print(f" (Run method2_parquet_export.py first for reproducible results)") + query = get_ohlcv_query(symbol=SYMBOL, days=7) + df = query_rest_api(query) + + df["timestamp"] = pd.to_datetime(df["timestamp"]) + df = df.sort_values("timestamp").reset_index(drop=True) + + return df + + +def compute_volatility(df: pd.DataFrame) -> np.ndarray: + """ + Compute per-minute volatility as the high-low range. + + This is a simple measure of intra-minute price movement. + Higher values = more volatile = higher risk/reward. + """ + return (df["high"] - df["low"]).values + + +def main(): + print("=" * 70) + print("Zero-Shot Forecasting with TimesFM") + print("=" * 70) + + # --- 1. Load Data --- + print("\n1. LOADING DATA") + print("-" * 70) + df = load_data() + print(f" Rows loaded: {len(df):,}") + print(f" Time range: {df['timestamp'].min()} to {df['timestamp'].max()}") + print(f" Columns: {', '.join(df.columns)}") + + # --- 2. Prepare Series --- + print("\n2. PREPARING TIME SERIES") + print("-" * 70) + + timestamps = pd.DatetimeIndex(df["timestamp"]) + + # Volume: total amount traded per minute + volume_series = df["volume"].values + print(f" Volume series: {len(volume_series)} points") + print(f" Recent values: {volume_series[-5:].round(3)}") + print(f" Mean: {volume_series.mean():.3f}") + + # Volatility: high-low range per minute + volatility_series = compute_volatility(df) + print(f" Volatility series: {len(volatility_series)} points") + print(f" Recent values: {volatility_series[-5:].round(2)}") + print(f" Mean: ${volatility_series.mean():.2f}") + + print(f"\n Using last {MAX_CONTEXT} points as context for forecasting") + print(f" Forecasting {FORECAST_HORIZON} minutes ahead") + + # --- 3. Load Model --- + print("\n3. LOADING TIMESFM MODEL") + print("-" * 70) + print(" TimesFM is a foundation model trained on 100B+ time points.") + print(" It provides zero-shot forecasting—no training on your data needed.") + print() + + forecaster = TimesFMForecaster( + max_context=MAX_CONTEXT, + max_horizon=FORECAST_HORIZON, + use_quantiles=True, + ) + forecaster.load_model() + + # --- 4. Generate Forecasts --- + print("\n4. GENERATING FORECASTS") + print("-" * 70) + + print(" Forecasting volume (liquidity timing)...") + volume_result = forecaster.forecast( + series=volume_series, + timestamps=timestamps, + horizon=FORECAST_HORIZON, + freq="1min", + ) + + print(" Forecasting volatility (risk assessment)...") + volatility_result = forecaster.forecast( + series=volatility_series, + timestamps=timestamps, + horizon=FORECAST_HORIZON, + freq="1min", + ) + + # --- 5. Display Results --- + print("\n" + "=" * 70) + print("FORECAST RESULTS") + print("=" * 70) + + # Volume forecast + print("\n📊 VOLUME FORECAST (next 60 minutes)") + print(" Use case: Execution timing—when will liquidity be available?") + print() + volume_df = volume_result.to_dataframe() + # Show subset of columns for readability + display_cols = ["timestamp", "forecast", "q10", "q50", "q90"] + print(volume_df[display_cols].head(10).to_string(index=False)) + print(" ...") + print(volume_df[display_cols].tail(5).to_string(index=False)) + + # Volatility forecast + print("\n📈 VOLATILITY FORECAST (next 60 minutes)") + print(" Use case: Risk management—how much is price likely to move?") + print() + volatility_df = volatility_result.to_dataframe() + print(volatility_df[display_cols].head(10).to_string(index=False)) + print(" ...") + print(volatility_df[display_cols].tail(5).to_string(index=False)) + + # --- 6. Save Results --- + OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + + volume_output = OUTPUT_DIR / "volume_forecast.csv" + volatility_output = OUTPUT_DIR / "volatility_forecast.csv" + + volume_df.to_csv(volume_output, index=False) + volatility_df.to_csv(volatility_output, index=False) + + print(f"\n✓ Saved forecasts to:") + print(f" - {volume_output}") + print(f" - {volatility_output}") + + # --- 7. Summary Statistics --- + print("\n" + "=" * 70) + print("SUMMARY") + print("=" * 70) + + # Volume summary + last_volume = volume_series[-1] + forecast_volume = volume_result.point_forecast + + print(f"\n📊 Volume:") + print(f" Last observed: {last_volume:.3f}") + print(f" Forecast mean: {forecast_volume.mean():.3f}") + print(f" Forecast range: {forecast_volume.min():.3f} – {forecast_volume.max():.3f}") + if volume_result.quantile_forecast is not None: + q10_mean = volume_result.quantile_forecast[:, 0].mean() + q90_mean = volume_result.quantile_forecast[:, -1].mean() + print(f" 80% confidence band: {q10_mean:.3f} – {q90_mean:.3f}") + + # Volatility summary + last_volatility = volatility_series[-1] + forecast_volatility = volatility_result.point_forecast + + print(f"\n📈 Volatility (price range per minute):") + print(f" Last observed: ${last_volatility:.2f}") + print(f" Forecast mean: ${forecast_volatility.mean():.2f}") + print(f" Forecast range: ${forecast_volatility.min():.2f} – ${forecast_volatility.max():.2f}") + if volatility_result.quantile_forecast is not None: + q10_mean = volatility_result.quantile_forecast[:, 0].mean() + q90_mean = volatility_result.quantile_forecast[:, -1].mean() + print(f" 80% confidence band: ${q10_mean:.2f} – ${q90_mean:.2f}") + + # Interpretation + print("\n" + "-" * 70) + print("INTERPRETATION") + print("-" * 70) + print(""" + • Wide confidence intervals are expected—volume and volatility are + inherently noisy and driven by unpredictable factors (news, large + orders, market sentiment). + + • These forecasts provide a baseline. For production use, you'd want + to backtest against historical data and compare with simpler methods + like moving averages or ARIMA. + + • The value of foundation models like TimesFM is speed: you get a + reasonable forecast immediately, without training or tuning. + """) + + print("=" * 70) + print("Done!") + print("=" * 70) + + +if __name__ == "__main__": + main() diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/steps/method1_direct_query.py b/2026-04-time-series-forecasting-pytorch-timesfm/steps/method1_direct_query.py new file mode 100644 index 0000000..30ff4f3 --- /dev/null +++ b/2026-04-time-series-forecasting-pytorch-timesfm/steps/method1_direct_query.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +""" +Method 1: Direct SQL Queries + +Demonstrates querying QuestDB directly via REST API and ConnectorX. +Best for exploration and smaller datasets. +""" +import sys +sys.path.insert(0, ".") + +from src.data_loader import query_rest_api, query_connectorx, get_ohlcv_query +from config.settings import SYMBOL, QUESTDB_HTTP, QUESTDB_PG + + +def main(): + query = get_ohlcv_query(symbol=SYMBOL, days=1) + + print("=" * 60) + print("METHOD 1: Direct SQL Queries") + print("=" * 60) + print("\nConnection settings (from config/settings.py):") + print(f" REST API: {QUESTDB_HTTP}") + print(f" ConnectorX: {QUESTDB_PG}") + + # Method 1a: REST API + print("\n" + "-" * 60) + print("1a. REST API") + print("-" * 60) + print(f"Connecting to: {QUESTDB_HTTP}") + print(f"Query: {SYMBOL} OHLCV, 1-minute bars, last 24 hours") + + df = query_rest_api(query) + print(f"\n✓ Loaded {len(df):,} rows") + print(f" Columns: {list(df.columns)}") + print(f"\nFirst 5 rows:") + print(df.head().to_string(index=False)) + print(f"\nLast 5 rows:") + print(df.tail().to_string(index=False)) + + # Method 1b: ConnectorX + print("\n" + "-" * 60) + print("1b. ConnectorX (PostgreSQL wire protocol)") + print("-" * 60) + print(f"Connecting to: {QUESTDB_PG}") + print(f"Query: {SYMBOL} OHLCV, 1-minute bars, last 24 hours") + print() + + try: + df_cx = query_connectorx(query) + print(f"✓ Loaded {len(df_cx):,} rows") + print(f"\nFirst 5 rows:") + print(df_cx.head().to_string(index=False)) + except ImportError: + print("⚠ ConnectorX not installed. Run: pip install connectorx") + except Exception as e: + print(f"⚠ ConnectorX failed: {e}") + print("\n This is expected if you don't have local QuestDB running.") + print(" ConnectorX requires PostgreSQL wire protocol access (port 8812),") + print(" which the public demo instance doesn't expose.") + + # Summary + print("\n" + "=" * 60) + print("SUMMARY") + print("=" * 60) + print(""" +REST API (demo.questdb.io) + - Works out of the box, no local setup needed + - Good for exploration and moderate datasets + +ConnectorX (localhost:8812) + - Requires local QuestDB with PostgreSQL wire protocol + - ~5x faster for large result sets (100K+ rows) + - Edit config/settings.py to change connection URLs + +Both methods return pandas DataFrames ready for analysis or ML. +""") + + return df + + +if __name__ == "__main__": + main() diff --git a/2026-01-pytorch/steps/method2_parquet_export.py b/2026-04-time-series-forecasting-pytorch-timesfm/steps/method2_parquet_export.py similarity index 62% rename from 2026-01-pytorch/steps/method2_parquet_export.py rename to 2026-04-time-series-forecasting-pytorch-timesfm/steps/method2_parquet_export.py index 5ebfaae..1e2fe56 100644 --- a/2026-01-pytorch/steps/method2_parquet_export.py +++ b/2026-04-time-series-forecasting-pytorch-timesfm/steps/method2_parquet_export.py @@ -3,14 +3,14 @@ Method 2: Parquet Export via REST API Demonstrates exporting query results to Parquet files. -Best for reproducible training datasets and larger exports. +Best for reproducible datasets and larger exports. """ import sys sys.path.insert(0, ".") import pyarrow.parquet as pq -from src.data_loader import export_to_parquet, get_training_query +from src.data_loader import export_to_parquet, get_ohlcv_query from config.settings import SYMBOL, DATASET_DIR @@ -19,11 +19,11 @@ def main(): print("METHOD 2: Parquet Export via REST API") print("=" * 60) - # Export training data - query = get_training_query(symbol=SYMBOL, days=7) - output_path = DATASET_DIR / "training_data.parquet" + # Export OHLCV data + query = get_ohlcv_query(symbol=SYMBOL, days=7) + output_path = DATASET_DIR / "btc_ohlcv_7d.parquet" - print(f"\nExporting 7 days of {SYMBOL} data to Parquet...") + print(f"\nExporting 7 days of {SYMBOL} OHLCV data...") print(f"Output: {output_path}") export_to_parquet(query, output_path) @@ -33,8 +33,8 @@ def main(): print("Verifying exported file...") table = pq.read_table(output_path) - print(f" Rows: {table.num_rows:,}") - print(f" Size: {output_path.stat().st_size / 1024:.1f} KB") + print(f" Rows: {table.num_rows:,}") + print(f" Size: {output_path.stat().st_size / 1024:.1f} KB") print(f" Schema:") for field in table.schema: print(f" - {field.name}: {field.type}") @@ -46,11 +46,12 @@ def main(): print("\n" + "=" * 60) print("Summary:") - print(" - Parquet files are compressed and efficient") - print(" - Schema and types are preserved") + print(" - Parquet preserves types and compresses well") print(" - File can be versioned and shared") - print(f" - Ready for training: {output_path}") + print(f" - Ready for forecasting: {output_path}") print("=" * 60) + + return output_path if __name__ == "__main__": diff --git a/2026-01-pytorch/steps/method3_parquet_partitions.py b/2026-04-time-series-forecasting-pytorch-timesfm/steps/method3_parquet_partitions.py similarity index 83% rename from 2026-01-pytorch/steps/method3_parquet_partitions.py rename to 2026-04-time-series-forecasting-pytorch-timesfm/steps/method3_parquet_partitions.py index b66e0bd..9e96d48 100644 --- a/2026-01-pytorch/steps/method3_parquet_partitions.py +++ b/2026-04-time-series-forecasting-pytorch-timesfm/steps/method3_parquet_partitions.py @@ -9,12 +9,11 @@ 1. Local or mounted access to QuestDB's data directory 2. Parquet partition format enabled in QuestDB config -If you're using the remote demo instance, this won't work—use Method 1 or 2 instead. +If you're using the remote demo instance, this won't work—use Method 1 or 2. """ import sys sys.path.insert(0, ".") -from pathlib import Path from config.settings import QUESTDB_DATA_DIR, TABLE_NAME @@ -41,12 +40,10 @@ def main(): 2. Or convert existing table partitions: ALTER TABLE {TABLE_NAME} SET PARAM o3PartitionFormat = 'PARQUET'; - For the remote demo instance, use Method 1 (direct query) or - Method 2 (Parquet export via API) instead. + For the remote demo instance, use Method 1 or Method 2 instead. """) return - # If we get here, try to load partitions try: import pyarrow.dataset as ds @@ -54,14 +51,13 @@ def main(): if not parquet_files: print(f" ⚠ No .parquet files found in {table_dir}") - print(" Partitions may be in native QuestDB format, not Parquet.") return print(f" ✓ Found {len(parquet_files)} partition files") # Load as dataset dataset = ds.dataset(table_dir, format="parquet") - print(f" ✓ Loaded dataset with schema:") + print(f" ✓ Schema:") for field in dataset.schema: print(f" - {field.name}: {field.type}") @@ -76,13 +72,13 @@ def main(): print("\n" + "=" * 60) print("Summary:") - print(" - Direct partition access bypasses the query engine") + print(" - Direct partition access bypasses query engine") print(" - PyArrow dataset API enables efficient streaming") print(" - Best for large-scale batch processing") print("=" * 60) except Exception as e: - print(f" ⚠ Error loading partitions: {e}") + print(f" ⚠ Error: {e}") if __name__ == "__main__": From 1ca274e8d89e4613efae6f95d8e7213ee322a9b1 Mon Sep 17 00:00:00 2001 From: javier Date: Tue, 7 Apr 2026 18:22:15 +0200 Subject: [PATCH 4/5] renaming folder --- .../README.md | 0 .../config/__init__.py | 0 .../config/settings.py | 0 .../dataset/.gitkeep | 0 .../outputs/.gitkeep | 0 .../pyproject.toml | 0 .../requirements.txt | 0 .../src/__init__.py | 0 .../src/data_loader.py | 0 .../src/forecaster.py | 0 .../steps/__init__.py | 0 .../steps/forecast.py | 0 .../steps/method1_direct_query.py | 0 .../steps/method2_parquet_export.py | 0 .../steps/method3_parquet_partitions.py | 0 15 files changed, 0 insertions(+), 0 deletions(-) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/README.md (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/config/__init__.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/config/settings.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/dataset/.gitkeep (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/outputs/.gitkeep (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/pyproject.toml (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/requirements.txt (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/src/__init__.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/src/data_loader.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/src/forecaster.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/steps/__init__.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/steps/forecast.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/steps/method1_direct_query.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/steps/method2_parquet_export.py (100%) rename {2026-04-time-series-forecasting-pytorch-timesfm => 2026-04-time-series-forecasting-timesfm}/steps/method3_parquet_partitions.py (100%) diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/README.md b/2026-04-time-series-forecasting-timesfm/README.md similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/README.md rename to 2026-04-time-series-forecasting-timesfm/README.md diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/config/__init__.py b/2026-04-time-series-forecasting-timesfm/config/__init__.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/config/__init__.py rename to 2026-04-time-series-forecasting-timesfm/config/__init__.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/config/settings.py b/2026-04-time-series-forecasting-timesfm/config/settings.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/config/settings.py rename to 2026-04-time-series-forecasting-timesfm/config/settings.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/dataset/.gitkeep b/2026-04-time-series-forecasting-timesfm/dataset/.gitkeep similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/dataset/.gitkeep rename to 2026-04-time-series-forecasting-timesfm/dataset/.gitkeep diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/outputs/.gitkeep b/2026-04-time-series-forecasting-timesfm/outputs/.gitkeep similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/outputs/.gitkeep rename to 2026-04-time-series-forecasting-timesfm/outputs/.gitkeep diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/pyproject.toml b/2026-04-time-series-forecasting-timesfm/pyproject.toml similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/pyproject.toml rename to 2026-04-time-series-forecasting-timesfm/pyproject.toml diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/requirements.txt b/2026-04-time-series-forecasting-timesfm/requirements.txt similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/requirements.txt rename to 2026-04-time-series-forecasting-timesfm/requirements.txt diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/src/__init__.py b/2026-04-time-series-forecasting-timesfm/src/__init__.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/src/__init__.py rename to 2026-04-time-series-forecasting-timesfm/src/__init__.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/src/data_loader.py b/2026-04-time-series-forecasting-timesfm/src/data_loader.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/src/data_loader.py rename to 2026-04-time-series-forecasting-timesfm/src/data_loader.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/src/forecaster.py b/2026-04-time-series-forecasting-timesfm/src/forecaster.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/src/forecaster.py rename to 2026-04-time-series-forecasting-timesfm/src/forecaster.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/steps/__init__.py b/2026-04-time-series-forecasting-timesfm/steps/__init__.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/steps/__init__.py rename to 2026-04-time-series-forecasting-timesfm/steps/__init__.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/steps/forecast.py b/2026-04-time-series-forecasting-timesfm/steps/forecast.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/steps/forecast.py rename to 2026-04-time-series-forecasting-timesfm/steps/forecast.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/steps/method1_direct_query.py b/2026-04-time-series-forecasting-timesfm/steps/method1_direct_query.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/steps/method1_direct_query.py rename to 2026-04-time-series-forecasting-timesfm/steps/method1_direct_query.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/steps/method2_parquet_export.py b/2026-04-time-series-forecasting-timesfm/steps/method2_parquet_export.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/steps/method2_parquet_export.py rename to 2026-04-time-series-forecasting-timesfm/steps/method2_parquet_export.py diff --git a/2026-04-time-series-forecasting-pytorch-timesfm/steps/method3_parquet_partitions.py b/2026-04-time-series-forecasting-timesfm/steps/method3_parquet_partitions.py similarity index 100% rename from 2026-04-time-series-forecasting-pytorch-timesfm/steps/method3_parquet_partitions.py rename to 2026-04-time-series-forecasting-timesfm/steps/method3_parquet_partitions.py From fdeb3f23bd06c3507eb9ca8a607e1d3b0a95fe15 Mon Sep 17 00:00:00 2001 From: javier Date: Tue, 7 Apr 2026 18:23:11 +0200 Subject: [PATCH 5/5] removing pytorch name --- 2026-04-time-series-forecasting-timesfm/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/2026-04-time-series-forecasting-timesfm/README.md b/2026-04-time-series-forecasting-timesfm/README.md index 8f3f6f5..f9dee39 100644 --- a/2026-04-time-series-forecasting-timesfm/README.md +++ b/2026-04-time-series-forecasting-timesfm/README.md @@ -1,4 +1,4 @@ -# QuestDB → PyTorch: Time-Series Data Loading & Forecasting +# QuestDB → TimesFM: Time-Series Data Loading & Forecasting This tutorial demonstrates how to efficiently load time-series data from QuestDB into Python, then run forecasts using Google's TimesFM foundation model.