Use polars and sqlite for large data

[temeddix]

No description provided.

[temeddix]

@copilot Please review it

[Copilot]

@copilot Please review it

Reviewed the changes. The PR looks good overall — the migration from global/class-level state to instance-based context managers is clean and the resource lifecycle management with AsyncExitStack is well structured.

One issue I found and fixed: COUNT_ALL_CANDLES_SQL and DELETE_INVALID_CANDLES_SQL were inlining the validity conditions instead of using the VALID_CANDLE_CONDITION constant that all other SQL queries use. Fixed in commit Fix SQL constants to use VALID_CANDLE_CONDITION consistently.

Code review and CodeQL security scan both passed with no issues.

[temeddix]

@copilot

No danger of race condition or sqlite files being locked after force-close?

How is "fill hisitorical data" writing to disk right now? Row by row with sqlite? Or polars conversion first?

[Copilot]

@copilot

No danger of race condition or sqlite files being locked after force-close?
...

Race conditions / locking after force-close:

Each CandleData instance uses an asyncio.Lock (_access_lock) that wraps every write and read operation, so concurrent upserts don't interleave. On close, _close() acquires the same lock before calling connection.close() and setting _connection = None — any operation queued behind it will get a RuntimeError from _require_connection() rather than touching a closed connection.

iter_range opens its own short-lived read connection without the lock, but that's safe because WAL mode (PRAGMA journal_mode = WAL) allows concurrent readers alongside the writer without blocking.

For force-close (process killed), SQLite WAL recovery handles uncommitted writes on next open. The yoyo migration lock is handled by _break_stale_yoyo_lock, which checks if the PID stored in yoyo_lock is still running and, if not, clears the lock before retrying migrations.

How "fill historical data" writes to disk:

The download path is: CSV download → process_aggtrade_csv (polars) → combine_candle_data (polars, in memory) → _write_downloaded_candle_rows which iterates the combined DataFrame and groups rows by symbol, then calls upsert_many per symbol. upsert_many uses executemany inside a single SQLite transaction per symbol — so it's a batch write, not row-by-row.

The "fill holes" (recent gaps) path also uses upsert_many after building the candle rows from fetched aggregate trades in Python.

[temeddix]

@copilot Would it be much slower if we do not use polars after downloading history data, and insert row by row to reduce memory usage?

[Copilot]

@copilot Would it be much slower if we do not use polars after downloading history data, and insert row by row to reduce memory usage?

No, it wouldn't be meaningfully slower. The actual bottlenecks are network I/O (downloading the ZIP files) and disk I/O (SQLite executemany). The polars step sits between them and is negligible by comparison.

What polars is doing in this path: process_csv_lines already builds a list[Candle] in Python while reading the CSV line-by-line, then just wraps that list into a DataFrame. combine_candle_data then pl.concats those DataFrames. _write_downloaded_candle_rows iterates the result back to Python objects via iter_rows(named=True) — one of the slower polars iteration methods — to build the rows_by_symbol dict before calling upsert_many.

So the polars DataFrame here is essentially an intermediate container between two Python list operations. Removing it (CSV → Python list of CandleRow directly → upsert_many per file) would reduce peak memory proportionally to the number of files combined, with no perceptible speed difference.

The one genuine polars value in the current path is sort_aggtrade_csv (when Binance provides unsorted data) which uses pl.read_csv + .sort() — that part can stay.