Data·Data Storage·Beginner

Store in Timescaledb

Store and manage large-scale time series market data using TimescaleDB hypertables with automatic chunk partitioning, continuous aggregates for downsampled materialized views, and automated retention and compression policies.

data-engineeringdata-storage

Data Storage Framework — TimescaleDB

This notebook defines a standardized protocol for persisting cleaned OHLCV data into a TimescaleDB hypertable. It covers hypertable creation, chunk interval configuration, datetime-indexed insertion, compression policy definition, and read-back verification.

1. Dependency Installation

[1]

!pip install pandas sqlalchemy psycopg2-binary

Requirement already satisfied: pandas in /usr/local/lib/python3.12/dist-packages (2.2.2)
Requirement already satisfied: sqlalchemy in /usr/local/lib/python3.12/dist-packages (2.0.50)
Collecting psycopg2-binary
  Downloading psycopg2_binary-2.9.12-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl.metadata (4.9 kB)
Requirement already satisfied: numpy>=1.26.0 in /usr/local/lib/python3.12/dist-packages (from pandas) (2.0.2)
Requirement already satisfied: python-dateutil>=2.8.2 in /usr/local/lib/python3.12/dist-packages (from pandas) (2.9.0.post0)
Requirement already satisfied: pytz>=2020.1 in /usr/local/lib/python3.12/dist-packages (from pandas) (2025.2)
Requirement already satisfied: tzdata>=2022.7 in /usr/local/lib/python3.12/dist-packages (from pandas) (2026.2)
Requirement already satisfied: greenlet>=1 in /usr/local/lib/python3.12/dist-packages (from sqlalchemy) (3.5.1)
Requirement already satisfied: typing-extensions>=4.6.0 in /usr/local/lib/python3.12/dist-packages (from sqlalchemy) (4.15.0)
Requirement already satisfied: six>=1.5 in /usr/local/lib/python3.12/dist-packages (from python-dateutil>=2.8.2->pandas) (1.17.0)
Downloading psycopg2_binary-2.9.12-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl (4.3 MB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m4.3/4.3 MB[0m [31m12.1 MB/s[0m eta [36m0:00:00[0m
[?25hInstalling collected packages: psycopg2-binary
Successfully installed psycopg2-binary-2.9.12

2. Library Imports

[2]

import warnings
warnings.filterwarnings("ignore")

import pandas as pd
from sqlalchemy import create_engine, text

3. What Is TimescaleDB and Why Use It for Trading Data?

TimescaleDB is an open-source time-series database built as a PostgreSQL extension. It is installed on top of a standard PostgreSQL instance and adds time-series-specific optimizations while remaining fully compatible with standard SQL.

Why TimescaleDB outperforms plain PostgreSQL for OHLCV data:

Feature	PostgreSQL	TimescaleDB
Storage structure	Single table	Hypertable automatically partitioned into time-ordered chunks
Query performance on time ranges	Full table scan without manual partitioning	Only the relevant time chunks are scanned
Compression	Standard TOAST (row-level)	Columnar compression with delta-delta + gorilla encoding — 10–20× smaller than plain PostgreSQL for OHLCV data
Automated maintenance	Manual vacuuming and partitioning	Automated compression and retention policies
Continuous aggregates	Manual materialized views	Native time-bucketed materialized views that auto-refresh

Hypertable: A TimescaleDB hypertable is a standard PostgreSQL table that TimescaleDB automatically partitions into fixed-size time chunks behind the scenes. From the application's perspective it behaves exactly like a regular table — standard SQL INSERT, SELECT, JOIN, and GROUP BY queries work without modification. Internally, each chunk is a separate PostgreSQL table covering a defined time window (e.g., 7 days). Queries that filter on datetime only read the chunks whose window overlaps with the query range, eliminating full table scans.

Chunk interval: The chunk interval defines how much time each internal partition covers. For 1-minute OHLCV data, a 7-day chunk contains approximately 10,080 rows — a size that balances query scan efficiency against chunk management overhead. Smaller chunks reduce scan size per query but increase the number of chunks the planner must manage. Larger chunks reduce chunk count but negate the scan benefit for narrow time-range queries.

Columnar compression: TimescaleDB compresses old chunks using columnar encoding algorithms specifically selected for time-series data. datetime columns are encoded with delta-delta compression (stores the difference between differences — efficient for regular 1-minute intervals). Float columns (open, high, low, close) are encoded with gorilla compression (XOR-based, efficient for slowly changing float sequences). For OHLCV data these algorithms typically achieve 10–20× compression ratios — a 1 GB raw table compresses to 50–100 MB.

4. Dummy Dataset

[3]

raw_data = {
    "datetime": [
        "2024-01-01 00:00:00+00:00",
        "2024-01-01 00:01:00+00:00",
        "2024-01-01 00:02:00+00:00",
        "2024-01-01 00:03:00+00:00",
        "2024-01-01 00:04:00+00:00",
    ],
    "open":   [42100.0, 42200.0, 42150.0, 42300.0, 42250.0],
    "high":   [42300.0, 42400.0, 42350.0, 42500.0, 42450.0],
    "low":    [41900.0, 42000.0, 41950.0, 42100.0, 42050.0],
    "close":  [42200.0, 42150.0, 42300.0, 42250.0, 42400.0],
    "volume": [10.5,    8.2,     9.1,     11.3,    7.6    ],
}

df = pd.DataFrame(raw_data)
df["datetime"] = pd.to_datetime(df["datetime"], utc=True)

print("--- OHLCV DataFrame ---")
display(df)
df.info()

--- OHLCV DataFrame ---

	datetime	open	high	low	close	volume
0	2024-01-01 00:00:00+00:00	42100.0	42300.0	41900.0	42200.0	10.5
1	2024-01-01 00:01:00+00:00	42200.0	42400.0	42000.0	42150.0	8.2
2	2024-01-01 00:02:00+00:00	42150.0	42350.0	41950.0	42300.0	9.1
3	2024-01-01 00:03:00+00:00	42300.0	42500.0	42100.0	42250.0	11.3
4	2024-01-01 00:04:00+00:00	42250.0	42450.0	42050.0	42400.0	7.6

<class 'pandas.core.frame.DataFrame'>
RangeIndex: 5 entries, 0 to 4
Data columns (total 6 columns):
 #   Column    Non-Null Count  Dtype              
---  ------    --------------  -----              
 0   datetime  5 non-null      datetime64[ns, UTC]
 1   open      5 non-null      float64            
 2   high      5 non-null      float64            
 3   low       5 non-null      float64            
 4   close     5 non-null      float64            
 5   volume    5 non-null      float64            
dtypes: datetime64[ns, UTC](1), float64(5)
memory usage: 372.0 bytes

5. Database Configuration

[4]

# Replace with your db cofigrations.
DB_HOST = "localhost"
DB_PORT = 5432
DB_NAME = "bitpredict"
DB_USER = "postgres"
DB_PASS = "neuroglia"
SCHEMA  = "data_ohlcv"
TABLE   = "ohlcv_btcusdt_1m"

DB_URL  = f"postgresql+psycopg2://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
engine  = create_engine(DB_URL, echo=False)

Code Logic

TimescaleDB is a PostgreSQL extension — the connection URL, driver (psycopg2), and engine configuration are identical to a standard PostgreSQL connection. No TimescaleDB-specific driver is required.

6. Hypertable Initialization Function

[5]

def initialize_hypertable(engine, schema: str, table: str) -> None:
    ddl = f"""
        CREATE SCHEMA IF NOT EXISTS {schema};

        CREATE TABLE IF NOT EXISTS {schema}.{table} (
            datetime  TIMESTAMPTZ      NOT NULL,
            open      DOUBLE PRECISION NOT NULL,
            high      DOUBLE PRECISION NOT NULL,
            low       DOUBLE PRECISION NOT NULL,
            close     DOUBLE PRECISION NOT NULL,
            volume    DOUBLE PRECISION NOT NULL
        );

        SELECT create_hypertable(
            '{schema}.{table}',
            'datetime',
            chunk_time_interval => INTERVAL '7 days',
            if_not_exists       => TRUE
        );

        CREATE UNIQUE INDEX IF NOT EXISTS {table}_datetime_idx
            ON {schema}.{table} (datetime ASC);

        ALTER TABLE {schema}.{table}
            SET (
                timescaledb.compress,
                timescaledb.compress_orderby   = 'datetime ASC',
                timescaledb.compress_segmentby = ''
            );

        SELECT add_compression_policy(
            '{schema}.{table}',
            INTERVAL '7 days',
            if_not_exists => TRUE
        );
    """

    with engine.begin() as conn:
        conn.execute(text(ddl))

    print(f"Hypertable '{schema}.{table}' initialized with 7-day chunks and compression policy.")

Code Logic

CREATE TABLE IF NOT EXISTS: Defines the base relation. TimescaleDB requires a standard PostgreSQL table to exist before converting it to a hypertable.
TIMESTAMPTZ NOT NULL: TimescaleDB requires the partition column to be a non-null timezone-aware timestamp. All inserted values are normalized to UTC internally.
create_hypertable(...): TimescaleDB catalog function that converts the base table into a partitioned hypertable. See Section 3 for the full definition of hypertables and chunk intervals. if_not_exists => TRUE makes the call idempotent — safe on repeated execution.
chunk_time_interval => INTERVAL '7 days': Sets a 7-day chunk window. See Section 3 for chunk interval rationale.
CREATE UNIQUE INDEX ... (datetime ASC): Enforces row uniqueness on datetime. TimescaleDB discourages PRIMARY KEY on the partition column for performance reasons — a unique index achieves the same integrity guarantee without the overhead.
timescaledb.compress: Enables columnar compression on the hypertable. See Section 3 for compression algorithm definitions.
timescaledb.compress_orderby = 'datetime ASC': Defines the sort order within each compressed chunk. Ascending datetime ordering maximizes delta-delta compression efficiency for regular time intervals.
timescaledb.compress_segmentby = '': No segmentation column is used. Segmentation groups rows by a categorical column (e.g., symbol) within each chunk — for a single-symbol table this is unnecessary.
add_compression_policy(..., INTERVAL '7 days'): Registers a background job that automatically compresses chunks older than 7 days. Without this policy, compression must be triggered manually per chunk.
engine.begin(): All DDL and TimescaleDB catalog modifications are committed atomically on context exit.

7. Save Function

[6]

def save_to_timescaledb(df: pd.DataFrame, engine, schema: str, table: str) -> int:
    df = df.copy().set_index("datetime")

    df.to_sql(
        name        = table,
        con         = engine,
        schema      = schema,
        if_exists   = "append",
        index       = True,
        index_label = "datetime",
        method      = "multi",
    )

    print(f"Inserted {len(df)} row(s) → {schema}.{table}")
    return len(df)

Code Logic

df.set_index("datetime"): Promotes datetime to the DataFrame index. TimescaleDB routes each inserted row to its correct time chunk based on the value of the partition column — index_label="datetime" ensures the index is written as that column.
if_exists="append": Appends rows without truncation. Duplicate datetime values are rejected by the unique index.
method="multi": Emits a single multi-row INSERT per batch, reducing client-server round trips during bulk ingestion.

8. Execution

[7]

initialize_hypertable(engine, SCHEMA, TABLE)
rows_inserted = save_to_timescaledb(df, engine, SCHEMA, TABLE)

---------------------------------------------------------------------------
OperationalError                          Traceback (most recent call last)
/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/base.py in __init__(self, engine, connection, _has_events, _allow_revalidate, _allow_autobegin)
    143             try:
--> 144                 self._dbapi_connection = engine.raw_connection()
    145             except dialect.loaded_dbapi.Error as err:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/base.py in raw_connection(self)
   3318         """
-> 3319         return self.pool.connect()
   3320 

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in connect(self)
    447         """
--> 448         return _ConnectionFairy._checkout(self)
    449 

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in _checkout(cls, pool, threadconns, fairy)
   1271         if not fairy:
-> 1272             fairy = _ConnectionRecord.checkout(pool)
   1273 

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in checkout(cls, pool)
    711         else:
--> 712             rec = pool._do_get()
    713 

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/impl.py in _do_get(self)
    177             except:
--> 178                 with util.safe_reraise():
    179                     self._dec_overflow()

/usr/local/lib/python3.12/dist-packages/sqlalchemy/util/langhelpers.py in __exit__(self, type_, value, traceback)
    121             self._exc_info = None  # remove potential circular references
--> 122             raise exc_value.with_traceback(exc_tb)
    123         else:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/impl.py in _do_get(self)
    175             try:
--> 176                 return self._create_connection()
    177             except:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in _create_connection(self)
    388 
--> 389         return _ConnectionRecord(self)
    390 

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in __init__(self, pool, connect)
    673         if connect:
--> 674             self.__connect()
    675         self.finalize_callback = deque()

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in __connect(self)
    899         except BaseException as e:
--> 900             with util.safe_reraise():
    901                 pool.logger.debug("Error on connect(): %s", e)

/usr/local/lib/python3.12/dist-packages/sqlalchemy/util/langhelpers.py in __exit__(self, type_, value, traceback)
    121             self._exc_info = None  # remove potential circular references
--> 122             raise exc_value.with_traceback(exc_tb)
    123         else:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in __connect(self)
    895             self.starttime = time.time()
--> 896             self.dbapi_connection = connection = pool._invoke_creator(self)
    897             pool.logger.debug("Created new connection %r", connection)

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/create.py in connect(connection_record)
    666             else:
--> 667                 return dialect.connect(*cargs_tup, **cparams)
    668 

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/default.py in connect(self, *cargs, **cparams)
    629         # inherits the docstring from interfaces.Dialect.connect
--> 630         return self.loaded_dbapi.connect(*cargs, **cparams)  # type: ignore[no-any-return]  # NOQA: E501
    631 

/usr/local/lib/python3.12/dist-packages/psycopg2/__init__.py in connect(dsn, connection_factory, cursor_factory, **kwargs)
    121     dsn = _ext.make_dsn(dsn, **kwargs)
--> 122     conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
    123     if cursor_factory is not None:

OperationalError: connection to server at "localhost" (::1), port 5432 failed: Connection refused
	Is the server running on that host and accepting TCP/IP connections?
connection to server at "localhost" (127.0.0.1), port 5432 failed: Connection refused
	Is the server running on that host and accepting TCP/IP connections?


The above exception was the direct cause of the following exception:

OperationalError                          Traceback (most recent call last)
/tmp/ipykernel_6619/578533995.py in <cell line: 0>()
----> 1 initialize_hypertable(engine, SCHEMA, TABLE)
      2 rows_inserted = save_to_timescaledb(df, engine, SCHEMA, TABLE)

/tmp/ipykernel_6619/1723711135.py in initialize_hypertable(engine, schema, table)
     36     """
     37 
---> 38     with engine.begin() as conn:
     39         conn.execute(text(ddl))
     40 

/usr/lib/python3.12/contextlib.py in __enter__(self)
    135         del self.args, self.kwds, self.func
    136         try:
--> 137             return next(self.gen)
    138         except StopIteration:
    139             raise RuntimeError("generator didn't yield") from None

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/base.py in begin(self)
   3257 
   3258         """  # noqa: E501
-> 3259         with self.connect() as conn:
   3260             with conn.begin():
   3261                 yield conn

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/base.py in connect(self)
   3293         """
   3294 
-> 3295         return self._connection_cls(self)
   3296 
   3297     def raw_connection(self) -> PoolProxiedConnection:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/base.py in __init__(self, engine, connection, _has_events, _allow_revalidate, _allow_autobegin)
    144                 self._dbapi_connection = engine.raw_connection()
    145             except dialect.loaded_dbapi.Error as err:
--> 146                 Connection._handle_dbapi_exception_noconnection(
    147                     err, dialect, engine
    148                 )

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/base.py in _handle_dbapi_exception_noconnection(cls, e, dialect, engine, is_disconnect, invalidate_pool_on_disconnect, is_pre_ping)
   2448         elif should_wrap:
   2449             assert sqlalchemy_exception is not None
-> 2450             raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
   2451         else:
   2452             assert exc_info[1] is not None

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/base.py in __init__(self, engine, connection, _has_events, _allow_revalidate, _allow_autobegin)
    142         if connection is None:
    143             try:
--> 144                 self._dbapi_connection = engine.raw_connection()
    145             except dialect.loaded_dbapi.Error as err:
    146                 Connection._handle_dbapi_exception_noconnection(

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/base.py in raw_connection(self)
   3317 
   3318         """
-> 3319         return self.pool.connect()
   3320 
   3321 

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in connect(self)
    446 
    447         """
--> 448         return _ConnectionFairy._checkout(self)
    449 
    450     def _return_conn(self, record: ConnectionPoolEntry) -> None:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in _checkout(cls, pool, threadconns, fairy)
   1270     ) -> _ConnectionFairy:
   1271         if not fairy:
-> 1272             fairy = _ConnectionRecord.checkout(pool)
   1273 
   1274             if threadconns is not None:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in checkout(cls, pool)
    710             rec = cast(_ConnectionRecord, pool._do_get())
    711         else:
--> 712             rec = pool._do_get()
    713 
    714         try:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/impl.py in _do_get(self)
    176                 return self._create_connection()
    177             except:
--> 178                 with util.safe_reraise():
    179                     self._dec_overflow()
    180                 raise

/usr/local/lib/python3.12/dist-packages/sqlalchemy/util/langhelpers.py in __exit__(self, type_, value, traceback)
    120             assert exc_value is not None
    121             self._exc_info = None  # remove potential circular references
--> 122             raise exc_value.with_traceback(exc_tb)
    123         else:
    124             self._exc_info = None  # remove potential circular references

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/impl.py in _do_get(self)
    174         if self._inc_overflow():
    175             try:
--> 176                 return self._create_connection()
    177             except:
    178                 with util.safe_reraise():

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in _create_connection(self)
    387         """Called by subclasses to create a new ConnectionRecord."""
    388 
--> 389         return _ConnectionRecord(self)
    390 
    391     def _invalidate(

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in __init__(self, pool, connect)
    672         self.__pool = pool
    673         if connect:
--> 674             self.__connect()
    675         self.finalize_callback = deque()
    676 

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in __connect(self)
    898             self.fresh = True
    899         except BaseException as e:
--> 900             with util.safe_reraise():
    901                 pool.logger.debug("Error on connect(): %s", e)
    902         else:

/usr/local/lib/python3.12/dist-packages/sqlalchemy/util/langhelpers.py in __exit__(self, type_, value, traceback)
    120             assert exc_value is not None
    121             self._exc_info = None  # remove potential circular references
--> 122             raise exc_value.with_traceback(exc_tb)
    123         else:
    124             self._exc_info = None  # remove potential circular references

/usr/local/lib/python3.12/dist-packages/sqlalchemy/pool/base.py in __connect(self)
    894         try:
    895             self.starttime = time.time()
--> 896             self.dbapi_connection = connection = pool._invoke_creator(self)
    897             pool.logger.debug("Created new connection %r", connection)
    898             self.fresh = True

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/create.py in connect(connection_record)
    665                 return dialect.connect(*mutable_cargs, **mutable_cparams)
    666             else:
--> 667                 return dialect.connect(*cargs_tup, **cparams)
    668 
    669         creator = pop_kwarg("creator", connect)

/usr/local/lib/python3.12/dist-packages/sqlalchemy/engine/default.py in connect(self, *cargs, **cparams)
    628     def connect(self, *cargs: Any, **cparams: Any) -> DBAPIConnection:
    629         # inherits the docstring from interfaces.Dialect.connect
--> 630         return self.loaded_dbapi.connect(*cargs, **cparams)  # type: ignore[no-any-return]  # NOQA: E501
    631 
    632     def create_connect_args(self, url: URL) -> ConnectArgsType:

/usr/local/lib/python3.12/dist-packages/psycopg2/__init__.py in connect(dsn, connection_factory, cursor_factory, **kwargs)
    120 
    121     dsn = _ext.make_dsn(dsn, **kwargs)
--> 122     conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
    123     if cursor_factory is not None:
    124         conn.cursor_factory = cursor_factory

OperationalError: (psycopg2.OperationalError) connection to server at "localhost" (::1), port 5432 failed: Connection refused
	Is the server running on that host and accepting TCP/IP connections?
connection to server at "localhost" (127.0.0.1), port 5432 failed: Connection refused
	Is the server running on that host and accepting TCP/IP connections?

(Background on this error at: https://sqlalche.me/e/20/e3q8)

9. Verification — Read Back and Inspect

[ ]

query = f"SELECT * FROM {SCHEMA}.{TABLE} ORDER BY datetime ASC;"
df_db = pd.read_sql(text(query), con=engine.connect(), parse_dates=["datetime"])

print("--- TimescaleDB Read-Back ---")
display(df_db.head())

print("\n--- Schema Summary ---")
df_db.info()

10. Chunk and Compression Inspection

[ ]

chunk_query = f"""
    SELECT
        chunk_name,
        range_start,
        range_end,
        is_compressed
    FROM timescaledb_information.chunks
    WHERE hypertable_schema = '{SCHEMA}'
      AND hypertable_name   = '{TABLE}'
    ORDER BY range_start ASC;
"""

df_chunks = pd.read_sql(text(chunk_query), con=engine.connect())

print("--- Hypertable Chunk Map ---")
display(df_chunks)

Code Logic

timescaledb_information.chunks: TimescaleDB system catalog view exposing per-chunk metadata for all registered hypertables.
range_start / range_end: The TIMESTAMPTZ boundaries of each 7-day chunk window — confirms the chunk_time_interval was applied correctly during initialization.
is_compressed: Boolean flag per chunk. Newly inserted chunks display False until the background compression policy fires or compression is triggered manually via SELECT compress_chunk(...).

Conclusion

This notebook demonstrates the setup and usage of TimescaleDB for efficiently storing and querying OHLCV data. By leveraging hypertables, chunking, and columnar compression, TimescaleDB provides a robust and performant solution for time-series data management, significantly outperforming standard PostgreSQL for this specific use case.