chpy-orm 0.1.0

A Python wrapper for ClickHouse database operations

chpy-orm - ClickHouse Python Wrapper

A comprehensive Python library for programmatically querying and managing ClickHouse databases. chpy-orm provides a fluent, ORM-style interface with type safety, autocomplete support, and extensive ClickHouse function coverage.

Acknowledgments

This project is developed mainly using Cursor, an AI-powered code editor, but all code is thoroughly tested and verified by the maintainer to ensure quality and reliability.

Features

• 🚀 Fluent Query Builder - Method chaining for building complex queries
• 🔒 Type-Safe Columns - ORM-style column objects with autocomplete support
• 📊 Multiple Output Formats - DataFrame, NumPy, JSON, CSV, Parquet, and more
• 🛠️ DDL Operations - High-level table and database management
• 🎯 Comprehensive Functions - Full coverage of ClickHouse functions
• 🔄 Window Functions - Advanced window function support with OVER clauses
• 🔗 JOIN Support - INNER, LEFT, RIGHT, FULL, and CROSS joins
• 📦 Type System - Support for all ClickHouse data types (Arrays, Maps, Tuples, Nested, etc.)
• 🎨 Generic Table Wrapper - Works with any ClickHouse table, not just crypto_quotes

Installation

pip install chpy-orm

Or install in development mode:

pip install -e .

Requirements

• Python 3.8+
• clickhouse-connect
• pandas (optional, for DataFrame support)
• numpy (optional, for NumPy array support)

Quick Start

Basic Connection

from chpy import ClickHouseClient, CryptoQuotesTable

# Initialize the client
client = ClickHouseClient(
    host="localhost",
    port=8123,
    username="default",
    password="",
    database="stockhouse"
)

# Create a table wrapper
table = CryptoQuotesTable(client)

# Simple query
df = (table.query()
    .where(table.c.pair == "BTC-USDT")
    .limit(10)
    .to_dataframe())

print(df)

Using Context Manager

with ClickHouseClient(host="localhost", database="stockhouse") as client:
    table = CryptoQuotesTable(client)
    results = table.query().where(table.c.pair == "BTC-USDT").to_list()
    # Connection automatically closed

---

Basic Examples

Example 1: Simple Query with Column Objects

from chpy import ClickHouseClient, CryptoQuotesTable, crypto_quotes
from datetime import datetime, timedelta

client = ClickHouseClient(host="localhost", database="stockhouse")
table = CryptoQuotesTable(client)

# Query using column objects (type-safe with autocomplete)
df = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        crypto_quotes.best_ask_price,
        crypto_quotes.timestamp_ms
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=1))
    .order_by(crypto_quotes.timestamp_ms, desc=True)
    .limit(100)
    .to_dataframe())

print(df.head())

Example 2: Using Table Shortcuts

# Use table.c or table.columns for column access
results = (table.query()
    .where(table.c.pair == "BTC-USDT")
    .where(table.c.best_bid_price > 50000)
    .order_by(table.c.timestamp_ms, desc=True)
    .limit(5)
    .to_list())

for row in results:
    print(f"Pair: {row['pair']}, Bid: {row['best_bid_price']}")

Example 3: Comparison Operators

# All comparison operators are supported
results = (table.query()
    .where(crypto_quotes.best_bid_price > 50000)
    .where(crypto_quotes.best_bid_price < 60000)
    .where(crypto_quotes.best_bid_price >= 51000)
    .where(crypto_quotes.best_bid_price <= 59000)
    .where(crypto_quotes.pair != "ETH-USDT")
    .limit(10)
    .to_list())

Example 4: IN and NOT IN Operators

# IN operator for multiple values
results = (table.query()
    .where(crypto_quotes.pair.in_(["BTC-USDT", "ETH-USDT", "BNB-USDT"]))
    .where(crypto_quotes.exchange == "BINANCE")
    .to_list())

# NOT IN operator
results = (table.query()
    .where(crypto_quotes.exchange.not_in(["BINANCE", "KUCOIN"]))
    .limit(10)
    .to_list())

Example 5: LIKE Operator

# Pattern matching with LIKE
results = (table.query()
    .where(crypto_quotes.pair.like("BTC-%"))
    .limit(10)
    .to_list())

Example 6: Complex Expressions with AND/OR

# Combine conditions with & (AND) and | (OR)
results = (table.query()
    .where(
        (crypto_quotes.pair == "BTC-USDT") & 
        (crypto_quotes.exchange.in_(["BINANCE", "KUCOIN"])) &
        (crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=1))
    )
    .limit(10)
    .to_list())

# OR expression
results = (table.query()
    .where(
        (crypto_quotes.pair == "BTC-USDT") | 
        (crypto_quotes.pair == "ETH-USDT")
    )
    .where(crypto_quotes.exchange == "BINANCE")
    .limit(10)
    .to_list())

Example 7: Multiple WHERE Clauses

# Multiple where() calls are combined with AND
results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .where(crypto_quotes.best_bid_price > 50000)
    .where(crypto_quotes.best_ask_price < 60000)
    .limit(5)
    .to_list())

Example 8: Selecting Specific Columns

# Select only the columns you need
df = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        crypto_quotes.best_ask_price,
        crypto_quotes.exchange,
        crypto_quotes.timestamp_ms
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(10)
    .to_dataframe())

print(df.columns)  # Only selected columns

Example 9: Ordering Results

# Order by one or more columns
results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .order_by(crypto_quotes.exchange, desc=False)  # ASC
    .order_by(crypto_quotes.timestamp_ms, desc=True)  # DESC
    .limit(10)
    .to_list())

Example 10: Counting Rows

# Count rows matching conditions
count = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=1))
    .count())

print(f"Found {count} rows")

Example 11: Getting First Result

# Get the first matching row
first = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .order_by(crypto_quotes.timestamp_ms, desc=True)
    .first())

if first:
    print(f"Latest quote: {first['pair']} @ {first['best_bid_price']}")

Example 12: Checking Existence

# Check if any rows match
exists = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .exists())

print(f"Data exists: {exists}")

Example 13: Iterating Over Results

# Iterate over results (lazy evaluation)
for row in table.query().where(crypto_quotes.pair == "BTC-USDT").limit(10):
    print(f"Pair: {row['pair']}, Bid: {row['best_bid_price']}")

Example 14: Output Formats

# List of dictionaries
results = table.query().where(crypto_quotes.pair == "BTC-USDT").to_list()

# Pandas DataFrame
df = table.query().where(crypto_quotes.pair == "BTC-USDT").to_dataframe()

# NumPy array
arr = table.query().select(
    crypto_quotes.best_bid_price,
    crypto_quotes.best_ask_price
).to_numpy()

# JSON string
json_str = table.query().limit(10).to_json(indent=2)

# CSV string
csv_str = table.query().limit(100).to_csv()

# Write to file
table.query().limit(10000).to_csv("quotes.csv")
table.query().limit(10000).to_parquet("quotes.parquet")

# Dictionary (key-value mapping)
pair_dict = (table.query()
    .select(crypto_quotes.pair, crypto_quotes.best_bid_price)
    .to_dict(crypto_quotes.pair, crypto_quotes.best_bid_price))

---

Intermediate Examples

Example 15: Grouping and Aggregation

from chpy.functions import avg, count, min, max, sum

# Group by columns and aggregate
results = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.exchange,
        avg(crypto_quotes.best_bid_price).alias("avg_bid"),
        min(crypto_quotes.best_bid_price).alias("min_bid"),
        max(crypto_quotes.best_bid_price).alias("max_bid"),
        count().alias("cnt")
    )
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=7))
    .group_by(crypto_quotes.pair, crypto_quotes.exchange)
    .having("avg_bid > 0")
    .order_by("avg_bid", desc=True)
    .limit(10)
    .to_list())

for row in results:
    print(f"{row['pair']} on {row['exchange']}: "
          f"avg={row['avg_bid']:.2f}, count={row['cnt']}")

Example 16: String Functions

from chpy.functions import length, upper, lower, substring, concat, startsWith, endsWith

# Use string functions in SELECT
results = (table.query()
    .select(
        crypto_quotes.pair,
        length(crypto_quotes.pair).alias("pair_length"),
        upper(crypto_quotes.pair).alias("pair_upper"),
        lower(crypto_quotes.exchange).alias("exchange_lower"),
        substring(crypto_quotes.pair, 1, 3).alias("pair_prefix"),
        concat(crypto_quotes.pair, " on ", crypto_quotes.exchange).alias("description")
    )
    .where(crypto_quotes.pair.in_(["BTC-USDT", "ETH-USDT"]))
    .limit(5)
    .to_list())

Example 17: Date and Time Functions

from chpy.functions import toYear, toMonth, toDayOfMonth, toHour, toDateTime, divide

# Extract date/time components
results = (table.query()
    .select(
        crypto_quotes.timestamp_ms,
        toYear(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("year"),
        toMonth(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("month"),
        toDayOfMonth(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("day"),
        toHour(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("hour")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(5)
    .to_list())

Example 18: Mathematical Functions

from chpy.functions import abs, sqrt, round, floor, ceil

# Mathematical transformations
results = (table.query()
    .select(
        crypto_quotes.best_bid_price,
        crypto_quotes.best_ask_price,
        abs(crypto_quotes.best_bid_price).alias("abs_bid"),
        sqrt(crypto_quotes.best_bid_price).alias("sqrt_bid"),
        round(crypto_quotes.best_bid_price, 2).alias("rounded_bid"),
        floor(crypto_quotes.best_bid_price).alias("floor_bid"),
        ceil(crypto_quotes.best_ask_price).alias("ceil_ask")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.best_bid_price > 0)
    .limit(5)
    .to_list())

Example 19: Conditional Functions

from chpy.functions import if_ as if_func, coalesce

# Conditional logic
results = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        crypto_quotes.best_ask_price,
        if_func(
            crypto_quotes.best_bid_price > 50000,
            "high",
            "normal"
        ).alias("price_category"),
        coalesce(crypto_quotes.best_bid_price, 0).alias("safe_bid")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(5)
    .to_list())

Example 20: Type Conversion Functions

from chpy.functions import toString, toInt64, toFloat64

# Convert types
results = (table.query()
    .select(
        crypto_quotes.pair,
        toString(crypto_quotes.best_bid_price).alias("bid_as_string"),
        toInt64(crypto_quotes.best_bid_price).alias("bid_as_int"),
        toFloat64(crypto_quotes.best_bid_price).alias("bid_as_float")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(3)
    .to_list())

Example 21: Combining Multiple Functions

# Combine different function types in one query
results = (table.query()
    .select(
        upper(crypto_quotes.pair).alias("pair_upper"),
        length(crypto_quotes.pair).alias("pair_length"),
        round(crypto_quotes.best_bid_price, 2).alias("rounded_price"),
        toYear(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("year"),
        if_func(
            crypto_quotes.best_bid_price > 50000,
            "premium",
            "standard"
        ).alias("tier")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(3)
    .to_list())

Example 22: Raw SQL Conditions

# Use raw SQL for complex conditions
results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where("best_bid_price > 50000 AND best_ask_price < 60000")
    .where("timestamp_ms >= toUnixTimestamp(now()) * 1000 - 86400000")
    .limit(10)
    .to_list())

Example 23: Working with Generic Tables

from chpy import ClickHouseClient, TableWrapper
from chpy.orm import Table, Column

# Create a schema for any table
columns = [
    Column("id", "UInt64"),
    Column("name", "String"),
    Column("value", "Float64"),
    Column("created_at", "DateTime")
]
schema = Table("my_table", "my_db", columns)

# Create wrapper with schema
client = ClickHouseClient(host="localhost", database="my_db")
table = TableWrapper(client, "my_table", "my_db", schema=schema)

# Query with type-safe columns
df = (table.query()
    .where(schema.id > 100)
    .where(schema.name == "test")
    .to_dataframe())

# Or query without schema (using raw strings)
df = (table.query()
    .where("id > 100")
    .where("name = 'test'")
    .to_dataframe())

# Insert data
table.insert([
    {"id": 1, "name": "test", "value": 1.5, "created_at": datetime.now()}
])

---

Advanced Examples

Example 24: JOIN Operations

from chpy.orm import Table, Column

# Define another table schema
other_table_columns = [
    Column("symbol", "String"),
    Column("name", "String"),
    Column("market_cap", "Float64"),
]
other_table = Table("market_data", "stockhouse", other_table_columns)

# INNER JOIN with column expressions
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        other_table.name,
        other_table.market_cap
    )
    .join(
        other_table,
        condition=(crypto_quotes.pair == other_table.symbol),
        join_type="INNER"
    )
    .where(crypto_quotes.exchange == "BINANCE")
    .to_dataframe())

# LEFT JOIN with alias
result = (table.query()
    .select(
        crypto_quotes.pair,
        other_table.market_cap
    )
    .join(
        other_table,
        condition=(crypto_quotes.pair == other_table.symbol),
        join_type="LEFT",
        alias="md"
    )
    .to_list())

# Multiple JOINs
result = (table.query()
    .join(other_table, condition=(crypto_quotes.pair == other_table.symbol))
    .join(
        "stockhouse.exchanges",
        condition="crypto_quotes.exchange = exchanges.code",
        join_type="LEFT"
    )
    .to_dataframe())

# CROSS JOIN (no condition needed)
result = (table.query()
    .join("stockhouse.reference_table", join_type="CROSS")
    .limit(100)
    .to_list())

Example 25: Window Functions

from chpy.functions.base import WindowSpec
from chpy.functions.window import rowNumber, rank, denseRank
from chpy.functions.aggregate import avg, sum

# Basic window function with PARTITION BY
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec().partition_by(crypto_quotes.pair)
        ).alias("avg_by_pair")
    )
    .to_dataframe())

# Window function with ORDER BY
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        rank().over(
            WindowSpec()
                .partition_by(crypto_quotes.pair)
                .order_by(crypto_quotes.best_bid_price, desc=True)
        ).alias("price_rank")
    )
    .to_dataframe())

# Running average with frame specification
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.timestamp_ms,
        crypto_quotes.best_bid_price,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec()
                .partition_by(crypto_quotes.pair)
                .order_by(crypto_quotes.timestamp_ms)
                .rows_between("UNBOUNDED PRECEDING", "CURRENT ROW")
        ).alias("running_avg")
    )
    .where(crypto_quotes.exchange == "BINANCE")
    .to_dataframe())

# Multiple window functions in same query
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.exchange,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec().partition_by(crypto_quotes.pair)
        ).alias("avg_by_pair"),
        rowNumber().over(
            WindowSpec()
                .partition_by(crypto_quotes.exchange)
                .order_by(crypto_quotes.timestamp_ms)
        ).alias("row_num")
    )
    .to_dataframe())

# Window functions with GROUP BY
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.exchange,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec().partition_by(crypto_quotes.pair)
        ).alias("window_avg"),
        avg(crypto_quotes.best_bid_price).alias("group_avg")
    )
    .where(crypto_quotes.exchange == "BINANCE")
    .group_by(crypto_quotes.pair, crypto_quotes.exchange)
    .to_dataframe())

Example 26: Subqueries

from chpy.orm import Subquery

# Scalar subquery in SELECT
subquery_builder = (table.query()
    .select(avg(crypto_quotes.best_bid_price))
    .where(crypto_quotes.pair == "BTC-USDT"))

result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        Subquery(subquery_builder).alias("avg_btc_price")
    )
    .where(crypto_quotes.pair == "ETH-USDT")
    .to_list())

# Subquery in WHERE with IN
subquery_builder = (table.query()
    .select(crypto_quotes.pair)
    .where(crypto_quotes.exchange == "BINANCE")
    .group_by(crypto_quotes.pair)
    .having("count() > 100"))

result = (table.query()
    .where(crypto_quotes.pair.in_(Subquery(subquery_builder)))
    .to_list())

# EXISTS subquery
subquery_builder = (table.query()
    .where(crypto_quotes.exchange == "BINANCE"))

result = (table.query()
    .where(Subquery.exists(subquery_builder))
    .to_list())

# Derived table (subquery in FROM)
subquery_builder = (table.query()
    .select(
        crypto_quotes.pair,
        avg(crypto_quotes.best_bid_price).alias("avg_price")
    )
    .group_by(crypto_quotes.pair))

result = (table.query()
    .from_subquery(Subquery(subquery_builder), alias="avg_prices")
    .where("avg_price > 50000")
    .to_list())

Example 27: DDL Operations - Creating Tables

from chpy import ClickHouseClient, DDL
from chpy.orm import Table, Column
from chpy import LowCardinality, Nullable, Array, Map, DateTime

client = ClickHouseClient(host="localhost", database="my_db")
ddl = DDL(client)

# Create table from schema
columns = [
    Column("id", "UInt64"),
    Column("name", LowCardinality("String")),
    Column("value", "Float64"),
    Column("tags", Array("String")),
    Column("metadata", Map("String", "String")),
    Column("created_at", DateTime("UTC")),
    Column("description", Nullable("String"))
]
schema = Table("my_table", "my_db", columns)

ddl.create_table(
    schema,
    engine="MergeTree",
    order_by="id",
    partition_by="toYYYYMM(created_at)",
    primary_key="id"
)

# Or create table with string name
ddl.create_table(
    "users",
    columns=columns,
    database="my_db",
    order_by="id"
)

# Create table with advanced options
ddl.create_table(
    schema,
    engine="MergeTree",
    order_by=["created_at", "id"],
    partition_by="toYYYYMM(created_at)",
    primary_key=["id"],
    settings={"index_granularity": 8192}
)

Example 28: DDL Operations - Altering Tables

# Add column
ddl.add_column(
    "my_db.my_table",
    Column("new_col", "String"),
    after="id"
)

# Drop column
ddl.drop_column("my_db.my_table", "old_col")

# Modify column
ddl.modify_column(
    "my_db.my_table",
    Column("name", "FixedString(100)")
)

# Rename table
ddl.rename_table("my_db.old_table", "new_table")

Example 29: DDL Operations - Databases and Views

# Create database
ddl.create_database("my_database", engine="Atomic")

# Drop database
ddl.drop_database("my_database")

# Create materialized view
target_columns = [
    Column("pair", "String"),
    Column("avg_price", "Float64"),
]
target_table = Table("mv_target", "my_db", target_columns)

ddl.create_materialized_view(
    "my_view",
    target_table,
    "SELECT pair, avg(price) as avg_price FROM source_table GROUP BY pair",
    database="my_db",
    order_by="pair"
)

# Drop materialized view
ddl.drop_materialized_view("my_db.my_view")

# Create distributed table
columns = [Column("id", "UInt64"), Column("name", "String")]
schema = Table("dist_table", "my_db", columns)

ddl.create_distributed_table(
    schema,
    cluster="my_cluster",
    local_table="my_db.local_table",
    sharding_key="rand()"
)

Example 30: ClickHouse Type System

from chpy import (
    LowCardinality, Nullable, Array, Tuple, Map, Nested,
    FixedString, Enum, IPv4, IPv6, UUID, Date, DateTime, DateTime64,
    LowCardinalityNullable, NullableArray, ArrayNullable
)
from chpy.orm import Column, Table

# LowCardinality for string optimization
Column("exchange", LowCardinality("String"))
Column("name", LowCardinality(Nullable("String")))  # Nested types

# Nullable types
Column("description", Nullable("String"))
Column("tags", Nullable(Array("String")))

# Array types
Column("tags", Array("String"))
Column("prices", Array("Float64"))
Column("nested_tags", Array(Nullable("String")))

# Tuple types
Column("coordinates", Tuple("Float64", "Float64"))
Column("metadata", Tuple("String", "Int64", "Float64"))

# Map types
Column("settings", Map("String", "String"))
Column("counts", Map("String", "Int64"))

# Nested types
Column("user", Nested("name", "String", "age", "Int64"))
# Or with tuples:
Column("user", Nested(("name", "String"), ("age", "Int64")))

# FixedString
Column("code", FixedString(10))

# Enum types
Column("status", Enum("active", 1, "inactive", 0))
# Or with dict:
Column("status", Enum({"active": 1, "inactive": 0}))

# Special types
Column("ip_address", IPv4())
Column("ipv6_address", IPv6())
Column("user_id", UUID())
Column("created", Date())
Column("timestamp", DateTime("UTC"))
Column("precise_time", DateTime64(3, "UTC"))  # 3 decimal places

# Convenience functions
Column("name", LowCardinalityNullable("String"))  # LowCardinality(Nullable(String))
Column("tags", NullableArray("String"))  # Nullable(Array(String))
Column("items", ArrayNullable("String"))  # Array(Nullable(String))

# Use in table definitions
columns = [
    Column("id", "UInt64"),
    Column("name", LowCardinality("String")),
    Column("tags", Array("String")),
    Column("metadata", Map("String", "String")),
    Column("created_at", DateTime("UTC")),
]
table = Table("my_table", "my_db", columns)

Example 31: Row Objects with Schema

# When using a schema, results are returned as Row objects
# Row objects support both attribute and dictionary access

results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(5)
    .to_list())

for row in results:
    # Attribute-style access (when schema is available)
    print(row.pair)  # or row['pair']
    print(row.best_bid_price)  # or row['best_bid_price']
    
    # Dictionary-style access
    print(row['pair'])
    print(row.get('pair', 'default'))
    
    # Convert to dict
    row_dict = row.to_dict()

Example 32: Complex Aggregations

from chpy.functions import (
    quantile, quantileExact, stddevPop, stddevSamp,
    argMin, argMax, topK, uniq
)

# Advanced aggregations
results = (table.query()
    .select(
        crypto_quotes.pair,
        avg(crypto_quotes.best_bid_price).alias("avg_price"),
        quantile(0.5)(crypto_quotes.best_bid_price).alias("median_price"),
        quantile(0.95)(crypto_quotes.best_bid_price).alias("p95_price"),
        stddevPop(crypto_quotes.best_bid_price).alias("stddev"),
        argMin(crypto_quotes.timestamp_ms, crypto_quotes.best_bid_price).alias("min_price_time"),
        topK(5)(crypto_quotes.exchange).alias("top_exchanges"),
        uniq(crypto_quotes.exchange).alias("unique_exchanges")
    )
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=7))
    .group_by(crypto_quotes.pair)
    .to_list())

Example 33: Array Functions

from chpy.functions import (
    array, arraySum, arrayAvg, arrayMax, arrayMin,
    arrayElement, has, hasAll, hasAny, indexOf
)

# Working with array columns
results = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.bid_prices,
        arraySum(crypto_quotes.bid_prices).alias("total_bids"),
        arrayAvg(crypto_quotes.bid_prices).alias("avg_bid"),
        arrayMax(crypto_quotes.bid_prices).alias("max_bid"),
        arrayMin(crypto_quotes.bid_prices).alias("min_bid"),
        arrayElement(crypto_quotes.bid_prices, 1).alias("first_bid"),
        has(crypto_quotes.bid_prices, 50000).alias("has_50k")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .limit(5)
    .to_list())

Example 34: Time Series Analysis

from chpy.functions import (
    toStartOfHour, toStartOfDay, toStartOfWeek,
    addDays, subtractDays, dateDiff
)

# Time-based aggregations
results = (table.query()
    .select(
        toStartOfHour(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))).alias("hour"),
        crypto_quotes.pair,
        avg(crypto_quotes.best_bid_price).alias("avg_price"),
        min(crypto_quotes.best_bid_price).alias("min_price"),
        max(crypto_quotes.best_bid_price).alias("max_price"),
        count().alias("quote_count")
    )
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.timestamp_ms >= datetime.now() - timedelta(days=7))
    .group_by(
        toStartOfHour(toDateTime(divide(crypto_quotes.timestamp_ms, 1000))),
        crypto_quotes.pair
    )
    .order_by("hour", desc=True)
    .to_dataframe())

Example 35: Inserting Data

# Insert data into table
data = [
    {
        "pair": "BTC-USDT",
        "best_bid_price": 50000.0,
        "best_bid_size": 1.5,
        "best_ask_price": 50001.0,
        "best_ask_size": 2.0,
        "timestamp_ms": int(datetime.now().timestamp() * 1000),
        "exchange": "BINANCE",
        "sequence_number": 1,
        "inserted_at": int(datetime.now().timestamp() * 1000)
    },
    # ... more rows
]

table.insert(data)

# Or use the client directly
client.insert("stockhouse.crypto_quotes", data)

---

API Reference

ClickHouseClient

Base client for ClickHouse operations.

Initialization

client = ClickHouseClient(
    host="localhost",
    port=8123,
    username="default",
    password="",
    database="default",
    **kwargs  # Additional connection parameters
)

Methods

• execute(query, parameters=None) - Execute a SELECT query, returns list of dicts
• query_df(query, parameters=None) - Execute query, returns pandas DataFrame
• query_np(query, parameters=None) - Execute query, returns numpy array
• query_arrow(query, parameters=None) - Execute query, returns PyArrow Table
• execute_command(query, parameters=None) - Execute non-SELECT command (INSERT, CREATE, etc.)
• insert(table, data) - Insert data into a table
• close() - Close the connection

Context Manager

with ClickHouseClient(...) as client:
    # Use client
    pass  # Connection automatically closed

TableWrapper

Generic wrapper for any ClickHouse table.

Initialization

table = TableWrapper(
    client=client,
    table_name="my_table",
    database="my_db",
    schema=optional_table_schema  # Optional Table object for type safety
)

Methods

• query() - Start building a query, returns QueryBuilder
• insert(data) - Insert data into the table
• c or columns - Access to schema columns (if schema provided)

CryptoQuotesTable

Specialized wrapper for the crypto_quotes table, extends TableWrapper.

Initialization

table = CryptoQuotesTable(client, database="stockhouse")

Helper Methods

• get_valid_exchanges() - Get list of valid exchange names
• get_exchange_base_currencies(exchange) - Get base currencies for exchange
• get_exchange_currencies(exchange) - Get supported currencies for exchange
• get_exchange_pairs(exchange) - Get all valid trading pairs for exchange
• is_valid_pair(pair, exchange=None) - Check if trading pair is valid
• get_all_valid_pairs() - Get all valid trading pairs across all exchanges

QueryBuilder

Fluent query builder returned by table.query().

Filtering Methods

• where(condition) - Add WHERE condition (chainable)
  • Accepts: ColumnExpression, CombinedExpression, SubqueryExpression, or raw SQL string
• having(condition) - Add HAVING clause for GROUP BY queries (chainable)

Query Building Methods

• select(*columns) - Specify columns to select (chainable)
  • Accepts: Column objects, Function objects, AggregateFunction objects, Subquery objects, or raw SQL strings
• join(table, condition=None, join_type="INNER", alias=None) - Add JOIN clause (chainable)
  • table: Table object or table name string
  • condition: ColumnExpression, CombinedExpression, or raw SQL string (required except for CROSS JOIN)
  • join_type: "INNER", "LEFT", "RIGHT", "FULL", or "CROSS"
  • alias: Optional table alias
• from_subquery(subquery, alias) - Use subquery as FROM clause (chainable)
• order_by(column, desc=True) - Specify ordering (chainable)
• limit(n) - Limit number of results (chainable)
• group_by(*columns) - Group by columns (chainable)

Execution Methods

• to_list() - Execute and return as list of Row objects (if schema) or dictionaries
• to_dict(key_column, value_column=None) - Execute and return as dictionary
• to_dataframe() - Execute and return as pandas DataFrame
• to_numpy(columns=None, dtype=None) - Execute and return as numpy array
• to_json(indent=None) - Execute and return as JSON string
• to_csv(path=None, **kwargs) - Execute and return/write CSV
• to_parquet(path, **kwargs) - Execute and write Parquet file
• count() - Count rows matching filters
• first() - Get first result (or None)
• exists() - Check if any rows match (returns bool)
• __iter__() - Make query builder iterable

ORM Classes

Table

Represents a database table with columns.

table = Table(name="my_table", database="my_db", columns=[...])

Column

Represents a database column.

column = Column(name="pair", type_="String", table=optional_table)

Column objects support:

• Comparison operators: ==, !=, <, <=, >, >=
• in_(values) - IN operator
• not_in(values) - NOT IN operator
• like(pattern) - LIKE operator
• Can be combined with & (AND) and | (OR)

Row

Represents a row from a query result.

• Attribute access: row.column_name
• Dictionary access: row['column_name']
• get(key, default) - Get with default
• to_dict() - Convert to dictionary

DDL

High-level DDL operations for table and database management.

Methods

• create_table(table, columns=None, database=None, engine="MergeTree", order_by=None, ...)
• drop_table(table, database=None, if_exists=True)
• add_column(table, column, database=None, after=None, if_not_exists=True)
• drop_column(table, column_name, database=None, if_exists=True)
• modify_column(table, column, database=None)
• rename_table(old_table, new_name, database=None)
• create_database(database, if_not_exists=True, engine=None, settings=None)
• drop_database(database, if_exists=True)
• create_materialized_view(view, to_table, select_query, ...)
• drop_materialized_view(view, database=None, if_exists=True)
• create_distributed_table(table, cluster, local_table, ...)

Functions

The library provides comprehensive ClickHouse function coverage. Import from chpy.functions:

from chpy.functions import (
    # Aggregate
    count, sum, avg, min, max, quantile, stddevPop, stddevSamp,
    # String
    length, upper, lower, substring, concat, startsWith, endsWith,
    # Date/Time
    toYear, toMonth, toDayOfMonth, toHour, addDays, subtractDays,
    # Math
    abs, sqrt, round, floor, ceil,
    # Type conversion
    toString, toInt64, toFloat64, toDateTime,
    # Conditional
    if_ as if_func, coalesce,
    # Array
    array, arraySum, arrayAvg, arrayMax, arrayMin,
    # And many more...
)

All functions support:

• .alias(name) - Set column alias
• .over(window_spec) - Add OVER clause for window functions

WindowSpec

Window specification for OVER clauses.

from chpy.functions.base import WindowSpec

WindowSpec()
    .partition_by(column1, column2)
    .order_by(column3, desc=True)
    .rows_between("UNBOUNDED PRECEDING", "CURRENT ROW")

Methods

• partition_by(*columns) - Add PARTITION BY clause
• order_by(*columns, desc=False) - Add ORDER BY clause
• rows_between(start, end) - Add ROWS BETWEEN frame
• range_between(start, end) - Add RANGE BETWEEN frame
• to_sql() - Convert to SQL OVER clause

---

Best Practices

1. Use Schema Objects for Type Safety

# Good: Type-safe with autocomplete
df = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .to_dataframe())

# Avoid: Raw strings (no type checking)
df = (table.query()
    .where("pair = 'BTC-USDT'")
    .to_dataframe())

2. Use Method Chaining

# Good: Fluent interface
results = (table.query()
    .where(crypto_quotes.pair == "BTC-USDT")
    .where(crypto_quotes.exchange == "BINANCE")
    .order_by(crypto_quotes.timestamp_ms, desc=True)
    .limit(100)
    .to_list())

# Avoid: Multiple variables
builder = table.query()
builder = builder.where(crypto_quotes.pair == "BTC-USDT")
builder = builder.where(crypto_quotes.exchange == "BINANCE")
results = builder.to_list()

3. Use Context Managers

# Good: Automatic connection management
with ClickHouseClient(...) as client:
    table = CryptoQuotesTable(client)
    results = table.query().to_list()

# Avoid: Manual connection management
client = ClickHouseClient(...)
try:
    table = CryptoQuotesTable(client)
    results = table.query().to_list()
finally:
    client.close()

4. Select Only Needed Columns

# Good: Select specific columns
df = (table.query()
    .select(crypto_quotes.pair, crypto_quotes.best_bid_price)
    .to_dataframe())

# Avoid: Selecting all columns when you only need a few
df = (table.query()
    .to_dataframe())  # Selects all columns

5. Use Appropriate Output Formats

# For data analysis: DataFrame
df = table.query().to_dataframe()

# For numerical computation: NumPy
arr = table.query().select(...).to_numpy()

# For iteration: List
results = table.query().to_list()

# For export: CSV/Parquet
table.query().to_csv("output.csv")
table.query().to_parquet("output.parquet")

6. Use Functions for Transformations

# Good: Use library functions
results = (table.query()
    .select(
        upper(crypto_quotes.pair).alias("pair_upper"),
        round(crypto_quotes.best_bid_price, 2).alias("rounded_price")
    )
    .to_list())

# Avoid: Raw SQL when functions are available
results = (table.query()
    .select("upper(pair) as pair_upper, round(best_bid_price, 2) as rounded_price")
    .to_list())

7. Use Window Functions for Advanced Analytics

# Good: Window functions for running calculations
result = (table.query()
    .select(
        crypto_quotes.pair,
        crypto_quotes.best_bid_price,
        avg(crypto_quotes.best_bid_price).over(
            WindowSpec().partition_by(crypto_quotes.pair)
        ).alias("avg_price")
    )
    .to_dataframe())

---

Table Schema

The crypto_quotes table has the following columns:

• pair (String): Trading pair (e.g., "BTC-USDT")
• best_bid_price (Float64): Best bid price
• best_bid_size (Float64): Best bid size
• best_ask_price (Float64): Best ask price
• best_ask_size (Float64): Best ask size
• bid_prices (Array(Float64)): Array of bid prices
• bid_sizes (Array(Float64)): Array of bid sizes
• ask_prices (Array(Float64)): Array of ask prices
• ask_sizes (Array(Float64)): Array of ask sizes
• timestamp_ms (UInt64): Timestamp in milliseconds
• exchange (LowCardinality(String)): Exchange name
• sequence_number (UInt64): Sequence number
• inserted_at (UInt64): Insert timestamp

---

License

MIT

---

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

---

Support

For issues, questions, or contributions, please open an issue on GitHub.