SQL تحلیلی: Window Functions، تجمیع زمانی، ClickHouse و TimescaleDB

۱. چرا SQL؟

در درس ۱۴ داده Tick را با pandas و Polars پردازش کردیم. این روش تا وقتی خوب است که داده در حافظه یک ماشین جا شود و فقط خودت با آن کار کنی. وقتی داده به ماه‌ها و صدها میلیون ردیف می‌رسد، یا چند نفر و چند سرویس هم‌زمان از آن کوئری می‌گیرند، منطقی‌تر است محاسبه را پیش داده ببری به‌جای اینکه داده را پیش محاسبه بیاوری. ابزار این کار SQL و یک پایگاه داده تحلیلی است.

به‌عنوان برنامه‌نویس بک‌اند احتمالاً SQL بلدی؛ ولی SQL تحلیلی با SELECT ... WHERE id = ? فرق دارد. ستون فقرات آن Window Functionهاست: محاسبه روی مجموعه‌ای از ردیف‌های مرتبط، بدون اینکه ردیف‌ها مثل GROUP BY در هم ادغام شوند.

همه کوئری‌های بخش ۲ تا ۴ را با SQLite (از کتابخانه استاندارد Python) اجرا کرده‌ایم. جز تابع گرد کردن زمان که هر موتور نسخه خودش را دارد (بخش ۴)، همه SQL استاندارد هستند و در PostgreSQL هم کار می‌کنند. خروجی‌ها را برای خوانایی گرد کرده‌ایم.

۲. جدول نمونه و حذف ردیف تکراری

هفت معامله BTC/USDT در دو دقیقه. دو معامله اول همان سفارش خرید Market درس ۳ است که دو سطح Ask را با یک Timestamp مصرف کرد. معامله ۶ هم دو بار آمده؛ اتفاقی رایج وقتی اتصال قطع و وصل می‌شود و کلاینت پیام‌های اخیر را دوباره دریافت می‌کند (درس ۱۶).

CREATE TABLE trades (
    trade_id INTEGER,
    ts       TEXT,      -- UTC, 'YYYY-MM-DD HH:MM:SS.mmm'
    price    REAL,
    qty      REAL,
    side     TEXT       -- taker side
);

INSERT INTO trades VALUES
    (1, '2027-01-11 10:00:05.120', 100010, 0.3, 'buy'),
    (2, '2027-01-11 10:00:05.120', 100020, 0.5, 'buy'),
    (3, '2027-01-11 10:00:31.400',  99990, 0.2, 'sell'),
    (4, '2027-01-11 10:00:58.900', 100000, 0.1, 'buy'),
    (5, '2027-01-11 10:01:02.000', 100030, 0.2, 'buy'),
    (6, '2027-01-11 10:01:15.500',  99980, 0.4, 'sell'),
    (6, '2027-01-11 10:01:15.500',  99980, 0.4, 'sell'),  -- duplicate after a reconnect
    (7, '2027-01-11 10:01:44.250',  99970, 0.3, 'sell');

ROW_NUMBER() به هر ردیف در گروه خودش یک شماره ترتیبی می‌دهد. اگر گروه را trade_id بگیری، از هر معامله فقط ردیف شماره ۱ را نگه می‌داری:

CREATE VIEW trades_clean AS
SELECT trade_id, ts, price, qty, side
FROM (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY trade_id ORDER BY ts) AS rn
    FROM trades
) AS t
WHERE rn = 1;

اهمیت این کار را در بخش ۴ با عدد می‌بینیم: همین یک ردیف تکراری، VWAP دقیقه دوم را حدود ۲.۴ دلار جابه‌جا می‌کند. در داده واقعی کلید یکتا معمولاً ترکیب نماد و trade_id است، نه trade_id به‌تنهایی.

۳. Window Functions: LAG، جمع تجمعی و VWAP جاری

سه محاسبه رایج در یک کوئری:

LAG(price): قیمت ردیف قبلی؛ برای تغییر قیمت معامله‌به‌معامله یا بازده کندل‌ها.
جمع تجمعی حجم علامت‌دار: خرید Taker مثبت، فروش منفی. این جریان سفارش خالص (Net Order Flow) است؛ جریان سفارش موضوع درس ۴ بود. اگر به‌جای همه معامله‌های بازار، Fillهای خودت را جمع بزنی، موجودی لحظه‌به‌لحظه‌ات به دست می‌آید.
VWAP جاری: VWAP از ابتدای بازه تا هر ردیف؛ معیار رایج برای سنجیدن کیفیت اجرای سفارش‌های بزرگ.

SELECT
    trade_id,
    price,
    price - LAG(price) OVER (ORDER BY ts, trade_id)            AS change,
    SUM(CASE side WHEN 'buy' THEN qty ELSE -qty END)
        OVER (ORDER BY ts)                                     AS flow_default,
    SUM(CASE side WHEN 'buy' THEN qty ELSE -qty END) OVER w    AS flow,
    SUM(price * qty) OVER w / SUM(qty) OVER w                  AS vwap_running
FROM trades_clean
WINDOW w AS (ORDER BY ts, trade_id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
ORDER BY ts, trade_id;

trade_id  price    change  flow_default  flow  vwap_running
1         100010   NULL    0.8           0.3   100010.00
2         100020   10      0.8           0.8   100016.25
3          99990   -30     0.6           0.6   100011.00
4         100000   10      0.7           0.7   100010.00
5         100030   30      0.9           0.9   100013.08
6          99980   -50     0.5           0.5   100005.29
7          99970   -10     0.2           0.2   100000.00

به ستون flow_default در ردیف اول دقت کن: 0.8، در حالی که تا آن لحظه فقط 0.3 BTC خرید شده بود. دلیلش قاب پیش‌فرض (Window Frame) است. وقتی فقط ORDER BY می‌نویسی، قاب پیش‌فرض در استاندارد SQL برابر RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW است؛ و RANGE همه ردیف‌هایی را که مقدار ts برابر دارند (Peer) با هم حساب می‌کند. معامله‌های ۱ و ۲ Timestamp یکسان دارند، پس هر دو جمع 0.8 را می‌گیرند. ستون flow با ROWS و مرتب‌سازی کامل (ts, trade_id) ردیف‌به‌ردیف جلو می‌رود و درست است.

دو عادت: همیشه کلید مرتب‌سازی را یکتا کن (مثلاً ts, trade_id)، چون در داده Tick Timestamp تکراری عادی است؛ و وقتی جمع تجمعی می‌خواهی، قاب را صریحاً با ROWS بنویس. برای داده چند نمادی یا چندروزه، PARTITION BY symbol یا PARTITION BY روز را هم اضافه کن تا جمع‌ها از یک نماد یا یک روز به بعدی نشت نکنند.

۴. تجمیع زمانی: کندل OHLCV و VWAP

کندل‌سازی در SQL یعنی GROUP BY روی زمانِ گرد‌شده. High، Low، حجم و VWAP با تابع‌های تجمیعی معمولی به دست می‌آیند. Open و Close سخت‌ترند، چون «اولین» و «آخرین» در SQL استاندارد تابع تجمیعی ندارند. راه قابل‌حمل: با ROW_NUMBER اولین و آخرین معامله هر دقیقه را علامت بزن.

WITH ranked AS (
    SELECT *,
        strftime('%Y-%m-%d %H:%M', ts) AS bucket,          -- engine-specific: see below
        ROW_NUMBER() OVER (PARTITION BY strftime('%Y-%m-%d %H:%M', ts)
                           ORDER BY ts, trade_id)           AS rn_first,
        ROW_NUMBER() OVER (PARTITION BY strftime('%Y-%m-%d %H:%M', ts)
                           ORDER BY ts DESC, trade_id DESC) AS rn_last
    FROM trades_clean
)
SELECT
    bucket,
    MAX(CASE WHEN rn_first = 1 THEN price END) AS open,
    MAX(price)                                 AS high,
    MIN(price)                                 AS low,
    MAX(CASE WHEN rn_last = 1 THEN price END)  AS close,
    SUM(qty)                                   AS volume,
    SUM(price * qty) / SUM(qty)                AS vwap,
    COUNT(*)                                   AS n_trades
FROM ranked
GROUP BY bucket
ORDER BY bucket;

دقیقه	Open	High	Low	Close	حجم	VWAP	تعداد
10:00	100,010	100,020	99,990	100,000	1.1	100,010.00	۴
10:01	100,030	100,030	99,970	99,970	0.9	99,987.78	۳

VWAP دقیقه اول را با دست هم می‌شود چک کرد:

(30,003 + 50,010 + 19,998 + 10,000) / 1.1 = 110,011 / 1.1 = 100,010

اگر همین کوئری را روی جدول خام trades (بدون حذف تکراری) اجرا کنی، دقیقه دوم حجم 1.3 و VWAP برابر 99,985.38 می‌گیرد؛ یعنی حجم ۴۴٪ بیشتر و VWAP حدود ۲.۴ دلار پایین‌تر از مقدار درست. هیچ خطایی هم نمی‌بینی.

تساوی زمانی دوباره: Open دقیقه اول 100,010 است چون ترتیب را با trade_id شکسته‌ایم. اگر فقط با ts مرتب می‌کردی، بین معامله ۱ و ۲ (هر دو در 10:00:05.120) هر کدام ممکن بود Open شود و نتیجه از یک اجرا به اجرای دیگر فرق می‌کرد.

تابع گرد کردن زمان در هر موتور فرق دارد:

موتور	گرد کردن به دقیقه	بازه دلخواه
SQLite	`strftime('%Y-%m-%d %H:%M', ts)`	محاسبه دستی روی Epoch
PostgreSQL	`date_trunc('minute', ts)`	`date_bin('5 minutes', ts, origin)` (از نسخه ۱۴)
TimescaleDB	`time_bucket('1 minute', ts)`	`time_bucket('5 minutes', ts)`
ClickHouse	`toStartOfMinute(ts)`	`toStartOfInterval(ts, INTERVAL 5 MINUTE)`

۵. ذخیره سطری و ستونی

پایگاه داده‌های رایج بک‌اند مثل PostgreSQL و MySQL سطری (Row Store) هستند: همه ستون‌های یک ردیف کنار هم روی دیسک ذخیره می‌شوند. این برای کار تراکنشی (OLTP) ایده‌آل است: «سفارش شماره X را بخوان و وضعیتش را عوض کن» یعنی یک ردیف کامل را بخوان و بنویس.

کوئری تحلیلی (OLAP) الگوی دیگری دارد: «میانگین price و جمع qty در یک ماه». دو ستون از ده ستون، ولی روی صدها میلیون ردیف. پایگاه داده ستونی (Column Store) هر ستون را جدا ذخیره می‌کند، پس:

فقط ستون‌های لازم خوانده می‌شوند. همان Projection Pushdown درس ۱۴، این بار در سطح دیسک.
فشرده‌سازی خیلی بهتر است. مقادیر یک ستون شبیه هم‌اند؛ Timestampهای مرتب را می‌شود به‌صورت اختلاف با قبلی (Delta) ذخیره کرد و ستون side فقط دو مقدار دارد.
پردازش برداری. موتور، ستون را تکه‌تکه و با همان منطق برداری‌سازی درس ۱۳ پردازش می‌کند.

بهایش این است که به‌روزرسانی و حذف تک‌ردیف گران است و نوشتن ردیف‌به‌ردیف کارایی ندارد؛ داده باید دسته‌ای (Batch) وارد شود. خوشبختانه داده بازار تقریباً همیشه فقط اضافه می‌شود (Append-only) و بعداً تغییر نمی‌کند.

دو گزینه رایج متن‌باز برای این کار را با رویکرد متفاوت می‌بینیم: ClickHouse که از پایه ستونی است، و TimescaleDB که افزونه‌ای روی PostgreSQL است. سینتکس هر دو در نسخه‌های مختلف تغییر کرده؛ کدهای زیر مفهومی‌اند و پیش از استفاده، مستندات نسخه نصب‌شده خودت را مرجع بگیر.

۶. ClickHouse: MergeTree و کلید مرتب‌سازی

CREATE TABLE trades
(
    symbol   LowCardinality(String),
    ts       DateTime64(3, 'UTC'),
    trade_id UInt64,
    price    Float64,
    qty      Float64,
    side     Enum8('buy' = 1, 'sell' = 2)
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ts)
ORDER BY (symbol, ts, trade_id);

مفاهیم کلیدی:

MergeTree: خانواده اصلی موتورهای جدول ClickHouse. هر Insert یک «قطعه» (Part) مرتب‌شده روی دیسک می‌سازد و ClickHouse در پس‌زمینه قطعه‌ها را با هم ادغام (Merge) می‌کند. به همین دلیل Insertهای تک‌ردیفی و پرتعداد بدترین الگو هستند؛ داده را دسته‌ای وارد کن.
ORDER BY: مهم‌ترین تصمیم طراحی جدول. داده روی دیسک به این ترتیب مرتب می‌شود و یک ایندکس پراکنده (Sparse Index) روی آن ساخته می‌شود. کوئری‌ای که روی symbol و بازه ts فیلتر می‌کند، بخش عمده داده را اصلاً نمی‌خواند. ترتیب ستون‌ها باید با الگوی کوئری‌هایت جور باشد.
PARTITION BY: واحد مدیریت داده (حذف یا جابه‌جایی یک ماه کامل)، نه ابزار اصلی سرعت. پارتیشن را درشت نگه دار؛ ماهانه یا در داده خیلی حجیم روزانه. پارتیشن ریز (ساعتی یا بر اساس نماد) تعداد قطعه‌ها را زیاد و کارایی را کم می‌کند.
LowCardinality(String): همان ایده Category در pandas؛ برای ستون‌هایی با تعداد کم مقدار یکتا.

کندل‌ها در ClickHouse ساده‌ترند، چون تابع argMin و argMax دقیقاً «قیمتِ ردیفی که کمترین یا بیشترین زمان را دارد» را برمی‌گرداند. با Tuple، تساوی زمانی هم با trade_id شکسته می‌شود:

SELECT
    toStartOfMinute(ts)                AS bucket,
    argMin(price, (ts, trade_id))      AS open,
    max(price)                         AS high,
    min(price)                         AS low,
    argMax(price, (ts, trade_id))      AS close,
    sum(qty)                           AS volume,
    sum(price * qty) / sum(qty)        AS vwap
FROM trades
WHERE symbol = 'BTCUSDT'
  AND ts >= '2027-01-11 00:00:00' AND ts < '2027-01-12 00:00:00'
GROUP BY bucket
ORDER BY bucket;

ClickHouse Asof Join درس ۱۴ را هم مستقیم پشتیبانی می‌کند. شرط > (به‌جای >=) همان معادل allow_exact_matches=False است:

SELECT t.ts, t.price, q.bid, q.ask
FROM trades AS t
ASOF LEFT JOIN quotes AS q
    ON t.symbol = q.symbol AND t.ts > q.ts;

حذف تکراری در ClickHouse: موتور ReplacingMergeTree ردیف‌هایی را که کلید ORDER BY یکسان دارند حذف می‌کند، ولی فقط هنگام ادغام قطعه‌ها که زمانش معلوم نیست. تا آن موقع تکراری‌ها در نتیجه کوئری هستند، مگر اینکه از FINAL استفاده کنی که کوئری را کندتر می‌کند. پس آن را تضمین فوری یکتایی فرض نکن؛ بهتر است پیش از ورود به پایگاه داده هم تکراری‌ها را حذف کنی (درس ۱۶).

۷. TimescaleDB: Hypertable و Continuous Aggregate

TimescaleDB یک افزونه PostgreSQL است؛ یعنی همان SQL، همان ابزارها و همان Join با جدول‌های معمولی. ایده اصلی آن Hypertable است: جدولی که از بیرون یک جدول عادی است ولی در داخل بر اساس زمان به تکه‌هایی به نام Chunk تقسیم می‌شود. کوئری روی یک بازه زمانی فقط Chunkهای مربوط را می‌خواند و داده قدیمی را می‌توان Chunk به Chunk فشرده یا حذف کرد. Chunkهای فشرده‌شده در قالبی ستونی نگه داشته می‌شوند؛ یعنی ترکیبی از ذخیره سطری برای داده تازه و ستونی برای داده قدیمی.

CREATE TABLE trades (
    ts       TIMESTAMPTZ      NOT NULL,
    symbol   TEXT             NOT NULL,
    trade_id BIGINT           NOT NULL,
    price    DOUBLE PRECISION NOT NULL,
    qty      DOUBLE PRECISION NOT NULL,
    side     TEXT             NOT NULL,
    UNIQUE (symbol, trade_id, ts)
);

-- TimescaleDB 2.13 and later:
SELECT create_hypertable('trades', by_range('ts', INTERVAL '1 day'));
-- Older versions:
-- SELECT create_hypertable('trades', 'ts', chunk_time_interval => INTERVAL '1 day');

قید UNIQUE همراه با INSERT ... ON CONFLICT DO NOTHING تکراری‌ها را هنگام ورود رد می‌کند. یک محدودیت Hypertable: هر قید یکتا باید ستون زمان را هم شامل شود؛ به همین دلیل ts در آن آمده است.

کندل‌سازی با time_bucket و دو تابع first و last خود TimescaleDB:

SELECT
    time_bucket('1 minute', ts)         AS bucket,
    first(price, ts)                    AS open,
    max(price)                          AS high,
    min(price)                          AS low,
    last(price, ts)                     AS close,
    sum(qty)                            AS volume,
    sum(price * qty) / sum(qty)         AS vwap
FROM trades
WHERE symbol = 'BTCUSDT'
  AND ts >= '2027-01-11' AND ts < '2027-01-12'
GROUP BY bucket
ORDER BY bucket;

first(price, ts) قیمت ردیفی با کمترین ts را برمی‌گرداند؛ ولی اگر دو ردیف Timestamp برابر داشته باشند، معلوم نیست کدام انتخاب شود. اگر این برایت مهم است، روش ROW_NUMBER بخش ۴ را به کار ببر.

Continuous Aggregate

اگر داشبورد یا تحلیل‌گری مدام کندل یک‌دقیقه‌ای ماه گذشته را بخواهد، محاسبه دوباره آن از میلیون‌ها معامله در هر بار اتلاف است. Continuous Aggregate یک Materialized View است که TimescaleDB به‌صورت تدریجی به‌روز نگه می‌دارد: فقط بازه‌هایی را که داده تازه گرفته‌اند دوباره حساب می‌کند.

CREATE MATERIALIZED VIEW ohlcv_1m
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 minute', ts) AS bucket,
    symbol,
    first(price, ts)            AS open,
    max(price)                  AS high,
    min(price)                  AS low,
    last(price, ts)             AS close,
    sum(qty)                    AS volume,
    sum(price * qty)            AS notional
FROM trades
GROUP BY bucket, symbol
WITH NO DATA;

SELECT add_continuous_aggregate_policy('ohlcv_1m',
    start_offset      => INTERVAL '1 hour',
    end_offset        => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 minute');

Policy می‌گوید هر دقیقه، بازه یک ساعت قبل تا یک دقیقه قبل را تازه کن. end_offset عمداً دقیقه جاری را کنار می‌گذارد، چون هنوز کامل نشده است؛ همان مسئله کندل ناقص درس ۱۴.

به یک انتخاب دقت کن: در View به‌جای vwap، ستون notional را ذخیره کرده‌ایم. VWAP را نمی‌شود میانگین گرفت؛ VWAP پنج‌دقیقه‌ای از میانگین پنج VWAP یک‌دقیقه‌ای به دست نمی‌آید. ولی مجموع مبلغ و مجموع حجم را می‌شود جمع زد:

VWAP(5m) = Σ notional(1m) / Σ volume(1m)

قاعده کلی: در جدول‌های تجمیعی، کمیت‌های جمع‌پذیر را نگه دار (جمع، تعداد، کمینه، بیشینه) و نسبت‌ها را موقع کوئری بساز.

تفاوت نسخه‌ها و مجوز: دو نکته که پیش از به کار بردن TimescaleDB چک کن. اول، در نسخه‌های اخیر، حالت Real-time Aggregation (ترکیب داده Materialize‌شده با ردیف‌های تازه‌ای که هنوز وارد View نشده‌اند) برای Viewهای جدید به‌طور پیش‌فرض خاموش است و با تنظیم timescaledb.materialized_only = false روشن می‌شود. دوم، بخشی از قابلیت‌ها از جمله Continuous Aggregate و فشرده‌سازی، با مجوز متفاوتی از هسته منتشر می‌شوند و در نسخه‌هایی که فقط هسته Apache را دارند (مثلاً بعضی سرویس‌های مدیریت‌شده PostgreSQL) در دسترس نیستند.

کدام؟

ClickHouse روی اسکن و تجمیع حجم‌های خیلی بزرگ معمولاً سریع‌تر است و فشرده‌سازی قوی‌تری دارد، ولی مدل داده‌اش (ادغام پس‌زمینه، حذف تکراری تأخیری، Join محدودتر) با پایگاه داده‌های تراکنشی فرق دارد. TimescaleDB همان PostgreSQL است: اگر تیم PostgreSQL بلد است، داده‌ها باید با جدول‌های تراکنشی Join شوند و حجم در حد چند ترابایت است، شروع راحت‌تری است. این‌ها گرایش‌های کلی‌اند؛ تنها راه مطمئن، اجرای کوئری‌های واقعی خودت روی نمونه‌ای از داده خودت است.

۸. تمرین این درس

گزینه مناسب هر سؤال را انتخاب کن و در پایان نتیجه را ببین.

سؤال ۱: دو معامله Timestamp یکسان دارند. با SUM(qty) OVER (ORDER BY ts) هر دو جمع تجمعی یکسانی می‌گیرند. چرا؟

چون SQL معامله‌های هم‌زمان را یکی می‌کند چون SUM با Window Function سازگار نیست چون قاب پیش‌فرض RANGE است و ردیف‌های با ts برابر را با هم حساب می‌کند

سؤال ۲: دو کندل یک‌دقیقه‌ای داریم: VWAP برابر 100,010 با حجم 1.1 BTC و VWAP برابر 99,987.78 با حجم 0.9 BTC. VWAP دو دقیقه‌ای چقدر است؟

100,000 99,998.89 99,987.78

سؤال ۳: در جدول MergeTree با ORDER BY (symbol, ts)، این کلید چه نقشی دارد؟

یکتایی ردیف‌ها را تضمین می‌کند ترتیب ذخیره روی دیسک و ایندکس پراکنده را تعیین می‌کند تا کوئری‌های فیلترشده روی نماد و زمان بخش زیادی از داده را نخوانند فقط ترتیب پیش‌فرض خروجی SELECT را تعیین می‌کند

سؤال ۴: کدام کار برای پایگاه داده ستونی مناسب‌تر است؟

به‌روزرسانی وضعیت یک سفارش با شناسه‌اش ثبت ردیف‌به‌ردیف هر رویداد با تراکنش جداگانه محاسبه جمع حجم و VWAP روزانه روی صدها میلیون معامله

درس بعدی

در درس شانزدهم بخش چهارم را با ساختن کل مسیر داده تمام می‌کنیم: Market Data Pipeline. از اتصال WebSocket و ترکیب Snapshot با به‌روزرسانی‌های تدریجی، تا تشخیص پیام گم‌شده با شماره ترتیب، پاک‌سازی داده تکراری و نامرتب، چیدمان ذخیره‌سازی خام و نرمال‌شده، و بازسازی Order Book با ایده‌های درس ۳.

این مطلب صرفاً آموزشی است و توصیه خرید، فروش یا سرمایه‌گذاری نیست.